feat(three-stage-scan): overlay complexity ledger without hiding sev-5

.kiro/specs/three-stage-scan/tasks.md

@@ -31,7 +31,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
 
 ## Phase 2 — Core: Map, hunter tools, sandbox (parallel after foundation)
 
-- [ ] 2. Build the heuristic complexity map and test-tuning hints
+- [x] 2. Build the heuristic complexity map and test-tuning hints
 - [x] 2.1 (P) Compute per-file and per-function complexity signals
   - Signals include size, nesting, surface tags, reachability, inventory-hit density as a feature not the score, and whether an adjacent test appears to exist.
   - Function identity uses entry-point names when present and otherwise remains empty.

src/openultrasast/complexity/__init__.py

@@ -1,6 +1,16 @@
 """Heuristic complexity map for stage 2."""
 
 from .hints import TestHint, attach_test_hints
+from .ledger import (
+    LedgerEntry,
+    apply_overlay,
+    hotspot_key,
+    load_ledger,
+    must_keep_as_candidate,
+    record_verdict,
+    select_forced_candidates,
+    write_ledger,
+)
 from .map import ComplexityMap, Hotspot, build_complexity_map, write_complexity_map
 from .signals import ComplexitySignals, collect_signals
 
@@ -8,9 +18,17 @@ __all__ = [
     "ComplexityMap",
     "ComplexitySignals",
     "Hotspot",
+    "LedgerEntry",
     "TestHint",
+    "apply_overlay",
     "attach_test_hints",
     "build_complexity_map",
     "collect_signals",
+    "hotspot_key",
+    "load_ledger",
+    "must_keep_as_candidate",
+    "record_verdict",
+    "select_forced_candidates",
     "write_complexity_map",
+    "write_ledger",
 ]

src/openultrasast/complexity/ledger.py

@@ -0,0 +1,175 @@
+"""Loop-owned complexity ledger: overlay hotspot scores without hiding sev-5 inventory."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass, replace
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from ..findings import StaticFinding
+
+if TYPE_CHECKING:
+    from .map import Hotspot
+
+TRIGGERABLE_DELTA = 1.5
+NOT_TRIGGERABLE_DELTA = -1.5
+VERDICT_DELTAS = {
+    "triggerable": TRIGGERABLE_DELTA,
+    "not_triggerable": NOT_TRIGGERABLE_DELTA,
+}
+FORCED_CANDIDATE_SEVERITY = 5
+FORCED_CANDIDATE_REACHABILITY = frozenset({"reachable", "inferred-file-surface"})
+
+
+@dataclass(frozen=True)
+class LedgerEntry:
+    score_delta: float
+    last_verdict: str
+    round: int = 0
+
+
+def hotspot_key(path: str, function_name: str | None) -> str:
+    return f"{path}:{function_name or ''}"
+
+
+def load_ledger(path: Path | None) -> dict[str, LedgerEntry]:
+    if path is None or not path.exists():
+        return {}
+    payload = json.loads(path.read_text())
+    entries: dict[str, LedgerEntry] = {}
+    for key, value in payload.items():
+        entry = _coerce_entry(value)
+        if entry is not None:
+            entries[str(key)] = entry
+    return entries
+
+
+def write_ledger(path: Path, ledger: Mapping[str, LedgerEntry | Mapping[str, object]]) -> None:
+    payload: dict[str, dict[str, object]] = {}
+    for key, value in ledger.items():
+        entry = _coerce_entry(value)
+        if entry is None:
+            continue
+        payload[str(key)] = asdict(entry)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+
+
+def record_verdict(
+    ledger: Mapping[str, LedgerEntry | Mapping[str, object]],
+    *,
+    path: str,
+    function_name: str | None,
+    verdict: str,
+) -> dict[str, LedgerEntry]:
+    updated: dict[str, LedgerEntry] = {}
+    for key, value in ledger.items():
+        entry = _coerce_entry(value)
+        if entry is not None:
+            updated[str(key)] = entry
+    key = hotspot_key(path, function_name)
+    previous = updated.get(key)
+    updated[key] = LedgerEntry(
+        score_delta=round((previous.score_delta if previous else 0.0) + VERDICT_DELTAS.get(verdict, 0.0), 4),
+        last_verdict=verdict,
+        round=(previous.round if previous else 0) + 1,
+    )
+    return updated
+
+
+def apply_overlay(
+    hotspots: Sequence[Hotspot],
+    ledger: Mapping[str, LedgerEntry | Mapping[str, object]],
+) -> tuple[Hotspot, ...]:
+    if not ledger:
+        return tuple(hotspots)
+    adjusted: list[Hotspot] = []
+    for hotspot in hotspots:
+        entry = _coerce_entry(ledger.get(hotspot_key(hotspot.path, hotspot.function_name)))
+        if entry is None or entry.score_delta == 0.0:
+            adjusted.append(hotspot)
+            continue
+        score = round(hotspot.score + entry.score_delta, 4)
+        signals = dict(hotspot.signals)
+        signals["ledger_delta"] = entry.score_delta
+        adjusted.append(
+            replace(
+                hotspot,
+                score=score,
+                band=_band_for(score),
+                signals=signals,
+                rationale=(
+                    f"{hotspot.rationale} ledger overlay last_verdict={entry.last_verdict} "
+                    f"score_delta={entry.score_delta} (inventory retained)."
+                ),
+            )
+        )
+    return tuple(sorted(adjusted, key=lambda item: (-item.score, item.path, item.function_name or "")))
+
+
+def must_keep_as_candidate(
+    finding: StaticFinding,
+    policy_severity: int | object,
+    reachability: str | None = None,
+) -> bool:
+    status = finding.reachability_status if reachability is None else reachability
+    return _as_severity(policy_severity) >= FORCED_CANDIDATE_SEVERITY and status in FORCED_CANDIDATE_REACHABILITY
+
+
+def select_forced_candidates(
+    findings: Sequence[StaticFinding],
+    policy: Mapping[str, object],
+    *,
+    rule_cwe: Mapping[str, str] | None = None,
+) -> tuple[StaticFinding, ...]:
+    cwe_by_id = rule_cwe or {}
+    return tuple(
+        finding
+        for finding in findings
+        if must_keep_as_candidate(finding, _policy_severity_for(finding, policy, cwe_by_id), finding.reachability_status)
+    )
+
+
+def _policy_severity_for(finding: StaticFinding, policy: Mapping[str, object], cwe_by_id: Mapping[str, str]) -> int:
+    if finding.finding_id in policy:
+        return _as_severity(policy[finding.finding_id])
+    cwe = cwe_by_id.get(finding.finding_id)
+    if cwe is not None and cwe in policy:
+        return _as_severity(policy[cwe])
+    return 0
+
+
+def _as_severity(value: object) -> int:
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, int):
+        return value
+    severity = getattr(value, "severity", None)
+    if isinstance(severity, bool) or not isinstance(severity, int):
+        return 0
+    return severity
+
+
+def _coerce_entry(value: object) -> LedgerEntry | None:
+    if isinstance(value, LedgerEntry):
+        return value
+    if not isinstance(value, Mapping):
+        return None
+    raw_delta = value.get("score_delta")
+    if isinstance(raw_delta, bool) or not isinstance(raw_delta, int | float):
+        return None
+    raw_round = value.get("round", 0)
+    round_n = raw_round if isinstance(raw_round, int) and not isinstance(raw_round, bool) else 0
+    return LedgerEntry(
+        score_delta=float(raw_delta),
+        last_verdict=str(value.get("last_verdict", "")),
+        round=round_n,
+    )
+
+
+def _band_for(score: float) -> str:
+    from .map import _band
+
+    return _band(score)

src/openultrasast/complexity/map.py

@@ -8,6 +8,7 @@ from pathlib import Path
 from ..findings import StaticFinding
 from ..preprocess import FileTarget
 from .hints import TestHint, attach_test_hints
+from .ledger import apply_overlay, load_ledger
 from .signals import ComplexitySignals, collect_signals
 
 # Score cutoffs for hotspot bands; inventory density is not the rank driver.
@@ -40,11 +41,14 @@ def build_complexity_map(
     *,
     repo_files: Iterable[str] = (),
     sources: Mapping[str, str] | None = None,
+    ledger_path: Path | None = None,
 ) -> ComplexityMap:
     collected = collect_signals(targets, findings, repo_files=repo_files, sources=sources)
     ids_by_path = _inventory_ids_by_path(findings)
     hotspots = tuple(_hotspot_from_signals(item, ids_by_path.get(item.path, ())) for item in collected)
     hotspots = attach_test_hints(hotspots, targets, repo_files=repo_files, sources=sources)
+    if ledger_path is not None:
+        hotspots = apply_overlay(hotspots, load_ledger(ledger_path))
     complexity_map = ComplexityMap(hotspots=hotspots, heuristic_only=True)
     write_complexity_map(complexity_map, output_path)
     return complexity_map

tests/test_complexity_ledger.py

@@ -0,0 +1,260 @@
+import json
+from pathlib import Path
+
+from openultrasast.complexity.ledger import (
+    NOT_TRIGGERABLE_DELTA,
+    TRIGGERABLE_DELTA,
+    LedgerEntry,
+    apply_overlay,
+    hotspot_key,
+    load_ledger,
+    must_keep_as_candidate,
+    record_verdict,
+    select_forced_candidates,
+    write_ledger,
+)
+from openultrasast.complexity.map import Hotspot, build_complexity_map
+from openultrasast.findings import StaticFinding
+from openultrasast.policy import CwePolicy
+from openultrasast.preprocess import FileTarget
+
+
+def _target(
+    path: str,
+    *,
+    loc: int,
+    tags: list[str] | None = None,
+    language: str = "python",
+    hints: list[dict[str, object]] | None = None,
+    has_fuzz_entry_point: bool = False,
+) -> FileTarget:
+    return FileTarget(
+        path=path,
+        absolute_path=f"/repo/{path}",
+        language=language,
+        loc=loc,
+        tags=list(tags or []),
+        has_fuzz_entry_point=has_fuzz_entry_point,
+        reachability_hints=list(hints or []),
+    )
+
+
+def _hit(
+    path: str,
+    finding_id: str,
+    *,
+    function_name: str | None = None,
+    reachability: str = "unknown",
+    severity: str = "low",
+    rationale: str = "inventory",
+) -> StaticFinding:
+    return StaticFinding(
+        finding_id=finding_id,
+        path=path,
+        title="pattern hit",
+        severity=severity,
+        confidence="medium",
+        evidence_level="static_corroboration",
+        rationale=rationale,
+        line=1,
+        function_name=function_name,
+        reachability_status=reachability,
+        reachability_evidence=[],
+        reachability_conditions=[],
+        tags=[],
+        ranking_priority=1.0,
+    )
+
+
+def _hotspot(
+    path: str,
+    *,
+    function_name: str | None = None,
+    score: float = 4.0,
+    band: str = "medium",
+    inventory_finding_ids: tuple[str, ...] = (),
+) -> Hotspot:
+    return Hotspot(
+        path=path,
+        function_name=function_name,
+        score=score,
+        band=band,
+        signals={"loc": 10, "inventory_hit_count": len(inventory_finding_ids)},
+        rationale=f"{band} band: score {score}",
+        test_hint=None,
+        inventory_finding_ids=inventory_finding_ids,
+    )
+
+
+def test_hotspot_key_joins_path_and_function_or_empty() -> None:
+    assert hotspot_key("parser.c", "parse_header") == "parser.c:parse_header"
+    assert hotspot_key("src/app.py", None) == "src/app.py:"
+    assert hotspot_key("src/app.py", "") == "src/app.py:"
+
+
+def test_triggerable_overlay_raises_later_score() -> None:
+    hotspot = _hotspot("parser.c", function_name="parse_header", score=4.0, inventory_finding_ids=("hit:1",))
+    ledger = record_verdict({}, path="parser.c", function_name="parse_header", verdict="triggerable")
+
+    overlaid = apply_overlay((hotspot,), ledger)
+    raised = overlaid[0]
+
+    assert ledger[hotspot_key("parser.c", "parse_header")].score_delta == TRIGGERABLE_DELTA
+    assert TRIGGERABLE_DELTA > 0
+    assert raised.score == hotspot.score + TRIGGERABLE_DELTA
+    assert raised.score > hotspot.score
+    assert raised.inventory_finding_ids == ("hit:1",)
+
+
+def test_not_triggerable_overlay_drops_score_and_keeps_inventory_hit() -> None:
+    hotspot = _hotspot("src/auth.py", function_name="login", score=5.0, inventory_finding_ids=("cmd:src/auth.py:12",))
+    ledger = {
+        hotspot_key("src/auth.py", "login"): LedgerEntry(
+            score_delta=NOT_TRIGGERABLE_DELTA,
+            last_verdict="not_triggerable",
+            round=1,
+        )
+    }
+
+    overlaid = apply_overlay((hotspot,), ledger)
+    demoted = overlaid[0]
+
+    assert NOT_TRIGGERABLE_DELTA < 0
+    assert demoted.score == hotspot.score + NOT_TRIGGERABLE_DELTA
+    assert demoted.score < hotspot.score
+    assert demoted.inventory_finding_ids == ("cmd:src/auth.py:12",)
+    assert demoted.path == hotspot.path
+    assert demoted.function_name == hotspot.function_name
+
+
+def test_not_triggerable_overlay_drops_map_score_and_sev5_reachable_stays_candidate(tmp_path: Path) -> None:
+    auth = _target(
+        "src/auth.py",
+        loc=80,
+        tags=["auth_boundary"],
+        hints=[{"kind": "route", "access_level": "public", "function_name": "login"}],
+    )
+    helper = _target("src/helper.py", loc=12, tags=[])
+    sev5 = _hit(
+        "src/auth.py",
+        "cmd:src/auth.py:12",
+        function_name="login",
+        reachability="reachable",
+        severity="critical",
+        rationale="Static pattern cmd (CWE-78) matched os.system on line 12; manual verification still required.",
+    )
+    noise = _hit("src/helper.py", "print:src/helper.py:1", reachability="unknown", severity="low")
+    artifact = tmp_path / "complexity_map.json"
+
+    baseline = build_complexity_map([auth, helper], [sev5, noise], artifact)
+    original = next(item for item in baseline.hotspots if item.path == "src/auth.py")
+    assert sev5.finding_id in original.inventory_finding_ids
+
+    ledger_path = tmp_path / ".openultrasast" / "calibration" / "complexity_ledger.json"
+    write_ledger(
+        ledger_path,
+        record_verdict({}, path=original.path, function_name=original.function_name, verdict="not_triggerable"),
+    )
+
+    overlaid = build_complexity_map([auth, helper], [sev5, noise], artifact, ledger_path=ledger_path)
+    demoted = next(item for item in overlaid.hotspots if item.path == "src/auth.py")
+
+    assert demoted.score < original.score
+    assert demoted.score == original.score + NOT_TRIGGERABLE_DELTA
+    assert demoted.inventory_finding_ids == original.inventory_finding_ids
+    assert sev5.finding_id in demoted.inventory_finding_ids
+    assert sev5.severity == "critical"
+    assert sev5.rationale == ("Static pattern cmd (CWE-78) matched os.system on line 12; manual verification still required.")
+
+    policy = {
+        sev5.finding_id: 5,
+        noise.finding_id: 2,
+    }
+    forced = select_forced_candidates([sev5, noise], policy)
+    assert forced == (sev5,)
+    assert must_keep_as_candidate(sev5, 5, "reachable") is True
+    assert must_keep_as_candidate(noise, 2, "unknown") is False
+
+
+def test_must_keep_as_candidate_requires_sev5_and_reachable_surface() -> None:
+    reachable = _hit("a.py", "a:1", reachability="reachable")
+    inferred = _hit("b.py", "b:1", reachability="inferred-file-surface")
+    unknown = _hit("c.py", "c:1", reachability="unknown")
+
+    assert must_keep_as_candidate(reachable, 5, "reachable") is True
+    assert must_keep_as_candidate(inferred, 5, "inferred-file-surface") is True
+    assert must_keep_as_candidate(unknown, 5, "unknown") is False
+    assert must_keep_as_candidate(reachable, 4, "reachable") is False
+    assert must_keep_as_candidate(reachable, 5, None) is True
+
+
+def test_select_forced_candidates_resolves_policy_severity() -> None:
+    sev5 = _hit("src/auth.py", "cmd:src/auth.py:12", reachability="reachable")
+    inferred = _hit("src/upload.py", "sql:src/upload.py:4", reachability="inferred-file-surface")
+    mild = _hit("src/helper.py", "print:src/helper.py:1", reachability="reachable")
+    policy = {
+        "CWE-78": CwePolicy("Command Injection", 5, True, False),
+        "CWE-89": CwePolicy("SQL Injection", 5, True, False),
+        "CWE-116": CwePolicy("Encode Output", 2, True, False),
+    }
+
+    forced = select_forced_candidates(
+        [mild, sev5, inferred],
+        policy,
+        rule_cwe={
+            sev5.finding_id: "CWE-78",
+            inferred.finding_id: "CWE-89",
+            mild.finding_id: "CWE-116",
+        },
+    )
+
+    assert forced == (sev5, inferred)
+
+
+def test_ledger_round_trip_matches_design_schema(tmp_path: Path) -> None:
+    path = tmp_path / "complexity_ledger.json"
+    key = hotspot_key("parser.c", "parse_header")
+    write_ledger(
+        path,
+        {key: LedgerEntry(score_delta=1.5, last_verdict="triggerable", round=3)},
+    )
+    payload = json.loads(path.read_text())
+
+    assert payload == {
+        key: {
+            "last_verdict": "triggerable",
+            "round": 3,
+            "score_delta": 1.5,
+        }
+    }
+    loaded = load_ledger(path)
+    assert loaded[key] == LedgerEntry(score_delta=1.5, last_verdict="triggerable", round=3)
+    assert load_ledger(tmp_path / "missing.json") == {}
+    assert load_ledger(None) == {}
+
+
+def test_overlay_reorders_and_rebands_without_dropping_hits() -> None:
+    high = _hotspot("src/auth.py", function_name="login", score=3.2, band="medium", inventory_finding_ids=("sev5:1",))
+    other = _hotspot("src/helper.py", score=2.0, band="low", inventory_finding_ids=("low:1",))
+    ledger = record_verdict({}, path="src/auth.py", function_name="login", verdict="not_triggerable")
+
+    overlaid = apply_overlay((high, other), ledger)
+
+    assert [item.path for item in overlaid] == ["src/helper.py", "src/auth.py"]
+    demoted = overlaid[1]
+    assert demoted.score == round(3.2 + NOT_TRIGGERABLE_DELTA, 4)
+    assert demoted.band == "low"
+    assert demoted.inventory_finding_ids == ("sev5:1",)
+    assert overlaid[0].inventory_finding_ids == ("low:1",)
+
+
+def test_build_complexity_map_without_ledger_path_skips_overlay(tmp_path: Path) -> None:
+    target = _target("src/app.py", loc=20)
+    finding = _hit("src/app.py", "rule:src/app.py:1")
+    artifact = tmp_path / "complexity_map.json"
+
+    first = build_complexity_map([target], [finding], artifact)
+    second = build_complexity_map([target], [finding], artifact)
+
+    assert first.hotspots[0].score == second.hotspots[0].score
+    assert first.hotspots[0].inventory_finding_ids == (finding.finding_id,)