From 760df7c809d5f2d59957708cd2b5045bd11acc97 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 22 Jun 2026 17:36:58 +0200 Subject: [PATCH] =?UTF-8?q?feat(cli):=20add=20`ousast=20improve`=20?= =?UTF-8?q?=E2=80=94=20run=20the=20self-improvement=20loop=20from=20the=20?= =?UTF-8?q?CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expose the bounded, gated self-improvement loop as a first-class subcommand so it can be steered from the command line / opencode. - `ousast improve ` resolves the target from the benchmark manifest, runs the loop to convergence, and writes accepted edits to the loop-owned ledger at /.openultrasast/calibration/rule_policy.json — exactly where `scan` and `benchmark` read it, so a subsequent scan of that target automatically picks up the improved ruleset (the loop closes end-to-end). - Flags: --max-rounds, --recall-floor, --fp-ceiling, --ledger, --journal, --ruleset-dir, and --dry-run (runs against a throwaway ledger, touches nothing). - Per-round output: reason, recall/fp/score deltas, and the edits applied. - tests/test_cli_improve.py: accept path writes the loop-owned ledger (noisy rule auto-shadowed), --dry-run never touches it, missing manifest fails loudly. - README: quickstart entry + a "Rule-level loop: ousast improve" subsection documenting the bounded levers, hard acceptance gate, and the HarnessX MetaAgent as the optional proposer over the same gated machinery. 131 passed, 3 skipped; ruff + mypy clean; detection gate PASS. Co-Authored-By: Claude Opus 4.8 --- README.md | 49 +++++++++++++++++--- src/openultrasast/cli.py | 95 +++++++++++++++++++++++++++++++++++++++ tests/test_cli_improve.py | 60 +++++++++++++++++++++++++ 3 files changed, 199 insertions(+), 5 deletions(-) create mode 100644 tests/test_cli_improve.py diff --git a/README.md b/README.md index 01956d6..4eb520f 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,9 @@ uv run ousast index /path/to/repo # Score the detector against a benchmark manifest uv run ousast benchmark benchmarks/manifests/python-vulnerable.toml --mode quick + +# Run the bounded self-improvement loop over the ruleset (benchmark-driven) +uv run ousast improve benchmarks/manifests/python-vulnerable.toml --dry-run ``` Every run writes auditable artifacts under `.openultrasast/runs//` @@ -273,11 +276,47 @@ priority is demoted rather than the finding deleted, the audit trail (`applied_calibrations.json`, the ledger) shows exactly why a surface lost attention. Covered by `tests/test_pipeline_calibration.py`. -Beyond the automatic ranking loop, `openultrasast-triage` still lets OpenCode -adjust prompt constraints, retrieval filters, skill routing, and benchmark-miss -triage. Those `RankingCalibration` fields are already produced and will be -consumed directly once the language-aware LLM hunters land -(`standard_security_harness`). +### Rule-level loop: `ousast improve` + +The ranking loop above adjusts *attention*. A second, bounded loop adjusts the +**governed ruleset itself** from benchmark feedback — and writes its result to the +same loop-owned ledger a scan reads, so the loop closes end-to-end: + +```bash +# Preview what the loop would change (writes nothing) +uv run ousast improve benchmarks/manifests/java-spring-boot-vulnerable.toml --dry-run + +# Apply: accepted rounds land in /.openultrasast/calibration/rule_policy.json, +# which the next `ousast scan`/`benchmark` of that target loads automatically. +uv run ousast improve benchmarks/manifests/java-spring-boot-vulnerable.toml +``` + +Each round (`improve/evolve.py`): benchmark with the current ledger → per-rule +signals (`rule_signals.json`) → propose **bounded** edits → validate → replay +smoke → re-benchmark → a **hard acceptance gate** → accept or revert +byte-for-byte. The loop may only pull two levers (rule status, score constants); +it can never edit a rule's pattern text or the authoritative 0–5 CWE severity, and +must stage `enabled → shadow` before `disabled`. A round is accepted only if +**recall ≥ 90% AND FP < 10% AND the project score did not regress AND no matched +finding was lost** — otherwise the ledger is never written. Auto-shadowing removes +a precision-dragger's false-positive cost without deleting it (recall is +preserved); persistent misses are *nominated* through the signal file, leaving +pattern authoring to a human pull request. The project score is the optimization +reward; the recall/FP gate is a separate hard constraint that is never folded into +it — and the same gate runs in CI (`python -m openultrasast.gate`), so a loop +result that breached it could never merge. Covered by `tests/test_improve.py` and +`tests/test_cli_improve.py`. + +This is the deterministic substrate of the self-improvement loop. When the +optional `openultrasast[harnessx]` extra is present, the HarnessX `MetaAgent` +becomes the richer *proposer* that plugs into this same validated, gated +machinery — the safety contract (bounded levers, replay/novelty/journal gates, +hard acceptance gate, byte-for-byte revert) is identical either way. + +Beyond these loops, `openultrasast-triage` still lets OpenCode adjust prompt +constraints, retrieval filters, skill routing, and benchmark-miss triage. Those +`RankingCalibration` fields are already produced and will be consumed directly +once the language-aware LLM hunters land (`standard_security_harness`). ## Development diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 1060265..80105bd 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse import json +import tempfile from dataclasses import dataclass from pathlib import Path from time import perf_counter @@ -27,10 +28,12 @@ from .calibration import ( ) from .config import load_config from .findings import StaticFinding, quick_scan_findings, write_findings +from .gate import FALSE_POSITIVE_CEILING, RECALL_FLOOR from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config from .harness_ext import has_harnessx from .hunter import run_hunter_pool, write_hunter_trajectories from .hunter_harness import HxScanOrchestrator +from .improve import RoundOutcome, run_improvement from .index import build_code_chunks from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints from .policy import assert_rules_resolve, load_policy @@ -77,6 +80,23 @@ def main(argv: list[str] | None = None) -> int: benchmark.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick") benchmark.add_argument("--config", type=Path, default=Path("openultrasast.toml")) + improve = subparsers.add_parser( + "improve", + help="run the bounded self-improvement loop against a benchmark manifest and update the loop-owned ruleset ledger", + ) + improve.add_argument("manifest", type=Path) + improve.add_argument("--max-rounds", type=int, default=5) + improve.add_argument("--recall-floor", type=float, default=RECALL_FLOOR) + improve.add_argument("--fp-ceiling", type=float, default=FALSE_POSITIVE_CEILING) + improve.add_argument( + "--ledger", type=Path, default=None, help="ruleset ledger path (default: /.openultrasast/calibration/rule_policy.json)" + ) + improve.add_argument("--journal", type=Path, default=None, help="improvement journal path (default: alongside the ledger)") + improve.add_argument( + "--ruleset-dir", type=Path, default=DEFAULT_RULESET_DIR, help="ruleset directory to improve (default: the bundled ruleset)" + ) + improve.add_argument("--dry-run", action="store_true", help="run rounds against a throwaway ledger; never touch the target's ledger") + args = parser.parse_args(argv) if args.command == "scan": return _scan(args.path, args.config, args.mode, args.fail_on) @@ -84,6 +104,17 @@ def main(argv: list[str] | None = None) -> int: return _index(args.path, args.config, args.chunk_lines) if args.command == "benchmark": return _benchmark(args.manifest, args.config, args.mode) + if args.command == "improve": + return _improve( + args.manifest, + max_rounds=args.max_rounds, + recall_floor=args.recall_floor, + fp_ceiling=args.fp_ceiling, + ledger=args.ledger, + journal=args.journal, + ruleset_dir=args.ruleset_dir, + dry_run=args.dry_run, + ) return 2 @@ -318,6 +349,70 @@ def _benchmark(manifest_path: Path, config_path: Path, mode: str) -> int: return 0 +def _improve( + manifest_path: Path, + *, + max_rounds: int, + recall_floor: float, + fp_ceiling: float, + ledger: Path | None, + journal: Path | None, + ruleset_dir: Path, + dry_run: bool, +) -> int: + if not manifest_path.exists() or not manifest_path.is_file(): + raise SystemExit(f"benchmark manifest is not a file: {manifest_path}") + manifest = load_benchmark_manifest(manifest_path) + target = resolve_benchmark_source(manifest_path, manifest) + policy = load_policy() + + # The loop writes its accepted ledger where `scan`/`benchmark` read it, so a + # subsequent scan of this target automatically picks up the improved ruleset. + default_ledger = target / CALIBRATION_DIR / "rule_policy.json" + with tempfile.TemporaryDirectory(prefix="ousast-improve-") as scratch: + if dry_run: + ledger_path = Path(scratch) / "rule_policy.json" + journal_path = Path(scratch) / "improve_journal.json" + else: + ledger_path = ledger or default_ledger + journal_path = journal or ledger_path.with_name("improve_journal.json") + + outcomes = run_improvement( + target, + manifest, + ledger_path=ledger_path, + journal_path=journal_path, + ruleset_dir=ruleset_dir, + policy=policy, + max_rounds=max_rounds, + recall_floor=recall_floor, + fp_ceiling=fp_ceiling, + ) + _print_improve_outcomes(outcomes, manifest_path, target, ledger_path, dry_run=dry_run) + return 0 + + +def _print_improve_outcomes(outcomes: list[RoundOutcome], manifest_path: Path, target: Path, ledger_path: Path, *, dry_run: bool) -> None: + print(f"manifest={manifest_path}") + print(f"target={target}") + accepted = [o for o in outcomes if o.accepted] + for outcome in outcomes: + edits = ", ".join(f"{e.rule_id}:{e.from_status}->{e.to_status}" for e in outcome.edits) or "-" + print( + f"round {outcome.round}: {outcome.reason} | " + f"recall {outcome.recall_before:.2%}->{outcome.recall_after:.2%} " + f"fp {outcome.fp_before:.2%}->{outcome.fp_after:.2%} " + f"score {outcome.score_before}->{outcome.score_after} | edits: {edits}" + ) + print(f"rounds={len(outcomes)} accepted={len(accepted)}") + if dry_run: + print("dry_run=true (no ledger written)") + elif accepted: + print(f"ledger={ledger_path}") + else: + print("ledger=unchanged (no round accepted)") + + def _load_scan_findings(run_dir: Path) -> list[StaticFinding]: return load_findings(run_dir / "findings.json") diff --git a/tests/test_cli_improve.py b/tests/test_cli_improve.py new file mode 100644 index 0000000..52bbdac --- /dev/null +++ b/tests/test_cli_improve.py @@ -0,0 +1,60 @@ +"""`ousast improve` CLI: drives the self-improvement loop and the loop-owned ledger.""" + +from pathlib import Path + +from openultrasast.cli import main +from openultrasast.ruleset import PatternRule, read_rule_ledger, write_ruleset + + +def _rule(rule_id: str, pattern: str) -> PatternRule: + return PatternRule(rule_id=rule_id, title=rule_id, languages=("python",), cwe="CWE-95", tags=("injection",), pattern=pattern) + + +def _expected_block(rule_id: str) -> str: + return ( + '[[expected]]\ncwe = "CWE-95"\nclass = "code injection"\npath = "app.py"\n' + f'line = 2\nrule_id = "{rule_id}"\nsink = "eval"\nevidence = "eval executes input"\n' + ) + + +def _scaffold(tmp_path: Path) -> tuple[Path, Path]: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g():\n print('hello')\n") + ruleset_dir = tmp_path / "ruleset" + write_ruleset(ruleset_dir / "python" / "rules.toml", [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")]) + manifest = tmp_path / "bench.toml" + manifest.write_text('name = "t"\nlanguage = "python_web"\n\n[source]\npath = "repo"\n\n' + _expected_block("good-eval")) + return repo, ruleset_dir + + +def test_improve_accepts_and_writes_loop_owned_ledger(tmp_path: Path) -> None: + repo, ruleset_dir = _scaffold(tmp_path) + manifest = tmp_path / "bench.toml" + + exit_code = main(["improve", str(manifest), "--ruleset-dir", str(ruleset_dir)]) + + assert exit_code == 0 + # The ledger lands exactly where `scan`/`benchmark` read it for this target. + ledger = repo / ".openultrasast" / "calibration" / "rule_policy.json" + assert ledger.exists() + assert read_rule_ledger(ledger)["noisy-print"]["status"] == "shadow" + journal = repo / ".openultrasast" / "calibration" / "improve_journal.json" + assert journal.exists() + + +def test_improve_dry_run_never_touches_the_targets_ledger(tmp_path: Path) -> None: + repo, ruleset_dir = _scaffold(tmp_path) + manifest = tmp_path / "bench.toml" + + exit_code = main(["improve", str(manifest), "--ruleset-dir", str(ruleset_dir), "--dry-run"]) + + assert exit_code == 0 + assert not (repo / ".openultrasast" / "calibration" / "rule_policy.json").exists() + + +def test_improve_rejects_missing_manifest(tmp_path: Path) -> None: + import pytest + + with pytest.raises(SystemExit): + main(["improve", str(tmp_path / "nope.toml")])