feat: add load_state flag to Context

Add `load_state: bool = True` to `GenericContext.__init__` and AND the
new flag into the two `if any(self._projects):` guards in
`Context.load()`. With `load_state=False`, no remote-snapshot merge
runs, so `state_reader` is never read and the state-sync backend is
never instantiated.

The default preserves today's behavior for every existing caller. CLI
wiring lands in the next commit; this one just introduces the
mechanism and proves it.

Tests patch `EngineAdapterStateSync.get_versions` to raise and assert
it is never called when `load_state=False`. They set a non-empty
`project` on the test config so `any(self._projects)` is truthy and
the gate actually exercises `self._load_state` (with the default
empty project, the outer `any(...)` short-circuits and the test is
vacuous). Verified non-vacuous by flipping `load_state=True` and
confirming both tests fail with the patched RuntimeError.

The plan originally proposed an unreachable Postgres state
connection; `psycopg2` is not installed in the dev env so
`PostgresConnectionConfig` validates at construction time and the
config can't be built. Implementation uses the plan's documented
patch-based fallback instead. Plan updated to reflect the change.

Coding-Agent: pi
Model: anthropic/claude-sonnet-4-6
Signed-off-by: Joe Hartshorn <8881940+j-hartshorn@users.noreply.github.com>

Author: j-hartshorn

docs/2026-05-21_local-only-format/plan.md

@@ -29,10 +29,11 @@ The PR is done when all of these are true:
 - The `Linter` is built per-project in `Context.load()` at `context.py:670-674` before the state-merge block. Linters don't depend on the state-merge having run.
 - `format` and `lint` Click handlers are at `sqlmesh/cli/main.py:343-380` and `sqlmesh/cli/main.py:1168-1185` respectively. Neither needs argument changes.
 
-**Assumptions** (flagged for the implementer to confirm during work):
+**Decisions resolved during implementation:**
 
-- A Postgres connection config pointing at an unreachable host (e.g. `localhost:1`) is sufficient to make state access fail loudly without requiring `psycopg2` to actually attempt a real connection — the failure should happen at `get_versions` cursor-fetch time, not at config validation time. If config validation rejects the host, fall back to patching `EngineAdapterStateSync.get_versions` to raise.
-- No existing test currently exercises the "format/lint succeed despite broken state" path. Verified by `rg 'format.*state\|lint.*state' tests/` returning no relevant matches, but the implementer should confirm before adding the regression.
+- The plan originally suggested configuring an unreachable Postgres state connection. This doesn't work in the dev environment because `psycopg2` is not installed; `PostgresConnectionConfig` runs `_get_engine_import_validator("psycopg2", ...)` at config-validation time (`sqlmesh/core/config/connection.py:1424`) and raises before our code path is reached. Implementation uses the plan's documented fallback: patch `sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions` with `side_effect=RuntimeError(...)` and assert it is never called.
+- The state-merge blocks are guarded by `if any(self._projects):`, where `self._projects = {config.project for config in self.configs.values()}`. `Config.project` defaults to `""` (`sqlmesh/core/config/root.py:142`), and `any({""})` is `False` — so a `Config()` literal short-circuits the guard before the new `self._load_state` term is evaluated, making the test vacuous. The Context-level tests therefore set a non-empty `project` (`Config(project="local_only")` in the format test; an inline rewrite of sushi's `config.py` to add `project="sushi"` in the linter test) to ensure the guard is actually exercised.
+- No existing test exercised the "format/lint succeed despite broken state" path (verified by `rg 'format.*state\|lint.*state' tests/`).
 
 ## Existing patterns & conventions to follow
 
@@ -85,10 +86,10 @@ Two tasks, two commits, one PR.
 **Test cases** (red first, then implementation):
 
 `tests/core/test_format.py` (fast):
-- *`test_format_without_state_load`*: Construct a `Context` with `paths=tmp_path`, a `Config` whose state connection points at `localhost:1` (Postgres), and `load_state=False`. Place one `.sql` model file under `models/` containing SQL that is already in canonical formatted form (so `--check` succeeds for content reasons — the test is about state access, not about formatting decisions). Call `context.format(check=True)` and assert it returns `True` without raising. Assert the file's contents on disk are unchanged.
+- *`test_format_without_state_load`*: Patch `EngineAdapterStateSync.get_versions` to raise `RuntimeError`. Place one `.sql` model under `models/`. Construct `Context(paths=tmp_path, config=Config(project="local_only"), load_state=False)` (non-empty `project` so `any(self._projects)` is truthy and the gate is exercised). Call `context.format(check=True)`. Assert the patched mock has `assert_not_called()`. Verified non-vacuous by flipping to `load_state=True` and confirming the test fails with the patched `RuntimeError`.
 
 `tests/core/linter/test_builtin.py` (fast):
-- *`test_lint_without_state_load`*: Using `copy_to_temp_path("examples/sushi")`, rewrite the sushi `config.py` to enable the linter (existing pattern at `test_builtin.py:21-40`) AND set a Postgres state connection at `localhost:1`. Construct `Context(paths=[sushi_path], load_state=False)`. Call `context.lint_models(raise_on_error=False)` and assert it returns a list (i.e., the linter ran end-to-end) without raising.
+- *`test_lint_without_state_load`*: Using `copy_to_temp_path("examples/sushi")`, rewrite the sushi `config.py` to add `project="sushi"` to the `Config(...)` call (so `any(self._projects)` is truthy). Patch `EngineAdapterStateSync.get_versions` to raise. Construct `Context(paths=[sushi_path], load_state=False)`. Call `context.lint_models(raise_on_error=False)`. Assert the patched mock has `assert_not_called()`. Verified non-vacuous the same way as the format test.
 
 **Implementation outline:**
 

sqlmesh/core/context.py

@@ -363,6 +363,7 @@ class GenericContext(BaseContext, t.Generic[C]):
             connection as it appears in configuration will be used.
         concurrent_tasks: The maximum number of tasks that can use the connection concurrently.
         load: Whether or not to automatically load all models and macros (default True).
+        load_state: Whether to merge remote state into the local project during load (default True). Only meaningful when load=True.
         console: The rich instance used for printing out CLI command results.
         users: A list of users to make known to SQLMesh.
     """
@@ -383,6 +384,7 @@ def __init__(
         concurrent_tasks: t.Optional[int] = None,
         loader: t.Optional[t.Type[Loader]] = None,
         load: bool = True,
+        load_state: bool = True,
         users: t.Optional[t.List[User]] = None,
         config_loader_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
         selector: t.Optional[t.Type[Selector]] = None,
@@ -413,6 +415,7 @@ def __init__(
         self._engine_adapter: t.Optional[EngineAdapter] = None
         self._linters: t.Dict[str, Linter] = {}
         self._loaded: bool = False
+        self._load_state: bool = load_state
         self._selector_cls = selector or NativeSelector
 
         self.path, self.config = t.cast(t.Tuple[Path, C], next(iter(self.configs.items())))
@@ -674,7 +677,7 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
             )
 
         # Load environment statements from state for projects not in current load
-        if any(self._projects):
+        if self._load_state and any(self._projects):
             prod = self.state_reader.get_environment(c.PROD)
             if prod:
                 existing_statements = self.state_reader.get_environment_statements(c.PROD)
@@ -684,7 +687,7 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
 
         uncached = set()
 
-        if any(self._projects):
+        if self._load_state and any(self._projects):
             prod = self.state_reader.get_environment(c.PROD)
 
             if prod:

tests/core/linter/test_builtin.py

@@ -232,3 +232,32 @@ def test_no_missing_unit_tests(tmp_path, copy_to_temp_path):
         assert len(model_violations) == 0, (
             f"Model {model_name} should not have a violation since it has a test"
         )
+
+
+def test_lint_without_state_load(tmp_path, copy_to_temp_path, mocker) -> None:
+    """`lint_models` with `load_state=False` runs end-to-end without touching state sync."""
+    sushi_paths = copy_to_temp_path("examples/sushi")
+    sushi_path = sushi_paths[0]
+
+    with open(sushi_path / "config.py", "r") as f:
+        read_file = f.read()
+
+    # Set a non-empty project name so `any(self._projects)` is truthy and the
+    # state-merge guard in `Context.load()` actually exercises `self._load_state`.
+    read_file = read_file.replace(
+        "config = Config(\n    gateways=",
+        'config = Config(\n    project="sushi",\n    gateways=',
+    )
+    assert 'project="sushi"' in read_file
+
+    with open(sushi_path / "config.py", "w") as f:
+        f.writelines(read_file)
+
+    mock = mocker.patch(
+        "sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions",
+        side_effect=RuntimeError("state should not be accessed"),
+    )
+
+    context = Context(paths=[sushi_path], load_state=False)
+    context.lint_models(raise_on_error=False)
+    mock.assert_not_called()

tests/core/test_format.py

@@ -144,3 +144,21 @@ def test_ignore_formating_files(tmp_path: pathlib.Path):
         model3.read_text(encoding="utf-8")
         == "MODEL (\n  name this.model3,\n  dialect 'duckdb',\n  formatting TRUE\n);\n\nSELECT\n  1 AS col"
     )
+
+
+def test_format_without_state_load(tmp_path: pathlib.Path, mocker: MockerFixture):
+    """`format` with `load_state=False` runs end-to-end without touching state sync."""
+    mock = mocker.patch(
+        "sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions",
+        side_effect=RuntimeError("state should not be accessed"),
+    )
+
+    create_temp_file(
+        tmp_path,
+        pathlib.Path("models/example.sql"),
+        "MODEL(name local.example, dialect 'duckdb'); SELECT 1 AS col",
+    )
+
+    context = Context(paths=tmp_path, config=Config(project="local_only"), load_state=False)
+    context.format(check=True)
+    mock.assert_not_called()