feat(intelligence): add snapshot impact query

README.md

@@ -76,7 +76,7 @@ Statuses describe executable behaviour on the current `dev` branch:
 | AI provider integration | **Implemented, limited** | Per-user configuration for supported providers, encrypted API keys, and constrained outbound destinations. Free-form answers receive structural facts and observed paths—not source bytes or line spans—and return no automatic citations. |
 | Asynchronous processing | **Partially implemented** | Analysis runs off the request path. Import, extraction of the initial archive/clone, and file-tree parsing remain synchronous; one in-process worker handles analysis jobs. |
 | Incremental re-analysis and revision comparison | **Not implemented / not assessed** | The full repository is analysed again; no snapshot-to-snapshot product workflow is available. |
-| Change-impact or blast-radius analysis | **Not implemented / not assessed** | No product result is emitted. |
+| Change-impact or blast-radius analysis | **Implemented, limited** | Owner-scoped traversal over one sealed snapshot's resolved import and dependency edges. It does not compare revisions or calculate churn or trends. |
 | Vulnerability and outdated-dependency scanning | **Not implemented / not assessed** | Dependency responses report explicit `not_computed` states; Review keeps vulnerability scanning `not_assessed`. No clean bill of health or zero count is fabricated. |
 | Grounded, cited free-form AI answers | **Not implemented / not assessed** | Provider answers are intentionally uncited because providers do not receive source content or line numbers. |
 
@@ -250,7 +250,7 @@ The prototype browser suite exercises defined Architecture, Engineering Review,
 - **Pre-alpha, trusted-environment use.** PARTHA has not been operated or hardened as a public multi-tenant service.
 - **Narrow semantic coverage.** Supported Python and TypeScript/JavaScript constructs receive the deepest extraction. Other languages primarily contribute file inventory. Role, module, layer, framework, and entry-point classifications can be heuristic.
 - **Narrow dependency coverage.** Only direct declarations in three manifest formats are extracted. Lockfiles, transitive dependencies, vulnerability scanning, and outdated-version scanning are not implemented.
-- **No repository evolution workflow.** Analysis is whole-repository; incremental analysis, revision comparison, churn/trend analysis, and change impact are unavailable.
+- **No repository evolution workflow.** Analysis is whole-repository; incremental analysis, revision comparison, and churn/trend analysis are unavailable. The sealed-snapshot impact query does not compare revisions or calculate historical change.
 - **Surface-dependent evidence.** A sealed snapshot does not make every product sentence line-cited. In particular, generated structural documentation and free-form AI have stricter evidence limits.
 - **In-process execution.** A daemon worker thread inside the API process handles one analysis job at a time; there is no separate worker service or external job queue.
 

apps/backend/alembic/versions/0008_impact_query_edge_indexes.py

@@ -0,0 +1,38 @@
+"""add directional snapshot-edge indexes for impact traversal
+
+Revision ID: 0008_impact_query_edge_indexes
+Revises: 0007_remove_data_source
+Create Date: 2026-07-28
+
+The sealed-snapshot impact query filters resolved edges by snapshot, one
+endpoint direction, and predicate. These complementary composite indexes keep
+both outgoing dependency and incoming dependent lookups bounded as snapshots
+grow.
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0008_impact_query_edge_indexes"
+down_revision = "0007_remove_data_source"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "ix_ri_edges_snapshot_subject_predicate",
+        "ri_edges",
+        ["snapshot_id", "subject_key", "predicate"],
+    )
+    op.create_index(
+        "ix_ri_edges_snapshot_object_predicate",
+        "ri_edges",
+        ["snapshot_id", "object_key", "predicate"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_ri_edges_snapshot_object_predicate", table_name="ri_edges")
+    op.drop_index("ix_ri_edges_snapshot_subject_predicate", table_name="ri_edges")

apps/backend/app/api/routes/intelligence.py

@@ -6,13 +6,16 @@
 
 from app.api.deps import get_current_user, get_snapshot_query_service
 from app.api.openapi import documented_responses
-from app.intelligence.query_service import SnapshotQueryService
+from app.intelligence.query_service import IMPACT_MAX_DEPTH, SnapshotQueryService
 from app.schemas.intelligence import (
     RiAssertionResponse,
     RiAssertionsResponse,
     RiEdgeResponse,
     RiEvidenceResponse,
     RiEvidenceResponsePage,
+    RiImpactDirectionResponse,
+    RiImpactResponse,
+    RiImpactStepResponse,
     RiNeighboursResponse,
     RiNodeResponse,
     RiPagination,
@@ -33,6 +36,14 @@
 _MAX_LIMIT = 100
 PaginationOffset = Annotated[int, Query(ge=0, description="Zero-based deterministic result offset.")]
 PaginationLimit = Annotated[int, Query(ge=1, le=_MAX_LIMIT, description="Maximum 100 results per page.")]
+ImpactDepth = Annotated[
+    int,
+    Query(
+        ge=1,
+        le=IMPACT_MAX_DEPTH,
+        description=f"Directed import/dependency traversal depth (1-{IMPACT_MAX_DEPTH}).",
+    ),
+]
 
 
 def _metadata(snapshot) -> RiSnapshotMetadataResponse:
@@ -114,6 +125,20 @@ def _pagination(offset: int, limit: int, total: int) -> RiPagination:
     return RiPagination(offset=offset, limit=limit, total=total)
 
 
+def _impact_direction(snapshot, direction, evidence, derivations) -> RiImpactDirectionResponse:
+    return RiImpactDirectionResponse(
+        data=[
+            RiImpactStepResponse(
+                depth=step.depth,
+                node_key=step.node_key,
+                via=_edge(snapshot, step.edge, evidence, derivations),
+            )
+            for step in direction.steps
+        ],
+        limit_reached=direction.limit_reached,
+    )
+
+
 @router.get(
     "/{snapshot_id}",
     response_model=RiSnapshotMetadataResponse,
@@ -157,6 +182,49 @@ def list_neighbours(
     return RiNeighboursResponse(schema_version=snapshot.schema_version, node_key=node_key, data=[_edge(snapshot, edge, evidence, derivations) for edge in edges], pagination=_pagination(offset, limit, total))
 
 
+@router.get(
+    "/{snapshot_id}/impact",
+    response_model=RiImpactResponse,
+    responses=documented_responses(
+        200,
+        "Bounded, provenance-backed dependents and dependencies over stored import/dependency edges.",
+        {
+            "schemaVersion": "ri.v1",
+            "nodeKey": "file:src/api.py",
+            "depth": 1,
+            "dependents": {"data": [], "limitReached": False},
+            "dependencies": {"data": [], "limitReached": False},
+        },
+        401,
+        404,
+        422,
+        429,
+        500,
+    ),
+)
+def get_impact(
+    snapshot_id: str,
+    node_key: Annotated[str, Query(alias="nodeKey", min_length=1, max_length=1024)],
+    service: SnapshotQueryService = Depends(get_snapshot_query_service),
+    depth: ImpactDepth = 1,
+) -> RiImpactResponse:
+    impact = service.impact(snapshot_id, node_key=node_key, depth=depth)
+    edges = [
+        step.edge
+        for direction in (impact.dependents, impact.dependencies)
+        for step in direction.steps
+    ]
+    evidence = service.evidence_for_edges(impact.snapshot, edges)
+    derivations = service.derivations_for_edges(impact.snapshot, edges)
+    return RiImpactResponse(
+        schema_version=impact.snapshot.schema_version,
+        node_key=impact.node_key,
+        depth=impact.depth,
+        dependents=_impact_direction(impact.snapshot, impact.dependents, evidence, derivations),
+        dependencies=_impact_direction(impact.snapshot, impact.dependencies, evidence, derivations),
+    )
+
+
 @router.get(
     "/{snapshot_id}/references",
     response_model=RiReferencesResponse,

apps/backend/app/intelligence/query_service.py

@@ -61,6 +61,15 @@
 #: trusting the set to stay small.
 SNAPSHOT_IN_CLAUSE_BATCH_SIZE = 500
 ARCHITECTURE_EVIDENCE_BATCH_SIZE = SNAPSHOT_IN_CLAUSE_BATCH_SIZE
+#: Only these persisted, directed relationship facts participate in the #173
+#: blast-radius query. Unresolved observations remain diagnostics and are never
+#: traversed as if they were relationships.
+IMPACT_TRAVERSAL_PREDICATES = ("depends_on", "imports")
+#: A traversal is intentionally bounded by both depth and returned nodes. The
+#: endpoint exposes ``limit_reached`` whenever the node cap prevents a complete
+#: result at the requested depth.
+IMPACT_MAX_DEPTH = 10
+IMPACT_MAX_RESULTS_PER_DIRECTION = 100
 
 
 def batched_ids[T](values: Sequence[T], size: int = SNAPSHOT_IN_CLAUSE_BATCH_SIZE) -> Iterator[list[T]]:
@@ -163,6 +172,34 @@ class ProductSnapshotProjection:
     file_relationships: tuple[SnapshotFileRelationshipFact, ...] = ()
 
 
+@dataclass(frozen=True)
+class SnapshotImpactStep:
+    """One deterministically selected edge on a bounded graph traversal."""
+
+    depth: int
+    node_key: str
+    edge: RiEdge
+
+
+@dataclass(frozen=True)
+class SnapshotImpactDirection:
+    """One directional impact result and whether the response was capped."""
+
+    steps: tuple[SnapshotImpactStep, ...]
+    limit_reached: bool
+
+
+@dataclass(frozen=True)
+class SnapshotImpact:
+    """Persisted, provenance-backed impact facts for one snapshot node."""
+
+    snapshot: RiSnapshot
+    node_key: str
+    depth: int
+    dependents: SnapshotImpactDirection
+    dependencies: SnapshotImpactDirection
+
+
 class SnapshotQueryService:
     """Expose only persisted snapshot facts; this service never touches repository storage."""
 
@@ -652,6 +689,32 @@ def neighbours(
         )
         return snapshot, rows, total
 
+    def impact(self, snapshot_id: str, *, node_key: str, depth: int) -> SnapshotImpact:
+        """Traverse stored import/dependency edges from one known snapshot node.
+
+        The traversal is read-only, owner-scoped through :meth:`_snapshot`, and
+        only follows resolved edges. Each direction retains a canonical shortest
+        path to each reached node, so cycles cannot cause unbounded work or
+        duplicate result nodes.
+        """
+
+        snapshot = self._snapshot(snapshot_id)
+        node_exists = self.db.scalar(
+            select(RiNode.id).where(
+                RiNode.snapshot_id == snapshot.snapshot_id,
+                RiNode.stable_key == node_key,
+            )
+        )
+        if node_exists is None:
+            raise NotFoundError("Node not found in snapshot.")
+        return SnapshotImpact(
+            snapshot=snapshot,
+            node_key=node_key,
+            depth=depth,
+            dependents=self._impact_direction(snapshot, node_key=node_key, depth=depth, outbound=False),
+            dependencies=self._impact_direction(snapshot, node_key=node_key, depth=depth, outbound=True),
+        )
+
     def references(self, snapshot_id: str, *, offset: int, limit: int) -> tuple[RiSnapshot, list[RiEdge], int]:
         """Return only stored resolved relationship facts, never unresolved observations."""
 
@@ -806,6 +869,75 @@ 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 _impact_direction(
+        self,
+        snapshot: RiSnapshot,
+        *,
+        node_key: str,
+        depth: int,
+        outbound: bool,
+    ) -> SnapshotImpactDirection:
+        """Walk one edge direction with deterministic breadth-first ordering."""
+
+        frontier = [node_key]
+        visited = {node_key}
+        steps: list[SnapshotImpactStep] = []
+        endpoint = RiEdge.subject_key if outbound else RiEdge.object_key
+        adjacent = RiEdge.object_key if outbound else RiEdge.subject_key
+
+        for distance in range(1, depth + 1):
+            if not frontier:
+                break
+            # A reached node can be proven by many current-frontier edges.
+            # Ranking one canonical edge per adjacent node *before* the cap is
+            # essential: limiting raw edges could otherwise exhaust the query
+            # budget on duplicate paths and falsely claim the result complete.
+            canonical_rank = func.row_number().over(
+                partition_by=adjacent,
+                order_by=(
+                    RiEdge.subject_key,
+                    RiEdge.predicate,
+                    RiEdge.object_key,
+                    RiEdge.edge_id,
+                    RiEdge.id,
+                ),
+            ).label("canonical_rank")
+            ranked_edges = (
+                select(RiEdge.id.label("edge_row_id"), canonical_rank)
+                .where(
+                    RiEdge.snapshot_id == snapshot.snapshot_id,
+                    RiEdge.predicate.in_(IMPACT_TRAVERSAL_PREDICATES),
+                    endpoint.in_(frontier),
+                    adjacent.not_in(visited),
+                )
+                .subquery()
+            )
+            edges = self.db.scalars(
+                select(RiEdge)
+                .join(ranked_edges, RiEdge.id == ranked_edges.c.edge_row_id)
+                .where(ranked_edges.c.canonical_rank == 1)
+                .order_by(
+                    RiEdge.subject_key,
+                    RiEdge.predicate,
+                    RiEdge.object_key,
+                    RiEdge.edge_id,
+                    RiEdge.id,
+                )
+                .limit(IMPACT_MAX_RESULTS_PER_DIRECTION + 1)
+            ).all()
+
+            next_frontier: list[str] = []
+            for edge in edges:
+                adjacent_key = edge.object_key if outbound else edge.subject_key
+                if len(steps) >= IMPACT_MAX_RESULTS_PER_DIRECTION:
+                    return SnapshotImpactDirection(steps=tuple(steps), limit_reached=True)
+                visited.add(adjacent_key)
+                next_frontier.append(adjacent_key)
+                steps.append(SnapshotImpactStep(depth=distance, node_key=adjacent_key, edge=edge))
+            frontier = next_frontier
+
+        return SnapshotImpactDirection(steps=tuple(steps), limit_reached=False)
+
     def _architecture_covered_paths(self, snapshot: RiSnapshot) -> set[str]:
         """Return snapshot paths that carry non-inventory extraction evidence.
 

apps/backend/app/models/snapshot.py

@@ -216,6 +216,18 @@ class RiEdge(Base):
             ondelete="CASCADE",
         ),
         Index("ix_ri_edges_snapshot_id", "snapshot_id"),
+        Index(
+            "ix_ri_edges_snapshot_subject_predicate",
+            "snapshot_id",
+            "subject_key",
+            "predicate",
+        ),
+        Index(
+            "ix_ri_edges_snapshot_object_predicate",
+            "snapshot_id",
+            "object_key",
+            "predicate",
+        ),
     )
 
 

apps/backend/app/schemas/intelligence.py

@@ -101,6 +101,33 @@ class RiNeighboursResponse(CamelModel):
     pagination: RiPagination
 
 
+class RiImpactStepResponse(CamelModel):
+    """A reached node and the stored edge that proves the traversal hop."""
+
+    depth: int = Field(ge=1)
+    node_key: str
+    via: RiEdgeResponse
+
+
+class RiImpactDirectionResponse(CamelModel):
+    """One bounded traversal direction.
+
+    ``limit_reached`` means the response omitted further reachable nodes at the
+    requested depth rather than claiming that the listed nodes are exhaustive.
+    """
+
+    data: list[RiImpactStepResponse]
+    limit_reached: bool
+
+
+class RiImpactResponse(CamelModel):
+    schema_version: SchemaVersion
+    node_key: str
+    depth: int = Field(ge=1)
+    dependents: RiImpactDirectionResponse
+    dependencies: RiImpactDirectionResponse
+
+
 class RiReferencesResponse(CamelModel):
     schema_version: SchemaVersion
     data: list[RiEdgeResponse]