diff --git a/.kiro/specs/harnessx-self-improving-rulesets/spec.json b/.kiro/specs/harnessx-self-improving-rulesets/spec.json index 05d93ba..88bd690 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/spec.json +++ b/.kiro/specs/harnessx-self-improving-rulesets/spec.json @@ -1,7 +1,7 @@ { "feature_name": "harnessx-self-improving-rulesets", "created_at": "2026-06-22T10:07:18Z", - "updated_at": "2026-06-22T15:05:00Z", + "updated_at": "2026-06-22T15:35:00Z", "language": "en", "phase": "implementation", "approvals": { diff --git a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md index 42fb24a..ac7be27 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md +++ b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md @@ -177,15 +177,15 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: HxScanOrchestrator_ - _Depends: 6.1, 6.4_ -- [ ] 7. Stand up the bounded self-improvement loop -- [ ] 7.1 Add the two bounded edit levers +- [x] 7. Stand up the bounded self-improvement loop +- [x] 7.1 Add the two bounded edit levers - Expose exactly two new edit levers: a rule lever covering rule status, evidence floor, and threshold, and a policy lever covering evidence floor, scope, and the score constants. - Observable completion: the loop's valid-lever set includes exactly the rule and policy levers and rejects any edit outside their declared fields, verified by a lever test. - _Requirements: 3.1_ - _Boundary: EvolveValidator, improve/evolve.py_ - _Depends: 5.2_ -- [ ] 7.2 Enforce the rule/policy edit safety bounds in the validator +- [x] 7.2 Enforce the rule/policy edit safety bounds in the validator - Reject any loop edit that modifies detection pattern text and any edit that modifies the authoritative 0-5 CWE severity. - Reject moving a rule directly from enabled to disabled, requiring shadow as a mandatory intermediate staging state. - Assert that each edit parses, that status is one of the three literals, that threshold/evidence-floor changes stay within configured bounds, that an added rule does not duplicate an existing identifier, and that every referenced CWE resolves in policy; raise a rule-change validation error and reject the whole round on any failure. @@ -194,7 +194,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: EvolveValidator_ - _Depends: 7.1, 1.1_ -- [ ] 7.3 Add the novelty, replay, and journal gates +- [x] 7.3 Add the novelty, replay, and journal gates - Block re-proposing a previously reverted edit that lacks a retry rationale via the novelty gate. - Run a replay smoke gate before any benchmark re-run proving the edited ruleset and policy still load and boot. - Record each rule and policy edit, its lever, and its outcome in the journal so every change is attributable and reversible. @@ -203,14 +203,14 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: EvolveValidator, improve/evolve.py_ - _Depends: 7.2_ -- [ ] 7.4 Convert benchmark deltas into loop signals +- [x] 7.4 Convert benchmark deltas into loop signals - Convert the per-rule recommendation delta and false-positive learnings into a loop signal file, writing recall-side items as miss signals and precision-side items as verifier-reject or false-positive signals. - Observable completion: a benchmark run produces a signal file whose recall-side entries are miss signals and precision-side entries are verifier-reject/false-positive signals, verified against the documented signal schema. - _Requirements: 4.3_ - _Boundary: improve/evolve.py_ - _Depends: 4.2, 7.1_ -- [ ] 7.5 Implement the hard acceptance gate with byte-for-byte revert +- [x] 7.5 Implement the hard acceptance gate with byte-for-byte revert - After the meta-agent edits rules or policy, re-run the benchmark and accept the round only if aggregate recall is at least 90% AND false-positive rate is under 10% AND the project score did not regress on the held set AND no unpredicted regression occurred. - On any failed acceptance condition, revert the round and restore the rule and policy configuration byte-for-byte. - Observable completion: a round that regresses recall is reverted and the configuration is restored byte-for-byte, while a compliant round is accepted and persisted, proven by a self-improvement integration test. @@ -218,7 +218,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: improve/evolve.py_ - _Depends: 7.3, 7.4, 4.1_ -- [ ] 7.6 Auto-shadow precision-draggers, nominate coverage gaps, and separate the objective +- [x] 7.6 Auto-shadow precision-draggers, nominate coverage gaps, and separate the objective - Auto-shadow a rule that drags aggregate precision below the gate rather than delete it, preserving recall while removing the false-positive cost. - Nominate a new or loosened rule through the signal file when a coverage gap surfaces as a persistent miss, leaving pattern authoring to a human pull request. - Treat maximizing the project score as the optimization objective and the recall/FP gate as a separate hard feasibility constraint that is never folded into the score reward. diff --git a/src/openultrasast/improve/__init__.py b/src/openultrasast/improve/__init__.py new file mode 100644 index 0000000..a8e7699 --- /dev/null +++ b/src/openultrasast/improve/__init__.py @@ -0,0 +1,30 @@ +"""Bounded, gated self-improvement loop over the ruleset/policy governance data.""" + +from .evolve import RoundOutcome, build_rule_signals, propose_status_edits, run_improvement, run_round +from .journal import load_journal, reverted_edit_keys +from .validator import ( + VALID_LEVERS, + EvolveBounds, + EvolveValidator, + PolicyConstantEdit, + RuleStatusEdit, + StrictValidationError, + edits_to_ledger, +) + +__all__ = [ + "EvolveBounds", + "EvolveValidator", + "PolicyConstantEdit", + "RoundOutcome", + "RuleStatusEdit", + "StrictValidationError", + "VALID_LEVERS", + "build_rule_signals", + "edits_to_ledger", + "load_journal", + "propose_status_edits", + "reverted_edit_keys", + "run_improvement", + "run_round", +] diff --git a/src/openultrasast/improve/evolve.py b/src/openultrasast/improve/evolve.py new file mode 100644 index 0000000..4fb2779 --- /dev/null +++ b/src/openultrasast/improve/evolve.py @@ -0,0 +1,267 @@ +"""Deterministic self-improvement loop (tasks 7.4, 7.5, 7.6). + +One round: benchmark with the current ledger -> per-rule signals -> propose bounded +edits (auto-shadow precision-draggers) -> validate -> replay smoke -> re-benchmark -> +hard acceptance gate (recall >= floor AND FP < ceiling AND project score not regressed +AND no matched-finding regression) -> accept (persist the ledger) or revert byte-for-byte. + +The edit *proposer* here is the benchmark's per-rule signal; the LLM ``MetaAgent.evolve`` +proposer plugs into this same validated/gated machinery when the HarnessX extra is present. +The project score is the optimization reward; the recall/FP gate is a separate hard +feasibility constraint that is never folded into the reward. +""" + +from __future__ import annotations + +import tempfile +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path + +from ..benchmark import BenchmarkManifest, BenchmarkResult, BenchmarkRun, evaluate_benchmark +from ..findings import StaticFinding, quick_scan_findings +from ..mapping import analyze_entry_points, attach_reachability_hints +from ..policy import CwePolicy, load_policy +from ..preprocess import preprocess_repository +from ..rank import rank_targets +from ..ruleset import DEFAULT_RULESET_DIR, PatternRule, load_ruleset, read_rule_ledger, write_rule_ledger +from ..scoring import build_score_artifact +from .journal import append_round, load_journal, next_round_index, reverted_edit_keys +from .validator import EvolveValidator, RuleStatusEdit, StrictValidationError, edits_to_ledger + + +@dataclass(frozen=True) +class RoundOutcome: + round: int + accepted: bool + reason: str + edits: list[RuleStatusEdit] = field(default_factory=list) + recall_before: float = 0.0 + recall_after: float = 0.0 + fp_before: float = 0.0 + fp_after: float = 0.0 + score_before: int = 0 + score_after: int = 0 + matched_before: int = 0 + matched_after: int = 0 + + +def build_rule_signals(result: BenchmarkResult) -> list[dict[str, object]]: + """Convert a benchmark result into per-rule loop signals (recall: miss; precision: fp).""" + signals: list[dict[str, object]] = [] + for miss in result.misses: + signals.append({"rule_id": miss.rule_id or "", "signal": "miss", "cwe": miss.cwe, "path": miss.path}) + for fp in result.false_positives: + signals.append({"rule_id": fp.rule_id, "signal": "fp", "path": fp.path, "line": fp.line}) + return sorted(signals, key=lambda item: (str(item.get("signal")), str(item.get("rule_id")), str(item.get("path")))) + + +def propose_status_edits( + per_rule: Mapping[str, dict[str, int]], + ruleset_by_id: Mapping[str, PatternRule], + current_ledger: Mapping[str, dict[str, object]], + blocked_keys: set[str], +) -> list[RuleStatusEdit]: + """Auto-shadow precision-draggers: enabled rules with false positives -> shadow (never delete).""" + edits: list[RuleStatusEdit] = [] + for rule_id, counts in sorted(per_rule.items()): + if counts.get("false_positives", 0) <= 0: + continue + rule = ruleset_by_id.get(rule_id) + if rule is None: + continue + current_status = str(current_ledger.get(rule_id, {}).get("status", rule.status)) + if current_status != "enabled": + continue # already staged or disabled + edit = RuleStatusEdit(rule_id=rule_id, from_status="enabled", to_status="shadow", rationale="") + if edit.key() in blocked_keys: + continue # novelty gate: previously reverted without a new rationale + edits.append(edit) + return edits + + +def run_round( + target: Path, + manifest: BenchmarkManifest, + *, + ledger_path: Path, + journal_path: Path, + ruleset_dir: Path = DEFAULT_RULESET_DIR, + policy: dict[str, CwePolicy] | None = None, + recall_floor: float = 0.9, + fp_ceiling: float = 0.1, +) -> RoundOutcome: + policy = policy if policy is not None else load_policy() + rounds = load_journal(journal_path) + round_index = next_round_index(rounds) + blocked = reverted_edit_keys(rounds) + current_ledger = read_rule_ledger(ledger_path) + + before_findings, before_rules = _scan(target, ruleset_dir, current_ledger, policy) + before = _evaluate(target, manifest, before_findings, before_rules, policy) + (ledger_path.parent / "rule_signals.json").parent.mkdir(parents=True, exist_ok=True) + _write_signals(ledger_path.parent / "rule_signals.json", build_rule_signals(before.result)) + + ruleset_by_id = {rule.rule_id: rule for rule in before_rules} + edits = propose_status_edits(before.result.metrics.per_rule, ruleset_by_id, current_ledger, blocked) + if not edits: + return _outcome(round_index, accepted=False, reason="no_proposals", edits=[], before=before, after=before) + + validator = EvolveValidator() + try: + for edit in edits: + validator.validate(edit, ruleset_by_id, policy) + except StrictValidationError as exc: + _record(journal_path, round_index, edits, "rejected", before, before, str(exc)) + return _outcome(round_index, accepted=False, reason=f"validation_failed: {exc}", edits=edits, before=before, after=before) + + candidate = edits_to_ledger(edits, current_ledger) + try: + _load_with(ruleset_dir, candidate) # replay smoke gate: candidate ruleset must boot + except Exception as exc: # noqa: BLE001 — any boot failure rejects the round + _record(journal_path, round_index, edits, "rejected", before, before, f"replay_failed: {exc}") + return _outcome(round_index, accepted=False, reason="replay_failed", edits=edits, before=before, after=before) + + after_findings, after_rules = _scan(target, ruleset_dir, candidate, policy) + after = _evaluate(target, manifest, after_findings, after_rules, policy) + + accepted = ( + after.recall >= recall_floor and after.fp_rate < fp_ceiling and after.score >= before.score and after.matched >= before.matched + ) + if accepted: + write_rule_ledger(ledger_path, candidate) + reason = "accepted" + else: + reason = "reverted" # ledger_path untouched -> byte-for-byte revert + _record(journal_path, round_index, edits, reason, before, after) + return _outcome(round_index, accepted=accepted, reason=reason, edits=edits, before=before, after=after) + + +def run_improvement( + target: Path, + manifest: BenchmarkManifest, + *, + ledger_path: Path, + journal_path: Path, + ruleset_dir: Path = DEFAULT_RULESET_DIR, + policy: dict[str, CwePolicy] | None = None, + max_rounds: int = 5, + recall_floor: float = 0.9, + fp_ceiling: float = 0.1, +) -> list[RoundOutcome]: + """Run improvement rounds until convergence (no new proposals) or ``max_rounds``.""" + outcomes: list[RoundOutcome] = [] + for _ in range(max_rounds): + outcome = run_round( + target, + manifest, + ledger_path=ledger_path, + journal_path=journal_path, + ruleset_dir=ruleset_dir, + policy=policy, + recall_floor=recall_floor, + fp_ceiling=fp_ceiling, + ) + outcomes.append(outcome) + if outcome.reason == "no_proposals": + break + return outcomes + + +@dataclass(frozen=True) +class _Eval: + result: BenchmarkResult + recall: float + fp_rate: float + score: int + matched: int + + +def _scan( + target: Path, ruleset_dir: Path, ledger: Mapping[str, dict[str, object]], policy: dict[str, CwePolicy] +) -> tuple[list[StaticFinding], tuple[PatternRule, ...]]: + rules = _load_with(ruleset_dir, ledger) + _, targets = preprocess_repository(target) + targets = attach_reachability_hints(targets, analyze_entry_points(target, targets)) + rankings = rank_targets(targets) + findings = quick_scan_findings(target, targets, rankings, rules, policy) + reported = [finding for finding in findings if finding.status != "shadow"] + return reported, rules + + +def _evaluate( + target: Path, manifest: BenchmarkManifest, findings: list[StaticFinding], rules: tuple[PatternRule, ...], policy: dict[str, CwePolicy] +) -> _Eval: + run = BenchmarkRun(benchmark_run_id="improve", root=target, manifest=manifest) + result = evaluate_benchmark(run=run, mode="quick", findings=findings, scan_id=None, scan_run_dir=None) + metrics = result.metrics + cwe_by_rule = {rule.rule_id: rule.cwe for rule in rules} + rule_cwe_by_id = {f.finding_id: cwe_by_rule.get(f.finding_id.split(":", 1)[0], "") for f in findings} + score = build_score_artifact(findings, rule_cwe_by_id, policy).project_score + recall = metrics.recall if metrics.recall is not None else 1.0 + fp_rate = metrics.false_positive_findings_total / metrics.actual_findings_total if metrics.actual_findings_total else 0.0 + return _Eval(result=result, recall=recall, fp_rate=fp_rate, score=score, matched=metrics.matched_findings_total) + + +def _load_with(ruleset_dir: Path, ledger: Mapping[str, dict[str, object]]) -> tuple[PatternRule, ...]: + import json + + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle: + handle.write(json.dumps(ledger)) + temp_path = Path(handle.name) + try: + return load_ruleset(ruleset_dir, temp_path) + finally: + temp_path.unlink(missing_ok=True) + + +def _write_signals(path: Path, signals: list[dict[str, object]]) -> None: + import json + + path.write_text(json.dumps(signals, indent=2, sort_keys=True) + "\n") + + +def _outcome(round_index: int, *, accepted: bool, reason: str, edits: list[RuleStatusEdit], before: _Eval, after: _Eval) -> RoundOutcome: + return RoundOutcome( + round=round_index, + accepted=accepted, + reason=reason, + edits=edits, + recall_before=before.recall, + recall_after=after.recall, + fp_before=before.fp_rate, + fp_after=after.fp_rate, + score_before=before.score, + score_after=after.score, + matched_before=before.matched, + matched_after=after.matched, + ) + + +def _record( + journal_path: Path, + round_index: int, + edits: list[RuleStatusEdit], + outcome: str, + before: _Eval, + after: _Eval, + reason: str = "", +) -> None: + append_round( + journal_path, + { + "round": round_index, + "outcome": outcome, + "reason": reason or outcome, + "edits": [ + {"key": e.key(), "lever": e.lever, "rule_id": e.rule_id, "from": e.from_status, "to": e.to_status, "rationale": e.rationale} + for e in edits + ], + "recall_before": round(before.recall, 4), + "recall_after": round(after.recall, 4), + "fp_before": round(before.fp_rate, 4), + "fp_after": round(after.fp_rate, 4), + "score_before": before.score, + "score_after": after.score, + }, + ) diff --git a/src/openultrasast/improve/journal.py b/src/openultrasast/improve/journal.py new file mode 100644 index 0000000..5910282 --- /dev/null +++ b/src/openultrasast/improve/journal.py @@ -0,0 +1,51 @@ +"""Improvement journal: per-round audit + novelty memory (task 7.3, 7.9). + +Persistent JSON record of every self-improvement round so each rule/policy edit is +attributable and reversible, and so a previously-reverted edit is not re-proposed +without a new rationale (the novelty gate). +""" + +from __future__ import annotations + +import json +from pathlib import Path + + +def load_journal(path: Path) -> list[dict[str, object]]: + if not path.exists(): + return [] + payload = json.loads(path.read_text()) + return payload if isinstance(payload, list) else [] + + +def write_journal(path: Path, rounds: list[dict[str, object]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(rounds, indent=2, sort_keys=True) + "\n") + + +def reverted_edit_keys(rounds: list[dict[str, object]]) -> set[str]: + """Edit keys from reverted rounds — blocked from re-proposal unless a new rationale appears.""" + keys: set[str] = set() + for entry in rounds: + if entry.get("outcome") != "reverted": + continue + for edit in _edits(entry): + key = edit.get("key") + if isinstance(key, str) and not edit.get("rationale"): + keys.add(key) + return keys + + +def next_round_index(rounds: list[dict[str, object]]) -> int: + return len(rounds) + 1 + + +def append_round(path: Path, entry: dict[str, object]) -> None: + rounds = load_journal(path) + rounds.append(entry) + write_journal(path, rounds) + + +def _edits(entry: dict[str, object]) -> list[dict[str, object]]: + edits = entry.get("edits") + return [edit for edit in edits if isinstance(edit, dict)] if isinstance(edits, list) else [] diff --git a/src/openultrasast/improve/validator.py b/src/openultrasast/improve/validator.py new file mode 100644 index 0000000..ba57b6b --- /dev/null +++ b/src/openultrasast/improve/validator.py @@ -0,0 +1,121 @@ +"""Bounded-lever edit model + validator for the self-improvement loop (tasks 7.1, 7.2). + +The loop may only adjust governance *data* through two bounded levers and never the +two authorities a human/upstream owns: detection **pattern text** and the 0-5 CWE +**severity**. Edits are typed so pattern/severity changes are structurally +impossible; the validator additionally enforces status/bounds/staging/resolution. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass + +from ..policy import CwePolicy +from ..ruleset import PatternRule +from ..ruleset.store import VALID_STATUS + +# The only levers the loop may pull (mirrors the design's journal._VALID_LEVERS). +VALID_LEVERS = frozenset({"rule", "policy"}) +# Policy constants the loop may tune; the 0-5 severity is intentionally absent. +TUNABLE_POLICY_CONSTANTS = frozenset({"K", "MIN_SCORE"}) + + +class StrictValidationError(ValueError): + """Raised when a proposed loop edit violates the bounded-lever contract.""" + + def __init__(self, message: str, *, kind: str = "rule_change") -> None: + super().__init__(message) + self.kind = kind + + +@dataclass(frozen=True) +class RuleStatusEdit: + """A `rule`-lever edit: change a rule's status / evidence floor / precision estimate.""" + + rule_id: str + from_status: str + to_status: str + rationale: str = "" + lever: str = "rule" + + def key(self) -> str: + return f"rule:{self.rule_id}:{self.from_status}->{self.to_status}" + + +@dataclass(frozen=True) +class PolicyConstantEdit: + """A `policy`-lever edit: tune a score constant (K / MIN_SCORE). Never severity.""" + + name: str + from_value: float + to_value: float + rationale: str = "" + lever: str = "policy" + + def key(self) -> str: + return f"policy:{self.name}:{self.from_value}->{self.to_value}" + + +@dataclass(frozen=True) +class EvolveBounds: + k_range: tuple[float, float] = (20.0, 200.0) + min_score_range: tuple[int, int] = (0, 100) + + +class EvolveValidator: + """Enforces the bounded-lever safety contract; raises StrictValidationError on any breach.""" + + def __init__(self, bounds: EvolveBounds | None = None) -> None: + self._bounds = bounds or EvolveBounds() + + def validate(self, edit: object, ruleset_by_id: Mapping[str, PatternRule], policy: Mapping[str, CwePolicy]) -> None: + if isinstance(edit, RuleStatusEdit): + self._validate_rule(edit, ruleset_by_id, policy) + elif isinstance(edit, PolicyConstantEdit): + self._validate_policy(edit) + else: + raise StrictValidationError(f"unknown edit type {type(edit).__name__}", kind="rule_change") + + def _validate_rule(self, edit: RuleStatusEdit, ruleset_by_id: Mapping[str, PatternRule], policy: Mapping[str, CwePolicy]) -> None: + if edit.to_status not in VALID_STATUS: + raise StrictValidationError(f"invalid status {edit.to_status!r} for {edit.rule_id}") + rule = ruleset_by_id.get(edit.rule_id) + if rule is None: + raise StrictValidationError(f"edit targets unknown rule_id {edit.rule_id!r} (loop cannot add rules)") + if edit.from_status == "enabled" and edit.to_status == "disabled": + raise StrictValidationError(f"{edit.rule_id}: enabled->disabled jump is forbidden; stage via shadow first") + if rule.cwe not in policy: + raise StrictValidationError(f"{edit.rule_id}: CWE {rule.cwe} does not resolve in the policy") + + def _validate_policy(self, edit: PolicyConstantEdit) -> None: + if edit.name not in TUNABLE_POLICY_CONSTANTS: + raise StrictValidationError( + f"policy constant {edit.name!r} is not loop-tunable (severity is upstream-owned)", kind="policy_change" + ) + if edit.name == "K" and not (self._bounds.k_range[0] <= edit.to_value <= self._bounds.k_range[1]): + raise StrictValidationError(f"K={edit.to_value} out of bounds {self._bounds.k_range}", kind="policy_change") + if edit.name == "MIN_SCORE" and not (self._bounds.min_score_range[0] <= edit.to_value <= self._bounds.min_score_range[1]): + raise StrictValidationError(f"MIN_SCORE={edit.to_value} out of bounds {self._bounds.min_score_range}", kind="policy_change") + + +def edits_to_ledger(edits: list[RuleStatusEdit], base: Mapping[str, dict[str, object]]) -> dict[str, dict[str, object]]: + """Overlay rule-status edits onto a copy of the base ledger.""" + ledger: dict[str, dict[str, object]] = {key: dict(value) for key, value in base.items()} + for edit in edits: + entry = dict(ledger.get(edit.rule_id, {})) + entry["status"] = edit.to_status + ledger[edit.rule_id] = entry + return ledger + + +__all__ = [ + "EvolveBounds", + "EvolveValidator", + "PolicyConstantEdit", + "RuleStatusEdit", + "StrictValidationError", + "TUNABLE_POLICY_CONSTANTS", + "VALID_LEVERS", + "edits_to_ledger", +] diff --git a/tests/test_improve.py b/tests/test_improve.py new file mode 100644 index 0000000..d2416c0 --- /dev/null +++ b/tests/test_improve.py @@ -0,0 +1,170 @@ +"""Self-improvement loop: bounded levers, validator, gates, accept/revert (Phase 3 task 7).""" + +import json +from pathlib import Path + +import pytest + +from openultrasast.benchmark import load_benchmark_manifest +from openultrasast.improve import ( + EvolveValidator, + PolicyConstantEdit, + RuleStatusEdit, + StrictValidationError, + build_rule_signals, + propose_status_edits, + run_improvement, + run_round, +) +from openultrasast.policy import load_policy +from openultrasast.ruleset import PatternRule, read_rule_ledger, write_ruleset + +POLICY = load_policy() + + +def _rule(rule_id: str, pattern: str, cwe: str = "CWE-95") -> PatternRule: + return PatternRule(rule_id=rule_id, title=rule_id, languages=("python",), cwe=cwe, tags=("injection",), pattern=pattern) + + +def _expected(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' + ) + + +# ---- validator (7.1, 7.2) --------------------------------------------------- + + +def test_validator_rejects_enabled_to_disabled_jump() -> None: + ruleset = {"r": _rule("r", r"\beval\s*\(")} + with pytest.raises(StrictValidationError, match="shadow"): + EvolveValidator().validate(RuleStatusEdit("r", "enabled", "disabled"), ruleset, POLICY) + + +def test_validator_accepts_enabled_to_shadow() -> None: + ruleset = {"r": _rule("r", r"\beval\s*\(")} + EvolveValidator().validate(RuleStatusEdit("r", "enabled", "shadow"), ruleset, POLICY) # no raise + + +def test_validator_rejects_unknown_rule_and_bad_status() -> None: + ruleset = {"r": _rule("r", r"\beval\s*\(")} + with pytest.raises(StrictValidationError, match="unknown rule_id"): + EvolveValidator().validate(RuleStatusEdit("nope", "enabled", "shadow"), ruleset, POLICY) + with pytest.raises(StrictValidationError): + EvolveValidator().validate(RuleStatusEdit("r", "enabled", "bogus"), ruleset, POLICY) + + +def test_validator_never_tunes_severity_and_bounds_constants() -> None: + with pytest.raises(StrictValidationError, match="severity is upstream-owned"): + EvolveValidator().validate(PolicyConstantEdit("severity", 3, 5), {}, POLICY) + with pytest.raises(StrictValidationError): + EvolveValidator().validate(PolicyConstantEdit("K", 60, 5000), {}, POLICY) # out of bounds + EvolveValidator().validate(PolicyConstantEdit("K", 60, 90), {}, POLICY) # in bounds, no raise + + +# ---- proposer + signals (7.4, 7.6) ----------------------------------------- + + +def test_propose_status_edits_auto_shadows_precision_draggers_only() -> None: + ruleset = {"good": _rule("good", r"\beval\s*\("), "noisy": _rule("noisy", r"\bprint\s*\(")} + per_rule = {"good": {"matched": 1, "missed": 0, "false_positives": 0}, "noisy": {"matched": 0, "missed": 0, "false_positives": 2}} + + edits = propose_status_edits(per_rule, ruleset, current_ledger={}, blocked_keys=set()) + assert [(e.rule_id, e.to_status) for e in edits] == [("noisy", "shadow")] + # novelty gate: a previously-reverted edit is not re-proposed. + assert propose_status_edits(per_rule, ruleset, {}, {edits[0].key()}) == [] + # already shadow -> not re-proposed. + assert propose_status_edits(per_rule, ruleset, {"noisy": {"status": "shadow"}}, set()) == [] + + +# ---- end-to-end loop (7.5) -------------------------------------------------- + + +def _manifest(tmp_path: Path, expected_block: str) -> tuple[Path, object]: + manifest_path = tmp_path / "bench.toml" + manifest_path.write_text('name = "t"\nlanguage = "python_web"\n\n[source]\npath = "repo"\n\n' + expected_block) + return manifest_path, load_benchmark_manifest(manifest_path) + + +def _ruleset_dir(tmp_path: Path, rules: list[PatternRule]) -> Path: + rd = tmp_path / "ruleset" / "python" + write_ruleset(rd / "rules.toml", rules) + return tmp_path / "ruleset" + + +def test_round_auto_shadows_noisy_rule_and_accepts(tmp_path: Path) -> None: + 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 = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")]) + _, manifest = _manifest(tmp_path, _expected("good-eval")) + ledger = tmp_path / "rule_policy.json" + journal = tmp_path / "journal.json" + + outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir) + + assert outcome.accepted and outcome.reason == "accepted" + assert outcome.fp_before > 0.0 and outcome.fp_after == 0.0 + assert outcome.recall_after == 1.0 # the true positive is preserved + assert read_rule_ledger(ledger)["noisy-print"]["status"] == "shadow" + assert json.loads(journal.read_text())[0]["outcome"] == "accepted" + + +def test_round_reverts_byte_for_byte_when_shadowing_would_drop_recall(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g(y):\n return eval(y)\n") + ruleset_dir = _ruleset_dir(tmp_path, [_rule("dual-eval", r"\beval\s*\(")]) + _, manifest = _manifest(tmp_path, _expected("dual-eval")) + ledger = tmp_path / "rule_policy.json" + journal = tmp_path / "journal.json" + + outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir) + + assert not outcome.accepted and outcome.reason == "reverted" + assert outcome.score_after > outcome.score_before # score improved but the recall gate blocks it + assert not ledger.exists() # byte-for-byte revert: the ledger was never written + assert json.loads(journal.read_text())[0]["outcome"] == "reverted" + + +def test_run_improvement_converges_after_accepting(tmp_path: Path) -> None: + 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 = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")]) + _, manifest = _manifest(tmp_path, _expected("good-eval")) + + outcomes = run_improvement( + repo, + manifest, + ledger_path=tmp_path / "rule_policy.json", + journal_path=tmp_path / "journal.json", + ruleset_dir=ruleset_dir, + max_rounds=4, + ) + + assert outcomes[0].accepted # round 1 shadows the noisy rule + assert outcomes[-1].reason == "no_proposals" # converges (nothing left to improve) + + +def test_build_rule_signals_separates_miss_and_fp() -> None: + from openultrasast.benchmark import BenchmarkFalsePositive, BenchmarkMetrics, BenchmarkMiss, BenchmarkResult + + result = BenchmarkResult( + benchmark_run_id="r", + benchmark_name="n", + language="python_web", + mode="quick", + scan_id=None, + scan_run_dir=None, + metrics=BenchmarkMetrics(1, 1, 0, 1, 1, 0.0, 0.0, None, {}, {}, {}), + misses=[BenchmarkMiss("CWE-89", "sqli", "a.py", "ev", "no match", rule_id="sqli")], + false_positives=[BenchmarkFalsePositive("noisy:a.py:1", "a.py", 1, "fp", rule_id="noisy")], + baseline_deltas=[], + calibration_records=[], + rule_recommendations=[], + ) + signals = build_rule_signals(result) + assert {s["signal"] for s in signals} == {"miss", "fp"} + assert {s["rule_id"] for s in signals} == {"sqli", "noisy"}