diff --git a/src/openultrasast/learning/rounds.py b/src/openultrasast/learning/rounds.py index 2272c62..59e2a59 100644 --- a/src/openultrasast/learning/rounds.py +++ b/src/openultrasast/learning/rounds.py @@ -24,8 +24,10 @@ from pathlib import Path from ..findings import StaticFinding from ..pairs import PairCase from .classify import classify_pair -from .detectors import write_default_configs +from .detectors import FamilyConfig, write_default_configs from .families import FamilyTaxonomy +from .journal import Archive, Attribution, LearningJournal, RoundRecord +from .proposer import Proposal, Proposer from .scoring import FamilyMetrics, PairFamilyScore, aggregate, score_pair_family DEFAULT_SLICES = ("vibe-py", "agent-vfc") # the web families, where real-world code and dangerous fruit coincide @@ -213,8 +215,267 @@ def load_noise_floors(out_dir: Path, model: str) -> dict[str, NoiseFloor]: return floors +MIN_K_RUNS = 3 # Req 3.5: a single run misses most real changes, so no stage may score with fewer + + +def run_learning_round( + cases: Sequence[PairCase], + *, + family: str, + taxonomy: FamilyTaxonomy, + configs_dir: Path, + proposer: Proposer, + scan_factory: Callable[[FamilyConfig], Scan], + model: str, + journal: LearningJournal, + archive: Archive, + floors: Mapping[str, NoiseFloor], + out_dir: Path, + cost_cap_usd: float = 10.0, + minibatch: int = 8, + k_runs: int = MIN_K_RUNS, + sweep_families: Sequence[str] = (), + spent_usd: Callable[[], float] | None = None, + trajectories: Callable[[], Sequence[Mapping[str, object]]] | None = None, +) -> object: + """One round: propose one change for one family, then let the evidence decide whether it stays. + + Staged so that the cheap answer comes first: a train minibatch, then the family's held-out pairs, + then every other family. Each stage scores several runs per pair, because a single run misses most + real changes. The proposer never sees a held-out pair, and the family directory is snapshotted before + the change so any outcome other than acceptance leaves the tree exactly as it was found. + """ + from ..redaction import redact_secrets + from .acceptance import DirectorySnapshot, bump_version, decide, within_budget + from .detectors import load_family_configs + from .proposer import apply_proposal, build_failure_facts, refuse_proposal + from .split import is_teacher + + if k_runs < MIN_K_RUNS: + raise ValueError(f"a stage scores at least three runs per pair; {k_runs} was asked for") + round_number = journal.next_round() # type: ignore[attr-defined] + directory = out_dir / "rounds" / str(round_number) + if directory.exists(): + from .journal import JournalError + + raise JournalError(f"round {round_number} already has a directory; history is append-only") + configs = load_family_configs(configs_dir, taxonomy) + config = configs[family] + train = [case for case in cases if is_teacher(case) and _family_of(case, taxonomy) == family] + holdout = [case for case in cases if case.split == "holdout" and _family_of(case, taxonomy) == family] + batch = train[:minibatch] + + def _score(rows: Sequence[PairCase], scan: Scan, name: str) -> list[PairFamilyScore]: + return [score_case(case, scan, taxonomy=taxonomy, runs=k_runs, family=name) for case in rows] + + before_scan = scan_factory(config) + before_train = _score(batch, before_scan, family) + before_hold = _score(holdout, before_scan, family) + facts = build_failure_facts(before_train, family=family, config=config, journal=journal, train_pairs={case.name for case in train}) + proposal = proposer.propose(facts) # type: ignore[attr-defined] + record = _round_record(round_number, family, proposal, taxonomy, config) + if proposal is None: + return _finish(journal, directory, record, outcome="rejected", reason="no_proposal", trajectories=trajectories) + refusal = refuse_proposal(proposal, config) + if refusal is not None: + return _finish(journal, directory, record, outcome="rejected", reason=refusal, proposal=proposal, trajectories=trajectories) + snapshot = DirectorySnapshot.of(configs_dir / family) + apply_proposal(proposal, configs_dir / family) + try: + after_config = load_family_configs(configs_dir, taxonomy)[family] + after_scan = scan_factory(after_config) + after_train = _score(batch, after_scan, family) + after_hold = _score(holdout, after_scan, family) + sweep, flipped_unpredicted = _sweep(cases, taxonomy, configs, scan_factory, sweep_families, family, k_runs) + cost = spent_usd() if spent_usd is not None else 0.0 + train_delta = _delta(before_train, after_train) + holdout_delta = _delta(before_hold, after_hold) + if not within_budget(cost, cost_cap_usd): + snapshot.restore() + return _finish( + journal, + directory, + record, + outcome="reverted_cost", + reason="cost_cap_exceeded", + proposal=proposal, + cost=cost, + scores=after_train + after_hold, + trajectories=trajectories, + ) + verdict = decide(target=family, train_delta=train_delta, holdout_delta=holdout_delta, sweep=sweep, floors=floors) # type: ignore[arg-type] + flipped = _flipped(before_train + before_hold, after_train + after_hold) + attribution = Attribution( + flipped_predicted=len(flipped), + flipped_unpredicted=len(flipped_unpredicted), + precision=len(flipped) / (len(flipped) + len(flipped_unpredicted)) if flipped or flipped_unpredicted else 0.0, + ) + if not verdict.accepted: + snapshot.restore() + return _finish( + journal, + directory, + record, + outcome="rejected", + reason=verdict.reason, + proposal=proposal, + cost=cost, + train_delta=train_delta, + holdout_delta=holdout_delta, + sweep=sweep, + attribution=attribution, + flipped=flipped, + unpredicted=flipped_unpredicted, + scores=after_train + after_hold, + trajectories=trajectories, + ) + version = str(int(config.version) + 1) if config.version.isdigit() else f"{config.version}+1" + bump_version(configs_dir / family, version) + for score in after_train + after_hold: + archive.record(family, version, score) # type: ignore[attr-defined] + return _finish( + journal, + directory, + record, + outcome="accepted", + reason=verdict.reason, + proposal=proposal, + cost=cost, + train_delta=train_delta, + holdout_delta=holdout_delta, + sweep=sweep, + attribution=attribution, + flipped=flipped, + unpredicted=flipped_unpredicted, + scores=after_train + after_hold, + version=version, + trajectories=trajectories, + ) + except Exception: + snapshot.restore() + raise + finally: + del redact_secrets + + +def _sweep( + cases: Sequence[PairCase], + taxonomy: FamilyTaxonomy, + configs: Mapping[str, FamilyConfig], + scan_factory: Callable[[FamilyConfig], Scan], + families: Sequence[str], + target: str, + k_runs: int, +) -> tuple[dict[str, int], list[str]]: + """Every other family's held-out pairs, so a change cannot help one family by quietly breaking another.""" + flips: dict[str, int] = {} + unpredicted: list[str] = [] + for name in families: + if name == target or name not in configs: + continue + rows = [case for case in cases if case.split == "holdout" and _family_of(case, taxonomy) == name] + scan = scan_factory(configs[name]) + broken = [ + item + for item in (score_case(case, scan, taxonomy=taxonomy, runs=k_runs, family=name) for case in rows) + if item.outcome != "pair_correct" + ] + flips[name] = len(broken) + unpredicted.extend(item.pair for item in broken) + return flips, unpredicted + + +def _delta(before: Sequence[PairFamilyScore], after: Sequence[PairFamilyScore]) -> float: + """The change in how many pairs came out right, as a fraction of the pairs that could be scored.""" + scorable = [item for item in before if item.outcome != "unscorable"] + if not scorable: + return 0.0 + was = sum(1 for item in before if item.outcome == "pair_correct") + now = sum(1 for item in after if item.outcome == "pair_correct") + return (now - was) / len(scorable) + + +def _flipped(before: Sequence[PairFamilyScore], after: Sequence[PairFamilyScore]) -> list[str]: + was = {item.pair: item.outcome for item in before} + return [item.pair for item in after if item.outcome == "pair_correct" and was.get(item.pair) not in {"pair_correct", None}] + + +def _round_record( + round_number: int, family: str, proposal: Proposal | None, taxonomy: FamilyTaxonomy, config: FamilyConfig +) -> dict[str, object]: + return { + "round": round_number, + "family": family, + "hypothesis": proposal.hypothesis if proposal is not None else "", + "levers": (proposal.lever,) if proposal is not None else (), + "predicted_affected": tuple(proposal.predicted_affected) if proposal is not None else (), + "predicted_at_risk": tuple(proposal.predicted_at_risk) if proposal is not None else (), + "taxonomy_version": taxonomy.version, + "config_version": config.version, + } + + +def _finish( + journal: LearningJournal, + directory: Path, + base: dict[str, object], + *, + outcome: str, + reason: str, + proposal: Proposal | None = None, + cost: float = 0.0, + train_delta: float = 0.0, + holdout_delta: float = 0.0, + sweep: Mapping[str, int] | None = None, + attribution: Attribution | None = None, + flipped: Sequence[str] = (), + unpredicted: Sequence[str] = (), + scores: Sequence[PairFamilyScore] = (), + version: str | None = None, + trajectories: Callable[[], Sequence[Mapping[str, object]]] | None = None, +) -> object: + """Write the round directory and journal the outcome. Every outcome is journalled, not just the good ones.""" + from ..redaction import redact_secrets + + record = RoundRecord( + **base, # type: ignore[arg-type] + outcome=outcome, + reason=reason, + target_train_delta=train_delta, + target_holdout_delta=holdout_delta, + sweep=dict(sweep or {}), + attribution=attribution or Attribution(), + cost_usd=cost, + ) + if version is not None: + record = RoundRecord(**{**record.__dict__, "config_version": version}) + directory.mkdir(parents=True, exist_ok=True) + if proposal is not None: + (directory / "proposal.json").write_text( + redact_secrets(json.dumps(proposal.to_dict(), indent=2, sort_keys=True)) + "\n", + encoding="utf-8", + ) + (directory / "scores.json").write_text( + json.dumps([{"pair": item.pair, "family": item.family, "outcome": item.outcome} for item in scores], indent=2, sort_keys=True) + + "\n", + encoding="utf-8", + ) + (directory / "attribution.json").write_text( + json.dumps({"flipped_predicted": list(flipped), "flipped_unpredicted": list(unpredicted)}, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + rows = list(trajectories() if trajectories is not None else ()) + (directory / "trajectories.jsonl").write_text( + "".join(redact_secrets(json.dumps(row, sort_keys=True)) + "\n" for row in rows), encoding="utf-8" + ) + journal.append(record) # type: ignore[attr-defined] + return record + + __all__ = [ "DEFAULT_K_RUNS", + "MIN_K_RUNS", + "run_learning_round", "DEFAULT_SLICES", "NON_GATING_FAMILIES", "BaselineReport", diff --git a/tests/test_learning_round_loop.py b/tests/test_learning_round_loop.py new file mode 100644 index 0000000..bef601c --- /dev/null +++ b/tests/test_learning_round_loop.py @@ -0,0 +1,223 @@ +"""learning-harness task 3.7: one evolve round, end to end (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.findings import StaticFinding +from openultrasast.learning.detectors import FamilyConfig, write_default_configs +from openultrasast.learning.families import load_families +from openultrasast.learning.journal import Archive, LearningJournal +from openultrasast.learning.proposer import Proposal, ScriptedProposer +from openultrasast.learning.rounds import NoiseFloor +from openultrasast.pairs import PairCase + +VULN = "from flask import request\nimport os\n\n\ndef run():\n os.system(request.args.get('cmd'))\n" +FIXED = "from flask import request\nimport subprocess\n\n\ndef run():\n subprocess.run(['echo', 'x'], check=False)\n" + + +def _case(tmp_path: Path, name: str, *, split: str, family: str = "injection") -> PairCase: + (tmp_path / f"{name}-v.py").write_text(VULN) + (tmp_path / f"{name}-f.py").write_text(FIXED) + return PairCase( + name=name, + slice="vibe-py", + language="python", + origin="test", + vuln_file=tmp_path / f"{name}-v.py", + fixed_file=tmp_path / f"{name}-f.py", + relpath="app.py", + expected=( + ExpectedFinding( + cwe="CWE-78", vulnerability_class="x", path="app.py", evidence="", function="run", mechanism="other", family=family + ), + ), + min_recall=1.0, + fix_policy="silent", + split=split, + review_tier="seeded", + ) + + +def _finding(family: str) -> StaticFinding: + return StaticFinding( + finding_id="detector:app.py:6", + path="app.py", + title="t", + severity="high", + confidence="low", + evidence_level="suspicion", + rationale="r", + line=6, + function_name="run", + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=[f"family:{family}", "detector:x@0"], + ranking_priority=0.0, + ) + + +def _factory(finds: dict[str, bool]): # type: ignore[no-untyped-def] + """A detector whose behaviour depends on the configuration version it was built from.""" + + def build(config: FamilyConfig): # type: ignore[no-untyped-def] + def scan(root: Path) -> list[StaticFinding]: + if root.name != "vuln": + return [] + return [_finding(config.family)] if finds.get(config.checklist.strip(), False) else [] + + return scan + + return build + + +def _proposal(text: str) -> Proposal: + return Proposal(hypothesis="ask about quoting", lever="checklist", change={"checklist.md": text}, predicted_affected=("injection",)) + + +@pytest.fixture +def world(tmp_path: Path): # type: ignore[no-untyped-def] + configs = tmp_path / "configs" + write_default_configs(configs, load_families(), prompt="p", version="0") + (configs / "injection" / "checklist.md").write_text("- old\n") + cases = [_case(tmp_path, "train-a", split="train"), _case(tmp_path, "hold-a", split="holdout")] + return { + "configs": configs, + "cases": cases, + "journal": LearningJournal(tmp_path / "journal.jsonl"), + "archive": Archive(tmp_path / "archive.jsonl"), + "out": tmp_path / "out", + "floors": {"injection": NoiseFloor(family="injection", k_runs=5, pairs=1, flaky_pairs=0, negative_flip_rate=0.0, budget_flips=0)}, + } + + +def _run(world, proposal: Proposal | None, finds: dict[str, bool], **kwargs): # type: ignore[no-untyped-def] + from openultrasast.learning.rounds import run_learning_round + + return run_learning_round( + world["cases"], + family="injection", + taxonomy=load_families(), + configs_dir=world["configs"], + proposer=ScriptedProposer([proposal] if proposal else []), + scan_factory=_factory(finds), + model="m", + journal=world["journal"], + archive=world["archive"], + floors=world["floors"], + out_dir=world["out"], + **kwargs, + ) + + +def test_a_change_that_helps_is_accepted_versioned_and_journalled(world) -> None: # type: ignore[no-untyped-def] + record = _run(world, _proposal("- ask about quoting\n"), {"- ask about quoting": True}) + assert record.outcome == "accepted" and record.family == "injection" + assert record.target_train_delta > 0 and record.target_holdout_delta > 0 + assert (world["configs"] / "injection" / "checklist.md").read_text() == "- ask about quoting\n" + assert 'version = "1"' in (world["configs"] / "injection" / "family.toml").read_text() + assert [item.outcome for item in world["journal"].rounds()] == ["accepted"] + assert world["archive"].winners("injection") + + +def test_a_change_that_helps_nothing_is_rejected_and_reverted(world) -> None: # type: ignore[no-untyped-def] + record = _run(world, _proposal("- useless\n"), {}) + assert record.outcome == "rejected" and record.reason == "tie" + assert (world["configs"] / "injection" / "checklist.md").read_text() == "- old\n" + assert 'version = "0"' in (world["configs"] / "injection" / "family.toml").read_text() + assert world["journal"].rejected_buffer("injection") == ("ask about quoting",) + + +def test_a_round_is_rejected_when_the_sweep_finds_another_family_below_its_floor(world) -> None: # type: ignore[no-untyped-def] + """The sweep is a floor check over every other family, not a causal claim about this change. + + Per-family directories make a change that reaches another family structurally impossible, so what the + sweep really guards is a family that is failing for any reason at the moment this round would land. + """ + world["cases"].append(_case(world["configs"].parent, "other-hold", split="holdout", family="access_control")) + world["floors"]["access_control"] = NoiseFloor( + family="access_control", k_runs=5, pairs=1, flaky_pairs=0, negative_flip_rate=0.0, budget_flips=0 + ) + + def build(config: FamilyConfig): # type: ignore[no-untyped-def] + def scan(root: Path) -> list[StaticFinding]: + if root.name != "vuln" or config.family == "access_control": + return [] # access control finds nothing, so its holdout pair is below its floor + return [_finding("injection")] if config.checklist.strip() == "- ask about quoting" else [] + + return scan + + from openultrasast.learning.rounds import run_learning_round + + record = run_learning_round( + world["cases"], + family="injection", + taxonomy=load_families(), + configs_dir=world["configs"], + proposer=ScriptedProposer([_proposal("- ask about quoting\n")]), + scan_factory=build, + model="m", + journal=world["journal"], + archive=world["archive"], + floors=world["floors"], + out_dir=world["out"], + sweep_families=("access_control",), + ) + assert record.outcome == "rejected" and record.reason == "collateral_regression" + assert record.sweep["access_control"] == 1 + assert (world["configs"] / "injection" / "checklist.md").read_text() == "- old\n" + + +def test_a_round_over_its_cost_cap_is_reverted_and_says_so(world) -> None: # type: ignore[no-untyped-def] + record = _run(world, _proposal("- ask about quoting\n"), {"- ask about quoting": True}, cost_cap_usd=0.0, spent_usd=lambda: 1.0) + assert record.outcome == "reverted_cost" and record.cost_usd == 1.0 + assert (world["configs"] / "injection" / "checklist.md").read_text() == "- old\n" + + +def test_a_proposal_that_reaches_beyond_its_family_never_runs(world) -> None: # type: ignore[no-untyped-def] + record = _run(world, Proposal(hypothesis="h", lever="checklist", change={"../escape.md": "x"}), {}) + assert record.outcome == "refused_holdout" or record.outcome == "rejected" + assert "outside" in record.reason + assert not (world["configs"].parent / "escape.md").exists() + + +def test_a_round_with_no_proposal_is_recorded_rather_than_skipped(world) -> None: # type: ignore[no-untyped-def] + record = _run(world, None, {}) + assert record.outcome == "rejected" and record.reason == "no_proposal" + assert [item.round for item in world["journal"].rounds()] == [1] + + +def test_the_round_directory_holds_what_a_replay_needs(world) -> None: # type: ignore[no-untyped-def] + record = _run( + world, + _proposal("- ask about quoting\n"), + {"- ask about quoting": True}, + trajectories=lambda: [{"pair": "train-a", "prompt": "api_key = " + "'" + "B" * 24 + "'"}], + ) + directory = world["out"] / "rounds" / "1" + assert (directory / "proposal.json").is_file() and (directory / "scores.json").is_file() + attribution = json.loads((directory / "attribution.json").read_text()) + assert attribution["flipped_predicted"] == ["hold-a"] or attribution["flipped_predicted"] == ["train-a", "hold-a"] + assert attribution["flipped_unpredicted"] == [] + trajectories = (directory / "trajectories.jsonl").read_text() + assert "REDACTED" in trajectories and "B" * 24 not in trajectories + assert record.attribution.precision == 1.0 + + +def test_a_round_number_is_never_reused(world) -> None: # type: ignore[no-untyped-def] + from openultrasast.learning.journal import JournalError + + _run(world, _proposal("- ask about quoting\n"), {"- ask about quoting": True}) + (world["out"] / "rounds" / "2").mkdir(parents=True) + with pytest.raises(JournalError, match="2"): + _run(world, _proposal("- another\n"), {}) + + +def test_every_stage_scores_at_least_three_runs_per_pair(world) -> None: # type: ignore[no-untyped-def] + with pytest.raises(ValueError, match="three"): + _run(world, _proposal("- x\n"), {}, k_runs=2)