diff --git a/docs/features/impact-analysis.md b/docs/features/impact-analysis.md
index f235b0f..d5b26ec 100644
--- a/docs/features/impact-analysis.md
+++ b/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:**
diff --git a/parrant/artifacts/manifest.py b/parrant/artifacts/manifest.py
index 07e1f38..59f164a 100644
--- a/parrant/artifacts/manifest.py
+++ b/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)
diff --git a/parrant/artifacts/registry.py b/parrant/artifacts/registry.py
index 09b4f2a..d46c960 100644
--- a/parrant/artifacts/registry.py
+++ b/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
diff --git a/parrant/cli/main.py b/parrant/cli/main.py
index 7b96fc0..00b8df6 100644
--- a/parrant/cli/main.py
+++ b/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(
diff --git a/parrant/lineage/backtest.py b/parrant/lineage/backtest.py
index f7cd2c8..bf92575 100644
--- a/parrant/lineage/backtest.py
+++ b/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),
)
diff --git a/parrant/lineage/changeset.py b/parrant/lineage/changeset.py
index 3b5e7c0..9017e92 100644
--- a/parrant/lineage/changeset.py
+++ b/parrant/lineage/changeset.py
@@ -15,6 +15,7 @@
import re
import subprocess
from dataclasses import dataclass, field
+from datetime import datetime
from enum import Enum
from typing import Any
@@ -22,15 +23,17 @@
from parrant.lineage.semantic_diff import (
_UNPARSEABLE_PREFIX,
canonical_key,
+ comment_free_token_signature,
compare_expressions,
)
from parrant.models.schema import (
+ LogicDiffStatus,
OverrideDirective,
OverrideVerb,
SemanticChangeKind,
+ StructuralDiffStatus,
)
from parrant.parser.sql_parser import parse_override_directives
-from parrant.parser.sql_parser_utils import strip_sql_comments
logger = logging.getLogger(__name__)
@@ -87,6 +90,11 @@ class ColumnChange:
# equality/hashing (``compare=False``) so it never perturbs the sort key or dedup, and so a
# frozen ``ColumnChange`` stays hashable even though ``OverrideDirective`` (pydantic) is not.
override: OverrideDirective | None = field(default=None, compare=False)
+ # Set to ``"indeterminate"`` when the model-level logic diff was IMPOSSIBLE (compiled SQL
+ # unavailable on a side for a model node): "no logic change" was unprovable, so the change
+ # is emitted fail-safe rather than silently dropped. ``None`` for every ordinary change.
+ # Consumed by the service to degrade confidence and widen the rebuild selection.
+ logic_diff_status: LogicDiffStatus | None = None
def to_dict(self) -> dict[str, object]:
payload: dict[str, object] = {
@@ -104,6 +112,10 @@ def to_dict(self) -> dict[str, object]:
"base": self.base_expression,
"head": self.head_expression,
}
+ # Only attach ``logic_diff_status`` when set (the indeterminate missing-SQL case), so
+ # ordinary changes keep their existing JSON shape byte-for-byte.
+ if self.logic_diff_status is not None:
+ payload["logic_diff_status"] = self.logic_diff_status
# Only attach ``override`` when one is present, so JSON stays byte-stable when absent.
# Flows into service.by_change automatically (by_change spreads change.to_dict()).
if self.override is not None:
@@ -117,18 +129,6 @@ def to_dict(self) -> dict[str, object]:
return payload
-def _normalize_sql(sql: str | None) -> str | None:
- """Normalize compiled SQL so cosmetic reformatting isn't read as a logic change.
-
- ``strip_sql_comments`` already removes comments and collapses whitespace runs,
- which is exactly the noise we want to ignore when deciding whether the logic
- that produces a model actually changed.
- """
- if sql is None:
- return None
- return strip_sql_comments(sql)
-
-
def _registry_dialect(registry: object) -> str | None:
"""Best-effort SQL dialect from a registry, ``None`` when it exposes no getter.
@@ -343,13 +343,21 @@ def record(change: ColumnChange) -> None:
)
)
+ # Logic diff: three-state, never silently safe. ``indeterminate`` (a MODEL whose
+ # compiled SQL is unavailable on a side) emits fail-safe changes below — treating
+ # it as "no change" would hide the edit AND leave the model out of the rebuild.
+ logic_status = self._logic_diff_status(model_name, head_model)
+ if logic_status == "indeterminate":
+ for change in self._indeterminate_logic_changes(model_name, head_model):
+ record(change)
+
# Logic change: the model's compiled SQL differs. Rather than flag EVERY output
# column — which floods the downstream blast radius with unrelated pass-throughs
# (editing one column must not implicate every other column of the model) — diff each output
# column's derivation between base and head and flag ONLY the columns that actually
# changed. Falls back to flagging all columns when neither side exposes per-column
# lineage (nothing to diff precisely).
- if self._logic_changed(model_name):
+ if logic_status == "changed":
classified = self._logic_changed_columns(base_model, head_model)
if not classified:
# The compiled SQL provably changed, yet no output column's per-column
@@ -419,12 +427,61 @@ def structural_diff_available(self) -> bool:
def _side_has_catalog(registry: LineageProvider) -> bool:
return any(registry.is_catalog_backed(name) for name in registry.get_models())
- def _logic_changed(self, model_name: str) -> bool:
- base_sql = _normalize_sql(self._safe_compiled_sql(self.base, model_name))
- head_sql = _normalize_sql(self._safe_compiled_sql(self.head, model_name))
+ def _logic_diff_status(self, model_name: str, head_model: Any) -> LogicDiffStatus:
+ """Three-state logic diff for a node present on both sides — never silently safe.
+
+ ``"changed"`` / ``"unchanged"`` when both sides expose compiled SQL to compare.
+ When either side's compiled SQL is unavailable the answer depends on the node kind:
+ sources / seeds / snapshots legitimately have no logic to diff (``"unchanged"``),
+ but a MODEL without compiled SQL cannot be proven unchanged — that is
+ ``"indeterminate"``, and the caller emits fail-safe changes for it. The resource
+ type is read defensively (``getattr``) so lightweight stubs without one keep the
+ historical no-op behaviour; every real registry model carries it.
+ """
+ base_sql = self._safe_compiled_sql(self.base, model_name)
+ head_sql = self._safe_compiled_sql(self.head, model_name)
if not base_sql or not head_sql:
- return False
- return base_sql != head_sql
+ if getattr(head_model, "resource_type", None) == "model":
+ return "indeterminate"
+ return "unchanged"
+ if base_sql == head_sql:
+ return "unchanged"
+ # Gate on the string-literal-safe token signature, NOT a regex comment strip: the
+ # tokenizer keeps string/quoted-identifier CONTENTS intact, so a `--` or `/*` inside a
+ # quoted string is data — a regex strip would swallow the real tokens around it and
+ # silently report "no logic change" (a false SAFE). Equal signatures prove the edit is
+ # comment/whitespace-only; anything else — including a side we cannot even tokenize —
+ # is treated as changed (fail-safe; the per-column diff classifies it precisely).
+ base_sig = comment_free_token_signature(base_sql, self._dialect)
+ head_sig = comment_free_token_signature(head_sql, self._dialect)
+ if base_sig is not None and head_sig is not None:
+ return "changed" if base_sig != head_sig else "unchanged"
+ return "changed"
+
+ def _indeterminate_logic_changes(self, model_name: str, head_model: Any) -> list[ColumnChange]:
+ """Fail-safe changes for a model whose logic diff was impossible (missing SQL).
+
+ Every head output column is flagged ``LOGIC_CHANGED`` / ``INDETERMINATE`` with the
+ ``logic_diff_status`` marker so the service degrades confidence and widens the rebuild
+ selection. A column-less model (no catalog entry AND no SQL to parse) still emits one
+ ``*`` sentinel change — an unprovable model must never vanish from the report.
+ """
+ reason = (
+ f"compiled SQL for model '{model_name}' is unavailable on at least one side — "
+ "the logic diff is impossible, so 'no logic change' cannot be proven (fail-safe)"
+ )
+ columns = sorted(head_model.columns) or ["*"]
+ return [
+ ColumnChange(
+ model_name,
+ column,
+ ChangeKind.LOGIC_CHANGED,
+ semantic=SemanticChangeKind.INDETERMINATE,
+ reason=reason,
+ logic_diff_status="indeterminate",
+ )
+ for column in columns
+ ]
def _logic_changed_columns(self, base_model, head_model) -> dict[str, _ColumnDiff]:
"""Which output columns changed derivation, each with a semantic classification.
@@ -596,6 +653,90 @@ def _safe_compiled_sql(registry: LineageProvider, model_name: str) -> str | None
return None
+def _provider_call(provider: object, method: str) -> Any:
+ """Call a zero-arg accessor on a provider defensively (``None`` when absent/failing).
+
+ Mirrors :func:`_registry_dialect`: a real ``ModelRegistry`` answers, a lightweight test
+ stub without the accessor simply makes no claim rather than raising.
+ """
+ getter = getattr(provider, method, None)
+ if not callable(getter):
+ return None
+ try:
+ return getter()
+ except Exception:
+ return None
+
+
+def _parse_artifact_timestamp(raw: str | None) -> datetime | None:
+ """Parse a dbt ``generated_at`` ISO timestamp, ``None`` when absent/unparseable."""
+ if not raw or not isinstance(raw, str):
+ return None
+ try:
+ # dbt stamps UTC with a trailing ``Z``, which py3.10's fromisoformat rejects.
+ return datetime.fromisoformat(raw.replace("Z", "+00:00"))
+ except ValueError:
+ return None
+
+
+def detect_structural_degradation(
+ base: LineageProvider, head: LineageProvider
+) -> StructuralDiffStatus | None:
+ """Detect a base/head catalog pair that structurally CANNOT show column changes.
+
+ Two real-deployment failure modes make the structural (added/removed/type_changed) diff
+ silently blind while ``structural_diff_available()`` still reports True:
+
+ * **Identical catalogs** — one prod ``catalog.json`` mounted on both sides. Diffing a
+ catalog against itself can never surface a removed or retyped column, so
+ ``provable_break_count`` is stuck at 0 and ``--fail-on tests`` can never fire.
+ * **Stale head catalog** — the head ``catalog.json`` predates the head manifest (and comes
+ from a different dbt invocation): its column truth describes an older build.
+
+ Returns a ``degraded`` stamp with a human-readable reason, or ``None`` when neither
+ condition is provable (providers without artifact identity — e.g. stubs — make no claim).
+ Advisory only: consumed by the report/JSON/markdown surfaces, never by exit codes.
+ """
+ base_fingerprint = _provider_call(base, "get_catalog_fingerprint")
+ head_fingerprint = _provider_call(head, "get_catalog_fingerprint")
+ if base_fingerprint is not None and base_fingerprint == head_fingerprint:
+ return StructuralDiffStatus(
+ status="degraded",
+ reason=(
+ "base and head catalog.json are identical — one catalog backs both sides, so "
+ "removed/type_changed columns can never be detected"
+ ),
+ )
+
+ stamps = _provider_call(head, "get_artifact_stamps")
+ if stamps is not None:
+ catalog_stamp = stamps.catalog
+ manifest_stamp = stamps.manifest
+ same_invocation = (
+ catalog_stamp.invocation_id is not None
+ and catalog_stamp.invocation_id == manifest_stamp.invocation_id
+ )
+ if not same_invocation:
+ catalog_ts = _parse_artifact_timestamp(catalog_stamp.generated_at)
+ manifest_ts = _parse_artifact_timestamp(manifest_stamp.generated_at)
+ try:
+ stale = (
+ catalog_ts is not None and manifest_ts is not None and catalog_ts < manifest_ts
+ )
+ except TypeError: # mixed aware/naive timestamps — no provable claim
+ stale = False
+ if stale:
+ return StructuralDiffStatus(
+ status="degraded",
+ reason=(
+ f"head catalog.json (generated_at {catalog_stamp.generated_at}) predates "
+ f"the head manifest ({manifest_stamp.generated_at}) — its column truth "
+ "is stale, so removed/type_changed detection is unreliable"
+ ),
+ )
+ return None
+
+
def _path_to_model_map(head: LineageProvider) -> dict[str, str]:
"""Map each model's ``resource_path`` (dbt ``original_file_path``) to its name."""
mapping: dict[str, str] = {}
@@ -763,10 +904,132 @@ def scope_changes_to_models(changes: list[ColumnChange], models: set[str]) -> li
return [change for change in changes if change.model in models]
+# Changed files that can influence models' compiled SQL. A change to one of these that
+# cannot be pinned to specific models makes git scoping unsound (see resolve_git_scope):
+# *.sql — a model (mappable) OR a macro/generic test that can rewrite ANY model's SQL
+# *.py — a Python model (mappable when in the manifest; otherwise unknown logic)
+# *.csv — a seed: column changes ripple into downstream models without any .sql edit
+_LOGIC_BEARING_SUFFIXES = (".sql", ".py", ".csv")
+# Project-level files whose edits (vars, packages, profiles/targets) can change the
+# compiled SQL of arbitrarily many models and are never mappable to one model.
+_LOGIC_BEARING_BASENAMES = frozenset(
+ {
+ "dbt_project.yml",
+ "dbt_project.yaml",
+ "packages.yml",
+ "packages.yaml",
+ "package-lock.yml",
+ "dependencies.yml",
+ "profiles.yml",
+ "profiles.yaml",
+ }
+)
+
+
+@dataclass
+class GitScopeResolution:
+ """How the git diff resolved for ``--scope-git``.
+
+ ``models`` is the union of models the diff could be pinned to (model files by
+ ``resource_path``, macros by their dependent models when the backend can map them).
+ ``unmappable`` lists logic-bearing changed files that could NOT be pinned to models;
+ when non-empty, narrowing the changeset to ``models`` is unsound and the caller must
+ disable scoping for the run (fail-safe) rather than silently drop changes.
+ Docs/property-only files (schema.yml, *.md, ...) influence neither compiled SQL nor
+ the catalog, so they are ignored and never disable scoping.
+ """
+
+ models: set[str] = field(default_factory=set)
+ unmappable: list[str] = field(default_factory=list)
+
+
+def _is_logic_bearing(path: str) -> bool:
+ """Whether a changed file can influence some model's compiled SQL / columns."""
+ basename = path.rsplit("/", 1)[-1].lower()
+ return basename in _LOGIC_BEARING_BASENAMES or basename.endswith(_LOGIC_BEARING_SUFFIXES)
+
+
+def _macro_dependents_or_empty(head: LineageProvider) -> dict[str, set[str]]:
+ """Macro-file -> dependent-model map, when the backend can provide it.
+
+ Capability lookup (not part of the :class:`LineageProvider` protocol): backends that
+ cannot map macros return ``{}``, which downgrades macro changes to *unmappable* — the
+ fail-safe floor — instead of crashing or, worse, silently narrowing.
+ """
+ getter = getattr(head, "get_macro_dependents", None)
+ if not callable(getter):
+ return {}
+ try:
+ dependents = getter()
+ except Exception: # pragma: no cover - honest degrade to the fail-safe floor
+ logger.warning("get_macro_dependents failed; macro changes treated as unmappable")
+ return {}
+ return dependents if isinstance(dependents, dict) else {}
+
+
+def resolve_git_scope(
+ head: LineageProvider,
+ git_base: str,
+ repo_dir: str | None = None,
+ git_head: str = "HEAD",
+) -> GitScopeResolution:
+ """Classify the FULL git diff (not just ``*.sql``) for ``--scope-git``.
+
+ Unlike :func:`git_changed_models` (which silently ignores files it cannot map — fine
+ for the coarse ``--git-base`` fallback where unmapped files simply add no changes),
+ scoping *subtracts* from a correct two-manifest changeset, so it must prove that every
+ logic-bearing changed file was accounted for. Each changed file is either:
+
+ * mapped to a model via its ``resource_path`` (``.sql`` and ``.py`` models, seeds), or
+ * mapped to its dependent models via the manifest macro graph (when available), or
+ * logic-bearing but unmappable -> recorded in ``unmappable`` (disables scoping), or
+ * non-logic (schema.yml/docs/README/...) -> ignored.
+ """
+ path_to_model = _path_to_model_map(head)
+ macro_dependents = _macro_dependents_or_empty(head)
+ models: set[str] = set()
+ unmappable: list[str] = []
+ for changed_file in _git_changed_files(git_base, repo_dir, git_head):
+ path = _norm_path(changed_file)
+ model = path_to_model.get(path)
+ if model:
+ models.add(model)
+ continue
+ if path in macro_dependents:
+ # Ceiling: the macro is known, so scope to every model it (transitively)
+ # feeds instead of disabling scoping outright.
+ models.update(macro_dependents[path])
+ continue
+ if _is_logic_bearing(path):
+ unmappable.append(path)
+ return GitScopeResolution(models=models, unmappable=unmappable)
+
+
def _norm_path(path: str) -> str:
return re.sub(r"^\./", "", path.strip()).lstrip("/")
+def _git_changed_files(git_base: str, repo_dir: str | None, git_head: str = "HEAD") -> list[str]:
+ """Every changed file between ``git_base`` and ``git_head`` (no pathspec filter).
+
+ :func:`resolve_git_scope` needs the complete diff — macros, Python models,
+ ``dbt_project.yml`` — not only the ``*.sql`` view that :func:`_git_changed_sql_files`
+ serves to the coarse git fallback and the backtest.
+ """
+ try:
+ result = subprocess.run(
+ ["git", "diff", "--name-only", f"{git_base}...{git_head}"],
+ cwd=repo_dir,
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ except (subprocess.CalledProcessError, FileNotFoundError) as exc:
+ raise RuntimeError(f"Failed to compute git diff against '{git_base}': {exc}") from exc
+
+ return [line.strip() for line in result.stdout.splitlines() if line.strip()]
+
+
def _git_changed_sql_files(
git_base: str, repo_dir: str | None, git_head: str = "HEAD"
) -> list[str]:
@@ -784,6 +1047,16 @@ def _git_changed_sql_files(
return [line.strip() for line in result.stdout.splitlines() if line.strip()]
+def _parrant_version() -> str | None:
+ """The installed parrant version, or ``None`` when metadata is unavailable."""
+ try:
+ from importlib.metadata import version
+
+ return version("parrant")
+ except Exception:
+ return None
+
+
def build_changeset_report(
source: str,
changes: list[ColumnChange],
@@ -794,18 +1067,22 @@ def build_changeset_report(
The impact keys (``summary``, ``affected_models``, ``affected_columns``,
``affected_exposures``) are a superset of the single-column ``impact`` block,
so existing consumers keep working; ``changeset`` and ``by_change`` are added.
+ ``report_version`` / ``parrant_version`` identify the report shape and producer so a
+ machine consumer can pin what it parses.
"""
by_kind: dict[str, int] = {}
for change in changes:
by_kind[change.kind.value] = by_kind.get(change.kind.value, 0) + 1
report: dict[str, object] = {
+ "report_version": 1,
+ "parrant_version": _parrant_version(),
"changeset": {
"source": source,
"total_changes": len(changes),
"by_kind": by_kind,
"changes": [change.to_dict() for change in changes],
- }
+ },
}
report.update(aggregated)
return report
diff --git a/parrant/lineage/display/backtest.py b/parrant/lineage/display/backtest.py
index 09c7ed1..19246ae 100644
--- a/parrant/lineage/display/backtest.py
+++ b/parrant/lineage/display/backtest.py
@@ -11,7 +11,7 @@
from __future__ import annotations
-from parrant.models.schema import BacktestReport, BacktestRuleStat
+from parrant.models.schema import BacktestReport, BacktestRuleStat, BacktestSelectionStats
def _rule_flag(stat: BacktestRuleStat) -> str:
@@ -34,6 +34,28 @@ def _totals_line(report: BacktestReport) -> str:
)
+def _selection_lines(stats: BacktestSelectionStats | None) -> list[str]:
+ """The selection widen-rate block (both renderers), or ``[]`` when no point carried one.
+
+ Measurement surface only: how often the rebuild selection widened to every reachable model,
+ the median skippable count, and the ranked resolution reasons forcing the widening.
+ """
+ if stats is None:
+ return []
+ lines = [
+ (
+ f"Selection: widened to all-reachable in {stats.widened_points}/"
+ f"{stats.points_with_selection} point(s) ({stats.widen_rate_pct}%); "
+ f"median skippable models {stats.median_skippable_models}; "
+ f"rebuilds forced by non-resolution: {stats.rebuild_forced_by_nonresolution_total}"
+ )
+ ]
+ if stats.top_reasons:
+ ranked = ", ".join(f"{entry.reason} ({entry.count})" for entry in stats.top_reasons)
+ lines.append(f"Top widening/forcing reasons: {ranked}")
+ return lines
+
+
def render_backtest_table(report: BacktestReport) -> str:
"""A fixed-width text table for terminals — the per-rule aggregate + totals + fidelity note."""
lines: list[str] = []
@@ -42,6 +64,7 @@ def render_backtest_table(report: BacktestReport) -> str:
lines.append(f"Range: {report.base}..{report.head}")
lines.append("")
lines.append(_totals_line(report))
+ lines.extend(_selection_lines(report.selection_stats))
lines.append("")
if not report.rule_stats:
@@ -80,6 +103,8 @@ def render_backtest_markdown(report: BacktestReport) -> str:
if report.base or report.head:
lines.append(f"- **Range:** `{report.base}..{report.head}`")
lines.append(f"- {_totals_line(report)}")
+ for selection_line in _selection_lines(report.selection_stats):
+ lines.append(f"- {selection_line}")
lines.append("")
lines.append(
diff --git a/parrant/lineage/display/markdown.py b/parrant/lineage/display/markdown.py
index 0db980c..548014f 100644
--- a/parrant/lineage/display/markdown.py
+++ b/parrant/lineage/display/markdown.py
@@ -53,6 +53,22 @@ def _structural_checks_skipped(report: dict[str, Any]) -> bool:
return not report.get("structural_checks_available", True)
+def _structural_diff_degraded_note(report: dict[str, Any]) -> str | None:
+ """The honesty line for a degraded structural diff (identical/stale catalogs), or None.
+
+ A degraded stamp means removed/type_changed columns may be INVISIBLE in this report even
+ though structural checks nominally ran — a quiet run must not read as proof.
+ """
+ block = report.get("structural_diff")
+ if not isinstance(block, dict) or block.get("status") != "degraded":
+ return None
+ reason = block.get("reason") or "base and head catalogs are not independent"
+ return (
+ f"⚠️ Structural diff degraded — {reason}. "
+ "Removed/retyped columns may be missing from this report."
+ )
+
+
def _plural(n: int, word: str) -> str:
return f"{n} {word}" + ("" if n == 1 else "s")
@@ -423,6 +439,44 @@ def _render_policy_section(verdict: dict[str, Any]) -> list[str]:
"",
]
+ out += _render_unproven_conditions(verdict)
+
+ return out
+
+
+def _render_unproven_conditions(verdict: dict[str, Any]) -> list[str]:
+ """Suppressed-unknown telemetry (additive section, issue #124): warn-rule conditions that a
+ ``fail_closed`` knob would have fired were the rule blocking. An all-warn pilot policy must
+ never read clean while block mode would fail closed here — so the count line is loud and the
+ per-rule list is folded (visibility without warning-flood)."""
+ unproven = verdict.get("unproven") or []
+ if not unproven:
+ return []
+ out: list[str] = [
+ "> ⚠️ **"
+ + _plural(len(unproven), "unproven warn-rule condition")
+ + " (unknown metadata under `fail_closed`)** — suppressed, not fired; block mode "
+ "would fail closed here.",
+ "",
+ f"Unproven conditions ({len(unproven)})
",
+ "",
+ ]
+ for record in unproven:
+ rule_id = record.get("rule_id", "?")
+ model = record.get("change_model")
+ column = record.get("change_column")
+ if model and column:
+ subject = f"`{model}.{column}`"
+ elif model:
+ subject = f"`{model}`"
+ else:
+ subject = "_changeset_"
+ cause = record.get("unknown_cause")
+ cause_txt = "evaluation error" if cause == "error" else "meta missing"
+ leaves = record.get("unknown_leaves") or []
+ leaves_txt = " — unknown: " + ", ".join(f"`{leaf}`" for leaf in leaves) if leaves else ""
+ out.append(f"- **{rule_id}** on {subject} ({cause_txt}){leaves_txt}")
+ out += ["", "Unproven conditions (2)
" in md
+ assert "warn-pii" in md
+ assert "`stg_accounts.ssn`" in md
+ assert "meta.pii" in md
+ assert "meta missing" in md
+ assert "evaluation error" in md
+
+
+def test_unproven_singular_line():
+ verdict = {
+ "decision": "allow",
+ "hits": [],
+ "unproven_count": 1,
+ "unproven": [
+ {
+ "rule_id": "warn-pii",
+ "change_model": "m",
+ "change_column": "c",
+ "unknown_cause": "missing",
+ "unknown_leaves": ["meta.pii"],
+ }
+ ],
+ }
+ md = render_changeset_markdown(_report_with_policy(verdict))
+ assert "1 unproven warn-rule condition (unknown metadata under `fail_closed`)" in md
+
+
+def test_no_unproven_renders_no_unproven_section():
+ verdict = {"decision": "allow", "hits": [], "unproven_count": 0, "unproven": []}
+ md = render_changeset_markdown(_report_with_policy(verdict))
+ assert "unproven warn-rule condition" not in md
+ assert "Unproven conditions" not in md
diff --git a/tests/unit/lineage/test_selection.py b/tests/unit/lineage/test_selection.py
index 319bf6e..a999976 100644
--- a/tests/unit/lineage/test_selection.py
+++ b/tests/unit/lineage/test_selection.py
@@ -209,6 +209,24 @@ def test_unresolved_change_still_rebuilds_its_own_model() -> None:
_assert_partition(selection, changed | reachable)
+def test_unresolved_change_widens_the_selection() -> None:
+ # A change parrant could not analyze (impact resolution raised) has an UNKNOWN reach: its
+ # downstream cone is invisible to the fold, so nothing in the reachable universe may be
+ # proven safe to skip. The selection must widen, exactly like non-full confidence.
+ changed = {"orphan"}
+ reachable = {"downstream_a", "downstream_b"}
+ by_change = [
+ _change(kind="removed", semantic=None, resolved=False),
+ _change(kind="added", semantic=None, reached=["downstream_a"]),
+ ]
+ selection = build_selection(reachable, changed, by_change, _confidence())
+
+ assert selection["widened_to_all_reachable"] is True
+ assert selection["skippable_models"] == []
+ assert set(selection["rebuild_models"]) == changed | reachable
+ _assert_partition(selection, changed | reachable)
+
+
def test_selection_is_wired_into_get_changeset_impact() -> None:
# End-to-end through the service: a real logic change on mart.id surfaces a selection block
# with the edited model in the rebuild set.