feat: add target execution limits

Author: kvlonge

Makefile

@@ -64,6 +64,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_dynamic_column_contract.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_dag_json_behavior.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_enum_contract.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_execution_limits.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_expression_sources.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_lifecycle_commands.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_manifest_artifact_gating.py \

README.md

@@ -24,7 +24,7 @@ All state is persisted as append-only tables in the warehouse alongside your dat
 - **Cursor-based incremental processing.** Automatic gap detection and resume, with microbatch mode for large ranges. No external checkpoint to maintain.
 - **Ingestion and Python nodes.** Load external data with Python `@loader` functions, and run `@task`, `@asset`, and `@check` nodes as first-class members of the same DAG as your SQL models.
 
-See the [documentation](https://docs.sqlbuild.com) for the full feature set, including providers, lifecycle hooks, Python macros, UDFs, custom materializations, data diffs, zero-copy cloning, and virtual environments. This repository also includes a focused guide to [model and raw-query diffs](docs/query-diffs.md). To coordinate dbt and SQLBuild projects, see the [dbt compatibility guide](https://docs.sqlbuild.com/concepts/dbt-compatibility/overview).
+See the [documentation](https://docs.sqlbuild.com) for the full feature set, including providers, lifecycle hooks, Python macros, UDFs, custom materializations, data diffs, zero-copy cloning, and virtual environments. This repository also includes focused guides to [target execution limits](docs/execution-limits.md) and [model and raw-query diffs](docs/query-diffs.md). To coordinate dbt and SQLBuild projects, see the [dbt compatibility guide](https://docs.sqlbuild.com/concepts/dbt-compatibility/overview).
 
 ## Quick start
 

docs/execution-limits.md

@@ -0,0 +1,68 @@
+# Target execution limits
+
+Named targets can set optional limits for `sqb build`. Limits are useful for development targets
+where an unexpectedly broad selector or long-running command should stop instead of consuming an
+unbounded amount of warehouse compute.
+
+```toml
+[targets.dev]
+connection = "warehouse"
+database = "analytics"
+schema = "dev_alice"
+
+[targets.dev.execution_limits]
+max_models = 125
+max_duration = "45m"
+remediation = """
+Review the expanded plan and confirm that the selected scope is intentional.
+Automated tools must request approval before changing this policy.
+"""
+```
+
+All fields are optional. A target without an `execution_limits` section retains unlimited existing
+behavior, so production or orchestrated targets do not need placeholder values.
+
+## Model limit
+
+`max_models` is a positive integer. SQLBuild expands selectors and counts model entries that would
+execute. When the count exceeds the target limit, the build exits before clone, load, or model
+execution begins. The error reports the target, selected count, configured maximum, configuration
+key, and confirms that no warehouse changes were made.
+
+`sqb plan` remains unrestricted and can be used to inspect a broad selection safely.
+
+## Duration limit
+
+`max_duration` is a positive fixed duration such as `30m`, `2h`, or `90s`. It covers planning and
+warehouse-mutating build phases after target resolution. SQLBuild stops dispatching work at the
+deadline and passes the remaining duration to each warehouse statement so in-flight work is
+cancelled by the adapter.
+
+Safe in-flight cancellation is currently supported by the Snowflake adapter, whose maximum accepted
+duration is seven days. SQLBuild rejects an unsupported duration or an adapter that cannot guarantee
+active statement cancellation rather than presenting a limit that may leave warehouse work running.
+
+A duration failure may occur after earlier models have completed. Its error therefore distinguishes
+the timeout from a pre-execution model-limit failure and states that warehouse changes may already
+have occurred.
+
+## Custom remediation
+
+`remediation` is optional project-authored text. SQLBuild appends it to model-count, duration, and
+unsupported-adapter errors after the factual error details. This lets each project define its own
+approval or recovery process without SQLBuild recommending that users weaken local policy.
+
+## Local overrides
+
+The same section is accepted in `sqlbuild_local.toml` and follows normal field-by-field local target
+precedence:
+
+```toml
+[targets.dev.execution_limits]
+max_models = 200
+max_duration = "90m"
+```
+
+This permits an intentional developer override while keeping the shared target policy visible in
+`sqlbuild_project.toml`. Other target fields and any execution-limit fields omitted locally continue
+to inherit their project values.

src/sqlbuild/adapter/contract/classes/base_adapter.py

@@ -85,6 +85,7 @@ class BaseAdapter(RetentionAdapterMixin, StrictAdapter):
     max_identifier_length: ClassVar[int] = 63
     state_tables_transient: ClassVar[bool] = False
     allows_implicit_managed_write_schema: ClassVar[bool] = False
+    execution_duration_limit_seconds: ClassVar[int | None] = None
 
     def supports_zero_copy_clone(self) -> bool:
         return False

src/sqlbuild/adapter/contract/classes/connection.py

@@ -9,6 +9,10 @@
 
 from sqlbuild.adapter.contract.main._attribute_failed_sql import attribute_failed_sql
 from sqlbuild.adapter.contract.models import QueryResult
+from sqlbuild.runtime.execution_limits.exceptions import ExecutionDurationLimitError
+from sqlbuild.runtime.execution_limits.main.enforce_execution_deadline import (
+    enforce_execution_deadline,
+)
 from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle
 
 
@@ -27,8 +31,13 @@ def execute(self, *, connection: Any, sql: str) -> Any:
 
         with StatementLifecycle(adapter=self.adapter_name, sql=sql, intent="execute") as lifecycle:
             try:
+                enforce_execution_deadline()
                 result: Any = self._execute(connection=connection, sql=sql)
             except Exception as error:
+                try:
+                    enforce_execution_deadline()
+                except ExecutionDurationLimitError as deadline_error:
+                    raise deadline_error from error
                 attributed_error: Exception = attribute_failed_sql(error=error, sql=sql)
                 if attributed_error is error:
                     raise

src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py

@@ -70,6 +70,7 @@
 from sqlbuild.adapters.snowflake.constants import (
     BASE_TABLE_METADATA_TYPE,
     EXTERNAL_BROWSER_AUTHENTICATOR,
+    MAX_STATEMENT_TIMEOUT_SECONDS,
     MFA_AUTHENTICATOR,
     NUMBER_TYPE_NAME,
     OAUTH_AUTHORIZATION_CODE_AUTHENTICATOR,
@@ -102,6 +103,8 @@ class SnowflakeAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter):
 
     adapter_name: ClassVar[str] = BuiltinAdapter.SNOWFLAKE.value
     sql_analysis_dialect_name: ClassVar[str | None] = "snowflake"
+    execution_duration_limit_seconds: ClassVar[int | None] = MAX_STATEMENT_TIMEOUT_SECONDS
+
     max_identifier_length: ClassVar[int] = 255
     state_tables_transient: ClassVar[bool] = True
     connection_routing_keys: ClassVar[frozenset[str]] = frozenset(

src/sqlbuild/adapters/snowflake/classes/snowflake_cursor.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import math
 import time
 from collections.abc import Callable, Iterator
 from datetime import UTC, datetime
@@ -15,6 +16,10 @@
 from sqlbuild.cost.classes.statement_ledger import StatementLedger
 from sqlbuild.cost.constants import SQLBUILD_QUERY_TAG_APP
 from sqlbuild.cost.models import CostResourceContext, StatementExecutionTelemetry
+from sqlbuild.runtime.execution_limits.main.current_execution_deadline import (
+    current_execution_deadline,
+)
+from sqlbuild.runtime.execution_limits.models import ExecutionDeadline
 from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle
 from sqlbuild.runtime.observability.main.current_execution_identity import (
     current_execution_identity,
@@ -23,6 +28,7 @@
 
 _LOGGER: logging.Logger = logging.getLogger("sqlbuild.cost")
 _QUERY_TAG_PARAMETER: str = "QUERY_TAG"
+_STATEMENT_TIMEOUT_PARAMETER: str = "STATEMENT_TIMEOUT_IN_SECONDS"
 _STATEMENT_PARAMETER_ERROR_FRAGMENT: str = "STATEMENT PARAMETER"
 _RAW_CURSOR_ATTRIBUTE: str = "raw_cursor"
 _EXECUTEMANY_INTENT: str = "executemany"
@@ -50,8 +56,9 @@ def __setattr__(self, name: str, value: Any) -> None:
     def execute(self, sql: str, *args: Any, **kwargs: Any) -> Any:
         context: CostResourceContext | None = CostContext.current()
         statement_id: str = _statement_id()
+        deadline_kwargs: dict[str, Any] = _with_execution_deadline(kwargs=kwargs)
         tagged_kwargs, query_tag_injected = _with_query_tag(
-            kwargs=kwargs, context=context, statement_id=statement_id
+            kwargs=deadline_kwargs, context=context, statement_id=statement_id
         )
         return self._execute_with_telemetry(
             operation=self.raw_cursor.execute,
@@ -67,8 +74,9 @@ def execute(self, sql: str, *args: Any, **kwargs: Any) -> Any:
     def executemany(self, sql: str, *args: Any, **kwargs: Any) -> Any:
         context: CostResourceContext | None = CostContext.current()
         statement_id: str = _statement_id()
+        deadline_kwargs: dict[str, Any] = _with_execution_deadline(kwargs=kwargs)
         tagged_kwargs, query_tag_injected = _with_query_tag(
-            kwargs=kwargs, context=context, statement_id=statement_id
+            kwargs=deadline_kwargs, context=context, statement_id=statement_id
         )
         return self._execute_with_telemetry(
             operation=self.raw_cursor.executemany,
@@ -120,7 +128,9 @@ def _execute_with_telemetry(
                         "Snowflake query tagging was rejected before submission; "
                         "continuing without a query tag"
                     )
-                    retry_kwargs: dict[str, Any] = _without_query_tag(kwargs=kwargs)
+                    retry_kwargs: dict[str, Any] = _with_execution_deadline(
+                        kwargs=_without_query_tag(kwargs=kwargs)
+                    )
                     try:
                         result = operation(sql, *args, **retry_kwargs)
                     except Exception as retry_error:
@@ -284,6 +294,20 @@ def _with_query_tag(
     return tagged_kwargs, True
 
 
+def _with_execution_deadline(*, kwargs: dict[str, Any]) -> dict[str, Any]:
+    deadline: ExecutionDeadline | None = current_execution_deadline()
+    if deadline is None:
+        return dict(kwargs)
+    deadline.enforce()
+    statement_params: dict[str, str] = dict(kwargs.get("_statement_params") or {})
+    statement_params[_STATEMENT_TIMEOUT_PARAMETER] = str(
+        max(1, math.ceil(deadline.remaining_seconds))
+    )
+    deadline_kwargs: dict[str, Any] = dict(kwargs)
+    deadline_kwargs["_statement_params"] = statement_params
+    return deadline_kwargs
+
+
 def _without_query_tag(*, kwargs: dict[str, Any]) -> dict[str, Any]:
     retry_kwargs: dict[str, Any] = dict(kwargs)
     statement_params: dict[str, str] = dict(retry_kwargs.get("_statement_params") or {})

src/sqlbuild/adapters/snowflake/constants.py

@@ -1,5 +1,7 @@
 """Snowflake adapter constants."""
 
+MAX_STATEMENT_TIMEOUT_SECONDS: int = 604_800
+
 BASE_TABLE_METADATA_TYPE: str = "BASE TABLE"
 VIEW_RELATION_TYPE_TOKEN: str = "view"
 EXTERNAL_BROWSER_AUTHENTICATOR: str = "externalbrowser"

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

@@ -7,6 +7,9 @@
 from typing import Any
 
 from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
+from sqlbuild.cli.commands._helpers.build_planning.execution_limits import (
+    enforce_model_execution_limit,
+)
 from sqlbuild.cli.commands._helpers.planning.external_refs import (
     resolve_external_sql_reference_resolver,
 )
@@ -69,7 +72,7 @@ def build_defer_clone_boundary_selectors(
     cli_vars: dict[str, object] | None,
     project_dir: Path,
     auto_load_sources: bool,
-) -> tuple[CompiledProject, tuple[str, ...], tuple[str, ...]]:
+) -> tuple[CompiledProject, tuple[str, ...], tuple[str, ...], int]:
     """Resolve out-of-selection clone boundaries and the view chain to rebuild over them."""
 
     project: CompiledProject = build_compiled_project(
@@ -94,6 +97,7 @@ def build_defer_clone_boundary_selectors(
         project,
         defer_clone_boundary_selectors(scope=scope),
         defer_clone_view_chain_selectors(scope=scope),
+        selected_executable_model_count(scope=scope),
     )
 
 
@@ -142,6 +146,20 @@ def defer_clone_view_chain_selectors(*, scope: PlannerScope) -> tuple[str, ...]:
     return tuple(sorted(key.name for key in view_keys))
 
 
+def selected_executable_model_count(*, scope: PlannerScope) -> int:
+    """Count selected models except those explicitly disabled."""
+
+    count: int = 0
+    for key in scope.selected_keys:
+        if key.resource_type != CompiledResourceType.MODEL:
+            continue
+        model: CompiledModel | None = scope.models_by_name.get(key.name)
+        if model is not None and model.config.values.get("enabled") is False:
+            continue
+        count += 1
+    return count
+
+
 def run_defer_clone_boundary_prephase(
     *,
     request: BuildCommandRequest,
@@ -153,7 +171,12 @@ def run_defer_clone_boundary_prephase(
     cloned_project: CompiledProject
     boundary_selectors: tuple[str, ...]
     view_chain_selectors: tuple[str, ...]
-    cloned_project, boundary_selectors, view_chain_selectors = build_defer_clone_boundary_selectors(
+    (
+        cloned_project,
+        boundary_selectors,
+        view_chain_selectors,
+        selected_model_count,
+    ) = build_defer_clone_boundary_selectors(
         discovered_inputs=invocation.discovered_inputs,
         adapter=invocation.adapter,
         selected_target=request.selected_target,
@@ -165,6 +188,11 @@ def run_defer_clone_boundary_prephase(
         project_dir=invocation.effective_project_dir,
         auto_load_sources=invocation.should_load_sources,
     )
+    enforce_model_execution_limit(
+        model_count=selected_model_count,
+        target_name=invocation.effective_target_name,
+        limits=invocation.execution_limits,
+    )
     run_defer_clone_prephase(
         inputs=DeferClonePrephaseInputs(
             discovered_inputs=invocation.discovered_inputs,

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

@@ -0,0 +1,39 @@
+"""Target-scoped build execution-limit enforcement."""
+
+from __future__ import annotations
+
+from sqlbuild.cli.commands.exceptions import CliUserError
+from sqlbuild.compiler.planner.models import PlanOutput
+from sqlbuild.compiler.planner.types import PlanAction
+from sqlbuild.spec.contracts.models import ExecutionLimitsConfig
+
+
+def executable_model_count(*, plan: PlanOutput) -> int:
+    """Return the number of model entries that would execute."""
+
+    return sum(entry.action != PlanAction.SKIP for entry in plan.model_entries)
+
+
+def enforce_model_execution_limit(
+    *,
+    model_count: int,
+    target_name: str | None,
+    limits: ExecutionLimitsConfig,
+) -> None:
+    """Reject a build whose expanded model count exceeds its target policy."""
+
+    maximum: int | None = limits.max_models
+    if maximum is None or model_count <= maximum:
+        return
+    target_label: str = target_name or "default"
+    raise CliUserError(
+        "Build plan exceeds the configured model limit\n\n"
+        f"Target:          {target_label}\n"
+        f"Selected models: {model_count}\n"
+        f"Maximum models:  {maximum}\n"
+        "Configuration:   "
+        f"targets.{target_label}.execution_limits.max_models\n\n"
+        "No warehouse changes were made.",
+        code="C413",
+        help=limits.remediation,
+    )