fix(scenario): run scenario fixture and check SQL as authored

Scenario expected, assertion, helper and fixture SQL was parsed and
regenerated with the generic dialect to swap relation markers, so
dialect syntax was rewritten or rejected (S506) and the executed SQL
differed from what was authored. Markers are now replaced in the
authored text, skipping comments and quoted text under the project
adapter's lexical rules, also for --local replay: backslash escapes on
Snowflake, BigQuery and Databricks, E'...' strings on DuckDB and
Postgres, nested block comments where the dialect nests them, and
BigQuery '#' and Snowflake '//' line comments. A marker the scenario
cannot resolve, or SQL with an unclosed string or comment, fails
planning with S511 naming the scenario file instead of being left in
the executed SQL. Local replay no longer regenerates SQL captured from
DuckDB.

Author: kvlonge

Makefile

@@ -116,6 +116,7 @@ E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
 E2E_DUCKDB_CLI_DATA_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/load \
 	tests/e2e/src/sqlbuild/cli/commands/main/providers \
+	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

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

@@ -269,7 +269,7 @@ This:
 1. Checks snapshot freshness via the input fingerprint (missing/stale snapshots are skipped by default)
 2. Creates a temporary DuckDB database at `target/run/scenarios/<scenario_name>/local.duckdb`
 3. Loads JSONL snapshots into typed DuckDB tables using column metadata from `scenario.json`
-4. Transpiles model and check SQL from the project adapter dialect to DuckDB
+4. Transpiles model and check SQL from the project adapter dialect to DuckDB; SQL captured from a DuckDB project runs as authored
 5. Builds functions, models, and runs expected/assertion checks in DuckDB
 6. Keeps the local DuckDB file for inspection (it lives under `target/`, so it's always retained)
 
@@ -329,3 +329,4 @@ See the [CLI reference](../cli/scenario.md) for full command documentation.
 
 - Custom materializations are not supported in scenarios yet. Scenario models using custom materializations will fail with a clear error.
 - Local replay transpiles SQL from the project adapter dialect to DuckDB. Adapter-specific SQL that cannot be translated will produce a clear error with the failing resource name and reason.
+- Outside local replay, scenario fixture, expected and assertion SQL runs as authored: SQLBuild replaces `__ref`, `__source`, `__seed` and `__dbt_ref` calls with relation names and may wrap a partial fixture to add its missing columns, but never regenerates the query. Calls inside comments and quoted text are left unchanged. Quoted text and comments follow the project adapter's rules, including with `--local`: backslash escapes on Snowflake, BigQuery and Databricks, `E'...'` strings on DuckDB and Postgres, nested block comments on DuckDB, Postgres, Databricks and SQL Server, `#` line comments on BigQuery, and `//` line comments on Snowflake. A call naming a relation the scenario does not provide, or SQL with an unclosed string or comment, fails planning with `S511` naming the scenario file.

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

@@ -75,6 +75,7 @@
 )
 from sqlbuild.compiler.compile.types import FunctionLanguage
 from sqlbuild.compiler.source_freshness.models import SourceFreshnessRecord
+from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from sqlbuild.spec.contracts.constants import DEFAULT_SEED_CSV_SETTINGS
 from sqlbuild.spec.contracts.models import SeedCsvSettings
 from sqlbuild.sql_values.exceptions import SqlValueRenderingError
@@ -101,6 +102,7 @@ class BaseAdapter(RetentionAdapterMixin, StrictAdapter):
 
     adapter_name: ClassVar[str]
     sql_analysis_dialect_name: ClassVar[str | None] = None
+    sql_lexical_syntax: ClassVar[SqlLexicalSyntax] = SqlLexicalSyntax()
     max_identifier_length: ClassVar[int] = 63
     state_tables_transient: ClassVar[bool] = False
     relation_grants_supported: ClassVar[bool] = True

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

@@ -80,6 +80,7 @@
 from sqlbuild.adapter.type_system.main.types_equal import types_equal
 from sqlbuild.compiler.compile.types import FunctionLanguage
 from sqlbuild.compiler.source_freshness.models import SourceFreshnessRecord
+from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from sqlbuild.diagnostics.main.log_sql import log_sql
 from sqlbuild.spec.contracts.constants import DEFAULT_SEED_CSV_SETTINGS
 from sqlbuild.spec.contracts.models import SeedCsvSettings
@@ -100,6 +101,9 @@ def get_columns_for_relations(
         )
 
     sql_analysis_dialect_name: ClassVar[str | None] = "duckdb"
+    sql_lexical_syntax: ClassVar[SqlLexicalSyntax] = SqlLexicalSyntax(
+        escape_string_prefix=True, nested_block_comments=True
+    )
     relation_grants_supported: ClassVar[bool] = False
 
     def supports_zero_copy_clone(self) -> bool:

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

@@ -127,6 +127,7 @@
 )
 from sqlbuild.compiler.compile.types import FunctionLanguage
 from sqlbuild.compiler.source_freshness.models import SourceFreshnessRecord
+from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from sqlbuild.diagnostics.main.log_sql import log_sql
 from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle
 from sqlbuild.spec.contracts.constants import DEFAULT_SEED_CSV_SETTINGS
@@ -150,6 +151,11 @@ def get_columns_for_relations(
 
     adapter_name: ClassVar[str] = BuiltinAdapter.BIGQUERY.value
     sql_analysis_dialect_name: ClassVar[str | None] = "bigquery"
+    sql_lexical_syntax: ClassVar[SqlLexicalSyntax] = SqlLexicalSyntax(
+        backslash_escape_quotes=frozenset({"'", '"', "`"}),
+        triple_quoted_strings=True,
+        line_comment_prefixes=frozenset({"--", "#"}),
+    )
     views_read_with_reader_access: ClassVar[bool] = True
     max_identifier_length: ClassVar[int] = 1024
     _snapshot_sql_dialect: ClassVar[SnapshotSqlDialect] = SnapshotSqlDialect(

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

@@ -115,6 +115,7 @@
 )
 from sqlbuild.compiler.compile.types import FunctionLanguage
 from sqlbuild.compiler.source_freshness.models import SourceFreshnessRecord
+from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from sqlbuild.diagnostics.main.log_sql import log_sql
 from sqlbuild.spec.contracts.constants import DEFAULT_SEED_CSV_SETTINGS
 from sqlbuild.spec.contracts.models import SeedCsvSettings
@@ -136,6 +137,11 @@ def get_columns_for_relations(
 
     adapter_name: ClassVar[str] = BuiltinAdapter.DATABRICKS.value
     sql_analysis_dialect_name: ClassVar[str | None] = "databricks"
+    sql_lexical_syntax: ClassVar[SqlLexicalSyntax] = SqlLexicalSyntax(
+        backslash_escape_quotes=frozenset({"'", '"'}),
+        raw_string_prefix=True,
+        nested_block_comments=True,
+    )
     max_identifier_length: ClassVar[int] = 255
     _snapshot_sql_dialect: ClassVar[SnapshotSqlDialect] = SnapshotSqlDialect(
         timestamp_type="TIMESTAMP",

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

@@ -98,6 +98,7 @@
 from sqlbuild.adapters.postgres.constants import TABLE_FUNCTION_RETURN_TYPE
 from sqlbuild.compiler.compile.types import FunctionLanguage
 from sqlbuild.compiler.source_freshness.models import SourceFreshnessRecord
+from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from sqlbuild.diagnostics.main.log_sql import log_sql
 from sqlbuild.spec.contracts.constants import DEFAULT_SEED_CSV_SETTINGS
 from sqlbuild.spec.contracts.models import SeedCsvSettings
@@ -119,6 +120,9 @@ def get_columns_for_relations(
 
     adapter_name: ClassVar[str] = BuiltinAdapter.POSTGRES.value
     sql_analysis_dialect_name: ClassVar[str | None] = "postgres"
+    sql_lexical_syntax: ClassVar[SqlLexicalSyntax] = SqlLexicalSyntax(
+        escape_string_prefix=True, nested_block_comments=True
+    )
     max_identifier_length: ClassVar[int] = 63
     _snapshot_sql_dialect: ClassVar[SnapshotSqlDialect] = SnapshotSqlDialect(
         timestamp_type="TIMESTAMP",

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

@@ -138,6 +138,7 @@
 )
 from sqlbuild.compiler.compile.types import FunctionLanguage
 from sqlbuild.compiler.source_freshness.models import SourceFreshnessRecord
+from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from sqlbuild.cost.main.collection import collect_snowflake_cost
 from sqlbuild.cost.models import RunCostSummary
 from sqlbuild.cost.types import CostCapability, CostStatus
@@ -164,6 +165,9 @@ class SnowflakeAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter):
 
     adapter_name: ClassVar[str] = BuiltinAdapter.SNOWFLAKE.value
     sql_analysis_dialect_name: ClassVar[str | None] = "snowflake"
+    sql_lexical_syntax: ClassVar[SqlLexicalSyntax] = SqlLexicalSyntax(
+        backslash_escape_quotes=frozenset({"'"}), line_comment_prefixes=frozenset({"--", "//"})
+    )
     execution_duration_limit_seconds: ClassVar[int | None] = MAX_STATEMENT_TIMEOUT_SECONDS
     metadata_inspection_concurrency: ClassVar[int] = _METADATA_INSPECTION_CONCURRENCY
     exact_column_inspection_limit: ClassVar[int] = _EXACT_COLUMN_INSPECTION_LIMIT

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

@@ -105,6 +105,7 @@
 )
 from sqlbuild.compiler.compile.types import FunctionLanguage
 from sqlbuild.compiler.source_freshness.models import SourceFreshnessRecord
+from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from sqlbuild.diagnostics.main.log_sql import log_sql
 from sqlbuild.spec.contracts.constants import DEFAULT_SEED_CSV_SETTINGS
 from sqlbuild.spec.contracts.models import SeedCsvSettings
@@ -135,6 +136,7 @@ def get_columns_for_relations(
 
     adapter_name: ClassVar[str] = BuiltinAdapter.SQLSERVER.value
     sql_analysis_dialect_name: ClassVar[str | None] = "tsql"
+    sql_lexical_syntax: ClassVar[SqlLexicalSyntax] = SqlLexicalSyntax(nested_block_comments=True)
     max_identifier_length: ClassVar[int] = 128
     _snapshot_sql_dialect: ClassVar[SnapshotSqlDialect] = SnapshotSqlDialect(
         timestamp_type="DATETIME2",

src/sqlbuild/cli/commands/_helpers/scenario_execution/local_run.py

@@ -19,7 +19,7 @@
 from sqlbuild.compiler.compile.models import CompiledSqlScenario
 from sqlbuild.compiler.pipeline.models import CompilePipelineResult
 from sqlbuild.executor.pipeline.main.run import run_scenario_local_test_pipeline
-from sqlbuild.executor.scenario.models import ScenarioRunResult
+from sqlbuild.executor.scenario.models import ScenarioLocalReplaySource, ScenarioRunResult
 from sqlbuild.executor.scenario.types import ScenarioLocalRunStatus
 from sqlbuild.presentation.classes.cli_style import CliStyle
 from sqlbuild.presentation.classes.transient_status_reporter import TransientStatusReporter
@@ -34,8 +34,7 @@ def run_local_scenarios(
     adapter: BaseAdapter,
     project_name: str,
     strict: bool,
-    capture_adapter: str,
-    capture_dialect: str,
+    replay_source: ScenarioLocalReplaySource,
     target_dir: Path,
     output_context: ScenarioRunOutputContext,
 ) -> int:
@@ -66,8 +65,7 @@ def run_local_scenarios(
         adapter=adapter,
         project_name=project_name,
         strict=strict,
-        capture_adapter=capture_adapter,
-        capture_dialect=capture_dialect,
+        replay_source=replay_source,
         on_scenario_start=lambda _scenario: (
             scenario_status.start(activity) if status_is_tty else None
         ),