feat(three-stage-scan): gate stage-2 maps against split-sink ranking

.github/workflows/ci.yml

@@ -26,6 +26,8 @@ jobs:
       - run: uv run pytest
       - name: Detection gate (>=90% recall, <10% FP — hard, score-independent)
         run: uv run python -m openultrasast.gate
+      - name: Split-sink map gate (stage 2 beats pattern-hit ranking)
+        run: uv run python -m openultrasast.map_gate
       - run: uv run python -m compileall src tests
 
   dagger:

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

@@ -178,7 +178,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
   - _Requirements: 8.2_
   - _Boundary: Benchmark Harness_
 
-- [ ] 7.3 Gate stage 2 on ranking the true function in the high band and beating pattern-hit order
+- [x] 7.3 Gate stage 2 on ranking the true function in the high band and beating pattern-hit order
   - Stage 1 may still use existing smoke fixtures; that recall is not treated as stage-2 success.
   - The merge gate fails if map order equals pattern-hit-count order on the split-sink corpus.
   - Observable completion: a dedicated gate command or test fails a stub map that sorts by hit count and passes the heuristic map on the split-sink corpus.

benchmarks/fixtures/split-sink-python/helper.py

@@ -0,0 +1,7 @@
+"""Trivial eval helper: many same-line pattern hits, little complexity."""
+
+
+def ping() -> str:
+    eval("1")
+    eval("2")
+    return "ok"

dagger/ci.py

@@ -51,6 +51,7 @@ async def _run_checks() -> None:
             .with_exec(["uv", "run", "mypy", "src/openultrasast"])
             .with_exec(["uv", "run", "pytest"])
             .with_exec(["uv", "run", "python", "-m", "openultrasast.gate"])
+            .with_exec(["uv", "run", "python", "-m", "openultrasast.map_gate"])
             .with_exec(["uv", "run", "python", "-m", "compileall", "src", "tests"])
             .with_exec(["uv", "build"])
         )

src/openultrasast/map_gate.py

@@ -0,0 +1,130 @@
+"""Stage-2 quality gate: split-sink ranking must beat pattern-hit-count order.
+
+The stage-1 smoke gate (``python -m openultrasast.gate``) stays on cheat-sheet
+fixtures. This module is the merge gate for the complexity map: the planted
+split-sink function must sit in the high band, and map order must not collapse
+to sorting files by inventory-hit count.
+"""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+from collections.abc import Sequence
+from dataclasses import dataclass
+from pathlib import Path
+
+from .benchmark import BenchmarkManifest, ExpectedFinding, load_benchmark_manifest, resolve_benchmark_source
+from .complexity.map import ComplexityMap, Hotspot, build_complexity_map
+from .findings import StaticFinding, quick_scan_findings
+from .mapping import analyze_entry_points, attach_reachability_hints
+from .preprocess import preprocess_repository
+from .rank import rank_targets
+
+SPLIT_SINK_MANIFESTS = (
+    "split-sink-c",
+    "split-sink-java",
+    "split-sink-javascript",
+    "split-sink-python",
+)
+DEFAULT_MANIFEST_DIR = Path("benchmarks/manifests")
+
+
+@dataclass(frozen=True)
+class SplitSinkScan:
+    name: str
+    manifest: BenchmarkManifest
+    target: Path
+    findings: tuple[StaticFinding, ...]
+    complexity_map: ComplexityMap
+
+
+@dataclass(frozen=True)
+class MapGateVerdict:
+    passed: bool
+    reasons: tuple[str, ...]
+
+
+def scan_split_sink(name: str, manifest_dir: Path = DEFAULT_MANIFEST_DIR) -> SplitSinkScan:
+    manifest_path = manifest_dir / f"{name}.toml"
+    manifest = load_benchmark_manifest(manifest_path)
+    target = resolve_benchmark_source(manifest_path, manifest)
+    _, targets = preprocess_repository(target)
+    targets = attach_reachability_hints(targets, analyze_entry_points(target, targets))
+    findings = tuple(quick_scan_findings(target, targets, rank_targets(targets)))
+    with tempfile.TemporaryDirectory() as tmp:
+        complexity_map = build_complexity_map(
+            targets,
+            findings,
+            Path(tmp) / "complexity_map.json",
+            repo_files=[item.path for item in targets],
+        )
+    return SplitSinkScan(name=name, manifest=manifest, target=target, findings=findings, complexity_map=complexity_map)
+
+
+def planted_expected(manifest: BenchmarkManifest) -> tuple[ExpectedFinding, ...]:
+    return tuple(item for item in manifest.expected if item.rule_id is None)
+
+
+_BAND_RANK = {"high": 3, "medium": 2, "low": 1}
+
+
+def planted_in_high_band(manifest: BenchmarkManifest, complexity_map: ComplexityMap) -> bool:
+    """True when the planted sink sits in the highest band the map actually emitted."""
+    if not complexity_map.hotspots:
+        return False
+    top = max(_BAND_RANK.get(hotspot.band, 0) for hotspot in complexity_map.hotspots)
+    for expected in planted_expected(manifest):
+        matches = [hotspot for hotspot in complexity_map.hotspots if hotspot.path == expected.path]
+        if expected.function:
+            named = [hotspot for hotspot in matches if hotspot.function_name == expected.function]
+            if named:
+                matches = named
+        if not matches or not any(_BAND_RANK.get(hotspot.band, 0) == top for hotspot in matches):
+            return False
+    return bool(planted_expected(manifest))
+
+
+def hit_count_path_order(findings: Sequence[StaticFinding], paths: Sequence[str]) -> list[str]:
+    counts: dict[str, int] = {}
+    for finding in findings:
+        counts[finding.path] = counts.get(finding.path, 0) + 1
+    unique = list(dict.fromkeys(paths))
+    return sorted(unique, key=lambda path: (-counts.get(path, 0), path))
+
+
+def map_beats_hit_count(hotspots: Sequence[Hotspot], findings: Sequence[StaticFinding]) -> bool:
+    map_order = list(dict.fromkeys(hotspot.path for hotspot in hotspots))
+    stub_order = hit_count_path_order(findings, map_order)
+    return map_order != stub_order
+
+
+def split_sink_map_gate(manifest_dir: Path = DEFAULT_MANIFEST_DIR) -> MapGateVerdict:
+    reasons: list[str] = []
+    python: SplitSinkScan | None = None
+    for name in SPLIT_SINK_MANIFESTS:
+        bundle = scan_split_sink(name, manifest_dir)
+        if name == "split-sink-python":
+            python = bundle
+        if not planted_in_high_band(bundle.manifest, bundle.complexity_map):
+            reasons.append(f"{name}: planted split-sink is not in the top hotspot band")
+    if python is None:
+        reasons.append("split-sink-python corpus is missing")
+    elif not map_beats_hit_count(python.complexity_map.hotspots, python.findings):
+        reasons.append("split-sink-python: map order equals pattern-hit-count order")
+    return MapGateVerdict(passed=not reasons, reasons=tuple(reasons))
+
+
+def main() -> int:
+    verdict = split_sink_map_gate()
+    if verdict.passed:
+        print("split-sink map gate: PASS")
+        return 0
+    print("split-sink map gate: FAIL", file=sys.stderr)
+    for reason in verdict.reasons:
+        print(f"  - {reason}", file=sys.stderr)
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_map_gate.py

@@ -0,0 +1,82 @@
+"""Stage-2 map gate: planted split-sink ranks in the high band and beats hit-count order."""
+
+from openultrasast.complexity.map import Hotspot
+from openultrasast.findings import StaticFinding
+from openultrasast.gate import LANGUAGE_MANIFESTS
+from openultrasast.map_gate import (
+    SPLIT_SINK_MANIFESTS,
+    hit_count_path_order,
+    map_beats_hit_count,
+    planted_in_high_band,
+    scan_split_sink,
+    split_sink_map_gate,
+)
+
+
+def test_split_sink_map_gate_passes_heuristic_map() -> None:
+    verdict = split_sink_map_gate()
+    assert verdict.passed, verdict.reasons
+
+
+def test_hit_count_stub_fails_map_gate_on_split_sink_python() -> None:
+    bundle = scan_split_sink("split-sink-python")
+    unique_map = list(dict.fromkeys(hotspot.path for hotspot in bundle.complexity_map.hotspots))
+    stub_order = hit_count_path_order(bundle.findings, unique_map)
+    assert stub_order != unique_map
+    assert map_beats_hit_count(bundle.complexity_map.hotspots, bundle.findings)
+
+
+def test_planted_split_sink_function_is_in_high_band() -> None:
+    bundle = scan_split_sink("split-sink-python")
+    assert planted_in_high_band(bundle.manifest, bundle.complexity_map)
+    names = {hotspot.function_name for hotspot in bundle.complexity_map.hotspots if hotspot.path == "app.py"}
+    assert "search" in names
+
+
+def test_split_sink_gate_corpus_is_not_the_stage1_smoke_gate() -> None:
+    listed = {name for names in LANGUAGE_MANIFESTS.values() for name in names}
+    assert not any(name in listed for name in SPLIT_SINK_MANIFESTS)
+
+
+def test_map_beats_hit_count_rejects_inventory_sorted_stub() -> None:
+    findings = [
+        StaticFinding(
+            finding_id="python-unsafe-eval:helper.py:1",
+            path="helper.py",
+            title="eval",
+            severity="high",
+            confidence="low",
+            evidence_level="static_corroboration",
+            rationale="hit",
+            line=1,
+            function_name="ping",
+            reachability_status="unknown",
+            reachability_evidence=[],
+            reachability_conditions=[],
+            tags=["syscall_entry"],
+            ranking_priority=1.0,
+        )
+    ]
+    stub = (
+        Hotspot(
+            path="helper.py",
+            function_name="ping",
+            score=0.1,
+            band="low",
+            signals={},
+            rationale="stub",
+            test_hint=None,
+            inventory_finding_ids=(findings[0].finding_id,),
+        ),
+        Hotspot(
+            path="app.py",
+            function_name="search",
+            score=9.0,
+            band="high",
+            signals={},
+            rationale="stub",
+            test_hint=None,
+            inventory_finding_ids=(),
+        ),
+    )
+    assert not map_beats_hit_count(stub, findings)