Add an optional Neo4j graph read model behind a default-off flag

.env.example

@@ -49,6 +49,22 @@
 # insert: text-embedding-3-small is 1536, text-embedding-3-large is 3072.
 # OPENAI_EMBEDDING_DIMENSIONS=1536
 
+# -----------------------
+# Neo4j graph read model (optional)
+# -----------------------
+# Off by default. SQL (code_dependencies) stays authoritative; Neo4j is a projection
+# rebuilt at index time, and reads fall back to SQL when it is unreachable.
+# Enables transitive traversals the SQL path cannot do: blast radius, shortest path
+# between two files, and import-cycle detection.
+#
+# Do NOT use AuraDB Free for a public demo: it auto-pauses after 72h idle and a paused
+# instance's hostname stops resolving, so the graph silently degrades to SQL.
+# NEO4J_ENABLED=false
+# NEO4J_URI=bolt://localhost:7687
+# NEO4J_USER=neo4j
+# NEO4J_PASSWORD=
+# NEO4J_DATABASE=
+
 # -----------------------
 # Embedding Providers
 # -----------------------
@@ -62,7 +78,10 @@
 # OPENAI_EMBEDDING_RATE_LIMIT_MAX_RETRIES=6
 # OPENAI_EMBEDDING_RATE_LIMIT_BASE_BACKOFF_SECONDS=1.0
 # OPENAI_EMBEDDING_RATE_LIMIT_MAX_BACKOFF_SECONDS=30.0
-# LOCAL_EMBEDDING_MODEL=nomic-ai/nomic-embed-text-v1.5
+# Ollama TAG (as in `ollama pull <tag>`), never a HuggingFace repo id. A HF id such as
+# nomic-ai/nomic-embed-text-v1.5 404s on every request, and fail-open then lands every
+# chunk in the index as a zero vector -- which scores every chunk identically.
+# LOCAL_EMBEDDING_MODEL=nomic-embed-text
 
 # -----------------------
 # GitHub (optional, for private repos)

apps/api/requirements.txt

@@ -46,6 +46,9 @@ tree-sitter-ruby>=0.21.0
 # HTTP Client
 httpx>=0.27.0
 
+# Graph read model (optional; only imported when NEO4J_ENABLED=true)
+neo4j>=5.28
+
 # Utilities
 python-dotenv>=1.0.0
 cachetools>=5.3.0

apps/api/src/api/routes/repos.py

@@ -17,7 +17,7 @@
     is_demo_mode,
 )
 from src.core.github.repo_manager import RepoManager
-from src.dependencies import get_db, get_session_factory, get_vector_store
+from src.dependencies import get_db, get_graph_store, get_session_factory, get_vector_store
 from src.models.database import IndexingStatus, Repository
 from src.models.schemas import RepoCreate, RepoListResponse, RepoResponse
 from src.services.indexing_service import IndexingService
@@ -202,7 +202,16 @@ async def delete_repository(
     # Delete from vector store
     await vector_store.delete_collection(repo_id)
 
-    # Delete from database
+    # Remove the Neo4j projection before the SQL rows go, so a failure here leaves the
+    # authoritative data intact and retryable rather than orphaning a subgraph.
+    graph_store = get_graph_store()
+    if graph_store is not None:
+        try:
+            await graph_store.delete_repository(repo_id)
+        except Exception as exc:
+            logger.warning("Failed to delete Neo4j subgraph for %s: %s", repo_id, exc)
+
+    # Delete from database (cascades to files, chunks, dependencies and chat sessions)
     db.delete(repo)
     db.commit()
 

apps/api/src/config.py

@@ -30,6 +30,8 @@
     "repos_dir",
     "vector_db_type",
     "azure_openai_tokenizer_model",
+    "neo4j_uri",
+    "neo4j_user",
 )
 
 
@@ -231,6 +233,22 @@ class Settings(BaseSettings):
     demo_quiz_requests: int = 8
     demo_quiz_window_seconds: int = 60
 
+    # Neo4j graph read model (optional)
+    #
+    # Off by default. code_dependencies (SQL) stays authoritative; this is a projection
+    # rebuilt at index time, and every read falls back to SQL when Neo4j is unreachable,
+    # so enabling it cannot take the graph endpoint down.
+    #
+    # Do not point this at AuraDB Free for anything public: a Free instance auto-pauses
+    # after 72 hours idle and a paused instance's hostname stops resolving, so the graph
+    # silently falls back to SQL until someone resumes it by hand.
+    neo4j_enabled: bool = False
+    neo4j_uri: str = "bolt://localhost:7687"
+    neo4j_user: str = "neo4j"
+    neo4j_password: Optional[str] = None
+    neo4j_database: Optional[str] = None  # None uses the server default
+    neo4j_max_traversal_hops: int = 5
+
     # Learning V2 controls
     learning_v2_enabled: bool = False
     learning_cache_ttl_days: int = 7

apps/api/src/core/graph/__init__.py

@@ -0,0 +1 @@
+"""Graph read model (Neo4j). Optional: disabled unless neo4j_enabled is set."""

apps/api/src/core/graph/neo4j_store.py

@@ -0,0 +1,224 @@
+"""
+Neo4j read model for the repository dependency graph.
+
+WHY THIS EXISTS
+The graph questions that matter for understanding a codebase are traversals:
+"what breaks if I change this file", "how does auth reach the database", "are there
+import cycles". Those are transitive, and the Python implementation cannot answer them
+-- hops is hard-capped at 2 in the API because each hop rescans the entire edge list, so
+depth is quadratic in Python and a single variable-length pattern in Cypher.
+
+WHAT IT DOES NOT DO
+This is a read model, not a source of truth. code_dependencies (SQL) remains
+authoritative; this is projected from it at index time and can be rebuilt at any point
+by re-indexing. Every read falls back to the SQL path when Neo4j is unavailable, so
+enabling this cannot take the graph endpoint down.
+
+Degree and centrality come from COUNT {} subqueries rather than Graph Data Science
+deliberately: the in-database GDS plugin is AuraDB Professional and above, and Aura
+Graph Analytics sessions are an offline batch shape (2GB, one concurrent session,
+30-minute TTL) that does not fit a synchronous request.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional, Sequence
+
+logger = logging.getLogger(__name__)
+
+# Applied once at startup. MERGE on (repo_id, path) is the hot write pattern, so the
+# uniqueness constraints are also the indexes that make it fast -- without them every
+# MERGE degrades to a label scan.
+SCHEMA_STATEMENTS: Sequence[str] = (
+    "CREATE CONSTRAINT file_unique IF NOT EXISTS "
+    "FOR (f:File) REQUIRE (f.repo_id, f.path) IS UNIQUE",
+    "CREATE CONSTRAINT module_unique IF NOT EXISTS "
+    "FOR (m:Module) REQUIRE (m.repo_id, m.key) IS UNIQUE",
+    "CREATE INDEX file_repo IF NOT EXISTS FOR (f:File) ON (f.repo_id)",
+    "CREATE INDEX file_module IF NOT EXISTS FOR (f:File) ON (f.repo_id, f.module_key)",
+)
+
+# Batched so a large repository does not build one enormous transaction.
+_INGEST_BATCH = 500
+
+_MERGE_FILES = """
+UNWIND $rows AS row
+MERGE (f:File {repo_id: $repo_id, path: row.path})
+SET f.filename    = row.filename,
+    f.extension   = row.extension,
+    f.language    = row.language,
+    f.loc         = row.loc,
+    f.module_key  = row.module_key
+"""
+
+_MERGE_EDGES = """
+UNWIND $rows AS row
+MATCH (s:File {repo_id: $repo_id, path: row.source})
+MATCH (t:File {repo_id: $repo_id, path: row.target})
+MERGE (s)-[r:IMPORTS {relation: row.relation}]->(t)
+SET r.weight = row.weight, r.confidence = row.confidence
+"""
+
+# COUNT {} rather than GDS degreeCentrality -- see the module docstring.
+_READ_NODES_WITH_DEGREE = """
+MATCH (f:File {repo_id: $repo_id})
+RETURN f.path AS path,
+       f.language AS language,
+       f.loc AS loc,
+       f.module_key AS module_key,
+       COUNT { (f)-[:IMPORTS]->(:File) } AS out_degree,
+       COUNT { (f)<-[:IMPORTS]-(:File) } AS in_degree
+"""
+
+_READ_EDGES = """
+MATCH (s:File {repo_id: $repo_id})-[r:IMPORTS]->(t:File {repo_id: $repo_id})
+RETURN s.path AS source, t.path AS target, r.relation AS relation,
+       r.weight AS weight, r.confidence AS confidence
+"""
+
+# The capability the Python path structurally cannot provide: arbitrary-depth
+# traversal. $hops is interpolated rather than parameterised because Cypher does not
+# allow a parameter inside a variable-length pattern bound; it is coerced to a bounded
+# int by the caller before it reaches here.
+_REACHABILITY = """
+MATCH path = (s:File {repo_id: $repo_id, path: $path})-[:IMPORTS*1..%(hops)d]->(t:File)
+RETURN DISTINCT t.path AS path, length(path) AS distance
+ORDER BY distance, path
+"""
+
+_BLAST_RADIUS = """
+MATCH path = (s:File {repo_id: $repo_id, path: $path})<-[:IMPORTS*1..%(hops)d]-(t:File)
+RETURN DISTINCT t.path AS path, length(path) AS distance
+ORDER BY distance, path
+"""
+
+_SHORTEST_PATH = """
+MATCH (a:File {repo_id: $repo_id, path: $from_path}),
+      (b:File {repo_id: $repo_id, path: $to_path}),
+      p = shortestPath((a)-[:IMPORTS*..%(max_hops)d]->(b))
+RETURN [n IN nodes(p) | n.path] AS path_nodes, length(p) AS distance
+"""
+
+# Import cycles: a real code-health signal the SQL path cannot express at all.
+_CYCLES = """
+MATCH (f:File {repo_id: $repo_id})
+MATCH p = (f)-[:IMPORTS*2..%(max_len)d]->(f)
+RETURN [n IN nodes(p) | n.path] AS cycle, length(p) AS size
+ORDER BY size, cycle
+LIMIT $limit
+"""
+
+_DELETE_REPO = """
+MATCH (n {repo_id: $repo_id})
+CALL (n) { DETACH DELETE n } IN TRANSACTIONS OF 1000 ROWS
+"""
+
+
+class Neo4jGraphStore:
+    """
+    Thin async wrapper around the Neo4j driver.
+
+    Holds no state beyond the driver, so it is safe to construct per request; the driver
+    itself is a long-lived singleton (see dependencies.get_graph_driver) because it owns
+    the connection pool.
+    """
+
+    def __init__(self, driver, database: Optional[str] = None):
+        self._driver = driver
+        self._database = database
+
+    async def _run(self, query: str, **params) -> List[Dict[str, Any]]:
+        records, _, _ = await self._driver.execute_query(
+            query, database_=self._database, **params
+        )
+        return [dict(r) for r in records]
+
+    # --- lifecycle ---------------------------------------------------------------
+
+    async def verify(self) -> bool:
+        """True when the server is reachable and authenticated."""
+        try:
+            await self._driver.verify_connectivity()
+            return True
+        except Exception as exc:
+            logger.warning("Neo4j connectivity check failed: %s", exc)
+            return False
+
+    async def ensure_schema(self) -> None:
+        """Idempotent; every statement is IF NOT EXISTS."""
+        for statement in SCHEMA_STATEMENTS:
+            await self._run(statement)
+
+    # --- ingest ------------------------------------------------------------------
+
+    async def sync_repository(
+        self,
+        repo_id: str,
+        files: Sequence[Dict[str, Any]],
+        edges: Sequence[Dict[str, Any]],
+    ) -> Dict[str, int]:
+        """
+        Project a repository's files and edges into the graph.
+
+        Replaces rather than merges: the previous subgraph is deleted first, because a
+        MERGE-only sync would leave edges for files that no longer exist and silently
+        accumulate a graph that no commit ever had.
+        """
+        await self.delete_repository(repo_id)
+
+        for start in range(0, len(files), _INGEST_BATCH):
+            await self._run(
+                _MERGE_FILES, repo_id=repo_id, rows=list(files[start:start + _INGEST_BATCH])
+            )
+        for start in range(0, len(edges), _INGEST_BATCH):
+            await self._run(
+                _MERGE_EDGES, repo_id=repo_id, rows=list(edges[start:start + _INGEST_BATCH])
+            )
+
+        logger.info(
+            "Synced %d files and %d edges into Neo4j for repo %s",
+            len(files), len(edges), repo_id,
+        )
+        return {"files": len(files), "edges": len(edges)}
+
+    async def delete_repository(self, repo_id: str) -> None:
+        """Remove a repository's subgraph. Called on re-sync and on repo deletion."""
+        await self._run(_DELETE_REPO, repo_id=repo_id)
+
+    # --- reads -------------------------------------------------------------------
+
+    async def nodes_with_degree(self, repo_id: str) -> List[Dict[str, Any]]:
+        return await self._run(_READ_NODES_WITH_DEGREE, repo_id=repo_id)
+
+    async def edges(self, repo_id: str) -> List[Dict[str, Any]]:
+        return await self._run(_READ_EDGES, repo_id=repo_id)
+
+    # --- traversals the SQL path cannot answer -----------------------------------
+
+    @staticmethod
+    def _bounded(value: int, lo: int, hi: int) -> int:
+        return max(lo, min(hi, int(value)))
+
+    async def reachable_from(self, repo_id: str, path: str, hops: int = 3) -> List[Dict[str, Any]]:
+        """What this file transitively imports."""
+        q = _REACHABILITY % {"hops": self._bounded(hops, 1, 10)}
+        return await self._run(q, repo_id=repo_id, path=path)
+
+    async def blast_radius(self, repo_id: str, path: str, hops: int = 3) -> List[Dict[str, Any]]:
+        """What transitively imports this file -- i.e. what a change here can affect."""
+        q = _BLAST_RADIUS % {"hops": self._bounded(hops, 1, 10)}
+        return await self._run(q, repo_id=repo_id, path=path)
+
+    async def shortest_path(
+        self, repo_id: str, from_path: str, to_path: str, max_hops: int = 10
+    ) -> Optional[Dict[str, Any]]:
+        q = _SHORTEST_PATH % {"max_hops": self._bounded(max_hops, 1, 15)}
+        rows = await self._run(q, repo_id=repo_id, from_path=from_path, to_path=to_path)
+        return rows[0] if rows else None
+
+    async def import_cycles(
+        self, repo_id: str, max_length: int = 6, limit: int = 20
+    ) -> List[Dict[str, Any]]:
+        q = _CYCLES % {"max_len": self._bounded(max_length, 2, 10)}
+        return await self._run(q, repo_id=repo_id, limit=max(1, int(limit)))

apps/api/src/dependencies.py

@@ -122,3 +122,40 @@ def get_redis_client():
 def get_chat_cache() -> ChatCache:
     """Get chat cache service with Redis+memory fallback."""
     return ChatCache(redis_client=get_redis_client())
+
+
+@lru_cache()
+def get_graph_driver():
+    """
+    Long-lived Neo4j async driver, or None when the graph read model is disabled.
+
+    The driver owns a connection pool, so it must be a singleton and must be closed on
+    shutdown (see main.py lifespan). Returns None rather than raising so that every
+    caller can treat "no graph store" as an ordinary fallback path.
+    """
+    if not settings.neo4j_enabled:
+        return None
+    if not settings.neo4j_password:
+        logger.warning("NEO4J_ENABLED is set but NEO4J_PASSWORD is empty; graph store disabled")
+        return None
+
+    try:
+        from neo4j import AsyncGraphDatabase
+
+        return AsyncGraphDatabase.driver(
+            settings.neo4j_uri,
+            auth=(settings.neo4j_user, settings.neo4j_password),
+        )
+    except Exception as exc:
+        logger.warning("Neo4j driver unavailable, falling back to SQL graph: %s", exc)
+        return None
+
+
+def get_graph_store():
+    """Neo4jGraphStore bound to the shared driver, or None when disabled."""
+    driver = get_graph_driver()
+    if driver is None:
+        return None
+    from src.core.graph.neo4j_store import Neo4jGraphStore
+
+    return Neo4jGraphStore(driver, database=settings.neo4j_database)