rune/tests/test_applicability_executor.py

"""Tests: RP4 APPLICABILITY executor + gate (#267).

Gate semantics: candidates must be classified with evidence, route must be
a valid decision, viable candidates must cite real OUTWARD sources.
"""

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"]


@pytest.fixture
def tree(tmp_path):
    t = rtt.ResearchTaskTree(issue=970, round_no=1, artifact_dir=str(tmp_path / "research" / "issue-970"))
    # advance through ISSUE→INWARD→OUTWARD
    for stage in ("ISSUE", "INWARD", "OUTWARD"):
        t.advance({"test": True, "route": None} if False else {"test": True})
    assert t.current_stage == "APPLICABILITY"
    # seed an OUTWARD artifact with known sources
    t.state["artifacts"]["OUTWARD"] = {
        "discovery": {
            "queries": [],
            "sources_discovered": [
                {"id": "O1", "url": "https://example.org/a"},
                {"id": "O2", "url": "https://example.org/b"},
            ],
            "source_count": 2,
        },
        "reading": {"sources_read": ["O1"], "sources_selected": [], "evidence_extracted": 1},
    }
    return t


def _executor(tree):
    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=tree.round)
    # The executor loads its own tree from disk — seed OUTWARD there too so
    # gate source-ID resolution has the same ground truth.
    ex.tree.state["artifacts"]["OUTWARD"] = tree.state["artifacts"].get("OUTWARD", {})
    return ex


def test_valid_applicability_artifact_passes_gate(tree):
    ex = _executor(tree)
    artifact = ex._normalize_applicability_artifact(
        {
            "route": "STANDARD_SOLUTION_SUFFICIENT",
            "candidates": [
                {
                    "id": "M1",
                    "source_ids": ["O1"],
                    "classification": "DIRECTLY_APPLICABLE",
                    "integration_point": "memory_store.recall",
                    "benefit": "x",
                    "cost": "low",
                    "architectural_fit": "ok",
                    "authority_implications": "none",
                    "risks": [],
                    "required_adaptation": [],
                    "evidence": ["E1"],
                }
            ],
        },
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    assert ex._validate_applicability(artifact) == []
    assert artifact["stage"] == "APPLICABILITY"


def test_invalid_route_fails_gate(tree):
    ex = _executor(tree)
    artifact = ex._normalize_applicability_artifact(
        {"route": "MAYBE", "candidates": []}, {"receipts": [], "event_counts": {}, "elapsed": 0.0}, {"attempt": 1}
    )
    errors = ex._validate_applicability(artifact)
    assert any("invalid route" in e for e in errors)


def test_unknown_source_citation_fails_gate(tree):
    ex = _executor(tree)
    artifact = ex._normalize_applicability_artifact(
        {
            "route": "STANDARD_SOLUTION_SUFFICIENT",
            "candidates": [
                {
                    "id": "M1",
                    "source_ids": ["O99"],
                    "classification": "DIRECTLY_APPLICABLE",
                    "integration_point": "x",
                    "evidence": ["E1"],
                }
            ],
        },
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    errors = ex._validate_applicability(artifact)
    assert any("unknown source" in e for e in errors)


def test_viable_candidate_needs_integration_point_and_evidence(tree):
    ex = _executor(tree)
    artifact = ex._normalize_applicability_artifact(
        {
            "route": "INNOVATION_WARRANTED",
            "candidates": [{"id": "M1", "source_ids": ["O1"], "classification": "APPLICABLE_WITH_ADAPTATION"}],
        },
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    errors = ex._validate_applicability(artifact)
    assert any("no integration_point" in e for e in errors)
    assert any("no evidence ids" in e for e in errors)


def test_route_decision_drives_advance(tree):
    """Tree-level: APPLICABILITY advance with INNOVATION_WARRANTED goes to INNOVATION;
    STANDARD_SOLUTION_SUFFICIENT goes to ADVERSARIAL (skip)."""
    tree.advance({"candidates": [], "route": "INNOVATION_WARRANTED"})
    assert tree.current_stage == "INNOVATION"
    t2 = rtt.ResearchTaskTree(issue=971, round_no=1, artifact_dir=str(tree.artifact_dir) + "-2")
    for stage in ("ISSUE", "INWARD", "OUTWARD"):
        t2.advance({"test": True})
    t2.advance({"candidates": [], "route": "STANDARD_SOLUTION_SUFFICIENT"})
    assert t2.current_stage == "ADVERSARIAL"
    assert "INNOVATION" not in t2.state["completed_stages"]


def test_applicability_requires_outward_artifact(tree, mock_gitea=None):
    """Stage-ordering guard: no OUTWARD artifact → STAGE_FAILED, no MiMo call."""
    ex = _executor(tree)
    ex.tree.state["artifacts"]["OUTWARD"] = {}
    result = ex._execute_applicability({"attempt": 1})
    assert result["status"] == "STAGE_FAILED"
    assert "OUTWARD artifact missing" in result["error"]