diff --git a/src/openultrasast/learning/proposer.py b/src/openultrasast/learning/proposer.py new file mode 100644 index 0000000..bf32727 --- /dev/null +++ b/src/openultrasast/learning/proposer.py @@ -0,0 +1,213 @@ +"""What a round proposes, and who may propose it (learning-harness, Req 9.1, 9.2, 5.4, 7.6). + +A proposer is handed facts, never scores it is about to be judged on: the misses and leaks of one family +on the train split, what each run actually did, and the hypotheses this family already tried and lost. +Self-diagnosis from free narrative confabulates; structured facts are what make a proposal answerable. + +A proposal is one hypothesis, one lever, and one file inside one family's directory. The narrowness is +the safety property: a round can only touch what it targets, so every other family is frozen by +construction rather than by promise, and the verifier code and the corpus stay outside every write root. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Literal, Protocol + +from ..harness_ext import has_harnessx +from .detectors import MAX_PROMPT_CHARS, FamilyConfig +from .journal import LearningJournal +from .scoring import PairFamilyScore + +Lever = Literal["tool", "checklist", "counterexample", "memory", "prompt"] +LEVERS: tuple[Lever, ...] = ("tool", "checklist", "counterexample", "memory", "prompt") +# The only files a round may write, one per lever. Anything else is a proposal reaching beyond its family. +EDITABLE_FILES = { + "tool": "family.toml", + "checklist": "checklist.md", + "counterexample": "counterexamples.jsonl", + "memory": "hard_negatives.jsonl", + "prompt": "prompt.md", +} + + +@dataclass(frozen=True) +class PairFact: + """What one pair did, in terms a proposer can act on.""" + + pair: str + outcome: str + runs: tuple[str, ...] = () + other_findings: int = 0 + rung: str = "suspicion" + + def to_dict(self) -> dict[str, object]: + return { + "pair": self.pair, + "outcome": self.outcome, + "runs": list(self.runs), + "other_findings": self.other_findings, + "rung": self.rung, + } + + +@dataclass(frozen=True) +class FailureFacts: + family: str + config: FamilyConfig + misses: tuple[PairFact, ...] = () + leaks: tuple[PairFact, ...] = () + rejected: tuple[str, ...] = () + + def to_dict(self) -> dict[str, object]: + return { + "family": self.family, + "config_version": self.config.version, + "tools": list(self.config.tools), + "misses": [item.to_dict() for item in self.misses], + "leaks": [item.to_dict() for item in self.leaks], + "rejected": list(self.rejected), + } + + +@dataclass(frozen=True) +class Proposal: + hypothesis: str + lever: Lever + change: Mapping[str, str] # file name inside the family directory -> its new contents + predicted_affected: tuple[str, ...] = () + predicted_at_risk: tuple[str, ...] = () + + def to_dict(self) -> dict[str, object]: + return { + "hypothesis": self.hypothesis, + "lever": self.lever, + "change": dict(self.change), + "predicted_affected": list(self.predicted_affected), + "predicted_at_risk": list(self.predicted_at_risk), + } + + +class Proposer(Protocol): + def propose(self, facts: FailureFacts) -> Proposal | None: ... + + +def build_failure_facts( + scores: Sequence[PairFamilyScore], + *, + family: str, + config: FamilyConfig, + journal: LearningJournal, + train_pairs: set[str], +) -> FailureFacts: + """The facts of one family's failures on the train split. A holdout pair never appears here (Req 5.4).""" + mine = [item for item in scores if item.family == family and item.pair in train_pairs and item.outcome != "unscorable"] + return FailureFacts( + family=family, + config=config, + misses=tuple(_fact(item, item.other_findings_vuln) for item in mine if item.outcome in {"both_silent", "reversed"}), + leaks=tuple(_fact(item, item.other_findings_fixed) for item in mine if item.outcome in {"both_flagged", "reversed"}), + rejected=journal.rejected_buffer(family), + ) + + +def refuse_proposal(proposal: Proposal, config: FamilyConfig) -> str | None: + """None when the proposal is one legal change to its own family; otherwise why it is refused.""" + if proposal.lever not in LEVERS: + return f"unknown lever {proposal.lever!r}; expected one of {', '.join(LEVERS)}" + if len(proposal.change) != 1: + return f"a round changes one file, not {len(proposal.change)}" + name, content = next(iter(proposal.change.items())) + if "/" in name or "\\" in name or name.startswith("."): + return f"{name!r} reaches outside the family directory" + if name != EDITABLE_FILES[proposal.lever]: + return f"{name!r} is not the file the {proposal.lever!r} lever writes ({EDITABLE_FILES[proposal.lever]})" + if name == "prompt.md" and len(content) + len(config.checklist) > MAX_PROMPT_CHARS: + return f"the proposed prompt exceeds the cap of {MAX_PROMPT_CHARS} characters" + if name == "checklist.md" and len(content) + len(config.prompt) > MAX_PROMPT_CHARS: + return f"the proposed checklist exceeds the cap of {MAX_PROMPT_CHARS} characters" + return None + + +def apply_proposal(proposal: Proposal, directory: Path) -> None: + """Write the one file, inside the one directory. Callers snapshot first so a rejection can revert.""" + directory.mkdir(parents=True, exist_ok=True) + for name, content in proposal.change.items(): + (directory / name).write_text(content, encoding="utf-8") + + +def write_roots(configs_dir: Path, family: str) -> tuple[Path, ...]: + """Everything a round may write: exactly one family's directory, and never the verifiers or the corpus.""" + return (configs_dir / family,) + + +@dataclass +class ScriptedProposer: + """A fixed sequence of proposals, for tests and for the first offline rounds.""" + + proposals: list[Proposal] = field(default_factory=list) + used: int = 0 + + def propose(self, facts: FailureFacts) -> Proposal | None: + del facts + if self.used >= len(self.proposals): + return None + self.used += 1 + return self.proposals[self.used - 1] + + +@dataclass +class HarnessXProposer: + """The meta-agent, with write access limited to the family directory it is targeting.""" + + configs_dir: Path + model: str + reason: str = "" + + def write_roots(self, family: str) -> tuple[Path, ...]: + return write_roots(self.configs_dir, family) + + def propose(self, facts: FailureFacts) -> Proposal | None: + if not has_harnessx(): + self.reason = "harnessx_unavailable" + return None + try: + from ..harness_ext import require_harnessx + + require_harnessx() + except Exception: # noqa: BLE001 — a proposer that cannot start proposes nothing and says so + self.reason = "harnessx_unavailable" + return None + self.reason = "harnessx_proposer_not_wired" # the evolve call lands with the live rounds (task 5.3) + del facts + return None + + +def _fact(score: PairFamilyScore, other: int) -> PairFact: + return PairFact(pair=score.pair, outcome=score.outcome, runs=score.runs, other_findings=other, rung=score.rung) + + +def facts_prompt(facts: FailureFacts) -> str: + """The facts as the proposer sees them: structured, and free of anything it will be judged on.""" + return json.dumps(facts.to_dict(), indent=2, sort_keys=True) + + +__all__ = [ + "EDITABLE_FILES", + "LEVERS", + "FailureFacts", + "HarnessXProposer", + "Lever", + "PairFact", + "Proposal", + "Proposer", + "ScriptedProposer", + "apply_proposal", + "build_failure_facts", + "facts_prompt", + "refuse_proposal", + "write_roots", +] diff --git a/tests/test_learning_proposer.py b/tests/test_learning_proposer.py new file mode 100644 index 0000000..add50c8 --- /dev/null +++ b/tests/test_learning_proposer.py @@ -0,0 +1,139 @@ +"""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",)