feat: depth-limited selection, run reports and config reuse

.gitignore

@@ -206,9 +206,11 @@ marimo/_static/
 marimo/_lsp/
 __marimo__/
 
-dbt
-!dbt_ci/**/dbt
-!dbt_ci/**/dbt
+# A dbt project kept at the repository root for local testing. Anchored with a
+# leading slash: an unanchored "dbt" matches every directory of that name at any
+# depth, which silently excluded dbt_ci/dbt/ and tests/unit/dbt/ and needed
+# negation rules to undo.
+/dbt/
 dependency_graph.json
 logs
 dbt.duckdb

README.md

@@ -173,8 +173,34 @@ dbt-ci run --dbt-project-dir dbt --mode models
 | Flag | Aliases | Env Var(s) | Default | Description |
 |------|---------|-----------|---------|-------------|
 | `--mode` | `-m`, `--nodes`, `-n` | `DBT_NODES` | `all` | What to run: `all`, `models`, `seeds`, `snapshots`, `tests` |
+| `--downstream-depth` | | `DBT_DOWNSTREAM_DEPTH` | Full graph | How many levels of downstream dependencies to include (dbt's `model+N`) |
 | `--filters` | `-f` | | `None` | Extra resource-type filter (repeatable, choices: `models`, `seeds`, `snapshots`, `tests`). E.g. `--mode tests -f snapshots` to run only tests that have a snapshot dependency |
 
+#### Limiting blast radius
+
+By default a change selects its **entire** downstream graph. On a large project a change
+to a core staging model therefore rebuilds almost everything, which is the opposite of
+what change-based CI is for. `--downstream-depth` caps how far a change propagates:
+
+```bash
+dbt-ci run --downstream-depth 0   # only what changed
+dbt-ci run --downstream-depth 1   # changed models and their direct children
+dbt-ci run --downstream-depth 2   # two levels out
+dbt-ci run                        # unlimited (default)
+```
+
+For a chain `customers → l1 → l2 → l3` where only `customers` changed:
+
+| Flag | Models run |
+|------|------------|
+| `--downstream-depth 0` | `customers` |
+| `--downstream-depth 1` | `customers`, `l1` |
+| `--downstream-depth 2` | `customers`, `l1`, `l2` |
+| *(omitted)* | `customers`, `l1`, `l2`, `l3` |
+
+New and deleted nodes are always included regardless of depth — a new model has to run
+whether or not anything depends on it yet.
+
 > All [common options](#common-options) also apply.
 
 **Examples:**
@@ -262,6 +288,38 @@ dbt-ci migration            # apply the changes
 
 ---
 
+### `report` - Summarise the Run
+
+Renders the change set detected by `init` and the status of every command that has run
+so far. In GitHub Actions the report is appended to the job summary automatically, so no
+workflow wiring is needed beyond calling it.
+
+```bash
+dbt-ci report                       # → $GITHUB_STEP_SUMMARY, or stdout locally
+dbt-ci report --output report.md    # → a file, e.g. to post as a PR comment
+dbt-ci report --format json         # → machine-readable
+```
+
+The report covers:
+
+- **Change counts and node names**, grouped as modified / new / deleted. Long lists fold
+  into a collapsible block.
+- **Exposure impact** — the exposures downstream of the change set, including those
+  downstream of *deleted* nodes, which is usually the case worth catching.
+- **Command status and duration** for each of `init`, `run`, `delete`, `ephemeral` and
+  `migration` that has run.
+
+**Flags:**
+
+| Flag | Aliases | Env Var(s) | Default | Description |
+|------|---------|-----------|---------|-------------|
+| `--output` | `-o` | `DBT_REPORT_OUTPUT` | `$GITHUB_STEP_SUMMARY` or stdout | Where to write the report |
+| `--format` | `-F` | `DBT_REPORT_FORMAT` | `markdown` | `markdown` or `json` |
+
+> All [common options](#common-options) also apply.
+
+---
+
 ### `config` - Generate a Config File
 
 Writes a commented `dbt-ci.config.yaml` skeleton with the common options pre-filled, so
@@ -460,6 +518,10 @@ init:
   comparison-strategy: hybrid
   base-ref: main
 
+run:
+  nodes: models
+  downstream-depth: 2
+
 finalize:
   artifacts-uri: s3://my-bucket/dbt-artifacts/
   files:
@@ -642,6 +704,19 @@ dbt-ci run
 
 **Note:** State management is cache-based. Run `init` once, then subsequent commands automatically use the cached state.
 
+### Settings Inherited From `init`
+
+`init` records the `--target` and `--vars` it ran with, and later commands reuse them, so
+they only need to be given once:
+
+```bash
+dbt-ci init --target ci --vars '{"use_production_data": false}' --state dbt/.dbtstate
+dbt-ci run          # runs against target 'ci' with the same vars
+dbt-ci delete       # likewise
+```
+
+Passing the flag explicitly still wins — the cache only fills in what was left unset.
+
 ### Cache Location
 
 `init` writes its cache (state comparison, manifests, run report and log file) to
@@ -729,6 +804,8 @@ dbt-ci:
 - **💬 Notifications**: Slack webhook integration for CI/CD alerts
 - **♻️ Ephemeral Environments**: Test changes in isolated environments
 - **🧹 Cleanup**: Automatically remove deleted models from target warehouse
+- **🎯 Blast-Radius Control**: Cap how far a change propagates with `--downstream-depth`
+- **📝 Run Reports**: Markdown summary of the change set, exposure impact and command status
 - **🔀 Partition Migrations**: Rebuild tables whose partitioning configuration changed (BigQuery)
 
 ## Use Cases

dbt_ci/cli/config/schema.py

@@ -53,6 +53,7 @@
         "type": "section",
         "fields": {
             "nodes": {"type": "enum", "choices": ["all", "models", "seeds", "snapshots", "tests"]},
+            "downstream-depth": {"aliases": ["downstream_depth"], "type": "int"},
         },
     },
     "finalize": {
@@ -95,6 +96,10 @@ def _validate_value(path: str, value: Any, rule: dict, errors: list[str]) -> Non
     elif kind == "bool":
         if not isinstance(value, bool):
             errors.append(f"'{path}' must be a boolean (true/false), got {value!r}")
+    elif kind == "int":
+        # bool is a subclass of int, so reject it explicitly.
+        if isinstance(value, bool) or not isinstance(value, int):
+            errors.append(f"'{path}' must be an integer, got {value!r}")
     elif kind == "enum":
         choices = rule["choices"]
         if isinstance(value, str):

dbt_ci/commands/delete/index.py

@@ -8,6 +8,7 @@
 from argparse import Namespace
 import click
 from dbt_ci.utilities.cache import CacheManager
+from dbt_ci.dbt.flags import apply_cached_config
 from dbt_ci.utilities.logging import redact_namespace
 from dbt_ci.connectors import get_connector
 from dbt_ci.graph.dependency_graph import DbtGraph
@@ -24,6 +25,7 @@ def delete(args: Namespace):
         logger.debug(f"Running with the following arguments: {redact_namespace(args)}")
         cache = CacheManager(args)
         cache.start_report("delete", args)
+        apply_cached_config(args)
         connector_type = cast(SupportedConnectors, get_profile(args)["type"])
         delete_connector = get_connector(connector_type)
         delete_map = generate_delete_map(args, cache)

dbt_ci/commands/ephemeral/index.py

@@ -13,6 +13,7 @@
 from typing import Optional, cast
 import click
 from dbt_ci.utilities.cache import CacheManager
+from dbt_ci.dbt.flags import apply_cached_config
 from dbt_ci.connectors import DB_CONNECTORS
 from dbt_ci.graph.dependency_graph import DbtGraph
 from dbt_ci.utilities.logging import print_exception, redact_namespace
@@ -56,6 +57,7 @@ def ephemeral(args: Namespace):
             )
         cache = CacheManager(args)
         cache.start_report("ephemeral", args)
+        apply_cached_config(args)
         cache_dict: StateChangeSummary = cast(StateChangeSummary, cache.get_cache())
 
         if cache_dict is None:

dbt_ci/commands/init/index.py

@@ -64,6 +64,16 @@ def init(args: Namespace):
 
             # Reload reference manifest file after downloading from storage
             cache.write_reference_manifest(get_reference_manifest_file(str(local_state_dir)))
+        elif reference_state_path:
+            # A local --state directory is just as much a baseline as a downloaded one.
+            # Caching it is what lets later commands resolve nodes that no longer exist
+            # in the target - without it, `delete` cannot look up what it should drop.
+            # Best-effort: if the state isn't there, the comparison below raises a
+            # clearer error than failing here would.
+            try:
+                cache.write_reference_manifest(get_reference_manifest_file(reference_state_path))
+            except FileNotFoundError:
+                logger.debug(f"No reference manifest at '{reference_state_path}' to cache yet.")
 
         # Compile dbt and generate reference manifest.json file
         #run_multiprocessed
@@ -77,12 +87,17 @@ def init(args: Namespace):
             # Different targets - will compile again later with actual target
             cache.write_target_manifest(target_manifest_file)
         else:
-            # Same target or no reference target specified - reference and target are the same
+            # Same target, or none specified. The reference *target* (which warehouse to
+            # compile against) is independent of the reference *state* (the baseline to
+            # compare with), so an already-cached baseline must not be replaced here -
+            # doing so discarded the only copy of the deleted nodes' metadata.
             logger.debug("Reference target is the same as current target!")
-            logger.debug("Using the same manifest for both reference and target state.")
-            cache.write_reference_manifest(target_manifest_file)
             cache.write_target_manifest(target_manifest_file)
 
+            if cache.get_cache("reference_manifest.json") is None:
+                logger.debug("No reference state available. Using the target manifest as the reference.")
+                cache.write_reference_manifest(target_manifest_file)
+
         # Will generate summary and output it in the logs. It also covers:
         # 1. Migration plan for partitioning changes
         # 2. Ephemeral plan

dbt_ci/commands/migration/index.py

@@ -5,6 +5,7 @@
 from typing import cast
 import click
 from dbt_ci.utilities.cache import CacheManager
+from dbt_ci.dbt.flags import apply_cached_config
 from dbt_ci.graph.dependency_graph import DbtGraph
 from dbt_ci.utilities.logging import print_exception, redact_namespace
 from dbt_ci.schema import MigrationMap, SupportedConnectors
@@ -33,6 +34,7 @@ def migration(args: Namespace):
         logger.debug(f"Running with the following arguments: {redact_namespace(args)}")
         cache = CacheManager(args)
         cache.start_report("migrate", args)
+        apply_cached_config(args)
         connector_type = cast(SupportedConnectors, get_profile(args)["type"])
         connector = get_connector(connector_type)
 

dbt_ci/commands/report/cli.py

@@ -0,0 +1,48 @@
+import click
+from dbt_ci.main import cli
+from dbt_ci.cli.common_options import common_options
+from dbt_ci.cli.config import make_config_callback
+from dbt_ci.cli.namespace import to_namespace
+from dbt_ci.utilities.logging import setup_logging
+from dbt_ci.commands.report.index import report
+
+
+@cli.command(name="report")
+@common_options
+@click.option(
+    "--output", "-o",
+    envvar=["DBT_REPORT_OUTPUT"],
+    default=None,
+    type=str,
+    callback=make_config_callback("DBT_REPORT_OUTPUT"),
+    help=(
+        "Where to write the report. Defaults to $GITHUB_STEP_SUMMARY when set, "
+        "otherwise stdout."
+    ),
+)
+@click.option(
+    "--format", "-F", "format",
+    envvar=["DBT_REPORT_FORMAT"],
+    type=click.Choice(["markdown", "json"], case_sensitive=False),
+    default="markdown",
+    callback=make_config_callback("DBT_REPORT_FORMAT", then=str.lower),
+    help="Output format (default: markdown).",
+)
+def report_cmd(**kwargs):
+    """Summarise the current dbt-ci run
+
+    Renders the change set from 'init' and the status of each command that has run so
+    far. In GitHub Actions the report is appended to the job summary automatically.
+
+    Examples:
+        # Append to the GitHub Actions job summary
+        dbt-ci report
+
+        # Write markdown to a file, e.g. to post as a pull request comment
+        dbt-ci report --output report.md
+
+        # Machine-readable output
+        dbt-ci report --format json
+    """
+    setup_logging(to_namespace(kwargs).log_level)
+    return report(to_namespace(kwargs, command="report"))