diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c69e8650e..ab9263ce6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,4 +61,4 @@ jobs: run: pnpm test - name: Test Python skill helpers - run: python -m unittest tests.skill.understand.test_merge_batch_graphs tests.skill.understand.test_merge_subdomain_graphs tests.skill.knowledge.test_parse_knowledge_base -v + run: python -m unittest tests.skill.understand.test_merge_batch_graphs tests.skill.understand.test_merge_subdomain_graphs tests.skill.knowledge.test_parse_knowledge_base tests.skill.diff.test_graph_query_helpers tests.skill.diff.test_graph_query -v diff --git a/tests/skill/diff/test_graph_query.py b/tests/skill/diff/test_graph_query.py new file mode 100644 index 000000000..f87c9b019 --- /dev/null +++ b/tests/skill/diff/test_graph_query.py @@ -0,0 +1,619 @@ +#!/usr/bin/env python3 +""" +test_graph_query.py — Tests for the optional graph query backend. + +Run from the repo root: + python -m unittest tests.skill.diff.test_graph_query -v + +Every test is skipped unless a FalkorDB backend is importable, so this is a +no-op on a checkout without the optional dependency installed. Install it with +`pip install falkordblite` (needs Python >= 3.12), or point UA_FALKORDB_URL at a +running instance. + +The graphs here are synthetic and small so the expected traversal results can be +worked out by hand rather than pinned to a snapshot of some real repo. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +_HERE = Path(__file__).resolve().parent +_REPO_ROOT = _HERE.parent.parent.parent +_SCRIPT = ( + _REPO_ROOT + / "understand-anything-plugin" + / "skills" + / "understand-diff" + / "graph-query.py" +) + + +def _backend_available() -> bool: + if os.environ.get("UA_FALKORDB_URL"): + return True + try: + import redislite.falkordb_client # noqa: F401 + except Exception: + return False + return True + + +# ── Fixtures ────────────────────────────────────────────────────────────── +# A deliberate dependency chain, so blast radius is countable by hand: +# +# types.ts <-imports- a.ts +# <-imports- b.ts <-imports- c.ts <-imports- d.ts +# +# a.ts contains runA, which calls runB in b.ts, which calls runC in c.ts. + + +def _node(node_id: str, node_type: str, name: str, file_path: str) -> dict: + return { + "id": node_id, + "type": node_type, + "name": name, + "filePath": file_path, + "summary": f"{name} defined in {file_path}", + "tags": [], + "complexity": "simple", + } + + +def _edge(source: str, target: str, edge_type: str) -> dict: + return { + "source": source, + "target": target, + "type": edge_type, + "direction": "forward", + "weight": 1.0, + } + + +def _chain_graph() -> dict: + files = ["types.ts", "a.ts", "b.ts", "c.ts", "d.ts"] + nodes = [_node(f"file:src/{f}", "file", f, f"src/{f}") for f in files] + # A second file with the same basename, imported by e.ts only. Seeding by + # name picks up both; seeding by path must not. + nodes.append(_node("file:src/nested/types.ts", "file", "types.ts", + "src/nested/types.ts")) + nodes.append(_node("file:src/e.ts", "file", "e.ts", "src/e.ts")) + nodes += [ + _node("function:src/a.ts:runA", "function", "runA", "src/a.ts"), + _node("function:src/b.ts:runB", "function", "runB", "src/b.ts"), + _node("function:src/c.ts:runC", "function", "runC", "src/c.ts"), + ] + edges = [ + _edge("file:src/e.ts", "file:src/nested/types.ts", "imports"), + _edge("file:src/a.ts", "file:src/types.ts", "imports"), + _edge("file:src/b.ts", "file:src/types.ts", "imports"), + _edge("file:src/c.ts", "file:src/b.ts", "imports"), + _edge("file:src/d.ts", "file:src/c.ts", "imports"), + _edge("file:src/a.ts", "function:src/a.ts:runA", "contains"), + _edge("file:src/b.ts", "function:src/b.ts:runB", "contains"), + _edge("file:src/c.ts", "function:src/c.ts:runC", "contains"), + _edge("function:src/a.ts:runA", "function:src/b.ts:runB", "calls"), + _edge("function:src/b.ts:runB", "function:src/c.ts:runC", "calls"), + ] + layers = [ + {"id": "core", "name": "Core", "description": "shared types", + "nodeIds": ["file:src/types.ts", "file:src/a.ts"]}, + {"id": "edge", "name": "Edge", "description": "leaves", + "nodeIds": ["file:src/d.ts"]}, + ] + return {"version": "1.0.0", "project": {"name": "chain"}, + "nodes": nodes, "edges": edges, "layers": layers} + + +def _member_graph(name: str) -> dict: + return { + "version": "1.0.0", + "project": {"name": name}, + "nodes": [ + _node(f"file:src/{name}.ts", "file", f"{name}.ts", f"src/{name}.ts"), + _node(f"function:src/{name}.ts:go", "function", "go", f"src/{name}.ts"), + ], + "edges": [_edge(f"file:src/{name}.ts", f"function:src/{name}.ts:go", "contains")], + } + + +# ── Harness ─────────────────────────────────────────────────────────────── + +@unittest.skipUnless(_backend_available(), "no FalkorDB backend installed") +class GraphQueryTestCase(unittest.TestCase): + """Shared plumbing: a temp project holding a knowledge graph.""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) + self.addCleanup(self._tmp.cleanup) + + def write_graph(self, graph: dict, root: Path | None = None) -> Path: + root = root or self.root + (root / ".ua").mkdir(parents=True, exist_ok=True) + path = root / ".ua" / "knowledge-graph.json" + path.write_text(json.dumps(graph, indent=2)) + return path + + def run_cli(self, *args: str, expect_success: bool = True, + stdin: str | None = None, env: dict | None = None, + root: Path | None = None) -> subprocess.CompletedProcess: + environ = dict(os.environ) + environ.pop("UA_EMBED_URL", None) # semantic search is tested separately + environ.update(env or {}) + proc = subprocess.run( + [sys.executable, str(_SCRIPT), *args, "--root", str(root or self.root)], + capture_output=True, text=True, input=stdin, env=environ, + ) + if expect_success: + self.assertEqual(proc.returncode, 0, f"{args} failed:\n{proc.stderr[-800:]}") + return proc + + def cli_json(self, *args: str, **kwargs): + return json.loads(self.run_cli(*args, **kwargs).stdout) + + +# ── Traversal correctness ───────────────────────────────────────────────── + +class TraversalTests(GraphQueryTestCase): + def setUp(self) -> None: + super().setUp() + self.write_graph(_chain_graph()) + + def test_blast_radius_follows_the_import_chain(self) -> None: + """a and b import types directly; c and d reach it in 2 and 3 hops.""" + self.assertEqual( + self.cli_json("blast-radius", "--path", "src/types.ts", "--hops", "3"), + ["file:src/a.ts", "file:src/b.ts", "file:src/c.ts", "file:src/d.ts"], + ) + + def test_blast_radius_by_path_is_scoped_to_one_file(self) -> None: + """Basenames are not unique, so a path must not seed from namesakes. + + Two different files are named types.ts here; seeding by name unions both + and overstates the impact. + """ + by_path = self.cli_json("blast-radius", "--path", "src/types.ts", "--hops", "3") + self.assertEqual(by_path, ["file:src/a.ts", "file:src/b.ts", + "file:src/c.ts", "file:src/d.ts"]) + + by_name = self.cli_json("blast-radius", "--name", "types.ts", "--hops", "1") + self.assertIn("file:src/e.ts", by_name) # only the namesake reaches this + self.assertNotIn("file:src/e.ts", + self.cli_json("blast-radius", "--path", "src/types.ts", + "--hops", "1")) + + def test_blast_radius_needs_a_path_or_a_name(self) -> None: + proc = self.run_cli("blast-radius", expect_success=False) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("path", proc.stdout + proc.stderr) + + def test_blast_radius_respects_the_hop_limit(self) -> None: + self.assertEqual( + self.cli_json("blast-radius", "--path", "src/types.ts", "--hops", "1"), + ["file:src/a.ts", "file:src/b.ts"], + ) + + def test_nodes_for_file_returns_the_file_and_its_members(self) -> None: + ids = [n["id"] for n in self.cli_json("nodes-for-file", "--q", "src/a.ts")] + self.assertEqual(ids, ["file:src/a.ts", "function:src/a.ts:runA"]) + + def test_nodes_for_file_matches_a_longer_caller_path(self) -> None: + """Callers may pass a path prefixed by the repo directory.""" + ids = [n["id"] for n in + self.cli_json("nodes-for-file", "--q", "some/prefix/src/a.ts")] + self.assertIn("file:src/a.ts", ids) + + def test_calls_from_and_calls_to_are_inverses(self) -> None: + self.assertEqual( + self.cli_json("calls-from", "--name", "runA", "--hops", "2"), + ["function:src/b.ts:runB", "function:src/c.ts:runC"], + ) + self.assertEqual( + self.cli_json("calls-to", "--name", "runC", "--hops", "2"), + ["function:src/a.ts:runA", "function:src/b.ts:runB"], + ) + + def test_search_matches_name_and_path(self) -> None: + self.assertTrue(self.cli_json("search", "--q", "runA")) + self.assertTrue(self.cli_json("search", "--q", "src/d.ts")) + + def test_batch_answers_several_questions_in_order(self) -> None: + out = self.cli_json("batch", "--q", json.dumps([ + {"op": "blast-radius", "path": "src/types.ts", "hops": 1}, + {"op": "calls-from", "name": "runA", "hops": 1}, + ])) + self.assertEqual(out[0], ["file:src/a.ts", "file:src/b.ts"]) + self.assertEqual(out[1], ["function:src/b.ts:runB"]) + + def test_batch_accepts_stdin(self) -> None: + out = self.cli_json( + "batch", stdin=json.dumps([{"op": "blast-radius", "path": "src/types.ts", + "hops": 1}])) + self.assertEqual(out[0], ["file:src/a.ts", "file:src/b.ts"]) + + def test_layers_for_reports_only_matching_layers(self) -> None: + """Step 6 of the skill. Read from the JSON, so no sync involvement.""" + out = self.cli_json("layers-for", "--ids", "file:src/types.ts,file:src/d.ts") + self.assertEqual([l["name"] for l in out], ["Core", "Edge"]) + self.assertEqual(out[0]["matched"], ["file:src/types.ts"]) + + self.assertEqual(self.cli_json("layers-for", "--ids", "file:src/b.ts"), []) + + def test_search_honours_an_explicit_limit(self) -> None: + """A limit passed in a batch spec must reach the query. + + It used to be dropped, so a caller asking for 2 results silently got 25 -- + the opposite of what a context-conscious skill wants. + """ + out = self.cli_json("batch", "--q", json.dumps([ + {"op": "search", "q": "src", "limit": 2}, + {"op": "search", "q": "src"}, + ])) + self.assertEqual(len(out[0]), 2) + self.assertGreater(len(out[1]), 2) + + def test_ids_are_not_namespaced_for_a_single_repo(self) -> None: + """Single-repo ids must stay byte-identical to the JSON's own ids.""" + for node_id in self.cli_json("blast-radius", "--path", "src/types.ts"): + self.assertNotIn("::", node_id) + + +# ── Incremental sync ────────────────────────────────────────────────────── + +class SyncTests(GraphQueryTestCase): + def setUp(self) -> None: + super().setUp() + self.graph = _chain_graph() + self.path = self.write_graph(self.graph) + self.baseline = self.cli_json("stats") + + def test_first_sync_is_full(self) -> None: + self.assertEqual(self.baseline["sync"]["mode"], "full") + self.assertEqual(self.baseline["nodes"], 10) + self.assertEqual(self.baseline["edges"], 10) + + def test_unchanged_graph_is_not_resynced(self) -> None: + again = self.cli_json("stats") + self.assertEqual(again["sync"]["mode"], "cached") + self.assertEqual(again["sync"]["files"], 0) + self.assertEqual(again["nodes"], self.baseline["nodes"]) + self.assertEqual(again["edges"], self.baseline["edges"]) + + def test_editing_one_file_resyncs_only_that_file(self) -> None: + for node in self.graph["nodes"]: + if node["filePath"] == "src/b.ts": + node["summary"] += " [edited]" + self.write_graph(self.graph) + + after = self.cli_json("stats") + self.assertEqual(after["sync"]["mode"], "incremental") + self.assertEqual(after["sync"]["files"], 1) + + def test_incremental_sync_preserves_incoming_edges(self) -> None: + """DETACH DELETE drops edges from untouched files; they must come back.""" + for node in self.graph["nodes"]: + if node["filePath"] == "src/b.ts": + node["summary"] += " [edited]" + self.write_graph(self.graph) + after = self.cli_json("stats") + + self.assertEqual(after["nodes"], self.baseline["nodes"]) + self.assertEqual(after["edges"], self.baseline["edges"]) + # c.ts -> b.ts is an edge owned by an untouched file. + self.assertIn("file:src/c.ts", + self.cli_json("blast-radius", "--path", "src/b.ts", "--hops", "1")) + + def test_removing_a_file_removes_its_nodes(self) -> None: + dropped = {n["id"] for n in self.graph["nodes"] if n["filePath"] == "src/d.ts"} + self.graph["nodes"] = [n for n in self.graph["nodes"] + if n["filePath"] != "src/d.ts"] + self.graph["edges"] = [e for e in self.graph["edges"] + if e["source"] not in dropped + and e["target"] not in dropped] + self.write_graph(self.graph) + + after = self.cli_json("stats") + self.assertEqual(after["nodes"], self.baseline["nodes"] - len(dropped)) + self.assertEqual(self.cli_json("nodes-for-file", "--q", "src/d.ts"), []) + + def test_adding_only_an_edge_is_synced(self) -> None: + """A commit that adds an import and changes nothing else must be seen. + + Digests cover each file's incident edges for exactly this reason. When + they covered only node fields, this was invisible: every digest matched, + sync reported "cached", and the new edge never reached the graph. + """ + self.graph["edges"].append( + _edge("file:src/d.ts", "file:src/types.ts", "imports")) + self.write_graph(self.graph) + + after = self.cli_json("stats") + self.assertEqual(after["sync"]["mode"], "incremental") + self.assertEqual(after["edges"], self.baseline["edges"] + 1) + # d.ts now depends on types.ts directly rather than only through c.ts + self.assertIn("file:src/d.ts", + self.cli_json("blast-radius", "--path", "src/types.ts", + "--hops", "1")) + + def test_removing_only_an_edge_is_synced(self) -> None: + self.graph["edges"] = [e for e in self.graph["edges"] + if not (e["source"] == "file:src/a.ts" + and e["target"] == "file:src/types.ts")] + self.write_graph(self.graph) + + after = self.cli_json("stats") + self.assertEqual(after["edges"], self.baseline["edges"] - 1) + self.assertNotIn("file:src/a.ts", + self.cli_json("blast-radius", "--path", "src/types.ts")) + + def test_changing_an_edge_property_is_synced(self) -> None: + for edge in self.graph["edges"]: + if edge["type"] == "imports": + edge["weight"] = 0.25 + break + self.write_graph(self.graph) + self.assertEqual(self.cli_json("stats")["sync"]["mode"], "incremental") + + def test_a_stamp_without_nodes_is_pruned(self) -> None: + """A stamp whose nodes are gone must not mask the file on the next sync. + + Simulates a sync interrupted after stamping: the stamp says the file is + present while its nodes are not, which would otherwise be reported as + 'cached' forever. + """ + self.cli_json("cypher", "--q", + "MATCH (n:Node) WHERE n.__key = 'src/b.ts' DELETE n") + after = self.cli_json("stats") + self.assertEqual(after["sync"]["mode"], "incremental") + self.assertEqual(after["nodes"], self.baseline["nodes"]) + self.assertEqual(after["edges"], self.baseline["edges"]) + + def test_node_without_a_file_path_still_loads(self) -> None: + self.graph["nodes"].append({ + "id": "concept:orphan", "type": "concept", "name": "Orphan", + "summary": "no filePath", "tags": [], "complexity": "simple", + }) + self.write_graph(self.graph) + + self.assertEqual(self.cli_json("stats")["nodes"], self.baseline["nodes"] + 1) + self.assertTrue(self.cli_json("search", "--q", "Orphan")) + + +# ── Workspaces ──────────────────────────────────────────────────────────── + +class WorkspaceTests(GraphQueryTestCase): + """Repos stay in separate graphs; only their topology is duplicated. + + The chain is web -> api -> shared, so `shared` has one direct dependent and + one that is only reachable transitively. + """ + + def setUp(self) -> None: + super().setUp() + for name, pkg, deps in ( + ("web", "@acme/web", {"@acme/api": "^1.0.0"}), + ("api", "@acme/api", {"@acme/shared": "^1.0.0"}), + ("shared", "@acme/shared", {}), + ): + member = self.root / name + member.mkdir() + (member / "package.json").write_text(json.dumps( + {"name": pkg, "version": "1.0.0", "dependencies": deps})) + self.write_graph(_member_graph(name), root=member) + + self.manifest = self.root / "workspace.json" + self.manifest.write_text(json.dumps({"repos": ["web", "api", "shared"]})) + + def cli_ws(self, *args: str): + return self.cli_json(*args, "--workspace", str(self.manifest)) + + def test_each_repo_keeps_its_own_graph(self) -> None: + stats = self.cli_ws("stats") + self.assertEqual(sorted(stats["isolatedGraphs"]), ["api", "shared", "web"]) + for name in ("web", "api", "shared"): + self.assertEqual(stats["repos"][name]["nodes"], 2) + self.assertEqual(stats["repos"][name]["edges"], 1) + + def test_ids_are_never_namespaced(self) -> None: + """A repo's ids must read the same inside a workspace as outside one.""" + per_repo = self.cli_ws("nodes-for-file", "--q", "src/api.ts") + self.assertEqual([n["id"] for n in per_repo["api"]], + ["file:src/api.ts", "function:src/api.ts:go"]) + + def test_repos_are_isolated_from_each_other(self) -> None: + """Asking api for a file that lives in shared must return nothing.""" + per_repo = self.cli_ws("nodes-for-file", "--q", "src/shared.ts") + self.assertEqual(per_repo["api"], []) + self.assertTrue(per_repo["shared"]) + + def test_index_finds_direct_and_transitive_dependents(self) -> None: + self.assertEqual(self.cli_ws("affected-repos", "--repo", "shared"), + ["api", "web"]) + self.assertEqual(self.cli_ws("affected-repos", "--repo", "api"), ["web"]) + self.assertEqual(self.cli_ws("affected-repos", "--repo", "web"), []) + + def test_index_respects_the_hop_limit(self) -> None: + """One hop from shared reaches api but not web.""" + self.assertEqual( + self.cli_ws("affected-repos", "--repo", "shared", "--hops", "1"), + ["api"]) + + def test_cross_repo_blast_radius_reports_both_scopes(self) -> None: + out = self.cli_ws("blast-radius", "--name", "shared.ts") + self.assertEqual(out["definedIn"], ["shared"]) + self.assertEqual(out["downstreamRepos"], ["api", "web"]) + self.assertIn("shared", out["sameRepo"]) + + def test_index_is_reused_when_manifests_are_unchanged(self) -> None: + """The index is derived from package.json, so it should not be rebuilt + on every command — a rebuild also briefly empties it for other readers.""" + first = self.cli_ws("stats") + second = self.cli_ws("stats") + self.assertEqual(first["indexEdges"], second["indexEdges"]) + self.assertEqual(self.cli_ws("affected-repos", "--repo", "shared"), + ["api", "web"]) + + def test_index_is_rebuilt_when_a_dependency_changes(self) -> None: + self.cli_ws("stats") + pkg = self.root / "web" / "package.json" + data = json.loads(pkg.read_text()) + data["dependencies"] = {} # web no longer depends on api + pkg.write_text(json.dumps(data)) + + self.assertEqual(self.cli_ws("affected-repos", "--repo", "shared"), ["api"]) + + def test_workspace_second_run_is_cached(self) -> None: + first = self.cli_ws("stats") + second = self.cli_ws("stats") + for name in ("web", "api", "shared"): + self.assertEqual(first["repos"][name]["sync"]["mode"], "full") + self.assertEqual(second["repos"][name]["sync"]["mode"], "cached") + + +# ── Semantic search ─────────────────────────────────────────────────────── + +def _embedder_reachable() -> bool: + url = os.environ.get("UA_EMBED_URL") + if not url: + return False + import urllib.request + try: + req = urllib.request.Request( + url, data=json.dumps({"inputs": ["probe"], "truncate": True}).encode(), + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req, timeout=10) + return True + except Exception: + return False + + +@unittest.skipUnless(_embedder_reachable(), "UA_EMBED_URL not set or unreachable") +class SemanticTests(GraphQueryTestCase): + """Only runs when an embedding endpoint is configured.""" + + def setUp(self) -> None: + super().setUp() + self.write_graph(_chain_graph()) + self.embed_env = {"UA_EMBED_URL": os.environ["UA_EMBED_URL"]} + + def test_vectors_are_built_during_sync(self) -> None: + stats = self.cli_json("stats", env=self.embed_env) + self.assertTrue(stats["semantic"]) + self.assertGreater(stats["embedDimension"], 0) + + def test_semantic_search_returns_ranked_nodes(self) -> None: + self.cli_json("stats", env=self.embed_env) + hits = self.cli_json("semantic", "--q", "run a function", + "--k", "3", env=self.embed_env) + self.assertTrue(hits) + self.assertIn("id", hits[0]) + + def test_a_stale_vector_index_is_rebuilt(self) -> None: + """Switching embedding models must not leave a wrong-dimension index. + + Writing 384-dimension vectors into a 4-dimension index used to succeed + silently and fail only at query time, with an error naming neither the + cause nor the fix. + """ + self.cli_json("stats", env=self.embed_env) + + # Plant the wrong index from processes with no embedder configured, so + # they do not repair it on the way in. + self.cli_json("cypher", "--q", "DROP VECTOR INDEX FOR (n:Node) ON (n.emb)") + self.cli_json("cypher", "--q", + "CREATE VECTOR INDEX FOR (n:Node) ON (n.emb) " + "OPTIONS {dimension: 4, similarityFunction: 'cosine'}") + + rebuilt = self.cli_json("stats", env=self.embed_env) + self.assertEqual(rebuilt["sync"]["vectorIndexRebuilt"]["from"], 4) + self.assertEqual(rebuilt["sync"]["vectorIndexRebuilt"]["to"], + rebuilt["embedDimension"]) + # and the query works rather than raising a dimension mismatch + self.assertTrue(self.cli_json("semantic", "--q", "types", env=self.embed_env)) + + def test_an_embedder_configured_later_backfills(self) -> None: + """Enabling the embedder after a plain sync must still index the graph. + + The vector index used to be created only on a first sync, so this path + left existing nodes without vectors and raised a raw driver error. + """ + first = self.cli_json("stats") # no embedder + self.assertFalse(first["semantic"]) + + second = self.cli_json("stats", env=self.embed_env) # embedder appears + self.assertTrue(second["semantic"]) + self.assertEqual(second["sync"]["mode"], "cached") + self.assertEqual(second["sync"]["vectorsBackfilled"], first["nodes"]) + + # and semantic search now works rather than erroring + self.assertTrue(self.cli_json("semantic", "--q", "types", env=self.embed_env)) + + +# ── Graceful failure ────────────────────────────────────────────────────── + +class FailureModeTests(GraphQueryTestCase): + def test_missing_graph_explains_itself(self) -> None: + proc = self.run_cli("stats", expect_success=False) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("/understand", proc.stdout + proc.stderr) + + def test_semantic_search_requires_an_endpoint(self) -> None: + self.write_graph(_chain_graph()) + proc = self.run_cli("semantic", "--q", "anything", expect_success=False) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("UA_EMBED_URL", proc.stdout + proc.stderr) + + def test_an_unreachable_embedder_does_not_break_plain_queries(self) -> None: + """A misconfigured endpoint must warn, not take the whole command down.""" + self.write_graph(_chain_graph()) + proc = self.run_cli("blast-radius", "--path", "src/types.ts", + env={"UA_EMBED_URL": "http://127.0.0.1:9/embed"}) + self.assertIn("warning", proc.stderr.lower()) + self.assertEqual(json.loads(proc.stdout), + ["file:src/a.ts", "file:src/b.ts", "file:src/c.ts", + "file:src/d.ts"]) + + def test_an_unsafe_node_type_is_refused(self) -> None: + """Types become labels, which cannot be parameterised. + + knowledge-graph.json is committed and shared, so a type that is not a + plain identifier is rejected rather than interpolated into Cypher. + """ + graph = _chain_graph() + graph["nodes"].append({ + "id": "evil", "type": "file) MATCH (n) DETACH DELETE n //", + "name": "evil", "filePath": "evil.ts", "summary": "", + "tags": [], "complexity": "simple", + }) + self.write_graph(graph) + + proc = self.run_cli("stats", expect_success=False) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("unsafe node type", proc.stdout + proc.stderr) + + def test_an_unsafe_edge_type_is_refused(self) -> None: + graph = _chain_graph() + graph["edges"].append( + _edge("file:src/a.ts", "file:src/b.ts", "imports] () //")) + self.write_graph(graph) + + proc = self.run_cli("stats", expect_success=False) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("unsafe edge type", proc.stdout + proc.stderr) + + def test_unknown_batch_op_is_rejected(self) -> None: + self.write_graph(_chain_graph()) + proc = self.run_cli("batch", "--q", json.dumps([{"op": "nope"}]), + expect_success=False) + self.assertNotEqual(proc.returncode, 0) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/skill/diff/test_graph_query_helpers.py b/tests/skill/diff/test_graph_query_helpers.py new file mode 100644 index 000000000..f38972d60 --- /dev/null +++ b/tests/skill/diff/test_graph_query_helpers.py @@ -0,0 +1,266 @@ +#!/usr/bin/env python3 +""" +test_graph_query_helpers.py — Unit tests for graph-query.py's pure helpers. + +Run from the repo root: + python -m unittest tests.skill.diff.test_graph_query_helpers -v + +These need no database and no optional dependency, so unlike the integration +tests in test_graph_query.py they always execute. The module is importable +without a backend because every driver import sits inside the function that +uses it. +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from typing import Any + +# ── Module loader ───────────────────────────────────────────────────────── +# `graph-query.py` has a hyphen in its name, so we cannot `import` it directly. + +_HERE = Path(__file__).resolve().parent +_REPO_ROOT = _HERE.parent.parent.parent +_MODULE_PATH = ( + _REPO_ROOT + / "understand-anything-plugin" + / "skills" + / "understand-diff" + / "graph-query.py" +) + + +def _load_module() -> Any: + spec = importlib.util.spec_from_file_location("graph_query", _MODULE_PATH) + if spec is None or spec.loader is None: + raise RuntimeError(f"Could not load module from {_MODULE_PATH}") + module = importlib.util.module_from_spec(spec) + sys.modules["graph_query"] = module + spec.loader.exec_module(module) + return module + + +gq = _load_module() + + +class SafeTypeTests(unittest.TestCase): + """Types become Cypher labels and relationship types, which cannot be + parameterised, so they are validated rather than escaped.""" + + def test_plain_identifiers_pass_through(self) -> None: + for value in ("file", "function", "defines_schema", "Component", "_x9"): + self.assertEqual(gq.safe_type(value, "node"), value) + + def test_injection_attempts_are_refused(self) -> None: + hostile = [ + "file) MATCH (n) DETACH DELETE n //", + "imports] () //", + "a b", + "a-b", + "a.b", + "a`b", + "", + "1leading_digit", + ] + for value in hostile: + with self.subTest(value=value): + with self.assertRaises(SystemExit): + gq.safe_type(value, "node") + + def test_non_strings_are_refused(self) -> None: + for value in (None, 7, ["file"]): + with self.subTest(value=value): + with self.assertRaises(SystemExit): + gq.safe_type(value, "node") + + def test_the_message_names_the_kind(self) -> None: + with self.assertRaises(SystemExit) as caught: + gq.safe_type("a b", "edge") + self.assertIn("edge", str(caught.exception)) + + +class ReexecGuardTests(unittest.TestCase): + """The handover must not fire when it would be wrong or looping. + + Only the early-return guards are exercised here: the probing branch ends in + os.execve, which would replace the test runner itself. + """ + + def setUp(self) -> None: + self._saved = {k: os.environ.get(k) for k in + (gq.REEXEC_FLAG, "UA_FALKORDB_URL", "UA_PYTHON")} + self.addCleanup(self._restore) + for key in self._saved: + os.environ.pop(key, None) + + def _restore(self) -> None: + for key, value in self._saved.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + def test_does_nothing_once_already_handed_over(self) -> None: + os.environ[gq.REEXEC_FLAG] = "1" + self.assertIsNone(gq.reexec_under_capable_interpreter()) + + def test_does_nothing_when_a_server_is_configured(self) -> None: + """A server needs no local driver, so the interpreter is irrelevant.""" + os.environ["UA_FALKORDB_URL"] = "redis://localhost:6379" + self.assertIsNone(gq.reexec_under_capable_interpreter()) + + def test_candidates_are_newest_first(self) -> None: + self.assertEqual(list(gq.INTERPRETER_CANDIDATES), + sorted(gq.INTERPRETER_CANDIDATES, reverse=True)) + for name in gq.INTERPRETER_CANDIDATES: + major, minor = name.removeprefix("python").split(".") + self.assertGreaterEqual((int(major), int(minor)), (3, 12)) + + +class LabelTests(unittest.TestCase): + def test_snake_case_becomes_pascal_case(self) -> None: + self.assertEqual(gq.label_for("file"), "File") + self.assertEqual(gq.label_for("component_set"), "ComponentSet") + self.assertEqual(gq.label_for("defines_schema"), "DefinesSchema") + + def test_relationship_types_are_upper_case(self) -> None: + self.assertEqual(gq.rel_for("imports"), "IMPORTS") + self.assertEqual(gq.rel_for("depends_on"), "DEPENDS_ON") + + def test_labels_are_validated_too(self) -> None: + with self.assertRaises(SystemExit): + gq.label_for("file) //") + + +class DigestTests(unittest.TestCase): + """Digests decide what gets resynced, so stability is the whole point.""" + + def test_same_payload_gives_same_digest(self) -> None: + self.assertEqual(gq.digest({"a": 1, "b": [2, 3]}), + gq.digest({"a": 1, "b": [2, 3]})) + + def test_key_order_does_not_matter(self) -> None: + self.assertEqual(gq.digest({"a": 1, "b": 2}), gq.digest({"b": 2, "a": 1})) + + def test_list_order_does_matter(self) -> None: + """Callers sort before digesting; the digest itself must stay faithful.""" + self.assertNotEqual(gq.digest([1, 2]), gq.digest([2, 1])) + + def test_a_changed_value_changes_the_digest(self) -> None: + self.assertNotEqual(gq.digest({"summary": "x"}), gq.digest({"summary": "y"})) + + def test_digests_are_short_and_hex(self) -> None: + value = gq.digest({"a": 1}) + self.assertEqual(len(value), 16) + int(value, 16) # raises if not hex + + +class LayersContainingTests(unittest.TestCase): + LAYERS = [ + {"id": "core", "name": "Core", "description": "shared", + "nodeIds": ["file:a.ts", "file:b.ts"]}, + {"id": "ui", "name": "UI", "description": "views", + "nodeIds": ["file:c.tsx"]}, + ] + + def test_only_layers_with_a_match_are_returned(self) -> None: + out = gq.layers_containing(self.LAYERS, ["file:c.tsx"]) + self.assertEqual([l["name"] for l in out], ["UI"]) + + def test_matches_are_named_and_sorted(self) -> None: + out = gq.layers_containing(self.LAYERS, ["file:b.ts", "file:a.ts"]) + self.assertEqual(out[0]["matched"], ["file:a.ts", "file:b.ts"]) + + def test_no_overlap_returns_nothing(self) -> None: + self.assertEqual(gq.layers_containing(self.LAYERS, ["file:zz.ts"]), []) + + def test_missing_or_empty_layers_are_tolerated(self) -> None: + self.assertEqual(gq.layers_containing([], ["file:a.ts"]), []) + self.assertEqual(gq.layers_containing(None, ["file:a.ts"]), []) + self.assertEqual(gq.layers_containing([{"id": "x"}], ["file:a.ts"]), []) + + +class FindGraphJsonTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) + self.addCleanup(self._tmp.cleanup) + + def _write(self, directory: str) -> Path: + target = self.root / directory + target.mkdir(parents=True) + path = target / "knowledge-graph.json" + path.write_text("{}") + return path + + def test_finds_the_current_directory_name(self) -> None: + expected = self._write(".ua") + self.assertEqual(gq.find_graph_json(self.root), expected) + + def test_legacy_directory_wins_when_both_exist(self) -> None: + """Projects analysed before the rename keep reading their old directory.""" + self._write(".ua") + legacy = self._write(".understand-anything") + self.assertEqual(gq.find_graph_json(self.root), legacy) + + def test_absence_points_at_understand(self) -> None: + with self.assertRaises(SystemExit) as caught: + gq.find_graph_json(self.root) + self.assertIn("/understand", str(caught.exception)) + + +class EmbedTextTests(unittest.TestCase): + def test_name_summary_and_tags_are_combined(self) -> None: + text = gq.Embedder.text_for( + {"name": "isStale", "summary": "checks freshness", "tags": ["git", "cache"]}) + for part in ("isStale", "checks freshness", "git", "cache"): + self.assertIn(part, text) + + def test_long_summaries_are_truncated(self) -> None: + text = gq.Embedder.text_for({"name": "n", "summary": "x" * 500, "tags": []}) + self.assertLess(len(text), 300) + + def test_missing_fields_do_not_raise(self) -> None: + self.assertEqual(gq.Embedder.text_for({}), "") + self.assertEqual(gq.Embedder.text_for({"name": "n", "summary": None}), "n") + + +class RowShapeTests(unittest.TestCase): + """Every node row must carry the same keys, or the batched UNWIND writes + would set different properties per row.""" + + def test_line_range_is_split_into_two_columns(self) -> None: + row = gq.RepoGraph._row_for( + {"id": "x", "type": "file", "lineRange": [3, 9]}, "src/x.ts") + self.assertEqual((row["lineStart"], row["lineEnd"]), (3, 9)) + + def test_a_missing_line_range_becomes_sentinels(self) -> None: + row = gq.RepoGraph._row_for({"id": "x", "type": "file"}, "src/x.ts") + self.assertEqual((row["lineStart"], row["lineEnd"]), (-1, -1)) + + def test_a_malformed_line_range_becomes_sentinels(self) -> None: + row = gq.RepoGraph._row_for( + {"id": "x", "type": "file", "lineRange": [5]}, "src/x.ts") + self.assertEqual((row["lineStart"], row["lineEnd"]), (-1, -1)) + + def test_rows_always_have_the_same_keys(self) -> None: + sparse = gq.RepoGraph._row_for({"id": "a", "type": "file"}, "k") + full = gq.RepoGraph._row_for( + {"id": "b", "type": "function", "name": "f", "filePath": "p", + "summary": "s", "tags": ["t"], "complexity": "complex", + "lineRange": [1, 2]}, "k") + self.assertEqual(sorted(sparse), sorted(full)) + + def test_the_grouping_key_is_carried(self) -> None: + row = gq.RepoGraph._row_for({"id": "x", "type": "file"}, "src/x.ts") + self.assertEqual(row["__key"], "src/x.ts") + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/understand-anything-plugin/skills/understand-diff/SKILL.md b/understand-anything-plugin/skills/understand-diff/SKILL.md index 13880f31b..bf5662de2 100644 --- a/understand-anything-plugin/skills/understand-diff/SKILL.md +++ b/understand-anything-plugin/skills/understand-diff/SKILL.md @@ -52,6 +52,52 @@ The knowledge graph JSON has this structure: - If the committed diff or any working-tree command reports project files, warn before impact analysis that the graph may omit those changes. Suggest: Run `/understand` to refresh the graph. - Run the commit diff only when `GRAPH_COMMIT_RAW` resolves successfully. If the graph commit or Git metadata is missing, invalid, or unavailable, give a brief best-effort warning and continue instead of blocking. + **Optional fast path for steps 4–6.** Those steps walk the graph by grepping + `knowledge-graph.json` once per node id, which pulls a lot of JSON into context on + large graphs. If the optional query backend is configured, ask for the same thing in + one call instead: + + ```bash + "${UA_PYTHON:-python}" "/graph-query.py" batch --q '[ + {"op": "nodes-for-file", "path": ""}, + {"op": "blast-radius", "path": "", "hops": 3}, + {"op": "layers-for", "ids": [""]} + ]' + ``` + + The embedded backend needs Python 3.12 or newer. If the `python` you invoke is older, + the script looks for a newer one on `PATH` and hands over to it by itself, so this + usually needs no setup. `UA_PYTHON` overrides that choice when the right interpreter + is somewhere unusual, and a FalkorDB server via `UA_FALKORDB_URL` works on any version. + + `nodes-for-file` returns the file node plus every function and class defined in it, + which is what step 4 assembles by grepping. `blast-radius` returns the affected node + ids from step 5, and `layers-for` the architectural layers from step 6 — send it the + ids the first two calls returned, which means a second `batch` call once you have them. Put every changed file in one `batch` call rather than calling once + per file — process startup dominates, the queries themselves are milliseconds. + + Pass `blast-radius` a **path**, not a basename. Basenames are not unique — this + repository has eleven files called `index.ts` and five called `types.ts` — and a + name seeds from all of them at once, overstating the impact without saying so. A + changed path is what git reports anyway. + + To check availability, run `"${UA_PYTHON:-python}" "/graph-query.py" stats`. If it exits + non-zero the backend is not configured — that is the normal default, so just continue + with steps 4–6 as written. The backend is read-only, reads the same + `knowledge-graph.json`, and writes nothing; the JSON remains the source of truth. + + Re-syncing after a `/understand` refresh is incremental: only files whose nodes + actually changed are replaced, so a one-file edit costs one file of work. + + Two further options, both off unless configured — see `graph-query.py --help`: + `semantic` and `semantic-traverse` rank nodes by meaning rather than substring when + `UA_EMBED_URL` points at an embedding endpoint, which helps when a diff touches code + the user describes in their own words. `--workspace ` keeps one graph + per repo and adds a small index of the dependencies between them, so a changed file + in one repo can report which other repos are downstream — `blast-radius` then returns + `sameRepo` detail plus a `downstreamRepos` list, and `affected-repos` answers the + repo-level question on its own. + 4. **Find nodes for changed files** — for each changed file path, use Grep to search the knowledge graph for: - Nodes with matching `"filePath"` values (e.g., `grep "changed/file/path"`) - This finds file-level nodes (including non-code types) AND function/class nodes defined in those files diff --git a/understand-anything-plugin/skills/understand-diff/graph-query.py b/understand-anything-plugin/skills/understand-diff/graph-query.py new file mode 100644 index 000000000..882bff988 --- /dev/null +++ b/understand-anything-plugin/skills/understand-diff/graph-query.py @@ -0,0 +1,926 @@ +#!/usr/bin/env python3 +"""FalkorDB query adapter for Understand-Anything knowledge graphs. + +Read-only consumer of `.ua/knowledge-graph.json`. The JSON stays the source of +truth; this mirrors it into a graph so the skills can ask multi-hop questions +with a query instead of a chain of greps. Nothing is ever written back. + +Backends, picked automatically: + embedded FalkorDBLite, in-process, no server, no config (needs Python >= 3.12) + server any FalkorDB instance, via UA_FALKORDB_URL (e.g. redis://localhost:6379) + +Sync is incremental. A digest is stored per source file, so re-syncing after an +edit replaces only the nodes of files that actually changed. + +Semantic search is optional. Set UA_EMBED_URL to a text-embedding endpoint that +accepts {"inputs": [...]} and returns a list of vectors (e.g. a local +text-embeddings-inference server) and node vectors are built during sync. +Without it, everything except the `semantic*` commands works unchanged. + +Workspaces keep one graph per repo rather than merging them. Every repo stays +isolated, with ids exactly as they appear in its own JSON, and a small index +graph holds just the repos and the dependencies between them. A cross-repo +question is answered in two cheap stages: traverse the index to find which repos +are affected, then query only those repos' graphs. Point --workspace at a +manifest: {"repos": ["../api", "../web"]}. + +CLI + python graph-query.py stats + python graph-query.py search --q auth + python graph-query.py nodes-for-file --q src/types.ts + python graph-query.py neighbors --id "file:src/a.ts" + python graph-query.py layers-for --ids "file:src/a.ts,file:src/b.ts" + python graph-query.py blast-radius --path src/types.ts --hops 3 + python graph-query.py calls-from --name registerAllParsers + python graph-query.py calls-to --name validateGraph + python graph-query.py path --from "file:a.ts" --to "file:b.ts" + python graph-query.py semantic --q "how are imports resolved" + python graph-query.py semantic-traverse --q "graph persistence" --hops 2 + python graph-query.py cypher --q "MATCH (n:File) RETURN count(n)" + python graph-query.py batch --q '[{"op":"blast-radius","path":"src/a.ts"}]' + + # workspace-only + python graph-query.py affected-repos --repo shared --workspace ws.json + python graph-query.py blast-radius --path src/auth.ts --workspace ws.json + +Every command prints JSON on stdout. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +import urllib.request +from collections import defaultdict +from pathlib import Path + +# Legacy first, matching resolveUaDirName in packages/core/src/persistence/index.ts: +# a project analysed before the `.ua` rename keeps reading its old directory, so +# checking `.ua` first would make this adapter disagree with the rest of UA about +# which graph is authoritative. +UA_DIRS = (".understand-anything", ".ua") +GRAPH_FILE = "knowledge-graph.json" +FILE_STAMP = "__ua_file__" +INDEX_GRAPH = "__ua_workspace__" + +# UA edge types that mean "A depends on B", used by the blast-radius query. +DEPENDENCY_EDGES = ("IMPORTS", "DEPENDS_ON") + +EMBED_BATCH = 8 # the local TEI server rejects much larger payloads +EMBED_TEXT_CHARS = 150 # summaries are long; the head of one is enough to rank on + + +REEXEC_FLAG = "UA_GRAPH_QUERY_REEXEC" +INTERPRETER_CANDIDATES = ("python3.14", "python3.13", "python3.12") + + +def reexec_under_capable_interpreter() -> None: + """Re-run under an interpreter that can actually load the backend. + + The embedded backend needs Python 3.12 or newer, but a project's default + `python` is frequently older, and a newer one is usually installed alongside + it. Rather than making that the user's problem, look for an interpreter that + can import the driver and hand over to it. Silent when nothing needs doing. + """ + if os.environ.get(REEXEC_FLAG): + return # already handed over once + if os.environ.get("UA_FALKORDB_URL"): + return # a server needs no local driver + try: + import redislite.falkordb_client # noqa: F401 + return # this interpreter is fine + except ImportError: + pass + + probe = "import redislite.falkordb_client" + candidates = [os.environ["UA_PYTHON"]] if os.environ.get("UA_PYTHON") else [] + candidates += [c for c in (shutil.which(name) for name in INTERPRETER_CANDIDATES) if c] + + for candidate in candidates: + try: + ok = subprocess.run([candidate, "-c", probe], capture_output=True, + timeout=30).returncode == 0 + except (OSError, subprocess.SubprocessError): + continue + if ok: + print(f"note: handing over to {candidate}, which can load the backend", + file=sys.stderr) + os.execve(candidate, [candidate, os.path.abspath(__file__), *sys.argv[1:]], + {**os.environ, REEXEC_FLAG: "1"}) + + # Nothing suitable found; Backend will explain what to do. + + +def find_graph_json(project_root: Path) -> Path: + """Locate the graph, honouring the legacy .understand-anything/ directory.""" + for d in UA_DIRS: + candidate = project_root / d / GRAPH_FILE + if candidate.exists(): + return candidate + raise SystemExit( + f"No {GRAPH_FILE} under {'/ or '.join(UA_DIRS)}/ in {project_root}. " + "Run /understand first." + ) + + +_SAFE_TYPE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def safe_type(raw: str, kind: str) -> str: + """Validate a type before it is interpolated into Cypher. + + Node and edge types become labels and relationship types, which cannot be + parameterised. `knowledge-graph.json` is committed to repositories and shared + between teammates, so it is not automatically trusted input. Anything that is + not a plain identifier is rejected rather than escaped. + """ + if not isinstance(raw, str) or not _SAFE_TYPE.match(raw): + raise SystemExit(f"refusing to build graph: unsafe {kind} type {raw!r}") + return raw + + +def label_for(node_type: str) -> str: + return "".join(part.capitalize() for part in safe_type(node_type, "node").split("_")) + + +def rel_for(edge_type: str) -> str: + return safe_type(edge_type, "edge").upper() + + +def digest(payload) -> str: + return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()[:16] + + +def layers_containing(layers: list[dict], ids: list[str]) -> list[dict]: + """Layers holding any of these node ids, with the matches named. + + A flat intersection rather than a traversal, so it stays out of the graph and + out of the incremental sync's invariants. + """ + wanted = set(ids) + out = [] + for layer in layers or []: + hits = sorted(wanted.intersection(layer.get("nodeIds", []))) + if hits: + out.append({"id": layer.get("id"), "name": layer.get("name"), + "description": layer.get("description"), "matched": hits}) + return out + + +# --------------------------------------------------------------------------- # +# embeddings (optional) +# --------------------------------------------------------------------------- # + +class Embedder: + """Thin client for an {"inputs": [...]} -> [[float]] embedding endpoint.""" + + def __init__(self, url: str): + self.url = url + self.dimension = len(self.encode(["dimension probe"])[0]) + + def encode(self, texts: list[str]) -> list[list[float]]: + out: list[list[float]] = [] + for i in range(0, len(texts), EMBED_BATCH): + req = urllib.request.Request( + self.url, + data=json.dumps( + {"inputs": texts[i:i + EMBED_BATCH], "truncate": True} + ).encode(), + headers={"Content-Type": "application/json"}, + ) + out.extend(json.loads(urllib.request.urlopen(req, timeout=120).read())) + return out + + @staticmethod + def text_for(node: dict) -> str: + return " ".join([ + node.get("name", ""), + (node.get("summary") or "")[:EMBED_TEXT_CHARS], + " ".join(node.get("tags") or []), + ]).strip() + + +def make_embedder() -> Embedder | None: + url = os.environ.get("UA_EMBED_URL") + if not url: + return None + try: + return Embedder(url) + except Exception as exc: # an unreachable endpoint must not break plain queries + print(f"warning: UA_EMBED_URL unreachable ({exc}); semantic commands disabled", + file=sys.stderr) + return None + + +# --------------------------------------------------------------------------- # +# backend +# --------------------------------------------------------------------------- # + +class Backend: + """One FalkorDB connection handing out graph handles by key. + + A single instance holds many graphs, which is what keeps repos isolated + without paying for a separate server per repo. + """ + + def __init__(self, anchor_dir: Path): + url = os.environ.get("UA_FALKORDB_URL") + if url: + from urllib.parse import urlparse + from falkordb import FalkorDB + + parsed = urlparse(url) + self._db = FalkorDB(host=parsed.hostname or "localhost", + port=parsed.port or 6379) + self.kind = "server" + return + + try: + from redislite.falkordb_client import FalkorDB as EmbeddedFalkorDB + except ImportError: + running = ".".join(str(v) for v in sys.version_info[:3]) + hint = ( + f"this interpreter is Python {running}, and the embedded backend " + "needs 3.12 or newer -- point UA_PYTHON at a newer interpreter, or " + "set UA_FALKORDB_URL to use a server on any version" + if sys.version_info < (3, 12) else + "install it with `pip install falkordblite`, or set UA_FALKORDB_URL " + "to point at a FalkorDB server" + ) + raise SystemExit(f"No backend available: {hint}.") + + anchor_dir.mkdir(parents=True, exist_ok=True) + self._db = EmbeddedFalkorDB(str(anchor_dir / "falkordb.db")) + self.kind = "embedded" + + def graph(self, key: str): + return self._db.select_graph(key) + + +# --------------------------------------------------------------------------- # +# one repo, one graph +# --------------------------------------------------------------------------- # + +class RepoGraph: + """A single UA knowledge graph. Node ids are exactly the JSON's own ids.""" + + def __init__(self, backend: Backend, key: str, graph_json: Path, + embedder: Embedder | None): + self.key = key + self.graph_json = graph_json + self.embedder = embedder + self.backend_kind = backend.kind + self.graph = backend.graph(key) + self.raw = json.loads(graph_json.read_text()) + self.synced = self._sync() + + # ---- incremental sync -------------------------------------------------- + + def _stored_digests(self) -> dict[str, str]: + try: + rows = self.graph.query( + f"MATCH (s:{FILE_STAMP}) RETURN s.key, s.digest" + ).result_set + return {k: d for k, d in rows} + except Exception: + return {} # graph does not exist yet + + def _vector_index_dimension(self) -> int | None: + """Dimension of the existing vector index on :Node(emb), if there is one.""" + try: + rows = self.graph.query("CALL db.indexes()").result_set + except Exception: + return None + for label, _props, types, options, *_rest in rows: + if label != "Node" or not isinstance(types, dict): + continue + if "VECTOR" in (types.get("emb") or []): + return (options or {}).get("emb", {}).get("dimension") + return None + + def _ensure_indexes(self) -> dict: + """Bring indexes in line with the current configuration. + + Checked on every sync rather than only the first, because the + configuration can change under a graph that already exists. Two cases + matter: an embedder added after a plain sync needs an index created, and + an embedder swapped for a different model needs the index *rebuilt* -- + writing 384-dimension vectors into a 4-dimension index succeeds silently + and only fails later, at query time, with an error that points nowhere. + """ + try: + self.graph.query("CREATE INDEX FOR (n:Node) ON (n.id)") + except Exception: + pass # already indexed + + if not self.embedder: + return {} + + existing = self._vector_index_dimension() + wanted = self.embedder.dimension + + if existing == wanted: + return {} + + if existing is not None: + # Stale vectors are unusable at the new dimension, so drop them and + # let the backfill re-embed from scratch. + self.graph.query("DROP VECTOR INDEX FOR (n:Node) ON (n.emb)") + self.graph.query("MATCH (n:Node) WHERE n.emb IS NOT NULL SET n.emb = NULL") + + self.graph.query( + "CREATE VECTOR INDEX FOR (n:Node) ON (n.emb) " + "OPTIONS {dimension: $d, similarityFunction: 'cosine'}", {"d": wanted}) + + if existing is None: + return {} + return {"vectorIndexRebuilt": {"from": existing, "to": wanted}} + + def _prune_orphan_stamps(self) -> int: + """Drop stamps whose files have no nodes left. + + A stamp without nodes would make the next sync believe that file is + already present and skip it. + """ + live = {r[0] for r in self.rows("MATCH (n:Node) RETURN DISTINCT n.__key")} + stamped = {r[0] for r in self.rows(f"MATCH (s:{FILE_STAMP}) RETURN s.key")} + orphans = sorted(stamped - live) + if orphans: + self.graph.query( + f"UNWIND $keys AS k MATCH (s:{FILE_STAMP} {{key: k}}) DELETE s", + {"keys": orphans}) + return len(orphans) + + def _backfill_vectors(self, nodes_by_id: dict[str, dict]) -> int: + """Embed any node that has no vector yet. + + Covers the case where the embedder is configured after a plain sync: the + nodes are already there and unchanged, so nothing else would revisit them. + """ + if not self.embedder: + return 0 + missing = [r[0] for r in self.rows( + "MATCH (n:Node) WHERE n.emb IS NULL RETURN n.id")] + pending = [i for i in missing if i in nodes_by_id] + if not pending: + return 0 + vectors = self.embedder.encode( + [Embedder.text_for(nodes_by_id[i]) for i in pending]) + self.graph.query( + "UNWIND $rows AS r MATCH (n:Node {id: r.id}) SET n.emb = vecf32(r.emb)", + {"rows": [{"id": i, "emb": v} for i, v in zip(pending, vectors)]}) + return len(pending) + + def _write_nodes(self, rows: list[dict], vectors: dict[str, list[float]]) -> None: + """Insert nodes in one query per label, rather than one query per node.""" + grouped: dict[str, list[dict]] = defaultdict(list) + for n in rows: + grouped[label_for(n["type"])].append(n) + + props = ("id: r.id, type: r.type, name: r.name, filePath: r.filePath, " + "summary: r.summary, tags: r.tags, complexity: r.complexity, " + "lineStart: r.lineStart, lineEnd: r.lineEnd, __key: r.__key") + for label, group in grouped.items(): + # Vectors cannot be applied conditionally inside one UNWIND, so the + # group is split by whether a vector is present. + plain = [r for r in group if r["id"] not in vectors] + embedded = [dict(r, emb=vectors[r["id"]]) for r in group if r["id"] in vectors] + if plain: + self.graph.query( + f"UNWIND $rows AS r CREATE (x:Node:{label} {{{props}}})", + {"rows": plain}) + if embedded: + self.graph.query( + f"UNWIND $rows AS r " + f"CREATE (x:Node:{label} {{{props}, emb: vecf32(r.emb)}})", + {"rows": embedded}) + + def _write_edges(self, edges: list[dict]) -> None: + """Insert edges in one query per relationship type.""" + grouped: dict[str, list[dict]] = defaultdict(list) + for e in edges: + grouped[rel_for(e["type"])].append({ + "src": e["source"], "dst": e["target"], "type": e["type"], + "direction": e.get("direction", "forward"), + "weight": e.get("weight", 0.0), + }) + for rel, group in grouped.items(): + self.graph.query( + "UNWIND $rows AS r MATCH (a:Node {id: r.src}), (b:Node {id: r.dst}) " + f"CREATE (a)-[:{rel} {{type: r.type, direction: r.direction, " + "weight: r.weight}]->(b)", + {"rows": group}) + + @staticmethod + def _row_for(node: dict, key: str) -> dict: + line_range = node.get("lineRange") or [] + return { + "id": node["id"], "type": node["type"], "name": node.get("name", ""), + "filePath": node.get("filePath", ""), "summary": node.get("summary", ""), + "tags": node.get("tags", []), "complexity": node.get("complexity", ""), + "lineStart": line_range[0] if len(line_range) == 2 else -1, + "lineEnd": line_range[1] if len(line_range) == 2 else -1, + "__key": key, + } + + def _sync(self) -> dict: + nodes = self.raw.get("nodes", []) + edges = self.raw.get("edges", []) + + # The file a node belongs to is the unit of replacement. + by_key: dict[str, list[dict]] = defaultdict(list) + key_of: dict[str, str] = {} + for n in nodes: + key = n.get("filePath") or n["id"] + by_key[key].append(n) + key_of[n["id"]] = key + + # An edge is incident to the files at both of its ends, so adding an + # import marks both files changed. Without this, an edge-only change -- + # the most common thing a commit does -- would leave every digest intact + # and never be synced at all. + edges_by_key: dict[str, list[dict]] = defaultdict(list) + for e in edges: + fingerprint = (e["source"], e["target"], e["type"], + e.get("direction", "forward"), e.get("weight", 0.0)) + for endpoint in (e["source"], e["target"]): + key = key_of.get(endpoint) + if key is not None: + edges_by_key[key].append(fingerprint) + + current = { + k: digest({ + "nodes": [ + {kk: n.get(kk) for kk in + ("id", "type", "name", "summary", "tags", "complexity", + "filePath", "lineRange")} + for n in sorted(v, key=lambda x: x["id"]) + ], + "edges": sorted(edges_by_key.get(k, [])), + }) + for k, v in by_key.items() + } + + index_state = self._ensure_indexes() + + stored = self._stored_digests() + if stored: + self._prune_orphan_stamps() + stored = self._stored_digests() + changed = {k for k, d in current.items() if stored.get(k) != d} + removed = set(stored) - set(current) + first_run = not stored + + if not changed and not removed: + backfilled = self._backfill_vectors({n["id"]: n for n in nodes}) + return {"mode": "cached", "files": 0, "nodes": 0, **index_state, + **({"vectorsBackfilled": backfilled} if backfilled else {})} + + # Drop the nodes of changed/removed files. DETACH also drops edges + # arriving from untouched files, so those are re-created below. Stamps go + # last, after edges: a crash mid-sync must leave the affected files + # unstamped so the next run repairs them instead of reporting "cached". + for key in changed | removed: + self.graph.query( + "MATCH (n:Node) WHERE n.__key = $k DETACH DELETE n", {"k": key}) + self.graph.query( + f"MATCH (s:{FILE_STAMP} {{key: $k}}) DELETE s", {"k": key}) + + fresh = [n for k in changed for n in by_key[k]] + vectors: dict[str, list[float]] = {} + if self.embedder and fresh: + vectors = dict(zip( + (n["id"] for n in fresh), + self.embedder.encode([Embedder.text_for(n) for n in fresh]), + )) + + self._write_nodes([self._row_for(n, key_of[n["id"]]) for n in fresh], vectors) + + touched = {n["id"] for n in fresh} + self._write_edges([e for e in edges + if e["source"] in touched or e["target"] in touched]) + + self.graph.query( + f"UNWIND $rows AS r CREATE (:{FILE_STAMP} {{key: r.key, digest: r.digest}})", + {"rows": [{"key": k, "digest": current[k]} for k in changed]}) + + backfilled = self._backfill_vectors({n["id"]: n for n in nodes}) + return { + "mode": "full" if first_run else "incremental", + "files": len(changed) + len(removed), + "nodes": len(fresh), + **index_state, + **({"vectorsBackfilled": backfilled} if backfilled else {}), + } + + # ---- queries ----------------------------------------------------------- + + def rows(self, cypher: str, params: dict | None = None) -> list: + return self.graph.query(cypher, params or {}).result_set + + def search(self, term: str, limit: int = 25) -> list[dict]: + rows = self.rows( + "MATCH (n:Node) WHERE toLower(n.name) CONTAINS toLower($t) " + "OR toLower(n.summary) CONTAINS toLower($t) " + "OR toLower(n.filePath) CONTAINS toLower($t) " + "RETURN n.id, n.type, n.name, n.filePath, n.summary " + "ORDER BY n.id LIMIT $lim", + {"t": term, "lim": limit}) + return [dict(zip(("id", "type", "name", "filePath", "summary"), r)) for r in rows] + + def nodes_for_file(self, path: str) -> list[dict]: + """The file node plus every function and class defined in it. + + Paths are matched from the right in both directions, because the caller's + path and the graph's may be rooted differently. `git diff` reports paths + relative to the repository, while graph paths are relative to whatever + directory `/understand` was pointed at — a subdirectory in a scoped + monorepo run. So the caller's path may be longer or shorter than the + stored one. + """ + clean = path.lstrip("/") + rows = self.rows( + "MATCH (n:Node) WHERE n.filePath <> '' AND (" + " n.filePath = $p" + " OR n.filePath ENDS WITH $suffix" # caller gave a shorter path + " OR $p ENDS WITH ('/' + n.filePath)" # caller gave a longer path + ") RETURN n.id, n.type, n.name, n.filePath ORDER BY n.id", + {"p": clean, "suffix": "/" + clean}) + return [dict(zip(("id", "type", "name", "filePath"), r)) for r in rows] + + def neighbors(self, node_id: str) -> list[dict]: + rows = self.rows( + "MATCH (n:Node {id: $id})-[r]-(m:Node) " + "RETURN type(r), m.id, m.type, m.name ORDER BY m.id", {"id": node_id}) + return [dict(zip(("edge", "id", "type", "name"), r)) for r in rows] + + _PATH_MATCH = ("t.filePath <> '' AND (t.filePath = $p " + "OR t.filePath ENDS WITH $suffix OR $p ENDS WITH ('/' + t.filePath))") + + def blast_radius(self, name: str | None = None, hops: int = 3, + path: str | None = None) -> list[str]: + """Everything in this repo that transitively depends on the target. + + Prefer `path`. Basenames are not unique -- this repo has eleven files + named index.ts and five named types.ts -- so seeding by name unions the + impact of every file sharing that name and silently overstates it. + `understand-diff` starts from a changed path, so it has the precise + answer available and should pass it. + """ + rels = "|".join(DEPENDENCY_EDGES) + if path: + clean = path.lstrip("/") + rows = self.rows( + f"MATCH (t:Node)<-[:{rels}*1..{hops}]-(d:Node) WHERE {self._PATH_MATCH} " + "RETURN DISTINCT d.id ORDER BY d.id", + {"p": clean, "suffix": "/" + clean}) + else: + rows = self.rows( + f"MATCH (t:Node)<-[:{rels}*1..{hops}]-(d:Node) " + "WHERE t.name = $n RETURN DISTINCT d.id ORDER BY d.id", {"n": name}) + return [r[0] for r in rows] + + def layers_for(self, ids: list[str]) -> list[dict]: + """Which architectural layers contain any of these nodes (step 6).""" + return layers_containing(self.raw.get("layers", []), ids) + + def has_path(self, path: str) -> bool: + clean = path.lstrip("/") + return bool(self.rows( + f"MATCH (t:Node) WHERE {self._PATH_MATCH} RETURN 1 LIMIT 1", + {"p": clean, "suffix": "/" + clean})) + + def calls_from(self, name: str, hops: int = 3) -> list[str]: + rows = self.rows( + f"MATCH (s:Node)-[:CALLS*1..{hops}]->(x:Node) " + "WHERE s.name = $n RETURN DISTINCT x.id ORDER BY x.id", {"n": name}) + return [r[0] for r in rows] + + def calls_to(self, name: str, hops: int = 2) -> list[str]: + rows = self.rows( + f"MATCH (t:Node)<-[:CALLS*1..{hops}]-(c:Node) " + "WHERE t.name = $n RETURN DISTINCT c.id ORDER BY c.id", {"n": name}) + return [r[0] for r in rows] + + def path(self, src: str, dst: str, max_hops: int = 6) -> list[str]: + # FalkorDB wants shortestPath in WITH/RETURN and a directed pattern. + rows = self.rows( + "MATCH (a:Node {id: $a}), (b:Node {id: $b}) " + f"RETURN [n IN nodes(shortestPath((a)-[*..{max_hops}]->(b))) | n.id]", + {"a": src, "b": dst}) + return rows[0][0] if rows and rows[0][0] else [] + + def has_name(self, name: str) -> bool: + return bool(self.rows( + "MATCH (n:Node) WHERE n.name = $n RETURN 1 LIMIT 1", {"n": name})) + + def _require_embedder(self) -> Embedder: + if not self.embedder: + raise SystemExit( + "Semantic search needs an embedding endpoint. " + "Set UA_EMBED_URL (e.g. http://localhost:8080/embed).") + return self.embedder + + def semantic(self, query: str, k: int = 10) -> list[dict]: + vec = self._require_embedder().encode([query])[0] + rows = self.rows( + "CALL db.idx.vector.queryNodes('Node','emb',$k,vecf32($q)) YIELD node " + "RETURN node.id, node.type, node.name, node.filePath", {"k": k, "q": vec}) + return [dict(zip(("id", "type", "name", "filePath"), r)) for r in rows] + + def semantic_traverse(self, query: str, k: int = 5, hops: int = 2) -> list[str]: + """Find nodes that mean the query, then traverse out from them. + + The point of putting vectors in the graph rather than beside it: seeding + and traversal happen in one statement. + """ + vec = self._require_embedder().encode([query])[0] + rows = self.rows( + "CALL db.idx.vector.queryNodes('Node','emb',$k,vecf32($q)) YIELD node AS seed " + f"MATCH (seed)-[:IMPORTS|CALLS|CONTAINS*1..{hops}]->(reached:Node) " + "RETURN DISTINCT reached.id ORDER BY reached.id", {"k": k, "q": vec}) + return [r[0] for r in rows] + + def counts(self) -> dict: + return { + "nodes": self.rows("MATCH (n:Node) RETURN count(n)")[0][0], + "edges": self.rows("MATCH ()-[r]->() RETURN count(r)")[0][0], + } + + +# --------------------------------------------------------------------------- # +# many repos, one instance +# --------------------------------------------------------------------------- # + +class Workspace: + """Isolated per-repo graphs plus a tiny index of the dependencies between them. + + Nothing is merged. Each repo keeps its own graph and its own ids, so a + single-repo query is identical whether or not a workspace exists. Only the + repo-level topology is duplicated into the index graph, which is what makes + a cross-repo question answerable by traversal rather than by scanning every + repo. + """ + + def __init__(self, members: list[tuple[str, Path]], backend: Backend, + embedder: Embedder | None, index_key: str = INDEX_GRAPH): + self.backend = backend + self.repos = { + name: RepoGraph(backend, name, path, embedder) for name, path in members + } + self.index = backend.graph(index_key) + self.index_edges = self._build_index(members) + + def _build_index(self, members: list[tuple[str, Path]]) -> int: + """Build the index from package manifests, skipping it when unchanged. + + The index is tiny, so it is rebuilt whole rather than diffed -- but only + when the manifests it derives from have actually changed, so repeated + commands do not pay for it and never observe it half-built. + + UA's graph records only resolved intra-repo imports, so file-level + cross-repo edges would need the scan phase's import map. Package + manifests give the repo-level dependency reliably. + """ + owner: dict[str, str] = {} + declared: dict[str, dict] = {} + for name, path in members: + manifest = path.parent.parent / "package.json" + if not manifest.exists(): + declared[name] = {} + continue + try: + data = json.loads(manifest.read_text()) + except json.JSONDecodeError: + declared[name] = {} + continue + if data.get("name"): + owner[data["name"]] = name + declared[name] = { + **data.get("dependencies", {}), + **data.get("devDependencies", {}), + **data.get("peerDependencies", {}), + } + + # Rebuild only when the manifests differ from what the index was built from. + stamp = digest({"repos": sorted(self.repos), "owner": owner, + "declared": {k: sorted(v) for k, v in declared.items()}}) + try: + rows = self.index.query( + f"MATCH (s:{FILE_STAMP}) RETURN s.digest LIMIT 1").result_set + if rows and rows[0][0] == stamp: + edges = self.index.query( + "MATCH ()-[r:DEPENDS_ON]->() RETURN count(r)").result_set + return edges[0][0] if edges else 0 + except Exception: + pass # index graph does not exist yet + + try: + self.index.delete() + except Exception: + pass + + for name in self.repos: + self.index.query("CREATE (:Repo {name: $n})", {"n": name}) + + created = 0 + for name, deps in declared.items(): + for spec in deps: + target = owner.get(spec) + if target and target != name: + self.index.query( + "MATCH (a:Repo {name: $a}), (b:Repo {name: $b}) " + "MERGE (a)-[:DEPENDS_ON {via: $via}]->(b)", + {"a": name, "b": target, "via": spec}) + created += 1 + + self.index.query(f"CREATE (:{FILE_STAMP} {{digest: $d}})", {"d": stamp}) + return created + + # ---- cross-repo --------------------------------------------------------- + + def affected_repos(self, repo: str, hops: int = 5) -> list[str]: + """Repos that transitively depend on `repo` — one query on a tiny graph.""" + rows = self.index.query( + f"MATCH (t:Repo {{name: $n}})<-[:DEPENDS_ON*1..{hops}]-(d:Repo) " + "RETURN DISTINCT d.name ORDER BY d.name", {"n": repo}).result_set + return [r[0] for r in rows] + + def repos_defining(self, name: str) -> list[str]: + return [r for r, g in self.repos.items() if g.has_name(name)] + + def repos_containing_path(self, path: str) -> list[str]: + return [r for r, g in self.repos.items() if g.has_path(path)] + + def blast_radius(self, name: str | None = None, hops: int = 3, + path: str | None = None) -> dict: + """Two stages: local impact where the file lives, then dependent repos. + + Downstream repos are reported at repo granularity because there are no + file-level cross-repo edges to follow — see _build_index. + """ + origins = (self.repos_containing_path(path) if path + else self.repos_defining(name)) + local = {r: self.repos[r].blast_radius(name, hops, path) for r in origins} + downstream = sorted({ + d for r in origins for d in self.affected_repos(r) if d not in origins + }) + return { + "definedIn": origins, + "sameRepo": local, + "downstreamRepos": downstream, + "note": "downstream is repo-level; file-level cross-repo edges need " + "the scan phase's import map", + } + + def fan_out(self, method: str, *args) -> dict: + """Run a per-repo query against every member, keyed by repo.""" + return {r: getattr(g, method)(*args) for r, g in self.repos.items()} + + def stats(self) -> dict: + return { + "mode": "workspace", + "backend": self.backend.kind, + "repos": { + r: {**g.counts(), "sync": g.synced} for r, g in self.repos.items() + }, + "isolatedGraphs": list(self.repos), + "indexEdges": self.index_edges, + } + + +# --------------------------------------------------------------------------- # +# CLI +# --------------------------------------------------------------------------- # + +PER_REPO_OPS = { + "search": ("search", ("q", "limit")), + "nodes-for-file": ("nodes_for_file", ("path",)), + "neighbors": ("neighbors", ("id",)), + "layers-for": ("layers_for", ("ids",)), + "calls-from": ("calls_from", ("name", "hops")), + "calls-to": ("calls_to", ("name", "hops")), + "path": ("path", ("from", "to", "hops")), + "semantic": ("semantic", ("q", "k")), + "semantic-traverse": ("semantic_traverse", ("q", "k", "hops")), +} +DEFAULTS = {"hops": 3, "k": 10, "limit": 25} + + +def run_one(target, spec: dict): + """Dispatch a single {op, ...} request. `target` is a RepoGraph or Workspace.""" + op = spec.get("op") + if op == "stats": + return target.stats() if isinstance(target, Workspace) else single_stats(target) + if op == "cypher": + if isinstance(target, Workspace): + return {r: g.rows(spec["q"]) for r, g in target.repos.items()} + return target.rows(spec["q"]) + if op == "blast-radius": + name, path = spec.get("name"), spec.get("path") + if not (name or path): + raise SystemExit("blast-radius requires a path (preferred) or a name") + return target.blast_radius(name, spec.get("hops", 3), path) + if op == "affected-repos": + if not isinstance(target, Workspace): + raise SystemExit("affected-repos needs --workspace") + return target.affected_repos(spec["repo"], spec.get("hops", 5)) + + if op not in PER_REPO_OPS: + raise SystemExit(f"unknown op: {op}") + method, params = PER_REPO_OPS[op] + args = [spec.get(p, DEFAULTS.get(p)) for p in params] + if any(a is None for a in args): + missing = [p for p, a in zip(params, args) if a is None] + raise SystemExit(f"{op} requires: {', '.join(missing)}") + + if isinstance(target, Workspace): + # blast-radius has real cross-repo meaning; the rest simply fan out. + if op == "blast-radius": + return target.blast_radius(*args) + return target.fan_out(method, *args) + return getattr(target, method)(*args) + + +def single_stats(repo: RepoGraph) -> dict: + return { + "graph": repo.key, + "backend": repo.backend_kind, + "repos": [""], + **repo.counts(), + "sync": repo.synced, + "semantic": bool(repo.embedder), + "embedDimension": repo.embedder.dimension if repo.embedder else None, + } + + +def main() -> None: + reexec_under_capable_interpreter() + p = argparse.ArgumentParser(description=__doc__.split("\n")[0]) + p.add_argument("command", choices=[ + "stats", "search", "nodes-for-file", "neighbors", "blast-radius", + "layers-for", "calls-from", "calls-to", "path", "semantic", + "semantic-traverse", "affected-repos", "cypher", "batch", + ]) + p.add_argument("--root", default=".", help="project root (default: cwd)") + p.add_argument("--workspace", help="manifest listing several repos") + p.add_argument("--q", help="search term, query text, raw Cypher, or batch JSON") + p.add_argument("--id", help="node id") + p.add_argument("--name", help="node name (ambiguous for common basenames)") + p.add_argument("--path", help="file path (preferred for blast-radius)") + p.add_argument("--ids", help="comma-separated node ids (layers-for)") + p.add_argument("--repo", help="repo name (affected-repos)") + p.add_argument("--hops", type=int, default=3) + p.add_argument("--limit", type=int, default=25) + p.add_argument("--k", type=int, default=10) + p.add_argument("--from", dest="src") + p.add_argument("--to", dest="dst") + args = p.parse_args() + + embedder = make_embedder() + + if args.workspace: + manifest_path = Path(args.workspace).resolve() + manifest = json.loads(manifest_path.read_text()) + members = [] + for entry in manifest.get("repos", []): + repo_root = (manifest_path.parent / entry).resolve() + members.append((repo_root.name, find_graph_json(repo_root))) + if not members: + raise SystemExit(f"{manifest_path} lists no repos") + # One embedded instance beside the manifest holds every repo's graph. + backend = Backend(manifest_path.parent / ".ua") + target = Workspace(members, backend, embedder) + else: + graph_json = find_graph_json(Path(args.root).resolve()) + backend = Backend(graph_json.parent) + key = json.loads(graph_json.read_text()).get("project", {}).get("name", "ua") + target = RepoGraph(backend, key, graph_json, embedder) + + spec_from_flags = { + "q": args.q, "id": args.id, "name": args.name, + "path": args.path or (args.q if args.command == "nodes-for-file" else None), + "repo": args.repo, "hops": args.hops, "k": args.k, "limit": args.limit, + "ids": [i for i in (args.ids or "").split(",") if i] or None, + "from": args.src, "to": args.dst, + } + + if args.command == "batch": + # Many questions, one process. The embedded server's boot and shutdown + # dominate a single-shot invocation, so a skill answering a multi-part + # question should send its whole plan at once. + specs = json.loads(args.q) if args.q else json.load(sys.stdin) + out = [run_one(target, s) for s in specs] + else: + out = run_one(target, {"op": args.command, **spec_from_flags}) + + print(json.dumps(out, indent=2, default=str)) + + +if __name__ == "__main__": + main()