From 8e2b3e920994f508d5a3daf4b7e0b838f2a89274 Mon Sep 17 00:00:00 2001 From: hardikuppal04 <262667963+hardikuppal04@users.noreply.github.com> Date: Tue, 28 Jul 2026 02:13:04 +0530 Subject: [PATCH] feat(intelligence): add snapshot impact query (#173) --- README.md | 4 +- .../0008_impact_query_edge_indexes.py | 38 ++ apps/backend/app/api/routes/intelligence.py | 70 +++- .../backend/app/intelligence/query_service.py | 132 +++++++ apps/backend/app/models/snapshot.py | 12 + apps/backend/app/schemas/intelligence.py | 27 ++ .../tests/test_intelligence_query_api.py | 330 +++++++++++++++++- apps/backend/tests/test_migrations.py | 14 + apps/backend/tests/test_openapi_contract.py | 1 + docs/architecture/REPOSITORY_INTELLIGENCE.md | 11 + 10 files changed, 634 insertions(+), 5 deletions(-) create mode 100644 apps/backend/alembic/versions/0008_impact_query_edge_indexes.py diff --git a/README.md b/README.md index f3c636d..1fe90b0 100644 --- a/README.md +++ b/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. diff --git a/apps/backend/alembic/versions/0008_impact_query_edge_indexes.py b/apps/backend/alembic/versions/0008_impact_query_edge_indexes.py new file mode 100644 index 0000000..ce93817 --- /dev/null +++ b/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") diff --git a/apps/backend/app/api/routes/intelligence.py b/apps/backend/app/api/routes/intelligence.py index 9cbd257..cce9e58 100644 --- a/apps/backend/app/api/routes/intelligence.py +++ b/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, diff --git a/apps/backend/app/intelligence/query_service.py b/apps/backend/app/intelligence/query_service.py index ac8af4f..ef4b4f7 100644 --- a/apps/backend/app/intelligence/query_service.py +++ b/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. diff --git a/apps/backend/app/models/snapshot.py b/apps/backend/app/models/snapshot.py index 93f6469..98305c9 100644 --- a/apps/backend/app/models/snapshot.py +++ b/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", + ), ) diff --git a/apps/backend/app/schemas/intelligence.py b/apps/backend/app/schemas/intelligence.py index a6b46ea..dcf70c5 100644 --- a/apps/backend/app/schemas/intelligence.py +++ b/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] diff --git a/apps/backend/tests/test_intelligence_query_api.py b/apps/backend/tests/test_intelligence_query_api.py index e7efbd2..d8acffe 100644 --- a/apps/backend/tests/test_intelligence_query_api.py +++ b/apps/backend/tests/test_intelligence_query_api.py @@ -4,6 +4,7 @@ import pytest +from app.intelligence.query_service import IMPACT_MAX_DEPTH from tests.conftest import register_user @@ -90,13 +91,196 @@ def evidence(path: str, start: int, end: int, producer: str = "inventory") -> Ev db.close() +def _seed_impact_snapshot(owner_id: str) -> tuple[str, str]: + """Persist an intentionally cyclic graph without a readable worktree.""" + + from app.core.database import SessionLocal + from app.intelligence.snapshot_store import Evidence, Revision, SnapshotStore, observation_ref + from app.models.repository import RepositoryRecord + + db = SessionLocal() + try: + repository_id = str(uuid4()) + revision_value = "sha256:" + "c" * 64 + db.add( + RepositoryRecord( + id=repository_id, + owner_id=owner_id, + name="impact-query", + source="upload", + revision_kind="upload", + revision_value=revision_value, + local_path=f"/definitely/inaccessible/{repository_id}", + status="completed", + file_tree=[], + ) + ) + db.commit() + store = SnapshotStore(db) + snapshot = store.begin( + repository_id=repository_id, + revision=Revision("upload", revision_value), + producer_version_set=["inventory@1.0.0", "resolver@1.0.0"], + ) + + def evidence(path: str, line: int, producer: str = "inventory") -> Evidence: + return Evidence(path, line, line, producer, "1.0.0", logical_line_count=40) + + store.add_node(snapshot, node_kind="repository", stable_key="repo:root", evidence=[evidence("README.md", 1)]) + for index, path in enumerate(("a.py", "b.py", "c.py", "d.py"), start=2): + store.add_node( + snapshot, + node_kind="file", + stable_key=f"file:src/{path}", + name=path, + language="python", + evidence=[evidence(f"src/{path}", index)], + ) + store.add_node( + snapshot, + node_kind="dependency", + stable_key="dep:pypi:requests", + name="requests", + evidence=[evidence("pyproject.toml", 6)], + ) + + def edge( + subject_kind: str, + subject_key: str, + predicate: str, + object_kind: str, + object_key: str, + line: int, + ) -> None: + path = "pyproject.toml" if predicate == "depends_on" else "src/relationships.py" + observation = store.add_observation( + snapshot, + observed_kind="dependency" if predicate == "depends_on" else "import", + subject_kind=subject_kind, + subject_key=subject_key, + referent_text=object_key, + evidence=evidence(path, line), + ) + store.add_edge( + snapshot, + subject_kind=subject_kind, + subject_key=subject_key, + predicate=predicate, + object_kind=object_kind, + object_key=object_key, + producer="resolver", + producer_version="1.0.0", + evidence=[evidence(path, line, "resolver")], + derived_from=[observation_ref(observation.observation_id)], + ) + + # A structural edge is deliberately present and must not enter the + # impact result. The import edges form a cycle A -> B -> C -> A, with + # a second direct relationship A -> D and a separate manifest edge. + edge("repository", "repo:root", "contains", "file", "file:src/a.py", 7) + edge("repository", "repo:root", "depends_on", "dependency", "dep:pypi:requests", 8) + edge("file", "file:src/a.py", "imports", "file", "file:src/b.py", 9) + edge("file", "file:src/b.py", "imports", "file", "file:src/c.py", 10) + edge("file", "file:src/c.py", "imports", "file", "file:src/a.py", 11) + edge("file", "file:src/a.py", "imports", "file", "file:src/d.py", 12) + edge("file", "file:src/d.py", "imports", "file", "file:src/a.py", 13) + return repository_id, store.seal(snapshot).snapshot_id + finally: + db.close() + + +def _seed_duplicate_heavy_impact_snapshot(owner_id: str) -> tuple[str, str]: + """Persist a graph where duplicate paths would hide the overflow node.""" + + from app.core.database import SessionLocal + from app.intelligence.snapshot_store import Evidence, Revision, SnapshotStore, observation_ref + from app.models.repository import RepositoryRecord + + db = SessionLocal() + try: + repository_id = str(uuid4()) + revision_value = "sha256:" + "d" * 64 + db.add( + RepositoryRecord( + id=repository_id, + owner_id=owner_id, + name="impact-duplicates", + source="upload", + revision_kind="upload", + revision_value=revision_value, + local_path=f"/definitely/inaccessible/{repository_id}", + status="completed", + file_tree=[], + ) + ) + db.commit() + store = SnapshotStore(db) + snapshot = store.begin( + repository_id=repository_id, + revision=Revision("upload", revision_value), + producer_version_set=["inventory@1.0.0", "resolver@1.0.0"], + ) + + def evidence(line: int, producer: str = "inventory") -> Evidence: + return Evidence("src/graph.py", line, line, producer, "1.0.0", logical_line_count=300) + + store.add_node(snapshot, node_kind="repository", stable_key="repo:root", evidence=[evidence(1)]) + for number in range(4): + key = f"file:frontier/{number:02d}.py" + store.add_node(snapshot, node_kind="file", stable_key=key, name=key.rsplit("/", 1)[-1], evidence=[evidence(number + 2)]) + for number in range(96): + key = f"file:shared/{number:03d}.py" + store.add_node(snapshot, node_kind="file", stable_key=key, name=key.rsplit("/", 1)[-1], evidence=[evidence(number + 6)]) + store.add_node( + snapshot, + node_kind="file", + stable_key="file:overflow.py", + name="overflow.py", + evidence=[evidence(102)], + ) + + def edge(subject_key: str, object_key: str, line: int) -> None: + observation = store.add_observation( + snapshot, + observed_kind="import", + subject_kind="file" if subject_key != "repo:root" else "repository", + subject_key=subject_key, + referent_text=object_key, + evidence=evidence(line), + ordinal=line, + ) + store.add_edge( + snapshot, + subject_kind="file" if subject_key != "repo:root" else "repository", + subject_key=subject_key, + predicate="imports", + object_kind="file", + object_key=object_key, + producer="resolver", + producer_version="1.0.0", + evidence=[evidence(line, "resolver")], + derived_from=[observation_ref(observation.observation_id)], + ) + + for number in range(4): + edge("repo:root", f"file:frontier/{number:02d}.py", 103 + number) + for number in range(96): + edge(f"file:frontier/00.py", f"file:shared/{number:03d}.py", 107 + number) + for number in range(13): + edge(f"file:frontier/01.py", f"file:shared/{number:03d}.py", 203 + number) + edge("file:frontier/03.py", "file:overflow.py", 216) + return repository_id, store.seal(snapshot).snapshot_id + finally: + db.close() + + def test_snapshot_query_endpoints_are_authenticated_owner_scoped_and_filesystem_independent(client, make_auth_headers): alice = make_auth_headers("alice-query@example.com") bob = make_auth_headers("bob-query@example.com") _, alice_snapshot = _seed_snapshot(alice["user"]["id"]) _, bob_snapshot = _seed_snapshot(bob["user"]["id"], suffix="two") - endpoints = ["", "/symbols", "/neighbours?nodeKey=repo:root", "/references", "/assertions", "/paths", "/evidence"] + endpoints = ["", "/symbols", "/neighbours?nodeKey=repo:root", "/impact?nodeKey=repo:root", "/references", "/assertions", "/paths", "/evidence"] for endpoint in endpoints: unauthenticated = client.get(f"/intelligence/v1/snapshots/{alice_snapshot}{endpoint}") assert unauthenticated.status_code == 401 @@ -172,7 +356,7 @@ def test_query_rejects_owner_visible_unsupported_schema_without_cross_owner_disc _, snapshot_id = _seed_snapshot(owner["user"]["id"], schema_version="ri.v2") path = f"/intelligence/v1/snapshots/{snapshot_id}" - for suffix in ["", "/symbols", "/neighbours?nodeKey=repo:root", "/references", "/assertions", "/paths", "/evidence"]: + for suffix in ["", "/symbols", "/neighbours?nodeKey=repo:root", "/impact?nodeKey=repo:root", "/references", "/assertions", "/paths", "/evidence"]: rejected = client.get(f"{path}{suffix}", headers=owner["headers"]) assert rejected.status_code == 422 assert rejected.json()["code"] == "unsupported_schema_version" @@ -195,9 +379,151 @@ def test_snapshot_query_rejects_invalid_pagination(client, make_auth_headers, qu assert response.json()["code"] == "request_validation_error" +def test_impact_query_returns_bounded_direct_and_transitive_relationships_with_provenance(client, make_auth_headers): + owner = make_auth_headers("impact-owner@example.com") + _, snapshot_id = _seed_impact_snapshot(owner["user"]["id"]) + base = f"/intelligence/v1/snapshots/{snapshot_id}/impact?nodeKey=file:src/a.py" + + depth_one = client.get(f"{base}&depth=1", headers=owner["headers"]) + depth_two = client.get(f"{base}&depth=2", headers=owner["headers"]) + repeated = client.get(f"{base}&depth=2", headers=owner["headers"]) + + assert depth_one.status_code == depth_two.status_code == repeated.status_code == 200 + assert depth_two.json() == repeated.json() + assert depth_two.json()["schemaVersion"] == "ri.v1" + assert depth_one.json()["depth"] == 1 + assert [item["nodeKey"] for item in depth_one.json()["dependencies"]["data"]] == [ + "file:src/b.py", + "file:src/d.py", + ] + assert [item["nodeKey"] for item in depth_two.json()["dependencies"]["data"]] == [ + "file:src/b.py", + "file:src/d.py", + "file:src/c.py", + ] + assert [item["depth"] for item in depth_two.json()["dependencies"]["data"]] == [1, 1, 2] + assert [item["nodeKey"] for item in depth_two.json()["dependents"]["data"]] == [ + "file:src/c.py", + "file:src/d.py", + "file:src/b.py", + ] + assert [item["depth"] for item in depth_two.json()["dependents"]["data"]] == [1, 1, 2] + assert "file:src/a.py" not in { + item["nodeKey"] + for direction in ("dependencies", "dependents") + for item in depth_two.json()[direction]["data"] + } + first_hop = depth_two.json()["dependencies"]["data"][0] + assert first_hop["via"]["predicate"] == "imports" + assert first_hop["via"]["evidence"][0]["extractor"] == "resolver" + assert first_hop["via"]["derivedFrom"][0]["kind"] == "observation" + assert depth_two.json()["dependencies"]["limitReached"] is False + assert depth_two.json()["dependents"]["limitReached"] is False + + repository = client.get( + f"/intelligence/v1/snapshots/{snapshot_id}/impact?nodeKey=repo:root", + headers=owner["headers"], + ) + assert repository.status_code == 200 + assert [item["nodeKey"] for item in repository.json()["dependencies"]["data"]] == ["dep:pypi:requests"] + + +def test_impact_query_enforces_depth_and_result_bounds(client, make_auth_headers, monkeypatch): + owner = make_auth_headers("impact-bounds@example.com") + _, snapshot_id = _seed_impact_snapshot(owner["user"]["id"]) + path = f"/intelligence/v1/snapshots/{snapshot_id}/impact?nodeKey=file:src/a.py" + + at_cap = client.get(f"{path}&depth={IMPACT_MAX_DEPTH}", headers=owner["headers"]) + too_deep = client.get(f"{path}&depth={IMPACT_MAX_DEPTH + 1}", headers=owner["headers"]) + zero_depth = client.get(f"{path}&depth=0", headers=owner["headers"]) + + assert at_cap.status_code == 200 + assert [item["nodeKey"] for item in at_cap.json()["dependencies"]["data"]] == [ + "file:src/b.py", + "file:src/d.py", + "file:src/c.py", + ] + assert len(at_cap.json()["dependencies"]["data"]) == len( + {item["nodeKey"] for item in at_cap.json()["dependencies"]["data"]} + ) + assert too_deep.status_code == zero_depth.status_code == 422 + assert too_deep.json()["code"] == zero_depth.json()["code"] == "request_validation_error" + + monkeypatch.setattr("app.intelligence.query_service.IMPACT_MAX_RESULTS_PER_DIRECTION", 1) + capped = client.get(f"{path}&depth=1", headers=owner["headers"]) + assert capped.status_code == 200 + assert [item["nodeKey"] for item in capped.json()["dependencies"]["data"]] == ["file:src/b.py"] + assert capped.json()["dependencies"]["limitReached"] is True + + +def test_impact_query_detects_cap_after_duplicate_paths_and_uses_canonical_hop(client, make_auth_headers): + """The cap applies after SQL deduplication, not to raw duplicate edges.""" + + owner = make_auth_headers("impact-duplicate-paths@example.com") + _, snapshot_id = _seed_duplicate_heavy_impact_snapshot(owner["user"]["id"]) + path = f"/intelligence/v1/snapshots/{snapshot_id}/impact?nodeKey=repo:root&depth=2" + + response = client.get(path, headers=owner["headers"]) + repeated = client.get(path, headers=owner["headers"]) + + assert response.status_code == repeated.status_code == 200 + assert response.json() == repeated.json() + dependencies = response.json()["dependencies"] + assert dependencies["limitReached"] is True + assert len(dependencies["data"]) == 100 + assert [item["nodeKey"] for item in dependencies["data"]] == [ + *(f"file:frontier/{number:02d}.py" for number in range(4)), + *(f"file:shared/{number:03d}.py" for number in range(96)), + ] + assert "file:overflow.py" not in {item["nodeKey"] for item in dependencies["data"]} + assert dependencies["data"][4]["via"]["subjectKey"] == "file:frontier/00.py" + assert dependencies["data"][4]["via"]["predicate"] == "imports" + + +@pytest.mark.parametrize( + "suffix", + [ + "", + "?nodeKey=", + "?nodeKey=file:src/a.py&depth=0", + f"?nodeKey=file:src/a.py&depth={IMPACT_MAX_DEPTH + 1}", + ], +) +def test_impact_query_rejects_malformed_parameters(client, make_auth_headers, suffix): + owner = make_auth_headers(f"impact-invalid-{uuid4().hex}@example.com") + _, snapshot_id = _seed_impact_snapshot(owner["user"]["id"]) + response = client.get(f"/intelligence/v1/snapshots/{snapshot_id}/impact{suffix}", headers=owner["headers"]) + + assert response.status_code == 422 + assert response.json()["code"] == "request_validation_error" + + +def test_impact_query_hides_unknown_nodes_and_cross_owner_snapshots(client, make_auth_headers): + owner = make_auth_headers("impact-visible@example.com") + other_owner = make_auth_headers("impact-hidden@example.com") + _, snapshot_id = _seed_impact_snapshot(owner["user"]["id"]) + path = f"/intelligence/v1/snapshots/{snapshot_id}/impact?nodeKey=file:src/missing.py" + + unknown_node = client.get(path, headers=owner["headers"]) + denied = client.get(path, headers=other_owner["headers"]) + missing = client.get( + "/intelligence/v1/snapshots/snap_missing/impact?nodeKey=file:src/missing.py", + headers=other_owner["headers"], + ) + + assert unknown_node.status_code == 404 + assert unknown_node.json()["code"] == "not_found" + assert unknown_node.json()["message"] == "Node not found in snapshot." + assert denied.status_code == missing.status_code == 404 + assert denied.json()["code"] == missing.json()["code"] == "not_found" + assert denied.json()["message"] == missing.json()["message"] == "Snapshot not found." + + def test_snapshot_query_openapi_documents_versioned_routes_and_schemas(client): document = client.get("/openapi.json").json() assert "/intelligence/v1/snapshots/{snapshot_id}/symbols" in document["paths"] + assert "/intelligence/v1/snapshots/{snapshot_id}/impact" in document["paths"] + assert "RiImpactResponse" in document["components"]["schemas"] assert "RiSymbolsResponse" in document["components"]["schemas"] assert "RiEvidenceResponse" in document["components"]["schemas"] assert document["components"]["schemas"]["RiSymbolsResponse"]["properties"]["schemaVersion"]["const"] == "ri.v1" diff --git a/apps/backend/tests/test_migrations.py b/apps/backend/tests/test_migrations.py index d056d76..450d1c1 100644 --- a/apps/backend/tests/test_migrations.py +++ b/apps/backend/tests/test_migrations.py @@ -86,6 +86,20 @@ def test_migrations_upgrade_and_downgrade_run_clean(tmp_path, monkeypatch): "revision_value", "config_hash", ] + edge_indexes = { + index["name"]: index["column_names"] + for index in inspect(probe_engine).get_indexes("ri_edges") + } + assert edge_indexes["ix_ri_edges_snapshot_subject_predicate"] == [ + "snapshot_id", + "subject_key", + "predicate", + ] + assert edge_indexes["ix_ri_edges_snapshot_object_predicate"] == [ + "snapshot_id", + "object_key", + "predicate", + ] command.downgrade(cfg, "base") assert "repositories" not in inspect(probe_engine).get_table_names() diff --git a/apps/backend/tests/test_openapi_contract.py b/apps/backend/tests/test_openapi_contract.py index 6e9339f..45159ef 100644 --- a/apps/backend/tests/test_openapi_contract.py +++ b/apps/backend/tests/test_openapi_contract.py @@ -39,6 +39,7 @@ ("GET", "/intelligence/v1/snapshots/{snapshot_id}"): {200, 401, 404, 422, 429, 500}, ("GET", "/intelligence/v1/snapshots/{snapshot_id}/symbols"): {200, 401, 404, 422, 429, 500}, ("GET", "/intelligence/v1/snapshots/{snapshot_id}/neighbours"): {200, 401, 404, 422, 429, 500}, + ("GET", "/intelligence/v1/snapshots/{snapshot_id}/impact"): {200, 401, 404, 422, 429, 500}, ("GET", "/intelligence/v1/snapshots/{snapshot_id}/references"): {200, 401, 404, 422, 429, 500}, ("GET", "/intelligence/v1/snapshots/{snapshot_id}/assertions"): {200, 401, 404, 422, 429, 500}, ("GET", "/intelligence/v1/snapshots/{snapshot_id}/paths"): {200, 401, 404, 422, 429, 500}, diff --git a/docs/architecture/REPOSITORY_INTELLIGENCE.md b/docs/architecture/REPOSITORY_INTELLIGENCE.md index 3b23b68..5046506 100644 --- a/docs/architecture/REPOSITORY_INTELLIGENCE.md +++ b/docs/architecture/REPOSITORY_INTELLIGENCE.md @@ -155,6 +155,17 @@ queries. Documentation and free-form AI context share a bounded immutable projection over the owner-scoped sealed snapshot for the repository's current revision. Missing or stale snapshots return the standard 404. +The authenticated `GET /intelligence/v1/snapshots/{snapshot_id}/impact` query +answers “what depends on this node?” and “what does this node depend on?” from +one named sealed snapshot. It follows only stored, resolved `imports` and +`depends_on` edges: outgoing edges are dependencies and incoming edges are +dependents. Every reached node carries the stored edge, evidence spans, and +immediate derivation that prove its traversal hop. The read is deterministic, +cycle-safe, and bounded to a requested depth from 1 through 10 and at most 100 +reached nodes in each direction. `limitReached: true` means that the result was +cut off at that node limit and must not be treated as exhaustive. The query +never reads a working tree, legacy metadata, or unresolved observations. + The architecture read is bounded to what the response actually renders (#133): the relationship predicates Architecture draws, the resolution diagnostics it displays, `classified_as` assertions, and the file, dependency and repository