fix(impact): close fail-open gaps in the offline gate (logic diff, catalog freshness, scope-git, selection)

docs/features/impact-analysis.md

@@ -127,6 +127,15 @@ for the exact rule and the honesty invariants.
     additive: it never posts a comment and never changes the exit code — gating stays with
     `fail-on`.
 
+!!! warning "Tests attached to skipped models do not run"
+    The selector selects **models**, and dbt only runs the tests attached to the models it
+    selects. A test that lives on a *skipped* model but references a *rebuilt* one — most
+    commonly a `relationships` test on a skipped child pointing at a rebuilt parent — will
+    **not** run in a selective CI build, even though the rebuilt side may have just broken it.
+    If such cross-model tests matter to your gate, widen the selection (e.g.
+    `dbt build --select "$SELECTOR" --indirect-selection=buildable`, or append `+1` to the
+    selector), or run a full `dbt test` on a schedule to catch what the selective build skipped.
+
 ## Use Cases
 
 **Before modifying a column:**

parrant/artifacts/manifest.py

@@ -49,7 +49,7 @@ def __init__(self, manifest_path: str | None = None):
     def load(self) -> None:
         if not self.manifest_path or not self.manifest_path.exists():
             raise FileNotFoundError(f"Manifest file not found: {self.manifest_path}")
-        with open(self.manifest_path, "r") as f:
+        with open(self.manifest_path) as f:
             self.manifest = json.load(f)
 
     def get_adapter(self) -> str | None:
@@ -80,6 +80,55 @@ def get_model_dependencies(self) -> dict[str, set[str]]:
             dependencies[model_id] = depends_on
         return dependencies
 
+    def get_macro_dependents(self) -> dict[str, set[str]]:
+        """Map each macro *file* to the model-like nodes whose compiled SQL it can affect.
+
+        Keys are the macros' normalized ``original_file_path`` (e.g. ``macros/money.sql``);
+        values are lowercased names of nodes (models/snapshots/seeds) that depend on a macro
+        in that file — directly (``node.depends_on.macros``) or transitively (a macro that a
+        used macro itself calls, via the ``macros[*].depends_on.macros`` graph). Feeds the
+        ``--scope-git`` fail-safe: a changed macro file scopes to exactly these dependents.
+        """
+        macros = self.manifest.get("macros", {})
+        macro_calls: dict[str, list[str]] = {
+            macro_id: list(macro.get("depends_on", {}).get("macros", []))
+            for macro_id, macro in macros.items()
+        }
+
+        # For each macro id, the closure of macros it (transitively) calls, itself included.
+        # Iterative BFS with a visited set so macro-call cycles terminate.
+        def _closure(macro_id: str) -> set[str]:
+            seen: set[str] = set()
+            stack = [macro_id]
+            while stack:
+                current = stack.pop()
+                if current in seen:
+                    continue
+                seen.add(current)
+                stack.extend(macro_calls.get(current, []))
+            return seen
+
+        def _norm(path: str) -> str:
+            return re.sub(r"^\./", "", path.strip()).lstrip("/")
+
+        dependents: dict[str, set[str]] = {
+            _norm(macro.get("original_file_path") or ""): set()
+            for macro in macros.values()
+            if macro.get("original_file_path")
+        }
+        for node in self.manifest.get("nodes", {}).values():
+            if node.get("resource_type") not in ("model", "snapshot", "seed"):
+                continue
+            node_name = (node.get("name") or "").lower()
+            if not node_name:
+                continue
+            for macro_id in node.get("depends_on", {}).get("macros", []):
+                for reached in _closure(macro_id):
+                    path = (macros.get(reached) or {}).get("original_file_path")
+                    if path:
+                        dependents[_norm(path)].add(node_name)
+        return dependents
+
     def get_model_upstream(self) -> dict[str, set[str]]:
         """Get upstream dependencies for each model."""
         upstream: dict[str, set[str]] = {}
@@ -109,7 +158,10 @@ def get_model_upstream(self) -> dict[str, set[str]]:
                             # Fallback to source name if identifier not found
                             source_name = parts[-1].lower()
                             upstream[model_name].add(source_name)
-                    elif parts[0] == "snapshot":
+                    elif parts[0] in ("snapshot", "seed"):
+                        # Seeds are upstream nodes like any other: without this edge a model
+                        # ref()ing a seed records no dependency, so the seed's consumers are
+                        # invisible to reachability and to the rebuild selection.
                         dep_name = parts[-1].lower()
                         upstream[model_name].add(dep_name)
 

parrant/artifacts/registry.py

@@ -1,3 +1,5 @@
+import hashlib
+import json
 import logging
 from dataclasses import dataclass, field
 from typing import Any
@@ -10,6 +12,8 @@
 )
 from parrant.artifacts.manifest import ManifestReader
 from parrant.models.schema import (
+    ArtifactStamp,
+    ArtifactStamps,
     Column,
     ColumnLineage,
     Coverage,
@@ -97,6 +101,8 @@ def __init__(
         # attached to it AND relationships tests referencing it. Column-level recovery can
         # miss a model's tested columns, but a wholly-removed model breaks all of its tests.
         self._model_tests: dict[str, list[TestNode]] = {}
+        # Lazily-built macro file -> dependent model-like nodes (see get_macro_dependents).
+        self._macro_dependents: dict[str, set[str]] | None = None
 
     @property
     def is_loaded(self) -> bool:
@@ -344,6 +350,19 @@ def get_column_tests(self, model: str, column: str) -> list[TestNode]:
         """
         return list(self._column_tests.get((model.lower(), column.lower()), []))
 
+    def get_macro_dependents(self) -> dict[str, set[str]]:
+        """Macro file path -> model-like nodes whose compiled SQL that file can affect.
+
+        Capability method consumed by the ``--scope-git`` fail-safe (looked up via
+        ``getattr``, not part of the provider protocols): a changed macro file scopes to
+        exactly its (transitive) dependents instead of disabling scoping. Delegates to the
+        manifest's macro graph; cached after the first call (the manifest is immutable
+        post-load).
+        """
+        if self._macro_dependents is None:
+            self._macro_dependents = self._manifest_reader.get_macro_dependents()
+        return self._macro_dependents
+
     def get_tests_referencing(self, model: str, column: str) -> list[TestNode]:
         """Return relationships tests whose *referenced* (parent) side is ``model.column``.
 
@@ -761,6 +780,39 @@ def get_dialect(self) -> str | None:
         """
         return self._dialect
 
+    def get_artifact_stamps(self) -> ArtifactStamps:
+        """Identity stamps (``generated_at`` / ``invocation_id``) of the loaded artifacts.
+
+        Read from each artifact's ``metadata`` block; a missing block yields empty stamps
+        (no identity claim). Consumed by the two-manifest diff to detect a stale or shared
+        catalog and stamp the report ``structural_diff: degraded`` instead of staying silent.
+        """
+        catalog_meta = (self._catalog_reader.catalog or {}).get("metadata") or {}
+        manifest_meta = (self._manifest_reader.manifest or {}).get("metadata") or {}
+        return ArtifactStamps(
+            catalog=ArtifactStamp(
+                generated_at=catalog_meta.get("generated_at"),
+                invocation_id=catalog_meta.get("invocation_id"),
+            ),
+            manifest=ArtifactStamp(
+                generated_at=manifest_meta.get("generated_at"),
+                invocation_id=manifest_meta.get("invocation_id"),
+            ),
+        )
+
+    def get_catalog_fingerprint(self) -> str | None:
+        """Content fingerprint (sha256 over canonical JSON) of the loaded ``catalog.json``.
+
+        Byte-formatting-insensitive: two catalogs with the same CONTENT (the real signal —
+        e.g. one prod catalog mounted on both diff sides) fingerprint equal even if
+        re-serialized. ``None`` when no catalog content is loaded (no identity claim).
+        """
+        content = self._catalog_reader.catalog
+        if not content:
+            return None
+        canonical = json.dumps(content, sort_keys=True, separators=(",", ":"))
+        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
     def get_model(self, model_name: str) -> Model:
         """Get a specific model by name."""
         if not self.is_loaded:
@@ -949,7 +1001,7 @@ def _find_compiled_sql(self, model_name: str) -> str | None:
         compiled_path = self._manifest_reader.get_model_path(model_name)
         if compiled_path:
             try:
-                with open(compiled_path, "r") as f:
+                with open(compiled_path) as f:
                     compiled_sql = f.read()
                 model.compiled_sql = compiled_sql
                 return compiled_sql

parrant/cli/main.py

@@ -12,7 +12,7 @@
     OverrideResolution,
     build_changeset_report,
     build_git_changeset,
-    git_changed_models,
+    resolve_git_scope,
     scope_changes_to_models,
 )
 from parrant.lineage.display import DotDisplay, JsonDisplay, TextDisplay
@@ -27,6 +27,7 @@
     decide_verdict,
     ineffective_overrides,
 )
+from parrant.models.schema import GitScopeStatus
 
 logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")
 
@@ -587,6 +588,9 @@ def impact(
 
         base_service: LineageService | None = None
         changes: list[ColumnChange]
+        # Populated only under --scope-git: whether the git intersection was applied or
+        # disabled fail-safe (unmappable logic-bearing files). Rendered as report["scope_git"].
+        scope_status: GitScopeStatus | None = None
         # Whether structural checks (added/removed/type_changed) could run. They need a
         # real catalog on both sides; the two-manifest path decides this from the builder
         # below. The git-diff fallback is a separate, self-evident coarse mode, so it is
@@ -630,10 +634,31 @@ def impact(
             if scope_git:
                 # Intersect the precise two-manifest changeset with the models the
                 # branch actually touched, so a stale base artifact can't leak
-                # already-merged changes into the report.
-                scoped_models = git_changed_models(head_service.registry, scope_git)
-                changes = scope_changes_to_models(changes, scoped_models)
-                source = f"two-manifest scoped to git-diff ({scope_git})"
+                # already-merged changes into the report. FAIL-SAFE: when the diff
+                # contains a logic-bearing file that cannot be pinned to models (a
+                # macro, an unmapped Python model, dbt_project.yml vars, ...), the
+                # narrowing is unprovable — keep the FULL changeset and surface the
+                # reason, instead of silently dropping real changes into a SAFE.
+                scope = resolve_git_scope(head_service.registry, scope_git)
+                if scope.unmappable:
+                    reason = "disabled — unmappable changed files: [{}]".format(
+                        ", ".join(scope.unmappable)
+                    )
+                    click.echo(f"Warning: --scope-git {reason}", err=True)
+                    source = "two-manifest (git scoping disabled: unmappable changed files)"
+                    scope_status = GitScopeStatus(
+                        base=scope_git,
+                        applied=False,
+                        changed_models=sorted(scope.models),
+                        unmappable_files=list(scope.unmappable),
+                        reason=reason,
+                    )
+                else:
+                    changes = scope_changes_to_models(changes, scope.models)
+                    source = f"two-manifest scoped to git-diff ({scope_git})"
+                    scope_status = GitScopeStatus(
+                        base=scope_git, applied=True, changed_models=sorted(scope.models)
+                    )
         elif git_base:
             override_collector = OverrideResolution()
             changes = build_git_changeset(
@@ -680,6 +705,8 @@ def impact(
         report = build_changeset_report(source, changes, aggregated)
         report["coverage"] = head_service.get_coverage().model_dump()
         report["structural_checks_available"] = structural_checks_available
+        if scope_status is not None:
+            report["scope_git"] = scope_status.model_dump()
 
         # Provable breaks + the SAFE/REVIEW/BLOCK ruling. Base registry (when present) is the
         # reliable source of the tests that existed before the change.
@@ -913,7 +940,7 @@ def policy_test(
 
         if baseline_path:
             try:
-                with open(baseline_path, "r", encoding="utf-8") as handle:
+                with open(baseline_path, encoding="utf-8") as handle:
                     baseline = BacktestReport.model_validate_json(handle.read())
             except Exception as exc:
                 raise click.ClickException(

parrant/lineage/backtest.py

@@ -21,6 +21,7 @@
 
 import json
 import os
+import statistics
 import subprocess
 import sys
 from collections import defaultdict
@@ -40,9 +41,11 @@
     BacktestPointResult,
     BacktestReport,
     BacktestRuleStat,
+    BacktestSelectionStats,
     GateDecision,
     Policy,
     PolicyVerdict,
+    ResolutionReasonCount,
     SemanticChangeKind,
 )
 
@@ -148,6 +151,7 @@ def _replay_point(
         )
         for hit in verdict.hits
     ]
+    widened, skippable_count, forced, reasons = _selection_facts(aggregated)
     point = BacktestPointResult(
         ref=ref,
         source=source,
@@ -158,10 +162,84 @@ def _replay_point(
         blast_radius=blast_radius,
         fired=fired,
         sample_reach=sample_reach,
+        selection_widened=widened,
+        skippable_models_count=skippable_count,
+        rebuild_forced_by_nonresolution=forced,
+        resolution_reasons=reasons,
     )
     return point, verdict
 
 
+def _selection_facts(
+    aggregated: Any,
+) -> tuple[bool | None, int | None, int | None, dict[str, int]]:
+    """Extract the per-point selection/resolution facts for the widen-rate stats.
+
+    Pure observation of the already-computed ``selection`` / ``resolution_summary`` blocks —
+    nothing here influences selection or gating. Returns all-``None``/empty when the impact
+    carried no selection block (e.g. a stub service without a registry), so such points are
+    honestly excluded from the aggregate rather than counted as "not widened".
+    """
+    if not isinstance(aggregated, dict):
+        return None, None, None, {}
+    selection = aggregated.get("selection")
+    if not isinstance(selection, dict):
+        return None, None, None, {}
+    widened = bool(selection.get("widened_to_all_reachable", False))
+    skippable_count = len(selection.get("skippable_models") or [])
+
+    forced: int | None = None
+    reasons: dict[str, int] = {}
+    summary = aggregated.get("resolution_summary")
+    if isinstance(summary, dict):
+        forced = int(summary.get("rebuild_forced_by_nonresolution", 0))
+        for entry in summary.get("top_reasons") or []:
+            if isinstance(entry, dict) and entry.get("reason") is not None:
+                reasons[str(entry["reason"])] = int(entry.get("count", 0))
+    return widened, skippable_count, forced, reasons
+
+
+# Cap the aggregated top-reasons list so a reason-rich range stays a readable breakdown.
+_TOP_REASONS_CAP = 10
+
+
+def _aggregate_selection_stats(
+    points: list[BacktestPointResult],
+) -> BacktestSelectionStats | None:
+    """Roll the per-point selection facts up into the range-wide widen-rate statistics.
+
+    Only points that actually carried a selection block participate (``selection_widened is
+    not None``); a range with none returns ``None`` so the report omits the surface instead of
+    fabricating a 0% widen rate. Median is over ``skippable_models_count``; reasons are summed
+    across points and ranked by frequency (ties broken alphabetically for determinism).
+    """
+    with_selection = [p for p in points if p.selection_widened is not None]
+    if not with_selection:
+        return None
+
+    widened_points = sum(1 for p in with_selection if p.selection_widened)
+    skippable_counts = [p.skippable_models_count or 0 for p in with_selection]
+    forced_total = sum(p.rebuild_forced_by_nonresolution or 0 for p in with_selection)
+
+    reason_totals: dict[str, int] = defaultdict(int)
+    for point in with_selection:
+        for reason, count in point.resolution_reasons.items():
+            reason_totals[reason] += count
+    top_reasons = [
+        ResolutionReasonCount(reason=reason, count=count)
+        for reason, count in sorted(reason_totals.items(), key=lambda kv: (-kv[1], kv[0]))
+    ][:_TOP_REASONS_CAP]
+
+    return BacktestSelectionStats(
+        points_with_selection=len(with_selection),
+        widened_points=widened_points,
+        widen_rate_pct=round(100.0 * widened_points / len(with_selection), 1),
+        median_skippable_models=float(statistics.median(skippable_counts)),
+        rebuild_forced_by_nonresolution_total=forced_total,
+        top_reasons=top_reasons,
+    )
+
+
 def _aggregate_rule_stats(
     points_verdicts: list[tuple[BacktestPointResult, PolicyVerdict]],
     policy: Policy,
@@ -231,7 +309,7 @@ def _load_changesets_dir(path: str) -> list[tuple[str, list[ColumnChange]]]:
         if not name.endswith(".json"):
             continue
         full = os.path.join(path, name)
-        with open(full, "r", encoding="utf-8") as handle:
+        with open(full, encoding="utf-8") as handle:
             data = json.load(handle)
         entries = _extract_change_entries(data)
         out.append((name, changes_from_dicts(entries)))
@@ -462,6 +540,7 @@ def _assemble_report(
         fidelity_note=fidelity_note,
         warnings=warnings,
         baseline_delta=baseline_delta,
+        selection_stats=_aggregate_selection_stats(points),
     )