diff --git a/graphify/analyze.py b/graphify/analyze.py index fd94c0ca..f3e08103 100644 --- a/graphify/analyze.py +++ b/graphify/analyze.py @@ -606,3 +606,107 @@ def graph_diff(G_old: nx.Graph, G_new: nx.Graph) -> dict: "removed_edges": removed_edges_list, "summary": summary, } + + +def find_import_cycles( + G: nx.Graph, + max_cycle_length: int = 5, + top_n: int = 20, +) -> list[dict]: + """Detect circular import dependencies at the file level. + + Collapses symbol-level nodes to their parent file (using source_file attr + or 'contains' edges), builds a directed file-level graph from imports_from + edges, then finds simple cycles. + + Args: + G: The full knowledge graph (may be undirected or directed). + max_cycle_length: Only report cycles with at most this many files. + top_n: Maximum number of cycles to return (shortest first). + + Returns: + List of cycle records with stable structure: + { + "cycle": ["a.ts", "b.ts"], + "length": 2, + "why": "circular dependency" + } + """ + def _endpoint_source_file(node_id: str) -> str: + attrs = G.nodes.get(node_id, {}) + src_file = attrs.get("source_file", "") + return src_file if isinstance(src_file, str) else "" + + # Step 1: Build a directed file-level graph from import/re-export edges. + # IMPORTANT: resolve endpoints using source_file only; never infer from label/id. + file_graph = nx.DiGraph() + + for u, v, data in G.edges(data=True): + rel = data.get("relation", "") + if rel not in ("imports_from", "re_exports"): + continue + + src_file_attr = data.get("source_file", "") + if not isinstance(src_file_attr, str) or not src_file_attr: + continue + + u_file = _endpoint_source_file(u) + v_file = _endpoint_source_file(v) + + # Works for both DiGraph and Graph inputs: + # orient edge from edge.source_file endpoint to the opposite endpoint. + if u_file == src_file_attr: + tgt_file = v_file + elif v_file == src_file_attr: + tgt_file = u_file + else: + # Fallback: if source endpoint cannot be matched exactly, + # still treat edge.source_file as source and pick the opposite endpoint + # only if one endpoint has a real source_file. + tgt_file = v_file if v_file and v_file != src_file_attr else u_file + + if not tgt_file: + continue + + file_graph.add_edge(src_file_attr, tgt_file) + + if not file_graph.edges(): + return [] + + # Step 2: Find simple cycles, bounded by length. + cycles: list[list[str]] = [] + for cycle in nx.simple_cycles(file_graph): + if len(cycle) <= max_cycle_length: + cycles.append(cycle) + if len(cycles) >= top_n * 10: + # Stop early to avoid combinatorial explosion + break + + # Step 3: Sort by length (shortest = tightest coupling), then deduplicate. + cycles.sort(key=len) + + # Deduplicate rotations: normalize each cycle by starting from the + # lexicographically smallest element. + seen: set[tuple[str, ...]] = set() + unique_cycles: list[list[str]] = [] + for cycle in cycles: + core = list(cycle) + if not core: + continue + min_idx = core.index(min(core)) + normalized = tuple(core[min_idx:] + core[:min_idx]) + if normalized not in seen: + seen.add(normalized) + unique_cycles.append(list(normalized)) + if len(unique_cycles) >= top_n: + break + + result: list[dict] = [] + for cycle in unique_cycles: + result.append({ + "cycle": cycle, + "length": len(cycle), + "why": "circular dependency", + }) + + return result diff --git a/graphify/report.py b/graphify/report.py index 9ca7a2fc..f0210897 100644 --- a/graphify/report.py +++ b/graphify/report.py @@ -119,6 +119,21 @@ def generate( else: lines.append("- None detected - all connections are within the same source files.") + # Circular imports surfaced from file-level dependency graph. + from .analyze import find_import_cycles + cycles = find_import_cycles(G) + lines += ["", "## Import Cycles"] + if cycles: + for c in cycles: + cycle = c.get("cycle", []) + length = c.get("length", len(cycle)) + if not cycle: + continue + cycle_path = " -> ".join(cycle + [cycle[0]]) + lines.append(f"- {length}-file cycle: `{cycle_path}`") + else: + lines.append("- None detected.") + hyperedges = G.graph.get("hyperedges", []) if hyperedges: lines += ["", "## Hyperedges (group relationships)"] diff --git a/tests/test_analyze.py b/tests/test_analyze.py index e0608976..ecf1555d 100644 --- a/tests/test_analyze.py +++ b/tests/test_analyze.py @@ -5,7 +5,8 @@ import pytest from pathlib import Path from graphify.build import build_from_json from graphify.cluster import cluster -from graphify.analyze import god_nodes, surprising_connections, _is_concept_node, graph_diff, _surprise_score, _file_category, _is_json_key_node +from graphify.analyze import god_nodes, surprising_connections, _is_concept_node, graph_diff, _surprise_score, _file_category, _is_json_key_node, find_import_cycles +from graphify.extract import _make_id FIXTURES = Path(__file__).parent / "fixtures" @@ -600,3 +601,123 @@ def test_god_nodes_filter_is_case_insensitive(): labels = [r["label"] for r in result] for variant in ("Start", "START", "Name", "ID"): assert variant not in labels, f"`{variant}` should be filtered as JSON-key noise" + + +# ── find_import_cycles tests ────────────────────────────────────────────────── + + +def _make_file_node(path: str) -> tuple[str, dict]: + """Create a graph node resembling real graphify schema.""" + nid = _make_id(path) + return nid, {"label": Path(path).name, "source_file": path, "file_type": "code"} + + +def _make_cycle_graph_directed() -> nx.DiGraph: + G = nx.DiGraph() + + a_id, a = _make_file_node("src/a.ts") + b_id, b = _make_file_node("src/b.ts") + c_id, c = _make_file_node("src/c.ts") + d_id, d = _make_file_node("src/d.ts") + ext_id = _make_id("react") + + G.add_node(a_id, **a) + G.add_node(b_id, **b) + G.add_node(c_id, **c) + G.add_node(d_id, **d) + # External-like node (no source_file): must be skipped safely. + G.add_node(ext_id, label="react", file_type="code") + + # 2-cycle: a <-> b + G.add_edge(a_id, b_id, relation="imports_from", source_file="src/a.ts", confidence="EXTRACTED") + G.add_edge(b_id, a_id, relation="imports_from", source_file="src/b.ts", confidence="EXTRACTED") + + # 3-cycle: b -> c -> d -> b + G.add_edge(b_id, c_id, relation="imports_from", source_file="src/b.ts", confidence="EXTRACTED") + G.add_edge(c_id, d_id, relation="imports_from", source_file="src/c.ts", confidence="EXTRACTED") + G.add_edge(d_id, b_id, relation="imports_from", source_file="src/d.ts", confidence="EXTRACTED") + + # Self-loop: c imports itself + G.add_edge(c_id, c_id, relation="imports_from", source_file="src/c.ts", confidence="EXTRACTED") + + # Mixed edge types: must not bleed into cycle graph + G.add_edge(a_id, ext_id, relation="calls", source_file="src/a.ts", confidence="INFERRED") + G.add_edge(a_id, ext_id, relation="contains", source_file="src/a.ts", confidence="EXTRACTED") + + # Edge whose target has no source_file: must be skipped, no garbage label fallback + G.add_edge(a_id, ext_id, relation="imports_from", source_file="src/a.ts", confidence="EXTRACTED") + + return G + + +def test_find_import_cycles_returns_structured_records(): + G = _make_cycle_graph_directed() + cycles = find_import_cycles(G) + assert isinstance(cycles, list) + assert cycles + assert isinstance(cycles[0], dict) + assert "cycle" in cycles[0] + assert "length" in cycles[0] + assert "why" in cycles[0] + + +def test_find_import_cycles_detects_2_and_3_cycles(): + G = _make_cycle_graph_directed() + cycles = find_import_cycles(G) + cycle_sets = [set(c["cycle"]) for c in cycles] + assert any({"src/a.ts", "src/b.ts"}.issubset(s) for s in cycle_sets) + assert any({"src/b.ts", "src/c.ts", "src/d.ts"}.issubset(s) for s in cycle_sets) + + +def test_find_import_cycles_includes_self_loop_cycle(): + G = _make_cycle_graph_directed() + cycles = find_import_cycles(G) + assert any(c["cycle"] == ["src/c.ts"] and c["length"] == 1 for c in cycles) + + +def test_find_import_cycles_respects_max_cycle_length(): + G = _make_cycle_graph_directed() + cycles = find_import_cycles(G, max_cycle_length=2) + assert all(c["length"] <= 2 for c in cycles) + + +def test_find_import_cycles_skips_nodes_without_source_file(): + G = _make_cycle_graph_directed() + cycles = find_import_cycles(G) + flat = " ".join(" ".join(c["cycle"]) for c in cycles) + assert "react" not in flat + + +def test_find_import_cycles_handles_undirected_graph_input(): + Gd = _make_cycle_graph_directed() + Gu = nx.Graph() + Gu.add_nodes_from(Gd.nodes(data=True)) + Gu.add_edges_from(Gd.edges(data=True)) + cycles = find_import_cycles(Gu) + assert cycles # should still resolve orientation via edge.source_file + + +def test_find_import_cycles_ignores_non_import_relations(): + G = nx.DiGraph() + a_id, a = _make_file_node("src/a.ts") + b_id, b = _make_file_node("src/b.ts") + G.add_node(a_id, **a) + G.add_node(b_id, **b) + # Bidirectional non-import edges should not be considered a dependency cycle. + G.add_edge(a_id, b_id, relation="calls", source_file="src/a.ts", confidence="INFERRED") + G.add_edge(b_id, a_id, relation="contains", source_file="src/b.ts", confidence="EXTRACTED") + assert find_import_cycles(G) == [] + + +def test_find_import_cycles_empty_graph(): + assert find_import_cycles(nx.DiGraph()) == [] + + +def test_find_import_cycles_no_cycles(): + G = nx.DiGraph() + x_id, x = _make_file_node("x.ts") + y_id, y = _make_file_node("y.ts") + G.add_node(x_id, **x) + G.add_node(y_id, **y) + G.add_edge(x_id, y_id, relation="imports_from", source_file="x.ts", confidence="EXTRACTED") + assert find_import_cycles(G) == []