rune/tests/test_adversarial_executor.py

"""Tests: RP6 ADVERSARIAL executor + gate (#269).

Gate semantics: a real attack (every dimension addressed, every leading
candidate targeted), coherent survivors (no unrebutted fatal attacks),
no rubber-stamping.
"""

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=980):
    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",
                    "integration_point": "memory_store.recall",
                    "evidence": ["E1"],
                }
            ],
            "route": "STANDARD_SOLUTION_SUFFICIENT",
        }
    )
    assert t.current_stage == "ADVERSARIAL"
    return t


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


GOOD_ATTACKS = [
    {
        "id": "A1",
        "target_candidate": "M1",
        "dimension": "contradicting evidence",
        "finding": "vendor benchmarks conflict",
        "evidence": ["E1"],
        "severity": "minor",
    },
    {
        "id": "A2",
        "target_candidate": "M1",
        "dimension": "weak assumptions",
        "finding": "assumes stable corpus",
        "evidence": [],
        "severity": "moderate",
    },
    {
        "id": "A3",
        "target_candidate": "M1",
        "dimension": "failure modes",
        "finding": "RRF degrades on empty FTS results",
        "evidence": [],
        "severity": "moderate",
    },
    {
        "id": "A4",
        "target_candidate": "M1",
        "dimension": "simpler alternative",
        "finding": "plain FTS5 may suffice",
        "evidence": [],
        "severity": "minor",
    },
    {
        "id": "A5",
        "target_candidate": "M1",
        "dimension": "duplicate authority",
        "finding": "harness could become second truth",
        "evidence": [],
        "severity": "moderate",
    },
    {
        "id": "A6",
        "target_candidate": "M1",
        "dimension": "Rune constraint violations",
        "finding": "none found — projection rule respected",
        "evidence": [],
        "severity": "none",
    },
    {
        "id": "A7",
        "target_candidate": "M1",
        "dimension": "falsification",
        "finding": "harness showing <5% delta falsifies",
        "evidence": [],
        "severity": "major",
    },
]


def test_full_attack_passes_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    artifact = ex._normalize_adversarial_artifact(
        {
            "surviving_candidates": ["M1"],
            "rejected_candidates": [],
            "attacks": GOOD_ATTACKS,
            "critical_assumptions": ["stable corpus"],
            "required_mitigations": ["seeded corpus"],
            "new_research_gaps": [],
            "falsification_tests": ["<5% delta test"],
        },
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    assert ex._validate_adversarial(artifact) == []


def test_rubber_stamp_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    artifact = ex._normalize_adversarial_artifact(
        {"surviving_candidates": ["M1"], "attacks": []},
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    errors = ex._validate_adversarial(artifact)
    assert any("rubber-stamp" in e for e in errors)
    assert any("dimensions not addressed" in e for e in errors)


def test_unattacked_leading_candidate_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    tree.state["artifacts"]["APPLICABILITY"]["candidates"].append(
        {"id": "M2", "classification": "APPLICABLE_WITH_ADAPTATION", "integration_point": "x", "evidence": ["E1"]}
    )
    artifact = ex._normalize_adversarial_artifact(
        {"surviving_candidates": ["M1", "M2"], "attacks": GOOD_ATTACKS},  # only M1 attacked
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    errors = ex._validate_adversarial(artifact)
    assert any("never attacked" in e and "M2" in e for e in errors)


def test_unrebutted_fatal_on_survivor_fails_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    attacks = GOOD_ATTACKS + [
        {
            "id": "A8",
            "target_candidate": "M1",
            "dimension": "failure modes",
            "finding": "corrupts memory on crash",
            "evidence": ["E2"],
            "severity": "fatal",
        }
    ]
    artifact = ex._normalize_adversarial_artifact(
        {"surviving_candidates": ["M1"], "attacks": attacks},
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    errors = ex._validate_adversarial(artifact)
    assert any("fatal" in e and "without rebuttal" in e for e in errors)


def test_rebutted_fatal_on_survivor_passes_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    attacks = GOOD_ATTACKS + [
        {
            "id": "A8",
            "target_candidate": "M1",
            "dimension": "failure modes",
            "finding": "corrupts memory on crash",
            "evidence": ["E2"],
            "severity": "fatal",
            "rebuttal": "write path is append-only; crash loses at most the current batch",
        }
    ]
    artifact = ex._normalize_adversarial_artifact(
        {"surviving_candidates": ["M1"], "attacks": attacks},
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    assert ex._validate_adversarial(artifact) == []


def test_missing_fields_in_attack_fail_gate(tmp_path):
    tree = _make_tree(tmp_path)
    ex = _executor(tree)
    artifact = ex._normalize_adversarial_artifact(
        {
            "surviving_candidates": [],
            "rejected_candidates": [{"id": "M1", "reason": "broke"}],
            "attacks": [{"id": "A1", "target_candidate": "M1", "dimension": "falsification"}],
        },
        {"receipts": [], "event_counts": {}, "elapsed": 0.0},
        {"attempt": 1},
    )
    errors = ex._validate_adversarial(artifact)
    assert any("missing finding" in e for e in errors)
    assert any("dimensions not addressed" in e for e in errors)


def test_advance_after_adversarial_goes_to_synthesis(tmp_path):
    tree = _make_tree(tmp_path)
    tree.advance(
        {
            "disposition": "SURVIVES",
            "surviving_candidates": ["M1"],
            "required_mitigations": [],
            "falsification_tests": [],
            "attacks": GOOD_ATTACKS,
        }
    )
    assert tree.current_stage == "SYNTHESIS"