feat(query): --seed to start the traversal at given nodes

An external ranker (embeddings, a hybrid or learned retriever, or an agent
that already knows the file) can choose where graphify query starts instead
of the question's keyword matches; the graph then does the expansion.

Each --seed value resolves exactly as affected resolves its seed
(affected.resolve_seed: node id, unique label, callable name, repo-relative
or absolute file path anchored to the graph's root), so there is one
interpretation of a node reference. Fail closed: a value that does not
resolve to exactly one node is reported and skipped; if none resolve, the
query says so and does not fall back to keyword seeds. Without --seed the
output is unchanged (rendering moved into _render_query as is).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

graphify/__main__.py

@@ -908,6 +908,8 @@ def _run_cli() -> None:
         print("  query \"<question>\"       BFS traversal of graph.json for a question")
         print("    --dfs                   use depth-first instead of breadth-first")
         print("    --context C             explicit edge-context filter (repeatable)")
+        print("    --seed X                start at node X (id, label or file path; repeatable)")
+        print("                            instead of the question's keyword matches")
         print("    --budget N              cap output at N tokens (default 2000)")
         print("    --graph <path>          path to graph.json (default graphify-out/graph.json)")
         print("  query, path and explain exit 0 with an answer, 3 when a name is ambiguous")

graphify/cli.py

@@ -143,6 +143,12 @@ def _print_cloud_cta(out_dir: "Path") -> None:
         pass
 
 
+def _graph_root_for(gp: Path) -> Path:
+    """The analysed repo root for a graph file: <root>/<GRAPHIFY_OUT_NAME>/graph.json, else the file's own directory."""
+    from graphify.paths import GRAPHIFY_OUT_NAME
+    return gp.parent.parent if gp.parent.name == GRAPHIFY_OUT_NAME else gp.parent
+
+
 def _default_graph_path() -> str:
     return str(Path(_GRAPHIFY_OUT) / "graph.json")
 
@@ -1314,7 +1320,7 @@ def dispatch_command(cmd: str) -> None:
             sys.exit(1)
     elif cmd == "query":
         if len(sys.argv) < 3:
-            print("Usage: graphify query \"<question>\" [--dfs] [--context C] [--budget N] [--graph path]", file=sys.stderr)
+            print("Usage: graphify query \"<question>\" [--dfs] [--context C] [--seed X] [--budget N] [--graph path]", file=sys.stderr)
             sys.exit(1)
         from graphify.serve import _QUERY_NO_MATCH, _query_graph_text
         from graphify.security import sanitize_label
@@ -1326,6 +1332,7 @@ def dispatch_command(cmd: str) -> None:
         budget = 2000
         graph_path = _default_graph_path()
         context_filters: list[str] = []
+        seeds: list[str] = []
         args = sys.argv[3:]
         i = 0
         while i < len(args):
@@ -1349,6 +1356,12 @@ def dispatch_command(cmd: str) -> None:
             elif args[i].startswith("--context="):
                 context_filters.append(args[i].split("=", 1)[1])
                 i += 1
+            elif args[i] == "--seed" and i + 1 < len(args):
+                seeds.append(args[i + 1])
+                i += 2
+            elif args[i].startswith("--seed="):
+                seeds.append(args[i].split("=", 1)[1])
+                i += 1
             elif args[i] == "--graph" and i + 1 < len(args):
                 graph_path = args[i + 1]
                 i += 2
@@ -1420,6 +1433,9 @@ def dispatch_command(cmd: str) -> None:
             token_budget=budget,
             context_filters=context_filters,
             graph_path=str(gp),
+            seeds=seeds or None,
+            # Same root derivation as `affected` (#2706): <root>/graphify-out/graph.json.
+            seed_root=_graph_root_for(gp),
         )
         querylog.log_query(
             kind="query",

graphify/serve.py

@@ -1388,7 +1388,30 @@ def _query_graph_text(
     token_budget: int = 2000,
     context_filters: list[str] | None = None,
     graph_path: str | None = None,
+    seeds: list[str] | None = None,
+    seed_root: "Path | None" = None,
 ) -> str:
+    # Explicit start nodes: an external ranker -- embeddings, a learned
+    # retriever, grep -- can choose where the traversal begins instead of the
+    # question's keyword matches. Each value resolves exactly as `affected`
+    # resolves its seed (node id, unique label, or repo-relative / absolute file
+    # path; `seed_root` anchors absolute paths), so there is one interpretation
+    # of a node reference. Fail closed: a value that does not resolve to exactly
+    # one node is reported and skipped, never guessed.
+    unresolved: list[str] = []
+    if seeds:
+        from graphify.affected import resolve_seed
+        start_nodes: list[str] = []
+        for value in seeds:
+            nid = resolve_seed(G, value, seed_root)
+            if nid is None:
+                unresolved.append(value)
+            elif nid not in start_nodes:
+                start_nodes.append(nid)
+        if not start_nodes:
+            return "No unique node match for --seed: " + ", ".join(seeds)
+        return _render_query(G, question, start_nodes, mode=mode, depth=depth, token_budget=token_budget,
+                             context_filters=context_filters, graph_path=graph_path, given=True, unresolved=unresolved)
     terms = _query_terms(question)
     # One graph scoring pass produces both the combined ranking (used to drive
     # the gap-based seed selection below) and the per-token singleton winners
@@ -1413,13 +1436,32 @@ def _query_graph_text(
     start_nodes = _pick_seeds(qs.ranked, G=G, best_seed_by_term=best_seed_by_term)
     if not start_nodes:
         return _QUERY_NO_MATCH
+    return _render_query(G, question, start_nodes, mode=mode, depth=depth, token_budget=token_budget,
+                         context_filters=context_filters, graph_path=graph_path)
+
+
+def _render_query(
+    G: nx.Graph,
+    question: str,
+    start_nodes: list[str],
+    *,
+    mode: str,
+    depth: int,
+    token_budget: int,
+    context_filters: list[str] | None,
+    graph_path: str | None,
+    given: bool = False,
+    unresolved: list[str] | None = None,
+) -> str:
     resolved_filters, filter_source = _resolve_context_filters(question, context_filters)
     traversal_graph = _filter_graph_by_context(_traversal_view(G), resolved_filters)
     nodes, edges = _dfs(traversal_graph, start_nodes, depth) if mode == "dfs" else _bfs(traversal_graph, start_nodes, depth)
     header_parts = [
         f"Traversal: {mode.upper()} depth={depth}",
-        f"Start: {[G.nodes[n].get('label', n) for n in start_nodes]}",
+        f"Start{' (given)' if given else ''}: {[G.nodes[n].get('label', n) for n in start_nodes]}",
     ]
+    if unresolved:
+        header_parts.append(f"Unresolved seeds (skipped): {unresolved}")
     # Name the graph this answer came from. `graphify-out/` resolves against the
     # CWD, so running a query from a parent project while thinking about a
     # vendored subproject silently answers from the wrong corpus — the output is

tests/test_query_seeds.py

@@ -0,0 +1,87 @@
+"""`graphify query --seed`: explicit start nodes chosen by the caller (an external ranker) instead of the question's
+keyword matches. A seed resolves exactly as `affected` resolves its seed, and fails closed when it does not."""
+from __future__ import annotations
+
+import json
+
+import networkx as nx
+from networkx.readwrite import json_graph
+
+import graphify.__main__ as mainmod
+from graphify.serve import _query_graph_text
+
+
+def _graph():
+    G = nx.Graph()
+    G.add_node("n1", label="extract", source_file="pkg/extract.py", source_location="L10", community=0)
+    G.add_node("n2", label="cluster", source_file="pkg/cluster.py", source_location="L5", community=0)
+    G.add_node("n3", label="build", source_file="pkg/build.py", source_location="L1", community=1)
+    G.add_node("n4", label="report", source_file="pkg/report.py", source_location="L1", community=2)
+    G.add_edge("n1", "n2", relation="calls", confidence="EXTRACTED", context="call")
+    G.add_edge("n3", "n4", relation="imports", confidence="EXTRACTED", context="import")
+    return G
+
+
+def _write(tmp_path, G):
+    out = tmp_path / "graphify-out"
+    out.mkdir()
+    graph_path = out / "graph.json"
+    graph_path.write_text(json.dumps(json_graph.node_link_data(G, edges="links")))
+    return graph_path
+
+
+def test_given_seed_replaces_keyword_seeds():
+    G = _graph()
+    plain = _query_graph_text(G, "extract", depth=1)
+    assert "Start: ['extract']" in plain and "NODE build" not in plain
+    seeded = _query_graph_text(G, "extract", depth=1, seeds=["build"])
+    assert "Start (given): ['build']" in seeded
+    assert "NODE build" in seeded and "NODE report" in seeded       # traversal from the given node
+    assert "NODE extract" not in seeded                              # the keyword match is not used
+
+
+def test_seed_by_file_path_and_by_id_dedupes_in_order():
+    G = _graph()
+    text = _query_graph_text(G, "anything", depth=1, seeds=["pkg/build.py", "n1", "n3"])
+    assert "Start (given): ['build', 'extract']" in text             # n3 is pkg/build.py: kept once, first
+
+
+def test_unresolved_seed_is_reported_and_skipped():
+    G = _graph()
+    text = _query_graph_text(G, "anything", depth=1, seeds=["no_such_node", "cluster"])
+    assert "Start (given): ['cluster']" in text
+    assert "Unresolved seeds (skipped): ['no_such_node']" in text
+
+
+def test_no_resolvable_seed_fails_closed():
+    G = _graph()
+    text = _query_graph_text(G, "extract", depth=1, seeds=["no_such_node"])
+    assert text == "No unique node match for --seed: no_such_node"   # no fallback to the keyword seeds
+
+
+def test_absolute_path_seed_resolves_against_the_graph_root(tmp_path):
+    G = _graph()
+    text = _query_graph_text(G, "anything", depth=1, seeds=[str(tmp_path / "pkg" / "report.py")], seed_root=tmp_path)
+    assert "Start (given): ['report']" in text
+
+
+def test_query_cli_seed_flag_is_repeatable(monkeypatch, tmp_path, capsys):
+    graph_path = _write(tmp_path, _graph())
+    monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None)
+    monkeypatch.setattr(mainmod.sys, "argv", ["graphify", "query", "extract", "--seed", "build", "--seed=pkg/cluster.py",
+                                              "--graph", str(graph_path)])
+    mainmod.main()
+    out = capsys.readouterr().out
+    assert "Start (given): ['build', 'cluster']" in out
+
+
+def test_query_cli_absolute_seed_uses_the_graph_location_not_cwd(monkeypatch, tmp_path, capsys):
+    graph_path = _write(tmp_path, _graph())
+    elsewhere = tmp_path / "elsewhere"
+    elsewhere.mkdir()
+    monkeypatch.chdir(elsewhere)
+    monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None)
+    monkeypatch.setattr(mainmod.sys, "argv", ["graphify", "query", "q", "--seed", str(tmp_path / "pkg" / "build.py"),
+                                              "--graph", str(graph_path)])
+    mainmod.main()
+    assert "Start (given): ['build']" in capsys.readouterr().out