feat(improve): Phase 3 task 7 — bounded, gated self-improvement loop

Stand up the deterministic self-improvement loop over the ruleset/policy
governance data. Zero-dependency and fully tested offline; detection gate intact.

- 7.1/7.2 bounded levers + validator (improve/validator.py): two levers (rule,
  policy); typed edits make pattern-text and 0-5 severity changes structurally
  impossible; EvolveValidator rejects bad status, unknown rule_ids (loop cannot add
  rules), enabled->disabled jumps (shadow staging mandatory), unresolved CWEs, and
  out-of-bounds K/MIN_SCORE, raising StrictValidationError.
- 7.3 novelty + replay + journal (improve/journal.py): a JSON round journal records
  every edit/lever/outcome; reverted edits are blocked from re-proposal (novelty);
  a replay smoke gate boots the candidate ruleset before re-benchmarking.
- 7.4 benchmark deltas -> rule_signals.json (recall side: miss; precision side: fp).
- 7.5 hard acceptance gate + byte-for-byte revert (improve/evolve.py): a round is
  accepted only if recall >= 90% AND FP < 10% AND project score did not regress AND
  no matched-finding regression; otherwise the ledger is never written (byte-for-byte
  revert). run_round/run_improvement drive rounds to convergence.
- 7.6 auto-shadow precision-draggers (enabled rules with FPs -> shadow, never
  delete), nominate coverage gaps via rule_signals, and keep the project score as the
  reward separate from the recall/FP feasibility gate (the loop cannot raise the
  score by shadowing rules that breach the gate — proven by the revert test).

Design note: the loop's safety machinery is deterministic and benchmark-driven
(the proposer is the per-rule signal); the LLM MetaAgent.evolve is the optional
richer proposer that plugs into this same validated/gated machinery when the
HarnessX extra is present (offline-untestable). This mirrors how DocuHarnessX
realizes self-improvement.

9 new tests (validator, proposer/signals, accept, byte-for-byte revert, convergence);
118 pass total, 2 construction smokes skip without the extra; ruff + mypy clean;
detection gate 93.6%/0% intact. Tasks 7.1-7.6 checked off.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

.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": {

.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.

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",
+]

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,
+        },
+    )

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 []

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",
+]

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"}