fix(scenario): honour promotion mode and run scenarios concurrently

Makefile

@@ -121,7 +121,11 @@ E2E_DUCKDB_CLI_DATA_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_authored_check_sql.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_scenario.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_capture.py \
-	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_namespace.py
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_namespace.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_promotion.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_concurrency.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_interrupts.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_empty_fixture.py
 
 E2E_DUCKDB_CLI_PATHS := \
 	tests/e2e/scripts/cli_preview \

fensu.toml

@@ -99,6 +99,12 @@ path = "src/sqlbuild/compiler/planner/_helpers/warehouse/source_columns.py"
 symbols = ["gather_source_inspection"]
 reason = "Each type-enforced expression source is its own query whose output columns can only be read by running it; table sources are listed in one batch per database."
 
+[[targets.python.rule_exceptions]]
+rule = "XSB051"
+path = "src/sqlbuild/executor/scenario/_helpers/lifecycle/cleanup.py"
+symbols = ["existing_scenario_cleanup_targets"]
+reason = "Scenario cleanup lists every scenario-owned relation in one batch per database instead of dropping each planned relation blindly."
+
 [[targets.python.rule_exceptions]]
 rule = "XSB051"
 path = "src/sqlbuild/executor/migrations/_helpers/columns.py"

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

@@ -32,6 +32,7 @@ sqb scenario test [flags]
 | `--scenario-namespace NAMESPACE` | Isolate warehouse artifacts for this run; overrides environment and config |
 | `--local` | Run locally against DuckDB using captured JSONL snapshots |
 | `--strict` | Treat missing/stale local snapshots as errors instead of skips |
+| `--concurrency N` | Run up to `N` scenarios at once (default: the project `concurrency` setting) |
 | `--sync-snapshots` | Capture missing/stale snapshots before local run (requires `--local`) |
 | `--refresh` | Recapture all selected snapshots before local run (requires `--local`) |
 | `--force` | Bypass snapshot capture safety limits |
@@ -93,17 +94,20 @@ sqb scenario test --local --strict
 Remote scenarios report per-scenario PASS/FAIL with nested check rows:
 
 ```
-daily_revenue_minimal                                            PASS
+daily_revenue_minimal                                            PASS  0.84s
     expect    expected daily_revenue                             PASS
     expect    assertion no_negative_revenue                      PASS
 
-PASS=1  FAIL=0  TOTAL=1
+PASS=1  FAIL=0  TOTAL=1  (0.91s)
 ```
 
+Each scenario row shows its wall time and the summary shows the total. `--json` reports the same
+values as `scenarios[].duration_ms` and `summary.duration_ms`.
+
 Local scenarios add ERROR and SKIP statuses:
 
 ```
-PASS=2  FAIL=0  ERROR=0  SKIP=1  TOTAL=3
+PASS=2  FAIL=0  ERROR=0  SKIP=1  TOTAL=3  (1.20s)
 ```
 
 Failed scenarios suggest rerunning with `--retain` for inspection. Local runs always keep the DuckDB file at `target/run/scenarios/<scenario_name>/local.duckdb`.

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/models/materializations.md

@@ -33,7 +33,7 @@ FROM __ref("stg_orders")
 GROUP BY customer_id
 ```
 
-Projects may opt into `settings.table_promotion_mode = "immediate"`. Immediate promotion replaces the destination before audits, so a failed audit does not restore the old table. Immediate promotion rejects models that require declared-type enforcement or `contract enforced`; use staged promotion for those guarantees.
+Projects may opt into `settings.table_promotion_mode = "immediate"`. Immediate promotion replaces the destination before audits, so a failed audit does not restore the old table. Immediate promotion rejects models that require declared-type enforcement or `contract enforced`; use staged promotion for those guarantees. A table or incremental model with `contract enforced` under immediate promotion is a `K011` compile error, so `sqb compile`, `sqb plan`, `sqb build`, and `sqb scenario test` report it before any warehouse work.
 
 ## Incremental
 

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/scenarios.md

@@ -106,6 +106,10 @@ SELECT 1
 
 For sources with two-part identity, use double underscores: `__source__raw__orders`.
 
+To mock a contracted source, ref, or seed with no rows, use `SELECT * FROM __empty_fixture()` as
+the complete fixture body. SQLBuild expands it to the relation's declared, typed columns with a
+false filter, as in [unit tests](testing.md).
+
 Every scenario must have at least one fixture CTE and at least one `__expected__` or `__assert__` CTE.
 
 Scenario SQL uses macros, enums, and constants available from the scenario file's directory under
@@ -145,13 +149,32 @@ Scenario artifacts are physically isolated from production:
 
 ### Execution flow
 
-1. Clean any existing scenario artifacts from a previous run in the effective run namespace
+1. Drop any scenario artifacts a previous run left in the effective run namespace (one catalog read finds them)
 2. Materialize source, ref, and seed fixtures as physical tables
 3. Load required project seeds
 4. Build required models in dependency order (incremental models run as full-refresh in scenarios)
 5. Run expected-output comparisons (order-insensitive)
 6. Run zero-row assertions
-7. Clean up this namespace's scenario-owned artifacts (unless `--retain`)
+7. Clean up this namespace's scenario-owned artifacts (unless `--retain`), including after a failure or Ctrl-C
+
+Table models use the same promotion as `sqb build`: the project's `table_promotion_mode`, or the
+adapter default (staged) when it is unset. An enforced contract needs staged promotion, so its
+runtime contract check runs before the scenario reaches its expectations; with
+`table_promotion_mode = "immediate"` such a model is a `K011` compile error, reported before any
+scenario runs, exactly as in `sqb compile`, `sqb plan`, and `sqb build`.
+
+Independent scenarios run concurrently, up to `--concurrency` or the project `concurrency`
+setting. Each scenario writes only its own prefixed relations, so concurrent scenarios never
+share an artifact. Results are reported in selection order with each scenario's wall time, and the
+summary includes the total time. On DuckDB, concurrent scenarios use separate connections to the
+same database file; an in-memory database (`:memory:`) gives each connection its own database, so
+fixtures still isolate every scenario.
+
+On Ctrl-C, SQLBuild prints one notice, cancels scenarios that have not started, cancels the
+statements running scenarios have in flight, drops their relations (unless `--retain`), and only
+then exits. A second Ctrl-C skips that cleanup and exits immediately; the next run of the same
+scenarios drops whatever they left behind. `--local` replay and `sqb scenario capture` run one
+scenario at a time.
 
 ### Parallel CI runs
 

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

@@ -149,6 +149,18 @@ def supports_unqualified_function_fingerprints(self) -> bool:
     def supports_table_functions(self) -> bool:
         return False
 
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+
+        del connection
+        return False
+
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+
+        del database, schema
+        return False
+
     def supports_concurrent_microbatch_dml(self) -> bool:
         """Return whether disjoint same-target delete/insert batches may run concurrently."""
 

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

@@ -112,6 +112,18 @@ def get_columns_for_relations(
     )
     relation_grants_supported: ClassVar[bool] = False
 
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+
+        connection.interrupt()
+        return True
+
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+
+        del database, schema
+        return True
+
     def supports_zero_copy_clone(self) -> bool:
         return False
 

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

@@ -101,6 +101,16 @@ def maximum_identifier_length(self) -> int:
         """Return the maximum unqualified identifier length supported by the adapter."""
         ...
 
+    @abstractmethod
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+        ...
+
+    @abstractmethod
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+        ...
+
     @abstractmethod
     def supports_concurrent_microbatch_dml(self) -> bool:
         """Return whether disjoint delete/insert microbatches may execute concurrently."""

src/sqlbuild/adapter/relations/_helpers/identifier_fitting.py

@@ -1,4 +1,4 @@
-"""Identifier-limit fitting implementations for generated relation names."""
+"""Fit generated relation names within identifier limits measured in UTF-8 bytes."""
 
 from __future__ import annotations
 
@@ -13,13 +13,13 @@ def fit_artifact_logical_name_impl(
 ) -> str:
     """Fit a readable logical component with a deterministic hash suffix."""
 
-    max_logical_length: int = identifier_limit - len(fixed_prefix)
+    max_logical_length: int = identifier_limit - _byte_length(fixed_prefix)
     if max_logical_length < 1:
         raise SharedInputError(
             f"{artifact_label} prefix '{fixed_prefix}' does not fit within identifier "
             f"limit {identifier_limit}"
         )
-    if len(logical_name) <= max_logical_length:
+    if _byte_length(logical_name) <= max_logical_length:
         return logical_name
 
     suffix_length: int = SHORTENED_LOGICAL_NAME_HASH_LENGTH + 1
@@ -32,4 +32,32 @@ def fit_artifact_logical_name_impl(
         :SHORTENED_LOGICAL_NAME_HASH_LENGTH
     ]
     prefix_length: int = max_logical_length - suffix_length
-    return f"{logical_name[:prefix_length]}_{logical_hash}"
+    return f"{_truncate_bytes(value=logical_name, max_bytes=prefix_length)}_{logical_hash}"
+
+
+def fit_auxiliary_relation_name_impl(*, base_name: str, suffix: str, identifier_limit: int) -> str:
+    """Append a suffix, shortening the base with a stable hash when the whole name is too long."""
+
+    if _byte_length(base_name) + _byte_length(suffix) <= identifier_limit:
+        return f"{base_name}{suffix}"
+    hash_part: str = (
+        "_"
+        + hashlib.sha256(base_name.encode("utf-8")).hexdigest()[:SHORTENED_LOGICAL_NAME_HASH_LENGTH]
+    )
+    kept_length: int = identifier_limit - _byte_length(suffix) - len(hash_part)
+    if kept_length < 1:
+        raise SharedInputError(
+            f"auxiliary relation suffix '{suffix}' for '{base_name}' cannot fit within "
+            f"identifier limit {identifier_limit}"
+        )
+    return f"{_truncate_bytes(value=base_name, max_bytes=kept_length)}{hash_part}{suffix}"
+
+
+def _byte_length(value: str) -> int:
+    return len(value.encode("utf-8"))
+
+
+def _truncate_bytes(*, value: str, max_bytes: int) -> str:
+    """Keep the longest prefix whose UTF-8 encoding fits, never splitting a character."""
+
+    return value.encode("utf-8")[:max_bytes].decode("utf-8", errors="ignore")

src/sqlbuild/adapter/relations/main/fit_auxiliary_relation_name.py

@@ -0,0 +1,15 @@
+"""Auxiliary relation name fitting entrypoint."""
+
+from __future__ import annotations
+
+from sqlbuild.adapter.relations._helpers.identifier_fitting import (
+    fit_auxiliary_relation_name_impl,
+)
+
+
+def fit_auxiliary_relation_name(*, base_name: str, suffix: str, identifier_limit: int) -> str:
+    """Return ``base_name + suffix`` fitted within the limit, keeping the suffix and a base hash."""
+
+    return fit_auxiliary_relation_name_impl(
+        base_name=base_name, suffix=suffix, identifier_limit=identifier_limit
+    )

src/sqlbuild/adapters/bigquery/classes/bigquery_adapter.py

@@ -1512,6 +1512,18 @@ def close(self, connection: _BigQueryConnection) -> None:
     def supports_transactions(self) -> bool:
         return False
 
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+
+        del connection
+        return False
+
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+
+        del database, schema
+        return True
+
     def supports_zero_copy_clone(self) -> bool:
         return True
 

src/sqlbuild/adapters/databricks/classes/databricks_adapter.py

@@ -1170,6 +1170,18 @@ def render_apply_historical_check_snapshot_changes(
             invalidate_hard_deletes=invalidate_hard_deletes,
         )
 
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+
+        del connection
+        return False
+
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+
+        del schema
+        return database is not None
+
     def supports_zero_copy_clone(self) -> bool:
         return True
 

src/sqlbuild/adapters/postgres/classes/postgres_adapter.py

@@ -140,6 +140,18 @@ def get_columns_for_relations(
         historical_insert=HistoricalSnapshotInsertStyle.WITH_INSERT,
     )
 
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+
+        connection.raw_connection.cancel_safe()
+        return True
+
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+
+        del database, schema
+        return True
+
     def supports_zero_copy_clone(self) -> bool:
         return False
 

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

@@ -2843,6 +2843,18 @@ def render_cursor_bound_literal(self, *, value: str, cursor_type: str | None) ->
             return f"'{escaped_value}'"
         return f"'{value}'"
 
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+
+        del connection
+        return False
+
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+
+        del database, schema
+        return True
+
     def supports_zero_copy_clone(self) -> bool:
         return True
 

src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py

@@ -2807,6 +2807,18 @@ def supports_table_functions(self) -> bool:
     def supports_unqualified_function_fingerprints(self) -> bool:
         return False
 
+    def interrupt_connection(self, connection: Any) -> bool:
+        """Cancel a connection's running statement from another thread; False if unsupported."""
+
+        del connection
+        return False
+
+    def lists_relations_authoritatively(self, *, database: str | None, schema: str) -> bool:
+        """Return whether ``list_relations`` reports every relation in this scope."""
+
+        del database, schema
+        return True
+
     def supports_zero_copy_clone(self) -> bool:
         return False
 

src/sqlbuild/cli/commands/_helpers/entry/dispatch.py

@@ -679,6 +679,7 @@ def _dispatch_scenario_command(
                 retain=args.scenario_retain,
                 local=args.scenario_local,
                 strict=args.scenario_strict,
+                concurrency=args.concurrency,
                 sync_snapshots=args.scenario_sync_snapshots,
                 refresh=args.scenario_refresh,
                 limit_inputs=ScenarioSnapshotLimitInputs(

src/sqlbuild/cli/commands/_helpers/entry/parser.py

@@ -792,6 +792,7 @@ def _add_workspace_parsers(
     scenario_test_parser.add_argument("--retain", dest="scenario_retain", action="store_true")
     scenario_test_parser.add_argument("--local", dest="scenario_local", action="store_true")
     scenario_test_parser.add_argument("--strict", dest="scenario_strict", action="store_true")
+    scenario_test_parser.add_argument("--concurrency", type=int, default=None)
     scenario_test_parser.add_argument("--json", action="store_true", default=False)
     _ = add_execution_json_output_arg(scenario_test_parser)
     _ = add_select_args(scenario_test_parser)