diff --git a/benchmarks/measurements/2026-09-06-mechanism-lever-split.json b/benchmarks/measurements/2026-09-06-mechanism-lever-split.json new file mode 100644 index 0000000..a737ac5 --- /dev/null +++ b/benchmarks/measurements/2026-09-06-mechanism-lever-split.json @@ -0,0 +1,19 @@ +{ + "after": { + "detected": 5, + "shapes": 6, + "silent": 11, + "youden": -0.0588 + }, + "before": { + "detected": 7, + "shapes": 10, + "silent": 11, + "youden": 0.0588 + }, + "command": "ousast mechanisms export --slice vibe-py, then score every holdout pair with semantic.loo.score_pair_with_store", + "generated": "2026-09-06", + "holdout_pairs": 17, + "note": "`before` is the behaviour this spec found: the exporter seeded candidate shapes from every vendored pair, so a shape a holdout pair taught could recover that same pair. `after` is the same measurement with the teacher rule enforced: only train-split, scorable, trusted pairs seed. The difference is the inflation.", + "slice": "vibe-py" +} diff --git a/src/openultrasast/improve/evolve.py b/src/openultrasast/improve/evolve.py index 2bc64f0..a630dc5 100644 --- a/src/openultrasast/improve/evolve.py +++ b/src/openultrasast/improve/evolve.py @@ -116,7 +116,8 @@ def propose_mechanism_edits(pair_cases: Sequence[object], candidates: object, st from ..semantic.loo import score_pair_with_store from ..semantic.mechanisms import MechanismStore, corpus_mechanisms - cases = list(select_tier(select_vendored([case for case in pair_cases if isinstance(case, PairCase)]), GATING_TIERS)) + cases_all = [case for case in pair_cases if isinstance(case, PairCase)] + cases = list(select_tier(select_vendored(cases_all), GATING_TIERS)) if not cases: return [] facts = _facts() @@ -133,9 +134,13 @@ def propose_mechanism_edits(pair_cases: Sequence[object], candidates: object, st ) ) missed = [case for case in cases if not current[case.name].detected] + from ..learning.split import refuse_if_holdout + for record in corpus_mechanisms(candidates.load()): # type: ignore[attr-defined] if record.id in admitted or not missed: continue + if refuse_if_holdout(record.pairs, cases_all) is not None: + continue # learning-harness Req 5.2: a shape a holdout pair taught can never recover that pair with tempfile.TemporaryDirectory(prefix="ousast-lever-") as scratch: trial = MechanismStore(Path(scratch) / "trial.jsonl") trial.append(record) diff --git a/src/openultrasast/learning/split.py b/src/openultrasast/learning/split.py new file mode 100644 index 0000000..af3129b --- /dev/null +++ b/src/openultrasast/learning/split.py @@ -0,0 +1,64 @@ +"""One rule for who may teach, and one refusal for anything that touches the holdout (Req 5.1, 5.2). + +Every learning path in the tool asks this module the same question, so the split cannot be enforced in +one place and forgotten in another. A teacher is a vendored pair, at a review tier a maintainer stands +behind, on the train split, and scorable. Everything else may be measured; nothing else may teach. + +A refusal is data, not an exception: the path records which holdout pairs a candidate touched and moves +on, so a leak is visible in the run rather than silently accepted. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Sequence +from dataclasses import dataclass + +GATING_TIERS = frozenset({"seeded", "reviewed"}) +TRAIN = "train" +HOLDOUT = "holdout" +REFUSED = "holdout_pair_refused" + + +@dataclass(frozen=True) +class Refusal: + """A candidate that a holdout pair taught, or that touches one.""" + + pairs: tuple[str, ...] + reason: str = REFUSED + + def degradation(self) -> dict[str, object]: + return { + "stage": "learning", + "reason": self.reason, + "pairs": list(self.pairs), + "detail": f"refused: taught by or touching holdout pair(s) {', '.join(self.pairs)}", + } + + +def teachers(cases: Sequence[object]) -> tuple[object, ...]: + """The pairs a learning path may learn from. Everything else is measured, never learned from.""" + return tuple(case for case in cases if is_teacher(case)) + + +def is_teacher(case: object) -> bool: + if not bool(getattr(case, "vendored", True)): + return False + if str(getattr(case, "review_tier", "")) not in GATING_TIERS: + return False + if str(getattr(case, "split", TRAIN)) != TRAIN: + return False + return getattr(case, "unscorable", None) is None + + +def holdout_names(cases: Sequence[object]) -> frozenset[str]: + return frozenset(str(getattr(case, "name", "")) for case in cases if str(getattr(case, "split", "")) == HOLDOUT) + + +def refuse_if_holdout(taught_by: Iterable[str], cases: Sequence[object]) -> Refusal | None: + """None when nothing in ``taught_by`` is a holdout pair; a refusal naming them when something is.""" + held = holdout_names(cases) + touched = tuple(sorted({name for name in taught_by if name in held})) + return Refusal(pairs=touched) if touched else None + + +__all__ = ["GATING_TIERS", "HOLDOUT", "REFUSED", "TRAIN", "Refusal", "holdout_names", "is_teacher", "refuse_if_holdout", "teachers"] diff --git a/src/openultrasast/semantic/loo.py b/src/openultrasast/semantic/loo.py index 96dbaf1..2884e05 100644 --- a/src/openultrasast/semantic/loo.py +++ b/src/openultrasast/semantic/loo.py @@ -20,7 +20,7 @@ from dataclasses import dataclass, field from pathlib import Path from ..findings import StaticFinding -from ..pairs import GATING_TIERS, PairCase, _materialize, _targets +from ..pairs import PairCase, _materialize, _targets from .facts import FactLoadError, SemanticFacts, load_facts from .functions import named_function_ranges, spans_named from .mechanisms import Mechanism, MechanismStore, append_from_pair, corpus_mechanisms @@ -99,7 +99,10 @@ def evaluate_loo(cases: Sequence[PairCase], *, facts: SemanticFacts | None = Non skipped.append((case.name, "pointer_pair_not_cached" if not case.vendored else "excerpt_missing")) else: targets.append(case) - lessons = {case.name: pair_lessons(case, loaded) for case in targets if case.review_tier in GATING_TIERS} + from ..learning.split import is_teacher + + # learning-harness Req 5.1: every pair is still scored, but only a train-split teacher may seed a store. + lessons = {case.name: pair_lessons(case, loaded) for case in targets if is_teacher(case)} outcomes = [_hold_out(case, targets, lessons, loaded) for case in targets] return LooResult( outcomes=tuple(outcomes), diff --git a/src/openultrasast/semantic/seed.py b/src/openultrasast/semantic/seed.py index f483fa5..42d167c 100644 --- a/src/openultrasast/semantic/seed.py +++ b/src/openultrasast/semantic/seed.py @@ -193,6 +193,10 @@ def _obligation_facts() -> ObligationFacts: def _skip_reason(case: PairCase) -> str | None: + from ..learning.split import is_teacher + + if not is_teacher(case) and case.review_tier in GATING_TIERS and case.vendored and not case.known_limit: + return f"split:{case.split}" # learning-harness Req 5.1: only train-split, scorable pairs teach if case.review_tier not in GATING_TIERS: return f"tier:{case.review_tier}" if case.known_limit: diff --git a/tests/test_learning_split.py b/tests/test_learning_split.py new file mode 100644 index 0000000..d3fdd9d --- /dev/null +++ b/tests/test_learning_split.py @@ -0,0 +1,109 @@ +"""learning-harness task 3.1: one teacher rule across every learning path (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.pairs import DEFAULT_CATALOG, PairCase, load_pair_catalog, select_slice, select_vendored + + +def _case(tmp_path: Path, name: str, *, split: str = "train", tier: str = "seeded", unscorable: str | None = None) -> PairCase: + (tmp_path / f"{name}.py").write_text(f"# {name}\nx = 1\n") + return PairCase( + name=name, + slice="vibe-py", + language="python", + origin="test", + vuln_file=tmp_path / f"{name}.py", + fixed_file=tmp_path / f"{name}.py", + relpath="app.py", + expected=(ExpectedFinding(cwe="CWE-89", vulnerability_class="x", path="app.py", evidence="", function="f", mechanism="other"),), + min_recall=1.0, + fix_policy="silent", + split=split, + review_tier=tier, + reviewer="m" if tier == "reviewed" else "", + unscorable=unscorable, + ) + + +def test_only_vendored_trusted_scorable_train_pairs_may_teach(tmp_path: Path) -> None: + from openultrasast.learning.split import teachers + + cases = [ + _case(tmp_path, "train-seeded"), + _case(tmp_path, "train-reviewed", tier="reviewed"), + _case(tmp_path, "holdout-seeded", split="holdout"), + _case(tmp_path, "train-untrusted", tier="title"), + _case(tmp_path, "train-twin", unscorable="identical_twin"), + ] + assert [case.name for case in teachers(cases)] == ["train-seeded", "train-reviewed"] + + +def test_a_candidate_taught_by_a_holdout_pair_is_refused_by_name(tmp_path: Path) -> None: + from openultrasast.learning.split import refuse_if_holdout + + cases = [_case(tmp_path, "train-one"), _case(tmp_path, "holdout-one", split="holdout")] + assert refuse_if_holdout(["train-one"], cases) is None + refusal = refuse_if_holdout(["train-one", "holdout-one"], cases) + assert refusal is not None and refusal.pairs == ("holdout-one",) + assert refusal.reason == "holdout_pair_refused" + degradation = refusal.degradation() + assert degradation["stage"] == "learning" and degradation["pairs"] == ["holdout-one"] + assert "holdout-one" in str(degradation["detail"]) + + +def test_the_exporter_seeds_from_train_pairs_only(tmp_path: Path) -> None: + from openultrasast.semantic.mechanisms import MechanismStore, corpus_mechanisms + from openultrasast.semantic.seed import export_mechanisms + + cases = select_vendored(select_slice(load_pair_catalog(DEFAULT_CATALOG), "vibe-py")) + holdout = {case.name for case in cases if case.split == "holdout"} + store = MechanismStore(tmp_path / "candidates.jsonl") + report = export_mechanisms(cases, store) + taught_by_holdout = [record.id for record in corpus_mechanisms(store.load()) if set(record.pairs) & holdout] + assert taught_by_holdout == [] + assert report.records > 0 and any(reason.startswith("split:") for _pair, reason in report.skipped) + + +def test_leave_one_out_teaches_from_train_pairs_only(tmp_path: Path) -> None: + from openultrasast.semantic.loo import evaluate_loo + + cases = [_case(tmp_path, "train-one"), _case(tmp_path, "holdout-one", split="holdout")] + result = evaluate_loo(cases) + assert {outcome.pair for outcome in result.outcomes} == {"train-one", "holdout-one"} # every pair is still scored + assert result.teaching_pairs <= 1 # only the train pair could ever have taught + + +def test_the_mechanism_lever_refuses_a_candidate_a_holdout_pair_taught(tmp_path: Path) -> None: + from openultrasast.improve.evolve import propose_mechanism_edits + from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair + from openultrasast.semantic.variants import Shape + + shape = Shape( + language="python", + sink_name="system", + arity=1, + source_positions=(0,), + source_kinds=("request",), + guard="none", + mechanism="source_reaches_sink", + ) + candidates = MechanismStore(tmp_path / "candidates.jsonl") + append_from_pair(candidates, shape, summary="s", cwe="CWE-78", pair="holdout-one", provenance="human", tier="seeded") + cases = [_case(tmp_path, "train-one"), _case(tmp_path, "holdout-one", split="holdout")] + edits = propose_mechanism_edits(cases, candidates, tmp_path / "scan.jsonl") + assert [edit.mechanism_id for edit in edits if edit.action == "admit"] == [] + + +def test_the_corrected_lever_number_is_measured_and_committed() -> None: + report = Path("benchmarks/measurements/2026-09-06-mechanism-lever-split.json") + assert report.is_file() + payload = json.loads(report.read_text()) + assert payload["slice"] == "vibe-py" and payload["holdout_pairs"] > 0 + assert set(payload["before"]) >= {"detected", "silent", "youden", "shapes"} + assert set(payload["after"]) >= {"detected", "silent", "youden", "shapes"} + assert payload["after"]["shapes"] < payload["before"]["shapes"] # holdout pairs no longer teach + assert payload["after"]["youden"] <= payload["before"]["youden"] diff --git a/tests/test_mechanism_loop.py b/tests/test_mechanism_loop.py index 56ef60e..558b735 100644 --- a/tests/test_mechanism_loop.py +++ b/tests/test_mechanism_loop.py @@ -195,14 +195,26 @@ def test_no_gain_admission_is_rejected_and_reverted_mechanism_edits_are_not_repr def _holdout_catalog(tmp_path: Path) -> Path: + """A train pair to teach and a holdout pair to recover. + + The catalog used to hold the holdout pair alone, so the only shape available was the one that pair + taught about itself; learning-harness Req 5.1 refuses that, and rightly. + """ (tmp_path / "x-v.py").write_text(EXEC_VULN) (tmp_path / "x-f.py").write_text(EXEC_FIX) + (tmp_path / "t-v.py").write_text(EXEC_VULN.replace("lookup", "search").replace("u = ", "name = ")) + (tmp_path / "t-f.py").write_text(EXEC_FIX.replace("lookup", "search").replace("u = ", "name = ")) + row = ( + '[[pair]]\nname = "{name}"\nslice = "github"\nlanguage = "python"\nvuln = "{name[0]}-v.py"\nfixed = "{name[0]}-f.py"\n' + 'relpath = "app.py"\nsplit = "{split}"\nreview_tier = "reviewed"\nreviewer = "t"\n\n' + '[[pair.expected]]\ncwe = "CWE-89"\nclass = "sql injection"\npath = "app.py"\nfunction = "{function}"\n' + 'sink = "execute"\nmechanism = "source_reaches_sink"\n' + ) catalog = tmp_path / "pairs.toml" catalog.write_text( - '[[pair]]\nname = "x"\nslice = "github"\nlanguage = "python"\nvuln = "x-v.py"\nfixed = "x-f.py"\nrelpath = "app.py"\n' - 'split = "holdout"\nreview_tier = "reviewed"\nreviewer = "t"\n\n' - '[[pair.expected]]\ncwe = "CWE-89"\nclass = "sql injection"\npath = "app.py"\nfunction = "lookup"\n' - 'sink = "execute"\nmechanism = "source_reaches_sink"\n' + row.replace("{name}", "x").replace("{name[0]}", "x").replace("{split}", "holdout").replace("{function}", "lookup") + + "\n" + + row.replace("{name}", "t").replace("{name[0]}", "t").replace("{split}", "train").replace("{function}", "search") ) return catalog diff --git a/tests/test_obligation_export.py b/tests/test_obligation_export.py index 70ac011..e96e6f0 100644 --- a/tests/test_obligation_export.py +++ b/tests/test_obligation_export.py @@ -111,29 +111,27 @@ def test_sink_search_ignores_obligation_rows(tmp_path: Path) -> None: assert result.mechanisms_searched == 0 and result.hits == () and result.degradations == () -def test_real_vibe_py_absence_pairs_seed_at_least_one_obligation_shape(tmp_path: Path) -> None: +def test_the_train_split_of_vibe_py_teaches_no_obligation_shape_today(tmp_path: Path) -> None: + """learning-harness Req 5.1 turned this number honest, and the honest number is zero. + + Every vibe-py absence pair that yields an obligation lesson sits on the holdout side, so once only + train-split pairs may teach, the corpus teaches no obligation shape at all. The earlier result + ("at least one obligation shape from vibe-py") was true only because holdout pairs were teaching. + The derivation itself stays covered by the fixture tests above; this pins the corpus gap, which the + absence re-harvest and a maintainer pass over the splits are what close. + """ from openultrasast.pairs import load_pair_catalog, select_slice, select_vendored from openultrasast.semantic.obligations.shapes import obligation_mechanisms from openultrasast.semantic.seed import export_mechanisms store = MechanismStore(tmp_path / "m.jsonl") - report = export_mechanisms(select_vendored(select_slice(load_pair_catalog(), "vibe-py")), store) - rows = obligation_mechanisms(store.load()) - assert rows and any("vampi-books-get-by-title" in r.pairs and "vampi-users-update-password" in r.pairs for r in rows) - vampi = next(r for r in rows if "vampi-books-get-by-title" in r.pairs) - assert vampi.shape is not None - assert (vampi.shape["operation_kind"], vampi.shape["discharger_kind"], vampi.shape["provenance"], vampi.shape["resource_class"]) == ( - "protected_read", - "identity_constraint", - "authenticated_context", - "owned", - ) - assert all( - r.shape is not None and r.shape["operation_kind"] in {"protected_read", "protected_write", "privileged_action", "security_setting"} - for r in rows - ) - flow_rows = [r for r in store.load() if r.shape is not None and r.shape.get("family", "sink") == "sink"] - assert report.records == len(flow_rows) + len(rows) and len(flow_rows) >= 1 # both families, nothing lost to the other + cases = select_vendored(select_slice(load_pair_catalog(), "vibe-py")) + report = export_mechanisms(cases, store) + assert obligation_mechanisms(store.load()) == () + refused = {pair for pair, reason in report.skipped if reason.startswith("split:")} + assert {"vampi-books-get-by-title", "vampi-users-update-password", "threatbyte-api-v1-get"} <= refused + flow_rows = [row for row in store.load() if row.shape is not None and row.shape.get("family", "sink") == "sink"] + assert report.records == len(flow_rows) and len(flow_rows) >= 1 # the flow family still teaches from the train split def test_absence_row_without_an_obligation_lesson_still_teaches_its_sink_shape(tmp_path: Path) -> None: