Open_Ultra_SAST/tests/test_learning_proposer.py

"""learning-harness task 3.5: failure facts, the scripted proposer and the HarnessX proposer (offline)."""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from openultrasast.learning.detectors import FamilyConfig
from openultrasast.learning.journal import LearningJournal, RoundRecord
from openultrasast.learning.scoring import PairFamilyScore


def _config(family: str = "access_control") -> FamilyConfig:
    return FamilyConfig(
        family=family,
        version="0",
        prompt="You hunt access control.",
        checklist="- did the handler check ownership?",
        tools=("read_definition", "obligations"),
        max_steps=3,
        max_cost_usd=0.5,
        max_chars=4000,
    )


def _score(pair: str, outcome: str, *, family: str = "access_control") -> PairFamilyScore:
    return PairFamilyScore(pair=pair, family=family, runs=(outcome,), outcome=outcome, slice="vibe-py")  # type: ignore[arg-type]


def test_failure_facts_carry_the_misses_the_leaks_and_the_rejected_buffer(tmp_path: Path) -> None:
    from openultrasast.learning.proposer import build_failure_facts

    journal = LearningJournal(tmp_path / "journal.jsonl")
    journal.append(
        RoundRecord(
            round=1,
            family="access_control",
            hypothesis="ask about ownership",
            levers=("checklist",),
            predicted_affected=("access_control",),
            predicted_at_risk=(),
            outcome="rejected",
            reason="tie",
        )
    )
    scores = [_score("miss-one", "both_silent"), _score("leak-one", "both_flagged"), _score("good", "pair_correct")]
    facts = build_failure_facts(
        scores, family="access_control", config=_config(), journal=journal, train_pairs={"miss-one", "leak-one", "good"}
    )
    assert [item.pair for item in facts.misses] == ["miss-one"]
    assert [item.pair for item in facts.leaks] == ["leak-one"]
    assert facts.rejected == ("ask about ownership",)
    assert facts.config.family == "access_control" and facts.family == "access_control"


def test_the_proposer_is_never_handed_a_holdout_pair(tmp_path: Path) -> None:
    from openultrasast.learning.proposer import build_failure_facts

    journal = LearningJournal(tmp_path / "journal.jsonl")
    scores = [_score("train-miss", "both_silent"), _score("holdout-miss", "both_silent")]
    facts = build_failure_facts(scores, family="access_control", config=_config(), journal=journal, train_pairs={"train-miss"})
    assert [item.pair for item in facts.misses] == ["train-miss"]
    assert "holdout-miss" not in json.dumps(facts.to_dict())


def test_the_scripted_proposer_returns_one_change_for_one_family(tmp_path: Path) -> None:
    from openultrasast.learning.proposer import Proposal, ScriptedProposer, build_failure_facts

    journal = LearningJournal(tmp_path / "journal.jsonl")
    facts = build_failure_facts([_score("a", "both_silent")], family="access_control", config=_config(), journal=journal, train_pairs={"a"})
    scripted = ScriptedProposer(
        [
            Proposal(
                hypothesis="h",
                lever="checklist",
                change={"checklist.md": "- ask about ownership\n"},
                predicted_affected=("access_control",),
            )
        ]
    )
    proposal = scripted.propose(facts)
    assert proposal is not None and proposal.lever == "checklist" and set(proposal.change) == {"checklist.md"}
    assert scripted.propose(facts) is None  # a scripted proposer runs out rather than repeating itself


@pytest.mark.parametrize(
    ("change", "lever", "needle"),
    [
        ({"../../etc/passwd": "x"}, "checklist", "outside"),
        ({"family.toml": "x", "prompt.md": "y"}, "prompt", "one file"),
        ({"secrets.md": "x"}, "checklist", "secrets.md"),
        ({"prompt.md": "x" * 9000}, "prompt", "cap"),
    ],
    ids=["escapes_the_family_directory", "two_files", "unknown_file", "over_the_cap"],
)
def test_a_proposal_that_reaches_beyond_its_family_is_refused_by_name(change: dict[str, str], lever: str, needle: str) -> None:
    from openultrasast.learning.proposer import Proposal, refuse_proposal

    refusal = refuse_proposal(Proposal(hypothesis="h", lever=lever, change=change), _config())  # type: ignore[arg-type]
    assert refusal is not None and needle in refusal


def test_a_well_formed_proposal_is_accepted_and_applies_inside_its_own_directory(tmp_path: Path) -> None:
    from openultrasast.learning.proposer import Proposal, apply_proposal, refuse_proposal

    directory = tmp_path / "access_control"
    directory.mkdir()
    (directory / "checklist.md").write_text("- old\n")
    proposal = Proposal(hypothesis="h", lever="checklist", change={"checklist.md": "- new\n"}, predicted_affected=("access_control",))
    assert refuse_proposal(proposal, _config()) is None
    apply_proposal(proposal, directory)
    assert (directory / "checklist.md").read_text() == "- new\n"


def test_the_write_roots_are_one_family_directory_and_exclude_the_checker_and_the_corpus(tmp_path: Path) -> None:
    from openultrasast.learning import canaries
    from openultrasast.learning.proposer import write_roots

    roots = write_roots(tmp_path / "configs", "access_control")
    assert roots == (tmp_path / "configs" / "access_control",)
    verifier_module = Path(canaries.__file__).resolve()
    assert not any(verifier_module.is_relative_to(root.resolve()) for root in roots)
    corpus = Path("benchmarks").resolve()
    assert not any(str(corpus).startswith(str(root.resolve())) for root in roots)


def test_the_harnessx_proposer_degrades_with_a_reason_when_the_extra_is_absent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    from openultrasast.learning import proposer as proposer_module
    from openultrasast.learning.proposer import HarnessXProposer, build_failure_facts

    journal = LearningJournal(tmp_path / "journal.jsonl")
    facts = build_failure_facts([_score("a", "both_silent")], family="access_control", config=_config(), journal=journal, train_pairs={"a"})
    monkeypatch.setattr(proposer_module, "has_harnessx", lambda: False)
    agent = HarnessXProposer(configs_dir=tmp_path / "configs", model="m")
    assert agent.propose(facts) is None
    assert agent.reason == "harnessx_unavailable"
    assert agent.write_roots("access_control") == (tmp_path / "configs" / "access_control",)


# --- the meta-agent proposer (review round 3, task 3.5 finding 1) ------------


def test_the_meta_agent_proposer_reads_its_change_out_of_a_scratch_copy(tmp_path: Path) -> None:
    """The meta-agent never writes into the live family directory.

    A proposer that edited the tree directly would have changed the configuration before the round snapshotted
    it, so a rejection could not put it back. It gets a copy, and the proposal is the diff of that copy."""
    from openultrasast.learning.proposer import HarnessXProposer

    configs = tmp_path / "configs"
    (configs / "injection").mkdir(parents=True)
    (configs / "injection" / "checklist.md").write_text("- old\n")
    (configs / "injection" / "family.toml").write_text('family = "injection"\nversion = "0"\n')
    seen: dict[str, object] = {}

    def agent(*, workspace: Path, facts, model: str):  # type: ignore[no-untyped-def]
        seen["workspace"] = workspace
        seen["family"] = facts.family
        (workspace / "checklist.md").write_text("- ask about quoting\n")
        return "ask about quoting"

    proposer = HarnessXProposer(configs_dir=configs, model="m", agent=agent)
    proposal = proposer.propose(_facts(tmp_path, "injection"))
    assert proposal is not None
    assert proposal.lever == "checklist" and proposal.change == {"checklist.md": "- ask about quoting\n"}
    assert proposal.hypothesis == "ask about quoting"
    assert seen["workspace"] != configs / "injection"
    assert (configs / "injection" / "checklist.md").read_text() == "- old\n", "the live tree was edited"


def test_the_meta_agent_proposer_refuses_more_than_one_changed_file(tmp_path: Path) -> None:
    from openultrasast.learning.proposer import HarnessXProposer

    configs = tmp_path / "configs"
    (configs / "injection").mkdir(parents=True)
    (configs / "injection" / "checklist.md").write_text("- old\n")
    (configs / "injection" / "prompt.md").write_text("p\n")

    def agent(*, workspace: Path, facts, model: str):  # type: ignore[no-untyped-def]
        (workspace / "checklist.md").write_text("- a\n")
        (workspace / "prompt.md").write_text("- b\n")
        return "two at once"

    proposer = HarnessXProposer(configs_dir=configs, model="m", agent=agent)
    assert proposer.propose(_facts(tmp_path, "injection")) is None
    assert "one file" in proposer.reason


def test_the_meta_agent_proposer_refuses_a_file_no_lever_owns(tmp_path: Path) -> None:
    from openultrasast.learning.proposer import HarnessXProposer

    configs = tmp_path / "configs"
    (configs / "injection").mkdir(parents=True)
    (configs / "injection" / "checklist.md").write_text("- old\n")

    def agent(*, workspace: Path, facts, model: str):  # type: ignore[no-untyped-def]
        (workspace / "notes.txt").write_text("x\n")
        return "sideways"

    proposer = HarnessXProposer(configs_dir=configs, model="m", agent=agent)
    assert proposer.propose(_facts(tmp_path, "injection")) is None
    assert "notes.txt" in proposer.reason


def test_the_meta_agent_proposer_says_so_when_it_changed_nothing(tmp_path: Path) -> None:
    from openultrasast.learning.proposer import HarnessXProposer

    configs = tmp_path / "configs"
    (configs / "injection").mkdir(parents=True)
    (configs / "injection" / "checklist.md").write_text("- old\n")
    proposer = HarnessXProposer(configs_dir=configs, model="m", agent=lambda **_kwargs: "nothing to do")
    assert proposer.propose(_facts(tmp_path, "injection")) is None
    assert proposer.reason == "no_change_proposed"


def _facts(tmp_path: Path, family: str):  # type: ignore[no-untyped-def]
    from openultrasast.learning.proposer import FailureFacts

    del tmp_path
    return FailureFacts(family=family, config=_config(family))