refactor(intelligence): bound architecture snapshot reads (#133)

apps/backend/app/analysis/architecture.py

@@ -1,7 +1,12 @@
 import posixpath
 
 from app.intelligence.engine import RepositoryIntelligenceEngine
-from app.intelligence.query_service import ArchitectureSnapshotFacts, SnapshotQueryService
+from app.intelligence.query_service import (
+    ARCHITECTURE_DIAGNOSTIC_CODES,
+    ARCHITECTURE_RELATIONSHIP_EDGE_TYPES,
+    ArchitectureSnapshotFacts,
+    SnapshotQueryService,
+)
 from app.intelligence.models import RepositoryModule
 from app.models.repository import RepositoryRecord
 from app.models.snapshot import RiDiagnostic, RiEvidence, RiNode
@@ -35,15 +40,6 @@
     "unknown": "shared-library",
 }
 
-RELATIONSHIP_EDGE_TYPES = {
-    "imports": "import",
-    "calls": "calls",
-    "routes_to": "api-call",
-    "implements": "dependency",
-    "depends_on": "dependency",
-}
-
-
 class ArchitectureAnalyzer:
     def __init__(
         self,
@@ -79,7 +75,12 @@ def build_architecture(self, record: RepositoryRecord) -> ArchitectureResponse:
             repository_intelligence.discovery.primary_language if repository_intelligence is not None else "Unknown"
         )
         entry_points = repository_intelligence.discovery.entry_points if repository_intelligence is not None else []
-        facts = self.snapshots.architecture_facts(record.id) if self.snapshots is not None else None
+        module_paths = {self._normalize_path(path) for module in modules for path in module.files}
+        facts = (
+            self.snapshots.architecture_facts(record.id, module_paths=module_paths)
+            if self.snapshots is not None
+            else None
+        )
         nodes = self._nodes_for_modules(modules)
         nodes.extend(self._dependency_nodes(facts))
         edges, diagnostics, unresolved_node_ids, covered_paths = self._edges_for_modules(modules, nodes, facts)
@@ -161,7 +162,7 @@ def _dependency_nodes(self, facts: ArchitectureSnapshotFacts | None) -> list[Arc
         relationship_keys = {
             key
             for edge in facts.edges
-            if edge.predicate in RELATIONSHIP_EDGE_TYPES
+            if edge.predicate in ARCHITECTURE_RELATIONSHIP_EDGE_TYPES
             for key in (edge.subject_key, edge.object_key)
         }
         result: list[ArchNode] = []
@@ -215,21 +216,19 @@ def _edges_for_modules(
                 modules_by_file.setdefault(self._normalize_path(path), []).append(module.id)
         snapshot_node_by_key = {item.stable_key: item for item in facts.nodes}
         node_ids = {node.id for node in nodes}
-        diagnostics = [self._architecture_diagnostic(item, modules_by_file, node_ids) for item in facts.diagnostics]
+        diagnostics = [
+            self._architecture_diagnostic(item, modules_by_file, node_ids)
+            for item in facts.diagnostics
+            if item.code in ARCHITECTURE_DIAGNOSTIC_CODES
+        ]
         unresolved_node_ids: set[str] = set()
         # Inventory-only file nodes prove that a path exists, not that a
         # relationship-capable extractor ran. Count only evidence emitted by a
         # syntax/manifest producer so unsupported files cannot look isolated.
-        covered_paths = {
-            item.path
-            for evidence_by_fact in (facts.node_evidence, facts.observation_evidence)
-            for evidence in evidence_by_fact.values()
-            for item in evidence
-            if item.extractor != "repository-inventory"
-        }
+        covered_paths = facts.covered_paths
 
         for item in facts.diagnostics:
-            if item.code not in {"RI-RES-UNRESOLVED", "RI-RES-AMBIGUOUS"}:
+            if item.code not in ARCHITECTURE_DIAGNOSTIC_CODES:
                 continue
             if item.path:
                 unresolved_node_ids.update(modules_by_file.get(self._normalize_path(item.path), []))
@@ -239,7 +238,7 @@ def _edges_for_modules(
 
         edges: list[ArchEdge] = []
         for fact in facts.edges:
-            if fact.predicate not in RELATIONSHIP_EDGE_TYPES:
+            if fact.predicate not in ARCHITECTURE_RELATIONSHIP_EDGE_TYPES:
                 continue
             evidence_rows = facts.edge_evidence.get(fact.id, [])
             source_ids = self._architecture_endpoint_ids(
@@ -333,7 +332,7 @@ def _edges_for_modules(
                         id=edge_id,
                         source=source,
                         target=target,
-                        type=RELATIONSHIP_EDGE_TYPES[fact.predicate],  # type: ignore[arg-type]
+                        type=ARCHITECTURE_RELATIONSHIP_EDGE_TYPES[fact.predicate],  # type: ignore[arg-type]
                         label=fact.predicate.replace("_", " "),
                         predicate=fact.predicate,
                         truth_class="inferred",

apps/backend/app/intelligence/query_service.py

@@ -1,6 +1,7 @@
 """Read-only, owner-scoped queries over sealed ``ri.v1`` snapshots (#92)."""
 
 from collections import defaultdict
+from collections.abc import Collection
 from dataclasses import dataclass
 
 from sqlalchemy import func, or_, select
@@ -19,18 +20,28 @@
 )
 
 
+ARCHITECTURE_RELATIONSHIP_EDGE_TYPES = {
+    "imports": "import",
+    "calls": "calls",
+    "routes_to": "api-call",
+    "implements": "dependency",
+    "depends_on": "dependency",
+}
+ARCHITECTURE_DIAGNOSTIC_CODES = frozenset({"RI-RES-UNRESOLVED", "RI-RES-AMBIGUOUS"})
+ARCHITECTURE_EVIDENCE_BATCH_SIZE = 500
+
+
 @dataclass(frozen=True)
 class ArchitectureSnapshotFacts:
     """Persisted facts needed to build evidence-backed architecture relationships."""
 
     snapshot: RiSnapshot
     nodes: list[RiNode]
-    observations: list[RiObservation]
     edges: list[RiEdge]
     node_evidence: dict[int, list[RiEvidence]]
-    observation_evidence: dict[int, list[RiEvidence]]
     edge_evidence: dict[int, list[RiEvidence]]
     diagnostics: list[RiDiagnostic]
+    covered_paths: set[str]
 
 
 class SnapshotQueryService:
@@ -45,7 +56,12 @@ def __init__(self, db: Session, owner_id: str) -> None:
     def metadata(self, snapshot_id: str) -> RiSnapshot:
         return self._snapshot(snapshot_id)
 
-    def architecture_facts(self, repository_id: str) -> ArchitectureSnapshotFacts | None:
+    def architecture_facts(
+        self,
+        repository_id: str,
+        *,
+        module_paths: Collection[str],
+    ) -> ArchitectureSnapshotFacts | None:
         """Return the newest sealed snapshot facts for an owner-scoped repository.
 
         This internal consumer query uses the normalized store behind the public
@@ -56,31 +72,35 @@ def architecture_facts(self, repository_id: str) -> ArchitectureSnapshotFacts |
         snapshot = self._latest_snapshot(repository_id)
         if snapshot is None:
             return None
-        nodes = list(
-            self.db.scalars(
-                select(RiNode)
-                .where(RiNode.snapshot_id == snapshot.snapshot_id)
-                .order_by(RiNode.stable_key, RiNode.id)
-            ).all()
-        )
         edges = list(
             self.db.scalars(
                 select(RiEdge)
-                .where(RiEdge.snapshot_id == snapshot.snapshot_id)
+                .where(
+                    RiEdge.snapshot_id == snapshot.snapshot_id,
+                    RiEdge.predicate.in_(ARCHITECTURE_RELATIONSHIP_EDGE_TYPES),
+                )
                 .order_by(RiEdge.subject_key, RiEdge.predicate, RiEdge.object_key, RiEdge.edge_id, RiEdge.id)
             ).all()
         )
-        observations = list(
-            self.db.scalars(
-                select(RiObservation)
-                .where(RiObservation.snapshot_id == snapshot.snapshot_id)
-                .order_by(RiObservation.observation_id, RiObservation.id)
-            ).all()
+        endpoint_keys = {key for edge in edges for key in (edge.subject_key, edge.object_key)}
+        nodes = (
+            list(
+                self.db.scalars(
+                    select(RiNode)
+                    .where(RiNode.snapshot_id == snapshot.snapshot_id, RiNode.stable_key.in_(endpoint_keys))
+                    .order_by(RiNode.stable_key, RiNode.id)
+                ).all()
+            )
+            if endpoint_keys
+            else []
         )
         diagnostics = list(
             self.db.scalars(
                 select(RiDiagnostic)
-                .where(RiDiagnostic.snapshot_id == snapshot.snapshot_id)
+                .where(
+                    RiDiagnostic.snapshot_id == snapshot.snapshot_id,
+                    RiDiagnostic.code.in_(ARCHITECTURE_DIAGNOSTIC_CODES),
+                )
                 .order_by(
                     RiDiagnostic.path,
                     RiDiagnostic.span_start_line,
@@ -89,19 +109,15 @@ def architecture_facts(self, repository_id: str) -> ArchitectureSnapshotFacts |
                 )
             ).all()
         )
+        covered_paths = self._architecture_covered_paths(snapshot, module_paths)
         return ArchitectureSnapshotFacts(
             snapshot=snapshot,
             nodes=nodes,
-            observations=observations,
             edges=edges,
             node_evidence=self._evidence_for(snapshot, "node_ref", [node.id for node in nodes]),
-            observation_evidence=self._evidence_for(
-                snapshot,
-                "observation_ref",
-                [observation.id for observation in observations],
-            ),
             edge_evidence=self._evidence_for(snapshot, "edge_ref", [edge.id for edge in edges]),
             diagnostics=diagnostics,
+            covered_paths=covered_paths,
         )
 
     def symbols(self, snapshot_id: str, *, offset: int, limit: int) -> tuple[RiSnapshot, list[RiNode], int]:
@@ -281,26 +297,50 @@ def _page(self, model, where: tuple, order_by: tuple, offset: int, limit: int):
         rows = self.db.scalars(select(model).where(*where).order_by(*order_by).offset(offset).limit(limit)).all()
         return list(rows), total
 
+    def _architecture_covered_paths(self, snapshot: RiSnapshot, module_paths: Collection[str]) -> set[str]:
+        """Return only module paths with non-inventory extraction evidence."""
+
+        paths = sorted(set(module_paths))
+        covered_paths: set[str] = set()
+        for start in range(0, len(paths), ARCHITECTURE_EVIDENCE_BATCH_SIZE):
+            covered_paths.update(
+                self.db.scalars(
+                    select(RiEvidence.path)
+                    .where(
+                        RiEvidence.snapshot_id == snapshot.snapshot_id,
+                        RiEvidence.path.in_(paths[start : start + ARCHITECTURE_EVIDENCE_BATCH_SIZE]),
+                        RiEvidence.extractor != "repository-inventory",
+                        or_(RiEvidence.node_ref.is_not(None), RiEvidence.observation_ref.is_not(None)),
+                    )
+                    .distinct()
+                ).all()
+            )
+        return covered_paths
+
     def _evidence_for(self, snapshot: RiSnapshot, column: str, ids: list[int]) -> dict[int, list[RiEvidence]]:
         if not ids:
             return {}
         field = getattr(RiEvidence, column)
-        rows = self.db.scalars(
-            select(RiEvidence)
-            .where(RiEvidence.snapshot_id == snapshot.snapshot_id, field.in_(ids))
-            .order_by(
-                RiEvidence.path,
-                RiEvidence.start_line,
-                RiEvidence.end_line,
-                RiEvidence.granularity,
-                RiEvidence.extractor,
-                RiEvidence.extractor_version,
-                RiEvidence.id,
-            )
-        ).all()
         grouped: dict[int, list[RiEvidence]] = defaultdict(list)
-        for row in rows:
-            parent = getattr(row, column)
-            if parent is not None:
-                grouped[parent].append(row)
+        for start in range(0, len(ids), ARCHITECTURE_EVIDENCE_BATCH_SIZE):
+            rows = self.db.scalars(
+                select(RiEvidence)
+                .where(
+                    RiEvidence.snapshot_id == snapshot.snapshot_id,
+                    field.in_(ids[start : start + ARCHITECTURE_EVIDENCE_BATCH_SIZE]),
+                )
+                .order_by(
+                    RiEvidence.path,
+                    RiEvidence.start_line,
+                    RiEvidence.end_line,
+                    RiEvidence.granularity,
+                    RiEvidence.extractor,
+                    RiEvidence.extractor_version,
+                    RiEvidence.id,
+                )
+            ).all()
+            for row in rows:
+                parent = getattr(row, column)
+                if parent is not None:
+                    grouped[parent].append(row)
         return grouped