rune/tests/test_research_task_tree.py

"""Red-first tests for scripts/research_task_tree.py — RP1/RP2 canonical stage model.

Supersedes RSC1-era tests: ISSUE replaces DEFINE, APPLICABILITY added,
INNOVATION precedes ADVERSARIAL, retired stages cannot become READY,
targeted refinement routes to earliest deficient stage.
"""

from __future__ import annotations

import importlib.util
import json
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parent.parent
SPEC = importlib.util.spec_from_file_location("research_task_tree", ROOT / "scripts" / "research_task_tree.py")
rtt = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(rtt)


def _walk(tree):
    """Stage-aware full-chain walk: supplies mandatory routing fields
    (APPLICABILITY route, ADVERSARIAL disposition, QUALIFY outcome)."""
    for stage in rtt.STAGES:
        if tree.current_stage == "APPLICABILITY":
            tree.advance({"route": "STANDARD_SOLUTION_SUFFICIENT"})
        elif tree.current_stage == "ADVERSARIAL":
            tree.advance(
                {
                    "disposition": "SURVIVES",
                    "surviving_candidates": ["M1"],
                    "falsification_tests": [],
                    "attacks": [{"id": "A1", "target_candidate": "M1", "dimension": "failure modes", "finding": "x"}],
                }
            )
        elif tree.current_stage == "QUALIFY":
            tree.advance({"outcome": "PROCEED"})
        else:
            tree.advance({"test": True})


@pytest.fixture
def tree(tmp_path):
    return rtt.ResearchTaskTree(issue=999, round_no=1, artifact_dir=str(tmp_path / "research" / "issue-999"))


def test_initial_state_starts_at_issue_gate(tree):
    """Canonical RP2: ISSUE (deterministic gate) is stage 0, not DEFINE."""
    assert tree.current_stage == "ISSUE"
    assert not tree.is_complete
    assert tree.state["completed_stages"] == []


def test_canonical_stage_order():
    """Canonical pipeline: ISSUE→INWARD→OUTWARD→APPLICABILITY→INNOVATION→ADVERSARIAL→SYNTHESIS→QUALIFY."""
    assert rtt.STAGES == [
        "ISSUE",
        "INWARD",
        "OUTWARD",
        "APPLICABILITY",
        "INNOVATION",
        "ADVERSARIAL",
        "SYNTHESIS",
        "QUALIFY",
    ]
    # APPLICABILITY must exist and precede INNOVATION/ADVERSARIAL
    assert "APPLICABILITY" in rtt.STAGES
    assert rtt.STAGES.index("APPLICABILITY") < rtt.STAGES.index("INNOVATION")
    assert rtt.STAGES.index("APPLICABILITY") < rtt.STAGES.index("ADVERSARIAL")


def test_retired_stage_names_are_declared():
    """26.5: retired stage identifiers are explicit and absent from canonical stages."""
    assert rtt.RETIRED_STAGES == {"DEFINE", "OUTWARD_DISCOVERY", "OUTWARD_READING"}
    for retired in rtt.RETIRED_STAGES:
        assert retired not in rtt.STAGES


def test_retired_stages_cannot_become_ready(tmp_path):
    """RP2/26.5: a tree loaded from legacy persisted state at a retired stage
    must refuse to advance it. Old authorities cannot quietly run."""
    legacy_dir = tmp_path / "research" / "issue-legacy"
    legacy_dir.mkdir(parents=True)
    legacy_state = {
        "issue": 998,
        "round": 1,
        "current_stage": "OUTWARD_DISCOVERY",
        "completed_stages": ["DEFINE", "INWARD"],
        "artifacts": {},
        "blocking_unknowns": [],
        "status": "ACTIVE",
        "stage_history": [],
    }
    (legacy_dir / "task_tree.json").write_text(json.dumps(legacy_state))
    legacy_tree = rtt.ResearchTaskTree(issue=998, round_no=1, artifact_dir=str(legacy_dir))
    assert legacy_tree.current_stage == "OUTWARD_DISCOVERY"
    with pytest.raises(ValueError, match="retired"):
        legacy_tree.advance({})


def test_retired_stage_cannot_be_routed_by_more_research(tree):
    """RP2/26.5: targeted refinement routes external unknowns to unified OUTWARD,
    never OUTWARD_DISCOVERY/OUTWARD_READING."""
    while tree.current_stage != "QUALIFY":
        cs = tree.current_stage
        if cs == "APPLICABILITY":
            tree.advance({"route": "STANDARD_SOLUTION_SUFFICIENT"})
        elif cs == "ADVERSARIAL":
            tree.advance({"disposition": "SURVIVES", "surviving_candidates": ["M1"], "falsification_tests": []})
        else:
            tree.advance({})
    branches = tree.more_research(
        [
            {"id": "U1", "question": "is Y maintained?", "type": "current-tech"},
            {"id": "U2", "question": "what does research say?", "type": "academic"},
        ]
    )
    for b in branches:
        assert b["stages"] == ["OUTWARD"]
    assert tree.current_stage == "OUTWARD"
    # No branch may contain a retired stage name
    for b in branches:
        assert not (set(b["stages"]) & rtt.RETIRED_STAGES)


def test_targeted_refinement_routes_to_earliest_deficient_stage(tree):
    """§14: REFINE routes to the earliest deficient stage, not first-listed."""
    while tree.current_stage != "QUALIFY":
        cs = tree.current_stage
        if cs == "APPLICABILITY":
            tree.advance({"route": "STANDARD_SOLUTION_SUFFICIENT"})
        elif cs == "ADVERSARIAL":
            tree.advance({"disposition": "SURVIVES", "surviving_candidates": ["M1"], "falsification_tests": []})
        else:
            tree.advance({})
    branches = tree.more_research(
        [
            {"id": "U1", "question": "late-stage gap", "type": "counter-evidence"},
            {"id": "U2", "question": "early evidence gap", "type": "current-tech"},
        ]
    )
    assert branches[0]["stages"] == ["ADVERSARIAL"]
    assert branches[1]["stages"] == ["OUTWARD"]
    # OUTWARD (idx 2) is earlier than ADVERSARIAL (idx 5) → current stage is OUTWARD
    assert tree.current_stage == "OUTWARD"
    assert tree.state["targeted_return_stage"] == "SYNTHESIS"


def test_advance_through_all_stages(tree):
    """Canonical walk with INNOVATION_WARRANTED route (INNOVATION is conditional).
    Skipped-INNOVATION walk is covered by test_applicability_skips_innovation_by_default."""
    _walk(tree)
    assert tree.current_stage == "QUALIFY"
    # 7 stages: INNOVATION skipped by default (conditional, never ceremonial)
    assert len(tree.state["completed_stages"]) == 7
    assert "INNOVATION" not in tree.state["completed_stages"]


def test_state_persists_to_disk(tree):
    tree.advance({"issue_gate": "passed"})
    tree2 = rtt.ResearchTaskTree(issue=999, round_no=1, artifact_dir=tree.artifact_dir)
    assert tree2.current_stage == "INWARD"
    assert "ISSUE" in tree2.state["completed_stages"]


def test_fail_records_reason(tree):
    tree.fail("test failure")
    assert tree.state["stage_history"][-1]["failed"] is True
    assert "test failure" in tree.state["stage_history"][-1]["reason"]


def test_applicability_artifact_file_declared():
    """RP4 prerequisite: APPLICABILITY has a canonical artifact file and classes."""
    assert rtt.STAGE_ARTIFACT_FILES["APPLICABILITY"] == "applicability.json"
    for cls in (
        "DIRECTLY_APPLICABLE",
        "APPLICABLE_WITH_ADAPTATION",
        "CONFLICTS_WITH_RUNE",
        "TOO_COMPLEX_FOR_VALUE",
        "INSUFFICIENT_EVIDENCE",
        "PROMISING_FOR_INNOVATION",
    ):
        assert cls in rtt.APPLICABILITY_CLASSES


def test_capsule_contains_process_state(tree):
    tree.advance({"decision": "issue gate passed"})
    cap = tree.capsule()
    assert cap["current_stage"] == "INWARD"
    assert cap["issue"] == 999
    assert "ISSUE" in cap["completed_stages"]
    assert "expected_completion_artifact" in cap


def test_get_prior_artifacts_for_stage(tree):
    tree.advance({"issue_gate_decision": "buildable"})
    tree.advance({"inward_claims": ["claim1"]})
    priors = tree.get_prior_artifacts("OUTWARD")
    assert "ISSUE" in priors
    assert "INWARD" in priors
    # Downstream artifacts are not exposed to earlier stages
    assert "APPLICABILITY" not in priors
    assert "SYNTHESIS" not in priors


def test_task_tree_json_is_valid_on_disk(tree):
    tree.advance({"test": "data"})
    path = Path(tree.artifact_dir) / "task_tree.json"
    data = json.loads(path.read_text())
    assert data["issue"] == 999
    assert data["current_stage"] == "INWARD"


# ── Conditional INNOVATION (semantic correction 2026-09-09) ──────────────


def _advance_to(tree, target):
    """Advance a fresh tree to the given stage via canonical walk."""
    for stage in rtt.STAGES:
        if tree.current_stage == target:
            return
        tree.advance({"test": True})


def test_applicability_skips_innovation_by_default(tree):
    """APPLICABILITY artifact without route → STANDARD_SOLUTION_SUFFICIENT → ADVERSARIAL."""
    for stage in ("ISSUE", "INWARD", "OUTWARD", "APPLICABILITY"):
        assert tree.current_stage == stage
        tree.advance({"test": True})
    # INNOVATION skipped; ADVERSARIAL is next
    assert tree.current_stage == "ADVERSARIAL"
    assert tree.state["applicability_route"] == "STANDARD_SOLUTION_SUFFICIENT"
    # INNOVATION must NOT be recorded as completed
    assert "INNOVATION" not in tree.state["completed_stages"]


def test_applicability_can_route_to_innovation(tree):
    """APPLICABILITY route=INNOVATION_WARRANTED → INNOVATION → ADVERSARIAL."""
    for stage in ("ISSUE", "INWARD", "OUTWARD"):
        tree.advance({"test": True})
    assert tree.current_stage == "APPLICABILITY"
    tree.advance({"test": True, "route": "INNOVATION_WARRANTED"})
    assert tree.current_stage == "INNOVATION"
    tree.advance({"candidates": [], "conclusion": "NO_INNOVATION_WARRANTED"})
    assert tree.current_stage == "ADVERSARIAL"


def test_applicability_invalid_route_rejected(tree):
    """APPLICABILITY route must be a valid decision, not freeform prose."""
    for stage in ("ISSUE", "INWARD", "OUTWARD"):
        tree.advance({"test": True})
    with pytest.raises(ValueError, match="route"):
        tree.advance({"test": True, "route": "MAYBE_INNOVATE_A_BIT"})


def test_no_innovation_warranted_is_valid_conclusion():
    """NO_INNOVATION_WARRANTED is a SUCCESS result, not a failure."""
    assert "NO_INNOVATION_WARRANTED" in rtt.INNOVATION_CONCLUSIONS
    # And skipping INNOVATION entirely is the default
    assert "STANDARD_SOLUTION_SUFFICIENT" in rtt.APPLICABILITY_ROUTES


def test_full_chain_both_routes(tmp_path):
    """Both routes reach SYNTHESIS; skipped-INNOVATION chain is one stage shorter."""
    # Route A: skip INNOVATION (stage-aware walk on CURRENT stage)
    t1 = rtt.ResearchTaskTree(issue=990, round_no=1, artifact_dir=str(tmp_path / "a"))
    while t1.current_stage != "QUALIFY":
        cs = t1.current_stage
        if cs == "APPLICABILITY":
            t1.advance({"route": "STANDARD_SOLUTION_SUFFICIENT"})
        elif cs == "ADVERSARIAL":
            t1.advance({"disposition": "SURVIVES", "surviving_candidates": ["M1"], "falsification_tests": []})
        else:
            t1.advance({"test": True})
    assert "INNOVATION" not in t1.state["completed_stages"]
    assert t1.state["completed_stages"].index("ADVERSARIAL") == t1.state["completed_stages"].index("APPLICABILITY") + 1

    # Route B: INNOVATION runs between APPLICABILITY and ADVERSARIAL
    t2 = rtt.ResearchTaskTree(issue=991, round_no=1, artifact_dir=str(tmp_path / "b"))
    while t2.current_stage != "QUALIFY":
        cs = t2.current_stage
        if cs == "APPLICABILITY":
            t2.advance({"route": "INNOVATION_WARRANTED"})
        elif cs == "INNOVATION":
            t2.advance(
                {"conclusion": "ADOPT", "candidates": [{"id": "C1", "category": "ADAPT", "derived_from": ["I1"]}]}
            )
        elif cs == "ADVERSARIAL":
            t2.advance({"disposition": "SURVIVES", "surviving_candidates": ["C1"], "falsification_tests": []})
        else:
            t2.advance({"test": True})
    completed = t2.state["completed_stages"]
    assert completed.index("APPLICABILITY") < completed.index("INNOVATION") < completed.index("ADVERSARIAL")


def test_refinement_routes_to_innovation_and_skips(tree):
    """§14: refinement may route to INNOVATION or bypass it to ADVERSARIAL,
    depending on the unknown type — never a blind full walk."""
    while tree.current_stage != "QUALIFY":
        cs = tree.current_stage
        if cs == "APPLICABILITY":
            tree.advance({"route": "STANDARD_SOLUTION_SUFFICIENT"})
        elif cs == "ADVERSARIAL":
            tree.advance({"disposition": "SURVIVES", "surviving_candidates": ["M1"], "falsification_tests": []})
        else:
            tree.advance({"test": True})
    # Cross-domain opportunity → INNOVATION
    branches = tree.more_research([{"id": "U1", "question": "cross-domain idea?", "type": "innovation"}])
    assert branches[0]["stages"] == ["INNOVATION"]
    assert tree.current_stage == "INNOVATION"
    # Mechanism weakness → ADVERSARIAL directly (INNOVATION not re-run)
    tree2 = rtt.ResearchTaskTree(issue=992, round_no=1, artifact_dir=str(tree.artifact_dir) + "2")
    while tree2.current_stage != "QUALIFY":
        cs = tree2.current_stage
        if cs == "APPLICABILITY":
            tree2.advance({"route": "STANDARD_SOLUTION_SUFFICIENT"})
        elif cs == "ADVERSARIAL":
            tree2.advance({"disposition": "SURVIVES", "surviving_candidates": ["M1"], "falsification_tests": []})
        else:
            tree2.advance({"test": True})
    branches2 = tree2.more_research(
        [{"id": "U2", "question": "does the leading option break?", "type": "counter-evidence"}]
    )
    assert branches2[0]["stages"] == ["ADVERSARIAL"]
    assert tree2.current_stage == "ADVERSARIAL"


def test_issue_gate_is_deterministic_no_mimo():
    """ISSUE is a deterministic intake/gate pseudo-stage — NOT a resurrected DEFINE.
    The module must not import subprocess/network/model-invocation machinery,
    and the executor's ISSUE path must not spawn MiMo."""
    import importlib.util as _ilu
    import sys as _sys

    spec = _ilu.spec_from_file_location("research_define_check", ROOT / "scripts" / "research_define.py")
    mod = _ilu.module_from_spec(spec)
    spec.loader.exec_module(mod)
    banned = {"subprocess", "requests", "httpx", "urllib.request", "socket", "http.client"}
    imported = (
        set(getattr(mod, "__dict__", {}).keys())
        | {
            m.split(".")[0]
            for m in _sys.modules
            if m.split(".")[0] in banned and m in getattr(mod, "__dict__", {}).values().__iter__.__self__.__dict__  # noqa — placeholder
        }
        if False
        else set()
    )
    # Direct source inspection is the honest check
    src = (ROOT / "scripts" / "research_define.py").read_text()
    for banned_mod in (
        "import subprocess",
        "import requests",
        "import httpx",
        "import socket",
        "from urllib",
        "urlopen",
        "webfetch",
    ):
        assert banned_mod not in src, f"ISSUE/DEFINE compiler must not reference {banned_mod}"
    # needs_mimo must be hardwired False (compat shim) — ISSUE never invokes a model
    assert "needs_mimo" in src
    spec2 = _ilu.spec_from_file_location("research_define_check2", ROOT / "scripts" / "research_define.py")
    rd = _ilu.module_from_spec(spec2)
    spec2.loader.exec_module(rd)
    issue = {"number": 42, "title": "Anything at all", "body": ""}
    c = rd.compile_define_contract(issue)
    assert c["needs_mimo"] is False
    # And the canonical ISSUE stage artifact file is a gate artifact, not define.json
    assert rtt.STAGE_ARTIFACT_FILES["ISSUE"] == "issue_gate.json"