feat: introduce GraphBackend protocol and refactor GraphStore as facade

Extract NetworkXBackend from GraphStore, define GraphBackend and CypherBackend
protocols, fix 3 NetworkX leaks in query.py, views.py, layer_classifier.py.
Zero behavioral changes — all 361 tests pass.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: aksOps

src/code_intelligence/classifiers/layer_classifier.py

@@ -5,7 +5,7 @@
 import re
 from typing import Any, Sequence
 
-from code_intelligence.models.graph import GraphNode, NodeKind, SourceLocation
+from code_intelligence.models.graph import GraphNode, NodeKind
 
 _FRONTEND_NODE_KINDS = {NodeKind.COMPONENT, NodeKind.HOOK}
 _BACKEND_NODE_KINDS = {NodeKind.GUARD, NodeKind.MIDDLEWARE, NodeKind.ENDPOINT, NodeKind.REPOSITORY, NodeKind.DATABASE_CONNECTION, NodeKind.QUERY}
@@ -35,25 +35,10 @@ def classify(self, nodes: Sequence[GraphNode]) -> None:
             node.properties["layer"] = self._classify_one(node)
 
     def classify_store(self, store: Any) -> None:
-        """Classify nodes directly in a GraphStore, updating networkx data in-place."""
-        for node_id, data in store.graph.nodes(data=True):
-            if "kind" not in data:
-                continue
-            # Build a lightweight GraphNode for classification
-            loc = data.get("location")
-            if isinstance(loc, dict):
-                loc = SourceLocation(**loc)
-            node = GraphNode(
-                id=data.get("id", node_id),
-                kind=NodeKind(data["kind"]),
-                label=data.get("label", ""),
-                location=loc,
-                properties=data.get("properties", {}),
-            )
+        """Classify nodes in a GraphStore, updating properties via public API."""
+        for node in store.all_nodes():
             layer = self._classify_one(node)
-            # Update the networkx data dict directly
-            props = data.setdefault("properties", {})
-            props["layer"] = layer
+            store.update_node_properties(node.id, {"layer": layer})
 
     def _classify_one(self, node: GraphNode) -> str:
         if node.kind in _FRONTEND_NODE_KINDS:

src/code_intelligence/graph/backend.py

@@ -0,0 +1,52 @@
+"""Protocol definitions for graph storage backends."""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, runtime_checkable
+
+from code_intelligence.models.graph import (
+    EdgeKind,
+    GraphEdge,
+    GraphNode,
+    NodeKind,
+)
+
+
+@runtime_checkable
+class GraphBackend(Protocol):
+    """Contract that every graph storage backend must satisfy."""
+
+    def add_node(self, node: GraphNode) -> None: ...
+    def add_edge(self, edge: GraphEdge) -> None: ...
+    def clear(self) -> None: ...
+    def get_node(self, node_id: str) -> GraphNode | None: ...
+    def has_node(self, node_id: str) -> bool: ...
+    def get_edges_between(self, source: str, target: str) -> list[GraphEdge]: ...
+    def all_nodes(self) -> list[GraphNode]: ...
+    def all_edges(self) -> list[GraphEdge]: ...
+    def nodes_by_kind(self, kind: NodeKind) -> list[GraphNode]: ...
+    def edges_by_kind(self, kind: EdgeKind) -> list[GraphEdge]: ...
+
+    @property
+    def node_count(self) -> int: ...
+    @property
+    def edge_count(self) -> int: ...
+
+    def neighbors(
+        self, node_id: str,
+        edge_kinds: set[EdgeKind] | None = None,
+        direction: str = "both",
+    ) -> list[str]: ...
+
+    def find_cycles(self, limit: int = 100) -> list[list[str]]: ...
+    def shortest_path(self, source: str, target: str) -> list[str] | None: ...
+    def subgraph(self, node_ids: set[str]) -> GraphBackend: ...
+    def update_node_properties(self, node_id: str, properties: dict[str, Any]) -> None: ...
+    def close(self) -> None: ...
+
+
+@runtime_checkable
+class CypherBackend(Protocol):
+    """Optional capability for backends supporting Cypher queries."""
+
+    def query_cypher(self, cypher: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: ...

src/code_intelligence/graph/backends/__init__.py

@@ -0,0 +1,23 @@
+"""Graph backend factory."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from code_intelligence.graph.backend import GraphBackend
+
+
+def create_backend(backend_name: str = "networkx", **kwargs) -> GraphBackend:
+    """Create a graph backend by name."""
+    if backend_name == "networkx":
+        from code_intelligence.graph.backends.networkx import NetworkXBackend
+        return NetworkXBackend()
+    elif backend_name == "kuzu":
+        from code_intelligence.graph.backends.kuzu import KuzuBackend
+        return KuzuBackend(db_path=kwargs.get("path", ".code-intelligence/graph.kuzu"))
+    elif backend_name == "sqlite":
+        from code_intelligence.graph.backends.sqlite_backend import SqliteGraphBackend
+        return SqliteGraphBackend(db_path=kwargs.get("path", ".code-intelligence/graph.db"))
+    else:
+        raise ValueError(f"Unknown graph backend: {backend_name}")

src/code_intelligence/graph/backends/networkx.py

@@ -0,0 +1,127 @@
+"""NetworkX-backed graph backend."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import networkx as nx
+
+from code_intelligence.models.graph import (
+    EdgeKind,
+    GraphEdge,
+    GraphNode,
+    NodeKind,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class NetworkXBackend:
+    """In-memory graph backend using NetworkX MultiDiGraph."""
+
+    def __init__(self) -> None:
+        self._g: nx.MultiDiGraph = nx.MultiDiGraph()
+
+    @property
+    def node_count(self) -> int:
+        return self._g.number_of_nodes()
+
+    @property
+    def edge_count(self) -> int:
+        return self._g.number_of_edges()
+
+    def add_node(self, node: GraphNode) -> None:
+        if node.id in self._g:
+            logger.debug("Duplicate node ID %s, keeping first", node.id)
+            return
+        self._g.add_node(node.id, **node.model_dump())
+
+    def add_edge(self, edge: GraphEdge) -> None:
+        self._g.add_edge(edge.source, edge.target, **edge.model_dump())
+
+    def clear(self) -> None:
+        self._g.clear()
+
+    def get_node(self, node_id: str) -> GraphNode | None:
+        if node_id not in self._g:
+            return None
+        return GraphNode(**self._g.nodes[node_id])
+
+    def has_node(self, node_id: str) -> bool:
+        return node_id in self._g
+
+    def get_edges_between(self, source: str, target: str) -> list[GraphEdge]:
+        if not self._g.has_edge(source, target):
+            return []
+        return [GraphEdge(**data) for _key, data in self._g[source][target].items()]
+
+    def all_nodes(self) -> list[GraphNode]:
+        return [
+            GraphNode(**data)
+            for _, data in self._g.nodes(data=True)
+            if "id" in data and "kind" in data
+        ]
+
+    def all_edges(self) -> list[GraphEdge]:
+        return [
+            GraphEdge(**data)
+            for _, _, data in self._g.edges(data=True)
+            if "source" in data and "target" in data
+        ]
+
+    def nodes_by_kind(self, kind: NodeKind) -> list[GraphNode]:
+        return [
+            GraphNode(**data)
+            for _, data in self._g.nodes(data=True)
+            if data.get("kind") == kind.value and "id" in data
+        ]
+
+    def edges_by_kind(self, kind: EdgeKind) -> list[GraphEdge]:
+        return [
+            GraphEdge(**data)
+            for _, _, data in self._g.edges(data=True)
+            if data.get("kind") == kind.value and "source" in data
+        ]
+
+    def neighbors(self, node_id: str, edge_kinds: set[EdgeKind] | None = None, direction: str = "both") -> list[str]:
+        result: set[str] = set()
+        if direction in ("out", "both"):
+            for _, target, data in self._g.out_edges(node_id, data=True):
+                if edge_kinds is None or EdgeKind(data.get("kind", "")) in edge_kinds:
+                    result.add(target)
+        if direction in ("in", "both"):
+            for source, _, data in self._g.in_edges(node_id, data=True):
+                if edge_kinds is None or EdgeKind(data.get("kind", "")) in edge_kinds:
+                    result.add(source)
+        return sorted(result)
+
+    def find_cycles(self, limit: int = 100) -> list[list[str]]:
+        cycles: list[list[str]] = []
+        for cycle in nx.simple_cycles(self._g):
+            cycles.append(cycle)
+            if len(cycles) >= limit:
+                break
+        return cycles
+
+    def shortest_path(self, source: str, target: str) -> list[str] | None:
+        try:
+            return nx.shortest_path(self._g, source, target)
+        except nx.NetworkXNoPath:
+            return None
+
+    def subgraph(self, node_ids: set[str]) -> NetworkXBackend:
+        new_backend = NetworkXBackend()
+        sub = self._g.subgraph(node_ids)
+        new_backend._g = nx.MultiDiGraph(sub)
+        return new_backend
+
+    def update_node_properties(self, node_id: str, properties: dict[str, Any]) -> None:
+        if node_id in self._g:
+            data = self._g.nodes[node_id]
+            props = data.get("properties", {})
+            props.update(properties)
+            data["properties"] = props
+
+    def close(self) -> None:
+        pass  # In-memory, nothing to close

src/code_intelligence/graph/query.py

@@ -162,11 +162,11 @@ def execute(self) -> GraphStore:
                 if spec.direction != "both":
                     # For directional focus, do a BFS in the specified direction
                     focused_ids.update(
-                        nid for nid, _ in ego_store.graph.nodes(data=True)
+                        n.id for n in ego_store.all_nodes()
                     )
                 else:
                     focused_ids.update(
-                        nid for nid, _ in ego_store.graph.nodes(data=True)
+                        n.id for n in ego_store.all_nodes()
                     )
             store = store.subgraph(focused_ids)