refactor: deduplicate expiration filter logic and add Context-level invalidate test

- Extend _create_expiration_filter_expr to accept optional name= param,
  consolidating the repeated AND-name filter from get_expired_environments
  and delete_expired_environments into a single place
- Add test_invalidate_environment_sync_calls_cleanup_with_name: verifies
  that invalidate_environment(..., sync=True) passes name= to
  _cleanup_environments, guarding the --sync bug fix at the Context level
- Add test_invalidate_environment_no_sync_skips_cleanup: verifies that
  sync=False does not trigger delete_expired_environments

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: mday-io

sqlmesh/core/state_sync/db/environment.py

@@ -181,10 +181,9 @@ def get_expired_environments(
         Returns:
             The list of environment summaries to remove.
         """
-        where: exp.Expr = self._create_expiration_filter_expr(current_ts)
-        if name is not None:
-            where = exp.and_(t.cast(exp.Condition, where), exp.column("name").eq(name))
-        return self._fetch_environment_summaries(where=where)
+        return self._fetch_environment_summaries(
+            where=self._create_expiration_filter_expr(current_ts, name=name)
+        )
 
     def delete_expired_environments(
         self, current_ts: t.Optional[int] = None, name: t.Optional[str] = None
@@ -201,12 +200,9 @@ def delete_expired_environments(
         current_ts = current_ts or now_timestamp()
         expired_environments = self.get_expired_environments(current_ts=current_ts, name=name)
 
-        where: exp.Expr = self._create_expiration_filter_expr(current_ts)
-        if name is not None:
-            where = exp.and_(t.cast(exp.Condition, where), exp.column("name").eq(name))
         self.engine_adapter.delete_from(
             self.environments_table,
-            where=where,
+            where=self._create_expiration_filter_expr(current_ts, name=name),
         )
 
         # Delete the expired environments' corresponding environment statements
@@ -325,16 +321,22 @@ def _environments_query(
             return query.lock(copy=False)
         return query
 
-    def _create_expiration_filter_expr(self, current_ts: int) -> exp.Expr:
+    def _create_expiration_filter_expr(
+        self, current_ts: int, name: t.Optional[str] = None
+    ) -> exp.Expr:
         """Creates a SQLGlot filter expression to find expired environments.
 
         Args:
             current_ts: The current timestamp.
+            name: If provided, adds an equality filter on the environment name.
         """
-        return exp.LTE(
+        where: exp.Expr = exp.LTE(
             this=exp.column("expiration_ts"),
             expression=exp.Literal.number(current_ts),
         )
+        if name is not None:
+            where = exp.and_(t.cast(exp.Condition, where), exp.column("name").eq(name))
+        return where
 
     def _fetch_environment_summaries(
         self, where: t.Optional[str | exp.Expr] = None

tests/core/test_context.py

@@ -1170,6 +1170,40 @@ def test_janitor_environment_not_expired_warning(sushi_context, mocker: MockerFi
     assert "nonexistent_env" in warning_mock.call_args[0][0]
 
 
+@pytest.mark.slow
+def test_invalidate_environment_sync_calls_cleanup_with_name(
+    sushi_context, mocker: MockerFixture
+) -> None:
+    """invalidate_environment(..., sync=True) must pass name= to _cleanup_environments so only the
+    target environment is deleted, not all expired environments."""
+    state_sync_mock = mocker.patch.object(
+        type(sushi_context), "state_sync", new_callable=mocker.PropertyMock
+    ).return_value
+    state_sync_mock.get_expired_environments.return_value = []
+
+    sushi_context.invalidate_environment("dev", sync=True)
+
+    state_sync_mock.invalidate_environment.assert_called_once_with("dev")
+    state_sync_mock.delete_expired_environments.assert_called_once()
+    _, kwargs = state_sync_mock.delete_expired_environments.call_args
+    assert kwargs.get("name") == "dev"
+
+
+@pytest.mark.slow
+def test_invalidate_environment_no_sync_skips_cleanup(
+    sushi_context, mocker: MockerFixture
+) -> None:
+    """invalidate_environment(..., sync=False) should not trigger _cleanup_environments at all."""
+    state_sync_mock = mocker.patch.object(
+        type(sushi_context), "state_sync", new_callable=mocker.PropertyMock
+    ).return_value
+
+    sushi_context.invalidate_environment("dev", sync=False)
+
+    state_sync_mock.invalidate_environment.assert_called_once_with("dev")
+    state_sync_mock.delete_expired_environments.assert_not_called()
+
+
 @pytest.mark.slow
 def test_plan_default_end(sushi_context_pre_scheduling: Context):
     prod_plan_builder = sushi_context_pre_scheduling.plan_builder("prod")