feat(benchmark): Phase 2 (task 4) — actionable per-rule signals, ledger, FP knob

Complete Phase 2: make benchmark output drive per-rule improvement, wire the
loop-owned ledger, and turn a confirmed false positive into a score knob. Zero
new dependencies; detection stays byte-identical (recall 44/47 = 93.6%, FP 0%)
and the 90/10 gate holds.

- Per-rule attribution (4.1): BenchmarkMiss/BenchmarkFalsePositive carry rule_id;
  BenchmarkMetrics.per_rule aggregates matched/missed/false_positives per rule.
- Recommendation delta (4.2): evaluate_benchmark emits RuleRecommendation
  (keep/shadow/loosen) written to rule_policy_delta.json; the inert
  next_improvement_candidate stub is now rule-aware ("rule:<id>" or
  "language_hunter").
- Loop ledger (4.3): cli loads the ruleset with the rule_policy.json overlay
  (empty -> byte-identical; an entry overlays a rule's status/floors at load);
  read_rule_ledger/write_rule_ledger added for the evolve loop. A disabled rule
  in the ledger no longer fires (covered by an integration test).
- FP -> reachability knob (4.4): a confirmed false-positive learning lowers a
  finding's effective reachability multiplier in the score
  (fp_reachability_overrides + scoring reachability_override) instead of deleting
  the rule, so recall stays high while the score stops over-penalizing.

93 tests pass; ruff + mypy clean. Spec tasks 4.1-4.4 checked off — Phase 2
(Core: rules-as-data + benchmark signals) is complete. The loop-driven decision
to apply ledger flips remains the gated evolve loop (Phase 3/4).

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-22T10:45:00Z",
+  "updated_at": "2026-06-22T13:45:00Z",
   "language": "en",
   "phase": "implementation",
   "approvals": {

.kiro/specs/harnessx-self-improving-rulesets/tasks.md

@@ -90,22 +90,22 @@ This plan splits the fused regex layer into three governed owners (rules detect,
   - _Boundary: PolicyScoringProcessor, RulesetStore, PolicyStore_
   - _Depends: 3.1, 3.2, 1.2_
 
-- [ ] 4. Make benchmark output an actionable per-rule signal
+- [x] 4. Make benchmark output an actionable per-rule signal
-- [ ] 4.1 Attribute misses and false positives to rules
+- [x] 4.1 Attribute misses and false positives to rules
   - Attribute every recorded miss and every recorded false positive to a rule identifier and aggregate per-rule precision and recall in the benchmark metrics.
   - Observable completion: a benchmark run reports per-rule precision/recall and every miss and false positive carries a rule identifier, verified by a metrics test.
   - _Requirements: 4.1_
   - _Boundary: Benchmark Harness_
   - _Depends: 3.1_
 
-- [ ] 4.2 Emit a per-rule recommendation delta artifact
+- [x] 4.2 Emit a per-rule recommendation delta artifact
   - Replace the inert improvement-candidate stub with a per-rule recommendation artifact recommending enable, disable, shadow, threshold, or evidence-floor changes per rule.
   - Observable completion: a benchmark run writes a recommendation artifact with per-rule suggested actions alongside its records, verified against the documented delta schema.
   - _Requirements: 4.2_
   - _Boundary: Benchmark Harness_
   - _Depends: 4.1_
 
-- [ ] 4.3 Introduce the loop-owned ledger and re-key calibration to rules
+- [x] 4.3 Introduce the loop-owned ledger and re-key calibration to rules
   - Add the loop-owned overlay ledger that records per-rule status, evidence floor, threshold, and precision, applied over the ruleset at load time.
   - Re-key calibration from the first-tag key to the rule identifier and write status flips into the ledger rather than only demoting ranking.
   - Observable completion: an empty ledger leaves behavior byte-identical to the prior phase while a ledger entry overlays a rule's status at load, proven by an integration test.
@@ -113,7 +113,7 @@ This plan splits the fused regex layer into three governed owners (rules detect,
   - _Boundary: RulesetStore, Calibration subsystem_
   - _Depends: 3.1, 3.2_
 
-- [ ] 4.4 Convert false-positive confirmation into a reachability knob
+- [x] 4.4 Convert false-positive confirmation into a reachability knob
   - When confirming a false positive, lower the finding's effective reachability multiplier rather than delete the rule, so recall stays high while the score stops over-penalizing.
   - Observable completion: confirming a false positive lowers the finding's effective reachability multiplier and the rule still fires, proven by a calibration test.
   - _Requirements: 5.6_

src/openultrasast/benchmark.py

@@ -60,6 +60,7 @@ class BenchmarkMiss:
     path: str
     evidence: str
     reason: str
+    rule_id: str | None = None
 
 
 @dataclass(frozen=True)
@@ -79,6 +80,7 @@ class BenchmarkCalibrationRecord:
     failed_stage: str
     next_improvement_candidate: str
     reason: str
+    rule_id: str | None = None
 
 
 @dataclass(frozen=True)
@@ -87,6 +89,17 @@ class BenchmarkFalsePositive:
     path: str
     line: int | None
     reason: str
+    rule_id: str = ""
+
+
+@dataclass(frozen=True)
+class RuleRecommendation:
+    rule_id: str
+    action: str  # keep | shadow | loosen
+    matched: int
+    missed: int
+    false_positives: int
+    rationale: str
 
 
 @dataclass(frozen=True)
@@ -101,6 +114,7 @@ class BenchmarkMetrics:
     runtime_seconds: float | None
     model_usage: dict[str, object]
     artifact_links: dict[str, str]
+    per_rule: dict[str, dict[str, int]]
 
 
 @dataclass(frozen=True)
@@ -123,6 +137,7 @@ class BenchmarkResult:
     false_positives: list[BenchmarkFalsePositive]
     baseline_deltas: list[BenchmarkBaselineDelta]
     calibration_records: list[BenchmarkCalibrationRecord]
+    rule_recommendations: list[RuleRecommendation]
 
 
 def load_benchmark_manifest(path: Path) -> BenchmarkManifest:
@@ -224,6 +239,8 @@ def evaluate_benchmark(
     false_positives = [_false_positive(finding) for index, finding in enumerate(findings) if index not in matched_finding_indexes]
     baseline_deltas = _baseline_deltas(run.manifest.expected, baseline_findings or [])
     calibration_records = [_calibration_record(miss) for miss in misses]
+    per_rule = _per_rule_metrics(findings, matched_finding_indexes, misses)
+    rule_recommendations = _rule_recommendations(per_rule)
     expected_total = len(run.manifest.expected)
     actual_total = len(findings)
     missed_total = len(misses)
@@ -249,11 +266,13 @@ def evaluate_benchmark(
             runtime_seconds=runtime_seconds,
             model_usage={},
             artifact_links=_artifact_links(scan_run_dir),
+            per_rule=per_rule,
         ),
         misses=misses,
         false_positives=false_positives,
         baseline_deltas=baseline_deltas,
         calibration_records=calibration_records,
+        rule_recommendations=rule_recommendations,
     )
 
 
@@ -266,6 +285,9 @@ def write_benchmark_artifacts(run: BenchmarkRun, manifest_path: Path, result: Be
     (run.root / "external_baseline_deltas.json").write_text(
         json.dumps(_jsonable([asdict(delta) for delta in result.baseline_deltas]), indent=2, sort_keys=True) + "\n"
     )
+    (run.root / "rule_policy_delta.json").write_text(
+        json.dumps(_jsonable([asdict(rec) for rec in result.rule_recommendations]), indent=2, sort_keys=True) + "\n"
+    )
 
 
 def _matching_finding_index(expected: ExpectedFinding, findings: list[StaticFinding], used_indexes: set[int]) -> int | None:
@@ -301,6 +323,7 @@ def _miss(expected: ExpectedFinding) -> BenchmarkMiss:
         path=expected.path,
         evidence=expected.evidence,
         reason="no OpenUltraSAST finding matched the expected benchmark vulnerability",
+        rule_id=expected.rule_id,
     )
 
 
@@ -310,9 +333,57 @@ def _false_positive(finding: StaticFinding) -> BenchmarkFalsePositive:
         path=finding.path,
         line=finding.line,
         reason="finding did not match any expected benchmark vulnerability",
+        rule_id=_rule_id_of(finding.finding_id),
     )
 
 
+def _rule_id_of(finding_id: str) -> str:
+    return finding_id.split(":", 1)[0]
+
+
+def _per_rule_metrics(
+    findings: list[StaticFinding],
+    matched_finding_indexes: set[int],
+    misses: list[BenchmarkMiss],
+) -> dict[str, dict[str, int]]:
+    per_rule: dict[str, dict[str, int]] = {}
+
+    def bucket(rule_id: str) -> dict[str, int]:
+        return per_rule.setdefault(rule_id, {"matched": 0, "missed": 0, "false_positives": 0})
+
+    for index, finding in enumerate(findings):
+        rule_id = _rule_id_of(finding.finding_id)
+        key = "matched" if index in matched_finding_indexes else "false_positives"
+        bucket(rule_id)[key] += 1
+    for miss in misses:
+        if miss.rule_id:
+            bucket(miss.rule_id)["missed"] += 1
+    return dict(sorted(per_rule.items()))
+
+
+def _rule_recommendations(per_rule: dict[str, dict[str, int]]) -> list[RuleRecommendation]:
+    recommendations: list[RuleRecommendation] = []
+    for rule_id, counts in per_rule.items():
+        matched, missed, fps = counts["matched"], counts["missed"], counts["false_positives"]
+        if fps > 0:
+            action, rationale = "shadow", "rule produced false positives; stage it before it drags precision"
+        elif missed > 0:
+            action, rationale = "loosen", "rule missed expected findings; widen or lower its evidence floor"
+        else:
+            action, rationale = "keep", "rule matched expected findings with no false positives"
+        recommendations.append(
+            RuleRecommendation(
+                rule_id=rule_id,
+                action=action,
+                matched=matched,
+                missed=missed,
+                false_positives=fps,
+                rationale=rationale,
+            )
+        )
+    return recommendations
+
+
 def _baseline_deltas(expected_findings: list[ExpectedFinding], baseline_findings: list[BaselineFinding]) -> list[BenchmarkBaselineDelta]:
     tools = sorted({finding.tool for finding in baseline_findings})
     deltas: list[BenchmarkBaselineDelta] = []
@@ -359,14 +430,18 @@ def _baseline_matches_expected(expected: ExpectedFinding, finding: BaselineFindi
 
 
 def _calibration_record(miss: BenchmarkMiss) -> BenchmarkCalibrationRecord:
+    # A miss tied to a known rule points the loop at that rule; an unattributed
+    # miss (no rule fired for the class) is a coverage gap for a language hunter.
+    candidate = f"rule:{miss.rule_id}" if miss.rule_id else "language_hunter"
     return BenchmarkCalibrationRecord(
         cwe=miss.cwe,
         vulnerability_class=miss.vulnerability_class,
         path=miss.path,
         evidence=miss.evidence,
         failed_stage="benchmark_ground_truth_matching",
-        next_improvement_candidate="rules_or_language_hunter",
+        next_improvement_candidate=candidate,
         reason=miss.reason,
+        rule_id=miss.rule_id,
     )
 
 

src/openultrasast/calibration.py

@@ -249,6 +249,30 @@ def learnings_from_verifications(
     return learnings
 
 
+# A confirmed false positive lowers a finding's effective reachability multiplier
+# by this factor in the project score (it does not delete the rule).
+FP_REACHABILITY_FACTOR = 0.4
+_BASE_REACH_MULT = {"reachable": 1.0, "inferred-file-surface": 0.6, "unknown": 0.4}
+
+
+def fp_reachability_overrides(
+    findings: list[StaticFinding],
+    learnings: list[FalsePositiveLearning],
+) -> dict[str, float]:
+    """Lowered score multipliers for findings whose scope has a confirmed FP.
+
+    Keyed by ``finding_id``; consumed by the scoring model so a confirmed false
+    positive reduces the score penalty instead of the rule being deleted.
+    """
+    overrides: dict[str, float] = {}
+    for finding in findings:
+        vulnerability_class = _vulnerability_class(finding)
+        if any(_learning_applies(finding.path, learning) and learning.vulnerability_class == vulnerability_class for learning in learnings):
+            base = _BASE_REACH_MULT.get(finding.reachability_status, _BASE_REACH_MULT["unknown"])
+            overrides[finding.finding_id] = round(base * FP_REACHABILITY_FACTOR, 3)
+    return overrides
+
+
 def _reason_from_verification(finding: StaticFinding, status: VerificationStatus) -> FalsePositiveReason:
     if finding.reachability_status == "unknown":
         return FalsePositiveReason.UNREACHABLE_PATH

src/openultrasast/cli.py

@@ -18,6 +18,7 @@ from .benchmark import (
 from .calibration import (
     FalsePositiveLearning,
     calibrate_rankings,
+    fp_reachability_overrides,
     learnings_from_verifications,
     load_false_positive_learnings,
     merge_false_positive_learnings,
@@ -105,7 +106,9 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     write_harness_config(config=config, processors=[], contract_mode="strict", path=run.root / "harness.json")
     runtime.start(mode=mode, target=run.target)
     policy = runtime.run_stage("policy_load", load_policy)
-    ruleset = runtime.run_stage("ruleset_load", lambda: load_ruleset(DEFAULT_RULESET_DIR))
+    ledger_path = run.target / CALIBRATION_DIR / "false_positive_learnings.json"
+    rule_ledger_path = run.target / CALIBRATION_DIR / "rule_policy.json"
+    ruleset = runtime.run_stage("ruleset_load", lambda: load_ruleset(DEFAULT_RULESET_DIR, rule_ledger_path))
     runtime.run_stage("policy_check", lambda: assert_rules_resolve(ruleset, policy))
     static_hints = runtime.run_stage("static_mapping", lambda: _load_static_hints(config.static_analysis.sarif_paths))
     write_static_hints(static_hints, run.root / "mapping" / "static_hints.json")
@@ -118,7 +121,6 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     targets = attach_reachability_hints(targets, entry_points)
     write_preprocess_artifact(snapshot, targets, run.root / "preprocess" / "file_targets.json")
     rankings = runtime.run_stage("rank", lambda: rank_targets(targets))
-    ledger_path = run.target / CALIBRATION_DIR / "false_positive_learnings.json"
     prior_learnings = load_false_positive_learnings(ledger_path)
     rankings, calibrations = runtime.run_stage("calibrate", lambda: calibrate_rankings(rankings, prior_learnings))
     applied_calibrations = [calibration for calibration in calibrations if calibration.applied_learning_ids]
@@ -168,6 +170,9 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     )
     cwe_by_rule = {rule.rule_id: rule.cwe for rule in ruleset}
     rule_cwe_by_id = {finding.finding_id: cwe_by_rule.get(finding.finding_id.split(":", 1)[0], "") for finding in findings}
+    # A confirmed false positive (prior-scan learning) lowers a finding's effective
+    # reachability multiplier in the score instead of deleting the rule.
+    reachability_override = fp_reachability_overrides(findings, prior_learnings)
     score_artifact = runtime.run_stage(
         "score",
         lambda: build_score_artifact(
@@ -178,6 +183,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
             min_score=config.score.min_score,
             block_severity_reachable=config.score.block_severity_reachable,
             blocking=config.score.blocking,
+            reachability_override=reachability_override,
         ),
     )
     score_path = run.root / "score.json"

src/openultrasast/ruleset/__init__.py

@@ -5,6 +5,8 @@ from .store import (
     PatternRule,
     RulesetError,
     load_ruleset,
+    read_rule_ledger,
+    write_rule_ledger,
     write_ruleset,
 )
 
@@ -13,5 +15,7 @@ __all__ = [
     "PatternRule",
     "RulesetError",
     "load_ruleset",
+    "read_rule_ledger",
+    "write_rule_ledger",
     "write_ruleset",
 ]

src/openultrasast/ruleset/store.py

@@ -81,6 +81,17 @@ def write_ruleset(path: Path, rules: Iterable[PatternRule]) -> None:
     path.write_text("\n\n".join(blocks) + "\n")
 
 
+def read_rule_ledger(path: Path) -> dict[str, dict[str, object]]:
+    """Read the loop-owned ``rule_policy.json`` overlay (status/floors per rule_id)."""
+    return _load_ledger(path)
+
+
+def write_rule_ledger(path: Path, entries: dict[str, dict[str, object]]) -> None:
+    """Write the loop-owned ``rule_policy.json`` overlay (used by the evolve loop)."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(entries, indent=2, sort_keys=True) + "\n")
+
+
 def _rule_from_dict(item: dict[str, object], path: Path) -> PatternRule:
     try:
         rule_id = str(item["rule_id"])

src/openultrasast/scoring/project_score.py

@@ -29,15 +29,26 @@ class ScoreArtifact:
         return asdict(self)
 
 
-def finding_penalty(finding: StaticFinding, rule_cwe: str | None, policy: Mapping[str, CwePolicy]) -> float:
+def finding_penalty(
+    finding: StaticFinding,
+    rule_cwe: str | None,
+    policy: Mapping[str, CwePolicy],
+    reachability_override: Mapping[str, float] | None = None,
+) -> float:
     """Penalty contributed by one finding: severity weight × reachability multiplier.
 
-    Dynamic-only or unmapped CWEs contribute zero (report-only, never scored).
+    Dynamic-only or unmapped CWEs contribute zero (report-only, never scored). A
+    confirmed false positive can supply a lowered effective reachability multiplier
+    via ``reachability_override`` (keyed by ``finding_id``) instead of deleting the rule.
     """
     pol = policy.get(rule_cwe) if rule_cwe else None
     if pol is None or not pol.static:
         return 0.0
-    return SEV_WEIGHT[pol.severity] * REACH_MULT.get(finding.reachability_status, REACH_MULT["unknown"])
+    if reachability_override and finding.finding_id in reachability_override:
+        multiplier = reachability_override[finding.finding_id]
+    else:
+        multiplier = REACH_MULT.get(finding.reachability_status, REACH_MULT["unknown"])
+    return SEV_WEIGHT[pol.severity] * multiplier
 
 
 def project_score(
@@ -45,13 +56,14 @@ def project_score(
     rule_cwe_by_id: Mapping[str, str],
     policy: Mapping[str, CwePolicy],
     k: float = 60.0,
+    reachability_override: Mapping[str, float] | None = None,
 ) -> int:
     """Compute the 0-100 project score (shadow findings excluded by the caller).
 
     Exponential decay of total penalty: no penalty → 100; one severity-5 reachable
     finding → ~43.
     """
-    total = sum(finding_penalty(finding, rule_cwe_by_id.get(finding.finding_id), policy) for finding in findings)
+    total = sum(finding_penalty(finding, rule_cwe_by_id.get(finding.finding_id), policy, reachability_override) for finding in findings)
     return round(100 * math.exp(-total / k))
 
 
@@ -92,8 +104,9 @@ def build_score_artifact(
     min_score: int = 80,
     block_severity_reachable: int = 5,
     blocking: bool = False,
+    reachability_override: Mapping[str, float] | None = None,
 ) -> ScoreArtifact:
-    score = project_score(findings, rule_cwe_by_id, policy, k=k)
+    score = project_score(findings, rule_cwe_by_id, policy, k=k, reachability_override=reachability_override)
     penalty_total = 0.0
     max_severity = 0
     by_category: dict[str, int] = {}
@@ -109,7 +122,7 @@ def build_score_artifact(
             if pol.dynamic:
                 out_of_scope_dynamic_only += 1
             continue
-        penalty_total += finding_penalty(finding, cwe, policy)
+        penalty_total += finding_penalty(finding, cwe, policy, reachability_override)
         max_severity = max(max_severity, pol.severity)
         by_category[pol.flaw_category] = by_category.get(pol.flaw_category, 0) + 1
     return ScoreArtifact(

tests/test_benchmark.py

@@ -322,15 +322,82 @@ evidence = "memcpy copies attacker-controlled input"
             "cwe": "CWE-120",
             "evidence": "memcpy copies attacker-controlled input",
             "failed_stage": "benchmark_ground_truth_matching",
-            "next_improvement_candidate": "rules_or_language_hunter",
+            "next_improvement_candidate": "rule:c-unsafe-memory-copy",
             "path": "src/vuln.cpp",
             "reason": "no OpenUltraSAST finding matched the expected benchmark vulnerability",
+            "rule_id": "c-unsafe-memory-copy",
             "vulnerability_class": "unsafe memory copy",
         }
     ]
     assert deltas == [{"extra_findings_total": 1, "matched_expected_total": 1, "missed_expected_total": 0, "tool": "clearwing"}]
 
 
+def test_benchmark_attributes_per_rule_signals_and_recommendations(tmp_path: Path) -> None:
+    manifest_path = tmp_path / "benchmark.toml"
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    manifest_path.write_text(
+        """
+name = "cpp-smoke"
+language = "c_cpp"
+
+[source]
+path = "repo"
+
+[[expected]]
+cwe = "CWE-120"
+class = "buffer overflow"
+path = "src/vuln.cpp"
+line = 9
+rule_id = "c-unsafe-memory-copy"
+sink = "memcpy"
+evidence = "memcpy copies attacker-controlled input"
+
+[[expected]]
+cwe = "CWE-78"
+class = "command injection"
+path = "src/vuln.cpp"
+line = 20
+rule_id = "c-shell-exec"
+sink = "system"
+evidence = "system runs attacker input"
+""".strip()
+        + "\n"
+    )
+    manifest = load_benchmark_manifest(manifest_path)
+    run = create_benchmark_run(repo, manifest)
+    findings = [
+        _finding(
+            finding_id="c-unsafe-memory-copy:src/vuln.cpp:9",
+            path="src/vuln.cpp",
+            line=9,
+            rationale="Static pattern c-unsafe-memory-copy matched memcpy with attacker-controlled input.",
+        ),
+        _finding(
+            finding_id="js-eval:src/app.js:3",
+            path="src/app.js",
+            line=3,
+            rationale="Static pattern js-eval matched eval.",
+        ),
+    ]
+
+    result = evaluate_benchmark(run=run, mode="quick", findings=findings, scan_id="scan-1", scan_run_dir=repo / ".runs" / "scan-1")
+    write_benchmark_artifacts(run, manifest_path, result)
+
+    per_rule = result.metrics.per_rule
+    assert per_rule["c-unsafe-memory-copy"] == {"matched": 1, "missed": 0, "false_positives": 0}
+    assert per_rule["c-shell-exec"]["missed"] == 1
+    assert per_rule["js-eval"]["false_positives"] == 1
+
+    actions = {rec.rule_id: rec.action for rec in result.rule_recommendations}
+    assert actions["c-unsafe-memory-copy"] == "keep"
+    assert actions["c-shell-exec"] == "loosen"
+    assert actions["js-eval"] == "shadow"
+
+    delta = json.loads((run.root / "rule_policy_delta.json").read_text())
+    assert {item["rule_id"] for item in delta} == {"c-unsafe-memory-copy", "c-shell-exec", "js-eval"}
+
+
 def _finding(
     *,
     finding_id: str,

tests/test_ruleset.py

@@ -0,0 +1,67 @@
+"""Rules-as-data store and the loop-owned ledger overlay (Phase 2 tasks 3.1, 4.3)."""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from openultrasast.cli import main
+from openultrasast.ruleset import (
+    DEFAULT_RULESET_DIR,
+    PatternRule,
+    load_ruleset,
+    read_rule_ledger,
+    write_rule_ledger,
+    write_ruleset,
+)
+
+
+def test_default_ruleset_loads_as_data() -> None:
+    rules = load_ruleset(DEFAULT_RULESET_DIR)
+    assert len(rules) == 32
+    assert all(rule.status == "enabled" for rule in rules)
+    assert all(not hasattr(rule, "severity") for rule in rules)  # severity is policy-governed, not rule-local
+
+
+def test_empty_ledger_is_byte_identical() -> None:
+    assert load_ruleset(DEFAULT_RULESET_DIR) == load_ruleset(DEFAULT_RULESET_DIR, None)
+
+
+def test_ledger_overlay_changes_only_targeted_rule(tmp_path: Path) -> None:
+    ledger = tmp_path / "rule_policy.json"
+    ledger.write_text(json.dumps({"python-unsafe-eval": {"status": "disabled", "precision_estimate": 0.3}}))
+
+    rules = {rule.rule_id: rule for rule in load_ruleset(DEFAULT_RULESET_DIR, ledger)}
+
+    assert rules["python-unsafe-eval"].status == "disabled"
+    assert rules["python-unsafe-eval"].precision_estimate == 0.3
+    assert rules["c-shell-exec"].status == "enabled"  # untargeted rule untouched
+
+
+def test_rule_ledger_round_trips(tmp_path: Path) -> None:
+    ledger = tmp_path / "rule_policy.json"
+    entries = {"python-unsafe-eval": {"status": "shadow", "precision_estimate": 0.41}}
+    write_rule_ledger(ledger, entries)
+    assert read_rule_ledger(ledger) == entries
+
+
+def test_write_ruleset_round_trips(tmp_path: Path) -> None:
+    rule = PatternRule(rule_id="x-demo", title="Demo", languages=("python",), cwe="CWE-89", tags=("injection",), pattern=r"\bfoo\(")
+    write_ruleset(tmp_path / "r" / "rules.toml", [rule])
+    assert load_ruleset(tmp_path / "r")[0] == rule
+
+
+def test_disabled_rule_via_ledger_does_not_fire_in_scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text("def f(x):\n    return eval(x)\n")
+    ledger_dir = repo / ".openultrasast" / "calibration"
+    ledger_dir.mkdir(parents=True)
+    (ledger_dir / "rule_policy.json").write_text(json.dumps({"python-unsafe-eval": {"status": "disabled"}}))
+    monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
+
+    assert main(["scan", str(repo), "--mode", "quick"]) == 0
+
+    run_dir = sorted((repo / ".runs").iterdir())[-1]
+    findings = json.loads((run_dir / "findings.json").read_text())["findings"]
+    assert all(not finding["finding_id"].startswith("python-unsafe-eval:") for finding in findings)

tests/test_scoring.py

@@ -1,3 +1,4 @@
+from openultrasast.calibration import FalsePositiveReason, fp_reachability_overrides, record_false_positive_learning
 from openultrasast.findings import StaticFinding
 from openultrasast.policy import CwePolicy
 from openultrasast.scoring import build_score_artifact, gate, project_score
@@ -62,6 +63,38 @@ def test_hard_gate_fails_on_severity5_reachable_only() -> None:
     assert ok["passed"] is True  # inferred reachability does not trip the hard gate
 
 
+def test_confirmed_false_positive_lowers_penalty_without_deleting_rule() -> None:
+    finding = StaticFinding(
+        finding_id="python-os-command:app.py:1",
+        path="app.py",
+        title="t",
+        severity="critical",
+        confidence="medium",
+        evidence_level="static_corroboration",
+        rationale="r",
+        line=1,
+        function_name=None,
+        reachability_status="reachable",
+        reachability_evidence=[],
+        reachability_conditions=[],
+        tags=["syscall_entry"],
+        ranking_priority=3.0,
+    )
+    learning = record_false_positive_learning(
+        finding,
+        reason=FalsePositiveReason.MISSING_ATTACKER_CONTROL,
+        evidence="input is a constant, not attacker-controlled",
+    )
+    overrides = fp_reachability_overrides([finding], [learning])
+    rule_cwe = {finding.finding_id: "CWE-78"}
+
+    assert overrides[finding.finding_id] < 1.0  # reachable multiplier lowered from 1.0
+    base = project_score([finding], rule_cwe, POLICY)
+    with_fp = project_score([finding], rule_cwe, POLICY, reachability_override=overrides)
+    assert with_fp > base  # confirmed FP reduces the penalty, raising the score
+    assert finding.finding_id in rule_cwe  # the rule still fires; it was not deleted
+
+
 def test_score_gate_blocks_only_when_blocking_enabled() -> None:
     finding = _finding("sql:a.py:1", "inferred-file-surface")
     rule_cwe = {"sql:a.py:1": "CWE-89"}