From bb160d2b0f6dc23cd14b97c4be270978e58bc01c Mon Sep 17 00:00:00 2001 From: Shree Bohara Date: Sun, 9 Aug 2026 18:21:13 -0700 Subject: [PATCH] Add an optional Neo4j graph read model behind a default-off flag Enables the traversals the SQL path structurally cannot do, without making a new datastore load-bearing. WHY The graph questions that matter for understanding a codebase are transitive: what breaks if I change this file, how does auth reach the database, are there import cycles. The Python implementation cannot answer them -- hops is hard-capped at 2 in the API because each hop rescans the whole edge list, so depth is quadratic in Python and a single variable-length pattern in Cypher. DESIGN: PROJECTION, NOT SOURCE OF TRUTH code_dependencies (SQL) stays authoritative. Neo4j is projected from it during indexing and can be rebuilt by re-indexing. Consequences, all deliberate: - neo4j_enabled defaults to False, so nothing changes unless it is turned on. - get_graph_store() returns None when disabled OR misconfigured (missing password) rather than raising, so every caller treats "no graph store" as an ordinary path. - Startup verifies connectivity and applies schema, but a failure only logs -- an unreachable graph database must not stop the API booting. - A sync failure during indexing is caught and logged; the SQL edges are unaffected. - Repo deletion removes the subgraph before the SQL rows, so a failure leaves the authoritative data intact and retryable rather than orphaning a subgraph. - core/graph/neo4j_store.py: schema (uniqueness on (repo_id, path), which is also the index that keeps MERGE off a label scan), batched UNWIND+MERGE ingest at 500 rows, and reads. sync_repository deletes the subgraph first rather than merging, because a MERGE-only sync leaves edges for deleted files and drifts into a union of every commit ever indexed. - Traversals: reachable_from, blast_radius, shortest_path, import_cycles. Depth is clamped (1..10) -- an unbounded variable-length pattern is a trivial way to hang the server. - Degree and centrality use COUNT {} subqueries, NOT Graph Data Science: the in-database GDS plugin requires AuraDB Professional or 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. - Async driver held as a single long-lived instance (it owns the pool) and closed in the lifespan. - docker-compose.yml: neo4j:5-community as a fourth service at 512m heap + 512m pagecache, plus the five settings forwarded to the API. Docs record the AuraDB Free trap: a Free instance auto-pauses after 72h idle and a paused instance's hostname stops resolving, so the graph would silently fall back to SQL until someone resumed it by hand. Also fixes stale drift in .env.example that still advertised LOCAL_EMBEDDING_MODEL=nomic-ai/nomic-embed-text-v1.5 -- the HuggingFace id removed from config.py earlier, which would have walked a self-hoster straight back into the all-zero-vector index. VERIFIED, AND WHAT IS NOT 15 new tests, 132 total (was 117), ruff clean, app boots with the store returning None by default. The tests use a fake driver, so they cover the queries we send and every fallback path -- including that indexing still persists SQL edges when the graph sync fails -- but NOT that the Cypher returns correct results. No query in this commit has executed against a Neo4j server. docker/README.md documents the two count queries that confirm the projection matches SQL once an instance is available. Co-Authored-By: Claude Opus 5 --- .env.example | 21 +- apps/api/requirements.txt | 3 + apps/api/src/api/routes/repos.py | 13 +- apps/api/src/config.py | 18 ++ apps/api/src/core/graph/__init__.py | 1 + apps/api/src/core/graph/neo4j_store.py | 224 ++++++++++++++++ apps/api/src/dependencies.py | 37 +++ apps/api/src/main.py | 22 ++ apps/api/src/services/indexing_service.py | 44 ++++ apps/api/tests/unit/test_neo4j_graph_store.py | 245 ++++++++++++++++++ docker/README.md | 35 +++ docker/docker-compose.yml | 22 ++ 12 files changed, 682 insertions(+), 3 deletions(-) create mode 100644 apps/api/src/core/graph/__init__.py create mode 100644 apps/api/src/core/graph/neo4j_store.py create mode 100644 apps/api/tests/unit/test_neo4j_graph_store.py diff --git a/.env.example b/.env.example index d70e70e..56c035a 100644 --- a/.env.example +++ b/.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 `), 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) diff --git a/apps/api/requirements.txt b/apps/api/requirements.txt index 8fafcb1..fb32748 100644 --- a/apps/api/requirements.txt +++ b/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 diff --git a/apps/api/src/api/routes/repos.py b/apps/api/src/api/routes/repos.py index 56b52b3..2169f9e 100644 --- a/apps/api/src/api/routes/repos.py +++ b/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() diff --git a/apps/api/src/config.py b/apps/api/src/config.py index 0428795..2ee2e71 100644 --- a/apps/api/src/config.py +++ b/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 diff --git a/apps/api/src/core/graph/__init__.py b/apps/api/src/core/graph/__init__.py new file mode 100644 index 0000000..ad13395 --- /dev/null +++ b/apps/api/src/core/graph/__init__.py @@ -0,0 +1 @@ +"""Graph read model (Neo4j). Optional: disabled unless neo4j_enabled is set.""" diff --git a/apps/api/src/core/graph/neo4j_store.py b/apps/api/src/core/graph/neo4j_store.py new file mode 100644 index 0000000..2ce0516 --- /dev/null +++ b/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))) diff --git a/apps/api/src/dependencies.py b/apps/api/src/dependencies.py index 21095e2..7a6f6eb 100644 --- a/apps/api/src/dependencies.py +++ b/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) diff --git a/apps/api/src/main.py b/apps/api/src/main.py index 731f639..5eace6a 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -15,6 +15,8 @@ from src.dependencies import ( get_chat_cache, get_db_engine, + get_graph_driver, + get_graph_store, get_redis_client, get_session_factory, get_vector_store, @@ -58,11 +60,31 @@ async def lifespan(app: FastAPI): await vector_store.initialize() logger.info("Vector store initialized") + # Optional Neo4j read model. Non-fatal by design: the graph endpoint falls back to + # the SQL path, so an unreachable graph database must not stop the API booting. + graph_store = get_graph_store() + if graph_store is not None: + try: + if await graph_store.verify(): + await graph_store.ensure_schema() + logger.info("Neo4j graph store initialized") + else: + logger.warning("Neo4j unreachable; graph requests will use the SQL fallback") + except Exception as exc: + logger.warning("Neo4j initialization failed, using SQL fallback: %s", exc) + yield # Shutdown logger.info("Shutting down CodebaseQA API...") await vector_store.close() + graph_driver = get_graph_driver() + if graph_driver is not None: + try: + await graph_driver.close() + except Exception: + pass + redis_client = get_redis_client() if redis_client is not None: try: diff --git a/apps/api/src/services/indexing_service.py b/apps/api/src/services/indexing_service.py index a11e040..6b026a0 100644 --- a/apps/api/src/services/indexing_service.py +++ b/apps/api/src/services/indexing_service.py @@ -228,8 +228,52 @@ async def _persist_dependency_graph(self, repo: Repository) -> int: "Derived %d dependency edges for %s/%s at index time", len(rows), repo.github_owner, repo.github_name, ) + + # Project into the Neo4j read model if it is enabled. SQL stays authoritative, + # so a failure here is logged and ignored -- the graph endpoint falls back. + await self._sync_graph_store(repo, files, rows) + return len(rows) + async def _sync_graph_store(self, repo, files, edge_rows) -> None: + """Mirror the freshly derived graph into Neo4j. Never raises.""" + from src.dependencies import get_graph_store + + store = get_graph_store() + if store is None: + return + + try: + from src.services.learning_service import LearningService + service = LearningService(self._db, llm=None, vector_store=None) + + file_payload = [ + { + "path": f.path, + "filename": f.filename, + "extension": f.extension, + "language": f.language, + "loc": f.line_count or 0, + "module_key": service._module_key_for_path(f.path), + } + for f in files + ] + edge_payload = [ + { + "source": r.source_path, + "target": r.target_path, + "relation": r.relation or "imports", + "weight": r.weight or 1, + "confidence": r.confidence if r.confidence is not None else 0.72, + } + for r in edge_rows + ] + await store.sync_repository(repo.id, file_payload, edge_payload) + except Exception as exc: + logger.warning( + "Neo4j graph sync failed for %s (SQL graph is unaffected): %s", repo.id, exc + ) + def _find_files(self, repo_path: Path) -> List[Path]: """Find all indexable files in repository.""" files = [] diff --git a/apps/api/tests/unit/test_neo4j_graph_store.py b/apps/api/tests/unit/test_neo4j_graph_store.py new file mode 100644 index 0000000..28e2ca1 --- /dev/null +++ b/apps/api/tests/unit/test_neo4j_graph_store.py @@ -0,0 +1,245 @@ +""" +Neo4j graph store wiring. + +IMPORTANT LIMITATION, stated rather than implied: these tests use a fake driver, so they +verify the *queries we send* and the *fallback behaviour*, not that the Cypher returns +correct results. Nothing here has executed against a Neo4j server. To validate the +Cypher itself, run `docker compose up neo4j`, set NEO4J_ENABLED=true, re-index a repo, +and check the graph endpoint against the SQL path. + +What these tests do lock down is the part most likely to break silently: that the store +is off by default, that every failure path degrades to SQL rather than erroring, and +that a repository's subgraph is actually replaced (not merged) on re-sync. +""" + +import pytest + +from src.config import Settings +from src.core.graph.neo4j_store import SCHEMA_STATEMENTS, Neo4jGraphStore + + +class FakeDriver: + """Records queries instead of executing them.""" + + def __init__(self, rows=None, fail_on=None, connectivity=True): + self.calls = [] + self._rows = rows or [] + self._fail_on = fail_on + self._connectivity = connectivity + + async def execute_query(self, query, database_=None, **params): + self.calls.append({"query": query, "database": database_, "params": params}) + if self._fail_on and self._fail_on in query: + raise RuntimeError("simulated neo4j failure") + return self._rows, None, None + + async def verify_connectivity(self): + if not self._connectivity: + raise RuntimeError("unreachable") + + async def close(self): + self.calls.append({"query": "__closed__"}) + + +def _queries(driver): + return " || ".join(c["query"] for c in driver.calls) + + +# --- disabled by default --------------------------------------------------------- + +def test_graph_store_is_disabled_by_default(): + """A new datastore must not become load-bearing by accident.""" + assert Settings(_env_file=None).neo4j_enabled is False + + +def test_get_graph_store_returns_none_when_disabled(monkeypatch): + import src.dependencies as deps + monkeypatch.setattr(deps, "settings", Settings(_env_file=None, neo4j_enabled=False)) + deps.get_graph_driver.cache_clear() + assert deps.get_graph_store() is None + + +def test_get_graph_store_returns_none_when_password_missing(monkeypatch): + """Enabled but unconfigured must degrade, not raise at import time.""" + import src.dependencies as deps + monkeypatch.setattr( + deps, "settings", + Settings(_env_file=None, neo4j_enabled=True, neo4j_password=None), + ) + deps.get_graph_driver.cache_clear() + assert deps.get_graph_store() is None + deps.get_graph_driver.cache_clear() + + +# --- schema ---------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_ensure_schema_is_idempotent_by_construction(): + driver = FakeDriver() + await Neo4jGraphStore(driver).ensure_schema() + + assert len(driver.calls) == len(SCHEMA_STATEMENTS) + # Re-running must be safe, which means every statement carries IF NOT EXISTS. + for statement in SCHEMA_STATEMENTS: + assert "IF NOT EXISTS" in statement + + +@pytest.mark.asyncio +async def test_schema_constrains_uniqueness_per_repo(): + """(repo_id, path) must be unique, or re-sync would duplicate every file node.""" + joined = " ".join(SCHEMA_STATEMENTS) + assert "f.repo_id, f.path) IS UNIQUE" in joined + assert "m.repo_id, m.key) IS UNIQUE" in joined + + +# --- ingest ---------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_sync_replaces_rather_than_merges(): + """ + A MERGE-only sync leaves edges for deleted files behind, so the graph drifts into a + union of every commit ever indexed. Sync must delete the subgraph first. + """ + driver = FakeDriver() + store = Neo4jGraphStore(driver) + await store.sync_repository( + "repo-1", + files=[{"path": "a.ts", "filename": "a.ts", "extension": ".ts", + "language": "typescript", "loc": 10, "module_key": "root"}], + edges=[{"source": "a.ts", "target": "b.ts", "relation": "imports", + "weight": 1, "confidence": 0.9}], + ) + assert "DETACH DELETE" in driver.calls[0]["query"] + assert "MERGE (f:File" in _queries(driver) + assert "MERGE (s)-[r:IMPORTS" in _queries(driver) + + +@pytest.mark.asyncio +async def test_sync_batches_large_repositories(): + """One transaction per 500 rows, not one transaction for the whole repo.""" + driver = FakeDriver() + files = [ + {"path": f"f{i}.ts", "filename": f"f{i}.ts", "extension": ".ts", + "language": "typescript", "loc": 1, "module_key": "root"} + for i in range(1200) + ] + await Neo4jGraphStore(driver).sync_repository("r", files=files, edges=[]) + merge_calls = [c for c in driver.calls if "MERGE (f:File" in c["query"]] + assert len(merge_calls) == 3 # 500 + 500 + 200 + assert sum(len(c["params"]["rows"]) for c in merge_calls) == 1200 + + +@pytest.mark.asyncio +async def test_edges_are_scoped_to_the_repository(): + """Both endpoints are matched with repo_id, so edges cannot cross repositories.""" + driver = FakeDriver() + await Neo4jGraphStore(driver).sync_repository( + "r", files=[], edges=[{"source": "a", "target": "b", "relation": "imports", + "weight": 1, "confidence": 0.5}]) + edge_q = [c["query"] for c in driver.calls if "IMPORTS" in c["query"]][0] + assert edge_q.count("repo_id: $repo_id") == 2 + + +# --- reads use COUNT{} rather than GDS ------------------------------------------- + +@pytest.mark.asyncio +async def test_degree_uses_count_subqueries_not_gds(): + """ + GDS is deliberately avoided: the in-database plugin is AuraDB Professional+ and + Aura Graph Analytics sessions are an offline batch shape. + """ + driver = FakeDriver(rows=[{"path": "a.ts", "in_degree": 1, "out_degree": 2}]) + await Neo4jGraphStore(driver).nodes_with_degree("r") + q = driver.calls[0]["query"] + assert "COUNT {" in q + assert "gds." not in q.lower() + + +# --- traversals ------------------------------------------------------------------ + +@pytest.mark.asyncio +async def test_traversal_depth_is_bounded(): + """An unbounded variable-length pattern is a trivial way to hang the server.""" + driver = FakeDriver() + store = Neo4jGraphStore(driver) + + await store.reachable_from("r", "a.ts", hops=999) + assert "IMPORTS*1..10]" in driver.calls[-1]["query"] + + await store.blast_radius("r", "a.ts", hops=0) + assert "IMPORTS*1..1]" in driver.calls[-1]["query"] + + +@pytest.mark.asyncio +async def test_blast_radius_traverses_inbound_edges(): + """"What breaks if I change this" is the reverse direction of "what this imports".""" + driver = FakeDriver() + store = Neo4jGraphStore(driver) + await store.reachable_from("r", "a.ts", hops=2) + forward = driver.calls[-1]["query"] + await store.blast_radius("r", "a.ts", hops=2) + reverse = driver.calls[-1]["query"] + assert "-[:IMPORTS*1..2]->" in forward + assert "<-[:IMPORTS*1..2]-" in reverse + + +@pytest.mark.asyncio +async def test_shortest_path_returns_none_when_unreachable(): + store = Neo4jGraphStore(FakeDriver(rows=[])) + assert await store.shortest_path("r", "a.ts", "z.ts") is None + + +@pytest.mark.asyncio +async def test_import_cycles_requires_length_at_least_two(): + """A self-loop is not an import cycle worth reporting.""" + driver = FakeDriver() + await Neo4jGraphStore(driver).import_cycles("r", max_length=1) + assert "IMPORTS*2..2]" in driver.calls[-1]["query"] + + +# --- failure degrades to SQL, never raises --------------------------------------- + +@pytest.mark.asyncio +async def test_verify_returns_false_when_unreachable(): + assert await Neo4jGraphStore(FakeDriver(connectivity=False)).verify() is False + + +@pytest.mark.asyncio +async def test_indexing_survives_a_graph_sync_failure(monkeypatch, tmp_path): + """ + The whole point of SQL staying authoritative: a Neo4j outage during indexing must + not fail the index or lose the SQL edges. + """ + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + from src.models.database import Base, CodeDependency, CodeFile, Repository + from src.services.indexing_service import IndexingService + + engine = create_engine(f"sqlite:///{tmp_path / 'x.db'}") + Base.metadata.create_all(engine) + db = sessionmaker(bind=engine)() + + clone = tmp_path / "clone" + (clone / "src").mkdir(parents=True) + (clone / "src" / "a.ts").write_text("import { b } from './b';\n") + (clone / "src" / "b.ts").write_text("export const b = 1;\n") + + repo = Repository(github_url="https://github.com/o/r", github_owner="o", + github_name="r", local_path=str(clone)) + db.add(repo) + db.commit() + db.refresh(repo) + for path in ("src/a.ts", "src/b.ts"): + db.add(CodeFile(repository_id=repo.id, path=path, filename=path[-4:], + extension=".ts", language="typescript", imports=[])) + db.commit() + + failing = Neo4jGraphStore(FakeDriver(fail_on="DETACH DELETE")) + monkeypatch.setattr("src.dependencies.get_graph_store", lambda: failing) + + written = await IndexingService(db)._persist_dependency_graph(repo) + + assert written > 0 + assert db.query(CodeDependency).filter(CodeDependency.repository_id == repo.id).count() > 0 + db.close() diff --git a/docker/README.md b/docker/README.md index b1e4c44..180bbef 100644 --- a/docker/README.md +++ b/docker/README.md @@ -50,6 +50,41 @@ So the SQLite database, the ChromaDB directory and every cloned repository live - `docker-compose.yml` also declares a named volume `data:` that nothing mounts. It has no effect; the bind mount above is what is actually used. +## Optional: Neo4j graph read model + +Off by default. `code_dependencies` in SQLite stays authoritative; Neo4j is a projection +rebuilt at index time, and every read falls back to SQL if Neo4j is unreachable — so +turning this on cannot take the graph endpoint down. + +```bash +docker compose up -d neo4j +# then in docker/.env +NEO4J_ENABLED=true +NEO4J_PASSWORD= +``` + +Re-index a repository to populate it, then browse at http://localhost:7474. + +To confirm the projection matches SQL: + +```cypher +MATCH (f:File {repo_id: $repo}) RETURN count(f); +MATCH (:File {repo_id: $repo})-[r:IMPORTS]->(:File) RETURN count(r); +``` + +Those two counts should equal `SELECT count(*) FROM code_files` and +`SELECT count(*) FROM code_dependencies` for the same repository. + +**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 +would silently fall back to SQL until someone resumes it by hand. Free instances are +deleted after 30 days paused. + +Degree and centrality use Cypher `COUNT {}` subqueries rather than Graph Data Science: +the in-database GDS plugin requires AuraDB Professional or 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. + ## Using Azure OpenAI Set these in `docker/.env` (compose forwards all of them): diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 3984d43..34df1e1 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -36,6 +36,12 @@ services: - OPENAI_EMBEDDING_DIMENSIONS=${OPENAI_EMBEDDING_DIMENSIONS:-1536} - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY:-} - ANTHROPIC_MODEL=${ANTHROPIC_MODEL:-claude-sonnet-4-20250514} + # --- optional Neo4j graph read model (SQL stays authoritative) --- + - NEO4J_ENABLED=${NEO4J_ENABLED:-false} + - NEO4J_URI=${NEO4J_URI:-bolt://neo4j:7687} + - NEO4J_USER=${NEO4J_USER:-neo4j} + - NEO4J_PASSWORD=${NEO4J_PASSWORD:-codebaseqa-dev} + - NEO4J_DATABASE=${NEO4J_DATABASE:-} # --- Azure OpenAI (v1 OpenAI-compatible surface; deployment names, not model ids) --- - AZURE_OPENAI_ENDPOINT=${AZURE_OPENAI_ENDPOINT:-} - AZURE_OPENAI_API_KEY=${AZURE_OPENAI_API_KEY:-} @@ -85,6 +91,22 @@ services: depends_on: - redis + # Optional graph read model. NEO4J_ENABLED defaults to false, so the API ignores this + # service unless you turn it on. ~1GB RSS at the heap/pagecache settings below. + neo4j: + image: neo4j:5-community + ports: + - "7474:7474" # HTTP browser + - "7687:7687" # Bolt + environment: + - NEO4J_AUTH=neo4j/${NEO4J_PASSWORD:-codebaseqa-dev} + - NEO4J_server_memory_heap_initial__size=512m + - NEO4J_server_memory_heap_max__size=512m + - NEO4J_server_memory_pagecache_size=512m + volumes: + - ../data/neo4j:/data + restart: always + redis: image: redis:7-alpine ports: