feat(cli): add ousast improve — run the self-improvement loop from the CLI

Expose the bounded, gated self-improvement loop as a first-class subcommand so it
can be steered from the command line / opencode.

- `ousast improve <manifest>` resolves the target from the benchmark manifest,
  runs the loop to convergence, and writes accepted edits to the loop-owned ledger
  at <target>/.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 <[email protected]>
This commit is contained in:
Marius Ciepluch
2026-06-22 17:36:58 +02:00
co-authored by Claude Opus 4.8
parent 649bd42fc3
commit 760df7c809
3 changed files with 199 additions and 5 deletions
+44 -5
View File
@@ -29,6 +29,9 @@ uv run ousast index /path/to/repo
# Score the detector against a benchmark manifest # Score the detector against a benchmark manifest
uv run ousast benchmark benchmarks/manifests/python-vulnerable.toml --mode quick 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/<scan-id>/` Every run writes auditable artifacts under `.openultrasast/runs/<scan-id>/`
@@ -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 (`applied_calibrations.json`, the ledger) shows exactly why a surface lost
attention. Covered by `tests/test_pipeline_calibration.py`. attention. Covered by `tests/test_pipeline_calibration.py`.
Beyond the automatic ranking loop, `openultrasast-triage` still lets OpenCode ### Rule-level loop: `ousast improve`
adjust prompt constraints, retrieval filters, skill routing, and benchmark-miss
triage. Those `RankingCalibration` fields are already produced and will be The ranking loop above adjusts *attention*. A second, bounded loop adjusts the
consumed directly once the language-aware LLM hunters land **governed ruleset itself** from benchmark feedback — and writes its result to the
(`standard_security_harness`). 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 <target>/.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 ## Development
+95
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import argparse import argparse
import json import json
import tempfile
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from time import perf_counter from time import perf_counter
@@ -27,10 +28,12 @@ from .calibration import (
) )
from .config import load_config from .config import load_config
from .findings import StaticFinding, quick_scan_findings, write_findings 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 import HarnessRuntime, HarnessTraceWriter, write_harness_config
from .harness_ext import has_harnessx from .harness_ext import has_harnessx
from .hunter import run_hunter_pool, write_hunter_trajectories from .hunter import run_hunter_pool, write_hunter_trajectories
from .hunter_harness import HxScanOrchestrator from .hunter_harness import HxScanOrchestrator
from .improve import RoundOutcome, run_improvement
from .index import build_code_chunks from .index import build_code_chunks
from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints 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 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("--mode", choices=("quick", "standard", "deep"), default="quick")
benchmark.add_argument("--config", type=Path, default=Path("openultrasast.toml")) 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: <target>/.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) args = parser.parse_args(argv)
if args.command == "scan": if args.command == "scan":
return _scan(args.path, args.config, args.mode, args.fail_on) 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) return _index(args.path, args.config, args.chunk_lines)
if args.command == "benchmark": if args.command == "benchmark":
return _benchmark(args.manifest, args.config, args.mode) 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 return 2
@@ -318,6 +349,70 @@ def _benchmark(manifest_path: Path, config_path: Path, mode: str) -> int:
return 0 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]: def _load_scan_findings(run_dir: Path) -> list[StaticFinding]:
return load_findings(run_dir / "findings.json") return load_findings(run_dir / "findings.json")
+60
View File
@@ -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")])