feat(query): --min-confidence filter on trace/impact (#112)

src/cgis/cli.py

@@ -43,6 +43,13 @@
     "--show-external/--no-show-external",
     help="Include stdlib and external nodes in output",
 )
+_OPT_MIN_CONFIDENCE: float | None = typer.Option(
+    None,
+    "--min-confidence",
+    min=0.0,
+    max=1.0,
+    help="Hide edges below this confidence (e.g. 0.5 drops unresolved raw_call edges)",
+)
 
 
 class OutputFormat(StrEnum):
@@ -244,6 +251,7 @@ def build_trace_tree(
     current_depth: int,
     allowed_edge_types: frozenset[EdgeType] | None = None,
     show_external: bool = True,
+    min_confidence: float | None = None,
 ) -> None:
     """Cycle-safe recursive tree builder for downstream flow tracing."""
     if current_depth >= max_depth:
@@ -252,6 +260,8 @@ def build_trace_tree(
     outgoing = store.get_outgoing_edges(current_id)
     if allowed_edge_types is not None:
         outgoing = [e for e in outgoing if e.type in allowed_edge_types]
+    if min_confidence is not None:
+        outgoing = [e for e in outgoing if e.confidence >= min_confidence]
     nodes_map = {n.id: n for n in store.get_nodes([e.target for e in outgoing])}
     for edge in outgoing:
         target_id = edge.target
@@ -290,6 +300,7 @@ def build_trace_tree(
             current_depth + 1,
             allowed_edge_types=allowed_edge_types,
             show_external=show_external,
+            min_confidence=min_confidence,
         )
         path_visited.remove(target_id)
 
@@ -307,6 +318,7 @@ def trace(
     ),
     show_structure: bool = _OPT_SHOW_STRUCTURE,
     show_external: bool = _OPT_SHOW_EXTERNAL,
+    min_confidence: float | None = _OPT_MIN_CONFIDENCE,
 ) -> None:
     """
     Trace execution flow starting from a specific code entity downwards.
@@ -326,7 +338,11 @@ def trace(
 
         if output_format == OutputFormat.MERMAID:
             nodes, edges = QueryEngine(store).get_flow_graph(
-                start, max_depth=depth, allowed_edge_types=allowed, show_external=show_external
+                start,
+                max_depth=depth,
+                allowed_edge_types=allowed,
+                show_external=show_external,
+                min_confidence=min_confidence,
             )
             if internal_only:
                 nodes, edges = _filter_internal(nodes, edges)
@@ -353,6 +369,7 @@ def trace(
                 0,
                 allowed_edge_types=allowed,
                 show_external=show_external,
+                min_confidence=min_confidence,
             )
             console.print(tree)
 
@@ -366,6 +383,7 @@ def build_impact_tree(
     current_depth: int,
     allowed_edge_types: frozenset[EdgeType] | None = None,
     show_external: bool = True,
+    min_confidence: float | None = None,
 ) -> None:
     """Cycle-safe recursive tree builder for upstream impact analysis."""
     if current_depth >= max_depth:
@@ -374,6 +392,8 @@ def build_impact_tree(
     incoming = store.get_incoming_edges(current_id)
     if allowed_edge_types is not None:
         incoming = [e for e in incoming if e.type in allowed_edge_types]
+    if min_confidence is not None:
+        incoming = [e for e in incoming if e.confidence >= min_confidence]
     nodes_map = {n.id: n for n in store.get_nodes([e.source for e in incoming])}
     for edge in incoming:
         source_id = edge.source
@@ -412,6 +432,7 @@ def build_impact_tree(
             current_depth + 1,
             allowed_edge_types=allowed_edge_types,
             show_external=show_external,
+            min_confidence=min_confidence,
         )
         path_visited.remove(source_id)
 
@@ -429,6 +450,7 @@ def impact(
     ),
     show_structure: bool = _OPT_SHOW_STRUCTURE,
     show_external: bool = _OPT_SHOW_EXTERNAL,
+    min_confidence: float | None = _OPT_MIN_CONFIDENCE,
 ) -> None:
     """
     Analyze transitive upstream impact (callers) of changing a specific code entity.
@@ -448,7 +470,11 @@ def impact(
 
         if output_format == OutputFormat.MERMAID:
             nodes, edges = QueryEngine(store).get_impact_graph(
-                target, max_depth=depth, allowed_edge_types=allowed, show_external=show_external
+                target,
+                max_depth=depth,
+                allowed_edge_types=allowed,
+                show_external=show_external,
+                min_confidence=min_confidence,
             )
             if internal_only:
                 nodes, edges = _filter_internal(nodes, edges)
@@ -475,6 +501,7 @@ def impact(
                 0,
                 allowed_edge_types=allowed,
                 show_external=show_external,
+                min_confidence=min_confidence,
             )
             console.print(tree)
 

src/cgis/query/engine.py

@@ -50,6 +50,17 @@ def _prune_external(
     return nodes, [e for e in filtered_edges if e.source in reachable and e.target in reachable]
 
 
+def _edge_accepted(
+    edge: Edge,
+    allowed_edge_types: frozenset[EdgeType] | None,
+    min_confidence: float | None,
+) -> bool:
+    """True when an edge passes the traversal filters: type allow-list and confidence floor."""
+    if allowed_edge_types is not None and edge.type not in allowed_edge_types:
+        return False
+    return not (min_confidence is not None and edge.confidence < min_confidence)
+
+
 class QueryEngine:
     """
     Performs graph traversals over the SQLite Code Graph.
@@ -66,6 +77,7 @@ def get_impact_graph(
         max_depth: int = 5,
         allowed_edge_types: frozenset[EdgeType] | None = None,
         show_external: bool = True,
+        min_confidence: float | None = None,
     ) -> tuple[list[Node], list[Edge]]:
         """
         Transitive upstream traversal (who calls me?).
@@ -78,6 +90,7 @@ def get_impact_graph(
             max_depth,
             allowed_edge_types=allowed_edge_types,
             show_external=show_external,
+            min_confidence=min_confidence,
         )
 
     def get_flow_graph(
@@ -86,6 +99,7 @@ def get_flow_graph(
         max_depth: int = 5,
         allowed_edge_types: frozenset[EdgeType] | None = None,
         show_external: bool = True,
+        min_confidence: float | None = None,
     ) -> tuple[list[Node], list[Edge]]:
         """
         Transitive downstream traversal (who do I call?).
@@ -98,6 +112,7 @@ def get_flow_graph(
             max_depth,
             allowed_edge_types=allowed_edge_types,
             show_external=show_external,
+            min_confidence=min_confidence,
         )
 
     def get_structural_graph(
@@ -118,10 +133,15 @@ def _bfs_traverse(
         max_depth: int,
         allowed_edge_types: frozenset[EdgeType] | None = None,
         show_external: bool = True,
+        min_confidence: float | None = None,
     ) -> tuple[list[Node], list[Edge]]:
         """
         Level-by-level BFS. Fetches edges for the entire frontier in one
         batch query per level — O(depth) DB roundtrips instead of O(nodes).
+
+        ``min_confidence`` prunes edges below the given confidence (e.g.
+        unresolved ``raw_call:`` edges at 0.1) so the traversal never crosses
+        them — low-confidence neighbours stay out of the result.
         """
         discovered_ids: set[str] = {start_id}
         visited_edges: dict[str, Edge] = {}
@@ -132,7 +152,7 @@ def _bfs_traverse(
             edges = get_edges_batch(current_frontier)
             next_frontier: list[str] = []
             for edge in edges:
-                if allowed_edge_types is not None and edge.type not in allowed_edge_types:
+                if not _edge_accepted(edge, allowed_edge_types, min_confidence):
                     continue
                 visited_edges[edge.id] = edge
                 neighbor_id = get_neighbor_id(edge)

tests/unit/test_cli.py

@@ -170,6 +170,23 @@ def test_trace_shows_unresolved_external_call(tmp_path: Path) -> None:
     assert "print" in result.output
 
 
+def test_trace_min_confidence_hides_low_conf_calls(tmp_path: Path) -> None:
+    """--min-confidence hides low-confidence (unresolved/raw_call) edges from the tree (#112)."""
+    (tmp_path / "mod.py").write_text("def greet(): print('hi')\n", encoding="utf-8")
+    db_file = tmp_path / "graph.db"
+    runner.invoke(app, ["ingest", str(tmp_path), "--output", str(db_file)])
+    fqn = f"{file_path_to_module_fqn('mod.py')}.greet"
+
+    shown = runner.invoke(app, ["trace", fqn, "--db", str(db_file), "--show-external"])
+    assert "print" in shown.output
+
+    hidden = runner.invoke(
+        app, ["trace", fqn, "--db", str(db_file), "--show-external", "--min-confidence", "0.9"]
+    )
+    assert hidden.exit_code == 0
+    assert "print" not in hidden.output
+
+
 def test_trace_detects_cycle(tmp_path: Path) -> None:
     """Mutually recursive functions trigger cycle detection in the trace tree."""
     code = "def func_a():\n    func_b()\n\ndef func_b():\n    func_a()\n"

tests/unit/test_sqlite_store.py

@@ -457,6 +457,42 @@ def test_flow_graph_filters_structural_edges(temp_store: SQLiteStore) -> None:
     assert filt_edges[0].type == EdgeType.CALLS
 
 
+def test_flow_graph_filters_by_min_confidence(temp_store: SQLiteStore) -> None:
+    """min_confidence drops low-confidence edges (e.g. unresolved raw_call) (#112)."""
+    nodes = [_make_node("caller"), _make_node("solid"), _make_node("weak")]
+    edges = [
+        Edge(id="c->s", source="caller", target="solid", type=EdgeType.CALLS, confidence=1.0),
+        Edge(id="c->w", source="caller", target="weak", type=EdgeType.CALLS, confidence=0.1),
+    ]
+    temp_store.save_graph(nodes, edges)
+    qe = QueryEngine(temp_store)
+
+    # No filter: both reached.
+    n_all, _ = qe.get_flow_graph("caller", max_depth=2)
+    assert {n.id for n in n_all} == {"caller", "solid", "weak"}
+
+    # min_confidence 0.5 prunes the 0.1 edge → weak is unreachable.
+    n_f, e_f = qe.get_flow_graph("caller", max_depth=2, min_confidence=0.5)
+    assert {n.id for n in n_f} == {"caller", "solid"}
+    assert all(e.confidence >= 0.5 for e in e_f)
+
+
+def test_impact_graph_filters_by_min_confidence(temp_store: SQLiteStore) -> None:
+    """min_confidence applies to upstream impact traversal too (#112)."""
+    nodes = [_make_node("strong_caller"), _make_node("weak_caller"), _make_node("target")]
+    edges = [
+        Edge(
+            id="s->t", source="strong_caller", target="target", type=EdgeType.CALLS, confidence=1.0
+        ),
+        Edge(id="w->t", source="weak_caller", target="target", type=EdgeType.CALLS, confidence=0.1),
+    ]
+    temp_store.save_graph(nodes, edges)
+    qe = QueryEngine(temp_store)
+
+    n_f, _ = qe.get_impact_graph("target", max_depth=2, min_confidence=0.5)
+    assert {n.id for n in n_f} == {"target", "strong_caller"}
+
+
 def test_impact_graph_filters_structural_edges(temp_store: SQLiteStore) -> None:
     """allowed_edge_types excludes CONTAINS/DECLARES in upstream impact traversal."""
     nodes = [_make_node("container"), _make_node("caller"), _make_node("target")]