From 6aa3d81405c9443d44092b0e6e01181e2e597add Mon Sep 17 00:00:00 2001 From: Shree Bohara Date: Sun, 9 Aug 2026 13:05:33 -0700 Subject: [PATCH] Derive dependency edges at index time instead of on every graph request Prerequisite for putting the graph in Neo4j, and worth landing on its own. generate_graph re-derived the whole edge set on every cache miss, and _build_deterministic_edges did it by reading every source file off disk inside the request. Three consequences: 1. Graph latency was proportional to repository size, with blocking file I/O on the event loop. The 45-second in-process TTL cache existed to hide this, and being per-process it bought nothing across workers. 2. The graph was coupled to the clone still being present. After a redeploy, a container restart or a volume reset, derivation silently fell back to the unresolved import strings on CodeFile.imports and produced a worse graph for the same repository, with nothing indicating it had happened. 3. Every graph request paid for work whose inputs only change when the repo is re-indexed. Derivation now runs once, at the end of indexing, while the clone is guaranteed to exist, into a new code_dependencies table. The read path is a single indexed query. - database.py: CodeDependency (source_path, target_path, relation, weight, confidence), indexed on repository_id and on each endpoint, unique on (repo, source, target, relation). Added to Repository.dependencies with delete-orphan cascade -- without it, DELETE /api/repos/{id} would leave orphaned edges that union into a later re-import of the same repository. - indexing_service.py: _persist_dependency_graph runs after embedding, wrapped so a derivation failure cannot fail an otherwise good index. It reuses LearningService's existing resolution helpers rather than reimplementing them, so there is still one definition of what an edge is, and offloads the blocking file reads with asyncio.to_thread. _reset_repository_index_data now clears edges too. - learning_service.py: _load_or_derive_edges prefers persisted rows and falls back to on-demand derivation for repositories indexed before this table existed or whose derivation step failed; both self-heal on the next re-index, and the fallback logs that it ran. Edges whose endpoints are not in the requested node set are dropped -- scope filters and the node cap can remove a node while its edges remain, which would otherwise ship an edge to a node the client never received. _build_deterministic_edges is unchanged, so its existing unit tests still cover the derivation logic directly. Verified: 6 new tests, 117 total (was 111), ruff clean. The important one deletes the clone and asserts the graph is still complete from the database, while asserting that on-demand derivation returns nothing in that same state -- which is exactly the silent degradation this removes. Others cover cascade-on-delete, stale-edge clearing on re-index, and out-of-scope edge filtering. Co-Authored-By: Claude Opus 5 --- apps/api/src/models/database.py | 55 +++++- apps/api/src/models/migrations.py | 12 ++ apps/api/src/services/indexing_service.py | 83 +++++++- apps/api/src/services/learning_service.py | 63 +++++- .../tests/unit/test_dependency_persistence.py | 181 ++++++++++++++++++ 5 files changed, 390 insertions(+), 4 deletions(-) create mode 100644 apps/api/tests/unit/test_dependency_persistence.py diff --git a/apps/api/src/models/database.py b/apps/api/src/models/database.py index 0d991cb..1ed8bb6 100644 --- a/apps/api/src/models/database.py +++ b/apps/api/src/models/database.py @@ -5,7 +5,7 @@ import enum import uuid -from datetime import datetime +from datetime import datetime, timezone from sqlalchemy import ( JSON, @@ -17,6 +17,7 @@ Integer, String, Text, + UniqueConstraint, ) from sqlalchemy import Enum as SQLEnum from sqlalchemy.orm import declarative_base, relationship @@ -72,6 +73,11 @@ class Repository(Base): files = relationship("CodeFile", back_populates="repository", cascade="all, delete-orphan") chunks = relationship("CodeChunk", back_populates="repository", cascade="all, delete-orphan") chat_sessions = relationship("ChatSession", back_populates="repository", cascade="all, delete-orphan") + # Cascade matters: without it, DELETE /api/repos/{id} would leave orphaned edges + # behind, and a later re-import of the same repo would union old and new graphs. + dependencies = relationship( + "CodeDependency", back_populates="repository", cascade="all, delete-orphan" + ) __table_args__ = ( Index("ix_repositories_github", "github_owner", "github_name"), @@ -116,6 +122,53 @@ class CodeFile(Base): ) +class CodeDependency(Base): + """ + A resolved import edge between two files, derived once at index time. + + Exists because graph generation used to re-derive every edge on each cache miss by + reading every source file off disk inside the request path (see + LearningService._build_deterministic_edges). That made graph latency proportional to + repository size, put blocking file I/O on the event loop, and coupled the graph to + the clone still being present -- which breaks after a redeploy or container restart. + + Derivation now happens once per index, while the clone definitely exists, and the + read path is a single indexed query. + """ + __tablename__ = "code_dependencies" + + id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) + repository_id = Column(String(36), ForeignKey("repositories.id"), nullable=False) + + # Repo-relative paths, matching CodeFile.path so they can be joined or mapped. + source_path = Column(String(1000), nullable=False) + target_path = Column(String(1000), nullable=False) + + # How the edge was classified (_infer_relation) and how much to trust it + # (_build_deterministic_edges): relative imports score higher than bare specifiers. + relation = Column(String(50), nullable=False, default="imports") + weight = Column(Integer, default=1) + confidence = Column(Float, default=0.72) + + # The raw specifier that produced this edge, kept for debugging why an edge exists. + specifier = Column(String(500), nullable=True) + + created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc)) + + repository = relationship("Repository", back_populates="dependencies") + + __table_args__ = ( + # The read path always filters by repository first. + Index("ix_code_dependencies_repo", "repository_id"), + Index("ix_code_dependencies_repo_source", "repository_id", "source_path"), + Index("ix_code_dependencies_repo_target", "repository_id", "target_path"), + UniqueConstraint( + "repository_id", "source_path", "target_path", "relation", + name="uq_code_dependencies_edge", + ), + ) + + class CodeChunk(Base): """Represents a semantic chunk of code for embedding.""" __tablename__ = "code_chunks" diff --git a/apps/api/src/models/migrations.py b/apps/api/src/models/migrations.py index 7adeab7..88517b2 100644 --- a/apps/api/src/models/migrations.py +++ b/apps/api/src/models/migrations.py @@ -110,6 +110,18 @@ def run_pending_migrations(engine: Engine) -> List[str]: connection.execute(text("ALTER TABLE learning_syllabi ADD COLUMN expires_at DATETIME")) applied.append("learning_syllabi.expires_at") + # code_dependencies itself is created by init_db/create_all, which runs first + # (main.py). These indexes are declared on the model too, so this block only + # matters for a database that predates the table and gets it via create_all + # without the accompanying indexes -- it is idempotent either way. + connection.execute( + text( + "CREATE INDEX IF NOT EXISTS ix_code_dependencies_repo " + "ON code_dependencies (repository_id)" + ) + ) + applied.append("ix_code_dependencies_repo") + if applied: logger.info("Applied runtime migrations: %s", ", ".join(applied)) diff --git a/apps/api/src/services/indexing_service.py b/apps/api/src/services/indexing_service.py index 4f6f141..a11e040 100644 --- a/apps/api/src/services/indexing_service.py +++ b/apps/api/src/services/indexing_service.py @@ -3,6 +3,7 @@ Handles cloning, parsing, and embedding of code repositories. """ +import asyncio import hashlib import logging import os @@ -17,7 +18,13 @@ from src.core.github.repo_manager import RepoManager from src.core.parser.tree_sitter_parser import get_parser_for_file from src.dependencies import get_embedding_service, get_vector_store -from src.models.database import CodeChunk, CodeFile, IndexingStatus, Repository +from src.models.database import ( + CodeChunk, + CodeDependency, + CodeFile, + IndexingStatus, + Repository, +) logger = logging.getLogger(__name__) @@ -122,6 +129,26 @@ async def index_repository(self, repo_id: str, force_reindex: bool = False): if chunks_data: await self._embed_and_store(repo_id, chunks_data) + # Derive the dependency graph while the clone is still on disk. + # + # This used to happen inside every graph request instead, which meant + # reading every source file off disk on the event loop behind a 45-second + # cache, and it silently produced a worse graph once the clone was gone + # (after a redeploy or container restart) because it fell back to the + # unresolved import strings on CodeFile. Doing it here is the only point + # where the working tree is guaranteed to exist. + # + # Deliberately non-fatal: a graph derivation failure must not fail an + # otherwise good index, and the read path can still fall back. + try: + await self._persist_dependency_graph(repo) + except Exception as exc: + logger.warning( + "Dependency graph derivation failed for %s; the graph endpoint will " + "fall back to deriving on request: %s", + repo_id, exc, + ) + # Complete repo.status = IndexingStatus.COMPLETED repo.last_indexed_at = datetime.now(timezone.utc) @@ -141,6 +168,12 @@ async def _reset_repository_index_data(self, repo_id: str) -> None: """Clear prior SQL/vector artifacts for a repository before full re-index.""" self._db.query(CodeChunk).filter(CodeChunk.repository_id == repo_id).delete(synchronize_session=False) self._db.query(CodeFile).filter(CodeFile.repository_id == repo_id).delete(synchronize_session=False) + # Stale edges would otherwise union with the newly derived ones, since the + # uniqueness constraint is on (repo, source, target, relation) and a renamed or + # deleted file produces edges that no longer have a counterpart. + self._db.query(CodeDependency).filter( + CodeDependency.repository_id == repo_id + ).delete(synchronize_session=False) self._db.commit() try: @@ -149,6 +182,54 @@ async def _reset_repository_index_data(self, repo_id: str) -> None: except Exception as exc: logger.warning("Failed to clear existing vector collection for repo %s: %s", repo_id, exc) + async def _persist_dependency_graph(self, repo: Repository) -> int: + """ + Derive import edges once and store them in code_dependencies. + + Reuses LearningService's resolution logic rather than reimplementing it, so + there is exactly one definition of what an edge is and how it is classified. + The file reading is offloaded to a worker thread because it is blocking and + proportional to repository size -- this runs inside the indexing task, which + shares an event loop with nothing, but keeping it off the loop means the same + method is safe to call from a request handler later if that is ever wanted. + """ + from src.services.learning_service import LearningService + + files = self._db.query(CodeFile).filter(CodeFile.repository_id == repo.id).all() + if not files: + return 0 + + service = LearningService(self._db, llm=None, vector_store=None) + file_map = {f.path: f for f in files} + all_paths = set(file_map) + source_paths = sorted(all_paths) + + edges = await asyncio.to_thread( + service._build_deterministic_edges, repo, source_paths, all_paths, file_map + ) + + rows = [ + CodeDependency( + repository_id=repo.id, + source_path=edge.source, + target_path=edge.target, + relation=getattr(edge, "relation", None) or "imports", + weight=int(getattr(edge, "weight", 1) or 1), + confidence=float(getattr(edge, "confidence", 0.72) or 0.72), + ) + for edge in edges + ] + + if rows: + self._db.bulk_save_objects(rows) + self._db.commit() + + logger.info( + "Derived %d dependency edges for %s/%s at index time", + len(rows), repo.github_owner, repo.github_name, + ) + return len(rows) + def _find_files(self, repo_path: Path) -> List[Path]: """Find all indexable files in repository.""" files = [] diff --git a/apps/api/src/services/learning_service.py b/apps/api/src/services/learning_service.py index 79f5148..053775c 100644 --- a/apps/api/src/services/learning_service.py +++ b/apps/api/src/services/learning_service.py @@ -16,7 +16,13 @@ from src.core.llm.openai_llm import OpenAILLM from src.core.vectorstore.chroma_store import ChromaStore from src.models.codetour_schemas import CodeTour, CodeTourStep -from src.models.database import CodeFile, LearningLesson, LearningSyllabus, Repository +from src.models.database import ( + CodeDependency, + CodeFile, + LearningLesson, + LearningSyllabus, + Repository, +) from src.models.learning import ( CacheInfo, CodeReference, @@ -1374,7 +1380,7 @@ async def generate_graph( file_nodes_by_id: Dict[str, GraphNode] = { path: self._build_graph_node_from_file(path, file_map[path]) for path in sorted(code_paths) } - file_edges = self._build_deterministic_edges(repo, sorted(code_paths), all_paths, file_map) + file_edges = self._load_or_derive_edges(repo, sorted(code_paths), all_paths, file_map) for edge in file_edges: if edge.source in file_map and edge.source not in file_nodes_by_id: @@ -1895,6 +1901,59 @@ def _build_graph_node_from_file(self, path: str, code_file: CodeFile) -> GraphNo module_key=self._module_key_for_path(path), ) + def _load_or_derive_edges( + self, + repo: Repository, + source_paths: List[str], + all_paths: Set[str], + file_map: Dict[str, CodeFile], + ) -> List[GraphEdge]: + """ + Prefer edges persisted at index time; derive on demand only as a fallback. + + The fallback exists for two real cases: repositories indexed before + code_dependencies existed, and an index whose graph derivation step failed. Both + self-heal on the next re-index. Deriving here is the slow path -- it reads every + source file off disk inside the request -- so it is worth knowing which one ran, + hence the debug log. + """ + persisted = ( + self._db.query(CodeDependency) + .filter(CodeDependency.repository_id == repo.id) + .all() + ) + + if persisted: + known = set(all_paths) + edges = [ + GraphEdge( + source=row.source_path, + target=row.target_path, + label=row.relation or "imports", + type=row.relation or "imports", + relation=row.relation or "imports", + weight=row.weight or 1, + confidence=row.confidence if row.confidence is not None else 0.72, + ) + # A file can disappear from the graph's node set (scope filters, the + # node cap) while its edges remain in the table, so both endpoints must + # still be present or the client receives an edge to a missing node. + for row in persisted + if row.source_path in known and row.target_path in known + ] + logger.debug( + "Graph edges for %s served from code_dependencies (%d of %d rows in scope)", + repo.id, len(edges), len(persisted), + ) + return edges + + logger.info( + "No persisted dependency edges for %s; deriving from disk on the request " + "path. Re-index to populate code_dependencies.", + repo.id, + ) + return self._build_deterministic_edges(repo, source_paths, all_paths, file_map) + def _build_deterministic_edges( self, repo: Repository, diff --git a/apps/api/tests/unit/test_dependency_persistence.py b/apps/api/tests/unit/test_dependency_persistence.py new file mode 100644 index 0000000..97360c1 --- /dev/null +++ b/apps/api/tests/unit/test_dependency_persistence.py @@ -0,0 +1,181 @@ +""" +Dependency edges are derived once at index time and served from the database. + +The behaviour that matters: graph generation no longer needs the clone on disk. It used +to read every source file inside the request, which made graph latency proportional to +repository size and silently degraded the graph once the working tree was gone. +""" + +import asyncio + +import pytest +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 +from src.services.learning_service import LearningService + + +@pytest.fixture() +def db(tmp_path): + engine = create_engine(f"sqlite:///{tmp_path / 'test.db'}") + Base.metadata.create_all(engine) + session = sessionmaker(bind=engine)() + yield session + session.close() + + +def _make_repo(db, clone_root): + """A tiny TS project: app imports both helpers, one of which imports the other.""" + (clone_root / "src").mkdir(parents=True, exist_ok=True) + (clone_root / "src" / "app.ts").write_text( + "import { a } from './alpha';\nimport { b } from './beta';\n" + ) + (clone_root / "src" / "alpha.ts").write_text("import { b } from './beta';\nexport const a = 1;\n") + (clone_root / "src" / "beta.ts").write_text("export const b = 2;\n") + + repo = Repository( + github_url="https://github.com/o/r", + github_owner="o", + github_name="r", + local_path=str(clone_root), + ) + db.add(repo) + db.commit() + db.refresh(repo) + + for path in ("src/app.ts", "src/alpha.ts", "src/beta.ts"): + db.add(CodeFile( + repository_id=repo.id, + path=path, + filename=path.rsplit("/", 1)[-1], + extension=".ts", + language="typescript", + imports=[], + )) + db.commit() + return repo + + +def test_edges_are_persisted_at_index_time(db, tmp_path): + repo = _make_repo(db, tmp_path / "clone") + + assert db.query(CodeDependency).count() == 0 + + written = asyncio.run(IndexingService(db)._persist_dependency_graph(repo)) + + rows = db.query(CodeDependency).filter(CodeDependency.repository_id == repo.id).all() + assert written == len(rows) + assert rows, "expected at least one resolved import edge" + + pairs = {(r.source_path, r.target_path) for r in rows} + assert ("src/app.ts", "src/alpha.ts") in pairs + assert ("src/app.ts", "src/beta.ts") in pairs + assert ("src/alpha.ts", "src/beta.ts") in pairs + + for r in rows: + assert r.relation + assert r.weight >= 1 + assert 0.0 < r.confidence <= 1.0 + + +def test_graph_survives_the_clone_being_deleted(db, tmp_path): + """ + The point of the change. Previously, losing the working tree (redeploy, container + restart, volume reset) silently degraded the graph to unresolved import strings. + """ + clone = tmp_path / "clone" + repo = _make_repo(db, clone) + asyncio.run(IndexingService(db)._persist_dependency_graph(repo)) + + persisted_pairs = { + (r.source_path, r.target_path) + for r in db.query(CodeDependency).filter(CodeDependency.repository_id == repo.id) + } + + # Simulate the clone going away. + import shutil + shutil.rmtree(clone) + assert not clone.exists() + + files = db.query(CodeFile).filter(CodeFile.repository_id == repo.id).all() + file_map = {f.path: f for f in files} + all_paths = set(file_map) + + service = LearningService(db, llm=None, vector_store=None) + edges = service._load_or_derive_edges(repo, sorted(all_paths), all_paths, file_map) + + assert {(e.source, e.target) for e in edges} == persisted_pairs + + # And the derive-from-disk path really would have returned nothing now. + derived = service._build_deterministic_edges(repo, sorted(all_paths), all_paths, file_map) + assert not derived, "clone is gone, so on-demand derivation has nothing to read" + + +def test_falls_back_to_derivation_when_nothing_is_persisted(db, tmp_path): + """Repositories indexed before this table existed must still produce a graph.""" + clone = tmp_path / "clone" + repo = _make_repo(db, clone) + + assert db.query(CodeDependency).count() == 0 + + files = db.query(CodeFile).filter(CodeFile.repository_id == repo.id).all() + file_map = {f.path: f for f in files} + all_paths = set(file_map) + + service = LearningService(db, llm=None, vector_store=None) + edges = service._load_or_derive_edges(repo, sorted(all_paths), all_paths, file_map) + + assert edges, "expected the on-demand fallback to derive edges from the clone" + + +def test_edges_outside_the_node_set_are_dropped(db, tmp_path): + """ + Scope filters and the node cap can remove a node while its edges remain in the + table; shipping those would give the client an edge to a node it never received. + """ + repo = _make_repo(db, tmp_path / "clone") + asyncio.run(IndexingService(db)._persist_dependency_graph(repo)) + + files = db.query(CodeFile).filter(CodeFile.repository_id == repo.id).all() + file_map = {f.path: f for f in files if f.path != "src/beta.ts"} + all_paths = set(file_map) + + service = LearningService(db, llm=None, vector_store=None) + edges = service._load_or_derive_edges(repo, sorted(all_paths), all_paths, file_map) + + endpoints = {e.source for e in edges} | {e.target for e in edges} + assert "src/beta.ts" not in endpoints + assert endpoints <= all_paths + + +def test_reindex_clears_stale_edges(db, tmp_path): + """Re-indexing must not union old and new graphs.""" + repo = _make_repo(db, tmp_path / "clone") + asyncio.run(IndexingService(db)._persist_dependency_graph(repo)) + before = db.query(CodeDependency).filter(CodeDependency.repository_id == repo.id).count() + assert before > 0 + + # A stale edge to a file that no longer exists. + db.add(CodeDependency( + repository_id=repo.id, + source_path="src/deleted.ts", + target_path="src/beta.ts", + relation="imports", + )) + db.commit() + + asyncio.run(IndexingService(db)._reset_repository_index_data(repo.id)) + assert db.query(CodeDependency).filter(CodeDependency.repository_id == repo.id).count() == 0 + + +def test_deleting_a_repository_cascades_to_its_edges(db, tmp_path): + repo = _make_repo(db, tmp_path / "clone") + asyncio.run(IndexingService(db)._persist_dependency_graph(repo)) + assert db.query(CodeDependency).count() > 0 + + db.delete(repo) + db.commit() + + assert db.query(CodeDependency).count() == 0, "orphaned edges would union into a re-import"