feat(review): rank untested by risk + new_untested_hotspot rule

codegraph/analysis/__init__.py

@@ -6,7 +6,7 @@
 from codegraph.analysis.hotspots import Hotspot, find_hotspots
 from codegraph.analysis.metrics import GraphMetrics, compute_metrics
 from codegraph.analysis.roles import classify_roles
-from codegraph.analysis.untested import UntestedNode, find_untested
+from codegraph.analysis.untested import UntestedNode, find_untested, rank_untested
 
 __all__ = [
     "BlastRadiusResult",
@@ -26,5 +26,6 @@
     "find_hotspots",
     "find_untested",
     "match_route",
+    "rank_untested",
     "trace",
 ]

codegraph/analysis/untested.py

@@ -1,7 +1,7 @@
 """Untested-symbol detection."""
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 import networkx as nx
 
@@ -27,6 +27,29 @@ class UntestedNode:
     file: str
     line_start: int
     incoming_calls: int
+    hotspot_score: int = field(default=0)
+    blast_radius_size: int = field(default=0)
+    blast_files: int = field(default=0)
+
+
+def _compute_hotspot_score(graph: nx.MultiDiGraph, nid: str) -> int:
+    """Compute hotspot score for a node: fan_in*2 + fan_out + loc//50.
+
+    Uses the same formula as :func:`codegraph.analysis.hotspots.Hotspot.score`.
+    """
+    attrs = graph.nodes.get(nid) or {}
+    fan_in = 0
+    fan_out = 0
+    for _src, _dst, key in graph.in_edges(nid, keys=True):
+        if key == EdgeKind.CALLS.value:
+            fan_in += 1
+    for _src, _dst, key in graph.out_edges(nid, keys=True):
+        if key == EdgeKind.CALLS.value:
+            fan_out += 1
+    line_start = int(attrs.get("line_start") or 0)
+    line_end = int(attrs.get("line_end") or 0)
+    loc = max(0, line_end - line_start + 1) if line_end else 0
+    return fan_in * 2 + fan_out + loc // 50
 
 
 def find_untested(graph: nx.MultiDiGraph) -> list[UntestedNode]:
@@ -77,3 +100,42 @@ def find_untested(graph: nx.MultiDiGraph) -> list[UntestedNode]:
         )
     out.sort(key=lambda u: (-u.incoming_calls, u.file, u.line_start))
     return out
+
+
+def rank_untested(
+    graph: nx.MultiDiGraph,
+    *,
+    blast_depth: int = 3,
+    top: int | None = None,
+) -> list[UntestedNode]:
+    """Return untested nodes ranked by risk: hotspot score x blast radius.
+
+    Enriches each :class:`UntestedNode` with ``hotspot_score``,
+    ``blast_radius_size``, and ``blast_files``, then sorts by
+    ``(-hotspot_score, -blast_radius_size, file, line_start)``.
+
+    Blast-radius computation is skipped for nodes with no callers (perf).
+
+    Args:
+        graph: The project call graph.
+        blast_depth: Maximum hop depth for blast-radius traversal.
+        top: If given, truncate result to this many entries.
+
+    Returns:
+        Sorted list of :class:`UntestedNode` with risk fields populated.
+    """
+    from codegraph.analysis.blast_radius import blast_radius
+
+    nodes = find_untested(graph)
+    for node in nodes:
+        node.hotspot_score = _compute_hotspot_score(graph, node.id)
+        if node.incoming_calls > 0:
+            br = blast_radius(graph, node.id, depth=blast_depth)
+            node.blast_radius_size = br.size
+            node.blast_files = len(br.files)
+    nodes.sort(
+        key=lambda u: (-u.hotspot_score, -u.blast_radius_size, u.file, u.line_start)
+    )
+    if top is not None:
+        nodes = nodes[:top]
+    return nodes

codegraph/cli.py

@@ -1359,21 +1359,32 @@ def query_subgraph(
 
 @query_app.command("untested")
 def query_untested(limit: int = typer.Option(50, "--limit")) -> None:
-    """List functions/methods with no test-side caller."""
-    from codegraph.analysis import find_untested
+    """List functions/methods with no test-side caller, ranked by risk."""
+    from codegraph.analysis import rank_untested
 
     graph = _open_graph(Path.cwd())
     if graph is None:
         raise typer.Exit(1)
-    rows = find_untested(graph)
+    rows = rank_untested(graph)
     console.print(f"[bold]{len(rows)} untested[/bold]")
     table = Table()
     table.add_column("qualname")
     table.add_column("file")
     table.add_column("line", justify="right")
     table.add_column("callers", justify="right")
+    table.add_column("hotspot", justify="right")
+    table.add_column("blast_r", justify="right")
+    table.add_column("blast_f", justify="right")
     for u in rows[:limit]:
-        table.add_row(u.qualname, u.file, str(u.line_start), str(u.incoming_calls))
+        table.add_row(
+            u.qualname,
+            u.file,
+            str(u.line_start),
+            str(u.incoming_calls),
+            str(u.hotspot_score),
+            str(u.blast_radius_size),
+            str(u.blast_files),
+        )
     console.print(table)
 
 

codegraph/mcp_server/server.py

@@ -331,17 +331,20 @@ def tool_untested(
     graph: nx.MultiDiGraph,
     limit: int = 50,
 ) -> list[dict[str, Any]]:
-    """Return untested functions/methods."""
-    from codegraph.analysis.untested import find_untested
+    """Return untested functions/methods ranked by risk."""
+    from codegraph.analysis.untested import rank_untested
 
-    items = find_untested(graph)
+    items = rank_untested(graph)
     return [
         {
             "qualname": u.qualname,
             "kind": u.kind,
             "file": u.file,
             "line": u.line_start,
             "incoming_calls": u.incoming_calls,
+            "hotspot_score": u.hotspot_score,
+            "blast_radius_size": u.blast_radius_size,
+            "blast_files": u.blast_files,
         }
         for u in items[:limit]
     ]

codegraph/review/rules.py

@@ -10,6 +10,8 @@
 import networkx as nx
 import yaml
 
+from codegraph.analysis._common import in_test_module
+from codegraph.graph.schema import EdgeKind, NodeKind
 from codegraph.review.differ import GraphDiff, NodeChange
 from codegraph.review.risk import (
     _count_callers,
@@ -87,6 +89,13 @@ class Finding:
         severity="med",
         message="Modified node signature change",
     ),
+    Rule(
+        id="new-untested-hotspot",
+        when="new_untested_hotspot",
+        severity="high",
+        message="New function with {fan_in} callers and no test coverage",
+        threshold=5,
+    ),
 ]
 
 
@@ -98,6 +107,7 @@ class Finding:
     "introduces_cycle",
     "high_fan_in",
     "new_dead_code",
+    "new_untested_hotspot",
 }
 
 
@@ -316,6 +326,39 @@ def evaluate_rules(
                         new_graph=new_graph, old_graph=old_graph, extra=extra,
                     )
                 )
+        elif when == "new_untested_hotspot":
+            threshold = rule.threshold or 5
+            for change in diff.added_nodes:
+                if not _node_matches(rule, change):
+                    continue
+                if change.kind not in (
+                    NodeKind.FUNCTION.value,
+                    NodeKind.METHOD.value,
+                ):
+                    continue
+                new_id = _find_node_id(change.qualname, change.kind, new_graph)
+                if new_id is None:
+                    continue
+                fan_in = _count_callers(new_id, new_graph)
+                if fan_in < threshold:
+                    continue
+                # Check whether any caller is from a test module.
+                has_test_caller = False
+                for src, _dst, key in new_graph.in_edges(new_id, keys=True):
+                    if key != EdgeKind.CALLS.value:
+                        continue
+                    if in_test_module(new_graph, src):
+                        has_test_caller = True
+                        break
+                if has_test_caller:
+                    continue
+                findings.append(
+                    _make_finding(
+                        rule, change,
+                        new_graph=new_graph, old_graph=old_graph, extra=extra,
+                        fmt_kwargs={"fan_in": fan_in},
+                    )
+                )
         elif when == "introduces_cycle":
             if not cycle_introduced:
                 continue

tests/fixtures/python_hotspot_v1/core.py

@@ -0,0 +1,12 @@
+"""Minimal v1 module — no hot_helper yet."""
+from __future__ import annotations
+
+
+def helper_a(x: int) -> int:
+    """Simple helper."""
+    return x + 1
+
+
+def helper_b(x: int) -> int:
+    """Another helper."""
+    return x + 2

tests/fixtures/python_hotspot_v1/test_core.py

@@ -0,0 +1,12 @@
+"""Tests for core (v1)."""
+from __future__ import annotations
+
+from core import helper_a, helper_b
+
+
+def test_helper_a() -> None:
+    assert helper_a(1) == 2
+
+
+def test_helper_b() -> None:
+    assert helper_b(1) == 3

tests/fixtures/python_hotspot_v2/core.py

@@ -0,0 +1,32 @@
+"""Minimal v2 module — hot_helper added with many callers but no test."""
+from __future__ import annotations
+
+
+def hot_helper(x: int) -> int:
+    """New critical helper used everywhere."""
+    return x * 10
+
+
+def helper_a(x: int) -> int:
+    """Simple helper."""
+    return hot_helper(x) + 1
+
+
+def helper_b(x: int) -> int:
+    """Another helper."""
+    return hot_helper(x) + 2
+
+
+def helper_c(x: int) -> int:
+    """Third helper."""
+    return hot_helper(x) + 3
+
+
+def helper_d(x: int) -> int:
+    """Fourth helper."""
+    return hot_helper(x) + 4
+
+
+def helper_e(x: int) -> int:
+    """Fifth helper."""
+    return hot_helper(x) + 5

tests/fixtures/python_hotspot_v2/test_core.py

@@ -0,0 +1,12 @@
+"""Tests for core (v2) — intentionally does NOT test hot_helper directly."""
+from __future__ import annotations
+
+from core import helper_a, helper_b
+
+
+def test_helper_a() -> None:
+    assert helper_a(1) == 11
+
+
+def test_helper_b() -> None:
+    assert helper_b(1) == 12

tests/fixtures/python_hotspot_v2_tested/core.py

@@ -0,0 +1,32 @@
+"""Minimal v2 module — hot_helper added with many callers but no test."""
+from __future__ import annotations
+
+
+def hot_helper(x: int) -> int:
+    """New critical helper used everywhere."""
+    return x * 10
+
+
+def helper_a(x: int) -> int:
+    """Simple helper."""
+    return hot_helper(x) + 1
+
+
+def helper_b(x: int) -> int:
+    """Another helper."""
+    return hot_helper(x) + 2
+
+
+def helper_c(x: int) -> int:
+    """Third helper."""
+    return hot_helper(x) + 3
+
+
+def helper_d(x: int) -> int:
+    """Fourth helper."""
+    return hot_helper(x) + 4
+
+
+def helper_e(x: int) -> int:
+    """Fifth helper."""
+    return hot_helper(x) + 5

tests/fixtures/python_hotspot_v2_tested/test_core.py

@@ -0,0 +1,16 @@
+"""Tests for core (v2) — this variant DOES test hot_helper directly."""
+from __future__ import annotations
+
+from core import helper_a, helper_b, hot_helper
+
+
+def test_hot_helper() -> None:
+    assert hot_helper(2) == 20
+
+
+def test_helper_a() -> None:
+    assert helper_a(1) == 11
+
+
+def test_helper_b() -> None:
+    assert helper_b(1) == 12

tests/test_mcp_server.py

@@ -222,6 +222,9 @@ def test_untested_returns_list(fixture_graph: nx.MultiDiGraph) -> None:
     for r in results:
         assert "qualname" in r
         assert "kind" in r
+        assert "hotspot_score" in r
+        assert "blast_radius_size" in r
+        assert "blast_files" in r
 
 
 def test_hotspots_returns_list(fixture_graph: nx.MultiDiGraph) -> None: