rune/tests/test_synthesis_executor.py

"""Tests: RP7 SYNTHESIS executor + gate (#270).

Gate semantics: concrete build instructions only; adversarial constraints
(mitigations, kill-tests) carried into the design; rejected candidates
cannot be resurrected; readiness invariant enforced before the MiMo call.
"""

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 _make_tree(tmp_path, issue=985):
    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_WITH_MITIGATIONS",
            "surviving_candidates": ["M1"],
            "rejected_candidates": [{"id": "BAD1", "reason": "duplicate authority"}],
            "required_mitigations": ["mode labeling", "cache controls"],
            "falsification_tests": ["FT1: mode switch must be labeled"],
            "attacks": [{"id": "A1", "target_candidate": "M1", "dimension": "failure modes", "finding": "x"}],
        }
    )
    assert t.current_stage == "SYNTHESIS"
    return t


def _executor(tree):
    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=tree.round)
    ex.tree.state["artifacts"]["ADVERSARIAL"] = tree.state["artifacts"].get("ADVERSARIAL", {})
    return ex


GOOD_SYNTHESIS = {
    "selected_design": "Measurement harness with mode labeling and cache controls",
    "why_this_design_won": "survivors align",
    "rejected_alternatives": [],
    "components_affected": ["src/rune/orchestration/benchmark.py"],
    "authority_boundaries": ["harness reads memory_selector, never writes"],
    "interfaces": ["def run_suite(corpus: list[dict]) -> dict"],
    "data_flow": ["corpus -> harness -> recall -> metrics"],
    "implementation_sequence": ["1. corpus fixture", "2. metric suite", "3. CI job"],
    "invariants": [
        "FT1: mode switch must be labeled",
        "mitigation: mode labeling — every result row records retrieval mode",
        "mitigation: cache controls — result/embedding caches bypassed per measurement",
    ],
    "tests_required": ["test_mode_labeling", "test_cache_bypass"],
    "acceptance_criteria": ["Recall@k reported", "p50/p95 latency reported"],
    "failure_modes": ["embedding backend down"],
    "rollback": "harness is additive; delete file to revert",
    "deliberately_excluded": ["qdrant production migration"],
}


def test_good_synthesis_passes_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    artifact = ex._normalize_synthesis_artifact(
        GOOD_SYNTHESIS, {"receipts": [], "event_counts": {}, "elapsed": 0.0}, {"attempt": 1}
    )
    assert ex._validate_synthesis(artifact) == []
    # adversarial constraints carried as provenance
    assert artifact["carried_mitigations"] == ["mode labeling", "cache controls"]
    assert artifact["carried_falsification_tests"] == ["FT1: mode switch must be labeled"]
    assert artifact["adversarial_disposition"] == "SURVIVES_WITH_MITIGATIONS"


def test_missing_rollback_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    bad = {**GOOD_SYNTHESIS, "rollback": ""}
    artifact = ex._normalize_synthesis_artifact(
        bad, {"receipts": [], "event_counts": {}, "elapsed": 0.0}, {"attempt": 1}
    )
    assert any("rollback" in e for e in ex._validate_synthesis(artifact))


def test_unconverted_falsification_test_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    bad = {**GOOD_SYNTHESIS, "invariants": ["cache bypassed per measurement"]}
    artifact = ex._normalize_synthesis_artifact(
        bad, {"receipts": [], "event_counts": {}, "elapsed": 0.0}, {"attempt": 1}
    )
    errors = ex._validate_synthesis(artifact)
    assert any("FT1" in e and "not converted" in e for e in errors)


def test_dropped_mitigation_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    # Remove ALL structural coverage of the cache-controls mitigation
    bad = {
        **GOOD_SYNTHESIS,
        "invariants": ["FT1: mode switch must be labeled"],
        "interfaces": ["def run_suite(corpus: list[dict]) -> dict"],
        "data_flow": ["corpus -> harness -> recall -> metrics"],
        "tests_required": ["test_mode_labeling"],
        "acceptance_criteria": ["Recall@k reported", "p50/p95 latency reported"],
    }
    artifact = ex._normalize_synthesis_artifact(
        bad, {"receipts": [], "event_counts": {}, "elapsed": 0.0}, {"attempt": 1}
    )
    errors = ex._validate_synthesis(artifact)
    assert any("cache controls" in e and "insufficiently represented" in e for e in errors)


def test_rejected_candidate_resurrection_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    bad = {**GOOD_SYNTHESIS, "components_affected": ["src/rune/orchestration/benchmark.py", "BAD1_adapter.py"]}
    artifact = ex._normalize_synthesis_artifact(
        bad, {"receipts": [], "event_counts": {}, "elapsed": 0.0}, {"attempt": 1}
    )
    errors = ex._validate_synthesis(artifact)
    assert any("BAD1" in e and "resurrection" in e for e in errors)


def test_not_concrete_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    bad = {**GOOD_SYNTHESIS, "interfaces": []}
    artifact = ex._normalize_synthesis_artifact(
        bad, {"receipts": [], "event_counts": {}, "elapsed": 0.0}, {"attempt": 1}
    )
    assert any("interfaces" in e for e in ex._validate_synthesis(artifact))


def test_readiness_invariant_blocks_synthesis_execution(tmp_path):
    """A stage execution with a broken ADVERSARIAL artifact fails BEFORE any MiMo call."""
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    ex.tree.state["artifacts"]["ADVERSARIAL"] = {"disposition": "EVIDENCE_DEFICIENT", "surviving_candidates": []}
    result = ex._execute_synthesis({"attempt": 1})
    assert result["status"] == "STAGE_FAILED"
    assert "readiness invariant" in result["error"]


def test_advance_after_synthesis_goes_to_qualify(tmp_path):
    tree = _make_tree(tmp_path)
    tree.advance(GOOD_SYNTHESIS)
    assert tree.current_stage == "QUALIFY"