graphify/graphify/analyze.py

"""Graph analysis: god nodes (most connected), surprising connections (cross-community), suggested questions."""
from __future__ import annotations
from pathlib import Path
import networkx as nx

from graphify.build import edge_data

# Builtin/mock names that can appear as annotation-derived nodes in pre-existing
# graphs. Excluded from god-node ranking so they don't displace real abstractions
# even if they weren't filtered at extraction time (#1147).
_BUILTIN_NOISE_LABELS = frozenset({
    "str", "int", "float", "bool", "bytes", "bytearray", "complex", "object",
    "True", "False",
    "MagicMock", "Mock", "AsyncMock", "NonCallableMock",
    "NonCallableMagicMock", "PropertyMock", "patch", "sentinel",
    # Python stdlib types commonly confused for project symbols
    "Path", "Any", "Optional", "List", "Dict", "Set", "Tuple", "Union",
    "Callable", "Type", "ClassVar", "Final", "Literal", "Protocol",
    "Counter", "defaultdict", "OrderedDict", "datetime", "Enum",
    "os", "sys", "re", "json", "io", "abc", "typing",
    # Swift / Foundation / SwiftUI framework symbols and module imports that
    # otherwise dominate god-node rankings on Swift codebases (#2147)
    "Foundation", "SwiftUI", "UIKit", "AppKit", "Combine",
    "String", "Int", "Double", "Float", "Bool", "Data", "URL", "Date", "UUID",
    "Sendable", "Codable", "Decodable", "Encodable", "Equatable", "Hashable",
    "Identifiable", "Comparable", "AnyObject", "Error", "LocalizedError",
    "NSObject", "NSString", "NSError", "NSLock",
    "View", "Color", "Font", "DispatchQueue",
})

# Language families — extensions sharing a runtime can legitimately call each other
_LANG_FAMILY: dict[str, str] = {
    **{e: "python" for e in (".py", ".pyw")},
    **{e: "js" for e in (".js", ".jsx", ".mjs", ".cjs", ".ejs", ".ts", ".tsx", ".mts", ".cts", ".vue", ".svelte")},
    **{e: "go" for e in (".go",)},
    **{e: "rust" for e in (".rs",)},
    **{e: "jvm" for e in (".java", ".kt", ".kts", ".scala")},
    **{e: "c" for e in (".c", ".h", ".cpp", ".cc", ".cxx", ".hpp")},
    **{e: "ruby" for e in (".rb", ".rake")},
    **{e: "swift" for e in (".swift",)},
    **{e: "dotnet" for e in (".cs", ".vb")},
    **{e: "php" for e in (".php",)},
    **{e: "r" for e in (".r",)},
    **{e: "cobol" for e in (".cbl", ".cob", ".cobol", ".cpy")},
    **{e: "solidity" for e in (".sol",)},
    **{e: "erlang" for e in (".erl", ".hrl", ".escript")},
}


def _cross_language(src_a: str, src_b: str) -> bool:
    """Return True if two source files belong to different language families."""
    ext_a = Path(src_a).suffix.lower()
    ext_b = Path(src_b).suffix.lower()
    fam_a = _LANG_FAMILY.get(ext_a)
    fam_b = _LANG_FAMILY.get(ext_b)
    if fam_a is None or fam_b is None:
        return False
    return fam_a != fam_b


def _node_community_map(communities: dict[int, list[str]]) -> dict[str, int]:
    """Invert communities dict: node_id -> community_id."""
    return {n: cid for cid, nodes in communities.items() for n in nodes}


def _is_file_node(G: nx.Graph, node_id: str) -> bool:
    """
    Return True if this node is a file-level hub node (e.g. 'client', 'models')
    or an AST method stub (e.g. '.auth_flow()', '.__init__()').

    These are synthetic nodes created by the AST extractor and should be excluded
    from god nodes, surprising connections, and knowledge gap reporting.
    """
    attrs = G.nodes[node_id]
    label = attrs.get("label", "")
    if not label:
        return False
    # File-level hub: label matches the actual source filename — bare basename OR
    # the directory-qualified form the #2032 disambiguation pass may assign.
    source_file = attrs.get("source_file", "")
    if source_file:
        from graphify.build import _is_file_node_label
        if _is_file_node_label(label, source_file):
            return True
    # Method stub: AST extractor labels methods as '.method_name()'
    if label.startswith(".") and label.endswith("()"):
        return True
    # Module-level function stub: labeled 'function_name()' - only has a contains edge
    # These are real functions but structurally isolated by definition; not a gap worth flagging
    if label.endswith("()") and G.degree(node_id) <= 1:
        return True
    return False


_JSON_NOISE_LABELS: frozenset[str] = frozenset({
    "start", "end", "name", "id", "type", "properties",
    "value", "key", "data", "items", "title", "description", "version",
    "dependencies", "devdependencies", "peerdependencies",
    "optionaldependencies", "bundleddependencies", "bundledependencies",
})


def _is_json_key_node(G: nx.Graph, node_id: str) -> bool:
    attrs = G.nodes[node_id]
    src = (attrs.get("source_file") or "").lower()
    if not src.endswith(".json"):
        return False
    label = (attrs.get("label") or "").strip().lower()
    return label in _JSON_NOISE_LABELS


def god_nodes(G: nx.Graph, top_n: int = 10,
              exclude_hubs_percentile: float | None = None) -> list[dict]:
    """Return the top_n most-connected real entities - the core abstractions.

    File-level hub nodes are excluded: they accumulate import/contains edges
    mechanically and don't represent meaningful architectural abstractions.

    ``exclude_hubs_percentile`` (0-100) suppresses nodes whose degree exceeds
    that percentile of the graph's degree distribution, using the same
    threshold computation ``cluster()`` applies (#3205) - so the one setting
    suppresses utility hubs in the ranking AND in community resolution,
    instead of only the latter. ``None`` keeps the historical ranking.
    """
    degree = dict(G.degree())
    hub_threshold: float | None = None
    if exclude_hubs_percentile is not None:
        degrees = sorted(degree.values())
        if degrees:
            idx = max(0, int(len(degrees) * exclude_hubs_percentile / 100) - 1)
            hub_threshold = degrees[idx]
    sorted_nodes = sorted(degree.items(), key=lambda x: x[1], reverse=True)
    result = []
    for node_id, deg in sorted_nodes:
        if hub_threshold is not None and deg > hub_threshold:
            continue
        if _is_file_node(G, node_id) or _is_concept_node(G, node_id) or _is_json_key_node(G, node_id):
            continue
        if G.nodes[node_id].get("label", "") in _BUILTIN_NOISE_LABELS:
            continue
        result.append({
            "id": node_id,
            "label": G.nodes[node_id].get("label", node_id),
            "degree": deg,
        })
        if len(result) >= top_n:
            break
    return result


def surprising_connections(
    G: nx.Graph,
    communities: dict[int, list[str]] | None = None,
    top_n: int = 5,
) -> list[dict]:
    """
    Find connections that are genuinely surprising - not obvious from file structure.

    Strategy:
    - Multi-file corpora: cross-file edges between real entities (not concept nodes).
      Sorted AMBIGUOUS → INFERRED → EXTRACTED.
    - Single-file / single-source corpora: cross-community edges that bridge
      distant parts of the graph (betweenness centrality on edges).
      These reveal non-obvious structural couplings.

    Concept nodes (empty source_file, or injected semantic annotations) are excluded
    from surprising connections because they are intentional, not discovered.
    """
    # Identify unique source files (ignore empty/null source_file)
    source_files = {
        data.get("source_file", "")
        for _, data in G.nodes(data=True)
        if data.get("source_file", "")
    }
    is_multi_source = len(source_files) > 1

    if is_multi_source:
        return _cross_file_surprises(G, communities or {}, top_n)
    else:
        return _cross_community_surprises(G, communities or {}, top_n)


def _is_concept_node(G: nx.Graph, node_id: str) -> bool:
    """
    Return True if this node is a manually-injected semantic concept node
    rather than a real entity found in source code.

    Signals:
    - Empty source_file
    - source_file doesn't look like a real file path (no extension)
    """
    data = G.nodes[node_id]
    source = data.get("source_file", "")
    if not source:
        return True
    # Has no file extension → probably a concept label, not a real file
    if "." not in source.split("/")[-1]:
        return True
    return False


def _is_contains_only_ast_node(G: nx.Graph, node_id: str) -> bool:
    """AST declarations (type aliases, enum members, local consts, JSON keys)
    whose only edge is the structural `contains` from their declaring file are
    working as designed, not documentation gaps (#4205)."""
    from graphify.build import _is_ast_tier
    attrs = G.nodes[node_id]
    if not _is_ast_tier(attrs):
        return False
    if G.degree(node_id) != 1:
        return False
    edges = list(G.edges(node_id, data=True))
    return len(edges) == 1 and edges[0][2].get("relation") == "contains"


from graphify.detect import CODE_EXTENSIONS, DOC_EXTENSIONS, PAPER_EXTENSIONS, IMAGE_EXTENSIONS


def _file_category(path: str) -> str:
    ext = ("." + path.rsplit(".", 1)[-1].lower()) if "." in path else ""
    if ext in CODE_EXTENSIONS:
        return "code"
    if ext in PAPER_EXTENSIONS:
        return "paper"
    if ext in IMAGE_EXTENSIONS:
        return "image"
    return "doc"


def _top_level_dir(path: str) -> str:
    """Return the first path component - used to detect cross-repo edges.

    A path with no "/" is a file at the scan root, so its top-level directory
    is the root itself ("."), not the filename: otherwise every pair of
    root-level files would look like it crosses repos/directories. "." keeps
    root files distinct from an absolute path outside the root, whose first
    component is "". Backslashes and a leading "./" are normalised first,
    since a graph.json loaded without build_from_json can still carry them.
    """
    path = path.replace("\\", "/")
    while path.startswith("./"):
        path = path[2:]
    return path.split("/")[0] if "/" in path else "."


def _surprise_score(
    G: nx.Graph,
    u: str,
    v: str,
    data: dict,
    node_community: dict[str, int],
    u_source: str,
    v_source: str,
    degrees: dict[str, int] | None = None,
) -> tuple[int, list[str]]:
    """Score how surprising a cross-file edge is. Returns (score, reasons)."""
    score = 0
    reasons: list[str] = []

    # 1. Confidence weight - uncertain connections are more noteworthy
    conf = data.get("confidence", "EXTRACTED")
    relation = data.get("relation", "")
    conf_bonus = {"AMBIGUOUS": 3, "INFERRED": 2, "EXTRACTED": 1}.get(conf, 1)

    cat_u = _file_category(u_source)
    cat_v = _file_category(v_source)

    # Suppress all structural bonuses for INFERRED calls/uses that cross language
    # boundaries or connect code to a doc file.  Both cases are resolver pollution:
    # label-matching fires across language families in monorepos, and code→doc
    # "calls" edges are extraction artefacts, not real architecture.
    # Excludes `semantically_similar_to` (genuine cross-boundary insight) and all
    # AMBIGUOUS/EXTRACTED edges (not from the resolver path).
    _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")

    # 2. Cross file-type bonus - code↔paper or code↔image is non-obvious
    if cat_u != cat_v and not _suppress_structural:
        score += 2
        reasons.append(f"crosses file types ({cat_u} ↔ {cat_v})")

    # 3. Cross-repo bonus - different top-level directory. merge-graphs /
    # global add keep source_file repo-relative and tag nodes with `repo`, so
    # the repo is part of the key (absent on a single-repo graph).
    top_u = (G.nodes[u].get("repo"), _top_level_dir(u_source))
    top_v = (G.nodes[v].get("repo"), _top_level_dir(v_source))
    if top_u != top_v and not _suppress_structural:
        score += 2
        reasons.append("connects across different repos/directories")

    # 4. Cross-community bonus - Leiden says these are structurally distant
    cid_u = node_community.get(u)
    cid_v = node_community.get(v)
    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")

    # 4b. Semantic similarity bonus - non-obvious conceptual links score higher
    if data.get("relation") == "semantically_similar_to":
        score = int(score * 1.5)
        reasons.append("semantically similar concepts with no structural link")

    # 5. Peripheral→hub: a low-degree node connecting to a high-degree one
    deg_u = degrees[u] if degrees is not None else G.degree(u)
    deg_v = degrees[v] if degrees is not None else G.degree(v)
    if min(deg_u, deg_v) <= 2 and max(deg_u, deg_v) >= 5:
        score += 1
        peripheral = G.nodes[u].get("label", u) if deg_u <= 2 else G.nodes[v].get("label", v)
        hub = G.nodes[v].get("label", v) if deg_u <= 2 else G.nodes[u].get("label", u)
        reasons.append(f"peripheral node `{peripheral}` unexpectedly reaches hub `{hub}`")

    return score, reasons


def _cross_file_surprises(G: nx.Graph, communities: dict[int, list[str]], top_n: int) -> list[dict]:
    """
    Cross-file edges between real code/doc entities, ranked by a composite
    surprise score rather than confidence alone.

    Surprise score accounts for:
    - Confidence (AMBIGUOUS > INFERRED > EXTRACTED)
    - Cross file-type (code↔paper is more surprising than code↔code)
    - Cross-repo (different top-level directory)
    - Cross-community (Leiden says structurally distant)
    - Peripheral→hub (low-degree node reaching a god node)

    Each result includes a 'why' field explaining what makes it non-obvious.
    """
    node_community = _node_community_map(communities)
    degrees = dict(G.degree())
    candidates = []

    for u, v, data in G.edges(data=True):
        relation = data.get("relation", "")
        if relation in ("imports", "imports_from", "contains", "method"):
            continue
        if _is_concept_node(G, u) or _is_concept_node(G, v):
            continue
        if _is_file_node(G, u) or _is_file_node(G, v):
            continue

        u_source = G.nodes[u].get("source_file", "")
        v_source = G.nodes[v].get("source_file", "")

        if not u_source or not v_source or u_source == v_source:
            continue

        score, reasons = _surprise_score(G, u, v, data, node_community, u_source, v_source, degrees)
        src_id = data.get("_src", u)
        if src_id not in G.nodes:
            src_id = u
        tgt_id = data.get("_tgt", v)
        if tgt_id not in G.nodes:
            tgt_id = v
        candidates.append({
            "_score": score,
            "source": G.nodes[src_id].get("label", src_id),
            "target": G.nodes[tgt_id].get("label", tgt_id),
            "source_files": [
                G.nodes[src_id].get("source_file", ""),
                G.nodes[tgt_id].get("source_file", ""),
            ],
            "confidence": data.get("confidence", "EXTRACTED"),
            "relation": relation,
            "why": "; ".join(reasons) if reasons else "cross-file semantic connection",
        })

    candidates.sort(key=lambda x: x["_score"], reverse=True)
    for c in candidates:
        c.pop("_score")

    if candidates:
        return candidates[:top_n]

    return _cross_community_surprises(G, communities, top_n)


def _cross_community_surprises(
    G: nx.Graph,
    communities: dict[int, list[str]],
    top_n: int,
) -> list[dict]:
    """
    For single-source corpora: find edges that bridge different communities.
    These are surprising because Leiden grouped everything else tightly -
    these edges cut across the natural structure.

    Falls back to high-betweenness edges if no community info is provided.
    """
    if not communities:
        # No community info - use edge betweenness centrality
        if G.number_of_edges() == 0:
            return []
        if G.number_of_nodes() > 5000:
            return []
        betweenness = nx.edge_betweenness_centrality(G)
        top_edges = sorted(betweenness.items(), key=lambda x: x[1], reverse=True)[:top_n]
        result = []
        for (u, v), score in top_edges:
            data = edge_data(G, u, v)
            result.append({
                "source": G.nodes[u].get("label", u),
                "target": G.nodes[v].get("label", v),
                "source_files": [
                    G.nodes[u].get("source_file", ""),
                    G.nodes[v].get("source_file", ""),
                ],
                "confidence": data.get("confidence", "EXTRACTED"),
                "relation": data.get("relation", ""),
                "note": f"Bridges graph structure (betweenness={score:.3f})",
            })
        return result

    # Build node → community map
    node_community = _node_community_map(communities)

    surprises = []
    for u, v, data in G.edges(data=True):
        cid_u = node_community.get(u)
        cid_v = node_community.get(v)
        if cid_u is None or cid_v is None or cid_u == cid_v:
            continue
        # Skip file hub nodes and plain structural edges
        if _is_file_node(G, u) or _is_file_node(G, v):
            continue
        relation = data.get("relation", "")
        if relation in ("imports", "imports_from", "contains", "method"):
            continue
        # This edge crosses community boundaries - interesting
        confidence = data.get("confidence", "EXTRACTED")
        src_id = data.get("_src", u)
        if src_id not in G.nodes:
            src_id = u
        tgt_id = data.get("_tgt", v)
        if tgt_id not in G.nodes:
            tgt_id = v
        surprises.append({
            "source": G.nodes[src_id].get("label", src_id),
            "target": G.nodes[tgt_id].get("label", tgt_id),
            "source_files": [
                G.nodes[src_id].get("source_file", ""),
                G.nodes[tgt_id].get("source_file", ""),
            ],
            "confidence": confidence,
            "relation": relation,
            "note": f"Bridges community {cid_u} → community {cid_v}",
            "_pair": tuple(sorted([cid_u, cid_v])),
        })

    # Sort: AMBIGUOUS first, then INFERRED, then EXTRACTED
    order = {"AMBIGUOUS": 0, "INFERRED": 1, "EXTRACTED": 2}
    surprises.sort(key=lambda x: order.get(x["confidence"], 3))

    # Deduplicate by community pair - one representative edge per (A→B) boundary.
    # Without this, a single high-betweenness god node dominates all results.
    seen_pairs: set[tuple] = set()
    deduped = []
    for s in surprises:
        pair = s.pop("_pair")
        if pair not in seen_pairs:
            seen_pairs.add(pair)
            deduped.append(s)
    return deduped[:top_n]


def suggest_questions(
    G: nx.Graph,
    communities: dict[int, list[str]],
    community_labels: dict[int, str],
    top_n: int = 7,
) -> list[dict]:
    """
    Generate questions the graph is uniquely positioned to answer.
    Based on: AMBIGUOUS edges, bridge nodes, underexplored god nodes, isolated nodes.
    Each question has a 'type', 'question', and 'why' field.
    Interleave question types before applying top_n so an abundant early type
    cannot crowd out later signals. Preserve the existing order within each type
    and category generation order when the limit is smaller than the type count.
    """
    if community_labels:
        community_labels = {int(k) if isinstance(k, str) else k: v for k, v in community_labels.items()}

    questions = []
    node_community = _node_community_map(communities)

    # 1. AMBIGUOUS edges → unresolved relationship questions.
    # Tagged ``low_confidence=True`` (#4199): the audit trail stays
    # intact but downstream renderers (report.py) move these into a
    # separate "Low-confidence Hints" section so they do not compete
    # with EXTRACTED/INFERRED bridges in the main list.
    for u, v, data in G.edges(data=True):
        if data.get("confidence") == "AMBIGUOUS":
            ul = G.nodes[u].get("label", u)
            vl = G.nodes[v].get("label", v)
            relation = data.get("relation", "related to")
            questions.append({
                "type": "ambiguous_edge",
                "question": f"What is the exact relationship between `{ul}` and `{vl}`?",
                "why": f"Edge tagged AMBIGUOUS (relation: {relation}) - confidence is low.",
                "low_confidence": True,
            })

    # 2. Bridge nodes (high betweenness) → cross-cutting concern questions
    if G.number_of_edges() > 0:
        k = min(100, G.number_of_nodes()) if G.number_of_nodes() > 1000 else None
        betweenness = nx.betweenness_centrality(G, k=k, seed=42)
        # Top bridge nodes that are NOT file-level hubs
        bridges = sorted(
            [(n, s) for n, s in betweenness.items()
             if not _is_file_node(G, n) and not _is_concept_node(G, n) and s > 0],
            key=lambda x: x[1],
            reverse=True,
        )[:3]
        for node_id, score in bridges:
            label = G.nodes[node_id].get("label", node_id)
            cid = node_community.get(node_id)
            comm_label = community_labels.get(cid, f"Community {cid}") if cid is not None else "unknown"
            neighbors = list(G.neighbors(node_id))
            neighbor_comms = {node_community.get(n) for n in neighbors if node_community.get(n) != cid}
            if neighbor_comms:
                other_labels = [community_labels.get(c, f"Community {c}") for c in neighbor_comms]
                questions.append({
                    "type": "bridge_node",
                    "question": f"Why does `{label}` connect `{comm_label}` to {', '.join(f'`{l}`' for l in other_labels)}?",
                    "why": f"High betweenness centrality ({score:.3f}) - this node is a cross-community bridge.",
                })

    # 3. God nodes with many INFERRED edges → verification questions
    degree = dict(G.degree())
    top_nodes = sorted(
        [(n, d) for n, d in degree.items() if not _is_file_node(G, n)],
        key=lambda x: x[1],
        reverse=True,
    )[:5]
    for node_id, _ in top_nodes:
        inferred = [
            (u, v, d) for u, v, d in G.edges(node_id, data=True)
            if d.get("confidence") == "INFERRED"
        ]
        if len(inferred) >= 2:
            label = G.nodes[node_id].get("label", node_id)
            # Use _src/_tgt to get the correct direction; fall back to v (the other node)
            others = []
            for u, v, d in inferred[:2]:
                src_id = d.get("_src", u)
                if src_id not in G.nodes:
                    src_id = u
                tgt_id = d.get("_tgt", v)
                if tgt_id not in G.nodes:
                    tgt_id = v
                other_id = tgt_id if src_id == node_id else src_id
                others.append(G.nodes[other_id].get("label", other_id))
            questions.append({
                "type": "verify_inferred",
                "question": f"Are the {len(inferred)} inferred relationships involving `{label}` (e.g. with `{others[0]}` and `{others[1]}`) actually correct?",
                "why": f"`{label}` has {len(inferred)} INFERRED edges - model-reasoned connections that need verification.",
            })

    # 4. Isolated or weakly-connected nodes → exploration questions
    isolated = [
        n for n in G.nodes()
        if G.degree(n) <= 1
        and not _is_file_node(G, n)
        and not _is_concept_node(G, n)
        and not _is_contains_only_ast_node(G, n)
        and G.nodes[n].get("file_type") != "rationale"
    ]
    if isolated:
        labels = [G.nodes[n].get("label", n) for n in isolated[:3]]
        questions.append({
            "type": "isolated_nodes",
            "question": f"What connects {', '.join(f'`{l}`' for l in labels)} to the rest of the system?",
            "why": f"{len(isolated)} weakly-connected nodes found - possible documentation gaps or missing edges.",
        })

    # 5. Low-cohesion communities → structural questions
    from .cluster import cohesion_score
    for cid, nodes in communities.items():
        score = cohesion_score(G, nodes)
        if score < 0.15 and len(nodes) >= 5:
            label = community_labels.get(cid, f"Community {cid}")
            questions.append({
                "type": "low_cohesion",
                "question": f"Should `{label}` be split into smaller, more focused modules?",
                "why": f"Cohesion score {score} - nodes in this community are weakly interconnected.",
            })

    if not questions:
        return [{
            "type": "no_signal",
            "question": None,
            "why": (
                "Not enough signal to generate questions. "
                "This usually means the corpus has no AMBIGUOUS edges, no bridge nodes, "
                "no INFERRED relationships, and all communities are tightly cohesive. "
                "Add more files or run with --mode deep to extract richer edges."
            ),
        }]

    from itertools import zip_longest

    by_type: dict[str, list[dict]] = {}
    for question in questions:
        by_type.setdefault(question["type"], []).append(question)
    diversified = [
        question for row in zip_longest(*by_type.values())
        for question in row if question is not None
    ]
    return diversified[:top_n]


def graph_diff(G_old: nx.Graph, G_new: nx.Graph) -> dict:
    """Compare two graph snapshots and return what changed.

    Returns:
        {
          "new_nodes": [{"id": ..., "label": ...}],
          "removed_nodes": [{"id": ..., "label": ...}],
          "new_edges": [{"source": ..., "target": ..., "relation": ..., "confidence": ...}],
          "removed_edges": [...],
          "summary": "3 new nodes, 5 new edges, 1 node removed"
        }
    """
    old_nodes = set(G_old.nodes())
    new_nodes = set(G_new.nodes())

    added_node_ids = new_nodes - old_nodes
    removed_node_ids = old_nodes - new_nodes

    new_nodes_list = [
        {"id": n, "label": G_new.nodes[n].get("label", n)}
        for n in added_node_ids
    ]
    removed_nodes_list = [
        {"id": n, "label": G_old.nodes[n].get("label", n)}
        for n in removed_node_ids
    ]

    def edge_key(G: nx.Graph, u: str, v: str, data: dict) -> tuple:
        if G.is_directed():
            return (u, v, data.get("relation", ""))
        return (min(u, v), max(u, v), data.get("relation", ""))

    def edge_dir(data: dict) -> tuple | None:
        # Stored direction of an undirected edge (build_from_json, load_node_link_graph).
        if "_src" in data and "_tgt" in data:
            return (data["_src"], data["_tgt"])
        return None

    old_edge_dirs = {
        edge_key(G_old, u, v, d): edge_dir(d)
        for u, v, d in G_old.edges(data=True)
    }
    new_edge_dirs = {
        edge_key(G_new, u, v, d): edge_dir(d)
        for u, v, d in G_new.edges(data=True)
    }
    old_edge_keys = set(old_edge_dirs)
    new_edge_keys = set(new_edge_dirs)

    # A reversed edge (a->b became b->a) keeps its undirected key. Report it as
    # removed + added only when both snapshots carry _src/_tgt and they differ,
    # so graphs without the markers diff exactly as before (#4067).
    reversed_edge_keys = {
        k for k in old_edge_keys & new_edge_keys
        if old_edge_dirs[k] and new_edge_dirs[k] and old_edge_dirs[k] != new_edge_dirs[k]
    }
    added_edge_keys = (new_edge_keys - old_edge_keys) | reversed_edge_keys
    removed_edge_keys = (old_edge_keys - new_edge_keys) | reversed_edge_keys

    new_edges_list = []
    for u, v, d in G_new.edges(data=True):
        if edge_key(G_new, u, v, d) in added_edge_keys:
            new_edges_list.append({
                "source": d.get("_src", u),
                "target": d.get("_tgt", v),
                "relation": d.get("relation", ""),
                "confidence": d.get("confidence", ""),
            })

    removed_edges_list = []
    for u, v, d in G_old.edges(data=True):
        if edge_key(G_old, u, v, d) in removed_edge_keys:
            removed_edges_list.append({
                "source": d.get("_src", u),
                "target": d.get("_tgt", v),
                "relation": d.get("relation", ""),
                "confidence": d.get("confidence", ""),
            })

    parts = []
    if new_nodes_list:
        parts.append(f"{len(new_nodes_list)} new node{'s' if len(new_nodes_list) != 1 else ''}")
    if new_edges_list:
        parts.append(f"{len(new_edges_list)} new edge{'s' if len(new_edges_list) != 1 else ''}")
    if removed_nodes_list:
        parts.append(f"{len(removed_nodes_list)} node{'s' if len(removed_nodes_list) != 1 else ''} removed")
    if removed_edges_list:
        parts.append(f"{len(removed_edges_list)} edge{'s' if len(removed_edges_list) != 1 else ''} removed")
    summary = ", ".join(parts) if parts else "no changes"

    return {
        "new_nodes": new_nodes_list,
        "removed_nodes": removed_nodes_list,
        "new_edges": new_edges_list,
        "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

        # Deferred `import(...)` edges are real dependencies but do not form a
        # hard file-level cycle, so they are excluded from cycle detection (#1241).
        if data.get("deferred"):
            continue
        # Type-only imports/re-exports (`import type` / `export type ... from`)
        # are erased at compile time - a cycle that closes through one cannot
        # exist at runtime (#3123). The edge itself stays in the graph.
        if data.get("type_only"):
            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.
    # Pass length_bound so networkx prunes during enumeration rather than
    # enumerating all elementary cycles and post-filtering — avoids exponential
    # blowup on dense graphs with many long cycles (#1196).
    cycles: list[list[str]] = []
    for cycle in nx.simple_cycles(file_graph, length_bound=max_cycle_length):
        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