fix(path): respect edge direction by default in path and shortest_path (#2487)

graphify path and the MCP shortest_path tool now build a digraph from the true _src/_tgt directions and respect edge direction by default, so a returned path no longer traverses edges backwards. --undirected (CLI) / undirected=true (MCP) opt out; no directed path is reported plainly instead of silently returning a reversed one. Thanks @luliaz0601. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-08-05 22:08:29 +01:00
parent e300587439
commit 94ebee18b7
3 changed files with 170 additions and 76 deletions
@@ -34,6 +34,11 @@ def _load_graph(graph_path: str) -> nx.Graph:
        data = json.loads(safe.read_text(encoding="utf-8"))
        if "links" not in data and "edges" in data:
            data = dict(data, links=data["edges"])
+        # Stash the on-disk logical flag before the load-time override below:
+        # `directed: True` exists only so renderers can recover stored arc
+        # order (#2309); tools that care about logical direction (#2487) must
+        # not mistake the override for graph truth.
+        _logical_directed = bool(data.get("directed", False))
        data = {**data, "directed": True}
        try:
            from graphify.build import graph_has_legacy_ids as _legacy
@@ -49,6 +54,7 @@ def _load_graph(graph_path: str) -> nx.Graph:
            G = json_graph.node_link_graph(data, edges="links")
        except TypeError:
            G = json_graph.node_link_graph(data)
+        G.graph["_logical_directed"] = _logical_directed
        # Attach the work-memory overlay (derived sidecar next to graph.json) so
        # the query/MCP read surface can annotate NODE lines display-only. Empty
        # when no sidecar exists, leaving un-annotated output byte-identical.
@@ -1230,6 +1236,107 @@ def find_node_ambiguity(G: nx.Graph, label: str) -> list[str]:
    return []


+def _shortest_path_text(G: nx.Graph, arguments: dict) -> str:
+    """Body of the `shortest_path` MCP tool (module-level so tests can call it
+    without an mcp install).
+
+    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:
+        return f"No node matching source '{arguments['source']}' found."
+    if not tgt_scored:
+        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).
+    if src_nid == tgt_nid:
+        return (
+            f"'{arguments['source']}' and '{arguments['target']}' both resolved to "
+            f"the same node '{src_nid}'. Use a more specific label or the exact node ID."
+        )
+    warnings: list[str] = []
+    for name, scored, nid in (
+        ("source", src_scored, src_nid),
+        ("target", tgt_scored, tgt_nid),
+    ):
+        # Only meaningful when the raw score head is what got picked — a
+        # full-token override was chosen on token coverage, not score.
+        if len(scored) >= 2 and nid == scored[0][1]:
+            top, runner = scored[0][0], scored[1][0]
+            if top > 0 and (top - runner) / top < 0.10:
+                warnings.append(
+                    f"warning: {name} match was ambiguous "
+                    f"(top score {top:g}, runner-up {runner:g})"
+                )
+    max_hops = int(arguments.get("max_hops", 8))
+    undirected = bool(arguments.get("undirected", False))
+    try:
+        # Deterministic path (#2074): the hash-seeded undirected view picked an
+        # arbitrary route among equal-length paths. Build a sorted, materialized
+        # graph so the chosen path is canonical. Serve's shared G is left
+        # untouched (its degree feeds query-seed tie-breaks).
+        if undirected:
+            _und = nx.Graph()
+            _und.add_nodes_from(sorted(G.nodes))
+            _und.add_edges_from(sorted((min(u, v), max(u, v)) for u, v in G.edges()))
+            path_nodes = nx.shortest_path(_und, src_nid, tgt_nid)
+        else:
+            # Directed by default (#2487). True direction is NOT raw arc
+            # order: legacy canonicalized files persist a flipped arc with
+            # _src/_tgt markers (#2309), so build the digraph from _src/_tgt
+            # (falling back to the loaded arc) rather than to_directed().
+            _dg = nx.DiGraph()
+            _dg.add_nodes_from(sorted(G.nodes))
+            _dg.add_edges_from(sorted(
+                (d.get("_src", u), d.get("_tgt", v)) for u, v, d in G.edges(data=True)
+            ))
+            path_nodes = nx.shortest_path(_dg, src_nid, tgt_nid)
+    except (nx.NetworkXNoPath, nx.NodeNotFound):
+        src_label = G.nodes[src_nid].get("label", src_nid)
+        tgt_label = G.nodes[tgt_nid].get("label", tgt_nid)
+        if undirected:
+            return f"No path found between '{src_label}' and '{tgt_label}'."
+        return (
+            f"No directed path found between '{src_label}' and '{tgt_label}'. "
+            "Retry with undirected=true to search ignoring edge direction."
+        )
+    hops = len(path_nodes) - 1
+    if hops > max_hops:
+        return f"Path exceeds max_hops={max_hops} ({hops} hops found)."
+    segments = []
+    for i in range(len(path_nodes) - 1):
+        u, v = path_nodes[i], path_nodes[i + 1]
+        # Report the actual stored relation(s), never a fabricated `calls`;
+        # fall back to an honest "related" when the edge has no relation (#2074).
+        # Direction truth lives in the per-link _src/_tgt markers (#2309): a
+        # legacy canonicalized file can persist a flipped arc, so classify each
+        # hop by _src (falling back to the arc tail) instead of raw arc order.
+        fwd, bwd = [], []
+        for a, b in ((u, v), (v, u)):
+            if G.has_edge(a, b):
+                for d in edge_datas(G, a, b):
+                    (fwd if d.get("_src", a) == u else bwd).append(d)
+        datas = fwd or bwd
+        forward = bool(fwd)
+        rels = sorted({d.get("relation") for d in datas if d.get("relation")})
+        rel = "/".join(rels) if rels else "related"
+        confs = sorted({d.get("confidence") for d in datas if d.get("confidence")})
+        conf_str = f" [{'/'.join(confs)}]" if confs else ""
+        if i == 0:
+            segments.append(G.nodes[u].get("label", u))
+        if forward:
+            segments.append(f"--{rel}{conf_str}--> {G.nodes[v].get('label', v)}")
+        else:
+            segments.append(f"<--{rel}{conf_str}-- {G.nodes[v].get('label', v)}")
+    prefix = ("\n".join(warnings) + "\n") if warnings else ""
+    return prefix + f"Shortest path ({hops} hops):\n  " + " ".join(segments)
+
+
 def _filter_blank_stdin() -> None:
    """Filter blank lines from stdin before MCP reads it.

@@ -1411,13 +1518,18 @@ def _build_server(graph_path: str):
            ),
            types.Tool(
                name="shortest_path",
-                description="Find the shortest path between two concepts in the knowledge graph.",
+                description=(
+                    "Find the shortest path between two concepts in the knowledge graph. "
+                    "Follows stored edge direction by default; set undirected=true to ignore it."
+                ),
                inputSchema={
                    "type": "object",
                    "properties": {
                        "source": {"type": "string", "description": "Source concept label or keyword"},
                        "target": {"type": "string", "description": "Target concept label or keyword"},
                        "max_hops": {"type": "integer", "default": 8, "description": "Maximum hops to consider"},
+                        "undirected": {"type": "boolean", "default": False,
+                                       "description": "Ignore stored edge direction when searching"},
                    },
                    "required": ["source", "target"],
                },
@@ -1623,74 +1735,7 @@ def _build_server(graph_path: str):
        )

    def _tool_shortest_path(arguments: dict) -> str:
-        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:
-            return f"No node matching source '{arguments['source']}' found."
-        if not tgt_scored:
-            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).
-        if src_nid == tgt_nid:
-            return (
-                f"'{arguments['source']}' and '{arguments['target']}' both resolved to "
-                f"the same node '{src_nid}'. Use a more specific label or the exact node ID."
-            )
-        warnings: list[str] = []
-        for name, scored, nid in (
-            ("source", src_scored, src_nid),
-            ("target", tgt_scored, tgt_nid),
-        ):
-            # Only meaningful when the raw score head is what got picked — a
-            # full-token override was chosen on token coverage, not score.
-            if len(scored) >= 2 and nid == scored[0][1]:
-                top, runner = scored[0][0], scored[1][0]
-                if top > 0 and (top - runner) / top < 0.10:
-                    warnings.append(
-                        f"warning: {name} match was ambiguous "
-                        f"(top score {top:g}, runner-up {runner:g})"
-                    )
-        max_hops = int(arguments.get("max_hops", 8))
-        try:
-            # Deterministic path (#2074): the hash-seeded undirected view picked an
-            # arbitrary route among equal-length paths. Build a sorted, materialized
-            # undirected graph so the chosen path is canonical. Serve's shared G is
-            # left untouched (its degree feeds query-seed tie-breaks).
-            _und = nx.Graph()
-            _und.add_nodes_from(sorted(G.nodes))
-            _und.add_edges_from(sorted((min(u, v), max(u, v)) for u, v in G.edges()))
-            path_nodes = nx.shortest_path(_und, src_nid, tgt_nid)
-        except (nx.NetworkXNoPath, nx.NodeNotFound):
-            return f"No path found between '{G.nodes[src_nid].get('label', src_nid)}' and '{G.nodes[tgt_nid].get('label', tgt_nid)}'."
-        hops = len(path_nodes) - 1
-        if hops > max_hops:
-            return f"Path exceeds max_hops={max_hops} ({hops} hops found)."
-        segments = []
-        for i in range(len(path_nodes) - 1):
-            u, v = path_nodes[i], path_nodes[i + 1]
-            # Report the actual stored relation(s), never a fabricated `calls`;
-            # fall back to an honest "related" when the edge has no relation (#2074).
-            if G.has_edge(u, v):
-                datas = edge_datas(G, u, v)
-                forward = True
-            else:
-                datas = edge_datas(G, v, u)
-                forward = False
-            rels = sorted({d.get("relation") for d in datas if d.get("relation")})
-            rel = "/".join(rels) if rels else "related"
-            confs = sorted({d.get("confidence") for d in datas if d.get("confidence")})
-            conf_str = f" [{'/'.join(confs)}]" if confs else ""
-            if i == 0:
-                segments.append(G.nodes[u].get("label", u))
-            if forward:
-                segments.append(f"--{rel}{conf_str}--> {G.nodes[v].get('label', v)}")
-            else:
-                segments.append(f"<--{rel}{conf_str}-- {G.nodes[v].get('label', v)}")
-        prefix = ("\n".join(warnings) + "\n") if warnings else ""
-        return prefix + f"Shortest path ({hops} hops):\n  " + " ".join(segments)
+        return _shortest_path_text(G, arguments)

    def _tool_list_prs(arguments: dict) -> str:
        from graphify.prs import fetch_prs, fetch_worktrees, format_prs_text, _detect_default_branch
@@ -30,10 +30,10 @@ def _write_graph(tmp_path):
    return p


-def _run(monkeypatch, graph_path, src, tgt, capsys):
+def _run(monkeypatch, graph_path, src, tgt, capsys, *extra):
    monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None)
    monkeypatch.setattr(mainmod.sys, "argv",
-        ["graphify", "path", src, tgt, "--graph", str(graph_path)])
+        ["graphify", "path", src, tgt, "--graph", str(graph_path), *extra])
    mainmod.main()
    return capsys.readouterr().out

@@ -47,7 +47,10 @@ def test_forward_arrow(monkeypatch, tmp_path, capsys):

 def test_reverse_arrow(monkeypatch, tmp_path, capsys):
    p = _write_graph(tmp_path)
-    out = _run(monkeypatch, p, "validateSanitySession", "createPatchHandler", capsys)
+    # #2487: path is directed by default, so walking the stored edge backwards
+    # needs the --undirected opt-out to exercise the reverse-arrow rendering.
+    out = _run(monkeypatch, p, "validateSanitySession", "createPatchHandler", capsys,
+               "--undirected")
    assert "Shortest path (1 hops):" in out
    assert "validateSanitySession() <--calls [EXTRACTED]-- createPatchHandler()" in out
    assert "validateSanitySession() --calls [EXTRACTED]--> createPatchHandler()" not in out
@@ -106,7 +109,10 @@ def test_endpoint_falls_back_to_score_head(monkeypatch, tmp_path, capsys):
    with pytest.raises(SystemExit) as exc_info:
        mainmod.main()
    assert exc_info.value.code == 0
-    assert "No path found" in capsys.readouterr().out
+    # #2487: path is directed by default, so the flagless no-path message is
+    # now the directed one (the decoy/target components stay disconnected
+    # either way — this test is about endpoint resolution, not direction).
+    assert "No directed path found" in capsys.readouterr().out


 # ── #2074: deterministic route + honest edge relation ────────────────────────
@@ -229,7 +235,9 @@ def test_path_direction_recovered_from_src_tgt_markers(monkeypatch, tmp_path, ca
    """#2309: a hop over a link stored in flipped order must render the TRUE
    direction from its _src/_tgt markers, not the persisted arc order."""
    p = _flipped_marker_graph(tmp_path)
-    out = _run(monkeypatch, p, "ingest", "draft-generator", capsys)
+    # #2487: the two hops point in opposite TRUE directions (ingest->logger,
+    # draft->logger), so this mixed-direction route only exists undirected.
+    out = _run(monkeypatch, p, "ingest", "draft-generator", capsys, "--undirected")
    assert "Shortest path (2 hops):" in out
    assert "ingest.ts --calls [EXTRACTED]--> logger.ts" in out
    # True direction is draft -> logger, so the logger->draft hop is reversed.
@@ -255,8 +263,9 @@ def test_path_canonical_marker_graph_still_forward(monkeypatch, tmp_path, capsys
    gp.write_text(json.dumps(data))
    out = _run(monkeypatch, gp, "Alpha", "Beta", capsys)
    assert "Alpha --calls [EXTRACTED]--> Beta" in out
-    # And walking the same edge backwards still reverses the arrow.
-    out = _run(monkeypatch, gp, "Beta", "Alpha", capsys)
+    # And walking the same edge backwards still reverses the arrow (#2487:
+    # backwards traversal now requires the --undirected opt-out).
+    out = _run(monkeypatch, gp, "Beta", "Alpha", capsys, "--undirected")
    assert "Beta <--calls [EXTRACTED]-- Alpha" in out


@@ -29,6 +29,7 @@ from graphify.serve import (
    _load_graph,
    _community_header,
    _search_tokens,
+    _shortest_path_text,
 )


@@ -1386,3 +1387,42 @@ def test_subgraph_to_text_honors_valid_src_tgt_direction():
    out = _subgraph_to_text(G, {"caller", "callee"}, [("callee", "caller")])
    edge_line = next(l for l in out.splitlines() if l.startswith("EDGE"))
    assert "caller --calls" in edge_line and "--> callee" in edge_line
+
+
+# --- _shortest_path_text direction (#2487) ---
+
+def _directed_chain() -> nx.DiGraph:
+    """alpha --calls--> beta --calls--> gamma, as _load_graph would load it
+    (directed storage, arc order = true direction on post-#563 files)."""
+    G = nx.DiGraph()
+    for n in ("alpha", "beta", "gamma"):
+        G.add_node(n, label=n)
+    G.add_edge("alpha", "beta", relation="calls")
+    G.add_edge("beta", "gamma", relation="calls")
+    return G
+
+
+def test_shortest_path_tool_directed_respects_direction():
+    out = _shortest_path_text(_directed_chain(), {"source": "alpha", "target": "gamma"})
+    assert "Shortest path (2 hops)" in out
+    assert out.count("-->") == 2
+    assert "<--" not in out
+
+
+def test_shortest_path_tool_directed_backwards_is_no_path():
+    # Directed is the default (#2487): walking the chain backwards must report
+    # no directed path, with the undirected opt-out hint, not a reversed path.
+    out = _shortest_path_text(_directed_chain(), {"source": "gamma", "target": "alpha"})
+    assert "No directed path found" in out
+    assert "undirected=true" in out
+    assert "-->" not in out
+    assert "<--" not in out
+
+
+def test_shortest_path_tool_undirected_opt_in():
+    out = _shortest_path_text(
+        _directed_chain(), {"source": "gamma", "target": "alpha", "undirected": True}
+    )
+    assert "Shortest path (2 hops)" in out
+    assert out.count("<--calls--") == 2
+    assert "-->" not in out