fix(path): resolve exact node IDs before fuzzy matching

graphify/cli.py

@@ -1405,7 +1405,7 @@ def dispatch_command(cmd: str) -> None:
                 file=sys.stderr,
             )
             sys.exit(1)
-        from graphify.serve import _pick_scored_endpoint, _score_nodes
+        from graphify.serve import _resolve_path_endpoint
         from networkx.readwrite import json_graph
         import networkx as _nx
 
@@ -1456,16 +1456,14 @@ def dispatch_command(cmd: str) -> None:
             G = json_graph.node_link_graph(_raw, edges="links")
         except TypeError:
             G = json_graph.node_link_graph(_raw)
-        src_scored = _score_nodes(G, [t.lower() for t in source_label.split()])
-        tgt_scored = _score_nodes(G, [t.lower() for t in target_label.split()])
-        if not src_scored:
+        src_nid, src_scored = _resolve_path_endpoint(G, source_label)
+        tgt_nid, tgt_scored = _resolve_path_endpoint(G, target_label)
+        if src_nid is None:
             print(f"No node matching '{source_label}' found.", file=sys.stderr)
             sys.exit(1)
-        if not tgt_scored:
+        if tgt_nid is None:
             print(f"No node matching '{target_label}' found.", file=sys.stderr)
             sys.exit(1)
-        src_nid = _pick_scored_endpoint(G, src_scored, source_label)
-        tgt_nid = _pick_scored_endpoint(G, tgt_scored, target_label)
         # Ambiguity guard: when both queries resolve to the same node, the
         # shortest path is trivially zero hops, which is almost never what the
         # caller wanted (see bug #828).

graphify/serve.py

@@ -653,6 +653,18 @@ def _pick_scored_endpoint(G: nx.Graph, scored: list[tuple[float, str]], query: s
     return scored[0][1]
 
 
+def _resolve_path_endpoint(
+    G: nx.Graph, query: str
+) -> tuple[str | None, list[tuple[float, str]]]:
+    """Resolve a path endpoint, giving an exact graph node ID absolute priority."""
+    if query in G:
+        return query, [(float("inf"), query)]
+    scored = _score_nodes(G, [t.lower() for t in query.split()])
+    if not scored:
+        return None, []
+    return _pick_scored_endpoint(G, scored, query), scored
+
+
 def _pick_seeds(
     scored: list[tuple[float, str]],
     max_k: int = 3,
@@ -1369,14 +1381,12 @@ def _shortest_path_text(G: nx.Graph, arguments: dict) -> str:
     Directed by default (#2487): the returned path must follow stored
     caller→callee direction; pass ``undirected=True`` to ignore it.
     """
-    src_scored = _score_nodes(G, [t.lower() for t in arguments["source"].split()])
-    tgt_scored = _score_nodes(G, [t.lower() for t in arguments["target"].split()])
-    if not src_scored:
+    src_nid, src_scored = _resolve_path_endpoint(G, arguments["source"])
+    tgt_nid, tgt_scored = _resolve_path_endpoint(G, arguments["target"])
+    if src_nid is None:
         return f"No node matching source '{arguments['source']}' found."
-    if not tgt_scored:
+    if tgt_nid is None:
         return f"No node matching target '{arguments['target']}' found."
-    src_nid = _pick_scored_endpoint(G, src_scored, arguments["source"])
-    tgt_nid = _pick_scored_endpoint(G, tgt_scored, arguments["target"])
     # Ambiguity guard: when both queries resolve to the same node, the
     # shortest path is trivially zero hops, which is almost never what the
     # caller wanted (see bug #828).

tests/test_path_cli.py

@@ -56,6 +56,30 @@ def test_reverse_arrow(monkeypatch, tmp_path, capsys):
     assert "validateSanitySession() --calls [EXTRACTED]--> createPatchHandler()" not in out
 
 
+def test_exact_node_ids_bypass_fuzzy_endpoint_scoring(monkeypatch, tmp_path, capsys):
+    """Exact graph node IDs must resolve directly, even when labels are stronger fuzzy matches."""
+    graph_data = {
+        "directed": True, "multigraph": False, "graph": {},
+        "nodes": [
+            {"id": "source_exact_id", "label": "Actual Source", "community": 0},
+            {"id": "target_exact_id", "label": "Actual Target", "community": 0},
+            {"id": "source_decoy", "label": "source exact id", "community": 1},
+            {"id": "target_decoy", "label": "target exact id", "community": 1},
+        ],
+        "links": [
+            {"source": "source_exact_id", "target": "target_exact_id",
+             "relation": "calls", "confidence": "EXTRACTED"},
+        ],
+    }
+    p = tmp_path / "graph.json"
+    p.write_text(json.dumps(graph_data))
+
+    out = _run(monkeypatch, p, "source_exact_id", "target_exact_id", capsys)
+
+    assert "Shortest path (1 hops):" in out
+    assert "Actual Source --calls [EXTRACTED]--> Actual Target" in out
+
+
 def _write_misranking_graph(tmp_path):
     """Graph where IDF scoring ranks a partial-token decoy above the full match.