feat(three-stage-scan): select bounded regress candidates with sev-5 extras

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

@@ -93,7 +93,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
   - _Boundary: DockerCliRunner_
   - _Depends: 4.1_
 
-- [ ] 4.3 Reject unsafe regression snippets before execution
+- [x] 4.3 Reject unsafe regression snippets before execution
   - Snippets that refer to the Docker socket, host network, or writes under the source mount are not executed and become `inconclusive`.
   - Observable completion: each forbidden shape is rejected in a unit test and no runner job is created.
   - _Requirements: 5.8, 6.6_
@@ -103,7 +103,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
 ## Phase 3 — Core: Regression verdicts
 
 - [ ] 5. Produce worth-fixing verdicts from sandboxed candidates
-- [ ] 5.1 Select a bounded candidate set from hotspots and severe reachable inventory
+- [x] 5.1 Select a bounded candidate set from hotspots and severe reachable inventory
   - Selection honors max-candidates and always includes reachable severity-5 inventory even when the ledger demoted the hotspot.
   - Observable completion: a unit test with six hotspots and max-candidates three returns three, plus the sev-5 reachable extra when present.
   - _Requirements: 5.1, 7.4, 9.5_

src/openultrasast/regress/__init__.py

@@ -1,6 +1,6 @@
 """Isolated regression snippets and verdicts."""
 
-from .candidate import INCONCLUSIVE, SAFETY_REJECTED, RegressionRunner, RegressionVerdict
+from .candidate import INCONCLUSIVE, SAFETY_REJECTED, RegressionRunner, RegressionVerdict, select_candidates
 from .safety import UnsafeSnippetError, check_snippet_safety
 
 __all__ = [
@@ -10,4 +10,5 @@ __all__ = [
     "RegressionVerdict",
     "UnsafeSnippetError",
     "check_snippet_safety",
+    "select_candidates",
 ]

src/openultrasast/regress/candidate.py

@@ -1,9 +1,13 @@
-"""Safety-gated regression execution (task 4.3)."""
+"""Candidate selection and safety-gated regression execution."""
 
 from __future__ import annotations
 
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 
+from ..complexity.ledger import hotspot_key, select_forced_candidates
+from ..complexity.map import Hotspot
+from ..findings import StaticFinding
 from ..sandbox import SandboxJob, SandboxResult, SandboxRunner
 from .safety import UnsafeSnippetError, check_snippet_safety
 
@@ -13,6 +17,58 @@ TRIGGERABLE = "triggerable"
 NOT_TRIGGERABLE = "not_triggerable"
 
 
+def select_candidates(
+    hotspots: Sequence[Hotspot],
+    findings: Sequence[StaticFinding],
+    *,
+    max_candidates: int,
+    policy_severity_by_id: Mapping[str, object] | None = None,
+    rule_cwe: Mapping[str, str] | None = None,
+) -> tuple[Hotspot, ...]:
+    """Select min(max_candidates, ranked hotspots) ∪ sev-5 reachable inventory.
+
+    The cap applies to the ranked hotspot slice. Forced policy-severity-5
+    reachable (or inferred-file-surface) inventory stays in the set even when a
+    ledger overlay demoted its hotspot out of that slice.
+    """
+    ranked = tuple(sorted(hotspots, key=lambda item: (-item.score, item.path, item.function_name or "")))
+    selected = list(ranked[: max(max_candidates, 0)])
+    seen = {hotspot_key(item.path, item.function_name) for item in selected}
+    extras: list[Hotspot] = []
+    for finding in select_forced_candidates(findings, policy_severity_by_id or {}, rule_cwe=rule_cwe):
+        hotspot = _hotspot_for_finding(finding, ranked) or _hotspot_from_forced_finding(finding)
+        key = hotspot_key(hotspot.path, hotspot.function_name)
+        if key in seen:
+            continue
+        seen.add(key)
+        extras.append(hotspot)
+    return tuple(selected + extras)
+
+
+def _hotspot_for_finding(finding: StaticFinding, hotspots: Sequence[Hotspot]) -> Hotspot | None:
+    for hotspot in hotspots:
+        if finding.finding_id in hotspot.inventory_finding_ids:
+            return hotspot
+    key = hotspot_key(finding.path, finding.function_name)
+    for hotspot in hotspots:
+        if hotspot_key(hotspot.path, hotspot.function_name) == key:
+            return hotspot
+    return None
+
+
+def _hotspot_from_forced_finding(finding: StaticFinding) -> Hotspot:
+    return Hotspot(
+        path=finding.path,
+        function_name=finding.function_name,
+        score=0.0,
+        band="low",
+        signals={},
+        rationale="forced reachable severity-5 inventory",
+        test_hint=None,
+        inventory_finding_ids=(finding.finding_id,),
+    )
+
+
 @dataclass(frozen=True)
 class RegressionVerdict:
     verdict: str

tests/test_regress_candidates.py

@@ -0,0 +1,159 @@
+"""Bounded regression candidate selection from hotspots and severe inventory (task 5.1)."""
+
+from openultrasast.complexity.ledger import NOT_TRIGGERABLE_DELTA, apply_overlay, record_verdict
+from openultrasast.complexity.map import Hotspot
+from openultrasast.findings import StaticFinding
+from openultrasast.regress.candidate import select_candidates
+
+
+def _finding(
+    path: str,
+    finding_id: str,
+    *,
+    function_name: str | None = None,
+    reachability: str = "unknown",
+    severity: str = "low",
+) -> StaticFinding:
+    return StaticFinding(
+        finding_id=finding_id,
+        path=path,
+        title="pattern hit",
+        severity=severity,
+        confidence="medium",
+        evidence_level="static_corroboration",
+        rationale="inventory",
+        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 _six_hotspots(*, sev5_id: str | None = None, sev5_score: float = 4.0) -> tuple[Hotspot, ...]:
+    return (
+        _hotspot("a.py", function_name="a", score=5.0),
+        _hotspot("b.py", function_name="b", score=4.8),
+        _hotspot("c.py", function_name="c", score=4.6),
+        _hotspot("d.py", function_name="d", score=4.4),
+        _hotspot("e.py", function_name="e", score=4.2),
+        _hotspot(
+            "f.py",
+            function_name="f",
+            score=sev5_score,
+            inventory_finding_ids=(sev5_id,) if sev5_id else (),
+        ),
+    )
+
+
+def test_six_hotspots_max_candidates_three_returns_three() -> None:
+    noise = _finding("f.py", "print:f.py:1", function_name="f", reachability="reachable")
+    selected = select_candidates(
+        _six_hotspots(),
+        [noise],
+        max_candidates=3,
+        policy_severity_by_id={noise.finding_id: 2},
+    )
+
+    assert len(selected) == 3
+    assert [item.path for item in selected] == ["a.py", "b.py", "c.py"]
+
+
+def test_sev5_reachable_extra_included_when_ledger_demoted_hotspot() -> None:
+    sev5_id = "cmd:f.py:1"
+    hotspots = _six_hotspots(sev5_id=sev5_id, sev5_score=5.2)
+    ledger = record_verdict({}, path="f.py", function_name="f", verdict="not_triggerable")
+    demoted = apply_overlay(hotspots, ledger)
+    finding = _finding(
+        "f.py",
+        sev5_id,
+        function_name="f",
+        reachability="reachable",
+        severity="critical",
+    )
+
+    selected = select_candidates(
+        demoted,
+        [finding],
+        max_candidates=3,
+        policy_severity_by_id={sev5_id: 5},
+    )
+
+    demoted_f = next(item for item in demoted if item.path == "f.py")
+    assert demoted_f.score == round(5.2 + NOT_TRIGGERABLE_DELTA, 4)
+    assert [item.path for item in demoted[:3]] == ["a.py", "b.py", "c.py"]
+    assert demoted_f not in demoted[:3]
+    assert len(selected) == 4
+    assert [item.path for item in selected[:3]] == ["a.py", "b.py", "c.py"]
+    assert selected[3] is demoted_f
+    assert sev5_id in selected[3].inventory_finding_ids
+
+
+def test_sev5_already_in_cap_is_not_duplicated() -> None:
+    sev5_id = "cmd:a.py:1"
+    hotspots = (
+        _hotspot("a.py", function_name="a", score=5.0, inventory_finding_ids=(sev5_id,)),
+        _hotspot("b.py", function_name="b", score=4.8),
+        _hotspot("c.py", function_name="c", score=4.6),
+        _hotspot("d.py", function_name="d", score=4.4),
+        _hotspot("e.py", function_name="e", score=4.2),
+        _hotspot("f.py", function_name="f", score=4.0),
+    )
+    finding = _finding("a.py", sev5_id, function_name="a", reachability="reachable", severity="critical")
+
+    selected = select_candidates(
+        hotspots,
+        [finding],
+        max_candidates=3,
+        policy_severity_by_id={sev5_id: 5},
+    )
+
+    assert len(selected) == 3
+    assert [item.path for item in selected] == ["a.py", "b.py", "c.py"]
+    assert sev5_id in selected[0].inventory_finding_ids
+
+
+def test_sev5_reachable_inventory_without_hotspot_is_still_selected() -> None:
+    finding = _finding(
+        "z.py",
+        "cmd:z.py:1",
+        function_name="z",
+        reachability="reachable",
+        severity="critical",
+    )
+
+    selected = select_candidates(
+        _six_hotspots(),
+        [finding],
+        max_candidates=3,
+        policy_severity_by_id={finding.finding_id: 5},
+    )
+
+    assert len(selected) == 4
+    assert [item.path for item in selected[:3]] == ["a.py", "b.py", "c.py"]
+    extra = selected[3]
+    assert extra.path == "z.py"
+    assert extra.function_name == "z"
+    assert extra.inventory_finding_ids == (finding.finding_id,)