fix(dbt): apply sqb build plan safety policies on dbt run and build

sqb dbt run and build enforced only the missing-origin and retention
decrease policies before executing SQLBuild work, so a
table_type_downgrade, snapshot full-refresh or max_models policy could
be bypassed through the dbt interop path. Both commands now share the
sqb build gate sequence. sqb dbt has no allow flags: a policy that needs
confirmation prompts on a terminal and otherwise refuses with guidance
naming the matching sqb build flag or policy setting.

Author: kvlonge

src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/dbt.md

@@ -34,6 +34,17 @@ Build dbt models first (including dbt tests), then SQLBuild models with audits.
 sqb dbt build [--select <selector>...] [--exclude <selector>...]
 ```
 
+### Plan safety policies
+
+After dbt finishes, `sqb dbt run` and `sqb dbt build` apply the same plan safety policies as
+[`sqb build`](build.md) before any SQLBuild model executes: blocked model and column
+migrations, compatibility-view conflicts, the target's `execution_limits.max_models`, the
+snapshot full-refresh policy, `table_type_downgrade`, `time_travel_retention_decrease`, and
+`missing_migration_origin`. `sqb dbt` has no `--allow-*` flags. A policy that requires
+confirmation asks for the typed confirmation on an interactive terminal and refuses otherwise;
+build the affected models with `sqb build` and its allow flag instead, or relax the policy. A
+refusal at this point leaves completed dbt work in place and builds no SQLBuild models.
+
 ## sqb dbt debug
 
 Run dbt diagnostics followed by SQLBuild diagnostics.

src/sqlbuild/cli/commands/_helpers/build_planning/defer_clone.py

@@ -13,6 +13,7 @@
 from sqlbuild.cli.commands._helpers.planning.external_refs import (
     resolve_external_sql_reference_resolver,
 )
+from sqlbuild.cli.commands.constants import EXECUTION_LIMIT_BUILD_NOTE
 from sqlbuild.cli.commands.exceptions import CliUserError
 from sqlbuild.cli.commands.models import (
     BuildCommandRequest,
@@ -196,6 +197,7 @@ def run_defer_clone_boundary_prephase(
         model_count=selected_model_count,
         target_name=invocation.effective_target_name,
         limits=invocation.execution_limits,
+        refusal_note=EXECUTION_LIMIT_BUILD_NOTE,
     )
     run_defer_clone_prephase(
         inputs=DeferClonePrephaseInputs(

src/sqlbuild/cli/commands/_helpers/build_planning/execution_limits.py

@@ -19,6 +19,7 @@ def enforce_model_execution_limit(
     model_count: int,
     target_name: str | None,
     limits: ExecutionLimitsConfig,
+    refusal_note: str,
 ) -> None:
     """Reject a build whose expanded model count exceeds its target policy."""
 
@@ -33,7 +34,7 @@ def enforce_model_execution_limit(
         f"Maximum models:  {maximum}\n"
         "Configuration:   "
         f"targets.{target_label}.execution_limits.max_models\n\n"
-        "No warehouse changes were made.",
+        f"{refusal_note}",
         code="C413",
         help=limits.remediation,
     )

src/sqlbuild/cli/commands/_helpers/build_planning/full_refresh.py

@@ -25,6 +25,7 @@ def enforce_snapshot_full_refresh_policy(
     plan: PlanOutput,
     snapshots_config: SnapshotsConfig,
     allow_snapshot_full_refresh: bool,
+    non_interactive_help: str,
     input_stream: TextIO,
     output_stream: TextIO,
 ) -> None:
@@ -69,7 +70,7 @@ def enforce_snapshot_full_refresh_policy(
         raise CliUserError(
             "snapshot full refresh requires confirmation",
             code="C239",
-            help="Pass --allow-snapshot-full-refresh to confirm in non-interactive runs.",
+            help=non_interactive_help,
         )
 
     if not _confirm(

src/sqlbuild/cli/commands/_helpers/build_planning/plan_policies.py

@@ -19,9 +19,20 @@
 from sqlbuild.cli.commands._helpers.build_planning.table_type import (
     enforce_table_type_downgrade_policy,
 )
-from sqlbuild.cli.commands.constants import MISSING_ORIGIN_BUILD_HELP
+from sqlbuild.cli.commands.constants import (
+    EXECUTION_LIMIT_BUILD_NOTE,
+    EXECUTION_LIMIT_DBT_NOTE,
+    MISSING_ORIGIN_BUILD_HELP,
+    MISSING_ORIGIN_DBT_HELP,
+    RETENTION_DECREASE_BUILD_HELP,
+    RETENTION_DECREASE_DBT_HELP,
+    SNAPSHOT_FULL_REFRESH_BUILD_HELP,
+    SNAPSHOT_FULL_REFRESH_DBT_HELP,
+    TABLE_TYPE_DOWNGRADE_BUILD_HELP,
+    TABLE_TYPE_DOWNGRADE_DBT_HELP,
+)
 from sqlbuild.cli.commands.exceptions import CliUserError
-from sqlbuild.cli.commands.models import BuildCommandRequest, BuildInvocation
+from sqlbuild.cli.commands.models import BuildCommandRequest, BuildInvocation, PlanSafetyGates
 from sqlbuild.compiler.migrations.types import ColumnMigrationDecision, MigrationDecision
 from sqlbuild.compiler.planner.constants import HIDDEN_ORIGIN_REMEDY
 from sqlbuild.compiler.planner.models import (
@@ -31,6 +42,7 @@
     PlanWarning,
 )
 from sqlbuild.compiler.planner.types import WarningSeverity
+from sqlbuild.spec.contracts.models import ExecutionLimitsConfig, SnapshotsConfig
 from sqlbuild.spec.contracts.types import MissingMigrationOriginPolicy
 
 _OLD_NAME_CODES: frozenset[str] = frozenset({"M114", "P008"})
@@ -44,40 +56,92 @@
 
 def enforce_build_plan_policies(
     *, request: BuildCommandRequest, invocation: BuildInvocation, plan: PlanOutput
+) -> None:
+    """Apply every plan safety gate before a direct build executes anything."""
+
+    enforce_plan_safety_policies(
+        plan=plan,
+        gates=PlanSafetyGates(
+            allow_missing_migration_origin=request.allow_missing_migration_origin,
+            allow_snapshot_full_refresh=request.allow_snapshot_full_refresh,
+            allow_table_type_downgrade=request.allow_table_type_downgrade,
+            allow_retention_decrease=request.allow_retention_decrease,
+            missing_origin_help=MISSING_ORIGIN_BUILD_HELP,
+            snapshot_full_refresh_help=SNAPSHOT_FULL_REFRESH_BUILD_HELP,
+            table_type_downgrade_help=TABLE_TYPE_DOWNGRADE_BUILD_HELP,
+            retention_decrease_help=RETENTION_DECREASE_BUILD_HELP,
+            execution_limit_note=EXECUTION_LIMIT_BUILD_NOTE,
+        ),
+        snapshots_config=invocation.discovered_inputs.project_config.snapshots,
+        target_name=invocation.effective_target_name,
+        execution_limits=invocation.execution_limits,
+        streams=(sys.stdin, sys.stdout),
+    )
+
+
+def dbt_interop_plan_safety_gates() -> PlanSafetyGates:
+    """Gates for sqb dbt, which has no allow flags and confirms only on a terminal."""
+
+    return PlanSafetyGates(
+        allow_missing_migration_origin=False,
+        allow_snapshot_full_refresh=False,
+        allow_table_type_downgrade=False,
+        allow_retention_decrease=False,
+        missing_origin_help=MISSING_ORIGIN_DBT_HELP,
+        snapshot_full_refresh_help=SNAPSHOT_FULL_REFRESH_DBT_HELP,
+        table_type_downgrade_help=TABLE_TYPE_DOWNGRADE_DBT_HELP,
+        retention_decrease_help=RETENTION_DECREASE_DBT_HELP,
+        execution_limit_note=EXECUTION_LIMIT_DBT_NOTE,
+    )
+
+
+def enforce_plan_safety_policies(
+    *,
+    plan: PlanOutput,
+    gates: PlanSafetyGates,
+    snapshots_config: SnapshotsConfig,
+    target_name: str | None,
+    execution_limits: ExecutionLimitsConfig,
+    streams: tuple[TextIO, TextIO],
 ) -> None:
     """Apply migration, execution-limit, and storage safety gates in their required order."""
 
+    input_stream, output_stream = streams
     enforce_migration_plan_policies(
         plan=plan,
-        allow_missing_migration_origin=request.allow_missing_migration_origin,
-        non_interactive_help=MISSING_ORIGIN_BUILD_HELP,
-        input_stream=sys.stdin,
-        output_stream=sys.stdout,
+        allow_missing_migration_origin=gates.allow_missing_migration_origin,
+        non_interactive_help=gates.missing_origin_help,
+        input_stream=input_stream,
+        output_stream=output_stream,
     )
     _enforce_old_name_policy(plan=plan)
     enforce_model_execution_limit(
         model_count=executable_model_count(plan=plan),
-        target_name=invocation.effective_target_name,
-        limits=invocation.execution_limits,
+        target_name=target_name,
+        limits=execution_limits,
+        refusal_note=gates.execution_limit_note,
     )
     enforce_snapshot_full_refresh_policy(
         plan=plan,
-        snapshots_config=invocation.discovered_inputs.project_config.snapshots,
-        allow_snapshot_full_refresh=request.allow_snapshot_full_refresh,
-        input_stream=sys.stdin,
-        output_stream=sys.stdout,
+        snapshots_config=snapshots_config,
+        allow_snapshot_full_refresh=gates.allow_snapshot_full_refresh,
+        non_interactive_help=gates.snapshot_full_refresh_help,
+        input_stream=input_stream,
+        output_stream=output_stream,
     )
     enforce_table_type_downgrade_policy(
         plan=plan,
-        allow_table_type_downgrade=request.allow_table_type_downgrade,
-        input_stream=sys.stdin,
-        output_stream=sys.stdout,
+        allow_table_type_downgrade=gates.allow_table_type_downgrade,
+        non_interactive_help=gates.table_type_downgrade_help,
+        input_stream=input_stream,
+        output_stream=output_stream,
     )
     enforce_retention_decrease_policy(
         plan=plan,
-        allow_retention_decrease=request.allow_retention_decrease,
-        input_stream=sys.stdin,
-        output_stream=sys.stdout,
+        allow_retention_decrease=gates.allow_retention_decrease,
+        non_interactive_help=gates.retention_decrease_help,
+        input_stream=input_stream,
+        output_stream=output_stream,
     )
 
 

src/sqlbuild/cli/commands/_helpers/build_planning/retention_decrease.py

@@ -13,6 +13,7 @@ def enforce_retention_decrease_policy(
     *,
     plan: PlanOutput,
     allow_retention_decrease: bool,
+    non_interactive_help: str,
     input_stream: TextIO,
     output_stream: TextIO,
 ) -> None:
@@ -42,7 +43,7 @@ def enforce_retention_decrease_policy(
     if not input_stream.isatty():
         raise CliUserError(
             "time travel retention decrease requires confirmation",
-            help="Pass --allow-retention-decrease to confirm in non-interactive runs.",
+            help=non_interactive_help,
         )
     expected: str = _confirmation_text(confirmation)
     output_stream.write(

src/sqlbuild/cli/commands/_helpers/build_planning/table_type.py

@@ -13,6 +13,7 @@ def enforce_table_type_downgrade_policy(
     *,
     plan: PlanOutput,
     allow_table_type_downgrade: bool,
+    non_interactive_help: str,
     input_stream: TextIO,
     output_stream: TextIO,
 ) -> None:
@@ -38,7 +39,7 @@ def enforce_table_type_downgrade_policy(
     if not input_stream.isatty():
         raise CliUserError(
             "table-type downgrade requires confirmation",
-            help="Pass --allow-table-type-downgrade to confirm in non-interactive runs.",
+            help=non_interactive_help,
         )
     expected: str = _confirmation_text(confirmation)
     output_stream.write(

src/sqlbuild/cli/commands/_helpers/dbt/sqlbuild_work.py

@@ -8,16 +8,13 @@
 
 from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
 from sqlbuild.cli.commands._helpers.build_planning.plan_policies import (
-    enforce_migration_plan_policies,
-)
-from sqlbuild.cli.commands._helpers.build_planning.retention_decrease import (
-    enforce_retention_decrease_policy,
+    dbt_interop_plan_safety_gates,
+    enforce_plan_safety_policies,
 )
 from sqlbuild.cli.commands.classes.build_progress_callbacks import (
     BuildProgressCallbacks,
     format_build_footer,
 )
-from sqlbuild.cli.commands.constants import MISSING_ORIGIN_DBT_HELP
 from sqlbuild.cli.commands.models import DbtSqlbuildWorkContext
 from sqlbuild.cli.progress.classes.connection_progress_reporter import ConnectionProgressReporter
 from sqlbuild.cli.progress.main._execution_header import format_execution_header
@@ -54,18 +51,13 @@ def execute_sqlbuild_build_work(
     adapter_name: str = context.adapter_name
     output_stream: TextIO = context.output_stream
     use_color: bool = context.use_color
-    enforce_migration_plan_policies(
-        plan=plan_output,
-        allow_missing_migration_origin=False,
-        non_interactive_help=MISSING_ORIGIN_DBT_HELP,
-        input_stream=sys.stdin,
-        output_stream=output_stream,
-    )
-    enforce_retention_decrease_policy(
+    enforce_plan_safety_policies(
         plan=plan_output,
-        allow_retention_decrease=False,
-        input_stream=sys.stdin,
-        output_stream=output_stream,
+        gates=dbt_interop_plan_safety_gates(),
+        snapshots_config=context.snapshots_config,
+        target_name=project.effective_target_name,
+        execution_limits=context.execution_limits,
+        streams=(sys.stdin, output_stream),
     )
     callbacks: BuildProgressCallbacks = BuildProgressCallbacks(
         plan=plan_output, use_color=use_color, verbose=verbose, debug=False

src/sqlbuild/cli/commands/constants.py

@@ -111,4 +111,30 @@
     "models with sqb build --allow-missing-migration-origin and leave them out of this dbt "
     "selection, or set the target's missing_migration_origin to 'allow'."
 )
+SNAPSHOT_FULL_REFRESH_BUILD_HELP: str = (
+    "Pass --allow-snapshot-full-refresh to confirm in non-interactive runs."
+)
+SNAPSHOT_FULL_REFRESH_DBT_HELP: str = (
+    "sqb dbt has no confirmation flag: confirm on an interactive terminal, build the affected "
+    "snapshots with sqb build --allow-snapshot-full-refresh and leave them out of this dbt "
+    "selection, or relax the snapshot_full_refresh policy."
+)
+TABLE_TYPE_DOWNGRADE_BUILD_HELP: str = (
+    "Pass --allow-table-type-downgrade to confirm in non-interactive runs."
+)
+TABLE_TYPE_DOWNGRADE_DBT_HELP: str = (
+    "sqb dbt has no confirmation flag: confirm on an interactive terminal, build the affected "
+    "models with sqb build --allow-table-type-downgrade and leave them out of this dbt "
+    "selection, or set the target's table_type_downgrade to 'allow'."
+)
+RETENTION_DECREASE_BUILD_HELP: str = (
+    "Pass --allow-retention-decrease to confirm in non-interactive runs."
+)
+RETENTION_DECREASE_DBT_HELP: str = (
+    "sqb dbt has no confirmation flag: confirm on an interactive terminal, build the affected "
+    "models with sqb build --allow-retention-decrease and leave them out of this dbt "
+    "selection, or set the target's time_travel_retention_decrease to 'allow'."
+)
+EXECUTION_LIMIT_BUILD_NOTE: str = "No warehouse changes were made."
+EXECUTION_LIMIT_DBT_NOTE: str = "dbt work has already run; no SQLBuild models were built."
 QUERY_DIFF_INCOMPLETE_EXECUTION_CODES: frozenset[str] = frozenset({"C237", "C238", "C239"})

src/sqlbuild/cli/commands/models.py

@@ -83,7 +83,12 @@
 from sqlbuild.presentation.classes.transient_status_reporter import TransientStatusReporter
 from sqlbuild.python_nodes.models import SqlResourceRef
 from sqlbuild.runtime.contracts.types import NodeStartCallback
-from sqlbuild.spec.contracts.models import CostConfig, ExecutionLimitsConfig, SourceEntry
+from sqlbuild.spec.contracts.models import (
+    CostConfig,
+    ExecutionLimitsConfig,
+    SnapshotsConfig,
+    SourceEntry,
+)
 
 
 @dataclass(frozen=True)
@@ -390,6 +395,23 @@ class DbtSqlbuildWorkContext:
     adapter_name: str
     output_stream: TextIO
     use_color: bool
+    snapshots_config: SnapshotsConfig
+    execution_limits: ExecutionLimitsConfig
+
+
+@dataclass(frozen=True)
+class PlanSafetyGates:
+    """Pre-confirmed plan safety gates and the guidance shown when one needs a terminal."""
+
+    allow_missing_migration_origin: bool
+    allow_snapshot_full_refresh: bool
+    allow_table_type_downgrade: bool
+    allow_retention_decrease: bool
+    missing_origin_help: str
+    snapshot_full_refresh_help: str
+    table_type_downgrade_help: str
+    retention_decrease_help: str
+    execution_limit_note: str
 
 
 @dataclass(frozen=True)