diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 080f46f223..b296f119c9 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -29,6 +29,7 @@ Signatures below are the real ones - `tests/test_architecture_doc.py` imports ev | `cache.py` | `check_semantic_cache(files, root)`, `save_semantic_cache(nodes, edges, ...)` | files → cached nodes / edges / hyperedges + the list of files still needing extraction | | `security.py` | `validate_url`, `safe_fetch`, `validate_graph_path`, `sanitize_label` | URL / path / label → validated value, or raises | | `validate.py` | `validate_extraction(data)`, `assert_valid(data)` | extraction dict → **list of schema error strings** (`validate_extraction` returns them; `assert_valid` raises) | +| `storage.py` | `init_db / ingest_extraction / ingest_communities` | extraction dict → NeuG `graph.db` (optional, requires `neug`) | | `serve.py` | `serve(graph_path)`, `serve_http(graph_path, *, host, port, ...)` | graph file path → MCP stdio server / HTTP server | | `watch.py` | `watch(watch_path, debounce=3.0)`, `check_update(watch_path)` | directory → rebuild on change; `check_update` reports whether a re-extraction is pending | | `benchmark.py` | `run_benchmark(graph_path)` | graph file → corpus vs subgraph token comparison | diff --git a/README.md b/README.md index 272c5f6f81..902346eb1c 100644 --- a/README.md +++ b/README.md @@ -254,6 +254,7 @@ Codex users also need `multi_agent = true` under `[features]` in `~/.codex/confi | `mcp` | MCP stdio server | `uv tool install "graphifyy[mcp]"` | | `neo4j` | Neo4j push support | `uv tool install "graphifyy[neo4j]"` | | `falkordb` | FalkorDB push support | `uv tool install "graphifyy[falkordb]"` | +| `neug` | [NeuG](https://github.com/alibaba/neug) embedded graph database — Cypher queries on your graph | `uv tool install "graphifyy[neug]"` | | `svg` | SVG graph export | `uv tool install "graphifyy[svg]"` | | `leiden` | Leiden community detection (Python < 3.13 only) | `uv tool install "graphifyy[leiden]"` | | `ollama` | Ollama local inference | `uv tool install "graphifyy[ollama]"` | @@ -788,6 +789,15 @@ graphify cluster-only ./my-project --backend=gemini # backend for com graphify cluster-only ./my-project --backend=gemini --model gemini-2.5-pro # specific model graphify label ./my-project # (re)name communities with the configured backend graphify label ./my-project --backend=openai --model gpt-4o # force a specific backend and model + +# NeuG embedded graph DB (requires the neug extra) +GRAPHIFY_NEUG=1 graphify extract ./raw # opt-in NeuG pipeline: build graph.db + cluster with neug GDS Leiden (stays active once graph.db exists, no env var needed after that) +GRAPHIFY_NEUG=1 graphify extract ./raw --resolution 1.2 # tune Leiden resolution +GRAPHIFY_NEUG=1 graphify extract ./raw --cluster-on-files # file-level communities +graphify delta-cluster ./raw # incremental community analysis on an existing graph.db +graphify delta-cluster ./raw --baseline communities.json # seed from an external clustering: old communities stay frozen, new nodes get assigned on top +graphify cypher "MATCH (n) RETURN n LIMIT 10" # query graph.db with Cypher +graphify cypher "MATCH (n:node)-[e:edge]->(m:node) RETURN n.id, e.relation, m.id LIMIT 10" --db path/to/graph.db # default: graphify-out/graph.db ``` > **Community names:** inside an agent (Claude Code, Gemini CLI) the agent names communities itself. When you run the bare CLI, `cluster-only` auto-names them with the configured backend (built-in or custom OpenAI-compatible provider) — pass `--no-label` to keep `Community N`, or run `graphify label` to (re)generate names on demand. diff --git a/graphify/__main__.py b/graphify/__main__.py index 155501a98d..f40fa90abf 100644 --- a/graphify/__main__.py +++ b/graphify/__main__.py @@ -552,6 +552,8 @@ def _run_cli() -> None: print(" --model= model to use for community naming") print(" --max-concurrency=N parallel labeling LLM calls (default 4; forced to 1 for ollama/claude-cli)") print(" --batch-size=N communities per labeling LLM call (default 100)") + print(" cypher \"MATCH ...\" execute a Cypher query against graph.db (requires neug)") + print(" --db path to graph.db (default graphify-out/graph.db)") print(" query \"\" 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)") diff --git a/graphify/cli.py b/graphify/cli.py index cb30420473..777328beea 100644 --- a/graphify/cli.py +++ b/graphify/cli.py @@ -967,6 +967,31 @@ def dispatch_command(cmd: str) -> None: else: print("Usage: graphify hook [install|uninstall|status]", file=sys.stderr) sys.exit(1) + elif cmd == "cypher": + if len(sys.argv) < 3: + print('Usage: graphify cypher "MATCH ..." [--db path]', file=sys.stderr) + sys.exit(1) + query_str = sys.argv[2] + db_path = str(Path(_GRAPHIFY_OUT) / "graph.db") + args = sys.argv[3:] + for i, a in enumerate(args): + if a == "--db" and i + 1 < len(args): + db_path = args[i + 1] + try: + from graphify.storage import init_db, execute_cypher, close_db + except ImportError: + print("error: neug is not installed. Run: pip install neug", file=sys.stderr) + sys.exit(1) + if not Path(db_path).exists(): + print(f"error: database not found: {db_path}", file=sys.stderr) + sys.exit(1) + db, conn = init_db(db_path) + try: + results = execute_cypher(conn, query_str) + for row in results: + print("\t".join(str(v) for v in row)) + finally: + close_db(db, conn) elif cmd == "query": if len(sys.argv) < 3: print("Usage: graphify query \"\" [--dfs] [--context C] [--budget N] [--graph path]", file=sys.stderr) @@ -2191,6 +2216,100 @@ def dispatch_command(cmd: str) -> None: check_update(Path(sys.argv[2]).resolve()) sys.exit(0) + elif cmd == "delta-cluster": + # Incremental community delta analysis (freeze-assign leiden). + # Requires: graphify extract (full) + graphify extract --no-cluster (DB updated). + # Outputs: .graphify_delta_analysis.json (does NOT modify DB or .graphify_analysis.json). + import json as _json + from graphify.storage import init_db as _init_db, close_db as _close_db, delta_analyze as _delta_analyze + + if len(sys.argv) < 3: + print("Usage: graphify delta-cluster [--resolution R] [--cluster-on-files] [--baseline ]", file=sys.stderr) + sys.exit(1) + + _delta_resolution: float = 1.0 + _delta_file_level: bool = False + _delta_baseline: str | None = None + _delta_args = sys.argv[2:] + _delta_pos: list[str] = [] + _di = 0 + while _di < len(_delta_args): + _da = _delta_args[_di] + if _da == "--resolution" and _di + 1 < len(_delta_args): + _delta_resolution = float(_delta_args[_di + 1]); _di += 2 + elif _da.startswith("--resolution="): + _delta_resolution = float(_da.split("=", 1)[1]); _di += 1 + elif _da == "--cluster-on-files": + _delta_file_level = True; _di += 1 + elif _da == "--baseline" and _di + 1 < len(_delta_args): + _delta_baseline = _delta_args[_di + 1]; _di += 2 + elif _da.startswith("--baseline="): + _delta_baseline = _da.split("=", 1)[1]; _di += 1 + else: + _delta_pos.append(_da); _di += 1 + if not _delta_pos: + print("Usage: graphify delta-cluster [--resolution R] [--cluster-on-files] [--baseline ]", file=sys.stderr) + sys.exit(1) + _target = Path(_delta_pos[0]).resolve() + _graphify_out = _target / _GRAPHIFY_OUT + _analysis_path = _graphify_out / ".graphify_analysis.json" + _delta_path = _graphify_out / ".graphify_delta_analysis.json" + _db_path = str(_graphify_out / "graph.db") + + # --baseline overrides the default analysis file + if _delta_baseline: + _baseline_path = Path(_delta_baseline) + if not _baseline_path.is_absolute(): + _baseline_path = _graphify_out / _delta_baseline + if not _baseline_path.exists(): + print( + f"[graphify delta-cluster] baseline file not found: {_baseline_path}", + file=sys.stderr, + ) + sys.exit(1) + _analysis_path = _baseline_path + + if not _analysis_path.exists(): + print( + f"[graphify delta-cluster] no baseline found at {_analysis_path}.\n" + f"Run 'graphify extract {_target}' first.", + file=sys.stderr, + ) + sys.exit(1) + if not Path(_db_path).exists(): + print( + f"[graphify delta-cluster] no graph.db found at {_db_path}.\n" + f"Run 'graphify extract --no-cluster {_target}' to update the DB.", + file=sys.stderr, + ) + sys.exit(1) + + print(f"[graphify delta-cluster] analyzing {_target}") + _prev_analysis = _json.loads(_analysis_path.read_text(encoding="utf-8")) + _stages = _StageTimer(False) + _neug_db, _neug_conn = _init_db(_db_path) + try: + _delta = _delta_analyze( + _neug_conn, + prev_analysis=_prev_analysis, + delta_analysis_path=_delta_path, + stages=_stages, + merged={"input_tokens": 0, "output_tokens": 0}, + resolution=_delta_resolution, + file_level=_delta_file_level, + ) + finally: + _close_db(_neug_db, _neug_conn) + + _s = _delta["summary"] + print(f"[graphify delta-cluster] wrote {_delta_path}") + print( + f"[graphify delta-cluster] communities: " + f"{_s['total_before']} before → {_s['total_after']} after " + f"({_s['stable']} stable, {_s['changed']} changed, " + f"{_s['new']} new, {_s['dissolved']} dissolved)" + ) + sys.exit(0) elif cmd == "tree": # Emit a D3 v7 collapsible-tree HTML view of graph.json: # expand-all / collapse-all / reset-view buttons, multi-line @@ -2890,6 +3009,7 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": cli_exclude_hubs: float | None = None cli_excludes: list[str] = [] cli_timing: bool = False + cli_cluster_on_files: bool = False # --force parity with `graphify update`: the flag or GRAPHIFY_FORCE=1 # disables the incremental gate and skips semantic-cache reads (#1894). force = os.environ.get("GRAPHIFY_FORCE", "").lower() in ("1", "true", "yes") @@ -2981,6 +3101,8 @@ def _parse_float(name: str, raw: str) -> float: cli_excludes.append(args[i + 1]); i += 2 elif a.startswith("--exclude="): cli_excludes.append(a.split("=", 1)[1]); i += 1 + elif a == "--cluster-on-files": + cli_cluster_on_files = True; i += 1 elif a == "--postgres" and i + 1 < len(args): cli_postgres_dsn = args[i + 1]; i += 2 elif a.startswith("--postgres="): @@ -3722,156 +3844,267 @@ def _invalidate_file_manifest_for_db_graph() -> None: print(f"error: could not invalidate file manifest: {exc}", file=sys.stderr) sys.exit(1) - if no_cluster: - # --no-cluster: dump the raw merged extraction as graph.json. - # No NetworkX, no community detection, no analysis sidecar. - # Dedupe nodes (by id) and parallel edges so the raw output matches the - # clustered path (whose DiGraph collapses both) and stays deterministic - # across modes (#1317; node dedup also collapses shared Swift module - # anchors emitted per importing file, #1327). - from graphify.build import dedupe_edges as _dedupe_edges, dedupe_nodes as _dedupe_nodes - from graphify.export import ( - backup_if_protected as _backup, - existing_graph_node_count as _existing_graph_node_count, + # --- neug / NetworkX path separation --- + # The neug pipeline (graph.db + GDS Leiden) is opt-in: enable it with + # GRAPHIFY_NEUG=1, or it stays active once a graph.db already exists in + # the output dir (continuity for projects that adopted it). A fresh + # extract without the env var always takes the upstream NetworkX path, + # which carries the latest extraction-fix behaviour. + try: + import neug as _neug_mod # noqa: F401 + _neug_available = True + except ImportError: + _neug_available = False + _use_neug = _neug_available and ( + os.environ.get("GRAPHIFY_NEUG") in ("1", "true", "yes", "on") + or (graphify_out / "graph.db").exists() + ) + + _neug_conn = None + _neug_db = None + if _use_neug: + from graphify.storage import ( + init_db as _init_db, ensure_schema as _ensure_schema, + ingest_extraction as _ingest, close_db as _close_db, + export_to_json as _export_to_json, ) - if ( - incremental_mode - and not code_files - and not semantic_files - and not deleted_files - and not pg_result.get("nodes") - and not pg_result.get("edges") - and not cargo_result.get("nodes") - and not cargo_result.get("edges") - ): - # An exclusion-only change reaches this gate (excluded files - # are deliberately NOT in deleted_files, #1908) but must still - # scrub the newly-excluded sources from the raw graph (#1909). - # This path never runs build_merge, so prune in place. - if graph_stale_sources: - _n_pruned = _prune_graph_json_sources( - existing_graph_path, graph_stale_sources - ) - if _n_pruned: - print( - f"[graphify extract] pruned {_n_pruned} node(s) from " - f"{len(graph_stale_sources)} source file(s) no longer " - "in the scan (deleted or excluded)." - ) + from graphify.export import backup_if_protected as _backup + _db_path = str(graphify_out / "graph.db") + _is_inc = Path(_db_path).exists() + _backup(graphify_out) + _neug_db, _neug_conn = _init_db(_db_path) + _known = _ensure_schema(_neug_conn, create_tables=not _is_inc) + _ingest(_neug_conn, merged, incremental=_is_inc, + prune_sources=deleted_files or None, root=target, + known_tables=_known) + print("[graphify extract] graph.db written (powered by NeuG)") + + if no_cluster: + # --no-cluster: no NetworkX, no community detection, no analysis sidecar. + if _use_neug: + # neug path: graph.db already built; export graph.json from it. + _data = _export_to_json(_neug_conn, hyperedges=merged.get("hyperedges", [])) + _close_db(_neug_db, _neug_conn) + graph_json_path.write_text(json.dumps(_data, indent=2), encoding="utf-8") + stages.mark("write") + cost = _estimate_cost( + backend, merged["input_tokens"], merged["output_tokens"] + ) print( - "[graphify extract] no incremental changes detected " - "(--no-cluster); outputs left untouched." + f"[graphify extract] wrote {graph_json_path} — " + f"{len(_data['nodes'])} nodes, {len(_data['links'])} edges " + f"(no clustering)" ) + if merged["input_tokens"] or merged["output_tokens"]: + print( + f"[graphify extract] tokens: " + f"{merged['input_tokens']:,} in / " + f"{merged['output_tokens']:,} out, " + f"est. cost: ${cost:.4f}" + ) try: - _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus, clear_semantic=_cleared_semantic, clear_ast=_cleared_ast or None) + if has_path: + _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus, clear_semantic=_cleared_semantic, clear_ast=_cleared_ast or None) except Exception as exc: print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr) + if global_merge: + from graphify.global_graph import global_add as _global_add + _tag = global_repo_tag or target.name + try: + result = _global_add(graphify_out / "graph.json", _tag) + if result["skipped"]: + print(f"[graphify global] '{_tag}' unchanged since last add - skipped.") + else: + print(f"[graphify global] '{_tag}' merged into global graph " + f"(+{result['nodes_added']} nodes, -{result['nodes_removed']} pruned).") + except Exception as exc: + print(f"[graphify global] warning: failed to merge into global graph: {exc}", file=sys.stderr) stages.total() sys.exit(0) + else: + # No NetworkX, no community detection, no analysis sidecar. + # Dedupe nodes (by id) and parallel edges so the raw output matches the + # clustered path (whose DiGraph collapses both) and stays deterministic + # across modes (#1317; node dedup also collapses shared Swift module + # anchors emitted per importing file, #1327). + from graphify.build import dedupe_edges as _dedupe_edges, dedupe_nodes as _dedupe_nodes + from graphify.export import ( + backup_if_protected as _backup, + existing_graph_node_count as _existing_graph_node_count, + ) + if ( + incremental_mode + and not code_files + and not semantic_files + and not deleted_files + and not pg_result.get("nodes") + and not pg_result.get("edges") + and not cargo_result.get("nodes") + and not cargo_result.get("edges") + ): + # An exclusion-only change reaches this gate (excluded files + # are deliberately NOT in deleted_files, #1908) but must still + # scrub the newly-excluded sources from the raw graph (#1909). + # This path never runs build_merge, so prune in place. + if graph_stale_sources: + _n_pruned = _prune_graph_json_sources( + existing_graph_path, graph_stale_sources + ) + if _n_pruned: + print( + f"[graphify extract] pruned {_n_pruned} node(s) from " + f"{len(graph_stale_sources)} source file(s) no longer " + "in the scan (deleted or excluded)." + ) + print( + "[graphify extract] no incremental changes detected " + "(--no-cluster); outputs left untouched." + ) + try: + _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus, clear_semantic=_cleared_semantic) + except Exception as exc: + print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr) + stages.total() + sys.exit(0) - if incremental_mode: - # #2169: this raw path used to write ONLY this run's extraction - # over graph.json — on an incremental run that is just the - # changed files, silently dropping every node/edge owned by an - # unchanged file. Merge the existing graph forward first, with - # the same replace/prune semantics as the clustered path's - # build_merge: re-extracted sources replaced, deleted + - # excluded + graph-stale sources pruned, everything else - # carried. Survivors are prepended, so the dedupe below keeps - # this run's fresh attributes for re-extracted nodes. - from graphify.build import merge_raw_extraction as _merge_raw_extraction - _raw_prune_sources: list[str] = list(deleted_files) - for _src in list(excluded_files) + graph_stale_sources: - if _src not in _raw_prune_sources: - _raw_prune_sources.append(_src) + if incremental_mode: + # #2169: this raw path used to write ONLY this run's extraction + # over graph.json — on an incremental run that is just the + # changed files, silently dropping every node/edge owned by an + # unchanged file. Merge the existing graph forward first, with + # the same replace/prune semantics as the clustered path's + # build_merge: re-extracted sources replaced, deleted + + # excluded + graph-stale sources pruned, everything else + # carried. Survivors are prepended, so the dedupe below keeps + # this run's fresh attributes for re-extracted nodes. + from graphify.build import merge_raw_extraction as _merge_raw_extraction + _raw_prune_sources: list[str] = list(deleted_files) + for _src in list(excluded_files) + graph_stale_sources: + if _src not in _raw_prune_sources: + _raw_prune_sources.append(_src) + try: + merged = _merge_raw_extraction( + merged, + graph_path=existing_graph_path, + prune_sources=_raw_prune_sources or None, + root=target, + ) + except RuntimeError as exc: + # Existing graph present but unparseable: refuse to + # raw-dump this run's partial extraction over it. + print(f"error: {exc}", file=sys.stderr) + sys.exit(1) + merged["nodes"] = _dedupe_nodes(merged["nodes"]) + merged["edges"] = _dedupe_edges(merged["edges"]) + # Disambiguate colliding-basename file-node labels (#2032). This raw + # --no-cluster path bypasses build_from_json (where the clustered path + # gets this), so apply it directly on the merged node list. + from graphify.build import disambiguate_file_labels_in_nodes as _disamb_labels + _disamb_labels(merged["nodes"]) + # Backfill source_file from endpoint nodes — this raw path bypasses + # build_from_json's backfill, and semantic edges sometimes omit it (#1279). + _node_sf = {n.get("id"): n.get("source_file") for n in merged["nodes"]} + for _e in merged["edges"]: + if not _e.get("source_file"): + _e["source_file"] = ( + _node_sf.get(_e.get("source")) or _node_sf.get(_e.get("target")) or "" + ) + # RT-parity for the raw path: an incomplete build must not force a + # partial graph over a larger complete one here either. The clustered + # path gets this from to_json's #479 guard; this path never calls + # to_json, so replicate the shrink check against the existing file and + # exit before the write/manifest unless --allow-partial is set. + if _extraction_incomplete and not cli_allow_partial: + from graphify.export import MALFORMED_GRAPH as _MALFORMED_GRAPH + _existing_n = _existing_graph_node_count(graph_json_path) + _malformed = _existing_n is _MALFORMED_GRAPH + _shrinks = isinstance(_existing_n, int) and len(merged["nodes"]) < _existing_n + if _malformed or _shrinks: + _detail = ( + f"the existing {graph_json_path} is present but unparseable " + "(corrupt or a mid-write), so a shrink cannot be ruled out" + if _malformed + else f"smaller than the existing {graph_json_path} " + f"({len(merged['nodes'])} < {_existing_n} nodes)" + ) + print( + "[graphify extract] error: extraction was incomplete (an AST/" + f"semantic pass failed) and the resulting --no-cluster graph is {_detail}. " + "Refusing to overwrite a complete graph with a partial one. Re-run after " + "fixing the failures, or pass --allow-partial to overwrite anyway.", + file=sys.stderr, + ) + sys.exit(1) + _backup(graphify_out) + _invalidate_file_manifest_for_db_graph() + from graphify.paths import write_json_atomic as _write_json_atomic + _write_json_atomic(graph_json_path, merged, indent=2) try: - merged = _merge_raw_extraction( - merged, - graph_path=existing_graph_path, - prune_sources=_raw_prune_sources or None, - root=target, - ) - except RuntimeError as exc: - # Existing graph present but unparseable: refuse to - # raw-dump this run's partial extraction over it. - print(f"error: {exc}", file=sys.stderr) - sys.exit(1) - merged["nodes"] = _dedupe_nodes(merged["nodes"]) - merged["edges"] = _dedupe_edges(merged["edges"]) - # Disambiguate colliding-basename file-node labels (#2032). This raw - # --no-cluster path bypasses build_from_json (where the clustered path - # gets this), so apply it directly on the merged node list. - from graphify.build import disambiguate_file_labels_in_nodes as _disamb_labels - _disamb_labels(merged["nodes"]) - # Backfill source_file from endpoint nodes — this raw path bypasses - # build_from_json's backfill, and semantic edges sometimes omit it (#1279). - _node_sf = {n.get("id"): n.get("source_file") for n in merged["nodes"]} - for _e in merged["edges"]: - if not _e.get("source_file"): - _e["source_file"] = ( - _node_sf.get(_e.get("source")) or _node_sf.get(_e.get("target")) or "" - ) - # RT-parity for the raw path: an incomplete build must not force a - # partial graph over a larger complete one here either. The clustered - # path gets this from to_json's #479 guard; this path never calls - # to_json, so replicate the shrink check against the existing file and - # exit before the write/manifest unless --allow-partial is set. - if _extraction_incomplete and not cli_allow_partial: - from graphify.export import MALFORMED_GRAPH as _MALFORMED_GRAPH - _existing_n = _existing_graph_node_count(graph_json_path) - _malformed = _existing_n is _MALFORMED_GRAPH - _shrinks = isinstance(_existing_n, int) and len(merged["nodes"]) < _existing_n - if _malformed or _shrinks: - _detail = ( - f"the existing {graph_json_path} is present but unparseable " - "(corrupt or a mid-write), so a shrink cannot be ruled out" - if _malformed - else f"smaller than the existing {graph_json_path} " - f"({len(merged['nodes'])} < {_existing_n} nodes)" + # Record the scan root so a later build_merge / update runbook can + # relativize deleted-file paths correctly even for a custom --out + # (its grandparent-of-graph.json fallback points at the wrong dir + # otherwise, and deleted files never prune — #2012/#1571). + (graphify_out / ".graphify_root").write_text( + str(Path(target).resolve()), encoding="utf-8" ) + except OSError: + pass + stages.mark("write") + cost = _estimate_cost( + backend, merged["input_tokens"], merged["output_tokens"] + ) + print( + f"[graphify extract] wrote {graph_json_path} — " + f"{len(merged['nodes'])} nodes, {len(merged['edges'])} edges " + f"(no clustering)" + ) + if merged["input_tokens"] or merged["output_tokens"]: print( - "[graphify extract] error: extraction was incomplete (an AST/" - f"semantic pass failed) and the resulting --no-cluster graph is {_detail}. " - "Refusing to overwrite a complete graph with a partial one. Re-run after " - "fixing the failures, or pass --allow-partial to overwrite anyway.", - file=sys.stderr, + f"[graphify extract] tokens: " + f"{merged['input_tokens']:,} in / " + f"{merged['output_tokens']:,} out, " + f"est. cost: ${cost:.4f}" ) - sys.exit(1) - _backup(graphify_out) - _invalidate_file_manifest_for_db_graph() - from graphify.paths import write_json_atomic as _write_json_atomic - _write_json_atomic(graph_json_path, merged, indent=2) - try: - # Record the scan root so a later build_merge / update runbook can - # relativize deleted-file paths correctly even for a custom --out - # (its grandparent-of-graph.json fallback points at the wrong dir - # otherwise, and deleted files never prune — #2012/#1571). - (graphify_out / ".graphify_root").write_text( - str(Path(target).resolve()), encoding="utf-8" - ) - except OSError: - pass - stages.mark("write") - cost = _estimate_cost( - backend, merged["input_tokens"], merged["output_tokens"] - ) - print( - f"[graphify extract] wrote {graph_json_path} — " - f"{len(merged['nodes'])} nodes, {len(merged['edges'])} edges " - f"(no clustering)" + try: + if has_path: + _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus, clear_semantic=_cleared_semantic) + except Exception as exc: + print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr) + if global_merge: + from graphify.global_graph import global_add as _global_add + _tag = global_repo_tag or target.name + try: + result = _global_add(graphify_out / "graph.json", _tag) + if result["skipped"]: + print(f"[graphify global] '{_tag}' unchanged since last add - skipped.") + else: + print(f"[graphify global] '{_tag}' merged into global graph " + f"(+{result['nodes_added']} nodes, -{result['nodes_removed']} pruned).") + except Exception as exc: + print(f"[graphify global] warning: failed to merge into global graph: {exc}", file=sys.stderr) + stages.total() + sys.exit(0) + + # Build graph + cluster + score + write. + if _use_neug: + # --- neug clustered path --- + from graphify.storage import cluster_by_neug as _cluster_by_neug + _data = _cluster_by_neug( + _neug_conn, + merged=merged, + graph_json_path=graph_json_path, + analysis_path=analysis_path, + stages=stages, + export_fn=_export_to_json, + hyperedges=merged.get("hyperedges", []), + resolution=cli_resolution, + file_level=cli_cluster_on_files, ) - if merged["input_tokens"] or merged["output_tokens"]: - print( - f"[graphify extract] tokens: " - f"{merged['input_tokens']:,} in / " - f"{merged['output_tokens']:,} out, " - f"est. cost: ${cost:.4f}" + _close_db(_neug_db, _neug_conn) + if merged.get("output_tokens", 0) > 0: + (graphify_out / ".graphify_semantic_marker").write_text( + json.dumps({"output_tokens": merged["output_tokens"]}), encoding="utf-8" ) - try: - if has_path: - _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus, clear_semantic=_cleared_semantic, clear_ast=_cleared_ast or None) - except Exception as exc: - print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr) if global_merge: from graphify.global_graph import global_add as _global_add _tag = global_repo_tag or target.name @@ -4004,65 +4237,201 @@ def _invalidate_file_manifest_for_db_graph() -> None: from graphify.global_graph import global_add as _global_add _tag = global_repo_tag or target.name try: - result = _global_add(graphify_out / "graph.json", _tag) - if result["skipped"]: - print(f"[graphify global] '{_tag}' unchanged since last add - skipped.") - else: - print(f"[graphify global] '{_tag}' merged into global graph " - f"(+{result['nodes_added']} nodes, -{result['nodes_removed']} pruned).") - except Exception as exc: - print(f"[graphify global] warning: failed to merge into global graph: {exc}", file=sys.stderr) - analysis = { - "communities": {str(k): v for k, v in communities.items()}, - "cohesion": {str(k): v for k, v in cohesion.items()}, - "gods": gods, - "surprises": surprises, - "tokens": { - "input": merged["input_tokens"], - "output": merged["output_tokens"], - }, - } - from graphify.paths import write_json_atomic as _wja - _wja(analysis_path, analysis, indent=2) - try: - if has_path: _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus, clear_semantic=_cleared_semantic, clear_ast=_cleared_ast or None) - except Exception as exc: - print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr) + except Exception as exc: + print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr) + cost = _estimate_cost(backend, merged["input_tokens"], merged["output_tokens"]) + _comm_count = len({n.get("community") for n in _data["nodes"] if n.get("community") is not None}) + print( + f"[graphify extract] wrote {graph_json_path}: " + f"{len(_data['nodes'])} nodes, {len(_data['links'])} edges, " + f"{_comm_count} communities" + ) + print(f"[graphify extract] wrote {analysis_path}") + if incremental_mode: + print( + f"[graphify extract] incremental summary: " + f"{sem_cache_hits + unchanged_total} files cached/unchanged, " + f"{len(code_files) + sem_cache_misses} re-extracted, " + f"{len(deleted_files)} deleted" + ) + elif sem_cache_hits: + print(f"[graphify extract] semantic cache: {sem_cache_hits} cached, {sem_cache_misses} re-extracted") + if merged["input_tokens"] or merged["output_tokens"]: + print( + f"[graphify extract] tokens: " + f"{merged['input_tokens']:,} in / " + f"{merged['output_tokens']:,} out, " + f"est. cost (~{backend}): ${cost:.4f}" + ) + stages.total() - cost = _estimate_cost(backend, merged["input_tokens"], merged["output_tokens"]) - print( - f"[graphify extract] wrote {graph_json_path}: " - f"{G.number_of_nodes()} nodes, {G.number_of_edges()} edges, " - f"{len(communities)} communities" - ) - print(f"[graphify extract] wrote {analysis_path}") - if incremental_mode: - _excl_note = f", {len(excluded_files)} excluded" if excluded_files else "" + else: + from graphify.build import ( + build as _build, + build_from_json as _build_from_json, + build_merge as _build_merge, + ) + from graphify.cluster import cluster as _cluster, score_all as _score_all + from graphify.export import to_json as _to_json + from graphify.analyze import god_nodes as _god_nodes, surprising_connections as _surprising + dedup_backend = backend if dedup_llm else None + if incremental_mode: + # Prune everything the current scan no longer covers: genuinely + # deleted manifest rows, excluded-but-alive manifest rows (#1908), + # and the graph's own stale sources — which catches files that + # became excluded without ever being manifest-listed (#1909). + _prune_sources: list[str] = list(deleted_files) + for _src in list(excluded_files) + graph_stale_sources: + if _src not in _prune_sources: + _prune_sources.append(_src) + G = _build_merge( + [merged], + graph_path=existing_graph_path, + prune_sources=_prune_sources or None, + dedup=True, + dedup_llm_backend=dedup_backend, + root=target, + ) + else: + G = _build([merged], dedup=True, dedup_llm_backend=dedup_backend, root=target) + stages.mark("build") + if G.number_of_nodes() == 0: + print( + "[graphify extract] graph is empty — extraction produced no nodes. " + "Possible causes: all files skipped, binary-only corpus, or LLM " + "returned no edges.", + file=sys.stderr, + ) + sys.exit(1) + + communities = _cluster(G, resolution=cli_resolution, exclude_hubs_percentile=cli_exclude_hubs) + stages.mark("cluster") + cohesion = _score_all(G, communities) + try: + gods = _god_nodes(G) + except Exception: + gods = [] + try: + surprises = _surprising(G, communities) + except Exception: + surprises = [] + stages.mark("analyze") + + from graphify.export import backup_if_protected as _backup + _backup(graphify_out) + _invalidate_file_manifest_for_db_graph() + # force=True bypasses the #479 shrink guard entirely. A full build + # legitimately shrinks (fuzzy dedup collapse, deleted code) so it keeps + # force=True — EXCEPT when this run's extraction was incomplete (an + # extractor pass crashed or some semantic chunks failed). Then a partial + # graph could silently overwrite a good complete one, so fall back to the + # shrink guard (force=False) unless the user opts in with --allow-partial. + # + # Both write paths are guarded: the clustered path here via to_json's + # #479 check, and the `--no-cluster` raw-dump path above via the same + # shrink check against the existing file (existing_graph_node_count). + # + # Trade-off: this reuses to_json's coarse node-count guard, not the + # source-aware _check_shrink that watch/update use. On an incremental run + # a legitimate deletion that coincides with an unrelated transient chunk + # failure can therefore be refused here — recoverable by re-running or + # passing --allow-partial (the good graph is preserved and the manifest + # is not stamped, so the retry re-extracts). + _force_write = cli_allow_partial or not _extraction_incomplete + _wrote = _to_json(G, communities, str(graph_json_path), force=_force_write) + if not _wrote: + # The shrink guard refused: this partial build is smaller than the + # existing graph. Exit before writing the manifest/marker below, which + # would otherwise stamp these files as done and make the next + # incremental run skip re-extracting them (poisoning the manifest + # against the graph we declined to write). Exit non-zero so a retry + # re-attempts. + print( + "[graphify extract] error: extraction was incomplete (an AST/semantic " + f"pass failed) and the resulting graph is smaller than the existing " + f"{graph_json_path}. Refusing to overwrite a complete graph with a " + "partial one. Re-run after fixing the failures, or pass --allow-partial " + "to overwrite anyway.", + file=sys.stderr, + ) + sys.exit(1) + try: + # See the --no-cluster path above: persist the scan root so build_merge + # can relativize deleted-file paths under a custom --out (#2012/#1571). + (graphify_out / ".graphify_root").write_text( + str(Path(target).resolve()), encoding="utf-8" + ) + except OSError: + pass + stages.mark("export") + if merged.get("output_tokens", 0) > 0: + (graphify_out / ".graphify_semantic_marker").write_text( + json.dumps({"output_tokens": merged["output_tokens"]}), encoding="utf-8" + ) + if global_merge: + from graphify.global_graph import global_add as _global_add + _tag = global_repo_tag or target.name + try: + result = _global_add(graphify_out / "graph.json", _tag) + if result["skipped"]: + print(f"[graphify global] '{_tag}' unchanged since last add - skipped.") + else: + print(f"[graphify global] '{_tag}' merged into global graph " + f"(+{result['nodes_added']} nodes, -{result['nodes_removed']} pruned).") + except Exception as exc: + print(f"[graphify global] warning: failed to merge into global graph: {exc}", file=sys.stderr) + analysis = { + "communities": {str(k): v for k, v in communities.items()}, + "cohesion": {str(k): v for k, v in cohesion.items()}, + "gods": gods, + "surprises": surprises, + "tokens": { + "input": merged["input_tokens"], + "output": merged["output_tokens"], + }, + } + from graphify.paths import write_json_atomic as _wja + _wja(analysis_path, analysis, indent=2) + try: + if has_path: + _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus, clear_semantic=_cleared_semantic) + except Exception as exc: + print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr) + + cost = _estimate_cost(backend, merged["input_tokens"], merged["output_tokens"]) print( - f"[graphify extract] incremental summary: " - f"{sem_cache_hits + unchanged_total} files cached/unchanged, " - f"{len(code_files) + sem_cache_misses} re-extracted, " - f"{len(deleted_files)} deleted{_excl_note}" + f"[graphify extract] wrote {graph_json_path}: " + f"{G.number_of_nodes()} nodes, {G.number_of_edges()} edges, " + f"{len(communities)} communities" ) - elif sem_cache_hits: - print(f"[graphify extract] semantic cache: {sem_cache_hits} cached, {sem_cache_misses} re-extracted") - if merged["input_tokens"] or merged["output_tokens"]: + print(f"[graphify extract] wrote {analysis_path}") + if incremental_mode: + _excl_note = f", {len(excluded_files)} excluded" if excluded_files else "" + print( + f"[graphify extract] incremental summary: " + f"{sem_cache_hits + unchanged_total} files cached/unchanged, " + f"{len(code_files) + sem_cache_misses} re-extracted, " + f"{len(deleted_files)} deleted{_excl_note}" + ) + elif sem_cache_hits: + print(f"[graphify extract] semantic cache: {sem_cache_hits} cached, {sem_cache_misses} re-extracted") + if merged["input_tokens"] or merged["output_tokens"]: + print( + f"[graphify extract] tokens: " + f"{merged['input_tokens']:,} in / " + f"{merged['output_tokens']:,} out, " + f"est. cost (~{backend}): ${cost:.4f}" + ) + # extract intentionally stops at graph.json + analysis; the report and + # community labels are produced by `cluster-only` (or an agent's Step 5). + # Point standalone users at it so communities get named (#1097). print( - f"[graphify extract] tokens: " - f"{merged['input_tokens']:,} in / " - f"{merged['output_tokens']:,} out, " - f"est. cost (~{backend}): ${cost:.4f}" + "[graphify extract] next: run " + f"`graphify cluster-only {graphify_out.parent}` " + "to generate GRAPH_REPORT.md and name communities" ) - # extract intentionally stops at graph.json + analysis; the report and - # community labels are produced by `cluster-only` (or an agent's Step 5). - # Point standalone users at it so communities get named (#1097). - print( - "[graphify extract] next: run " - f"`graphify cluster-only {graphify_out.parent}` " - "to generate GRAPH_REPORT.md and name communities" - ) - stages.total() + stages.total() elif cmd == "cache-check": # graphify cache-check [--root ] [--mode | --deep] diff --git a/graphify/serve.py b/graphify/serve.py index 4cf6d83968..1066dd5366 100644 --- a/graphify/serve.py +++ b/graphify/serve.py @@ -1533,6 +1533,21 @@ def _build_server(graph_path: str): _default_graph_path = str(Path(graph_path).resolve()) _ctx_cache = _GraphContextCache(_max_server_contexts()) + # NeuG embedded graph database for Cypher queries on graph.db + _neug_conn = None + _neug_db = None + _neug_execute = None + try: + from graphify.storage import init_db as _neug_init, execute_cypher as _neug_exec, close_db as _neug_close + _neug_db_path = str(Path(graph_path).parent / "graph.db") + if Path(_neug_db_path).exists(): + _neug_db, _neug_conn = _neug_init(_neug_db_path) + _neug_execute = _neug_exec + except ImportError: + pass + except Exception: + pass + def _load_ctx(path: str): """Return the current default or project graph context as a tool error. @@ -1706,6 +1721,20 @@ async def list_tools() -> list[types.Tool]: }, }, ), + types.Tool( + name="cypher_query", + description=( + "Execute a Cypher query against the NeuG graph database. " + "Returns tabular results. Requires neug to be installed and graph.db to exist." + ), + inputSchema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Cypher query string"}, + }, + "required": ["query"], + }, + ), ] # Multi-project support: every tool accepts an optional project_path. # Injected here (rather than repeated in 11 literal schemas) so the set @@ -1949,6 +1978,22 @@ def _tool_triage_prs(arguments: dict) -> str: ) return "\n\n".join(lines) + def _tool_cypher_query(arguments: dict) -> str: + if _neug_conn is None: + return "NeuG not available (not installed or graph.db not found)." + query = arguments["query"] + from graphify.storage import execute_cypher as _exec_cypher + try: + results = _exec_cypher(_neug_conn, query) + except RuntimeError as exc: + return f"Cypher error: {exc}" + if not results: + return "No results." + lines = [] + for row in results: + lines.append("\t".join(str(v) for v in row)) + return "\n".join(lines) + _handlers = { "query_graph": _tool_query_graph, "get_node": _tool_get_node, @@ -1960,6 +2005,7 @@ def _tool_triage_prs(arguments: dict) -> str: "list_prs": _tool_list_prs, "get_pr_impact": _tool_get_pr_impact, "triage_prs": _tool_triage_prs, + "cypher_query": _tool_cypher_query, } def _load_community_labels() -> dict[int, str]: diff --git a/graphify/storage.py b/graphify/storage.py new file mode 100644 index 0000000000..e4c5ea35b2 --- /dev/null +++ b/graphify/storage.py @@ -0,0 +1,1834 @@ +"""NeuG graph database adapter for graphify. + +Provides an optional parallel storage engine alongside NetworkX. +NeuG is lazily imported — when not installed, callers should catch +ImportError at the call site and skip silently. + +All property values interpolated into Cypher statements use NeuG's native +parameterised queries ($param syntax) to prevent injection. Table/label +names (which come from a fixed internal set, not user input) are still +interpolated as identifiers. + +Single-table schema: one node table + one edge table, with file_type and +relation as properties (not separate tables). This aligns with graphify's +NetworkX graph model and enables neug GDS algorithms that operate on a +single graph. +""" +from __future__ import annotations + +import csv +import os +import tempfile +from pathlib import Path + +from .build import _FILE_TYPE_SYNONYMS, _normalize_id, _norm_source_file +from .validate import VALID_FILE_TYPES + +# --------------------------------------------------------------------------- +# Single-table schema +# --------------------------------------------------------------------------- + +_NODE_DDL = """CREATE NODE TABLE IF NOT EXISTS node ( + id STRING PRIMARY KEY, label STRING, file_type STRING, + source_file STRING, source_location STRING, + community INT64, community_name STRING)""" + +_NODE_COLUMNS = ["id", "label", "file_type", "source_file", "source_location", + "community", "community_name"] + +_EDGE_DDL = """CREATE REL TABLE IF NOT EXISTS edge ( + FROM node TO node, + relation STRING, confidence STRING, + confidence_score DOUBLE, source_file STRING, weight DOUBLE)""" + +_EDGE_COLUMNS = ["from_id", "to_id", "relation", "confidence", + "confidence_score", "source_file", "weight"] + + +# --------------------------------------------------------------------------- +# CSV helpers for bulk COPY FROM +# --------------------------------------------------------------------------- + +def _sanitize_csv_value(v: object) -> str: + if isinstance(v, str): + return v.replace("\n", "\\n").replace("\r", "") + return str(v) + + +def _write_csv(path: str, rows: list[dict], columns: list[str]) -> int: + if not rows: + return 0 + with open(path, "w", newline="", encoding="utf-8") as f: + w = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore", + quoting=csv.QUOTE_ALL) + w.writeheader() + for row in rows: + w.writerow({k: _sanitize_csv_value(row.get(k, "")) for k in columns}) + return len(rows) + + +def _copy_node_csv(conn: object, csv_path: str) -> None: + conn.execute( + f'COPY node FROM "{csv_path}" (header=true, delim=",", escaping=false)' + ) + + +def _copy_rel_csv(conn: object, csv_path: str) -> None: + conn.execute( + f'COPY edge FROM "{csv_path}" ' + f'(from="node", to="node", ' + f'header=true, delim=",", escaping=false)' + ) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def init_db(db_path: str) -> tuple: + """Open (or create) a NeuG database and connect. + + Returns (db, conn). Raises ImportError if neug is not installed. + """ + import neug + db = neug.Database(db_path) + conn = db.connect() + return db, conn + + +def ensure_schema(conn: object, *, create_tables: bool = True) -> set[str]: + """Create the single node + edge tables if needed. + + Returns an empty set (kept for backward-compat with callers that + pass the return value to ingest_extraction's known_tables). + """ + if create_tables: + conn.execute(_NODE_DDL) + conn.execute(_EDGE_DDL) + return set() + + +def _fix_file_type(ft: str | None) -> str: + """Canonicalize file_type, matching build.py:138-146 logic.""" + if not ft or ft not in VALID_FILE_TYPES: + return _FILE_TYPE_SYNONYMS.get(ft, "concept") if ft else "concept" + return ft + + +def _bulk_ingest( + conn: object, + extraction: dict, + *, + root: str | None = None, + known_tables: set[str] | None = None, +) -> dict[str, str]: + """Full build via COPY FROM — much faster than per-row Cypher CREATE.""" + nodes = extraction.get("nodes") or [] + edges = extraction.get("edges") or [] + + # --- collect node rows (single table) --- + node_types: dict[str, str] = {} + node_rows: list[dict] = [] + written_ids: set[str] = set() + + for node in nodes: + nid = _normalize_id(node.get("id", "")) + if not nid or nid in written_ids: + continue + written_ids.add(nid) + ft = _fix_file_type(node.get("file_type")) + node_types[nid] = ft + node_rows.append({ + "id": nid, + "label": node.get("label", ""), + "file_type": ft, + "source_file": _norm_source_file(node.get("source_file"), root) or "", + "source_location": node.get("source_location") or "", + "community": 0, + "community_name": "", + }) + + # --- collect edge rows (single table) --- + edge_rows: list[dict] = [] + + for edge in edges: + src_id = _normalize_id(edge.get("source") or edge.get("from", "")) + tgt_id = _normalize_id(edge.get("target") or edge.get("to", "")) + if not src_id or not tgt_id: + continue + if src_id not in node_types or tgt_id not in node_types: + continue + edge_rows.append({ + "from_id": src_id, + "to_id": tgt_id, + "relation": edge.get("relation", ""), + "confidence": edge.get("confidence", ""), + "confidence_score": float(edge.get("confidence_score", 0.0)), + "source_file": _norm_source_file(edge.get("source_file"), root) or "", + "weight": float(edge.get("weight", 1.0)), + }) + + # --- write CSV + COPY FROM in a temp dir --- + tmp_dir = tempfile.mkdtemp(prefix="graphify_bulk_") + try: + if node_rows: + csv_path = os.path.join(tmp_dir, "nodes.csv") + _write_csv(csv_path, node_rows, _NODE_COLUMNS) + _copy_node_csv(conn, csv_path) + + if edge_rows: + csv_path = os.path.join(tmp_dir, "edges.csv") + _write_csv(csv_path, edge_rows, _EDGE_COLUMNS) + _copy_rel_csv(conn, csv_path) + finally: + import shutil + shutil.rmtree(tmp_dir, ignore_errors=True) + + return node_types + + +def _incremental_ingest( + conn: object, + extraction: dict, + *, + prune_sources: list[str] | None = None, + root: str | None = None, + known_tables: set[str] | None = None, +) -> dict[str, str]: + """Incremental update via DELETE affected source_files + COPY FROM. + + Much faster than per-row MERGE: deletes nodes whose source_file appears + in the incoming extraction (or in prune_sources), then bulk-inserts the + new data via COPY FROM. Incoming cross-file edges (from unchanged files + into affected nodes) are saved before deletion and restored afterwards. + """ + nodes = extraction.get("nodes") or [] + edges = extraction.get("edges") or [] + + # --- collect affected source_files from the incoming data --- + affected_sfs: set[str] = set() + if prune_sources: + for sf in prune_sources: + sf_norm = _norm_source_file(sf, root) or sf + affected_sfs.add(sf_norm) + + node_types: dict[str, str] = {} + node_rows: list[dict] = [] + written_ids: set[str] = set() + + for node in nodes: + nid = _normalize_id(node.get("id", "")) + if not nid or nid in written_ids: + continue + written_ids.add(nid) + ft = _fix_file_type(node.get("file_type")) + node_types[nid] = ft + sf = _norm_source_file(node.get("source_file"), root) or "" + if sf: + affected_sfs.add(sf) + node_rows.append({ + "id": nid, + "label": node.get("label", ""), + "file_type": ft, + "source_file": sf, + "source_location": node.get("source_location") or "", + "community": 0, + "community_name": "", + }) + + # --- resolve types for non-delta edge endpoints (before DELETE) --- + unknown_ids: set[str] = set() + for edge in edges: + for key in ("source", "from", "target", "to"): + eid = _normalize_id(edge.get(key, "")) + if eid and eid not in node_types: + unknown_ids.add(eid) + for nid in unknown_ids: + try: + rows = list(conn.execute( + "MATCH (n:node {id: $nid}) RETURN n.file_type", + parameters={"nid": nid}, + )) + if rows: + node_types[nid] = rows[0][0] + except RuntimeError: + pass + + # --- save incoming cross-file edges before DELETE --- + affected_node_ids: set[str] = set() + for sf in affected_sfs: + try: + for row in conn.execute( + "MATCH (n:node) WHERE n.source_file = $sf RETURN n.id", + parameters={"sf": sf}, + ): + affected_node_ids.add(row[0]) + except RuntimeError: + pass + + saved_edge_rows: list[dict] = [] + + for sf in affected_sfs: + try: + rows = list(conn.execute( + "MATCH (a:node)-[e:edge]->(b:node) " + "WHERE b.source_file = $sf " + "RETURN a.id, b.id, e.relation, e.confidence, " + "e.confidence_score, e.source_file, e.weight", + parameters={"sf": sf}, + )) + except RuntimeError: + continue + + for row in rows: + if row[0] in affected_node_ids: + continue + saved_edge_rows.append({ + "from_id": row[0], "to_id": row[1], + "relation": row[2] or "", + "confidence": row[3] or "", + "confidence_score": float(row[4] or 0.0), + "source_file": row[5] or "", + "weight": float(row[6] or 1.0), + }) + + # --- DELETE nodes from affected source_files --- + for sf in affected_sfs: + conn.execute( + "MATCH (n:node) WHERE n.source_file = $sf DETACH DELETE n", + parameters={"sf": sf}, + ) + + # --- collect delta edge rows --- + edge_rows: list[dict] = [] + + for edge in edges: + src_id = _normalize_id(edge.get("source") or edge.get("from", "")) + tgt_id = _normalize_id(edge.get("target") or edge.get("to", "")) + if not src_id or not tgt_id: + continue + if src_id not in node_types or tgt_id not in node_types: + continue + edge_rows.append({ + "from_id": src_id, + "to_id": tgt_id, + "relation": edge.get("relation", ""), + "confidence": edge.get("confidence", ""), + "confidence_score": float(edge.get("confidence_score", 0.0)), + "source_file": _norm_source_file(edge.get("source_file"), root) or "", + "weight": float(edge.get("weight", 1.0)), + }) + + # --- merge saved incoming edges back --- + edge_rows.extend(saved_edge_rows) + + # --- COPY FROM bulk insert --- + tmp_dir = tempfile.mkdtemp(prefix="graphify_inc_") + try: + if node_rows: + csv_path = os.path.join(tmp_dir, "nodes.csv") + _write_csv(csv_path, node_rows, _NODE_COLUMNS) + _copy_node_csv(conn, csv_path) + + if edge_rows: + csv_path = os.path.join(tmp_dir, "edges.csv") + _write_csv(csv_path, edge_rows, _EDGE_COLUMNS) + _copy_rel_csv(conn, csv_path) + finally: + import shutil + shutil.rmtree(tmp_dir, ignore_errors=True) + + return node_types + + +def ingest_extraction( + conn: object, + extraction: dict, + *, + incremental: bool = False, + prune_sources: list[str] | None = None, + root: str | Path | None = None, + known_tables: set[str] | None = None, +) -> dict[str, str]: + """Write an extraction dict into NeuG. + + incremental=False: first build — uses COPY FROM bulk loading. + incremental=True: update — uses DELETE + COPY FROM. + + Returns node_types dict (id -> file_type) for use by ingest_communities. + """ + _root = str(Path(root).resolve()) if root else None + + if incremental: + return _incremental_ingest( + conn, extraction, + prune_sources=prune_sources, root=_root, + known_tables=known_tables, + ) + else: + return _bulk_ingest( + conn, extraction, + root=_root, known_tables=known_tables, + ) + + +def ingest_communities( + conn: object, + communities: dict[int, list[str]], + community_labels: dict[int, str] | None = None, + node_types: dict[str, str] | None = None, + node_label: str = "node", +) -> None: + """Write community assignments into NeuG node properties. + + Uses per-community ``SET`` with ``IN`` clauses instead of a single giant + ``CASE WHEN`` (which is O(N) parse time for large graphs). + NeuG does not support ``UNWIND $param`` or ``SET n.prop = $param``, so + community IDs and names are inlined. + + If community_labels is provided, community_name is also written in a + separate per-community pass (inline values, not parameters). + + Args: + node_label: Target node table label (default 'node'; use 'TempFile' + for file-level clustering on temp tables). + """ + # Bulk writeback via parameterized per-node SET. + # Uses neug's primary-key index on n.id for O(log N) lookup per node. + # CASE WHEN is O(N*M) — too slow for large graphs (28K nodes = 63s). + # Parameterized SET with index lookup: ~27K queries × O(log N) ≈ 2-3s. + # For symbol-level nodes (label='node'), IDs are normalized during graph + # building, so we must normalize again to match. For file-level TempFile + # nodes, IDs are raw file paths — _normalize_id would corrupt them. + normalize = node_label == "node" + comm_map: dict[str, int] = {} + for cid, node_ids in communities.items(): + cid_int = int(cid) + for nid in node_ids: + nid_key = _normalize_id(nid) if normalize else nid + if nid_key: + comm_map[nid_key] = cid_int + + for nid, cid in comm_map.items(): + conn.execute( + f"MATCH (n:{node_label} {{id: '{nid}'}}) SET n.community = {cid}" + ) + + # Write community_name per-community (inline values, not $param) + if community_labels: + for cid, name in community_labels.items(): + cid_int = int(cid) + safe_name = (name or "").replace("'", "\\'") + conn.execute( + f"MATCH (n:{node_label}) WHERE n.community = {cid_int} " + f"SET n.community_name = '{safe_name}'" + ) + + +def execute_cypher(conn: object, query: str) -> list[list]: + """Execute a Cypher query and return results as list of lists.""" + try: + return list(conn.execute(query)) + except RuntimeError as exc: + raise RuntimeError(f"Cypher query failed: {exc}") from exc + + +def close_db(db: object, conn: object) -> None: + """Close the NeuG connection and database.""" + conn.close() + db.close() + + +def export_to_json(conn: object, *, hyperedges: list | None = None) -> dict: + """Export the NeuG graph to a NetworkX ``node_link_data``-compatible dict. + + This avoids any dependency on NetworkX — the dict is assembled directly + from Cypher query results and can be consumed by ``json_graph.node_link_graph``. + """ + from graphify.export import _strip_diacritics, _git_head + + nodes: list[dict] = [] + for row in conn.execute( + "MATCH (n:node) RETURN n.id, n.label, n.file_type, " + "n.source_file, n.source_location, n.community, n.community_name" + ): + nodes.append({ + "id": row[0], + "label": row[1] or "", + "file_type": row[2] or "concept", + "source_file": row[3] or "", + "source_location": row[4] or "", + "community": row[5] if row[5] is not None else 0, + "community_name": row[6] or "", + "norm_label": _strip_diacritics(row[1] or "").lower(), + }) + + links: list[dict] = [] + for row in conn.execute( + "MATCH (a:node)-[e:edge]->(b:node) " + "RETURN a.id, b.id, e.relation, e.confidence, " + "e.confidence_score, e.source_file, e.weight" + ): + links.append({ + "source": row[0], + "target": row[1], + "relation": row[2] or "", + "confidence": row[3] or "EXTRACTED", + "confidence_score": float(row[4]) if row[4] is not None else 1.0, + "source_file": row[5] or "", + "weight": float(row[6]) if row[6] is not None else 1.0, + }) + + commit = _git_head() + data: dict = { + "directed": False, + "multigraph": False, + "graph": {}, + "nodes": nodes, + "links": links, + "hyperedges": hyperedges or [], + } + if commit: + data["built_at_commit"] = commit + return data + + +# --------------------------------------------------------------------------- +# Shared row-based filter functions (mirror of analyze.py NetworkX filters) +# --------------------------------------------------------------------------- +# Used by find_god_nodes and find_surprising_connections to filter Cypher +# query results without depending on NetworkX graph objects. + +from graphify.analyze import _BUILTIN_NOISE_LABELS, _JSON_NOISE_LABELS + + +def _is_file_node_row(label: str, source_file: str, degree: int) -> bool: + """File hub / method stub — row-based mirror of analyze._is_file_node.""" + if not label: + return False + # File-level hub: label matches source filename + if source_file and label == Path(source_file).name: + return True + # Method stub: .method_name() — ALWAYS exclude regardless of degree (analyze.py:74) + if label.startswith(".") and label.endswith("()"): + return True + # Function stub: func() with degree <= 1 (analyze.py:78) + if label.endswith("()") and degree <= 1: + return True + return False + + +def _is_concept_node_row(source_file: str) -> bool: + """Concept node — row-based mirror of analyze._is_concept_node.""" + if not source_file: + return True + if "." not in source_file.split("/")[-1]: + return True + return False + + +def _is_json_key_node_row(label: str, source_file: str) -> bool: + """JSON key noise — row-based mirror of analyze._is_json_key_node.""" + src = (source_file or "").lower() + if not src.endswith(".json"): + return False + return (label or "").strip().lower() in _JSON_NOISE_LABELS + + +# --------------------------------------------------------------------------- +# Community detection via neug GDS Leiden +# --------------------------------------------------------------------------- + + +_GDS_GRAPH_COUNTER = 0 + + +def _next_graph_name() -> str: + """Return a unique projected-graph name. + + neug has a bug where re-creating a dropped projected graph with the same + name makes it invisible to subsequent GDS calls on the same connection. + Using a unique name each time avoids this. + """ + global _GDS_GRAPH_COUNTER + _GDS_GRAPH_COUNTER += 1 + return f"g{_GDS_GRAPH_COUNTER}" + + +def run_leiden(conn: object, *, resolution: float = 1.0) -> dict[int, list[str]]: + """Run neug GDS Leiden community detection. + + Args: + resolution: Leiden resolution parameter (gamma). > 1 favours smaller + communities, < 1 favours larger communities. Default 1.0. + + Returns ``{community_id: [node_ids]}``. + neug leiden guarantees stable community IDs — no re-indexing needed. + """ + # Check for empty graph + node_rows = list(conn.execute("MATCH (n:node) RETURN n.id")) + if not node_rows: + return {} + + edge_rows = list(conn.execute("MATCH ()-[e:edge]->() RETURN count(*)")) + if edge_rows and edge_rows[0][0] == 0: + # No edges: each node is its own community + return {i: [row[0]] for i, row in enumerate(node_rows)} + + # Load GDS extension first (needed for project_graph and leiden) + try: + conn.execute("LOAD gds;") + except RuntimeError: + conn.execute("INSTALL gds;") + conn.execute("LOAD gds;") + + # Use a unique projected-graph name to avoid neug's stale-graph bug + # (re-creating a dropped graph with the same name makes it invisible + # to subsequent GDS calls on the same connection). + gname = _next_graph_name() + + # Project graph for GDS algorithms + conn.execute( + f"CALL project_graph('{gname}', ['node'], {{'[node, edge, node]': ''}})" + ) + + # Run Leiden + results = list(conn.execute( + f"CALL leiden('{gname}', {{concurrency: 1, resolution: {resolution}}}) " + "YIELD node, community " + "RETURN node.id, community" + )) + + # Clean up projected graph + try: + conn.execute(f"CALL drop_projected_graph('{gname}')") + except RuntimeError: + pass + + communities: dict[int, list[str]] = {} + for nid, cid in results: + communities.setdefault(int(cid), []).append(nid) + + return communities + + +# --------------------------------------------------------------------------- +# File-level clustering (aggregate symbol edges → file graph → leiden) +# --------------------------------------------------------------------------- + + +def _aggregate_file_edges(conn: object, csv_path: Path) -> set[str]: + """Aggregate symbol-level edges into file-level edge table, write to CSV. + + Returns the set of all source_file values (for creating temp file nodes). + Edges where either endpoint has empty source_file (concept/stub nodes) + or where both endpoints share the same source_file (intra-file edges) + are excluded. + """ + import csv + + results = list(conn.execute( + "MATCH (a:node)-[:edge]->(b:node) " + "WHERE a.source_file <> '' AND b.source_file <> '' " + "AND a.source_file <> b.source_file " + "RETURN a.source_file, b.source_file, count(*) AS weight" + )) + + all_files: set[str] = set() + with open(csv_path, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(["from_file", "to_file", "weight"]) + for from_file, to_file, weight in results: + writer.writerow([from_file, to_file, float(weight)]) + all_files.add(from_file) + all_files.add(to_file) + + return all_files + + +def run_leiden_subgraph( + conn: object, + *, + node_label: str, + edge_label: str, + resolution: float = 1.0, + weight: str | None = None, + initial_community_property: str | None = None, +) -> dict[str, int] | dict[str, tuple[int, int | None]]: + """Run leiden on an existing node/edge label pair (project + leiden + cleanup). + + Does NOT create or drop temp tables — caller is responsible for that. + + Args: + node_label: Existing node table name in DB (persistent or temporary). + edge_label: Existing edge table name in DB. + resolution: Leiden resolution parameter (gamma). + weight: If set (e.g. 'weight'), run weighted leiden. + initial_community_property: If set (e.g. 'delta_comm'), run freeze-assign + leiden. When set, returns ``{node_id: (new_cid, prev_cid)}`` instead + of ``{node_id: community_id}``. + """ + # Load GDS extension + try: + conn.execute("LOAD gds;") + except RuntimeError: + conn.execute("INSTALL gds;") + conn.execute("LOAD gds;") + + gname = _next_graph_name() + conn.execute( + f"CALL project_graph('{gname}', ['{node_label}'], " + f"{{'[{node_label}, {edge_label}, {node_label}]': ''}})" + ) + + # Build leiden options + opts = f"concurrency: 1, resolution: {resolution}" + if weight: + opts += f", weight: '{weight}'" + if initial_community_property: + opts += f", initial_community_property: '{initial_community_property}'" + + if initial_community_property: + # Freeze-assign: also return previous_community + results = list(conn.execute( + f"CALL leiden('{gname}', {{{opts}}}) " + "YIELD node, community, previous_community " + "RETURN node.id, community, previous_community" + )) + try: + conn.execute(f"CALL drop_projected_graph('{gname}')") + except RuntimeError: + pass + return {nid: (int(cid), prev) for nid, cid, prev in results} + else: + results = list(conn.execute( + f"CALL leiden('{gname}', {{{opts}}}) " + "YIELD node, community RETURN node.id, community" + )) + try: + conn.execute(f"CALL drop_projected_graph('{gname}')") + except RuntimeError: + pass + return {nid: int(cid) for nid, cid in results} + + +def _maybe_dump_temp_csvs(edge_csv: Path, node_csv: Path, tag: str) -> None: + """If GRAPHIFY_KEEP_TEMP is set, copy temp CSVs to that directory.""" + import os, shutil + dest = os.environ.get("GRAPHIFY_KEEP_TEMP") + if not dest: + return + dest_dir = Path(dest) + dest_dir.mkdir(parents=True, exist_ok=True) + shutil.copy2(edge_csv, dest_dir / f"{tag}_edges.csv") + shutil.copy2(node_csv, dest_dir / f"{tag}_nodes.csv") + + +def cluster_on_files( + conn: object, *, resolution: float = 1.0 +) -> dict[int, list[str]]: + """File-level clustering. Returns ``{community_id: [file_paths]}``. + + Creates temp tables (TempFile / TEMP_FILE_EDGE) and keeps them alive + for subsequent analysis. Caller is responsible for cleaning up: + ``DROP TABLE TEMP_FILE_EDGE; DROP TABLE TempFile;`` + """ + import tempfile, csv + + _NODE_LABEL = "TempFile" + _EDGE_LABEL = "TEMP_FILE_EDGE" + + # Defensive: clean up any leftover temp tables from previous calls + conn.execute(f"DROP TABLE IF EXISTS {_EDGE_LABEL}") + conn.execute(f"DROP TABLE IF EXISTS {_NODE_LABEL}") + + with tempfile.TemporaryDirectory() as tmpdir: + edge_csv = Path(tmpdir) / "file_edges.csv" + node_csv = Path(tmpdir) / "file_nodes.csv" + + # 1. Aggregate symbol edges → file-level edge CSV + all_files = _aggregate_file_edges(conn, edge_csv) + + # Guard: if no files with inter-file edges, return empty — COPY TEMP + # with an empty CSV does not register the table in neug's catalog. + if not all_files: + return {} + + # 2. Write file node CSV (id = file path, must match edge CSV from/to) + # Include community + community_name columns for ingest_communities writeback + with open(node_csv, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(["id", "label", "community", "community_name"]) + for sf in sorted(all_files): + writer.writerow([sf, Path(sf).name, 0, ""]) + + # Keep temp CSVs for debugging if GRAPHIFY_KEEP_TEMP is set + _maybe_dump_temp_csvs(edge_csv, node_csv, "file_cluster") + + # 3. COPY TEMP to create temp tables (independent step) + conn.execute( + f"COPY TEMP {_NODE_LABEL} FROM '{node_csv}' " + "(header=true, delim=',')" + ) + conn.execute( + f"COPY TEMP {_EDGE_LABEL} FROM '{edge_csv}' " + f"(header=true, delim=',', from='{_NODE_LABEL}', to='{_NODE_LABEL}')" + ) + + # 4. Run leiden on the temp subgraph + file_communities = run_leiden_subgraph( + conn, + node_label=_NODE_LABEL, + edge_label=_EDGE_LABEL, + resolution=resolution, + weight=None, + ) + + # 5. Write community to TempFile nodes (for analysis queries) + communities: dict[int, list[str]] = {} + for file_path, cid in file_communities.items(): + communities.setdefault(cid, []).append(file_path) + ingest_communities(conn, communities, node_label=_NODE_LABEL) + + # NOTE: temp tables NOT dropped here — caller cleans up after analysis + return communities + + +# --------------------------------------------------------------------------- +# Cohesion + community labeling +# --------------------------------------------------------------------------- + + +def compute_cohesion( + conn: object, communities: dict[int, list[str]], + *, node_label: str = "node", edge_label: str = "edge", +) -> dict[int, float]: + """Per-community cohesion: undirected internal edges / max possible. + + Uses a single edge scan with ``frozenset`` deduplication to convert + directed edges to undirected, aligning with NetworkX's ``Graph`` semantics. + """ + node_comm = {n: cid for cid, nodes in communities.items() for n in nodes} + intra: dict[int, set] = {cid: set() for cid in communities} + + for row in conn.execute( + f"MATCH (a:{node_label})-[:{edge_label}]->(b:{node_label}) " + f"RETURN a.id, b.id" + ): + a, b = row[0], row[1] + if a == b: # skip self-loops + continue + ca = node_comm.get(a) + if ca is not None and ca == node_comm.get(b): + intra[ca].add(frozenset((a, b))) # deduplicate directed edges + + result: dict[int, float] = {} + for cid, nodes in communities.items(): + n = len(nodes) + if n <= 1: + result[cid] = 1.0 + continue + possible = n * (n - 1) / 2 + result[cid] = len(intra[cid]) / possible if possible > 0 else 1.0 + + return result + + +def label_communities_by_hub( + conn: object, communities: dict[int, list[str]], + *, node_label: str = "node", edge_label: str = "edge", +) -> dict[int, str]: + """Name each community after its highest-degree member. + + Requires community IDs to be written to the db beforehand. + """ + try: + rows = list(conn.execute( + f"MATCH (n:{node_label}) " + f"WHERE n.community IS NOT NULL " + f"OPTIONAL MATCH (n)-[e:{edge_label}]-() " + f"WITH n, n.community AS cid, count(e) AS degree " + f"ORDER BY cid, degree DESC, n.id ASC " + f"WITH cid, collect(n)[0] AS hub " + f"RETURN cid, hub.label, hub.id" + )) + labels: dict[int, str] = {} + for cid, label, nid in rows: + name = (label or nid or "").strip() + if name and name.endswith("()"): + name = name[:-2] + labels[int(cid)] = name or f"Community {cid}" + except RuntimeError: + # Fallback: two-step approach + deg_rows = list(conn.execute( + f"MATCH (n:{node_label})-[e:{edge_label}]-() " + f"WITH n, count(e) AS degree " + f"RETURN n.id, n.community, n.label, degree " + f"ORDER BY n.community, degree DESC, n.id" + )) + labels = {} + seen: set[int] = set() + for nid, cid, label, degree in deg_rows: + if cid is not None and int(cid) not in seen: + seen.add(int(cid)) + name = (label or nid or "").strip() + if name and name.endswith("()"): + name = name[:-2] + labels[int(cid)] = name or f"Community {cid}" + + # Communities with no nodes in the query result + for cid in communities: + if int(cid) not in labels: + labels[int(cid)] = f"Community {cid}" + + return labels + + +# --------------------------------------------------------------------------- +# God nodes +# --------------------------------------------------------------------------- + + +def find_god_nodes(conn: object, top_n: int = 10) -> list[dict]: + """Top-N most-connected real entities. + + Cypher pre-filters noise labels; Python applies complex filters + (file hub, concept node, JSON key, method stub) via shared row-based functions. + """ + # Build inline noise list (neug doesn't support IN $param) + noise_list = "[" + ", ".join(f"'{l}'" for l in _BUILTIN_NOISE_LABELS) + "]" + + rows = list(conn.execute( + f"MATCH (n:node)-[e:edge]-() " + f"WITH n, count(e) AS degree " + f"WHERE degree > 0 " + f" AND NOT (n.label IN {noise_list}) " + f"RETURN n.id, n.label, n.file_type, n.source_file, degree " + f"ORDER BY degree DESC, n.id ASC " + f"LIMIT {top_n * 5}" + )) + + gods: list[dict] = [] + for nid, label, ft, source_file, degree in rows: + label = label or "" + source_file = source_file or "" + if not label: + continue + if _is_file_node_row(label, source_file, degree): + continue + if _is_concept_node_row(source_file): + continue + if _is_json_key_node_row(label, source_file): + continue + if label in _BUILTIN_NOISE_LABELS: # Cypher already filtered, this is a safety net + continue + gods.append({"id": nid, "label": label, "degree": degree}) + if len(gods) >= top_n: + break + + return gods + + +# --------------------------------------------------------------------------- +# Surprising connections +# --------------------------------------------------------------------------- + + +def _surprise_score_row( + relation: str, + conf: str, + u_source: str, + v_source: str, + cid_u: int | None, + cid_v: int | None, + deg_u: int, + deg_v: int, +) -> tuple[int, list[str]]: + """Score how surprising a cross-file edge is (row-based mirror of analyze._surprise_score).""" + from graphify.analyze import _file_category, _top_level_dir, _cross_language + + score = 0 + reasons: list[str] = [] + + # 1. Confidence weight + conf_bonus = {"AMBIGUOUS": 3, "INFERRED": 2, "EXTRACTED": 1}.get(conf, 1) + + cat_u = _file_category(u_source) + cat_v = _file_category(v_source) + + # 2. Suppress structural bonuses for INFERRED calls/uses that cross language + # boundaries or connect code to doc (resolver pollution) + _suppress_structural = ( + conf == "INFERRED" + and relation in ("calls", "uses") + and (_cross_language(u_source, v_source) or {cat_u, cat_v} == {"code", "doc"}) + ) + if _suppress_structural: + conf_bonus = 0 + + score += conf_bonus + if conf in ("AMBIGUOUS", "INFERRED"): + reasons.append(f"{conf.lower()} connection - not explicitly stated in source") + + # 3. Cross file-type bonus + if cat_u != cat_v and not _suppress_structural: + score += 2 + reasons.append(f"crosses file types ({cat_u} ↔ {cat_v})") + + # 4. Cross-repo bonus + if _top_level_dir(u_source) != _top_level_dir(v_source) and not _suppress_structural: + score += 2 + reasons.append("connects across different repos/directories") + + # 5. Cross-community bonus + if (cid_u is not None and cid_v is not None and cid_u != cid_v + and not _suppress_structural): + score += 1 + reasons.append("bridges separate communities") + + # 6. Semantic similarity bonus + if relation == "semantically_similar_to": + score = int(score * 1.5) + reasons.append("semantically similar concepts with no structural link") + + # 7. Peripheral → hub + if min(deg_u, deg_v) <= 2 and max(deg_u, deg_v) >= 5: + score += 1 + reasons.append("peripheral node unexpectedly reaches hub") + + return score, reasons + + +def find_surprising_connections( + conn: object, + communities: dict[int, list[str]], + top_n: int = 5, +) -> list[dict]: + """Cross-file or cross-community edges ranked by composite surprise score.""" + # 0. Guard: skip tiny/edge-less graphs. The undirected degree query below + # crashes neug on graphs without edges, and surprising connections are + # meaningless without edges anyway. + edge_count = list(conn.execute( + "MATCH ()-[e:edge]->() RETURN count(e)" + )) + if not edge_count or edge_count[0][0] == 0: + return [] + if sum(len(v) for v in communities.values()) < 4: + return [] + + # 1. Determine multi-source vs single-source + source_count = list(conn.execute( + "MATCH (n:node) WHERE n.source_file <> '' " + "RETURN count(DISTINCT n.source_file) AS cnt" + )) + is_multi_source = source_count[0][0] > 1 if source_count else False + + # 2. Pre-compute degrees + deg_rows = list(conn.execute( + "MATCH (n:node)-[e:edge]-() " + "WITH n, count(e) AS degree " + "RETURN n.id, degree" + )) + degrees = {r[0]: r[1] for r in deg_rows} + + # 3. Get candidate edges + structural_list = "['imports', 'imports_from', 'contains', 'method']" + + if is_multi_source: + rows = list(conn.execute( + f"MATCH (a:node)-[e:edge]->(b:node) " + f"WHERE a.source_file <> '' AND b.source_file <> '' " + f" AND a.source_file <> b.source_file " + f" AND NOT (e.relation IN {structural_list}) " + f"RETURN a.id, a.label, a.source_file, a.community, " + f" b.id, b.label, b.source_file, b.community, " + f" e.relation, e.confidence" + )) + else: + rows = list(conn.execute( + f"MATCH (a:node)-[e:edge]->(b:node) " + f"WHERE a.community IS NOT NULL AND b.community IS NOT NULL " + f" AND a.community <> b.community " + f" AND NOT (e.relation IN {structural_list}) " + f"RETURN a.id, a.label, a.source_file, a.community, " + f" b.id, b.label, b.source_file, b.community, " + f" e.relation, e.confidence" + )) + + # 4. Python filtering + scoring + node_comm = {n: cid for cid, nodes in communities.items() for n in nodes} + candidates: list[dict] = [] + + for (a_id, a_label, a_src, a_comm, b_id, b_label, b_src, b_comm, + relation, conf) in rows: + a_label = a_label or "" + a_src = a_src or "" + b_label = b_label or "" + b_src = b_src or "" + relation = relation or "" + conf = conf or "EXTRACTED" + + deg_a = degrees.get(a_id, 0) + deg_b = degrees.get(b_id, 0) + + # Filter concept/file-hub nodes + if _is_concept_node_row(a_src) or _is_concept_node_row(b_src): + continue + if _is_file_node_row(a_label, a_src, deg_a) or _is_file_node_row(b_label, b_src, deg_b): + continue + + cid_u = a_comm if a_comm is not None else node_comm.get(a_id) + cid_v = b_comm if b_comm is not None else node_comm.get(b_id) + + score, reasons = _surprise_score_row( + relation, conf, a_src, b_src, cid_u, cid_v, deg_a, deg_b + ) + + candidates.append({ + "_score": score, + "source": a_label, + "target": b_label, + "source_files": [a_src, b_src], + "confidence": conf, + "relation": relation, + "why": "; ".join(reasons) if reasons else "cross-file semantic connection", + "_pair": tuple(sorted([cid_u or 0, cid_v or 0])) if not is_multi_source else None, + }) + + # Sort by score descending + candidates.sort(key=lambda x: x["_score"], reverse=True) + + # Single-source: deduplicate by community pair + if not is_multi_source: + seen_pairs: set[tuple] = set() + deduped: list[dict] = [] + for c in candidates: + pair = c.pop("_pair", None) + if pair not in seen_pairs: + seen_pairs.add(pair) + deduped.append(c) + candidates = deduped + else: + for c in candidates: + c.pop("_pair", None) + + # Strip _score from output + for c in candidates: + c.pop("_score", None) + + return candidates[:top_n] + + +# --------------------------------------------------------------------------- +# File-level analysis helpers (operate on TempFile / TEMP_FILE_EDGE) +# --------------------------------------------------------------------------- + + +def _find_god_files(conn: object, top_n: int = 10) -> list[dict]: + """File-level god nodes: top-N files by edge degree.""" + rows = list(conn.execute( + "MATCH (n:TempFile)-[e:TEMP_FILE_EDGE]-() " + "WITH n, count(e) AS degree " + "WHERE degree > 0 " + "RETURN n.id, n.label, degree " + "ORDER BY degree DESC, n.id ASC " + f"LIMIT {top_n}" + )) + return [{"id": nid, "label": label, "degree": deg} for nid, label, deg in rows] + + +def _find_surprising_file_connections( + conn: object, + communities: dict[int, list[str]], + top_n: int = 5, +) -> list[dict]: + """File-level surprising connections: cross-community edges ranked by weight.""" + node_comm = {n: cid for cid, nodes in communities.items() for n in nodes} + + rows = list(conn.execute( + "MATCH (a:TempFile)-[e:TEMP_FILE_EDGE]->(b:TempFile) " + "WHERE a.community IS NOT NULL AND b.community IS NOT NULL " + " AND a.community <> b.community " + "RETURN a.id, a.label, a.community, b.id, b.label, b.community, e.weight" + )) + + candidates: list[dict] = [] + for a_id, a_label, a_comm, b_id, b_label, b_comm, weight in rows: + candidates.append({ + "source": a_label or a_id, + "target": b_label or b_id, + "source_files": [a_id, b_id], + "weight": int(weight) if weight else 1, + "why": f"cross-community file edge (weight={int(weight) if weight else 1})", + }) + + candidates.sort(key=lambda x: x["weight"], reverse=True) + return candidates[:top_n] + + +# --------------------------------------------------------------------------- +# Orchestration: neug clustered path +# --------------------------------------------------------------------------- + + +def cluster_by_neug( + conn: object, + *, + merged: dict, + graph_json_path: object, + analysis_path: object, + stages: object, + export_fn: object, + hyperedges: list | None = None, + resolution: float = 1.0, + file_level: bool = False, +) -> None: + """Orchestrate the neug clustered path. + + Steps: leiden → writeback → label → analysis → export. + cli.py should call this directly instead of inlining the workflow. + Returns the exported graph data dict (for summary printing). + """ + import json + + _FILE_NODE = "TempFile" + _FILE_EDGE = "TEMP_FILE_EDGE" + + # 1. Leiden community detection (symbol-level or file-level) + if file_level: + communities = cluster_on_files(conn, resolution=resolution) + # communities = {cid: [file_paths]}, temp tables still alive + else: + communities = run_leiden(conn, resolution=resolution) + stages.mark("cluster") + + if file_level and communities: + # 2a. Label + analysis on temp tables (file-level graph) + labels = label_communities_by_hub( + conn, communities, node_label=_FILE_NODE, edge_label=_FILE_EDGE + ) + ingest_communities( + conn, communities, community_labels=labels, node_label=_FILE_NODE + ) + cohesion = compute_cohesion( + conn, communities, node_label=_FILE_NODE, edge_label=_FILE_EDGE + ) + gods = _find_god_files(conn) + surprises = _find_surprising_file_connections(conn, communities) + stages.mark("analyze") + + # 2b. Write community to symbol-level :node by source_file + # (so graph.json export includes community info per symbol) + for cid, file_paths in communities.items(): + if not file_paths: + continue + files_inline = ", ".join(f"'{f}'" for f in file_paths) + safe_name = (labels.get(cid, f"Community {cid}") or "").replace("'", "\\'") + conn.execute( + f"MATCH (n:node) WHERE n.source_file IN [{files_inline}] " + f"SET n.community = {cid}, n.community_name = '{safe_name}'" + ) + stages.mark("writeback") + elif file_level and not communities: + # No inter-file edges — skip file-level analysis (temp tables don't exist) + labels = {} + cohesion = {} + gods = [] + surprises = [] + stages.mark("analyze") + stages.mark("writeback") + else: + # 2. Batch writeback community IDs + ingest_communities(conn, communities) + + # 3. Label communities by hub + labels = label_communities_by_hub(conn, communities) + + # 4. Write community_name + ingest_communities(conn, communities, community_labels=labels) + + # 5. Analysis + cohesion = compute_cohesion(conn, communities) + gods = find_god_nodes(conn) + surprises = find_surprising_connections(conn, communities) + stages.mark("analyze") + + # 6. Export graph.json (with community + community_name) + data = export_fn(conn, hyperedges=hyperedges or []) + graph_json_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + stages.mark("export") + + # 7. Write .graphify_analysis.json (sorted by community ID) + analysis = { + "communities": {str(k): v for k, v in sorted(communities.items())}, + "cohesion": {str(k): v for k, v in sorted(cohesion.items())}, + "gods": gods, + "surprises": surprises, + "tokens": { + "input": merged["input_tokens"], + "output": merged["output_tokens"], + }, + } + analysis_path.write_text(json.dumps(analysis, indent=2), encoding="utf-8") + + # 8. Clean up temp tables (file-level only) + if file_level: + conn.execute(f"DROP TABLE IF EXISTS {_FILE_EDGE}") + conn.execute(f"DROP TABLE IF EXISTS {_FILE_NODE}") + + return data + + +# --------------------------------------------------------------------------- +# Incremental delta analysis (freeze-assign leiden) +# --------------------------------------------------------------------------- + + +def run_leiden_freeze_assign( + conn: object, + old_communities: dict[str, int], + *, + resolution: float = 1.0, +) -> list[tuple[str, int, int | None]]: + """Run freeze-assign leiden. Old nodes frozen, new nodes assigned. + + Args: + old_communities: {node_id: old_community_id} from .graphify_analysis.json. + resolution: Leiden resolution parameter (gamma). > 1 favours smaller + communities, < 1 favours larger communities. Default 1.0. + + Returns [(node_id, new_community, previous_community), ...]. + previous_community is None for new nodes (delta_comm was -1). + """ + # Add delta_comm property if not exists + try: + conn.execute("ALTER TABLE node ADD delta_comm INT64 DEFAULT -1") + except RuntimeError: + pass # Column already exists + + # Optimisation: most nodes already have the correct community in the DB's + # ``community`` column (written by the full extract). Instead of a giant + # CASE WHEN with N clauses (O(N) parse time), we: + # 1. Copy ``community`` → ``delta_comm`` for ALL nodes (one fast query) + # 2. Set ``delta_comm = -1`` for new nodes only (small IN clause) + # 3. Fix re-extracted nodes whose DB community is 0 but old_communities + # says different (per-community SET, usually a handful of queries) + conn.execute("MATCH (n:node) SET n.delta_comm = n.community") + + # Identify new nodes: in DB but not in old_communities + db_node_ids = {row[0] for row in conn.execute("MATCH (n:node) RETURN n.id")} + new_node_ids = db_node_ids - set(old_communities.keys()) + if new_node_ids: + # Batch in chunks of 500 to avoid query-length limits + new_list = sorted(new_node_ids) + for i in range(0, len(new_list), 500): + chunk = new_list[i:i + 500] + id_list = ", ".join(f"'{nid}'" for nid in chunk) + conn.execute( + f"MATCH (n:node) WHERE n.id IN [{id_list}] " + f"SET n.delta_comm = -1" + ) + + # Fix re-extracted nodes: in old_communities but DB community doesn't match + # These are nodes whose file was re-extracted (deleted + re-created with community=0) + re_extracted: dict[int, list[str]] = {} # {old_cid: [node_ids]} + for row in conn.execute( + "MATCH (n:node) WHERE n.community = 0 RETURN n.id" + ): + nid = row[0] + if nid in old_communities and old_communities[nid] != 0: + old_cid = old_communities[nid] + re_extracted.setdefault(old_cid, []).append(nid) + + for old_cid, node_ids in re_extracted.items(): + # Batch in chunks of 500 + for i in range(0, len(node_ids), 500): + chunk = node_ids[i:i + 500] + id_list = ", ".join(f"'{nid}'" for nid in chunk) + conn.execute( + f"MATCH (n:node) WHERE n.id IN [{id_list}] " + f"SET n.delta_comm = {old_cid}" + ) + + # Load GDS extension + try: + conn.execute("LOAD gds;") + except RuntimeError: + conn.execute("INSTALL gds;") + conn.execute("LOAD gds;") + + # Use a unique projected-graph name (see run_leiden for rationale) + gname = _next_graph_name() + + # Project graph (picks up delta_comm property) + conn.execute( + f"CALL project_graph('{gname}', ['node'], {{'[node, edge, node]': ''}})" + ) + + # Run freeze-assign leiden (allow_relocation defaults to false = frozen) + results = list(conn.execute( + f"CALL leiden('{gname}', {{concurrency: 1, resolution: {resolution}, " + "initial_community_property: 'delta_comm'}) " + "YIELD node, community, previous_community " + "RETURN node.id, community, previous_community" + )) + + # Clean up projected graph + try: + conn.execute(f"CALL drop_projected_graph('{gname}')") + except RuntimeError: + pass + + # Drop delta_comm column to avoid schema mismatch on subsequent extract + try: + conn.execute("ALTER TABLE node DROP delta_comm") + except RuntimeError: + pass + + return [(nid, int(cid), prev) for nid, cid, prev in results] + + +def _merge_changed_fragments( + conn: object, + leiden_results: list[tuple[str, int, int | None]], + old_communities: dict[str, int], + *, + min_size: int = 5, +) -> list[tuple[str, int, int | None]]: + """Merge small changed/new communities into their strongest neighbour. + + Only touches communities that are **new** or **changed** (not stable). + Stable communities may *receive* merged members but are never split. + + Args: + conn: neug connection (uses node/edge tables). + leiden_results: [(node_id, new_cid, prev_cid), ...] from freeze-assign. + old_communities: {node_id: old_cid} from baseline. + min_size: communities smaller than this get merged. + + Returns updated leiden_results with merged community assignments. + """ + # Build new_communities and node→prev mapping + new_communities: dict[int, list[str]] = {} + node_prev: dict[str, int | None] = {} + for nid, new_cid, prev_cid in leiden_results: + new_communities.setdefault(new_cid, []).append(nid) + node_prev[nid] = prev_cid + + # Build old community membership for comparison + old_comm_to_nodes: dict[int, set[str]] = {} + for nid, cid in old_communities.items(): + old_comm_to_nodes.setdefault(cid, set()).add(nid) + + # Identify changed/new CIDs + changed_cids: set[int] = set() + for cid, members in new_communities.items(): + if cid not in old_comm_to_nodes: + changed_cids.add(cid) # new + elif set(members) != old_comm_to_nodes[cid]: + changed_cids.add(cid) # changed + + if not changed_cids: + return leiden_results + + # Load edge weights from the database + edges: dict[tuple[str, str], float] = {} + try: + for row in conn.execute( + "MATCH (a:node)-[e:edge]->(b:node) " + "RETURN a.id, b.id, e.weight" + ): + w = row[2] if row[2] is not None else 1.0 + edges[(row[0], row[1])] = float(w) + except RuntimeError: + pass + + node_comm: dict[str, int] = { + n: cid for cid, nodes in new_communities.items() for n in nodes + } + communities = {k: list(v) for k, v in new_communities.items()} + + merged_any = True + while merged_any: + merged_any = False + small_cids = [ + cid for cid in communities + if cid in changed_cids and len(communities[cid]) < min_size + ] + for small_cid in small_cids: + if small_cid not in communities: + continue + members = communities[small_cid] + if len(members) >= min_size: + continue + + # Find strongest neighbour community by edge weight + comm_connections: dict[int, float] = {} + for node in members: + for (a, b), w in edges.items(): + if a == node: + other = node_comm.get(b, -1) + elif b == node: + other = node_comm.get(a, -1) + else: + continue + if other != small_cid and other >= 0: + comm_connections[other] = ( + comm_connections.get(other, 0.0) + w + ) + + if not comm_connections: + continue + + best_cid = max(comm_connections, key=lambda k: comm_connections[k]) + if best_cid not in communities: + continue + + # Merge small_cid → best_cid + communities[best_cid].extend(members) + del communities[small_cid] + for n in members: + node_comm[n] = best_cid + # If target was stable, mark it as changed now + changed_cids.add(best_cid) + merged_any = True + + # Rebuild leiden_results from merged communities + result: list[tuple[str, int, int | None]] = [] + for cid, members in communities.items(): + for nid in members: + result.append((nid, cid, node_prev.get(nid))) + return result + + +def analyze_community_changes( + leiden_results: list[tuple[str, int, int | None]], + old_communities: dict[str, int], +) -> dict: + """Classify communities into 4 orthogonal change types. + + Classification matrix: + Existed before? | Still exists? | Classification + Yes | Yes, no change | stable + Yes | Yes, changed | changed + Yes | No | dissolved + No | Yes | new + """ + # Build new_communities from leiden results + new_communities: dict[int, list[str]] = {} + prev_map: dict[str, int | None] = {} + for nid, new_cid, prev_cid in leiden_results: + new_communities.setdefault(new_cid, []).append(nid) + prev_map[nid] = prev_cid + + # Build old_comm_to_nodes from old_communities + old_comm_to_nodes: dict[int, list[str]] = {} + for nid, cid in old_communities.items(): + old_comm_to_nodes.setdefault(cid, []).append(nid) + + changed_communities: dict[str, dict] = {} + new_communities_out: dict[str, dict] = {} + stable_communities: list[str] = [] + dissolved_communities: list[dict] = [] + + # Classify communities in leiden results + for cid, current_members in new_communities.items(): + if cid not in old_comm_to_nodes: + # New community + new_communities_out[str(cid)] = { + "members": sorted(current_members), + } + continue + + # Existing community — compute grow/shrink + old_members_set = set(old_comm_to_nodes[cid]) + current_set = set(current_members) + + grow_members = sorted(current_set - old_members_set) + shrink_members = sorted(old_members_set - current_set) + + if not grow_members and not shrink_members: + stable_communities.append(str(cid)) + else: + changed_communities[str(cid)] = { + "grow_members": grow_members, + "shrink_members": shrink_members, + } + + # Find dissolved communities (old but not in new results) + for old_cid, old_members in old_comm_to_nodes.items(): + if old_cid not in new_communities: + dissolved_communities.append({ + "cid": old_cid, + "old_size": len(old_members), + }) + + # Build summary + summary = { + "total_before": len(old_comm_to_nodes), + "total_after": len(new_communities), + "stable": len(stable_communities), + "changed": len(changed_communities), + "new": len(new_communities_out), + "dissolved": len(dissolved_communities), + } + + return { + "changed_communities": changed_communities, + "new_communities": new_communities_out, + "stable_communities": stable_communities, + "dissolved_communities": dissolved_communities, + "summary": summary, + } + + +def _delta_analyze_file_level( + conn: object, + old_communities: dict[str, int], + *, + resolution: float = 1.0, +) -> list[tuple[str, int, int | None]]: + """File-level freeze-assign leiden for incremental analysis. + + 1. Map old file-level communities from prev_analysis (file paths → old_cid) + 2. Aggregate edges → CSV (same as full file-level clustering) + 3. Write file node CSV with delta_comm column (old_cid or -1 for new files) + 4. COPY TEMP → run_leiden_subgraph with freeze-assign + 5. Write community to TempFile (for analysis queries) + + Temp tables are NOT dropped — caller cleans up after analysis. + + Returns [(file_path, new_cid, prev_cid), ...]. + """ + import tempfile, csv + + _NODE_LABEL = "TempFile" + _EDGE_LABEL = "TEMP_FILE_EDGE" + + # Defensive: clean up any leftover temp tables from previous calls + conn.execute(f"DROP TABLE IF EXISTS {_EDGE_LABEL}") + conn.execute(f"DROP TABLE IF EXISTS {_NODE_LABEL}") + + # 1. old_communities is already {file_path: old_cid} in file-level mode + old_file_communities: dict[str, int] = old_communities + + with tempfile.TemporaryDirectory() as tmpdir: + edge_csv = Path(tmpdir) / "file_edges.csv" + node_csv = Path(tmpdir) / "file_nodes.csv" + + # 2. Aggregate edges → CSV + all_files = _aggregate_file_edges(conn, edge_csv) + + # Guard: if no files with inter-file edges, return empty — COPY TEMP + # with an empty CSV does not register the table in neug's catalog. + if not all_files: + return [] + + # 3. Write file node CSV with delta_comm + community columns + # id = file path, delta_comm = old community (or -1 for new files) + # community = 0 (placeholder, overwritten by ingest_communities) + with open(node_csv, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(["id", "label", "delta_comm", "community", "community_name"]) + for sf in sorted(all_files): + old_cid = old_file_communities.get(sf, -1) + writer.writerow([sf, Path(sf).name, old_cid, 0, ""]) + + # Keep temp CSVs for debugging if GRAPHIFY_KEEP_TEMP is set + _maybe_dump_temp_csvs(edge_csv, node_csv, "file_delta") + + # 4. COPY TEMP + conn.execute( + f"COPY TEMP {_NODE_LABEL} FROM '{node_csv}' " + "(header=true, delim=',')" + ) + conn.execute( + f"COPY TEMP {_EDGE_LABEL} FROM '{edge_csv}' " + f"(header=true, delim=',', from='{_NODE_LABEL}', to='{_NODE_LABEL}')" + ) + + # 5. Run freeze-assign leiden on temp subgraph + file_results = run_leiden_subgraph( + conn, + node_label=_NODE_LABEL, + edge_label=_EDGE_LABEL, + resolution=resolution, + weight=None, + initial_community_property="delta_comm", + ) + # file_results: {file_path: (new_cid, prev_cid)} + + # 6. Write community to TempFile nodes (for analysis queries) + new_communities: dict[int, list[str]] = {} + for file_path, (new_cid, _) in file_results.items(): + new_communities.setdefault(new_cid, []).append(file_path) + ingest_communities(conn, new_communities, node_label=_NODE_LABEL) + + # 7. Return file-level results (NOT expanded to symbol level) + return [ + (file_path, new_cid, prev_cid) + for file_path, (new_cid, prev_cid) in file_results.items() + ] + + +def delta_analyze( + conn: object, + *, + prev_analysis: dict, + delta_analysis_path: object, + stages: object, + merged: dict, + resolution: float = 1.0, + file_level: bool = False, +) -> dict: + """Orchestrate incremental delta analysis (preview mode). + + Steps: freeze-assign leiden → community change analysis → + partial cohesion → full gods/surprises → write delta. + Does NOT writeback to DB's community property (preview only). + """ + import json + + # 1. Build old_communities from prev_analysis + old_communities: dict[str, int] = {} + for cid_str, node_ids in prev_analysis.get("communities", {}).items(): + cid = int(cid_str) + for nid in node_ids: + old_communities[nid] = cid + + # 2. Run freeze-assign leiden (symbol-level or file-level) + if file_level: + leiden_results = _delta_analyze_file_level( + conn, old_communities, resolution=resolution + ) + else: + leiden_results = run_leiden_freeze_assign(conn, old_communities, resolution=resolution) + stages.mark("freeze-assign") + + # 2b. Merge small changed/new communities (only fragments, no split) + leiden_results = _merge_changed_fragments( + conn, leiden_results, old_communities, min_size=5, + ) + stages.mark("merge-fragments") + + # 3. Analyze community changes + changes = analyze_community_changes(leiden_results, old_communities) + stages.mark("analyze-changes") + + # 4. Build new_communities dict for cohesion + surprising_connections + new_communities: dict[int, list[str]] = {} + for nid, new_cid, _ in leiden_results: + new_communities.setdefault(new_cid, []).append(nid) + + # 5. Compute cohesion only for changed + new communities + changed_cids = set() + for cid_str in changes["changed_communities"]: + changed_cids.add(int(cid_str)) + for cid_str in changes["new_communities"]: + changed_cids.add(int(cid_str)) + + _FILE_NODE = "TempFile" + _FILE_EDGE = "TEMP_FILE_EDGE" + + if file_level and new_communities: + # File-level analysis on temp tables + all_cohesion = compute_cohesion( + conn, new_communities, node_label=_FILE_NODE, edge_label=_FILE_EDGE + ) + delta_cohesion = {cid: score for cid, score in all_cohesion.items() if cid in changed_cids} + + labels = label_communities_by_hub( + conn, new_communities, node_label=_FILE_NODE, edge_label=_FILE_EDGE + ) + gods = _find_god_files(conn) + surprises = _find_surprising_file_connections(conn, new_communities) + elif file_level and not new_communities: + # No inter-file edges — temp tables don't exist, skip file-level analysis + delta_cohesion = {} + labels = {} + gods = [] + surprises = [] + else: + # Symbol-level analysis (existing path) + all_cohesion = compute_cohesion(conn, new_communities) + delta_cohesion = {cid: score for cid, score in all_cohesion.items() if cid in changed_cids} + + # 6. Label communities (reuse label_communities_by_hub) + labels: dict[int, str] = {} + try: + node_degree: dict[str, int] = {} + for row in conn.execute( + "MATCH (n:node)-[e:edge]-() " + "WITH n, count(e) AS degree " + "RETURN n.id, degree" + ): + node_degree[row[0]] = row[1] + except RuntimeError: + node_degree = {} + + node_label: dict[str, str] = {} + try: + for row in conn.execute("MATCH (n:node) RETURN n.id, n.label"): + node_label[row[0]] = row[1] or row[0] + except RuntimeError: + pass + + for cid, members in new_communities.items(): + if cid not in changed_cids: + continue + best_nid = members[0] + best_deg = -1 + for nid in members: + deg = node_degree.get(nid, 0) + if deg > best_deg: + best_deg = deg + best_nid = nid + name = (node_label.get(best_nid, best_nid) or best_nid).strip() + if name.endswith("()"): + name = name[:-2] + labels[cid] = name or f"Community {cid}" + + gods = find_god_nodes(conn) + surprises = find_surprising_connections(conn, new_communities) + stages.mark("analyze-full") + + # 8. Build delta JSON + # Add cohesion + community_name to changed/new communities + for cid_str, info in changes["changed_communities"].items(): + cid = int(cid_str) + info["cohesion"] = delta_cohesion.get(cid, 0.0) + info["community_name"] = labels.get(cid, f"Community {cid}") + for cid_str, info in changes["new_communities"].items(): + cid = int(cid_str) + info["cohesion"] = delta_cohesion.get(cid, 0.0) + info["community_name"] = labels.get(cid, f"Community {cid}") + + # Sort all community sections by community ID (ascending) + sorted_changed = dict(sorted(changes["changed_communities"].items(), key=lambda x: int(x[0]))) + sorted_new = dict(sorted(changes["new_communities"].items(), key=lambda x: int(x[0]))) + sorted_stable = sorted(changes["stable_communities"], key=lambda x: int(x)) + sorted_dissolved = sorted(changes["dissolved_communities"], key=lambda x: x["cid"]) + + delta = { + "changed_communities": sorted_changed, + "new_communities": sorted_new, + "stable_communities": sorted_stable, + "dissolved_communities": sorted_dissolved, + "summary": changes["summary"], + "gods": gods, + "surprises": surprises, + "tokens": { + "input": merged.get("input_tokens", 0), + "output": merged.get("output_tokens", 0), + }, + } + + delta_analysis_path.write_text(json.dumps(delta, indent=2), encoding="utf-8") + + # Clean up temp tables (file-level only) + if file_level: + conn.execute(f"DROP TABLE IF EXISTS {_FILE_EDGE}") + conn.execute(f"DROP TABLE IF EXISTS {_FILE_NODE}") + + return delta diff --git a/pyproject.toml b/pyproject.toml index d6be867709..c68ada3e13 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,6 +75,7 @@ gemini = ["openai", "tiktoken"] openai = ["openai", "tiktoken"] chinese = ["jieba"] sql = ["tree-sitter-sql"] +neug = ["neug>=0.1.3"] # extract_pascal() uses tree-sitter-pascal for AST-quality extraction (more # accurate calls/inherits edges) and falls back to a regex extractor when it is # absent (#781), so this stays optional. Unlike tree-sitter-dm below, it ships @@ -91,7 +92,8 @@ ocaml = ["tree-sitter-ocaml"] # tree-sitter-commonlisp ships prebuilt abi3 wheels for every platform; optional # because Common Lisp is a niche corpus language. commonlisp = ["tree-sitter-commonlisp"] -all = ["mcp>=1,<3", "starlette>=1.3.1,<2", "neo4j", "falkordb", "pypdf>=6.12.0", "markdownify", "watchdog", "graspologic; python_version < '3.13'", "python-docx", "openpyxl", "faster-whisper; python_version >= '3.11'", "yt-dlp>=2026.6.9", "matplotlib", "numpy>=2.0; python_version >= '3.13'", "openai", "tiktoken", "boto3", "anthropic", "tree-sitter-sql", "jieba", "tree-sitter-dm", "tree-sitter-hcl", "tree-sitter-pascal", "tree-sitter-ocaml", "tree-sitter-commonlisp"] +neug = ["neug>=0.1.3"] +all = ["mcp>=1,<3", "starlette>=1.3.1,<2", "neo4j", "falkordb", "pypdf>=6.12.0", "markdownify", "watchdog", "graspologic; python_version < '3.13'", "python-docx", "openpyxl", "faster-whisper; python_version >= '3.11'", "yt-dlp>=2026.6.9", "matplotlib", "numpy>=2.0; python_version >= '3.13'", "openai", "tiktoken", "boto3", "anthropic", "tree-sitter-sql", "jieba", "tree-sitter-dm", "tree-sitter-hcl", "tree-sitter-pascal", "tree-sitter-ocaml", "tree-sitter-commonlisp", "neug>=0.1.3"] [project.scripts] graphify = "graphify.__main__:main" diff --git a/tests/test_cypher_cli.py b/tests/test_cypher_cli.py new file mode 100644 index 0000000000..4c77b953fc --- /dev/null +++ b/tests/test_cypher_cli.py @@ -0,0 +1,59 @@ +"""Tests for the `graphify cypher` CLI command.""" +import json +import subprocess +import sys +import tempfile +from pathlib import Path + +import pytest + +try: + import neug + _has_neug = True +except ImportError: + _has_neug = False + +pytestmark = pytest.mark.skipif(not _has_neug, reason="neug not installed") + +FIXTURES = Path(__file__).parent / "fixtures" +EXTRACTION_JSON = FIXTURES / "extraction.json" + + +def _build_db(tmp_path) -> str: + from graphify.storage import init_db, ensure_schema, ingest_extraction, close_db + db_path = str(tmp_path / "graph.db") + ext = json.loads(EXTRACTION_JSON.read_text()) + db, conn = init_db(db_path) + known = ensure_schema(conn) + ingest_extraction(conn, ext, incremental=False, known_tables=known) + close_db(db, conn) + return db_path + + +def test_cypher_command_basic(tmp_path): + db_path = _build_db(tmp_path) + result = subprocess.run( + [sys.executable, "-m", "graphify", "cypher", + "MATCH (n) RETURN count(n)", "--db", db_path], + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0 + assert "4" in result.stdout + + +def test_cypher_command_db_not_found(tmp_path): + result = subprocess.run( + [sys.executable, "-m", "graphify", "cypher", + "MATCH (n) RETURN n", "--db", str(tmp_path / "nonexistent.db")], + capture_output=True, text=True, timeout=30, + ) + assert result.returncode != 0 + assert "not found" in result.stderr.lower() or "error" in result.stderr.lower() + + +def test_cypher_command_no_query(): + result = subprocess.run( + [sys.executable, "-m", "graphify", "cypher"], + capture_output=True, text=True, timeout=30, + ) + assert result.returncode != 0 diff --git a/tests/test_storage.py b/tests/test_storage.py new file mode 100644 index 0000000000..8b61539045 --- /dev/null +++ b/tests/test_storage.py @@ -0,0 +1,717 @@ +"""Tests for graphify.storage — NeuG adapter layer.""" +import json +import shutil +import tempfile +from pathlib import Path + +import pytest + +try: + import neug + _has_neug = True +except ImportError: + _has_neug = False + +pytestmark = pytest.mark.skipif(not _has_neug, reason="neug not installed") + +FIXTURES = Path(__file__).parent / "fixtures" +EXTRACTION_JSON = FIXTURES / "extraction.json" + + +def _load_extraction() -> dict: + return json.loads(EXTRACTION_JSON.read_text()) + + +@pytest.fixture() +def tmp_db(tmp_path): + db_path = str(tmp_path / "test.db") + yield db_path + + +def _init(db_path): + from graphify.storage import init_db, ensure_schema + db, conn = init_db(db_path) + ensure_schema(conn) + return db, conn + + +def _close(db, conn): + from graphify.storage import close_db + close_db(db, conn) + + +def _query(conn, cypher): + from graphify.storage import execute_cypher + return execute_cypher(conn, cypher) + + +# --- init_db --- + +def test_init_db_creates_tables(tmp_db): + db, conn = _init(tmp_db) + rows = _query(conn, "MATCH (n:node) RETURN count(n)") + assert rows == [[0]] + _close(db, conn) + + +# --- ingest_extraction: CREATE mode --- + +def test_ingest_extraction_create_mode(tmp_db): + from graphify.storage import ingest_extraction + db, conn = _init(tmp_db) + ext = _load_extraction() + ingest_extraction(conn, ext, incremental=False) + rows = _query(conn, "MATCH (n:node {file_type: 'code'}) RETURN n.id ORDER BY n.id") + ids = sorted([r[0] for r in rows]) + assert "n_attention" in ids + assert "n_transformer" in ids + assert "n_layernorm" in ids + edge_rows = _query(conn, "MATCH (a:node)-[e:edge]->(b:node) WHERE e.relation = 'contains' RETURN count(e)") + assert edge_rows[0][0] == 2 + _close(db, conn) + + +# --- ingest_extraction: MERGE mode --- + +def test_ingest_extraction_merge_mode(tmp_db): + from graphify.storage import ingest_extraction + db, conn = _init(tmp_db) + ext = _load_extraction() + ingest_extraction(conn, ext, incremental=False) + ext["nodes"][0]["label"] = "TransformerV2" + ingest_extraction(conn, ext, incremental=True) + rows = _query(conn, "MATCH (n:node {file_type: 'code'}) WHERE n.id = 'n_transformer' RETURN n.label") + assert rows[0][0] == "TransformerV2" + count = _query(conn, "MATCH (n:node {file_type: 'code'}) RETURN count(n)") + assert count[0][0] == 3 + _close(db, conn) + + +# --- file_type routing --- + +def test_ingest_extraction_file_type_routing(tmp_db): + from graphify.storage import ingest_extraction + db, conn = _init(tmp_db) + ext = _load_extraction() + ingest_extraction(conn, ext, incremental=False) + doc_rows = _query(conn, "MATCH (n:node {file_type: 'document'}) RETURN n.id") + assert len(doc_rows) == 1 + assert doc_rows[0][0] == "n_concept_attn" + _close(db, conn) + + +# --- prune_sources --- + +def test_ingest_extraction_prune(tmp_db): + from graphify.storage import ingest_extraction + db, conn = _init(tmp_db) + ext = _load_extraction() + ingest_extraction(conn, ext, incremental=False) + before = _query(conn, "MATCH (n:node {file_type: 'code'}) RETURN count(n)")[0][0] + assert before == 3 + ingest_extraction(conn, ext, incremental=True, prune_sources=["model.py"]) + after_prune = _query(conn, "MATCH (n:node {file_type: 'code'}) RETURN count(n)")[0][0] + assert after_prune == 3 + _close(db, conn) + + +# --- communities --- + +def test_ingest_communities(tmp_db): + from graphify.storage import ingest_extraction, ingest_communities + db, conn = _init(tmp_db) + ext = _load_extraction() + ingest_extraction(conn, ext, incremental=False) + communities = {0: ["n_transformer", "n_attention"], 1: ["n_layernorm"]} + ingest_communities(conn, communities) + rows = _query(conn, "MATCH (n:node) WHERE n.id = 'n_transformer' RETURN n.community") + assert rows[0][0] == 0 + rows = _query(conn, "MATCH (n:node) WHERE n.id = 'n_layernorm' RETURN n.community") + assert rows[0][0] == 1 + _close(db, conn) + + +# --- execute_cypher --- + +def test_execute_cypher(tmp_db): + from graphify.storage import ingest_extraction + db, conn = _init(tmp_db) + ext = _load_extraction() + ingest_extraction(conn, ext, incremental=False) + rows = _query(conn, "MATCH (n:node {file_type: 'code'}) RETURN n.label ORDER BY n.id") + labels = [r[0] for r in rows] + assert "MultiHeadAttention" in labels + assert "Transformer" in labels + _close(db, conn) + + +def test_execute_cypher_bad_query(tmp_db): + db, conn = _init(tmp_db) + with pytest.raises(RuntimeError): + _query(conn, "THIS IS NOT VALID CYPHER") + _close(db, conn) + + +# --- roundtrip consistency --- + +def test_roundtrip_node_count(tmp_db): + from graphify.storage import ingest_extraction + db, conn = _init(tmp_db) + ext = _load_extraction() + ingest_extraction(conn, ext, incremental=False) + rows = _query(conn, "MATCH (n:node) RETURN count(n)") + assert rows[0][0] == len(ext["nodes"]) + _close(db, conn) + + +# --- community detection & analysis (neug GDS Leiden + Cypher) --- + +_TEST_NODES = [ + {"id": "n1", "label": "AuthService", "file_type": "code", "source_file": "src/auth.py"}, + {"id": "n2", "label": "login", "file_type": "code", "source_file": "src/auth.py"}, + {"id": "n3", "label": "token", "file_type": "code", "source_file": "src/auth.py"}, + {"id": "n4", "label": "UserModel", "file_type": "code", "source_file": "src/models.py"}, + {"id": "n5", "label": "save", "file_type": "code", "source_file": "src/models.py"}, + {"id": "n6", "label": "ApiClient", "file_type": "code", "source_file": "src/client.py"}, + {"id": "n7", "label": "request", "file_type": "code", "source_file": "src/client.py"}, + {"id": "n8", "label": "response", "file_type": "code", "source_file": "src/client.py"}, + {"id": "n9", "label": "parse", "file_type": "code", "source_file": "src/client.py"}, + {"id": "n10", "label": "fetch", "file_type": "code", "source_file": "src/client.py"}, + # Noise nodes + {"id": "n11", "label": "str", "file_type": "concept", "source_file": ""}, + {"id": "n12", "label": "auth.py", "file_type": "code", "source_file": "src/auth.py"}, + {"id": "n13", "label": ".init()", "file_type": "code", "source_file": "src/models.py"}, +] + +_TEST_EDGES = [ + {"from": "n1", "to": "n2", "relation": "calls", "confidence": "EXTRACTED"}, + {"from": "n2", "to": "n3", "relation": "uses", "confidence": "EXTRACTED"}, + {"from": "n4", "to": "n5", "relation": "calls", "confidence": "EXTRACTED"}, + {"from": "n6", "to": "n7", "relation": "calls", "confidence": "EXTRACTED"}, + {"from": "n7", "to": "n8", "relation": "uses", "confidence": "EXTRACTED"}, + {"from": "n8", "to": "n9", "relation": "calls", "confidence": "EXTRACTED"}, + {"from": "n9", "to": "n10", "relation": "uses", "confidence": "EXTRACTED"}, + {"from": "n3", "to": "n6", "relation": "calls", "confidence": "INFERRED"}, + {"from": "n3", "to": "n4", "relation": "uses", "confidence": "AMBIGUOUS"}, + {"from": "n1", "to": "n11", "relation": "uses", "confidence": "EXTRACTED"}, + {"from": "n1", "to": "n12", "relation": "contains", "confidence": "EXTRACTED"}, + {"from": "n4", "to": "n13", "relation": "contains", "confidence": "EXTRACTED"}, + {"from": "n12", "to": "n1", "relation": "contains", "confidence": "EXTRACTED"}, + {"from": "n12", "to": "n2", "relation": "contains", "confidence": "EXTRACTED"}, + {"from": "n12", "to": "n3", "relation": "contains", "confidence": "EXTRACTED"}, +] + + +def _populate_test_graph(conn): + """Insert test nodes and edges into the db.""" + for n in _TEST_NODES: + conn.execute( + "CREATE (n:node {id: $id, label: $label, file_type: $ft, " + "source_file: $sf, source_location: $sl, community: 0, community_name: ''})", + parameters={"id": n["id"], "label": n["label"], "ft": n["file_type"], + "sf": n["source_file"], "sl": ""} + ) + for e in _TEST_EDGES: + conn.execute( + "MATCH (a:node {id: $from}), (b:node {id: $to}) " + "CREATE (a)-[:edge {relation: $rel, confidence: $conf, " + "confidence_score: 1.0, source_file: '', weight: 1.0}]->(b)", + parameters={"from": e["from"], "to": e["to"], + "rel": e["relation"], "conf": e["confidence"]} + ) + + +def test_run_leiden(tmp_db): + from graphify.storage import run_leiden + db, conn = _init(tmp_db) + _populate_test_graph(conn) + communities = run_leiden(conn) + assert len(communities) >= 2, f"Expected >= 2 communities, got {len(communities)}" + total = sum(len(v) for v in communities.values()) + assert total == len(_TEST_NODES), f"Expected {len(_TEST_NODES)} nodes, got {total}" + _close(db, conn) + + +def test_compute_cohesion(tmp_db): + from graphify.storage import run_leiden, ingest_communities, compute_cohesion + db, conn = _init(tmp_db) + _populate_test_graph(conn) + communities = run_leiden(conn) + ingest_communities(conn, communities) + cohesion = compute_cohesion(conn, communities) + for cid, score in cohesion.items(): + assert 0.0 <= score <= 1.0, f"Cohesion {score} out of range for community {cid}" + _close(db, conn) + + +def test_find_god_nodes(tmp_db): + from graphify.storage import run_leiden, ingest_communities, find_god_nodes + db, conn = _init(tmp_db) + _populate_test_graph(conn) + communities = run_leiden(conn) + ingest_communities(conn, communities) + gods = find_god_nodes(conn, top_n=10) + assert len(gods) > 0, "Should have at least 1 god node" + god_ids = {g["id"] for g in gods} + assert "n11" not in god_ids, "n11 (str) should be filtered as noise" + assert "n12" not in god_ids, "n12 (auth.py) should be filtered as file hub" + assert "n13" not in god_ids, "n13 (.init()) should be filtered as method stub" + _close(db, conn) + + +def test_find_surprising_connections(tmp_db): + from graphify.storage import run_leiden, ingest_communities, find_surprising_connections + db, conn = _init(tmp_db) + _populate_test_graph(conn) + communities = run_leiden(conn) + ingest_communities(conn, communities) + surprises = find_surprising_connections(conn, communities, top_n=5) + # Multi-source graph (3 source files) -> should find cross-file edges + assert len(surprises) > 0, "Should find at least 1 surprising connection" + for s in surprises: + assert "source" in s and "target" in s + assert "source_files" in s and len(s["source_files"]) == 2 + assert "confidence" in s and "relation" in s + assert "why" in s + _close(db, conn) + + +def test_label_communities_by_hub(tmp_db): + from graphify.storage import run_leiden, ingest_communities, label_communities_by_hub + db, conn = _init(tmp_db) + _populate_test_graph(conn) + communities = run_leiden(conn) + ingest_communities(conn, communities) + labels = label_communities_by_hub(conn, communities) + assert len(labels) == len(communities) + for cid, name in labels.items(): + assert name, f"Community {cid} has empty label" + _close(db, conn) + + +def test_ingest_communities_batch_writeback(tmp_db): + from graphify.storage import ingest_communities, execute_cypher + db, conn = _init(tmp_db) + _populate_test_graph(conn) + # Include all 13 nodes (noise nodes get their own community) + communities = {0: ["n1", "n2", "n3"], 1: ["n4", "n5"], 2: ["n6", "n7", "n8", "n9", "n10"], 3: ["n11", "n12", "n13"]} + labels = {0: "AuthModule", 1: "Models", 2: "ApiClient", 3: "Noise"} + ingest_communities(conn, communities, community_labels=labels) + rows = execute_cypher(conn, "MATCH (n:node) WHERE n.community IS NOT NULL " + "RETURN n.community AS cid, n.community_name AS name, " + "count(*) AS cnt ORDER BY cid") + assert sum(r[2] for r in rows) == 13 + for cid, name, cnt in rows: + assert name, f"Community {cid} has empty name" + _close(db, conn) + + +# --- incremental delta analysis (freeze-assign leiden) --- + + +def test_run_leiden_freeze_assign(tmp_db): + """Freeze-assign: old nodes frozen, new nodes get previous_community=None.""" + from graphify.storage import run_leiden, ingest_communities, run_leiden_freeze_assign + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + # Phase 1: full leiden + communities = run_leiden(conn) + ingest_communities(conn, communities) + + # Build old_communities {node_id: cid} + old_communities = {} + for cid, node_ids in communities.items(): + for nid in node_ids: + old_communities[nid] = cid + + # Phase 2: add a new node + edge (simulating incremental extract) + conn.execute( + "CREATE (n:node {id: 'n14', label: 'NewFunc', file_type: 'code', " + "source_file: 'src/new.py', source_location: '', community: 0, community_name: ''})" + ) + conn.execute( + "MATCH (a:node {id: 'n14'}), (b:node {id: 'n1'}) " + "CREATE (a)-[:edge {relation: 'calls', confidence: 'EXTRACTED', " + "confidence_score: 1.0, source_file: 'src/new.py', weight: 1.0}]->(b)" + ) + + # Run freeze-assign leiden + results = run_leiden_freeze_assign(conn, old_communities) + + # All old nodes should have previous_community == their old community + results_map = {nid: (new_cid, prev_cid) for nid, new_cid, prev_cid in results} + for nid, old_cid in old_communities.items(): + assert nid in results_map, f"Old node {nid} missing from results" + new_cid, prev_cid = results_map[nid] + assert prev_cid == old_cid, ( + f"Node {nid}: prev_cid={prev_cid} should equal old_cid={old_cid}" + ) + # In freeze-assign mode, old nodes keep their community + assert new_cid == old_cid, ( + f"Node {nid}: new_cid={new_cid} should equal old_cid={old_cid} (frozen)" + ) + + # New node n14 should have previous_community = None + assert "n14" in results_map, "New node n14 missing from results" + new_cid, prev_cid = results_map["n14"] + assert prev_cid is None, ( + f"New node n14 should have prev=None, got {prev_cid}" + ) + _close(db, conn) + + +def test_run_leiden_resolution(tmp_db): + """run_leiden accepts a resolution parameter and passes it to neug GDS.""" + from graphify.storage import run_leiden + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + # Default resolution (1.0) + communities_default = run_leiden(conn) + # High resolution (5.0) — favours smaller communities + communities_high = run_leiden(conn, resolution=5.0) + # Low resolution (0.1) — favours larger communities + communities_low = run_leiden(conn, resolution=0.1) + + # All should produce valid community dicts + assert isinstance(communities_default, dict) + assert isinstance(communities_high, dict) + assert isinstance(communities_low, dict) + + # All nodes should be assigned in each case + all_nodes = {f"n{i}" for i in range(1, 14)} + for label, comms in [("default", communities_default), + ("high", communities_high), + ("low", communities_low)]: + assigned = set() + for node_ids in comms.values(): + assigned.update(node_ids) + assert assigned == all_nodes, f"{label}: missing nodes {all_nodes - assigned}" + + _close(db, conn) + + +def test_run_leiden_freeze_assign_resolution(tmp_db): + """run_leiden_freeze_assign accepts resolution and passes it to neug GDS.""" + from graphify.storage import run_leiden, ingest_communities, run_leiden_freeze_assign + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + # Phase 1: full leiden with resolution=0.5 + communities = run_leiden(conn, resolution=0.5) + ingest_communities(conn, communities) + + old_communities = {} + for cid, node_ids in communities.items(): + for nid in node_ids: + old_communities[nid] = cid + + # Phase 2: add new node + conn.execute( + "CREATE (n:node {id: 'n14', label: 'NewFunc', file_type: 'code', " + "source_file: 'src/new.py', source_location: '', community: 0, community_name: ''})" + ) + conn.execute( + "MATCH (a:node {id: 'n14'}), (b:node {id: 'n1'}) " + "CREATE (a)-[:edge {relation: 'calls', confidence: 'EXTRACTED', " + "confidence_score: 1.0, source_file: 'src/new.py', weight: 1.0}]->(b)" + ) + + # Run freeze-assign with resolution=0.5 (should match the full leiden resolution) + results = run_leiden_freeze_assign(conn, old_communities, resolution=0.5) + + # Old nodes should be frozen + results_map = {nid: (new_cid, prev_cid) for nid, new_cid, prev_cid in results} + for nid, old_cid in old_communities.items(): + assert nid in results_map + new_cid, prev_cid = results_map[nid] + assert new_cid == old_cid, f"Frozen node {nid} moved: {old_cid} -> {new_cid}" + + # New node should have prev=None + assert results_map["n14"][1] is None + + _close(db, conn) + + +def test_analyze_community_changes(): + """Classify communities into 4 types: stable, changed, new, dissolved.""" + from graphify.storage import analyze_community_changes + + # old_communities: {node_id: community_id} + old_communities = { + # Community 0: nodes A, B, C + "A": 0, "B": 0, "C": 0, + # Community 1: nodes D, E + "D": 1, "E": 1, + # Community 2: nodes F, G, H (will be dissolved — all deleted) + "F": 2, "G": 2, "H": 2, + } + + # leiden_results: [(node_id, new_community, previous_community), ...] + # Community 0: stable (A, B, C still there, no new members) + # Community 1: changed (D, E still there + new node Z joined) + # Community 2: dissolved (F, G, H all deleted, not in results) + # Community 3: new (new nodes X, Y form a new community) + leiden_results = [ + ("A", 0, 0), # old, same community + ("B", 0, 0), + ("C", 0, 0), + ("D", 1, 1), # old, same community + ("E", 1, 1), + ("Z", 1, None), # new node joined community 1 + ("X", 3, None), # new community + ("Y", 3, None), + ] + + changes = analyze_community_changes(leiden_results, old_communities) + + # Summary + s = changes["summary"] + assert s["total_before"] == 3, f"Expected 3 before, got {s['total_before']}" + assert s["total_after"] == 3, f"Expected 3 after, got {s['total_after']}" + assert s["stable"] == 1, f"Expected 1 stable, got {s['stable']}" + assert s["changed"] == 1, f"Expected 1 changed, got {s['changed']}" + assert s["new"] == 1, f"Expected 1 new, got {s['new']}" + assert s["dissolved"] == 1, f"Expected 1 dissolved, got {s['dissolved']}" + + # Stable: community 0 + assert "0" in changes["stable_communities"] + + # Changed: community 1 (grow_members=[Z], shrink_members=[]) + assert "1" in changes["changed_communities"] + ch1 = changes["changed_communities"]["1"] + assert ch1["grow_members"] == ["Z"], f"Expected grow=['Z'], got {ch1['grow_members']}" + assert ch1["shrink_members"] == [], f"Expected shrink=[], got {ch1['shrink_members']}" + + # New: community 3 (members=[X, Y]) + assert "3" in changes["new_communities"] + assert sorted(changes["new_communities"]["3"]["members"]) == ["X", "Y"] + + # Dissolved: community 2 + assert len(changes["dissolved_communities"]) == 1 + assert changes["dissolved_communities"][0]["cid"] == 2 + assert changes["dissolved_communities"][0]["old_size"] == 3 + + +def test_delta_analyze(tmp_db): + """End-to-end: full leiden → incremental update → delta_analyze.""" + from graphify.storage import ( + run_leiden, ingest_communities, delta_analyze, + ) + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + # Phase 1: full leiden + communities = run_leiden(conn) + ingest_communities(conn, communities) + + # Build prev_analysis dict (simulates .graphify_analysis.json) + prev_analysis = { + "communities": {str(k): v for k, v in communities.items()}, + "cohesion": {}, + "gods": [], + "surprises": [], + "tokens": {"input": 0, "output": 0}, + } + + # Phase 2: add new node + edge + conn.execute( + "CREATE (n:node {id: 'n14', label: 'NewFunc', file_type: 'code', " + "source_file: 'src/new.py', source_location: '', community: 0, community_name: ''})" + ) + conn.execute( + "MATCH (a:node {id: 'n14'}), (b:node {id: 'n1'}) " + "CREATE (a)-[:edge {relation: 'calls', confidence: 'EXTRACTED', " + "confidence_score: 1.0, source_file: 'src/new.py', weight: 1.0}]->(b)" + ) + + # Run delta_analyze + delta_path = Path(tmp_db).parent / "delta_analysis.json" + + class FakeStages: + def mark(self, stage): + pass + + delta = delta_analyze( + conn, + prev_analysis=prev_analysis, + delta_analysis_path=delta_path, + stages=FakeStages(), + merged={"input_tokens": 0, "output_tokens": 0}, + ) + + # Verify output structure + assert "changed_communities" in delta + assert "new_communities" in delta + assert "stable_communities" in delta + assert "dissolved_communities" in delta + assert "summary" in delta + assert "gods" in delta + assert "surprises" in delta + assert "tokens" in delta + + # Summary should be consistent + s = delta["summary"] + assert s["total_before"] == len(communities), ( + f"total_before={s['total_before']} should equal {len(communities)}" + ) + # total_after = stable + changed + new + assert s["total_after"] == s["stable"] + s["changed"] + s["new"], ( + f"total_after={s['total_after']} != stable+changed+new={s['stable']+s['changed']+s['new']}" + ) + + # Verify file was written + assert delta_path.exists(), f"Delta file not written at {delta_path}" + written = json.loads(delta_path.read_text()) + assert written["summary"] == delta["summary"] + + # Verify DB community property NOT modified (preview mode) + # n14 should still have community=0 (the default from creation) + rows = list(conn.execute("MATCH (n:node {id: 'n14'}) RETURN n.community")) + assert rows[0][0] == 0, f"n14 community should be 0 (preview), got {rows[0][0]}" + + _close(db, conn) + + +# --- file-level clustering --- + + +def test_aggregate_file_edges(tmp_db): + """_aggregate_file_edges should exclude concept nodes and intra-file edges.""" + from graphify.storage import _aggregate_file_edges + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + import tempfile + csv_path = Path(tmp_db).parent / "file_edges.csv" + all_files = _aggregate_file_edges(conn, csv_path) + + # 3 source files in test data + assert all_files == {"src/auth.py", "src/models.py", "src/client.py"} + + # Read CSV and verify edges + import csv as _csv + with open(csv_path) as f: + rows = list(_csv.DictReader(f)) + + # Cross-file edges: n3(auth)→n6(client), n3(auth)→n4(models) + # Intra-file and concept edges should be excluded + edge_pairs = {(r["from_file"], r["to_file"], float(r["weight"])) for r in rows} + assert ("src/auth.py", "src/client.py", 1.0) in edge_pairs + assert ("src/auth.py", "src/models.py", 1.0) in edge_pairs + + # No intra-file edges + for r in rows: + assert r["from_file"] != r["to_file"], "Intra-file edge should be excluded" + + # No concept nodes (source_file='') + for r in rows: + assert r["from_file"] != "", "Concept node should be excluded" + assert r["to_file"] != "", "Concept node should be excluded" + + _close(db, conn) + + +def test_cluster_on_files(tmp_db): + """File-level clustering: community members are file paths, not symbol IDs.""" + from graphify.storage import cluster_on_files + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + communities = cluster_on_files(conn) + + # Community members should be file paths (3 source files in test data) + all_members = set() + for members in communities.values(): + all_members.update(members) + assert all_members == {"src/auth.py", "src/models.py", "src/client.py"}, ( + f"Expected 3 file paths, got {all_members}" + ) + + # No symbol node IDs should appear + for n in _TEST_NODES: + assert n["id"] not in all_members, f"Symbol ID {n['id']} should not be in communities" + + _close(db, conn) + + +def test_cluster_on_files_resolution(tmp_db): + """cluster_on_files accepts resolution parameter.""" + from graphify.storage import cluster_on_files + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + # Just verify it doesn't crash with different resolution + communities = cluster_on_files(conn, resolution=0.5) + assert len(communities) >= 1 + _close(db, conn) + + +def test_delta_analyze_file_level(tmp_db): + """File-level delta analysis: freeze-assign on file-level graph.""" + from graphify.storage import ( + cluster_on_files, delta_analyze, + ) + db, conn = _init(tmp_db) + _populate_test_graph(conn) + + # Phase 1: full file-level clustering + communities = cluster_on_files(conn) + # communities = {cid: [file_paths]} + + # Build prev_analysis (simulates .graphify_analysis.json) + prev_analysis = { + "communities": {str(k): v for k, v in communities.items()}, + "cohesion": {}, + "gods": [], + "surprises": [], + "tokens": {"input": 0, "output": 0}, + } + + # Phase 2: add a new file with a node + edge to existing graph + conn.execute( + "CREATE (n:node {id: 'n14', label: 'NewFunc', file_type: 'code', " + "source_file: 'src/new.py', source_location: '', community: 0, community_name: ''})" + ) + conn.execute( + "MATCH (a:node {id: 'n14'}), (b:node {id: 'n1'}) " + "CREATE (a)-[:edge {relation: 'calls', confidence: 'EXTRACTED', " + "confidence_score: 1.0, source_file: 'src/new.py', weight: 1.0}]->(b)" + ) + + # Run file-level delta_analyze + delta_path = Path(tmp_db).parent / "delta_file_level.json" + + class FakeStages: + def mark(self, stage): + pass + + delta = delta_analyze( + conn, + prev_analysis=prev_analysis, + delta_analysis_path=delta_path, + stages=FakeStages(), + merged={"input_tokens": 0, "output_tokens": 0}, + file_level=True, + ) + + # Verify output structure + assert "changed_communities" in delta + assert "new_communities" in delta + assert "stable_communities" in delta + assert "dissolved_communities" in delta + assert "summary" in delta + + # Summary should be consistent + s = delta["summary"] + assert s["total_after"] == s["stable"] + s["changed"] + s["new"], ( + f"total_after={s['total_after']} != stable+changed+new={s['stable']+s['changed']+s['new']}" + ) + + # Verify file was written + assert delta_path.exists() + written = json.loads(delta_path.read_text()) + assert written["summary"] == delta["summary"] + + _close(db, conn)