feat(three-stage-scan): emit ordered complexity map artifact

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

@@ -39,7 +39,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
   - _Requirements: 3.1, 3.2_
   - _Boundary: ComplexityMapBuilder_
 
-- [ ] 2.2 (P) Emit the ordered hotspot map artifact
+- [x] 2.2 (P) Emit the ordered hotspot map artifact
   - Each hotspot carries score, band, rationale, signals, and linked inventory ids; the map records whether it is heuristic-only.
   - Observable completion: building a map writes a JSON artifact whose hotspot order is not identical to sorting by inventory-hit count on the same fixture.
   - _Requirements: 3.2, 3.3, 3.5, 8.6_
@@ -197,3 +197,4 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
 - `RegressConfig.images` is a sorted tuple of `(language, image)` pairs so frozen config stays hashable.
 - `scan_exit_code` duck-types `worth_fixing` on verdict objects; `_run_scan` still passes no verdicts until task 6.5. Dict-shaped verdicts would not fail the gate.
 - Map `top_k` default is 20 (not named in design.md); `max_hunter_hotspots=8` and `max_candidates=5` follow design.
+- `collect_signals` adjacent-test detection is false unless the caller also passes `repo_files`; task 2.2 must pass enumerated repo paths.

src/openultrasast/complexity/__init__.py

@@ -1,5 +1,13 @@
-"""Heuristic complexity signals for the stage-2 map."""
+"""Heuristic complexity map for stage 2."""
 
+from .map import ComplexityMap, Hotspot, build_complexity_map, write_complexity_map
 from .signals import ComplexitySignals, collect_signals
 
-__all__ = ["ComplexitySignals", "collect_signals"]
+__all__ = [
+    "ComplexityMap",
+    "ComplexitySignals",
+    "Hotspot",
+    "build_complexity_map",
+    "collect_signals",
+    "write_complexity_map",
+]

src/openultrasast/complexity/map.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+from ..findings import StaticFinding
+from ..preprocess import FileTarget
+from .signals import ComplexitySignals, collect_signals
+
+# Score cutoffs for hotspot bands; inventory density is not the rank driver.
+_HIGH_SCORE = 6.0
+_MEDIUM_SCORE = 3.0
+
+
+@dataclass(frozen=True)
+class TestHint:
+    path: str
+    function_name: str | None
+    gap: str
+    test_kind: str | None
+    reason: str
+
+
+@dataclass(frozen=True)
+class Hotspot:
+    path: str
+    function_name: str | None
+    score: float
+    band: str  # high | medium | low
+    signals: dict[str, float | int | bool | str]
+    rationale: str
+    test_hint: TestHint | None
+    inventory_finding_ids: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class ComplexityMap:
+    hotspots: tuple[Hotspot, ...]
+    heuristic_only: bool
+
+
+def build_complexity_map(
+    targets: Sequence[FileTarget],
+    findings: Sequence[StaticFinding],
+    output_path: Path,
+    *,
+    repo_files: Iterable[str] = (),
+    sources: Mapping[str, str] | 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)
+    complexity_map = ComplexityMap(hotspots=hotspots, heuristic_only=True)
+    write_complexity_map(complexity_map, output_path)
+    return complexity_map
+
+
+def write_complexity_map(complexity_map: ComplexityMap, path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(asdict(complexity_map), indent=2, sort_keys=True) + "\n")
+
+
+def _hotspot_from_signals(item: ComplexitySignals, inventory_finding_ids: tuple[str, ...]) -> Hotspot:
+    band = _band(item.score)
+    return Hotspot(
+        path=item.path,
+        function_name=item.function_name,
+        score=item.score,
+        band=band,
+        signals=_signal_payload(item),
+        rationale=_rationale(item, band),
+        test_hint=None,
+        inventory_finding_ids=inventory_finding_ids,
+    )
+
+
+def _band(score: float) -> str:
+    if score >= _HIGH_SCORE:
+        return "high"
+    if score >= _MEDIUM_SCORE:
+        return "medium"
+    return "low"
+
+
+def _signal_payload(item: ComplexitySignals) -> dict[str, float | int | bool | str]:
+    return {
+        "loc": item.loc,
+        "nesting": item.nesting,
+        "reachability": item.reachability,
+        "inventory_hit_count": item.inventory_hit_count,
+        "has_adjacent_test": item.has_adjacent_test,
+        "tags": ",".join(item.tags),
+    }
+
+
+def _rationale(item: ComplexitySignals, band: str) -> str:
+    tags = ",".join(item.tags) or "none"
+    return (
+        f"{band} band: score {item.score} against cutoffs high>={_HIGH_SCORE}, medium>={_MEDIUM_SCORE}; "
+        f"loc={item.loc}, nesting={item.nesting}, tags={tags}, reachability={item.reachability}, "
+        f"inventory_hit_count={item.inventory_hit_count} (density is a feature not the rank driver), "
+        f"has_adjacent_test={str(item.has_adjacent_test).lower()}."
+    )
+
+
+def _inventory_ids_by_path(findings: Sequence[StaticFinding]) -> dict[str, tuple[str, ...]]:
+    grouped: dict[str, list[str]] = {}
+    for finding in findings:
+        grouped.setdefault(finding.path, []).append(finding.finding_id)
+    return {path: tuple(ids) for path, ids in grouped.items()}

tests/test_complexity_map.py

@@ -0,0 +1,132 @@
+import json
+from pathlib import Path
+
+from openultrasast.complexity.map import build_complexity_map
+from openultrasast.findings import StaticFinding
+from openultrasast.preprocess import FileTarget
+
+_HOTSPOT_FIELDS = {
+    "path",
+    "function_name",
+    "score",
+    "band",
+    "signals",
+    "rationale",
+    "inventory_finding_ids",
+}
+
+
+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) -> StaticFinding:
+    return StaticFinding(
+        finding_id=finding_id,
+        path=path,
+        title="pattern hit",
+        severity="low",
+        confidence="medium",
+        evidence_level="static_corroboration",
+        rationale="inventory",
+        line=1,
+        function_name=None,
+        reachability_status="unknown",
+        reachability_evidence=[],
+        reachability_conditions=[],
+        tags=[],
+        ranking_priority=1.0,
+    )
+
+
+def _parser_helper_fixture() -> tuple[list[FileTarget], list[StaticFinding], tuple[str, ...]]:
+    parser = _target(
+        "src/parser.c",
+        loc=400,
+        language="c",
+        tags=["parser", "memory_unsafe"],
+        hints=[{"kind": "parser", "access_level": "public", "function_name": "parse_header"}],
+    )
+    helper = _target("src/helper.py", loc=1, tags=[])
+    findings = [_hit("src/helper.py", f"rule:helper:{index}") for index in range(8)]
+    repo_files = ("src/parser.c", "src/helper.py", "tests/test_parser.c")
+    return [parser, helper], findings, repo_files
+
+
+def test_map_order_differs_from_inventory_hit_count(tmp_path: Path) -> None:
+    targets, findings, repo_files = _parser_helper_fixture()
+    artifact = tmp_path / "complexity_map.json"
+
+    complexity_map = build_complexity_map(targets, findings, artifact, repo_files=repo_files)
+
+    map_paths = [hotspot.path for hotspot in complexity_map.hotspots]
+    hit_count_paths = [
+        hotspot.path
+        for hotspot in sorted(
+            complexity_map.hotspots,
+            key=lambda item: (-len(item.inventory_finding_ids), item.path),
+        )
+    ]
+    parser = next(hotspot for hotspot in complexity_map.hotspots if hotspot.path == "src/parser.c")
+    helper = next(hotspot for hotspot in complexity_map.hotspots if hotspot.path == "src/helper.py")
+
+    assert map_paths != hit_count_paths
+    assert map_paths[0] == "src/parser.c"
+    assert parser.inventory_finding_ids == ()
+    assert len(helper.inventory_finding_ids) == 8
+    assert helper.inventory_finding_ids == tuple(f"rule:helper:{index}" for index in range(8))
+    assert parser.score > helper.score
+    assert parser.function_name == "parse_header"
+    assert parser.band == "high"
+    assert helper.band == "low"
+    assert complexity_map.heuristic_only is True
+    assert parser.test_hint is None
+    assert "verified" not in parser.rationale.lower()
+    assert "worth fixing" not in parser.rationale.lower()
+    assert "worth_fixing" not in parser.rationale.lower()
+
+
+def test_complexity_map_json_has_required_fields(tmp_path: Path) -> None:
+    targets, findings, repo_files = _parser_helper_fixture()
+    artifact = tmp_path / "runs" / "complexity_map.json"
+
+    complexity_map = build_complexity_map(targets, findings, artifact, repo_files=repo_files)
+    raw = artifact.read_text()
+    payload = json.loads(raw)
+
+    assert raw == json.dumps(payload, indent=2, sort_keys=True) + "\n"
+    assert payload["heuristic_only"] is True
+    assert [item["path"] for item in payload["hotspots"]] == [hotspot.path for hotspot in complexity_map.hotspots]
+    for hotspot in payload["hotspots"]:
+        assert set(hotspot) >= _HOTSPOT_FIELDS
+        assert hotspot["band"] in {"high", "medium", "low"}
+        assert isinstance(hotspot["score"], float)
+        assert isinstance(hotspot["rationale"], str) and hotspot["rationale"]
+        assert isinstance(hotspot["signals"], dict)
+        assert set(hotspot["signals"]) >= {"loc", "nesting", "reachability", "inventory_hit_count", "has_adjacent_test"}
+        assert isinstance(hotspot["inventory_finding_ids"], list)
+        assert hotspot["test_hint"] is None
+        assert "6.0" in hotspot["rationale"]
+        assert "3.0" in hotspot["rationale"]
+    parser = next(item for item in payload["hotspots"] if item["path"] == "src/parser.c")
+    helper = next(item for item in payload["hotspots"] if item["path"] == "src/helper.py")
+    assert parser["signals"]["has_adjacent_test"] is True
+    assert helper["signals"]["has_adjacent_test"] is False
+    assert helper["inventory_finding_ids"] == [f"rule:helper:{index}" for index in range(8)]
+    assert parser["inventory_finding_ids"] == []