fix(scenario): cancel in-flight statements on Ctrl-C and fit names by bytes

Running every scenario through a worker pool and absorbing repeat
interrupts made long statements uninterruptible, even at concurrency 1.
Concurrency 1, --local replay and capture now run on the main thread again,
so Ctrl-C reaches the driver as before. A SIGINT handler records the stop
and cancels in-flight statements through the new adapter
interrupt_connection (DuckDB interrupt, Postgres cancel_safe); without it a
DuckDB query interrupted only by the signal kept running in the background
and blocked cleanup. Concurrent runs cancel queued scenarios, cancel or
close in-flight worker connections, wait for teardown, and sweep started
scenarios on a valid or fresh connection. A second Ctrl-C prints one notice
and exits at once; daemon workers never block exit, and the next run of
the same scenarios drops what was left behind.

Identifier fitting now measures UTF-8 bytes and cuts on character
boundaries, so non-ASCII names fit Postgres's 63-byte limit.

Author: kvlonge

Makefile

@@ -124,6 +124,7 @@ E2E_DUCKDB_CLI_DATA_PATHS := \
 	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 := \

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

@@ -170,9 +170,11 @@ summary includes the total time. On DuckDB, concurrent scenarios use separate co
 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, stops running
-scenarios at their next step, drops their relations (unless `--retain`), and only then exits.
-Further Ctrl-C presses during that cleanup are ignored.
+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,12 @@ 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."""
 

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

@@ -112,6 +112,12 @@ 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."""
 

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

@@ -101,6 +101,11 @@ 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."""

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,22 +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 len(base_name) + len(suffix) <= identifier_limit:
+    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 - len(suffix) - len(hash_part)
+    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"{base_name[:kept_length]}{hash_part}{suffix}"
+    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/adapters/bigquery/classes/bigquery_adapter.py

@@ -1512,6 +1512,12 @@ 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."""
 

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

@@ -1170,6 +1170,12 @@ 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."""
 

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

@@ -140,6 +140,12 @@ 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."""
 

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

@@ -2843,6 +2843,12 @@ 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."""