feat: preserve native route assembly provenance

README.md

@@ -103,11 +103,14 @@ fastapi-endpoint-detector list --app path/to/main.py --format html -o endpoints.
 
 ### Machine-readable endpoint provenance
 
-JSON and YAML inventories and analysis reports use schema version 3. Each endpoint may
-include the additive, nullable `dependency_graph` field: schema-v1 bounded evidence for
-the declared FastAPI dependency tree, with explicit status and limitations. Trusted
-runtime extraction transports this evidence through the private protocol-v2 worker;
-consumers should use the public inventory/report schema rather than that worker protocol.
+JSON and YAML inventories and analysis reports use schema version 4. Each endpoint may
+include two additive nullable evidence fields. `dependency_graph` is schema-v1 bounded
+runtime evidence for the declared FastAPI dependency tree. `native_provenance` is
+schema-v1 secure-AST evidence for the snapshot side, selected root, exact decorator or
+imperative registration span, app/router constructor chain, and ordered include/mount
+chain. Source columns use Python AST offsets and end coordinates are exclusive. Trusted
+runtime extraction transports dependency evidence through the private protocol-v2
+worker; it does not guess source-level route assembly occurrences.
 
 ## Commands
 

src/fastapi_endpoint_detector/analyzer/change_mapper.py

@@ -42,6 +42,7 @@
     Endpoint,
     EndpointDiscoveryStatus,
     EndpointInventory,
+    SnapshotSide,
 )
 from fastapi_endpoint_detector.models.report import (
     AffectedEndpoint,
@@ -98,14 +99,33 @@
 }
 
 
-def _endpoint_result_key(endpoint: Endpoint) -> tuple[str, str, int, str, str]:
+EndpointResultKey = tuple[str, str, int, str, str, str]
+
+
+def _endpoint_result_key(endpoint: Endpoint) -> EndpointResultKey:
     handler = endpoint.handler
+    provenance = endpoint.native_provenance
+    occurrence = ""
+    if provenance is not None:
+        registration = provenance.registration
+        occurrence = ":".join(
+            (
+                provenance.side.value,
+                str(registration.source_span.file_path.resolve()),
+                str(registration.occurrence_order),
+                *(
+                    f"{edge.source_span.file_path.resolve()}@{edge.occurrence_order}"
+                    for edge in provenance.assembly_chain
+                ),
+            )
+        )
     return (
         endpoint.identifier,
         str(handler.file_path.resolve()),
         handler.line_number,
         handler.name,
         handler.module,
+        occurrence,
     )
 
 
@@ -207,7 +227,7 @@ def materialize(self) -> AffectedEndpoint:
 
 
 def _merge_affected(
-    accumulated: dict[tuple[str, str, int, str, str], _AffectedAccumulator],
+    accumulated: dict[EndpointResultKey, _AffectedAccumulator],
     candidate: AffectedEndpoint,
 ) -> None:
     if (
@@ -572,6 +592,7 @@ def baseline_registry(self) -> EndpointRegistry:
                 app_variable=self.app_variable,
                 app_entry=self.app_entry,
                 bootstrap_entry=self.bootstrap_entry,
+                snapshot_side=SnapshotSide.BASELINE,
             )
             self._baseline_registry = EndpointRegistry()
             native = extractor.extract_inventory()
@@ -878,14 +899,55 @@ def _analyze_diff_file(
             Tuple of (affected endpoints, processed added lines, processed removed lines).
             Processed lines are those that were matched to any endpoint.
         """
-        affected: dict[tuple[str, str, int, str, str], _AffectedAccumulator] = {}
+        affected: dict[EndpointResultKey, _AffectedAccumulator] = {}
         processed_added_lines: set[int] = set()
         processed_removed_lines: set[int] = set()
 
         # Get changed lines
         added_lines, removed_lines = DiffParser.get_changed_line_numbers(diff_file)
 
-        # Find endpoints in the changed file
+        # Native route registrations and exact include/mount/object occurrences own
+        # their materialized descendants. Only target additions are queried here:
+        # removed coordinates require the explicit baseline path handled by SCIP.
+        for endpoint, kinds, overlap in self.registry.get_structural_overlaps(
+            diff_file.path, set(added_lines)
+        ):
+            matched_kinds = ", ".join(kinds)
+            changed_line = min(overlap)
+            _merge_affected(
+                affected,
+                AffectedEndpoint(
+                    endpoint=endpoint,
+                    confidence=ConfidenceLevel.HIGH,
+                    reason=(
+                        f"Native route assembly occurrence modified ({matched_kinds}) "
+                        f"in {diff_file.path}"
+                    ),
+                    dependency_chain=[str(diff_file.path), *kinds],
+                    changed_files=[str(diff_file.path)],
+                    effect_evidence=[
+                        EffectEvidence(
+                            producer=EvidenceProducer.STRUCTURAL,
+                            status=EvidenceStatus.ESTABLISHED,
+                            effect=ChangeEffectKind.ROUTE_ASSEMBLY,
+                            channel=ImpactChannel.UNKNOWN,
+                            disposition=EffectDisposition.INTERNAL_EFFECT,
+                            summary=(
+                                "Changed source overlaps exact secure-AST route assembly "
+                                "provenance for this endpoint occurrence."
+                            ),
+                            changed_location=CodeReference(
+                                file_path=str(diff_file.path),
+                                line_number=changed_line,
+                                symbol=matched_kinds,
+                            ),
+                        )
+                    ],
+                ),
+            )
+            processed_added_lines.update(overlap)
+
+        # Find endpoints whose handlers are defined in the changed file.
         file_endpoints = self.registry.get_by_file(diff_file.path)
 
         # Check for direct handler changes
@@ -948,7 +1010,7 @@ def _analyze_with_scip(
         if has_removed:
             self.baseline_scip_analyzer.ensure_index(force=not self.use_cache)
 
-        affected: dict[tuple[str, str, int, str, str], _AffectedAccumulator] = {}
+        affected: dict[EndpointResultKey, _AffectedAccumulator] = {}
         orphan_evidence: dict[str, _OrphanAccumulator] = {}
         target_root = self.app_path.parent if self.app_path.is_file() else self.app_path
         baseline_root = (
@@ -969,6 +1031,55 @@ def target_equivalent(endpoint: Endpoint) -> Endpoint:
             ]
             return matches[0] if len(matches) == 1 else endpoint
 
+        def analyze_structural_side(
+            registry: EndpointRegistry,
+            file_path: Path,
+            lines: list[int],
+            side: str,
+        ) -> set[int]:
+            processed: set[int] = set()
+            for discovered, kinds, overlap in registry.get_structural_overlaps(
+                file_path, set(lines)
+            ):
+                # Structural evidence remains attached to its source snapshot.
+                # Lifecycle reconciliation is deliberately deferred rather than
+                # replacing a baseline occurrence by a same-identifier target.
+                endpoint = discovered
+                matched_kinds = ", ".join(kinds)
+                _merge_affected(
+                    affected,
+                    AffectedEndpoint(
+                        endpoint=endpoint,
+                        confidence=ConfidenceLevel.HIGH,
+                        reason=(
+                            f"Secure-AST {side} route assembly occurrence modified "
+                            f"({matched_kinds}) in {file_path}"
+                        ),
+                        dependency_chain=[str(file_path), *kinds],
+                        changed_files=[str(file_path)],
+                        effect_evidence=[
+                            EffectEvidence(
+                                producer=EvidenceProducer.STRUCTURAL,
+                                status=EvidenceStatus.ESTABLISHED,
+                                effect=ChangeEffectKind.ROUTE_ASSEMBLY,
+                                channel=ImpactChannel.UNKNOWN,
+                                disposition=EffectDisposition.INTERNAL_EFFECT,
+                                summary=(
+                                    f"Changed {side} source overlaps exact secure-AST route "
+                                    "assembly provenance for this endpoint occurrence."
+                                ),
+                                changed_location=CodeReference(
+                                    file_path=str(file_path),
+                                    line_number=min(overlap),
+                                    symbol=matched_kinds,
+                                ),
+                            )
+                        ],
+                    ),
+                )
+                processed.update(overlap)
+            return processed
+
         def analyze_side(
             analyzer: SCIPAnalyzer,
             registry: EndpointRegistry,
@@ -1060,25 +1171,45 @@ def analyze_side(
             added_lines, removed_lines = DiffParser.get_changed_line_numbers(diff_file)
             processed_added: set[int] = set()
             if diff_file.path.suffix == ".py" and added_lines:
-                processed_added = analyze_side(
-                    self.scip_analyzer,
-                    self.registry,
-                    target_root,
-                    diff_file.path,
-                    added_lines,
-                    "target",
+                processed_added.update(
+                    analyze_structural_side(
+                        self.registry,
+                        diff_file.path,
+                        added_lines,
+                        "target",
+                    )
+                )
+                processed_added.update(
+                    analyze_side(
+                        self.scip_analyzer,
+                        self.registry,
+                        target_root,
+                        diff_file.path,
+                        added_lines,
+                        "target",
+                    )
                 )
             processed_removed: set[int] = set()
             source_path = diff_file.source_path or diff_file.path
             if removed_lines and source_path.suffix == ".py":
                 assert baseline_root is not None
-                processed_removed = analyze_side(
-                    self.baseline_scip_analyzer,
-                    self.baseline_registry,
-                    baseline_root,
-                    source_path,
-                    removed_lines,
-                    "baseline",
+                processed_removed.update(
+                    analyze_structural_side(
+                        self.baseline_registry,
+                        source_path,
+                        removed_lines,
+                        "baseline",
+                    )
+                )
+                processed_removed.update(
+                    analyze_side(
+                        self.baseline_scip_analyzer,
+                        self.baseline_registry,
+                        baseline_root,
+                        source_path,
+                        removed_lines,
+                        "baseline",
+                    )
                 )
 
             reason = "Changed lines did not resolve through SCIP to a registered endpoint"
@@ -1427,7 +1558,7 @@ def report_progress(current: int, total: int, desc: str) -> None:
 
         # Analyze each Python file
         report_progress(70, 100, f"Checking {len(python_files)} changed files...")
-        all_affected: dict[tuple[str, str, int, str, str], _AffectedAccumulator] = {}
+        all_affected: dict[EndpointResultKey, _AffectedAccumulator] = {}
         orphan_evidence: dict[str, _OrphanAccumulator] = {}
 
         for i, diff_file in enumerate(python_files):

src/fastapi_endpoint_detector/analyzer/endpoint_registry.py

@@ -21,6 +21,7 @@ def __init__(self) -> None:
         self._by_path: dict[str, list[Endpoint]] = {}
         self._by_file: dict[Path, list[Endpoint]] = {}
         self._by_module: dict[str, list[Endpoint]] = {}
+        self._by_structural_file: dict[Path, list[tuple[Endpoint, str, int, int]]] = {}
 
     def register(self, endpoint: Endpoint) -> None:
         """
@@ -48,6 +49,34 @@ def register(self, endpoint: Endpoint) -> None:
             self._by_module[module] = []
         self._by_module[module].append(endpoint)
 
+        provenance = endpoint.native_provenance
+        if provenance is not None:
+            occurrences = [
+                (
+                    f"route_{provenance.registration.operation}",
+                    provenance.registration.source_span,
+                ),
+                *(
+                    (f"object_{item.object_kind}", item.source_span)
+                    for item in provenance.object_chain
+                ),
+                *(
+                    (f"assembly_{item.operation}", item.source_span)
+                    for item in provenance.assembly_chain
+                ),
+            ]
+            if provenance.root.bootstrap_span is not None:
+                occurrences.append(("bootstrap", provenance.root.bootstrap_span))
+            seen: set[tuple[str, Path, int, int]] = set()
+            for kind, span in occurrences:
+                identity = (kind, span.file_path, span.start_line, span.end_line)
+                if identity in seen:
+                    continue
+                seen.add(identity)
+                self._by_structural_file.setdefault(span.file_path, []).append(
+                    (endpoint, kind, span.start_line, span.end_line)
+                )
+
     def register_many(self, endpoints: list[Endpoint]) -> None:
         """
         Register multiple endpoints.
@@ -122,6 +151,48 @@ def _has_path_suffix(path: tuple[str, ...], suffix: tuple[str, ...]) -> bool:
         """Return whether *suffix* identifies whole trailing path components."""
         return bool(suffix) and len(path) >= len(suffix) and path[-len(suffix) :] == suffix
 
+    def get_structural_overlaps(
+        self,
+        file_path: Path | str,
+        changed_lines: set[int],
+    ) -> list[tuple[Endpoint, tuple[str, ...], set[int]]]:
+        """Return exact native assembly occurrences intersecting target-side lines."""
+        if not changed_lines:
+            return []
+        query = Path(file_path)
+        try:
+            resolved = query.resolve()
+        except OSError:
+            resolved = query
+        if resolved in self._by_structural_file:
+            buckets = [self._by_structural_file[resolved]]
+        elif query in self._by_structural_file:
+            buckets = [self._by_structural_file[query]]
+        else:
+            query_parts = self._path_parts(query)
+            buckets = [
+                occurrences
+                for registered_path, occurrences in self._by_structural_file.items()
+                if self._has_path_suffix(self._path_parts(registered_path), query_parts)
+            ]
+        if len(buckets) != 1:
+            return []
+        matches: dict[int, tuple[Endpoint, set[str], set[int]]] = {}
+        for endpoint, kind, start_line, end_line in buckets[0]:
+            overlap = {line for line in changed_lines if start_line <= line <= end_line}
+            if not overlap:
+                continue
+            key = id(endpoint)
+            current = matches.get(key)
+            if current is None:
+                matches[key] = (endpoint, {kind}, overlap)
+            else:
+                current[1].add(kind)
+                current[2].update(overlap)
+        return [
+            (endpoint, tuple(sorted(kinds)), lines) for endpoint, kinds, lines in matches.values()
+        ]
+
     def get_by_module(self, module: str) -> list[Endpoint]:
         """
         Get endpoints defined in a specific module.

src/fastapi_endpoint_detector/models/__init__.py

@@ -60,7 +60,17 @@
     EndpointMethod,
     HandlerInfo,
     InventoryStatus,
+    NativeAssemblyMode,
+    NativeRegistrationKind,
+    NativeRootSelectionKind,
+    NativeRouteAssemblyEdgeEvidence,
+    NativeRouteObjectEvidence,
+    NativeRouteProvenance,
+    NativeRouteRegistrationEvidence,
+    NativeRouteRootEvidence,
+    NativeSourceSpan,
     RouteActivationEvidence,
+    SnapshotSide,
     SurfaceRegistrationEvidence,
 )
 from fastapi_endpoint_detector.models.report import (
@@ -131,7 +141,17 @@
     "EndpointMethod",
     "HandlerInfo",
     "InventoryStatus",
+    "NativeAssemblyMode",
+    "NativeRegistrationKind",
+    "NativeRootSelectionKind",
+    "NativeRouteAssemblyEdgeEvidence",
+    "NativeRouteObjectEvidence",
+    "NativeRouteProvenance",
+    "NativeRouteRegistrationEvidence",
+    "NativeRouteRootEvidence",
+    "NativeSourceSpan",
     "RouteActivationEvidence",
+    "SnapshotSide",
     "SurfaceRegistrationEvidence",
     # Dependency models
     "Dependency",