Repository: TobikoData/sqlmesh
Pull request #4732 (2025-06-13 15:55:05)

Chore: Update the plan evaluator to use stages
This update unifies plan evaluation and explanation by utilizing a common intermediate step: building the list of plan stages first.

The plan stages now contain a fully precomputed set of evaluation steps, removing dynamic behavior from the evaluation process. As a result, the following evaluation details are known ahead of time, before evaluation begins, and without producing any side effects:
- A set of snapshots for which physical tables will be checked and created if missing
- All missing intervals that will be backfilled
- A set of snapshots for which views will be created / deleted in the virtual layer
- A set of snapshots for which the schema will be migrated
- The order of operations

The only remaining dynamic step is the creation of physical tables, as it requires querying the data warehouse to determine which tables are missing. We want to avoid touch the data warehouse during the stage construction process.

Among the many benefits of the new approach is that plan evaluations and explanations are now always in sync, as they rely on the same underlying logic for deriving plan stages.


Patch:
diff --git a/pyproject.toml b/pyproject.toml
index 93e034de..ea20c21e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -22,7 +22,7 @@ dependencies = [
     "requests",
     "rich[jupyter]",
     "ruamel.yaml",
-    "sqlglot[rs]~=26.27.0",
+    "sqlglot[rs]~=26.27.1",
     "tenacity",
     "time-machine",
     "json-stream"
@@ -273,6 +273,7 @@ retry_delay = 10
 select = [
     "F401",
     "RET505",
+    "T100",
 ]
 extend-select = ["TID"]
 
diff --git a/sqlmesh/core/plan/__init__.py b/sqlmesh/core/plan/__init__.py
index 903e2e43..8b3ba63e 100644
--- a/sqlmesh/core/plan/__init__.py
+++ b/sqlmesh/core/plan/__init__.py
@@ -8,6 +8,5 @@ from sqlmesh.core.plan.definition import (
 from sqlmesh.core.plan.evaluator import (
     BuiltInPlanEvaluator as BuiltInPlanEvaluator,
     PlanEvaluator as PlanEvaluator,
-    update_intervals_for_new_snapshots as update_intervals_for_new_snapshots,
 )
 from sqlmesh.core.plan.explainer import PlanExplainer as PlanExplainer
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index c2e8846a..d1d9df8f 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -23,6 +23,7 @@ from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.environment import EnvironmentNamingInfo, execute_environment_statements
 from sqlmesh.core.macros import RuntimeStage
 from sqlmesh.core.snapshot.definition import Interval, to_view_mapping
+from sqlmesh.core.plan import stages
 from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.scheduler import Scheduler
 from sqlmesh.core.snapshot import (
@@ -35,11 +36,10 @@ from sqlmesh.core.snapshot import (
     SnapshotTableInfo,
     SnapshotCreationFailedError,
 )
-from sqlmesh.utils import CompletionStatus
-from sqlmesh.core.state_sync import StateSync, StateReader
-from sqlmesh.core.state_sync.base import PromotionResult
+from sqlmesh.utils import to_snake_case
+from sqlmesh.core.state_sync import StateSync
 from sqlmesh.utils.concurrency import NodeExecutionFailedError
-from sqlmesh.utils.errors import PlanError
+from sqlmesh.utils.errors import PlanError, SQLMeshError
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import now
 
@@ -77,12 +77,14 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         self.create_scheduler = create_scheduler
         self.default_catalog = default_catalog
         self.console = console or get_console()
+        self._circuit_breaker: t.Optional[t.Callable[[], bool]] = None
 
     def evaluate(
         self,
         plan: EvaluatablePlan,
         circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
     ) -> None:
+        self._circuit_breaker = circuit_breaker
         self.console.start_plan_evaluation(plan)
         analytics.collector.on_plan_apply_start(
             plan=plan,
@@ -92,110 +94,8 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         )
 
         try:
-            new_snapshots = {s.snapshot_id: s for s in plan.new_snapshots}
-            stored_snapshots = self.state_sync.get_snapshots(plan.environment.snapshots)
-            snapshots = {**new_snapshots, **stored_snapshots}
-            snapshots_by_name = {s.name: s for s in snapshots.values()}
-
-            all_names = {name for name in snapshots_by_name if plan.is_selected_for_backfill(name)}
-
-            deployability_index_for_evaluation = DeployabilityIndex.create(
-                snapshots, start=plan.start
-            )
-            deployability_index_for_creation = deployability_index_for_evaluation
-            if plan.is_dev:
-                before_promote_snapshots = all_names
-                after_promote_snapshots = set()
-            else:
-                before_promote_snapshots = {
-                    s.name
-                    for s in snapshots.values()
-                    if deployability_index_for_evaluation.is_representative(s)
-                    and plan.is_selected_for_backfill(s.name)
-                }
-                after_promote_snapshots = all_names - before_promote_snapshots
-                deployability_index_for_evaluation = DeployabilityIndex.all_deployable()
-
-            # Step 1: Run before_all environment statements before doing anything else
-            execute_environment_statements(
-                adapter=self.snapshot_evaluator.adapter,
-                environment_statements=plan.environment_statements or [],
-                runtime_stage=RuntimeStage.BEFORE_ALL,
-                environment_naming_info=plan.environment.naming_info,
-                default_catalog=self.default_catalog,
-                snapshots=snapshots_by_name,
-                start=plan.start,
-                end=plan.end,
-                execution_time=plan.execution_time,
-            )
-
-            # Step 2:Store new snapshot records in the state and create physical tables for them
-            push_completion_status = self._push(plan, snapshots, deployability_index_for_creation)
-            if push_completion_status.is_nothing_to_do:
-                self.console.log_success(
-                    "" if plan.restatements else "\nSKIP: No physical layer updates to perform"
-                )
-
-            # Step 3: Update the intervals for the new forward-only snapshots
-            update_intervals_for_new_snapshots(plan.new_snapshots, self.state_sync)
-
-            # Step 4: Run audits without evaluations for snapshots that capture audit metadata changes
-            self._run_audits_for_metadata_snapshots(plan, new_snapshots)
-
-            # Step 5: Remove intervals for snapshots that need to be restated
-            self._restate(plan, snapshots_by_name)
-
-            # Step 6: Backfill missing intervals for snapshots that can be backfilled before updating
-            # the schema of production tables
-            first_bf_completion_status = self._backfill(
-                plan,
-                snapshots_by_name,
-                before_promote_snapshots,
-                deployability_index_for_evaluation,
-                circuit_breaker=circuit_breaker,
-            )
-
-            # Step 7: Update the target environment record in the state and migrate table schemas for forward-only
-            # snapshots if deploying to production
-            promotion_result = self._promote(
-                plan, snapshots, before_promote_snapshots, deployability_index_for_creation
-            )
-
-            # Step 8: Backfill missing intervals for snapshots that can be backfilled only after updating
-            # the schema of production tables
-            second_bf_completion_status = self._backfill(
-                plan,
-                snapshots_by_name,
-                after_promote_snapshots,
-                deployability_index_for_evaluation,
-                circuit_breaker=circuit_breaker,
-            )
-
-            if (
-                first_bf_completion_status.is_nothing_to_do
-                and second_bf_completion_status.is_nothing_to_do
-            ):
-                self.console.log_success("SKIP: No model batches to execute")
-
-            # Step 9: Update environment views to point at new physical tables and finalize the environment
-            # record in the state
-            self._update_views(
-                plan, snapshots, promotion_result, deployability_index_for_evaluation
-            )
-
-            # Step 10: Run after_all environment statements
-            execute_environment_statements(
-                adapter=self.snapshot_evaluator.adapter,
-                environment_statements=plan.environment_statements or [],
-                runtime_stage=RuntimeStage.AFTER_ALL,
-                environment_naming_info=plan.environment.naming_info,
-                default_catalog=self.default_catalog,
-                snapshots=snapshots_by_name,
-                start=plan.start,
-                end=plan.end,
-                execution_time=plan.execution_time,
-            )
-
+            plan_stages = stages.build_plan_stages(plan, self.state_sync, self.default_catalog)
+            self._evaluate_stages(plan_stages, plan)
         except Exception as e:
             analytics.collector.on_plan_apply_end(plan_id=plan.plan_id, error=e)
             raise
@@ -204,101 +104,80 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         finally:
             self.console.stop_plan_evaluation()
 
-    def _backfill(
-        self,
-        plan: EvaluatablePlan,
-        snapshots_by_name: t.Dict[str, Snapshot],
-        selected_snapshots: t.Set[str],
-        deployability_index: DeployabilityIndex,
-        circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
-    ) -> CompletionStatus:
-        """Backfill missing intervals for snapshots that are part of the given plan.
-
-        Args:
-            plan: The plan to source snapshots from.
-            selected_snapshots: The snapshots to backfill.
-        """
-        if plan.empty_backfill:
-            intervals_to_add = []
-            for snapshot in snapshots_by_name.values():
-                if not snapshot.evaluatable or not plan.is_selected_for_backfill(snapshot.name):
-                    # Skip snapshots that are not evaluatable or not selected for backfill.
-                    continue
-                intervals = [
-                    snapshot.inclusive_exclusive(plan.start, plan.end, strict=False, expand=False)
-                ]
-                is_deployable = deployability_index.is_deployable(snapshot)
-                intervals_to_add.append(
-                    SnapshotIntervals(
-                        name=snapshot.name,
-                        identifier=snapshot.identifier,
-                        version=snapshot.version,
-                        dev_version=snapshot.dev_version,
-                        intervals=intervals if is_deployable else [],
-                        dev_intervals=intervals if not is_deployable else [],
-                    )
-                )
-            self.state_sync.add_snapshots_intervals(intervals_to_add)
-            return CompletionStatus.NOTHING_TO_DO
-
-        if not plan.requires_backfill or not selected_snapshots:
-            return CompletionStatus.NOTHING_TO_DO
-
-        scheduler = self.create_scheduler(snapshots_by_name.values())
-        completion_status = scheduler.run(
-            plan.environment,
-            plan.start,
-            plan.end,
+    def _evaluate_stages(
+        self, plan_stages: t.List[stages.PlanStage], plan: EvaluatablePlan
+    ) -> None:
+        for stage in plan_stages:
+            stage_name = stage.__class__.__name__
+            handler_name = f"visit_{to_snake_case(stage_name)}"
+            if not hasattr(self, handler_name):
+                raise SQLMeshError(f"Unexpected plan stage: {stage_name}")
+            logger.info("Evaluating plan stage %s", stage_name)
+            handler = getattr(self, handler_name)
+            handler(stage, plan)
+
+    def visit_before_all_stage(self, stage: stages.BeforeAllStage, plan: EvaluatablePlan) -> None:
+        execute_environment_statements(
+            adapter=self.snapshot_evaluator.adapter,
+            environment_statements=stage.statements,
+            runtime_stage=RuntimeStage.BEFORE_ALL,
+            environment_naming_info=plan.environment.naming_info,
+            default_catalog=self.default_catalog,
+            snapshots=stage.all_snapshots,
+            start=plan.start,
+            end=plan.end,
             execution_time=plan.execution_time,
-            restatements={
-                snapshots_by_name[name].snapshot_id: interval
-                for name, interval in plan.restatements.items()
-            },
-            selected_snapshots=selected_snapshots,
-            deployability_index=deployability_index,
-            circuit_breaker=circuit_breaker,
-            end_bounded=plan.end_bounded,
-            interval_end_per_model=plan.interval_end_per_model,
         )
-        if completion_status.is_failure:
-            raise PlanError("Plan application failed.")
 
-        return completion_status
-
-    def _push(
-        self,
-        plan: EvaluatablePlan,
-        snapshots: t.Dict[SnapshotId, Snapshot],
-        deployability_index: t.Optional[DeployabilityIndex] = None,
-    ) -> CompletionStatus:
-        """Push the snapshots to the state sync.
-
-        As a part of plan pushing, snapshot tables are created.
+    def visit_after_all_stage(self, stage: stages.AfterAllStage, plan: EvaluatablePlan) -> None:
+        execute_environment_statements(
+            adapter=self.snapshot_evaluator.adapter,
+            environment_statements=stage.statements,
+            runtime_stage=RuntimeStage.AFTER_ALL,
+            environment_naming_info=plan.environment.naming_info,
+            default_catalog=self.default_catalog,
+            snapshots=stage.all_snapshots,
+            start=plan.start,
+            end=plan.end,
+            execution_time=plan.execution_time,
+        )
 
-        Args:
-            plan: The plan to source snapshots from.
-            deployability_index: Indicates which snapshots are deployable in the context of this creation.
-        """
-        self.state_sync.push_snapshots(plan.new_snapshots)
+    def visit_create_snapshot_records_stage(
+        self, stage: stages.CreateSnapshotRecordsStage, plan: EvaluatablePlan
+    ) -> None:
+        self.state_sync.push_snapshots(stage.snapshots)
         analytics.collector.on_snapshots_created(
-            new_snapshots=plan.new_snapshots, plan_id=plan.plan_id
+            new_snapshots=stage.snapshots, plan_id=plan.plan_id
         )
+        # Update the intervals for the new forward-only snapshots
+        self._update_intervals_for_new_snapshots(stage.snapshots)
+
+    def visit_physical_layer_update_stage(
+        self, stage: stages.PhysicalLayerUpdateStage, plan: EvaluatablePlan
+    ) -> None:
+        skip_message = "" if plan.restatements else "\nSKIP: No physical layer updates to perform"
 
-        snapshots_to_create = get_snapshots_to_create(plan, snapshots)
+        snapshots_to_create = stage.snapshots
+        if not snapshots_to_create:
+            self.console.log_success(skip_message)
+            return
 
         completion_status = None
         progress_stopped = False
         try:
             completion_status = self.snapshot_evaluator.create(
                 snapshots_to_create,
-                snapshots,
+                stage.all_snapshots,
                 allow_destructive_snapshots=plan.allow_destructive_models,
-                deployability_index=deployability_index,
+                deployability_index=stage.deployability_index,
                 on_start=lambda x: self.console.start_creation_progress(
                     x, plan.environment, self.default_catalog
                 ),
                 on_complete=self.console.update_creation_progress,
             )
+            if completion_status.is_nothing_to_do:
+                self.console.log_success(skip_message)
+                return
         except SnapshotCreationFailedError as ex:
             self.console.stop_creation_progress(success=False)
             progress_stopped = True
@@ -316,71 +195,125 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                     success=completion_status is not None and completion_status.is_success
                 )
 
-        return completion_status
+    def visit_backfill_stage(self, stage: stages.BackfillStage, plan: EvaluatablePlan) -> None:
+        if plan.empty_backfill:
+            intervals_to_add = []
+            for snapshot in stage.all_snapshots.values():
+                if not snapshot.evaluatable or not plan.is_selected_for_backfill(snapshot.name):
+                    # Skip snapshots that are not evaluatable or not selected for backfill.
+                    continue
+                intervals = [
+                    snapshot.inclusive_exclusive(plan.start, plan.end, strict=False, expand=False)
+                ]
+                is_deployable = stage.deployability_index.is_deployable(snapshot)
+                intervals_to_add.append(
+                    SnapshotIntervals(
+                        name=snapshot.name,
+                        identifier=snapshot.identifier,
+                        version=snapshot.version,
+                        dev_version=snapshot.dev_version,
+                        intervals=intervals if is_deployable else [],
+                        dev_intervals=intervals if not is_deployable else [],
+                    )
+                )
+            self.state_sync.add_snapshots_intervals(intervals_to_add)
+            self.console.log_success("SKIP: No model batches to execute")
+            return
 
-    def _promote(
-        self,
-        plan: EvaluatablePlan,
-        snapshots: t.Dict[SnapshotId, Snapshot],
-        no_gaps_snapshot_names: t.Optional[t.Set[str]] = None,
-        deployability_index: t.Optional[DeployabilityIndex] = None,
-    ) -> PromotionResult:
-        """Promote a plan.
+        if not stage.snapshot_to_intervals:
+            self.console.log_success("SKIP: No model batches to execute")
+            return
 
-        Args:
-            plan: The plan to promote.
-            no_gaps_snapshot_names: The names of snapshots to check for gaps if the no gaps check is enabled in the plan.
-            If not provided, all snapshots are checked.
-        """
-        promotion_result = self.state_sync.promote(
+        scheduler = self.create_scheduler(stage.all_snapshots.values())
+        errors, _ = scheduler.run_merged_intervals(
+            merged_intervals=stage.snapshot_to_intervals,
+            deployability_index=stage.deployability_index,
+            environment_naming_info=plan.environment.naming_info,
+            execution_time=plan.execution_time,
+            circuit_breaker=self._circuit_breaker,
+            start=plan.start,
+            end=plan.end,
+        )
+        if errors:
+            raise PlanError("Plan application failed.")
+
+    def visit_audit_only_run_stage(
+        self, stage: stages.AuditOnlyRunStage, plan: EvaluatablePlan
+    ) -> None:
+        audit_snapshots = stage.snapshots
+        if not audit_snapshots:
+            return
+
+        # If there are any snapshots to be audited, we'll reuse the scheduler's internals to audit them
+        scheduler = self.create_scheduler(audit_snapshots)
+        completion_status = scheduler.audit(
             plan.environment,
-            no_gaps_snapshot_names=no_gaps_snapshot_names if plan.no_gaps else set(),
-            environment_statements=plan.environment_statements,
+            plan.start,
+            plan.end,
+            execution_time=plan.execution_time,
+            end_bounded=plan.end_bounded,
+            interval_end_per_model=plan.interval_end_per_model,
         )
 
-        if not plan.is_dev:
-            try:
-                self.snapshot_evaluator.migrate(
-                    [s for s in snapshots.values() if s.is_paused],
-                    snapshots,
-                    allow_destructive_snapshots=plan.allow_destructive_models,
-                    deployability_index=deployability_index,
-                )
-            except NodeExecutionFailedError as ex:
-                raise PlanError(str(ex.__cause__) if ex.__cause__ else str(ex))
+        if completion_status.is_failure:
+            raise PlanError("Plan application failed.")
 
-            if not plan.ensure_finalized_snapshots:
-                # Only unpause at this point if we don't have to use the finalized snapshots
-                # for subsequent plan applications. Otherwise, unpause right before finalizing
-                # the environment.
-                self.state_sync.unpause_snapshots(promotion_result.added, plan.end)
+    def visit_restatement_stage(
+        self, stage: stages.RestatementStage, plan: EvaluatablePlan
+    ) -> None:
+        snapshot_intervals_to_restate = {(s, i) for s, i in stage.snapshot_intervals.items()}
 
-        return promotion_result
+        # Restating intervals on prod plans should mean that the intervals are cleared across
+        # all environments, not just the version currently in prod
+        # This ensures that work done in dev environments can still be promoted to prod
+        # by forcing dev environments to re-run intervals that changed in prod
+        #
+        # Without this rule, its possible that promoting a dev table to prod will introduce old data to prod
+        snapshot_intervals_to_restate.update(
+            self._restatement_intervals_across_all_environments(
+                prod_restatements=plan.restatements,
+                disable_restatement_models=plan.disabled_restatement_models,
+                loaded_snapshots={s.snapshot_id: s for s in stage.all_snapshots.values()},
+            )
+        )
 
-    def _update_views(
-        self,
-        plan: EvaluatablePlan,
-        snapshots: t.Dict[SnapshotId, Snapshot],
-        promotion_result: PromotionResult,
-        deployability_index: t.Optional[DeployabilityIndex] = None,
+        self.state_sync.remove_intervals(
+            snapshot_intervals=list(snapshot_intervals_to_restate),
+            remove_shared_versions=plan.is_prod,
+        )
+
+    def visit_environment_record_update_stage(
+        self, stage: stages.EnvironmentRecordUpdateStage, plan: EvaluatablePlan
     ) -> None:
-        """Update environment views.
+        self.state_sync.promote(
+            plan.environment,
+            no_gaps_snapshot_names=stage.no_gaps_snapshot_names if plan.no_gaps else set(),
+            environment_statements=plan.environment_statements,
+        )
 
-        Args:
-            plan: The plan to promote.
-            promotion_result: The result of the promotion.
-            deployability_index: Indicates which snapshots are deployable in the context of this promotion.
-        """
-        if not plan.is_dev and plan.ensure_finalized_snapshots:
-            # Unpause right before finalizing the environment in case when
-            # we need to use the finalized snapshots for subsequent plan applications.
-            # Otherwise, unpause right after updatig the environment record.
-            self.state_sync.unpause_snapshots(promotion_result.added, plan.end)
+    def visit_migrate_schemas_stage(
+        self, stage: stages.MigrateSchemasStage, plan: EvaluatablePlan
+    ) -> None:
+        try:
+            self.snapshot_evaluator.migrate(
+                stage.snapshots,
+                stage.all_snapshots,
+                allow_destructive_snapshots=plan.allow_destructive_models,
+                deployability_index=stage.deployability_index,
+            )
+        except NodeExecutionFailedError as ex:
+            raise PlanError(str(ex.__cause__) if ex.__cause__ else str(ex))
+
+    def visit_unpause_stage(self, stage: stages.UnpauseStage, plan: EvaluatablePlan) -> None:
+        self.state_sync.unpause_snapshots(stage.promoted_snapshots, plan.end)
 
+    def visit_virtual_layer_update_stage(
+        self, stage: stages.VirtualLayerUpdateStage, plan: EvaluatablePlan
+    ) -> None:
         environment = plan.environment
 
         self.console.start_promotion_progress(
-            promotion_result.added + promotion_result.removed,
+            list(stage.promoted_snapshots) + list(stage.demoted_snapshots),
             environment.naming_info,
             self.default_catalog,
         )
@@ -389,25 +322,28 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         try:
             self._promote_snapshots(
                 plan,
-                [snapshots[s.snapshot_id] for s in promotion_result.added],
+                [stage.all_snapshots[s.snapshot_id] for s in stage.promoted_snapshots],
                 environment.naming_info,
-                deployability_index=deployability_index,
+                deployability_index=stage.deployability_index,
                 on_complete=lambda s: self.console.update_promotion_progress(s, True),
-                snapshots=snapshots,
+                snapshots=stage.all_snapshots,
             )
-            if promotion_result.removed_environment_naming_info:
+            if stage.demoted_environment_naming_info:
                 self._demote_snapshots(
-                    plan,
-                    promotion_result.removed,
-                    promotion_result.removed_environment_naming_info,
+                    stage.demoted_snapshots,
+                    stage.demoted_environment_naming_info,
                     on_complete=lambda s: self.console.update_promotion_progress(s, False),
                 )
 
-            self.state_sync.finalize(environment)
             completed = True
         finally:
             self.console.stop_promotion_progress(success=completed)
 
+    def visit_finalize_environment_stage(
+        self, stage: stages.FinalizeEnvironmentStage, plan: EvaluatablePlan
+    ) -> None:
+        self.state_sync.finalize(plan.environment)
+
     def _promote_snapshots(
         self,
         plan: EvaluatablePlan,
@@ -436,7 +372,6 @@ class BuiltInPlanEvaluator(PlanEvaluator):
 
     def _demote_snapshots(
         self,
-        plan: EvaluatablePlan,
         target_snapshots: t.Iterable[SnapshotTableInfo],
         environment_naming_info: EnvironmentNamingInfo,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
@@ -445,34 +380,6 @@ class BuiltInPlanEvaluator(PlanEvaluator):
             target_snapshots, environment_naming_info, on_complete=on_complete
         )
 
-    def _restate(self, plan: EvaluatablePlan, snapshots_by_name: t.Dict[str, Snapshot]) -> None:
-        if not plan.restatements or plan.is_dev:
-            return
-
-        snapshot_intervals_to_restate = {
-            (snapshots_by_name[name].table_info, intervals)
-            for name, intervals in plan.restatements.items()
-        }
-
-        # Restating intervals on prod plans should mean that the intervals are cleared across
-        # all environments, not just the version currently in prod
-        # This ensures that work done in dev environments can still be promoted to prod
-        # by forcing dev environments to re-run intervals that changed in prod
-        #
-        # Without this rule, its possible that promoting a dev table to prod will introduce old data to prod
-        snapshot_intervals_to_restate.update(
-            self._restatement_intervals_across_all_environments(
-                prod_restatements=plan.restatements,
-                disable_restatement_models=plan.disabled_restatement_models,
-                loaded_snapshots={s.snapshot_id: s for s in snapshots_by_name.values()},
-            )
-        )
-
-        self.state_sync.remove_intervals(
-            snapshot_intervals=list(snapshot_intervals_to_restate),
-            remove_shared_versions=plan.is_prod,
-        )
-
     def _restatement_intervals_across_all_environments(
         self,
         prod_restatements: t.Dict[str, Interval],
@@ -552,95 +459,19 @@ class BuiltInPlanEvaluator(PlanEvaluator):
 
         return set(snapshots_to_restate.values())
 
-    def _run_audits_for_metadata_snapshots(
-        self,
-        plan: EvaluatablePlan,
-        new_snapshots: t.Dict[SnapshotId, Snapshot],
-    ) -> None:
-        audit_snapshots = get_audit_only_snapshots(new_snapshots, self.state_sync)
-        if not audit_snapshots:
-            return
-
-        # If there are any snapshots to be audited, we'll reuse the scheduler's internals to audit them
-        scheduler = self.create_scheduler(audit_snapshots.values())
-        completion_status = scheduler.audit(
-            plan.environment,
-            plan.start,
-            plan.end,
-            execution_time=plan.execution_time,
-            end_bounded=plan.end_bounded,
-            interval_end_per_model=plan.interval_end_per_model,
-        )
-
-        if completion_status.is_failure:
-            raise PlanError("Plan application failed.")
-
-
-def update_intervals_for_new_snapshots(
-    snapshots: t.Collection[Snapshot], state_sync: StateSync
-) -> None:
-    snapshots_intervals: t.List[SnapshotIntervals] = []
-    for snapshot in state_sync.refresh_snapshot_intervals(snapshots):
-        if snapshot.is_forward_only:
-            snapshot.dev_intervals = snapshot.intervals.copy()
-            snapshots_intervals.append(
-                SnapshotIntervals(
-                    name=snapshot.name,
-                    identifier=snapshot.identifier,
-                    version=snapshot.version,
-                    dev_version=snapshot.dev_version,
-                    dev_intervals=snapshot.dev_intervals,
+    def _update_intervals_for_new_snapshots(self, snapshots: t.Collection[Snapshot]) -> None:
+        snapshots_intervals: t.List[SnapshotIntervals] = []
+        for snapshot in snapshots:
+            if snapshot.is_forward_only:
+                snapshots_intervals.append(
+                    SnapshotIntervals(
+                        name=snapshot.name,
+                        identifier=snapshot.identifier,
+                        version=snapshot.version,
+                        dev_version=snapshot.dev_version,
+                        dev_intervals=snapshot.dev_intervals,
+                    )
                 )
-            )
-
-    if snapshots_intervals:
-        state_sync.add_snapshots_intervals(snapshots_intervals)
-
-
-def get_audit_only_snapshots(
-    new_snapshots: t.Dict[SnapshotId, Snapshot], state_reader: StateReader
-) -> t.Dict[SnapshotId, Snapshot]:
-    metadata_snapshots = []
-    for snapshot in new_snapshots.values():
-        if not snapshot.is_metadata or not snapshot.is_model or not snapshot.evaluatable:
-            continue
-
-        metadata_snapshots.append(snapshot)
-
-    # Bulk load all the previous snapshots
-    previous_snapshots = state_reader.get_snapshots(
-        [s.previous_version.snapshot_id(s.name) for s in metadata_snapshots if s.previous_version]
-    ).values()
-
-    # Check if any of the snapshots have modifications to the audits field by comparing the hashes
-    audit_snapshots = {}
-    for snapshot, previous_snapshot in zip(metadata_snapshots, previous_snapshots):
-        new_audits_hash = snapshot.model.audit_metadata_hash()
-        previous_audit_hash = previous_snapshot.model.audit_metadata_hash()
-
-        if snapshot.model.audits and previous_audit_hash != new_audits_hash:
-            audit_snapshots[snapshot.snapshot_id] = snapshot
-
-    return audit_snapshots
-
-
-def get_snapshots_to_create(
-    plan: EvaluatablePlan, snapshots: t.Dict[SnapshotId, Snapshot]
-) -> t.List[Snapshot]:
-    promoted_snapshot_ids = (
-        set(plan.environment.promoted_snapshot_ids)
-        if plan.environment.promoted_snapshot_ids is not None
-        else None
-    )
-
-    def _should_create(s: Snapshot) -> bool:
-        if not s.is_model or s.is_symbolic:
-            return False
-        # Only create tables for snapshots that we're planning to promote or that were selected for backfill
-        return (
-            plan.is_selected_for_backfill(s.name)
-            or promoted_snapshot_ids is None
-            or s.snapshot_id in promoted_snapshot_ids
-        )
 
-    return [s for s in snapshots.values() if _should_create(s)]
+        if snapshots_intervals:
+            self.state_sync.add_snapshots_intervals(snapshots_intervals)
diff --git a/sqlmesh/core/plan/explainer.py b/sqlmesh/core/plan/explainer.py
index 32d80b5d..aa1d5775 100644
--- a/sqlmesh/core/plan/explainer.py
+++ b/sqlmesh/core/plan/explainer.py
@@ -17,7 +17,7 @@ from sqlmesh.core.state_sync import StateReader
 from sqlmesh.core.snapshot.definition import (
     SnapshotInfoMixin,
 )
-from sqlmesh.utils import Verbosity, rich as srich
+from sqlmesh.utils import Verbosity, rich as srich, to_snake_case
 from sqlmesh.utils.date import to_ts
 from sqlmesh.utils.errors import SQLMeshError
 
@@ -73,7 +73,7 @@ class RichExplainerConsole(ExplainerConsole):
     def explain(self, stages: t.List[stages.PlanStage]) -> None:
         tree = Tree("[bold]Explained plan[/bold]")
         for stage in stages:
-            handler_name = f"visit_{_to_snake_case(stage.__class__.__name__)}"
+            handler_name = f"visit_{to_snake_case(stage.__class__.__name__)}"
             if not hasattr(self, handler_name):
                 logger.error("Unexpected stage: %s", stage.__class__.__name__)
                 continue
@@ -97,6 +97,9 @@ class RichExplainerConsole(ExplainerConsole):
             "[bold]Validate SQL and create physical layer tables and views if they do not exist[/bold]"
         )
         for snapshot in stage.snapshots:
+            if snapshot.snapshot_id not in stage.snapshots_with_missing_intervals:
+                continue
+
             is_deployable = (
                 stage.deployability_index.is_deployable(snapshot)
                 if self.environment_naming_info.name != c.PROD
@@ -114,7 +117,9 @@ class RichExplainerConsole(ExplainerConsole):
 
             if snapshot.is_view:
                 create_tree = Tree("Create view if it doesn't exist")
-            elif snapshot.is_forward_only and snapshot.previous_versions:
+            elif (
+                snapshot.is_forward_only and snapshot.previous_versions and not snapshot.is_managed
+            ):
                 prod_table = snapshot.table_name(True)
                 create_tree = Tree(
                     f"Clone {prod_table} into {table_name} and then update its schema if it doesn't exist"
@@ -224,7 +229,7 @@ class RichExplainerConsole(ExplainerConsole):
             "[bold]Delete views in the virtual layer for models that were removed[/bold]"
         )
         for snapshot in stage.demoted_snapshots:
-            display_name = self._display_name(snapshot)
+            display_name = self._display_name(snapshot, stage.demoted_environment_naming_info)
             demote_tree.add(display_name)
 
         if stage.promoted_snapshots:
@@ -233,14 +238,31 @@ class RichExplainerConsole(ExplainerConsole):
             tree.add(self._limit_tree(demote_tree))
         return tree
 
+    def visit_create_snapshot_records_stage(
+        self, stage: stages.CreateSnapshotRecordsStage
+    ) -> t.Optional[Tree]:
+        return None
+
     def visit_environment_record_update_stage(
         self, stage: stages.EnvironmentRecordUpdateStage
     ) -> t.Optional[Tree]:
         return None
 
-    def _display_name(self, snapshot: SnapshotInfoMixin) -> str:
+    def visit_unpause_stage(self, stage: stages.UnpauseStage) -> t.Optional[Tree]:
+        return None
+
+    def visit_finalize_environment_stage(
+        self, stage: stages.FinalizeEnvironmentStage
+    ) -> t.Optional[Tree]:
+        return None
+
+    def _display_name(
+        self,
+        snapshot: SnapshotInfoMixin,
+        environment_naming_info: t.Optional[EnvironmentNamingInfo] = None,
+    ) -> str:
         return snapshot.display_name(
-            self.environment_naming_info,
+            environment_naming_info or self.environment_naming_info,
             self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
             dialect=self.dialect,
         )
@@ -273,9 +295,3 @@ def _get_explainer_console(
         verbosity=console.verbosity,
         console=console.console,
     )
-
-
-def _to_snake_case(name: str) -> str:
-    return "".join(
-        f"_{c.lower()}" if c.isupper() and idx != 0 else c.lower() for idx, c in enumerate(name)
-    )
diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py
index 964dadb1..d8c7cba8 100644
--- a/sqlmesh/core/plan/stages.py
+++ b/sqlmesh/core/plan/stages.py
@@ -1,11 +1,8 @@
 import typing as t
 
 from dataclasses import dataclass
+from sqlmesh.core.environment import EnvironmentStatements, EnvironmentNamingInfo
 from sqlmesh.core.plan.definition import EvaluatablePlan
-from sqlmesh.core.plan.evaluator import (
-    get_audit_only_snapshots,
-    get_snapshots_to_create,
-)
 from sqlmesh.core.state_sync import StateReader
 from sqlmesh.core.scheduler import merged_missing_intervals, SnapshotToIntervals
 from sqlmesh.core.snapshot.definition import (
@@ -19,68 +16,193 @@ from sqlmesh.core.snapshot.definition import (
 
 @dataclass
 class BeforeAllStage:
-    statements: t.List[str]
+    """Run environment statements before every other stage.
+
+    Args:
+        statements: Environment statements to run before every other stage.
+        all_snapshots: All snapshots in the plan by name.
+    """
+
+    statements: t.List[EnvironmentStatements]
+    all_snapshots: t.Dict[str, Snapshot]
 
 
 @dataclass
 class AfterAllStage:
-    statements: t.List[str]
+    """Run environment statements after all other stages.
+
+    Args:
+        statements: Environment statements to run after all other stages.
+        all_snapshots: All snapshots in the plan by name.
+    """
+
+    statements: t.List[EnvironmentStatements]
+    all_snapshots: t.Dict[str, Snapshot]
+
+
+@dataclass
+class CreateSnapshotRecordsStage:
+    """Create new snapshot reecords in the state.
+
+    Args:
+        snapshots: New snapshots to create records for.
+    """
+
+    snapshots: t.List[Snapshot]
 
 
 @dataclass
 class PhysicalLayerUpdateStage:
+    """Update the physical layer by creating physical tables and views for given snapshots.
+
+    Args:
+        snapshots: Snapshots to create physical tables and views for. This collection can be empty in which case
+            no physical layer update is needed. This can be useful to report the lack of physical layer updates
+            back to the user.
+        all_snapshots: All snapshots in the plan by snapshot ID.
+        snapshots_with_missing_intervals: Snapshots that have missing intervals.
+        deployability_index: Deployability index for this stage.
+    """
+
     snapshots: t.List[Snapshot]
+    all_snapshots: t.Dict[SnapshotId, Snapshot]
+    snapshots_with_missing_intervals: t.Set[SnapshotId]
     deployability_index: DeployabilityIndex
 
 
 @dataclass
 class AuditOnlyRunStage:
+    """Run audits only for given snapshots.
+
+    Args:
+        snapshots: Snapshots to run audits for.
+    """
+
     snapshots: t.List[Snapshot]
 
 
 @dataclass
 class RestatementStage:
+    """Restate intervals for given snapshots.
+
+    Args:
+        snapshot_intervals: Intervals to restate.
+        all_snapshots: All snapshots in the plan by name.
+    """
+
     snapshot_intervals: t.Dict[SnapshotTableInfo, Interval]
+    all_snapshots: t.Dict[str, Snapshot]
 
 
 @dataclass
 class BackfillStage:
+    """Backfill given missing intervals.
+
+    Args:
+        snapshot_to_intervals: Intervals to backfill. This collection can be empty in which case no backfill is needed.
+            This can be useful to report the lack of backfills back to the user.
+        all_snapshots: All snapshots in the plan by name.
+        deployability_index: Deployability index for this stage.
+        before_promote: Whether this stage is before the promotion stage.
+    """
+
     snapshot_to_intervals: SnapshotToIntervals
+    all_snapshots: t.Dict[str, Snapshot]
     deployability_index: DeployabilityIndex
     before_promote: bool = True
 
 
+@dataclass
+class EnvironmentRecordUpdateStage:
+    """Update the environment record in the state.
+
+    Args:
+        no_gaps_snapshot_names: Names of snapshots for which there should be no interval gaps.
+    """
+
+    no_gaps_snapshot_names: t.Set[str]
+
+
 @dataclass
 class MigrateSchemasStage:
+    """Migrate schemas of physical tables for given snapshots.
+
+    Args:
+        snapshots: Snapshots to migrate schemas for.
+        all_snapshots: All snapshots in the plan by snapshot ID.
+        deployability_index: Deployability index for this stage.
+    """
+
     snapshots: t.List[Snapshot]
+    all_snapshots: t.Dict[SnapshotId, Snapshot]
+    deployability_index: DeployabilityIndex
 
 
 @dataclass
 class VirtualLayerUpdateStage:
+    """Update the virtual layer by creating and deleting views for given snapshots.
+
+    Args:
+        promoted_snapshots: Snapshots to create views for.
+        demoted_snapshots: Snapshots to delete views for.
+        demoted_environment_naming_info: Environment naming info of the previous environment record.
+        all_snapshots: All snapshots in the plan by snapshot ID.
+        deployability_index: Deployability index for this stage.
+    """
+
     promoted_snapshots: t.Set[SnapshotTableInfo]
     demoted_snapshots: t.Set[SnapshotTableInfo]
+    demoted_environment_naming_info: t.Optional[EnvironmentNamingInfo]
+    all_snapshots: t.Dict[SnapshotId, Snapshot]
     deployability_index: DeployabilityIndex
 
 
 @dataclass
-class EnvironmentRecordUpdateStage:
+class UnpauseStage:
+    """Unpause given snapshots that are being deployed to prod.
+
+    Args:
+        promoted_snapshots: Snapshots to unpause.
+    """
+
+    promoted_snapshots: t.Set[SnapshotTableInfo]
+
+
+@dataclass
+class FinalizeEnvironmentStage:
+    """Finalize the enviornment record in the state.
+
+    Finalization means that all stages have been applied and that the environment has been transitioned
+    to the new state successfully. This should be the last stage in the plan application process.
+    """
+
     pass
 
 
 PlanStage = t.Union[
     BeforeAllStage,
     AfterAllStage,
+    CreateSnapshotRecordsStage,
     PhysicalLayerUpdateStage,
     AuditOnlyRunStage,
     RestatementStage,
     BackfillStage,
+    EnvironmentRecordUpdateStage,
     MigrateSchemasStage,
     VirtualLayerUpdateStage,
-    EnvironmentRecordUpdateStage,
+    UnpauseStage,
+    FinalizeEnvironmentStage,
 ]
 
 
 class PlanStagesBuilder:
+    """The builder for the plan stages.
+
+    Args:
+        state_reader: The state reader to use to read the snapshots and environment.
+        default_catalog: The default catalog to use for the snapshots.
+    """
+
     def __init__(
         self,
         state_reader: StateReader,
@@ -90,6 +212,16 @@ class PlanStagesBuilder:
         self.default_catalog = default_catalog
 
     def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
+        """Builds the plan stages for the given plan.
+
+        NOTE: Building the plan stages should NOT produce any side effects in the state or the data warehouse.
+
+        Args:
+            plan: The plan to build the stages for.
+
+        Returns:
+            A list of plan stages.
+        """
         new_snapshots = {s.snapshot_id: s for s in plan.new_snapshots}
         stored_snapshots = self.state_reader.get_snapshots(plan.environment.snapshots)
         snapshots = {**new_snapshots, **stored_snapshots}
@@ -99,6 +231,8 @@ class PlanStagesBuilder:
             s.snapshot_id for s in snapshots.values() if plan.is_selected_for_backfill(s.name)
         }
 
+        self._adjust_intervals(snapshots_by_name, plan)
+
         deployability_index = DeployabilityIndex.create(snapshots, start=plan.start)
         deployability_index_for_creation = deployability_index
         if plan.is_dev:
@@ -119,8 +253,9 @@ class PlanStagesBuilder:
                 s
                 for s in snapshots.values()
                 if s.is_paused
-                and s.is_materialized
-                and not deployability_index_for_creation.is_representative(s)
+                and s.is_model
+                and not s.is_symbolic
+                and (not deployability_index_for_creation.is_representative(s) or s.is_view)
             ]
 
         snapshots_to_intervals = self._missing_intervals(
@@ -138,19 +273,26 @@ class PlanStagesBuilder:
                 elif snapshot.snapshot_id in after_promote_snapshots:
                     missing_intervals_after_promote[snapshot] = intervals
 
+        promoted_snapshots, demoted_snapshots, demoted_environment_naming_info = (
+            self._get_promoted_demoted_snapshots(plan)
+        )
+
         stages: t.List[PlanStage] = []
 
-        before_all_stage = self._get_before_all_stage(plan)
+        before_all_stage = self._get_before_all_stage(plan, snapshots_by_name)
         if before_all_stage:
             stages.append(before_all_stage)
 
+        if plan.new_snapshots:
+            stages.append(CreateSnapshotRecordsStage(snapshots=plan.new_snapshots))
+
         stages.append(
             self._get_physical_layer_update_stage(
                 plan, snapshots, snapshots_to_intervals, deployability_index_for_creation
             )
         )
 
-        audit_only_snapshots = get_audit_only_snapshots(new_snapshots, self.state_reader)
+        audit_only_snapshots = self._get_audit_only_snapshots(new_snapshots)
         if audit_only_snapshots:
             stages.append(AuditOnlyRunStage(snapshots=list(audit_only_snapshots.values())))
 
@@ -162,53 +304,101 @@ class PlanStagesBuilder:
             stages.append(
                 BackfillStage(
                     snapshot_to_intervals=missing_intervals_before_promote,
+                    all_snapshots=snapshots_by_name,
                     deployability_index=deployability_index,
                 )
             )
         elif not needs_backfill:
             # Append an empty backfill stage so that explainer can show that the stage is skipped
             stages.append(
-                BackfillStage(snapshot_to_intervals={}, deployability_index=deployability_index)
+                BackfillStage(
+                    snapshot_to_intervals={},
+                    all_snapshots=snapshots_by_name,
+                    deployability_index=deployability_index,
+                )
             )
 
-        stages.append(EnvironmentRecordUpdateStage())
+        stages.append(
+            EnvironmentRecordUpdateStage(
+                no_gaps_snapshot_names={s.name for s in before_promote_snapshots}
+            )
+        )
 
         if snapshots_with_schema_migration:
-            stages.append(MigrateSchemasStage(snapshots=snapshots_with_schema_migration))
+            stages.append(
+                MigrateSchemasStage(
+                    snapshots=snapshots_with_schema_migration,
+                    all_snapshots=snapshots,
+                    deployability_index=deployability_index_for_creation,
+                )
+            )
+
+        if not plan.is_dev and not plan.ensure_finalized_snapshots and promoted_snapshots:
+            # Only unpause at this point if we don't have to use the finalized snapshots
+            # for subsequent plan applications. Otherwise, unpause right before updating
+            # the virtual layer.
+            stages.append(UnpauseStage(promoted_snapshots=promoted_snapshots))
 
         if missing_intervals_after_promote:
             stages.append(
                 BackfillStage(
                     snapshot_to_intervals=missing_intervals_after_promote,
+                    all_snapshots=snapshots_by_name,
                     deployability_index=deployability_index,
                 )
             )
 
-        virtual_layer_update_stage = self._get_virtual_layer_update_stage(plan, deployability_index)
+        if not plan.is_dev and plan.ensure_finalized_snapshots and promoted_snapshots:
+            # Unpause right before updating the virtual layer and finalizing the environment in case when
+            # we need to use the finalized snapshots for subsequent plan applications.
+            # Otherwise, unpause right after updatig the environment record.
+            stages.append(UnpauseStage(promoted_snapshots=promoted_snapshots))
+
+        virtual_layer_update_stage = self._get_virtual_layer_update_stage(
+            promoted_snapshots,
+            demoted_snapshots,
+            demoted_environment_naming_info,
+            snapshots,
+            deployability_index,
+        )
         if virtual_layer_update_stage:
             stages.append(virtual_layer_update_stage)
 
-        after_all_stage = self._get_after_all_stage(plan)
+        stages.append(FinalizeEnvironmentStage())
+
+        after_all_stage = self._get_after_all_stage(plan, snapshots_by_name)
         if after_all_stage:
             stages.append(after_all_stage)
 
         return stages
 
-    def _get_before_all_stage(self, plan: EvaluatablePlan) -> t.Optional[BeforeAllStage]:
+    def _get_before_all_stage(
+        self, plan: EvaluatablePlan, snapshots_by_name: t.Dict[str, Snapshot]
+    ) -> t.Optional[BeforeAllStage]:
         before_all = [
-            statement
+            environment_statements
             for environment_statements in plan.environment_statements or []
-            for statement in environment_statements.before_all
+            if environment_statements.before_all
         ]
-        return BeforeAllStage(statements=before_all) if before_all else None
+        return (
+            BeforeAllStage(statements=before_all, all_snapshots=snapshots_by_name)
+            if before_all
+            else None
+        )
 
-    def _get_after_all_stage(self, plan: EvaluatablePlan) -> t.Optional[AfterAllStage]:
+    def _get_after_all_stage(
+        self, plan: EvaluatablePlan, snapshots_by_name: t.Dict[str, Snapshot]
+    ) -> t.Optional[AfterAllStage]:
         after_all = [
-            statement
+            environment_statements
             for environment_statements in plan.environment_statements or []
-            for statement in environment_statements.after_all
+            if environment_statements.after_all
         ]
-        return AfterAllStage(statements=after_all) if after_all else None
+        return (
+            AfterAllStage(statements=after_all, all_snapshots=snapshots_by_name)
+            if after_all
+            else None
+        )
 
     def _get_restatement_stage(
         self, plan: EvaluatablePlan, snapshots_by_name: t.Dict[str, Snapshot]
@@ -220,7 +410,9 @@ class PlanStagesBuilder:
             snapshot_intervals_to_restate[restated_snapshot.table_info] = interval
         if not snapshot_intervals_to_restate or plan.is_dev:
             return None
-        return RestatementStage(snapshot_intervals=snapshot_intervals_to_restate)
+        return RestatementStage(
+            snapshot_intervals=snapshot_intervals_to_restate, all_snapshots=snapshots_by_name
+        )
 
     def _get_physical_layer_update_stage(
         self,
@@ -229,50 +421,80 @@ class PlanStagesBuilder:
         snapshots_to_intervals: SnapshotToIntervals,
         deployability_index: DeployabilityIndex,
     ) -> PhysicalLayerUpdateStage:
-        snapshots_to_create = [
-            s
-            for s in get_snapshots_to_create(plan, snapshots)
-            if s in snapshots_to_intervals and s.is_model and not s.is_symbolic
-        ]
         return PhysicalLayerUpdateStage(
-            snapshots=snapshots_to_create,
+            snapshots=self._get_snapshots_to_create(plan, snapshots),
+            all_snapshots=snapshots,
+            snapshots_with_missing_intervals={s.snapshot_id for s in snapshots_to_intervals},
             deployability_index=deployability_index,
         )
 
     def _get_virtual_layer_update_stage(
-        self, plan: EvaluatablePlan, deployability_index: DeployabilityIndex
+        self,
+        promoted_snapshots: t.Set[SnapshotTableInfo],
+        demoted_snapshots: t.Set[SnapshotTableInfo],
+        demoted_environment_naming_info: t.Optional[EnvironmentNamingInfo],
+        all_snapshots: t.Dict[SnapshotId, Snapshot],
+        deployability_index: DeployabilityIndex,
     ) -> t.Optional[VirtualLayerUpdateStage]:
-        promoted_snapshots, demoted_snapshots = self._get_promoted_demoted_snapshots(plan)
+        promoted_snapshots = {s for s in promoted_snapshots if s.is_model and not s.is_symbolic}
+        demoted_snapshots = {s for s in demoted_snapshots if s.is_model and not s.is_symbolic}
         if not promoted_snapshots and not demoted_snapshots:
             return None
         return VirtualLayerUpdateStage(
             promoted_snapshots=promoted_snapshots,
             demoted_snapshots=demoted_snapshots,
+            demoted_environment_naming_info=demoted_environment_naming_info,
+            all_snapshots=all_snapshots,
             deployability_index=deployability_index,
         )
 
     def _get_promoted_demoted_snapshots(
         self, plan: EvaluatablePlan
-    ) -> t.Tuple[t.Set[SnapshotTableInfo], t.Set[SnapshotTableInfo]]:
+    ) -> t.Tuple[
+        t.Set[SnapshotTableInfo], t.Set[SnapshotTableInfo], t.Optional[EnvironmentNamingInfo]
+    ]:
         existing_environment = self.state_reader.get_environment(plan.environment.name)
         if existing_environment:
-            snapshots_by_name = {s.name: s for s in existing_environment.snapshots}
-            demoted_snapshot_names = {s.name for s in existing_environment.promoted_snapshots} - {
+            new_table_infos = {
+                table_info.name: table_info for table_info in plan.environment.promoted_snapshots
+            }
+            existing_table_infos = {
+                table_info.name: table_info
+                for table_info in existing_environment.promoted_snapshots
+            }
+            views_that_changed_location = {
+                existing_table_info
+                for existing_table_info in existing_environment.promoted_snapshots
+                if existing_table_info.name in new_table_infos
+                and existing_table_info.qualified_view_name.for_environment(
+                    existing_environment.naming_info
+                )
+                != new_table_infos[existing_table_info.name].qualified_view_name.for_environment(
+                    plan.environment.naming_info
+                )
+            }
+            missing_model_names = set(existing_table_infos) - {
                 s.name for s in plan.environment.promoted_snapshots
             }
-            demoted_snapshots = {snapshots_by_name[name] for name in demoted_snapshot_names}
+            demoted_snapshots = {
+                existing_table_infos[name] for name in missing_model_names
+            } | views_that_changed_location
         else:
             demoted_snapshots = set()
+
         promoted_snapshots = set(plan.environment.promoted_snapshots)
         if existing_environment and plan.environment.can_partially_promote(existing_environment):
             promoted_snapshots -= set(existing_environment.promoted_snapshots)
 
-        def _snapshot_filter(snapshot: SnapshotTableInfo) -> bool:
-            return snapshot.is_model and not snapshot.is_symbolic
+        demoted_environment_naming_info = (
+            existing_environment.naming_info if demoted_snapshots and existing_environment else None
+        )
 
-        return {s for s in promoted_snapshots if _snapshot_filter(s)}, {
-            s for s in demoted_snapshots if _snapshot_filter(s)
-        }
+        return (
+            promoted_snapshots,
+            demoted_snapshots,
+            demoted_environment_naming_info,
+        )
 
     def _missing_intervals(
         self,
@@ -294,6 +516,69 @@ class PlanStagesBuilder:
             interval_end_per_model=plan.interval_end_per_model,
         )
 
+    def _get_audit_only_snapshots(
+        self, new_snapshots: t.Dict[SnapshotId, Snapshot]
+    ) -> t.Dict[SnapshotId, Snapshot]:
+        metadata_snapshots = []
+        for snapshot in new_snapshots.values():
+            if not snapshot.is_metadata or not snapshot.is_model or not snapshot.evaluatable:
+                continue
+
+            metadata_snapshots.append(snapshot)
+
+        # Bulk load all the previous snapshots
+        previous_snapshots = self.state_reader.get_snapshots(
+            [
+                s.previous_version.snapshot_id(s.name)
+                for s in metadata_snapshots
+                if s.previous_version
+            ]
+        ).values()
+
+        # Check if any of the snapshots have modifications to the audits field by comparing the hashes
+        audit_snapshots = {}
+        for snapshot, previous_snapshot in zip(metadata_snapshots, previous_snapshots):
+            new_audits_hash = snapshot.model.audit_metadata_hash()
+            previous_audit_hash = previous_snapshot.model.audit_metadata_hash()
+
+            if snapshot.model.audits and previous_audit_hash != new_audits_hash:
+                audit_snapshots[snapshot.snapshot_id] = snapshot
+
+        return audit_snapshots
+
+    def _get_snapshots_to_create(
+        self, plan: EvaluatablePlan, snapshots: t.Dict[SnapshotId, Snapshot]
+    ) -> t.List[Snapshot]:
+        promoted_snapshot_ids = (
+            set(plan.environment.promoted_snapshot_ids)
+            if plan.environment.promoted_snapshot_ids is not None
+            else None
+        )
+
+        def _should_create(s: Snapshot) -> bool:
+            if not s.is_model or s.is_symbolic:
+                return False
+            # Only create tables for snapshots that we're planning to promote or that were selected for backfill
+            return (
+                plan.is_selected_for_backfill(s.name)
+                or promoted_snapshot_ids is None
+                or s.snapshot_id in promoted_snapshot_ids
+            )
+
+        return [s for s in snapshots.values() if _should_create(s)]
+
+    def _adjust_intervals(
+        self, snapshots_by_name: t.Dict[str, Snapshot], plan: EvaluatablePlan
+    ) -> None:
+        # Make sure the intervals are up to date and restatements are reflected
+        self.state_reader.refresh_snapshot_intervals(snapshots_by_name.values())
+        for new_snapshot in plan.new_snapshots:
+            if new_snapshot.is_forward_only:
+                # Forward-only snapshots inherit intervals in dev because of cloning
+                new_snapshot.dev_intervals = new_snapshot.intervals.copy()
+        for s_name, interval in plan.restatements.items():
+            snapshots_by_name[s_name].remove_interval(interval)
+
 
 def build_plan_stages(
     plan: EvaluatablePlan,
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index a66f7c19..d0a121a4 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -441,12 +441,27 @@ class Scheduler:
 
         try:
             with self.snapshot_evaluator.concurrent_context():
-                return concurrent_apply_to_dag(
+                errors, skipped_intervals = concurrent_apply_to_dag(
                     dag,
                     evaluate_node,
                     self.max_workers,
                     raise_on_error=False,
                 )
+                self.console.stop_evaluation_progress(success=not errors)
+
+                skipped_snapshots = {i[0] for i in skipped_intervals}
+                self.console.log_skipped_models(skipped_snapshots)
+                for skipped in skipped_snapshots:
+                    logger.info(f"SKIPPED snapshot {skipped}\n")
+
+                for error in errors:
+                    if isinstance(error.__cause__, CircuitBreakerError):
+                        raise error.__cause__
+                    logger.info(str(error), exc_info=error)
+
+                self.console.log_failed_models(errors)
+
+                return errors, skipped_intervals
         finally:
             if run_environment_statements:
                 execute_environment_statements(
@@ -604,7 +619,7 @@ class Scheduler:
         if not merged_intervals:
             return CompletionStatus.NOTHING_TO_DO
 
-        errors, skipped_intervals = self.run_merged_intervals(
+        errors, _ = self.run_merged_intervals(
             merged_intervals=merged_intervals,
             deployability_index=deployability_index,
             environment_naming_info=environment_naming_info,
@@ -616,20 +631,6 @@ class Scheduler:
             audit_only=audit_only,
         )
 
-        self.console.stop_evaluation_progress(success=not errors)
-
-        skipped_snapshots = {i[0] for i in skipped_intervals}
-        self.console.log_skipped_models(skipped_snapshots)
-        for skipped in skipped_snapshots:
-            logger.info(f"SKIPPED snapshot {skipped}\n")
-
-        for error in errors:
-            if isinstance(error.__cause__, CircuitBreakerError):
-                raise error.__cause__
-            logger.info(str(error), exc_info=error)
-
-        self.console.log_failed_models(errors)
-
         return CompletionStatus.FAILURE if errors else CompletionStatus.SUCCESS
 
     def _audit_snapshot(
diff --git a/sqlmesh/core/state_sync/base.py b/sqlmesh/core/state_sync/base.py
index 05d46430..4f46ccf9 100644
--- a/sqlmesh/core/state_sync/base.py
+++ b/sqlmesh/core/state_sync/base.py
@@ -108,6 +108,17 @@ class StateReader(abc.ABC):
             A set of all the existing snapshot ids.
         """
 
+    @abc.abstractmethod
+    def refresh_snapshot_intervals(self, snapshots: t.Collection[Snapshot]) -> t.List[Snapshot]:
+        """Updates given snapshots with latest intervals from the state.
+
+        Args:
+            snapshots: The snapshots to refresh.
+
+        Returns:
+            The updated snapshots.
+        """
+
     @abc.abstractmethod
     def nodes_exist(self, names: t.Iterable[str], exclude_external: bool = False) -> t.Set[str]:
         """Returns the node names that exist in the state sync.
@@ -373,17 +384,6 @@ class StateSync(StateReader, abc.ABC):
             remove_shared_versions: Whether to remove intervals for snapshots that share the same version with the target snapshots.
         """
 
-    @abc.abstractmethod
-    def refresh_snapshot_intervals(self, snapshots: t.Collection[Snapshot]) -> t.List[Snapshot]:
-        """Updates given snapshots with latest intervals from the state.
-
-        Args:
-            snapshots: The snapshots to refresh.
-
-        Returns:
-            The updated snapshots.
-        """
-
     @abc.abstractmethod
     def promote(
         self,
diff --git a/sqlmesh/lsp/main.py b/sqlmesh/lsp/main.py
index f9bfe461..641b570c 100755
--- a/sqlmesh/lsp/main.py
+++ b/sqlmesh/lsp/main.py
@@ -294,32 +294,6 @@ class SQLMeshLanguageServer:
                     SQLMeshLanguageServer._diagnostics_to_lsp_diagnostics(diagnostics),
                 )
 
-        @self.server.feature(types.TEXT_DOCUMENT_DID_CHANGE)
-        def did_change(ls: LanguageServer, params: types.DidChangeTextDocumentParams) -> None:
-            if self.lsp_context is None:
-                current_path = Path.cwd()
-                self._ensure_context_in_folder(current_path)
-                if self.lsp_context is None:
-                    ls.log_trace("No context found after did_change")
-                    return
-
-            uri = URI(params.text_document.uri)
-            context = self._context_get_or_load(uri)
-
-            models = context.map[uri.to_path()]
-            if models is None or not isinstance(models, ModelTarget):
-                return
-
-            # Get diagnostics from context (which handles caching)
-            diagnostics = context.lint_model(uri)
-
-            # Only publish diagnostics if client doesn't support pull diagnostics
-            if not self.client_supports_pull_diagnostics:
-                ls.publish_diagnostics(
-                    params.text_document.uri,
-                    SQLMeshLanguageServer._diagnostics_to_lsp_diagnostics(diagnostics),
-                )
-
         @self.server.feature(types.TEXT_DOCUMENT_DID_SAVE)
         def did_save(ls: LanguageServer, params: types.DidSaveTextDocumentParams) -> None:
             uri = URI(params.text_document.uri)
@@ -753,15 +727,19 @@ class SQLMeshLanguageServer:
 
         loaded = False
         # Ascend directories to look for config
-        while path.parents and not loaded:
+        current_dir = path.parent  # Start from the file's parent directory
+        while current_dir.parents and not loaded:
             for ext in ("py", "yml", "yaml"):
-                config_path = path / f"config.{ext}"
+                config_path = current_dir / f"config.{ext}"
                 if config_path.exists():
-                    if self._create_lsp_context([path]):
+                    if self._create_lsp_context([current_dir]):
                         loaded = True
                         # Re-check context for the document now that it's loaded
                         return self._ensure_context_for_document(document_uri)
-            path = path.parent
+            # Check if we've reached the filesystem root to prevent infinite loops
+            if current_dir == current_dir.parent:
+                break
+            current_dir = current_dir.parent
 
         # If still no context found, try the workspace folders
         if not loaded:
diff --git a/sqlmesh/utils/__init__.py b/sqlmesh/utils/__init__.py
index 56e3cdf3..f102f232 100644
--- a/sqlmesh/utils/__init__.py
+++ b/sqlmesh/utils/__init__.py
@@ -376,3 +376,9 @@ class CompletionStatus(Enum):
     @property
     def is_nothing_to_do(self) -> bool:
         return self == CompletionStatus.NOTHING_TO_DO
+
+
+def to_snake_case(name: str) -> str:
+    return "".join(
+        f"_{c.lower()}" if c.isupper() and idx != 0 else c.lower() for idx, c in enumerate(name)
+    )
diff --git a/tooling/vscode/extensions.json b/tooling/vscode/extensions.json
index 02715704..b9df8890 100644
--- a/tooling/vscode/extensions.json
+++ b/tooling/vscode/extensions.json
@@ -5,6 +5,8 @@
     "dbaeumer.vscode-eslint",
     "amodio.tsl-problem-matcher",
     "ms-vscode.extension-test-runner",
-    "ms-playwright.playwright"
+    "ms-playwright.playwright",
+    "esbenp.prettier-vscode",
+    "charliermarsh.ruff"
   ]
 }
diff --git a/vscode/extension/src/utilities/errors.ts b/vscode/extension/src/utilities/errors.ts
index 7a0796a4..62d0d016 100644
--- a/vscode/extension/src/utilities/errors.ts
+++ b/vscode/extension/src/utilities/errors.ts
@@ -9,6 +9,7 @@ import { traceInfo } from './common/log'
 export type ErrorType =
   | ErrorTypeGeneric
   | { type: 'not_signed_in' }
+  | { type: 'sqlmesh_not_found' }
   | { type: 'sqlmesh_lsp_not_found' }
   // tcloud_bin_not_found is used when the tcloud executable is not found. This is likely to happen if the user
   // opens a project that has a `tcloud.yaml` file but doesn't have tcloud installed.
@@ -85,6 +86,8 @@ export async function handleError(
       return
     case 'not_signed_in':
       return handleNotSignedInError(authProvider)
+    case 'sqlmesh_not_found':
+      return handleSqlmeshNotFoundError()
     case 'sqlmesh_lsp_not_found':
       return handleSqlmeshLspNotFoundError()
     case 'sqlmesh_lsp_dependencies_missing':
@@ -118,6 +121,14 @@ const handleNotSignedInError = async (
   }
 }
 
+/**
+ * Handles the case where the sqlmesh executable is not found.
+ */
+const handleSqlmeshNotFoundError = async (): Promise<void> => {
+  traceInfo('handleSqlmeshNotFoundError')
+  await window.showErrorMessage('SQLMesh not found, please check installation')
+}
+
 /**
  * Handles the case where the sqlmesh_lsp is not found.
  */
diff --git a/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts b/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts
index 95cc94d3..f656937b 100644
--- a/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts
+++ b/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts
@@ -147,51 +147,69 @@ export const installSqlmeshEnterprise = async (
   return ok(true)
 }
 
+let installationLock: Promise<Result<boolean, ErrorType>> | undefined = undefined
+
 /**
  * Checks if sqlmesh enterprise is installed and updated. If not, it will install it.
  * This will also create a progress message in vscode in order to inform the user that sqlmesh enterprise is being installed.
+ * Uses a lock mechanism to prevent parallel executions.
  *
  * @returns A Result indicating whether sqlmesh enterprise was installed in the call.
  */
 export const ensureSqlmeshEnterpriseInstalled = async (): Promise<
   Result<boolean, ErrorType>
 > => {
-  traceInfo('Ensuring sqlmesh enterprise is installed')
-  const isInstalled = await isSqlmeshEnterpriseInstalled()
-  if (isErr(isInstalled)) {
-    return isInstalled
-  }
-  if (isInstalled.value) {
-    traceInfo('Sqlmesh enterprise is installed')
-    return ok(false)
+  // If there's an ongoing installation, wait for it to complete
+  if (installationLock) {
+    return installationLock
   }
-  traceInfo('Sqlmesh enterprise is not installed, installing...')
-  const abortController = new AbortController()
-  const installResult = await window.withProgress(
-    {
-      location: ProgressLocation.Notification,
-      title: 'Installing sqlmesh enterprise...',
-      cancellable: true,
-    },
-    async (progress, token) => {
-      // Connect the cancellation token to our abort controller
-      token.onCancellationRequested(() => {
-        abortController.abort()
-        traceInfo('Sqlmesh enterprise installation cancelled')
-        window.showInformationMessage('Installation cancelled')
-      })
-      progress.report({ message: 'Installing sqlmesh enterprise...' })
-      const result = await installSqlmeshEnterprise(abortController)
-      if (isErr(result)) {
-        return result
+
+  // Create a new lock
+  installationLock = (async () => {
+    try {
+      traceInfo('Ensuring sqlmesh enterprise is installed')
+      const isInstalled = await isSqlmeshEnterpriseInstalled()
+      if (isErr(isInstalled)) {
+        return isInstalled
+      }
+      if (isInstalled.value) {
+        traceInfo('Sqlmesh enterprise is installed')
+        return ok(false)
+      }
+      traceInfo('Sqlmesh enterprise is not installed, installing...')
+      const abortController = new AbortController()
+      const installResult = await window.withProgress(
+        {
+          location: ProgressLocation.Notification,
+          title: 'SQLMesh',
+          cancellable: true,
+        },
+        async (progress, token) => {
+          // Connect the cancellation token to our abort controller
+          token.onCancellationRequested(() => {
+            abortController.abort()
+            traceInfo('Sqlmesh enterprise installation cancelled')
+            window.showInformationMessage('Installation cancelled')
+          })
+          progress.report({ message: 'Installing enterprise python package...' })
+          const result = await installSqlmeshEnterprise(abortController)
+          if (isErr(result)) {
+            return result
+          }
+          return ok(true)
+        },
+      )
+      if (isErr(installResult)) {
+        return installResult
       }
       return ok(true)
-    },
-  )
-  if (isErr(installResult)) {
-    return installResult
-  }
-  return ok(true)
+    } finally {
+      // Clear the lock when done
+      installationLock = undefined
+    }
+  })()
+
+  return installationLock
 }
 
 /**
@@ -269,6 +287,12 @@ export const sqlmeshExec = async (): Promise<
       args: [],
     })
   } else {
+    const exists = await doesExecutableExist(sqlmesh)
+    if (!exists) {
+      return err({
+        type: 'sqlmesh_not_found',
+      })
+    }
     return ok({
       bin: sqlmesh,
       workspacePath,
@@ -371,10 +395,6 @@ export const sqlmeshLspExec = async (): Promise<
         return ensured
       }
     }
-    const ensuredDependencies = await ensureSqlmeshLspDependenciesInstalled()
-    if (isErr(ensuredDependencies)) {
-      return ensuredDependencies
-    }
     const binPath = path.join(interpreterDetails.binPath!, sqlmeshLSP)
     traceLog(`Bin path: ${binPath}`)
     if (!fs.existsSync(binPath)) {
@@ -382,6 +402,10 @@ export const sqlmeshLspExec = async (): Promise<
         type: 'sqlmesh_lsp_not_found',
       })
     }
+    const ensuredDependencies = await ensureSqlmeshLspDependenciesInstalled()
+    if (isErr(ensuredDependencies)) {
+      return ensuredDependencies
+    }
     return ok({
       bin: binPath,
       workspacePath,
@@ -393,6 +417,12 @@ export const sqlmeshLspExec = async (): Promise<
       args: [],
     })
   } else {
+    const exists = await doesExecutableExist(sqlmeshLSP)
+    if (!exists) {
+      return err({
+        type: 'sqlmesh_lsp_not_found',
+      })
+    }
     return ok({
       bin: sqlmeshLSP,
       workspacePath,
@@ -401,3 +431,18 @@ export const sqlmeshLspExec = async (): Promise<
     })
   }
 }
+
+async function doesExecutableExist(executable: string): Promise<boolean> {
+  const command = process.platform === 'win32' ? 'where.exe' : 'which'
+  traceLog(`Checking if ${executable} exists with ${command}`)
+  try {
+    const result = await execAsync(command, [executable])
+    traceLog(`Checked if ${executable} exists with ${command}, with result ${result.exitCode}`)
+    const exists = result.exitCode === 0
+    traceLog(`Checked if ${executable} exists with ${command}, with result ${exists}`)
+    return exists
+  } catch {
+    traceLog(`Checked if ${executable} exists with ${command}, errored, returning false`)
+    return false
+  }
+}
\ No newline at end of file
diff --git a/vscode/extension/src/webviews/lineagePanel.ts b/vscode/extension/src/webviews/lineagePanel.ts
index f9d84dc2..5bb3e169 100644
--- a/vscode/extension/src/webviews/lineagePanel.ts
+++ b/vscode/extension/src/webviews/lineagePanel.ts
@@ -16,37 +16,42 @@ export class LineagePanel implements WebviewViewProvider, Disposable {
 
   private panel: WebviewView | undefined
   private lsp: LSPClient
-  private _extensionUri: Uri
+  private readonly extensionUri: Uri
+
+  private disposables: Disposable[] = []
 
   public constructor(extensionUri: Uri, lsp: LSPClient) {
-    this._extensionUri = extensionUri
+    this.extensionUri = extensionUri
     this.lsp = lsp
 
     if (this.panel) {
       this.panel.webview.html = this.getHTML(this.panel.webview)
     }
 
-    workspace.onDidSaveTextDocument(document => {
-      this.panel?.webview.postMessage({
-        key: 'vscode_send',
-        payload: {
-          key: 'savedFile',
-          payload: { fileUri: document.uri.toString() },
-        },
-      })
-    })
-
-    window.onDidChangeActiveTextEditor(editor => {
-      if (editor) {
+    this.disposables.push(
+      workspace.onDidSaveTextDocument(document => {
         this.panel?.webview.postMessage({
           key: 'vscode_send',
           payload: {
-            key: 'changeFocusOnFile',
-            payload: { path: editor.document.uri.toString() },
+            key: 'savedFile',
+            payload: { fileUri: document.uri.toString() },
           },
         })
-      }
-    })
+      }),
+    )
+    this.disposables.push(
+      window.onDidChangeActiveTextEditor(editor => {
+        if (editor) {
+          this.panel?.webview.postMessage({
+            key: 'vscode_send',
+            payload: {
+              key: 'changeFocusOnFile',
+              payload: { path: editor.document.uri.toString() },
+            },
+          })
+        }
+      }),
+    )
   }
 
   public resolveWebviewView(webviewView: WebviewView) {
@@ -58,12 +63,12 @@ export class LineagePanel implements WebviewViewProvider, Disposable {
     webviewView.webview.options = {
       // Allow scripts in the webview
       enableScripts: true,
-      localResourceRoots: [this._extensionUri],
+      localResourceRoots: [this.extensionUri],
     }
 
     // Set content options for external URL access
     // Set up message listener for events from the iframe
-    webviewView.webview.onDidReceiveMessage(
+    const disposable = webviewView.webview.onDidReceiveMessage(
       async request => {
         if (!request) {
           return
@@ -99,7 +104,7 @@ export class LineagePanel implements WebviewViewProvider, Disposable {
                     result: response,
                   },
                 }
-                webviewView.webview.postMessage(responseCallback)
+                await webviewView.webview.postMessage(responseCallback)
                 break
               }
               case 'get_active_file': {
@@ -113,7 +118,7 @@ export class LineagePanel implements WebviewViewProvider, Disposable {
                     },
                   },
                 }
-                webviewView.webview.postMessage(responseCallback)
+                await webviewView.webview.postMessage(responseCallback)
                 break
               }
               default: {
@@ -132,21 +137,22 @@ export class LineagePanel implements WebviewViewProvider, Disposable {
       undefined,
       [],
     )
+    this.disposables.push(disposable)
     webviewView.webview.html = this.getHTML(webviewView.webview)
   }
 
   private getHTML(panel: Webview) {
     const cssUri = panel.asWebviewUri(
-      Uri.joinPath(this._extensionUri, 'src_react', 'assets', 'index.css'),
+      Uri.joinPath(this.extensionUri, 'src_react', 'assets', 'index.css'),
     )
     const jsUri = panel.asWebviewUri(
-      Uri.joinPath(this._extensionUri, 'src_react', 'assets', 'index.js'),
+      Uri.joinPath(this.extensionUri, 'src_react', 'assets', 'index.js'),
     )
     const faviconUri = panel.asWebviewUri(
-      Uri.joinPath(this._extensionUri, 'src_react', 'favicon.ico'),
+      Uri.joinPath(this.extensionUri, 'src_react', 'favicon.ico'),
     )
     const logoUri = panel.asWebviewUri(
-      Uri.joinPath(this._extensionUri, 'src_react', 'logo192.png'),
+      Uri.joinPath(this.extensionUri, 'src_react', 'logo192.png'),
     )
 
     // Handle query requests from the React app
@@ -179,5 +185,9 @@ export class LineagePanel implements WebviewViewProvider, Disposable {
     // WebviewView doesn't have a dispose method
     // We can clear references
     this.panel = undefined
+    this.disposables.forEach(disposable => {
+      disposable.dispose()
+    })
+    this.disposables = []
   }
 }


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 54f8db04..492da9db 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -33,14 +33,14 @@ from sqlmesh.core import lineage
 from sqlmesh.core.macros import macro
 from sqlmesh.core.model import IncrementalByTimeRangeKind, SqlModel, model
 from sqlmesh.core.model.kind import OnDestructiveChange
-from sqlmesh.core.plan import BuiltInPlanEvaluator, Plan
+from sqlmesh.core.plan import BuiltInPlanEvaluator, Plan, stages as plan_stages
 from sqlmesh.core.snapshot import (
+    DeployabilityIndex,
     Node,
     Snapshot,
     SnapshotChangeCategory,
     SnapshotDataVersion,
     SnapshotFingerprint,
-    DeployabilityIndex,
 )
 from sqlmesh.utils import random_id
 from sqlmesh.utils.date import TimeLike, to_date
@@ -258,11 +258,23 @@ def push_plan(context: Context, plan: Plan) -> None:
         context.default_catalog,
     )
     deployability_index = DeployabilityIndex.create(context.snapshots.values())
-    plan_evaluator._push(plan.to_evaluatable(), plan.snapshots, deployability_index)
-    promotion_result = plan_evaluator._promote(plan.to_evaluatable(), plan.snapshots)
-    plan_evaluator._update_views(
-        plan.to_evaluatable(), plan.snapshots, promotion_result, deployability_index
+    evaluatable_plan = plan.to_evaluatable()
+    stages = plan_stages.build_plan_stages(
+        evaluatable_plan, context.state_sync, context.default_catalog
     )
+    for stage in stages:
+        if isinstance(stage, plan_stages.CreateSnapshotRecordsStage):
+            plan_evaluator.visit_create_snapshot_records_stage(stage, evaluatable_plan)
+        elif isinstance(stage, plan_stages.PhysicalLayerUpdateStage):
+            stage.deployability_index = deployability_index
+            plan_evaluator.visit_physical_layer_update_stage(stage, evaluatable_plan)
+        elif isinstance(stage, plan_stages.EnvironmentRecordUpdateStage):
+            plan_evaluator.visit_environment_record_update_stage(stage, evaluatable_plan)
+        elif isinstance(stage, plan_stages.VirtualLayerUpdateStage):
+            stage.deployability_index = deployability_index
+            plan_evaluator.visit_virtual_layer_update_stage(stage, evaluatable_plan)
+        elif isinstance(stage, plan_stages.FinalizeEnvironmentStage):
+            plan_evaluator.visit_finalize_environment_stage(stage, evaluatable_plan)
 
 
 @pytest.fixture()
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index e88184c2..3ce1e488 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -41,7 +41,6 @@ from sqlmesh.core.model import load_sql_based_model, model, SqlModel, Model
 from sqlmesh.core.model.cache import OptimizedQueryCache
 from sqlmesh.core.renderer import render_statements
 from sqlmesh.core.model.kind import ModelKindName
-from sqlmesh.core.plan import BuiltInPlanEvaluator, PlanBuilder
 from sqlmesh.core.state_sync.cache import CachingStateSync
 from sqlmesh.core.state_sync.db import EngineAdapterStateSync
 from sqlmesh.utils.connection_pool import SingletonConnectionPool, ThreadLocalSharedConnectionPool
@@ -272,24 +271,6 @@ def test_diff(sushi_context: Context, mocker: MockerFixture):
     yesterday = yesterday_ds()
     success = sushi_context.run(start=yesterday, end=yesterday)
 
-    plan_evaluator = BuiltInPlanEvaluator(
-        sushi_context.state_sync,
-        sushi_context.snapshot_evaluator,
-        sushi_context.create_scheduler,
-        sushi_context.default_catalog,
-    )
-
-    plan = PlanBuilder(
-        context_diff=sushi_context._context_diff("prod"),
-    ).build()
-
-    # stringify used to trigger an unhashable exception due to
-    # https://github.com/pydantic/pydantic/issues/8016
-    assert str(plan) != ""
-
-    promotion_result = plan_evaluator._promote(plan.to_evaluatable(), plan.snapshots)
-    plan_evaluator._update_views(plan.to_evaluatable(), plan.snapshots, promotion_result)
-
     sushi_context.upsert_model("sushi.customers", query=parse_one("select 1 as customer_id"))
     sushi_context.diff("test")
     assert mock_console.show_environment_difference_summary.called
diff --git a/tests/core/test_plan_evaluator.py b/tests/core/test_plan_evaluator.py
index ae113e87..a784644b 100644
--- a/tests/core/test_plan_evaluator.py
+++ b/tests/core/test_plan_evaluator.py
@@ -8,10 +8,9 @@ from sqlmesh.core.plan import (
     BuiltInPlanEvaluator,
     Plan,
     PlanBuilder,
-    update_intervals_for_new_snapshots,
+    stages as plan_stages,
 )
 from sqlmesh.core.snapshot import SnapshotChangeCategory
-from sqlmesh.utils.date import to_timestamp
 
 
 @pytest.fixture
@@ -65,7 +64,14 @@ def test_builtin_evaluator_push(sushi_context: Context, make_snapshot):
         sushi_context.default_catalog,
         console=sushi_context.console,
     )
-    evaluator._push(plan.to_evaluatable(), plan.snapshots)
+    evaluatable_plan = plan.to_evaluatable()
+    stages = plan_stages.build_plan_stages(
+        evaluatable_plan, sushi_context.state_sync, sushi_context.default_catalog
+    )
+    assert isinstance(stages[0], plan_stages.CreateSnapshotRecordsStage)
+    evaluator.visit_create_snapshot_records_stage(stages[0], evaluatable_plan)
+    assert isinstance(stages[1], plan_stages.PhysicalLayerUpdateStage)
+    evaluator.visit_physical_layer_update_stage(stages[1], evaluatable_plan)
 
     assert (
         len(sushi_context.state_sync.get_snapshots([new_model_snapshot, new_view_model_snapshot]))
@@ -73,38 +79,3 @@ def test_builtin_evaluator_push(sushi_context: Context, make_snapshot):
     )
     assert sushi_context.engine_adapter.table_exists(new_model_snapshot.table_name())
     assert sushi_context.engine_adapter.table_exists(new_view_model_snapshot.table_name())
-
-
-@pytest.mark.parametrize(
-    "change_category", [SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.FORWARD_ONLY]
-)
-def test_update_intervals_for_new_snapshots(
-    sushi_context: Context,
-    mocker: MockerFixture,
-    change_category: SnapshotChangeCategory,
-    make_snapshot,
-):
-    model = SqlModel(
-        name="sushi.new_test_model",
-        query=parse_one("SELECT 1::INT AS one"),
-    )
-    snapshot = make_snapshot(model)
-    snapshot.categorize_as(change_category)
-
-    snapshot.add_interval("2023-01-01", "2023-01-01")
-
-    state_sync_mock = mocker.Mock()
-    state_sync_mock.refresh_snapshot_intervals.return_value = [snapshot]
-
-    update_intervals_for_new_snapshots([snapshot], state_sync_mock)
-
-    state_sync_mock.refresh_snapshot_intervals.assert_called_once_with([snapshot])
-
-    if change_category == SnapshotChangeCategory.FORWARD_ONLY:
-        assert snapshot.dev_intervals == [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))]
-        expected_intervals = snapshot.snapshot_intervals
-        expected_intervals.intervals.clear()
-        state_sync_mock.add_snapshots_intervals.assert_called_once_with([expected_intervals])
-    else:
-        assert not snapshot.dev_intervals
-        state_sync_mock.add_snapshots_intervals.assert_not_called()
diff --git a/tests/core/test_plan_stages.py b/tests/core/test_plan_stages.py
index 8c9cb9ac..5fa140cb 100644
--- a/tests/core/test_plan_stages.py
+++ b/tests/core/test_plan_stages.py
@@ -3,6 +3,7 @@ import typing as t
 from sqlglot import parse_one
 from pytest_mock.plugin import MockerFixture
 
+from sqlmesh.core.config import EnvironmentSuffixTarget
 from sqlmesh.core.model import SqlModel, ModelKindName
 from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.plan.stages import (
@@ -10,12 +11,15 @@ from sqlmesh.core.plan.stages import (
     AfterAllStage,
     AuditOnlyRunStage,
     PhysicalLayerUpdateStage,
+    CreateSnapshotRecordsStage,
     BeforeAllStage,
     BackfillStage,
     EnvironmentRecordUpdateStage,
     VirtualLayerUpdateStage,
     RestatementStage,
     MigrateSchemasStage,
+    FinalizeEnvironmentStage,
+    UnpauseStage,
 )
 from sqlmesh.core.snapshot.definition import (
     SnapshotChangeCategory,
@@ -55,6 +59,20 @@ def snapshot_b(make_snapshot, snapshot_a: Snapshot) -> Snapshot:
     return snapshot
 
 
+@pytest.fixture
+def snapshot_c(make_snapshot, snapshot_a: Snapshot) -> Snapshot:
+    snapshot = make_snapshot(
+        SqlModel(
+            name="c",
+            query=parse_one("select * from a"),
+            kind=dict(name=ModelKindName.VIEW),
+        ),
+        nodes={'"a"': snapshot_a.model},
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    return snapshot
+
+
 def test_build_plan_stages_basic(
     snapshot_a: Snapshot, snapshot_b: Snapshot, mocker: MockerFixture
 ) -> None:
@@ -104,10 +122,18 @@ def test_build_plan_stages_basic(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 4
+    assert len(stages) == 7
 
+    # Verify CreateSnapshotRecordsStage
+    create_snapshot_records_stage = stages[0]
+    assert isinstance(create_snapshot_records_stage, CreateSnapshotRecordsStage)
+    assert len(create_snapshot_records_stage.snapshots) == 2
+    assert {s.snapshot_id for s in create_snapshot_records_stage.snapshots} == {
+        snapshot_a.snapshot_id,
+        snapshot_b.snapshot_id,
+    }
     # Verify PhysicalLayerUpdateStage
-    physical_stage = stages[0]
+    physical_stage = stages[1]
     assert isinstance(physical_stage, PhysicalLayerUpdateStage)
     assert len(physical_stage.snapshots) == 2
     assert {s.snapshot_id for s in physical_stage.snapshots} == {
@@ -117,7 +143,7 @@ def test_build_plan_stages_basic(
     assert physical_stage.deployability_index == DeployabilityIndex.all_deployable()
 
     # Verify BackfillStage
-    backfill_stage = stages[1]
+    backfill_stage = stages[2]
     assert isinstance(backfill_stage, BackfillStage)
     assert backfill_stage.deployability_index == DeployabilityIndex.all_deployable()
     expected_interval = (to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))
@@ -126,14 +152,23 @@ def test_build_plan_stages_basic(
     assert backfill_stage.snapshot_to_intervals[snapshot_b] == [expected_interval]
 
     # Verify EnvironmentRecordUpdateStage
-    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert stages[3].no_gaps_snapshot_names == {snapshot_a.name, snapshot_b.name}
+
+    # Verify UnpauseStage
+    assert isinstance(stages[4], UnpauseStage)
+    assert {s.name for s in stages[4].promoted_snapshots} == {snapshot_a.name, snapshot_b.name}
 
     # Verify VirtualLayerUpdateStage
-    virtual_stage = stages[3]
+    virtual_stage = stages[5]
     assert isinstance(virtual_stage, VirtualLayerUpdateStage)
     assert len(virtual_stage.promoted_snapshots) == 2
     assert len(virtual_stage.demoted_snapshots) == 0
-    assert {s.name for s in virtual_stage.promoted_snapshots} == {'"a"', '"b"'}
+    assert {s.name for s in virtual_stage.promoted_snapshots} == {snapshot_a.name, snapshot_b.name}
+
+    state_reader.refresh_snapshot_intervals.assert_called_once()
+
+    assert isinstance(stages[6], FinalizeEnvironmentStage)
 
 
 def test_build_plan_stages_with_before_all_and_after_all(
@@ -154,6 +189,15 @@ def test_build_plan_stages_with_before_all_and_after_all(
         promoted_snapshot_ids=[snapshot_a.snapshot_id, snapshot_b.snapshot_id],
     )
 
+    environment_statements = [
+        EnvironmentStatements(
+            before_all=["BEFORE ALL A", "BEFORE ALL B"],
+            after_all=["AFTER ALL A", "AFTER ALL B"],
+            python_env={},
+            jinja_macros=None,
+        )
+    ]
+
     # Create evaluatable plan
     plan = EvaluatablePlan(
         start="2023-01-01",
@@ -177,14 +221,7 @@ def test_build_plan_stages_with_before_all_and_after_all(
         interval_end_per_model=None,
         execution_time="2023-01-02",
         disabled_restatement_models=set(),
-        environment_statements=[
-            EnvironmentStatements(
-                before_all=["BEFORE ALL A", "BEFORE ALL B"],
-                after_all=["AFTER ALL A", "AFTER ALL B"],
-                python_env={},
-                jinja_macros=None,
-            )
-        ],
+        environment_statements=environment_statements,
         user_provided_flags=None,
     )
 
@@ -192,15 +229,24 @@ def test_build_plan_stages_with_before_all_and_after_all(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 6
+    assert len(stages) == 9
 
     # Verify BeforeAllStage
     before_all_stage = stages[0]
     assert isinstance(before_all_stage, BeforeAllStage)
-    assert before_all_stage.statements == ["BEFORE ALL A", "BEFORE ALL B"]
+    assert before_all_stage.statements == environment_statements
+
+    # Verify CreateSnapshotRecordsStage
+    create_snapshot_records_stage = stages[1]
+    assert isinstance(create_snapshot_records_stage, CreateSnapshotRecordsStage)
+    assert len(create_snapshot_records_stage.snapshots) == 2
+    assert {s.snapshot_id for s in create_snapshot_records_stage.snapshots} == {
+        snapshot_a.snapshot_id,
+        snapshot_b.snapshot_id,
+    }
 
     # Verify PhysicalLayerUpdateStage
-    physical_stage = stages[1]
+    physical_stage = stages[2]
     assert isinstance(physical_stage, PhysicalLayerUpdateStage)
     assert len(physical_stage.snapshots) == 2
     assert {s.snapshot_id for s in physical_stage.snapshots} == {
@@ -210,7 +256,7 @@ def test_build_plan_stages_with_before_all_and_after_all(
     assert physical_stage.deployability_index == DeployabilityIndex.all_deployable()
 
     # Verify BackfillStage
-    backfill_stage = stages[2]
+    backfill_stage = stages[3]
     assert isinstance(backfill_stage, BackfillStage)
     assert backfill_stage.deployability_index == DeployabilityIndex.all_deployable()
     expected_interval = (to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))
@@ -219,19 +265,27 @@ def test_build_plan_stages_with_before_all_and_after_all(
     assert backfill_stage.snapshot_to_intervals[snapshot_b] == [expected_interval]
 
     # Verify EnvironmentRecordUpdateStage
-    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[4], EnvironmentRecordUpdateStage)
+    assert stages[4].no_gaps_snapshot_names == {snapshot_a.name, snapshot_b.name}
+
+    # Verify UnpauseStage
+    assert isinstance(stages[5], UnpauseStage)
+    assert {s.name for s in stages[5].promoted_snapshots} == {snapshot_a.name, snapshot_b.name}
 
     # Verify VirtualLayerUpdateStage
-    virtual_stage = stages[4]
+    virtual_stage = stages[6]
     assert isinstance(virtual_stage, VirtualLayerUpdateStage)
     assert len(virtual_stage.promoted_snapshots) == 2
     assert len(virtual_stage.demoted_snapshots) == 0
     assert {s.name for s in virtual_stage.promoted_snapshots} == {'"a"', '"b"'}
 
+    # Verify FinalizeEnvironmentStage
+    assert isinstance(stages[7], FinalizeEnvironmentStage)
+
     # Verify AfterAllStage
-    after_all_stage = stages[5]
+    after_all_stage = stages[8]
     assert isinstance(after_all_stage, AfterAllStage)
-    assert after_all_stage.statements == ["AFTER ALL A", "AFTER ALL B"]
+    assert after_all_stage.statements == environment_statements
 
 
 def test_build_plan_stages_select_models(
@@ -283,17 +337,26 @@ def test_build_plan_stages_select_models(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 4
+    assert len(stages) == 7
+
+    # Verify CreateSnapshotRecordsStage
+    create_snapshot_records_stage = stages[0]
+    assert isinstance(create_snapshot_records_stage, CreateSnapshotRecordsStage)
+    assert len(create_snapshot_records_stage.snapshots) == 2
+    assert {s.snapshot_id for s in create_snapshot_records_stage.snapshots} == {
+        snapshot_a.snapshot_id,
+        snapshot_b.snapshot_id,
+    }
 
     # Verify PhysicalLayerUpdateStage
-    physical_stage = stages[0]
+    physical_stage = stages[1]
     assert isinstance(physical_stage, PhysicalLayerUpdateStage)
     assert len(physical_stage.snapshots) == 1
     assert {s.snapshot_id for s in physical_stage.snapshots} == {snapshot_a.snapshot_id}
     assert physical_stage.deployability_index == DeployabilityIndex.all_deployable()
 
     # Verify BackfillStage
-    backfill_stage = stages[1]
+    backfill_stage = stages[2]
     assert isinstance(backfill_stage, BackfillStage)
     assert backfill_stage.deployability_index == DeployabilityIndex.all_deployable()
     expected_interval = (to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))
@@ -301,15 +364,23 @@ def test_build_plan_stages_select_models(
     assert backfill_stage.snapshot_to_intervals[snapshot_a] == [expected_interval]
 
     # Verify EnvironmentRecordUpdateStage
-    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert stages[3].no_gaps_snapshot_names == {snapshot_a.name}
+
+    # Verify UnpauseStage
+    assert isinstance(stages[4], UnpauseStage)
+    assert {s.name for s in stages[4].promoted_snapshots} == {snapshot_a.name}
 
     # Verify VirtualLayerUpdateStage
-    virtual_stage = stages[3]
+    virtual_stage = stages[5]
     assert isinstance(virtual_stage, VirtualLayerUpdateStage)
     assert len(virtual_stage.promoted_snapshots) == 1
     assert len(virtual_stage.demoted_snapshots) == 0
     assert {s.name for s in virtual_stage.promoted_snapshots} == {'"a"'}
 
+    # Verify FinalizeEnvironmentStage
+    assert isinstance(stages[6], FinalizeEnvironmentStage)
+
 
 @pytest.mark.parametrize("skip_backfill,empty_backfill", [(True, False), (False, True)])
 def test_build_plan_stages_basic_no_backfill(
@@ -365,10 +436,18 @@ def test_build_plan_stages_basic_no_backfill(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 4
+    assert len(stages) == 7
 
+    # Verify CreateSnapshotRecordsStage
+    create_snapshot_records_stage = stages[0]
+    assert isinstance(create_snapshot_records_stage, CreateSnapshotRecordsStage)
+    assert len(create_snapshot_records_stage.snapshots) == 2
+    assert {s.snapshot_id for s in create_snapshot_records_stage.snapshots} == {
+        snapshot_a.snapshot_id,
+        snapshot_b.snapshot_id,
+    }
     # Verify PhysicalLayerUpdateStage
-    physical_stage = stages[0]
+    physical_stage = stages[1]
     assert isinstance(physical_stage, PhysicalLayerUpdateStage)
     assert len(physical_stage.snapshots) == 2
     assert {s.snapshot_id for s in physical_stage.snapshots} == {
@@ -377,21 +456,29 @@ def test_build_plan_stages_basic_no_backfill(
     }
 
     # Verify BackfillStage
-    backfill_stage = stages[1]
+    backfill_stage = stages[2]
     assert isinstance(backfill_stage, BackfillStage)
     assert backfill_stage.deployability_index == DeployabilityIndex.all_deployable()
     assert backfill_stage.snapshot_to_intervals == {}
 
     # Verify EnvironmentRecordUpdateStage
-    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert stages[3].no_gaps_snapshot_names == {snapshot_a.name, snapshot_b.name}
+
+    # Verify UnpauseStage
+    assert isinstance(stages[4], UnpauseStage)
+    assert {s.name for s in stages[4].promoted_snapshots} == {snapshot_a.name, snapshot_b.name}
 
     # Verify VirtualLayerUpdateStage
-    virtual_stage = stages[3]
+    virtual_stage = stages[5]
     assert isinstance(virtual_stage, VirtualLayerUpdateStage)
     assert len(virtual_stage.promoted_snapshots) == 2
     assert len(virtual_stage.demoted_snapshots) == 0
     assert {s.name for s in virtual_stage.promoted_snapshots} == {'"a"', '"b"'}
 
+    # Verify FinalizeEnvironmentStage
+    assert isinstance(stages[6], FinalizeEnvironmentStage)
+
 
 def test_build_plan_stages_restatement(
     snapshot_a: Snapshot, snapshot_b: Snapshot, mocker: MockerFixture
@@ -458,7 +545,7 @@ def test_build_plan_stages_restatement(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 4
+    assert len(stages) == 5
 
     # Verify PhysicalLayerUpdateStage
     physical_stage = stages[0]
@@ -490,6 +577,9 @@ def test_build_plan_stages_restatement(
     # Verify EnvironmentRecordUpdateStage
     assert isinstance(stages[3], EnvironmentRecordUpdateStage)
 
+    # Verify FinalizeEnvironmentStage
+    assert isinstance(stages[4], FinalizeEnvironmentStage)
+
 
 def test_build_plan_stages_forward_only(
     snapshot_a: Snapshot, snapshot_b: Snapshot, make_snapshot, mocker: MockerFixture
@@ -566,10 +656,19 @@ def test_build_plan_stages_forward_only(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 5
+    assert len(stages) == 8
+
+    # Verify CreateSnapshotRecordsStage
+    create_snapshot_records_stage = stages[0]
+    assert isinstance(create_snapshot_records_stage, CreateSnapshotRecordsStage)
+    assert len(create_snapshot_records_stage.snapshots) == 2
+    assert {s.snapshot_id for s in create_snapshot_records_stage.snapshots} == {
+        new_snapshot_a.snapshot_id,
+        new_snapshot_b.snapshot_id,
+    }
 
     # Verify PhysicalLayerUpdateStage
-    physical_stage = stages[0]
+    physical_stage = stages[1]
     assert isinstance(physical_stage, PhysicalLayerUpdateStage)
     assert len(physical_stage.snapshots) == 2
     assert {s.snapshot_id for s in physical_stage.snapshots} == {
@@ -581,10 +680,11 @@ def test_build_plan_stages_forward_only(
     )
 
     # Verify EnvironmentRecordUpdateStage
-    assert isinstance(stages[1], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert stages[2].no_gaps_snapshot_names == set()
 
     # Verify MigrateSchemasStage
-    migrate_stage = stages[2]
+    migrate_stage = stages[3]
     assert isinstance(migrate_stage, MigrateSchemasStage)
     assert len(migrate_stage.snapshots) == 2
     assert {s.snapshot_id for s in migrate_stage.snapshots} == {
@@ -592,8 +692,15 @@ def test_build_plan_stages_forward_only(
         new_snapshot_b.snapshot_id,
     }
 
+    # Verify UnpauseStage
+    assert isinstance(stages[4], UnpauseStage)
+    assert {s.name for s in stages[4].promoted_snapshots} == {
+        new_snapshot_a.name,
+        new_snapshot_b.name,
+    }
+
     # Verify BackfillStage
-    backfill_stage = stages[3]
+    backfill_stage = stages[5]
     assert isinstance(backfill_stage, BackfillStage)
     assert len(backfill_stage.snapshot_to_intervals) == 2
     expected_interval = [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))]
@@ -602,12 +709,15 @@ def test_build_plan_stages_forward_only(
     assert backfill_stage.deployability_index == DeployabilityIndex.all_deployable()
 
     # Verify VirtualLayerUpdateStage
-    virtual_stage = stages[4]
+    virtual_stage = stages[6]
     assert isinstance(virtual_stage, VirtualLayerUpdateStage)
     assert len(virtual_stage.promoted_snapshots) == 2
     assert len(virtual_stage.demoted_snapshots) == 0
     assert {s.name for s in virtual_stage.promoted_snapshots} == {'"a"', '"b"'}
 
+    # Verify FinalizeEnvironmentStage
+    assert isinstance(stages[7], FinalizeEnvironmentStage)
+
 
 def test_build_plan_stages_forward_only_dev(
     snapshot_a: Snapshot, snapshot_b: Snapshot, make_snapshot, mocker: MockerFixture
@@ -674,10 +784,19 @@ def test_build_plan_stages_forward_only_dev(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 4
+    assert len(stages) == 6
+
+    # Verify CreateSnapshotRecordsStage
+    create_snapshot_records_stage = stages[0]
+    assert isinstance(create_snapshot_records_stage, CreateSnapshotRecordsStage)
+    assert len(create_snapshot_records_stage.snapshots) == 2
+    assert {s.snapshot_id for s in create_snapshot_records_stage.snapshots} == {
+        new_snapshot_a.snapshot_id,
+        new_snapshot_b.snapshot_id,
+    }
 
     # Verify PhysicalLayerUpdateStage
-    physical_stage = stages[0]
+    physical_stage = stages[1]
     assert isinstance(physical_stage, PhysicalLayerUpdateStage)
     assert len(physical_stage.snapshots) == 2
     assert {s.snapshot_id for s in physical_stage.snapshots} == {
@@ -689,7 +808,7 @@ def test_build_plan_stages_forward_only_dev(
     )
 
     # Verify BackfillStage
-    backfill_stage = stages[1]
+    backfill_stage = stages[2]
     assert isinstance(backfill_stage, BackfillStage)
     assert len(backfill_stage.snapshot_to_intervals) == 2
     expected_interval = [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))]
@@ -700,15 +819,18 @@ def test_build_plan_stages_forward_only_dev(
     )
 
     # Verify EnvironmentRecordUpdateStage
-    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
 
     # Verify VirtualLayerUpdateStage
-    virtual_stage = stages[3]
+    virtual_stage = stages[4]
     assert isinstance(virtual_stage, VirtualLayerUpdateStage)
     assert len(virtual_stage.promoted_snapshots) == 2
     assert len(virtual_stage.demoted_snapshots) == 0
     assert {s.name for s in virtual_stage.promoted_snapshots} == {'"a"', '"b"'}
 
+    # Verify FinalizeEnvironmentStage
+    assert isinstance(stages[5], FinalizeEnvironmentStage)
+
 
 def test_build_plan_stages_audit_only(
     snapshot_a: Snapshot, snapshot_b: Snapshot, make_snapshot, mocker: MockerFixture
@@ -785,30 +907,478 @@ def test_build_plan_stages_audit_only(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 5
+    assert len(stages) == 7
+
+    # Verify CreateSnapshotRecordsStage
+    create_snapshot_records_stage = stages[0]
+    assert isinstance(create_snapshot_records_stage, CreateSnapshotRecordsStage)
+    assert len(create_snapshot_records_stage.snapshots) == 2
+    assert {s.snapshot_id for s in create_snapshot_records_stage.snapshots} == {
+        new_snapshot_a.snapshot_id,
+        new_snapshot_b.snapshot_id,
+    }
 
     # Verify PhysicalLayerUpdateStage
-    physical_stage = stages[0]
+    physical_stage = stages[1]
     assert isinstance(physical_stage, PhysicalLayerUpdateStage)
-    assert len(physical_stage.snapshots) == 0
+    assert len(physical_stage.snapshots) == 2
+    assert {s.snapshot_id for s in physical_stage.snapshots} == {
+        new_snapshot_a.snapshot_id,
+        new_snapshot_b.snapshot_id,
+    }
+    assert physical_stage.deployability_index == DeployabilityIndex.create(
+        [new_snapshot_a, new_snapshot_b]
+    )
 
     # Verify AuditOnlyRunStage
-    audit_only_stage = stages[1]
+    audit_only_stage = stages[2]
     assert isinstance(audit_only_stage, AuditOnlyRunStage)
     assert len(audit_only_stage.snapshots) == 1
     assert audit_only_stage.snapshots[0].snapshot_id == new_snapshot_a.snapshot_id
 
     # Verify BackfillStage
-    backfill_stage = stages[2]
+    backfill_stage = stages[3]
     assert isinstance(backfill_stage, BackfillStage)
     assert len(backfill_stage.snapshot_to_intervals) == 0
 
     # Verify EnvironmentRecordUpdateStage
-    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[4], EnvironmentRecordUpdateStage)
 
     # Verify VirtualLayerUpdateStage
-    virtual_stage = stages[4]
+    virtual_stage = stages[5]
     assert isinstance(virtual_stage, VirtualLayerUpdateStage)
     assert len(virtual_stage.promoted_snapshots) == 2
     assert len(virtual_stage.demoted_snapshots) == 0
     assert {s.name for s in virtual_stage.promoted_snapshots} == {'"a"', '"b"'}
+
+    # Verify FinalizeEnvironmentStage
+    assert isinstance(stages[6], FinalizeEnvironmentStage)
+
+
+def test_build_plan_stages_forward_only_ensure_finalized_snapshots(
+    snapshot_a: Snapshot, snapshot_b: Snapshot, make_snapshot, mocker: MockerFixture
+) -> None:
+    # Categorize snapshot_a as forward-only
+    new_snapshot_a = make_snapshot(
+        snapshot_a.model.copy(update={"stamp": "new_version"}),
+    )
+    new_snapshot_a.previous_versions = snapshot_a.all_versions
+    new_snapshot_a.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
+
+    new_snapshot_b = make_snapshot(
+        snapshot_b.model.copy(),
+        nodes={'"a"': new_snapshot_a.model},
+    )
+    new_snapshot_b.previous_versions = snapshot_b.all_versions
+    new_snapshot_b.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
+
+    state_reader = mocker.Mock(spec=StateReader)
+    state_reader.get_snapshots.return_value = {}
+    existing_environment = Environment(
+        name="prod",
+        snapshots=[snapshot_a.table_info, snapshot_b.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="previous_plan",
+        previous_plan_id=None,
+        promoted_snapshot_ids=[snapshot_a.snapshot_id, snapshot_b.snapshot_id],
+        finalized_ts=to_timestamp("2023-01-02"),
+    )
+    state_reader.get_environment.return_value = existing_environment
+
+    # Create environment
+    environment = Environment(
+        name="prod",
+        snapshots=[new_snapshot_a.table_info, new_snapshot_b.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="test_plan",
+        previous_plan_id="previous_plan",
+        promoted_snapshot_ids=[new_snapshot_a.snapshot_id, new_snapshot_b.snapshot_id],
+    )
+
+    # Create evaluatable plan
+    plan = EvaluatablePlan(
+        start="2023-01-01",
+        end="2023-01-02",
+        new_snapshots=[new_snapshot_a, new_snapshot_b],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        empty_backfill=False,
+        restatements={},
+        is_dev=False,
+        allow_destructive_models=set(),
+        forward_only=False,
+        end_bounded=False,
+        ensure_finalized_snapshots=True,
+        directly_modified_snapshots=[new_snapshot_a.snapshot_id],
+        indirectly_modified_snapshots={
+            new_snapshot_a.name: [new_snapshot_b.snapshot_id],
+        },
+        removed_snapshots=[],
+        requires_backfill=True,
+        models_to_backfill=None,
+        interval_end_per_model=None,
+        execution_time="2023-01-02",
+        disabled_restatement_models=set(),
+        environment_statements=None,
+        user_provided_flags=None,
+    )
+
+    # Build plan stages
+    stages = build_plan_stages(plan, state_reader, None)
+
+    assert len(stages) == 8
+    assert isinstance(stages[0], CreateSnapshotRecordsStage)
+    assert isinstance(stages[1], PhysicalLayerUpdateStage)
+    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[3], MigrateSchemasStage)
+    assert isinstance(stages[4], BackfillStage)
+    assert isinstance(stages[5], UnpauseStage)
+    assert isinstance(stages[6], VirtualLayerUpdateStage)
+    assert isinstance(stages[7], FinalizeEnvironmentStage)
+
+
+def test_build_plan_stages_removed_model(
+    snapshot_a: Snapshot, snapshot_b: Snapshot, mocker: MockerFixture
+) -> None:
+    # Mock state reader
+    state_reader = mocker.Mock(spec=StateReader)
+    state_reader.get_snapshots.return_value = {
+        snapshot_a.snapshot_id: snapshot_a,
+        snapshot_b.snapshot_id: snapshot_b,
+    }
+    existing_environment = Environment(
+        name="prod",
+        snapshots=[snapshot_a.table_info, snapshot_b.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="previous_plan",
+        previous_plan_id=None,
+        promoted_snapshot_ids=[snapshot_a.snapshot_id, snapshot_b.snapshot_id],
+        finalized_ts=to_timestamp("2023-01-02"),
+    )
+    state_reader.get_environment.return_value = existing_environment
+
+    # Create environment
+    environment = Environment(
+        snapshots=[snapshot_a.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="test_plan",
+        previous_plan_id="previous_plan",
+        promoted_snapshot_ids=[snapshot_a.snapshot_id],
+    )
+
+    # Create evaluatable plan
+    plan = EvaluatablePlan(
+        start="2023-01-01",
+        end="2023-01-02",
+        new_snapshots=[],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        empty_backfill=False,
+        restatements={},
+        is_dev=False,
+        allow_destructive_models=set(),
+        forward_only=False,
+        end_bounded=False,
+        ensure_finalized_snapshots=False,
+        directly_modified_snapshots=[],
+        indirectly_modified_snapshots={},
+        removed_snapshots=[snapshot_b.snapshot_id],
+        requires_backfill=False,
+        models_to_backfill=None,
+        interval_end_per_model=None,
+        execution_time="2023-01-02",
+        disabled_restatement_models=set(),
+        environment_statements=None,
+        user_provided_flags=None,
+    )
+
+    # Build plan stages
+    stages = build_plan_stages(plan, state_reader, None)
+
+    # Verify stages
+    assert len(stages) == 5
+
+    assert isinstance(stages[0], PhysicalLayerUpdateStage)
+    assert isinstance(stages[1], BackfillStage)
+    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[3], VirtualLayerUpdateStage)
+    assert isinstance(stages[4], FinalizeEnvironmentStage)
+
+    virtual_layer_update_stage = stages[3]
+    assert virtual_layer_update_stage.promoted_snapshots == set()
+    assert virtual_layer_update_stage.demoted_snapshots == {snapshot_b.table_info}
+    assert (
+        virtual_layer_update_stage.demoted_environment_naming_info
+        == existing_environment.naming_info
+    )
+
+
+def test_build_plan_stages_environment_suffix_target_changed(
+    snapshot_a: Snapshot, snapshot_b: Snapshot, mocker: MockerFixture
+) -> None:
+    # Mock state reader
+    state_reader = mocker.Mock(spec=StateReader)
+    state_reader.get_snapshots.return_value = {
+        snapshot_a.snapshot_id: snapshot_a,
+        snapshot_b.snapshot_id: snapshot_b,
+    }
+    existing_environment = Environment(
+        name="dev",
+        snapshots=[snapshot_a.table_info, snapshot_b.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="previous_plan",
+        previous_plan_id=None,
+        promoted_snapshot_ids=[snapshot_a.snapshot_id, snapshot_b.snapshot_id],
+        finalized_ts=to_timestamp("2023-01-02"),
+    )
+    state_reader.get_environment.return_value = existing_environment
+
+    # Create environment
+    environment = Environment(
+        name="dev",
+        snapshots=[snapshot_a.table_info, snapshot_b.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="test_plan",
+        previous_plan_id="previous_plan",
+        promoted_snapshot_ids=[snapshot_a.snapshot_id, snapshot_b.snapshot_id],
+        suffix_target=EnvironmentSuffixTarget.TABLE,
+    )
+
+    # Create evaluatable plan
+    plan = EvaluatablePlan(
+        start="2023-01-01",
+        end="2023-01-02",
+        new_snapshots=[],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        empty_backfill=False,
+        restatements={},
+        is_dev=True,
+        allow_destructive_models=set(),
+        forward_only=False,
+        end_bounded=False,
+        ensure_finalized_snapshots=False,
+        directly_modified_snapshots=[],
+        indirectly_modified_snapshots={},
+        removed_snapshots=[],
+        requires_backfill=False,
+        models_to_backfill=None,
+        interval_end_per_model=None,
+        execution_time="2023-01-02",
+        disabled_restatement_models=set(),
+        environment_statements=None,
+        user_provided_flags=None,
+    )
+
+    # Build plan stages
+    stages = build_plan_stages(plan, state_reader, None)
+
+    # Verify stages
+    assert len(stages) == 5
+
+    assert isinstance(stages[0], PhysicalLayerUpdateStage)
+    assert isinstance(stages[1], BackfillStage)
+    assert isinstance(stages[2], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[3], VirtualLayerUpdateStage)
+    assert isinstance(stages[4], FinalizeEnvironmentStage)
+
+    virtual_layer_update_stage = stages[3]
+    assert virtual_layer_update_stage.promoted_snapshots == {
+        snapshot_a.table_info,
+        snapshot_b.table_info,
+    }
+    assert virtual_layer_update_stage.demoted_snapshots == {
+        snapshot_a.table_info,
+        snapshot_b.table_info,
+    }
+    assert (
+        virtual_layer_update_stage.demoted_environment_naming_info
+        == existing_environment.naming_info
+    )
+
+
+def test_build_plan_stages_indirect_non_breaking_no_migration(
+    snapshot_a: Snapshot, snapshot_b: Snapshot, make_snapshot, mocker: MockerFixture
+) -> None:
+    # Categorize snapshot_a as forward-only
+    new_snapshot_a = make_snapshot(
+        snapshot_a.model.copy(update={"stamp": "new_version"}),
+    )
+    new_snapshot_a.previous_versions = snapshot_a.all_versions
+    new_snapshot_a.categorize_as(SnapshotChangeCategory.NON_BREAKING)
+
+    new_snapshot_b = make_snapshot(
+        snapshot_b.model.copy(),
+        nodes={'"a"': new_snapshot_a.model},
+    )
+    new_snapshot_b.previous_versions = snapshot_b.all_versions
+    new_snapshot_b.change_category = SnapshotChangeCategory.INDIRECT_NON_BREAKING
+    new_snapshot_b.version = new_snapshot_b.previous_version.data_version.version
+
+    state_reader = mocker.Mock(spec=StateReader)
+    state_reader.get_snapshots.return_value = {}
+    existing_environment = Environment(
+        name="prod",
+        snapshots=[snapshot_a.table_info, snapshot_b.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="previous_plan",
+        previous_plan_id=None,
+        promoted_snapshot_ids=[snapshot_a.snapshot_id, snapshot_b.snapshot_id],
+        finalized_ts=to_timestamp("2023-01-02"),
+    )
+    state_reader.get_environment.return_value = existing_environment
+
+    # Create environment
+    environment = Environment(
+        name="prod",
+        snapshots=[new_snapshot_a.table_info, new_snapshot_b.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="test_plan",
+        previous_plan_id="previous_plan",
+        promoted_snapshot_ids=[new_snapshot_a.snapshot_id, new_snapshot_b.snapshot_id],
+    )
+
+    # Create evaluatable plan
+    plan = EvaluatablePlan(
+        start="2023-01-01",
+        end="2023-01-02",
+        new_snapshots=[new_snapshot_a, new_snapshot_b],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        empty_backfill=False,
+        restatements={},
+        is_dev=False,
+        allow_destructive_models=set(),
+        forward_only=False,
+        end_bounded=False,
+        ensure_finalized_snapshots=False,
+        directly_modified_snapshots=[new_snapshot_a.snapshot_id],
+        indirectly_modified_snapshots={
+            new_snapshot_a.name: [new_snapshot_b.snapshot_id],
+        },
+        removed_snapshots=[],
+        requires_backfill=True,
+        models_to_backfill=None,
+        interval_end_per_model=None,
+        execution_time="2023-01-02",
+        disabled_restatement_models=set(),
+        environment_statements=None,
+        user_provided_flags=None,
+    )
+
+    # Build plan stages
+    stages = build_plan_stages(plan, state_reader, None)
+
+    # Verify stages
+    assert len(stages) == 7
+
+    assert isinstance(stages[0], CreateSnapshotRecordsStage)
+    assert isinstance(stages[1], PhysicalLayerUpdateStage)
+    assert isinstance(stages[2], BackfillStage)
+    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[4], UnpauseStage)
+    assert isinstance(stages[5], VirtualLayerUpdateStage)
+    assert isinstance(stages[6], FinalizeEnvironmentStage)
+
+
+def test_build_plan_stages_indirect_non_breaking_view_migration(
+    snapshot_a: Snapshot, snapshot_c: Snapshot, make_snapshot, mocker: MockerFixture
+) -> None:
+    # Categorize snapshot_a as forward-only
+    new_snapshot_a = make_snapshot(
+        snapshot_a.model.copy(update={"stamp": "new_version"}),
+    )
+    new_snapshot_a.previous_versions = snapshot_a.all_versions
+    new_snapshot_a.categorize_as(SnapshotChangeCategory.NON_BREAKING)
+
+    new_snapshot_c = make_snapshot(
+        snapshot_c.model.copy(),
+        nodes={'"a"': new_snapshot_a.model},
+    )
+    new_snapshot_c.previous_versions = snapshot_c.all_versions
+    new_snapshot_c.change_category = SnapshotChangeCategory.INDIRECT_NON_BREAKING
+    new_snapshot_c.version = new_snapshot_c.previous_version.data_version.version
+
+    state_reader = mocker.Mock(spec=StateReader)
+    state_reader.get_snapshots.return_value = {}
+    existing_environment = Environment(
+        name="prod",
+        snapshots=[snapshot_a.table_info, snapshot_c.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="previous_plan",
+        previous_plan_id=None,
+        promoted_snapshot_ids=[snapshot_a.snapshot_id, snapshot_c.snapshot_id],
+        finalized_ts=to_timestamp("2023-01-02"),
+    )
+    state_reader.get_environment.return_value = existing_environment
+
+    # Create environment
+    environment = Environment(
+        name="prod",
+        snapshots=[new_snapshot_a.table_info, new_snapshot_c.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="test_plan",
+        previous_plan_id="previous_plan",
+        promoted_snapshot_ids=[new_snapshot_a.snapshot_id, new_snapshot_c.snapshot_id],
+    )
+
+    # Create evaluatable plan
+    plan = EvaluatablePlan(
+        start="2023-01-01",
+        end="2023-01-02",
+        new_snapshots=[new_snapshot_a, new_snapshot_c],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        empty_backfill=False,
+        restatements={},
+        is_dev=False,
+        allow_destructive_models=set(),
+        forward_only=False,
+        end_bounded=False,
+        ensure_finalized_snapshots=False,
+        directly_modified_snapshots=[new_snapshot_a.snapshot_id],
+        indirectly_modified_snapshots={
+            new_snapshot_a.name: [new_snapshot_c.snapshot_id],
+        },
+        removed_snapshots=[],
+        requires_backfill=True,
+        models_to_backfill=None,
+        interval_end_per_model=None,
+        execution_time="2023-01-02",
+        disabled_restatement_models=set(),
+        environment_statements=None,
+        user_provided_flags=None,
+    )
+
+    # Build plan stages
+    stages = build_plan_stages(plan, state_reader, None)
+
+    # Verify stages
+    assert len(stages) == 8
+
+    assert isinstance(stages[0], CreateSnapshotRecordsStage)
+    assert isinstance(stages[1], PhysicalLayerUpdateStage)
+    assert isinstance(stages[2], BackfillStage)
+    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[4], MigrateSchemasStage)
+    assert isinstance(stages[5], UnpauseStage)
+    assert isinstance(stages[6], VirtualLayerUpdateStage)
+    assert isinstance(stages[7], FinalizeEnvironmentStage)
+
+    migrate_schemas_stage = stages[4]
+    assert {s.snapshot_id for s in migrate_schemas_stage.snapshots} == {new_snapshot_c.snapshot_id}
diff --git a/vscode/extension/tests/bad_setup.spec.ts b/vscode/extension/tests/bad_setup.spec.ts
new file mode 100644
index 00000000..2c92e897
--- /dev/null
+++ b/vscode/extension/tests/bad_setup.spec.ts
@@ -0,0 +1,113 @@
+import { expect, test } from '@playwright/test'
+import fs from 'fs-extra'
+import os from 'os'
+import path from 'path'
+import {
+  createVirtualEnvironment,
+  openLineageView,
+  pipInstall,
+  REPO_ROOT,
+  startVSCode,
+  SUSHI_SOURCE_PATH,
+} from './utils'
+
+test('missing LSP dependencies shows install prompt', async ({}, testInfo) => {
+  testInfo.setTimeout(120_000) // 2 minutes for venv creation and package installation
+  const tempDir = await fs.mkdtemp(
+    path.join(os.tmpdir(), 'vscode-test-tcloud-'),
+  )
+  const pythonEnvDir = path.join(tempDir, '.venv')
+  const pythonDetails = await createVirtualEnvironment(pythonEnvDir)
+  const custom_materializations = path.join(
+    REPO_ROOT,
+    'examples',
+    'custom_materializations',
+  )
+  const sqlmeshWithExtras = `${REPO_ROOT}[bigquery]`
+  await pipInstall(pythonDetails, [sqlmeshWithExtras, custom_materializations])
+
+  try {
+    // Copy sushi project
+    await fs.copy(SUSHI_SOURCE_PATH, tempDir)
+
+    // Configure VS Code settings to use our Python environment
+    const settings = {
+      'python.defaultInterpreterPath': pythonDetails.pythonPath,
+      'sqlmesh.environmentPath': pythonEnvDir,
+    }
+    await fs.ensureDir(path.join(tempDir, '.vscode'))
+    await fs.writeJson(
+      path.join(tempDir, '.vscode', 'settings.json'),
+      settings,
+      { spaces: 2 },
+    )
+
+    // Start VS Code
+    const { window, close } = await startVSCode(tempDir)
+
+    // Open a SQL file to trigger SQLMesh activation
+    // Wait for the models folder to be visible
+    await window.waitForSelector('text=models')
+
+    // Click on the models folder
+    await window
+      .getByRole('treeitem', { name: 'models', exact: true })
+      .locator('a')
+      .click()
+
+    // Open the top_waiters model
+    await window
+      .getByRole('treeitem', { name: 'customers.sql', exact: true })
+      .locator('a')
+      .click()
+
+    // Wait for the message to show that LSP extras need to be installed
+    await window.waitForSelector('text=LSP dependencies missing')
+    expect(await window.locator('text=Install').count()).toBeGreaterThanOrEqual(
+      1,
+    )
+
+    await close()
+  } finally {
+    // Clean up
+    await fs.remove(tempDir)
+  }
+})
+
+test('lineage, no sqlmesh found', async ({}) => {
+  const tempDir = await fs.mkdtemp(
+    path.join(os.tmpdir(), 'vscode-test-tcloud-'),
+  )
+  const pythonEnvDir = path.join(tempDir, '.venv')
+  const pythonDetails = await createVirtualEnvironment(pythonEnvDir)
+
+  try {
+    // Copy sushi project
+    await fs.copy(SUSHI_SOURCE_PATH, tempDir)
+
+    // Configure VS Code settings to use our Python environment
+    const settings = {
+      'python.defaultInterpreterPath': pythonDetails.pythonPath,
+      'sqlmesh.environmentPath': pythonEnvDir,
+    }
+    await fs.ensureDir(path.join(tempDir, '.vscode'))
+    await fs.writeJson(
+      path.join(tempDir, '.vscode', 'settings.json'),
+      settings,
+      { spaces: 2 },
+    )
+
+    const { window, close } = await startVSCode(tempDir)
+
+    // Open lineage view
+    await openLineageView(window)
+
+    // Assert shows that sqlmesh is not installed
+    await window.waitForSelector('text=SQLMesh LSP not found')
+
+    await close()
+  } finally {
+    // Clean up
+    await fs.remove(tempDir)
+  }
+})
diff --git a/vscode/extension/tests/lineage.spec.ts b/vscode/extension/tests/lineage.spec.ts
index e2a049a9..a7540780 100644
--- a/vscode/extension/tests/lineage.spec.ts
+++ b/vscode/extension/tests/lineage.spec.ts
@@ -2,19 +2,14 @@ import { test, expect, Page } from '@playwright/test'
 import path from 'path'
 import fs from 'fs-extra'
 import os from 'os'
-import { startVSCode, SUSHI_SOURCE_PATH } from './utils'
+import { openLineageView, startVSCode, SUSHI_SOURCE_PATH } from './utils'
 import { writeFileSync } from 'fs'
 
 /**
  * Helper function to launch VS Code and test lineage with given project path config
  */
 async function testLineageWithProjectPath(window: Page): Promise<void> {
-  // Trigger lineage command
-  await window.keyboard.press(
-    process.platform === 'darwin' ? 'Meta+Shift+P' : 'Control+Shift+P',
-  )
-  await window.keyboard.type('Lineage: Focus On View')
-  await window.keyboard.press('Enter')
+  await openLineageView(window)
 
   // Wait for "Loaded SQLMesh context" text to appear
   const loadedContextText = window.locator('text=Loaded SQLMesh context')
diff --git a/vscode/extension/tests/tcloud.spec.ts b/vscode/extension/tests/tcloud.spec.ts
new file mode 100644
index 00000000..d047a9dc
--- /dev/null
+++ b/vscode/extension/tests/tcloud.spec.ts
@@ -0,0 +1,202 @@
+import { expect, test } from '@playwright/test'
+import path from 'path'
+import fs from 'fs-extra'
+import os from 'os'
+import {
+  createVirtualEnvironment,
+  pipInstall,
+  REPO_ROOT,
+  startVSCode,
+  SUSHI_SOURCE_PATH,
+} from './utils'
+
+/**
+ * Helper function to create and set up a Python virtual environment
+ */
+async function setupPythonEnvironment(envDir: string): Promise<string> {
+  // Create virtual environment
+  const pythonDetails = await createVirtualEnvironment(envDir)
+
+  // Install the mock tcloud package
+  const mockTcloudPath = path.join(__dirname, 'tcloud')
+  await pipInstall(pythonDetails, [mockTcloudPath])
+
+  // Install sqlmesh from the local repository with LSP support
+  const customMaterializations = path.join(
+    REPO_ROOT,
+    'examples',
+    'custom_materializations',
+  )
+  const sqlmeshWithExtras = `${REPO_ROOT}[lsp,bigquery]`
+  await pipInstall(pythonDetails, [sqlmeshWithExtras, customMaterializations])
+
+  return pythonDetails.pythonPath
+}
+
+/**
+ * Helper function to set up a pre-authenticated tcloud state
+ */
+async function setupAuthenticatedState(tempDir: string): Promise<void> {
+  const authStateFile = path.join(tempDir, '.tcloud_auth_state.json')
+  const authState = {
+    is_logged_in: true,
+    id_token: {
+      iss: 'https://mock.tobikodata.com',
+      aud: 'mock-audience',
+      sub: 'user-123',
+      scope: 'openid email profile',
+      iat: Math.floor(Date.now() / 1000),
+      exp: Math.floor(Date.now() / 1000) + 3600, // Valid for 1 hour
+      email: 'test@example.com',
+      name: 'Test User',
+    },
+  }
+  await fs.writeJson(authStateFile, authState)
+}
+
+test.describe('Tcloud', () => {
+  test('not signed in, shows sign in window', async ({}, testInfo) => {
+    testInfo.setTimeout(120_000) // 2 minutes for venv creation and package installation
+    const tempDir = await fs.mkdtemp(
+      path.join(os.tmpdir(), 'vscode-test-tcloud-'),
+    )
+    const pythonEnvDir = path.join(tempDir, '.venv')
+
+    try {
+      // Copy sushi project
+      await fs.copy(SUSHI_SOURCE_PATH, tempDir)
+
+      // Create a tcloud.yaml to mark this as a tcloud project
+      const tcloudConfig = {
+        url: 'https://mock.tobikodata.com',
+        org: 'test-org',
+        project: 'test-project',
+      }
+      await fs.writeFile(
+        path.join(tempDir, 'tcloud.yaml'),
+        `url: ${tcloudConfig.url}\norg: ${tcloudConfig.org}\nproject: ${tcloudConfig.project}\n`,
+      )
+
+      // Set up Python environment with mock tcloud and sqlmesh
+      const pythonPath = await setupPythonEnvironment(pythonEnvDir)
+
+      // Configure VS Code settings to use our Python environment
+      const settings = {
+        'python.defaultInterpreterPath': pythonPath,
+        'sqlmesh.environmentPath': pythonEnvDir,
+      }
+      await fs.ensureDir(path.join(tempDir, '.vscode'))
+      await fs.writeJson(
+        path.join(tempDir, '.vscode', 'settings.json'),
+        settings,
+        { spaces: 2 },
+      )
+
+      // Start VS Code
+      const { window, close } = await startVSCode(tempDir)
+
+      // Open a SQL file to trigger SQLMesh activation
+      // Wait for the models folder to be visible
+      await window.waitForSelector('text=models')
+
+      // Click on the models folder
+      await window
+        .getByRole('treeitem', { name: 'models', exact: true })
+        .locator('a')
+        .click()
+
+      // Open the top_waiters model
+      await window
+        .getByRole('treeitem', { name: 'customers.sql', exact: true })
+        .locator('a')
+        .click()
+
+      // Wait for the file to open
+      await window.waitForTimeout(2000)
+
+      await window.waitForSelector(
+        'text=Please sign in to Tobiko Cloud to use SQLMesh',
+      )
+
+      // Close VS Code
+      await close()
+    } finally {
+      // Clean up
+      await fs.remove(tempDir)
+    }
+  })
+
+  test('signed in and not installed shows installation window and can see output', async ({}, testInfo) => {
+    testInfo.setTimeout(120_000) // 2 minutes for venv creation and package installation
+    const tempDir = await fs.mkdtemp(
+      path.join(os.tmpdir(), 'vscode-test-tcloud-'),
+    )
+    const pythonEnvDir = path.join(tempDir, '.venv')
+
+    try {
+      // Copy sushi project
+      await fs.copy(SUSHI_SOURCE_PATH, tempDir)
+
+      // Create a tcloud.yaml to mark this as a tcloud project
+      const tcloudConfig = {
+        url: 'https://mock.tobikodata.com',
+        org: 'test-org',
+        project: 'test-project',
+      }
+      await fs.writeFile(
+        path.join(tempDir, 'tcloud.yaml'),
+        `url: ${tcloudConfig.url}\norg: ${tcloudConfig.org}\nproject: ${tcloudConfig.project}\n`,
+      )
+
+      // Write mock ".tcloud_auth_state.json" file
+      await setupAuthenticatedState(tempDir)
+
+      // Set up Python environment with mock tcloud and sqlmesh
+      const pythonPath = await setupPythonEnvironment(pythonEnvDir)
+
+      // Configure VS Code settings to use our Python environment
+      const settings = {
+        'python.defaultInterpreterPath': pythonPath,
+        'sqlmesh.environmentPath': pythonEnvDir,
+      }
+      await fs.ensureDir(path.join(tempDir, '.vscode'))
+      await fs.writeJson(
+        path.join(tempDir, '.vscode', 'settings.json'),
+        settings,
+        { spaces: 2 },
+      )
+
+      // Start VS Code
+      const { window, close } = await startVSCode(tempDir)
+
+      // Open a SQL file to trigger SQLMesh activation
+      // Wait for the models folder to be visible
+      await window.waitForSelector('text=models')
+
+      // Click on the models folder
+      await window
+        .getByRole('treeitem', { name: 'models', exact: true })
+        .locator('a')
+        .click()
+
+      // Open the top_waiters model
+      await window
+        .getByRole('treeitem', { name: 'customers.sql', exact: true })
+        .locator('a')
+        .click()
+
+      await window.waitForSelector('text=Installing enterprise python package')
+      expect(
+        await window.locator('text=Installing enterprise python package'),
+      ).toHaveCount(2)
+
+      await window.waitForSelector('text=Loaded SQLMesh context')
+
+      // Close VS Code
+      await close()
+    } finally {
+      // Clean up
+      await fs.remove(tempDir)
+    }
+  })
+})
diff --git a/vscode/extension/tests/tcloud/README.md b/vscode/extension/tests/tcloud/README.md
new file mode 100644
index 00000000..c3c723be
--- /dev/null
+++ b/vscode/extension/tests/tcloud/README.md
@@ -0,0 +1,53 @@
+# Mock tcloud CLI for Testing
+
+This directory contains a mock implementation of the tcloud CLI for testing the VSCode extension.
+
+## Implemented Commands
+
+The mock implements only the commands used by the VSCode extension:
+
+### Authentication Commands
+- `tcloud auth vscode status` - Returns authentication status
+- `tcloud auth vscode login-url` - Returns mock OAuth login URL
+- `tcloud auth vscode start-server <verifier_code>` - Simulates OAuth callback
+- `tcloud auth vscode device` - Returns mock device flow info
+- `tcloud auth vscode poll_device <device_code>` - Simulates device flow success
+- `tcloud auth logout` - Clears authentication state
+
+### SQLMesh Commands
+- `tcloud is_sqlmesh_installed` - Checks installation status
+- `tcloud install_sqlmesh` - Marks SQLMesh as installed
+- `tcloud sqlmesh <args>` - Echoes sqlmesh commands
+
+## State Management
+
+The mock maintains state in two files:
+- `.tcloud_auth_state.json` - Authentication state (logged in/out, ID token)
+- `.sqlmesh_installed` - SQLMesh installation marker
+
+## Usage in Tests
+
+To use this mock in tests:
+
+1. Ensure the mock is in PATH or reference it directly
+2. The mock will simulate successful authentication flows
+3. State persists between calls for realistic testing
+
+## Example
+
+```bash
+# Check auth status
+./tcloud auth vscode status
+# Output: {"is_logged_in": false, "id_token": null}
+
+# Simulate login
+./tcloud auth vscode login-url
+# Output: {"url": "https://mock-auth.example.com/auth?client_id=mock&redirect_uri=http://localhost:7890", "verifier_code": "mock_verifier_12345"}
+
+./tcloud auth vscode start-server mock_verifier_12345
+# Output: Mock server started successfully
+
+# Check status again
+./tcloud auth vscode status
+# Output: {"is_logged_in": true, "id_token": {"email": "test@example.com", "name": "Test User", "exp": 1736790123}}
+```
\ No newline at end of file
diff --git a/vscode/extension/tests/tcloud/mock_tcloud/__init__.py b/vscode/extension/tests/tcloud/mock_tcloud/__init__.py
new file mode 100644
index 00000000..98ad152e
--- /dev/null
+++ b/vscode/extension/tests/tcloud/mock_tcloud/__init__.py
@@ -0,0 +1,1 @@
+# Mock tcloud package
\ No newline at end of file
diff --git a/vscode/extension/tests/tcloud/mock_tcloud/cli.py b/vscode/extension/tests/tcloud/mock_tcloud/cli.py
new file mode 100755
index 00000000..b9b80ab8
--- /dev/null
+++ b/vscode/extension/tests/tcloud/mock_tcloud/cli.py
@@ -0,0 +1,257 @@
+"""
+Mock tcloud CLI for testing VSCode extension.
+Implements only the commands used by the extension.
+"""
+
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import click
+
+def get_auth_state_file():
+    """Get the path to the auth state file in the current working directory"""
+    return Path.cwd() / ".tcloud_auth_state.json"
+
+
+def load_auth_state():
+    """Load authentication state from file"""
+    auth_file = get_auth_state_file()
+    if auth_file.exists():
+        with open(auth_file, "r") as f:
+            return json.load(f)
+    return {"is_logged_in": False, "id_token": None}
+
+
+def save_auth_state(state):
+    """Save authentication state to file"""
+    auth_file = get_auth_state_file()
+    with open(auth_file, "w") as f:
+        json.dump(state, f)
+
+
+@click.group(no_args_is_help=True)
+@click.option(
+    "--project",
+    type=str,
+    help="The name of the project.",
+)
+@click.pass_context
+def cli(ctx: click.Context, project: str) -> None:
+    """Mock Tobiko Cloud CLI"""
+    ctx.ensure_object(dict)
+    ctx.obj["project"] = project
+
+
+@cli.command("is_sqlmesh_installed", hidden=True)
+@click.pass_context
+def is_sqlmesh_installed(ctx: click.Context) -> None:
+    """Check if SQLMesh Enterprise is installed"""
+    # For testing, we'll track installation state in a file in the current bin directory
+    # This matches where the test expects it to be
+    bin_dir = Path(sys.executable).parent
+    install_state_file = bin_dir / ".sqlmesh_installed"
+    is_installed = install_state_file.exists()
+
+    print(
+        json.dumps(
+            {
+                "is_installed": is_installed,
+            }
+        )
+    )
+
+
+@cli.command("install_sqlmesh")
+@click.pass_context
+def install_sqlmesh(ctx: click.Context) -> None:
+    """Install the correct version of SQLMesh Enterprise"""
+
+    # For 3 seconds output to stdout
+    for i in range(3):
+        print(f"Installing SQLMesh Enterprise logs {i + 1}/3", flush=True)
+        time.sleep(1)
+
+    # Simulate installation by creating a marker file in the bin directory
+    bin_dir = Path(sys.executable).parent
+    install_state_file = bin_dir / ".sqlmesh_installed"
+    install_state_file.touch()
+
+    print("Mock SQLMesh Enterprise installed successfully")
+
+
+@cli.command("sqlmesh")
+@click.argument("args", nargs=-1)
+@click.pass_context
+def sqlmesh(ctx: click.Context, args) -> None:
+    """Run SQLMesh Enterprise commands"""
+    # Pass through to the real sqlmesh command
+
+    # Get the path to sqlmesh in the same environment as this script
+    bin_dir = os.path.dirname(sys.executable)
+    sqlmesh_path = os.path.join(bin_dir, "sqlmesh")
+
+    if not os.path.exists(sqlmesh_path):
+        # Try with .exe extension on Windows
+        sqlmesh_path = os.path.join(bin_dir, "sqlmesh.exe")
+
+    if not os.path.exists(sqlmesh_path):
+        # Fall back to using sqlmesh from PATH
+        sqlmesh_path = "sqlmesh"
+
+    # Execute the real sqlmesh with the provided arguments
+    result = subprocess.run([sqlmesh_path] + list(args), capture_output=False)
+    sys.exit(result.returncode)
+
+
+@click.group()
+def auth() -> None:
+    """
+    Tobiko Cloud Authentication
+    """
+
+
+@auth.command()
+def logout() -> None:
+    """Logout of any current session"""
+    save_auth_state({"is_logged_in": False, "id_token": None})
+    print("Logged out successfully")
+
+
+### Methods for VSCode
+@auth.group(hidden=True)
+def vscode() -> None:
+    """Commands for VSCode integration"""
+    pass
+
+
+@vscode.command("login-url")
+def login_url() -> None:
+    """
+    Login to Tobiko Cloud.
+
+    This returns a JSON object with the following fields:
+    - url: The URL to login open
+    """
+    # Return mock OAuth URL and verifier
+    print(
+        json.dumps(
+            {
+                "url": "https://mock-auth.example.com/auth?client_id=mock&redirect_uri=http://localhost:7890",
+                "verifier_code": "mock_verifier_12345",
+            }
+        )
+    )
+
+
+@vscode.command("start-server")
+@click.argument("code_verifier", type=str, required=True)
+def start_server(code_verifier: str) -> None:
+    """
+    Start the server to catch the redirect from the browser.
+    """
+    # Simulate successful authentication after a short delay
+    time.sleep(0.5)
+
+    # Update auth state to logged in
+    save_auth_state(
+        {
+            "is_logged_in": True,
+            "id_token": {
+                "iss": "https://mock.tobikodata.com",
+                "aud": "mock-audience",
+                "sub": "user-123",
+                "scope": "openid email profile",
+                "iat": int(time.time()),
+                "exp": int(time.time()) + 3600,  # Token expires in 1 hour
+                "email": "test@example.com",
+                "name": "Test User",
+            },
+        }
+    )
+
+    # The real command would start a server, but for testing we just simulate success
+    print("Mock server started successfully")
+
+
+@vscode.command("status")
+def vscode_status() -> None:
+    """
+    Auth status for logged in
+    """
+    state = load_auth_state()
+    print(
+        json.dumps(
+            {"is_logged_in": state["is_logged_in"], "id_token": state["id_token"]}
+        )
+    )
+
+
+@vscode.command("device")
+def vscode_device() -> None:
+    """
+    Initiate device flow for VSCode integration
+    """
+    print(
+        json.dumps(
+            {
+                "device_code": "MOCK-DEVICE-CODE",
+                "user_code": "ABCD-1234",
+                "verification_uri": "https://mock-auth.example.com/device",
+                "verification_uri_complete": "https://mock-auth.example.com/device?user_code=ABCD-1234",
+                "expires_in": 600,
+                "interval": 5,
+            }
+        )
+    )
+
+
+@vscode.command("poll_device")
+@click.argument("device_code", type=str, required=True)
+@click.option(
+    "-i",
+    "--interval",
+    type=int,
+    default=5,
+    help="The interval between polling attempts in seconds",
+)
+@click.option(
+    "-t",
+    "--timeout",
+    type=int,
+    default=300,
+    help="The timeout for the device flow in seconds",
+)
+def vscode_poll_device(device_code: str, interval: int, timeout: int) -> None:
+    """
+    Poll the device flow for VSCode integration
+    """
+    # For testing, we'll just succeed immediately
+    save_auth_state(
+        {
+            "is_logged_in": True,
+            "id_token": {
+                "iss": "https://mock.tobikodata.com",
+                "aud": "mock-audience",
+                "sub": "device-user-123",
+                "scope": "openid email profile",
+                "iat": int(time.time()),
+                "exp": int(time.time()) + 3600,
+                "email": "device@example.com",
+                "name": "Device User",
+            },
+        }
+    )
+
+    print(json.dumps({"success": True}))
+
+
+# Add auth group to main CLI
+cli.add_command(auth)
+
+
+if __name__ == "__main__":
+    cli()
diff --git a/vscode/extension/tests/tcloud/pyproject.toml b/vscode/extension/tests/tcloud/pyproject.toml
new file mode 100644
index 00000000..505af5c7
--- /dev/null
+++ b/vscode/extension/tests/tcloud/pyproject.toml
@@ -0,0 +1,18 @@
+[build-system]
+requires = ["setuptools>=61.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "mock-tcloud"
+version = "0.1.0"
+description = "Mock tcloud CLI for testing VSCode extension"
+requires-python = ">=3.8"
+dependencies = [
+    "click>=8.0",
+]
+
+[project.scripts]
+tcloud = "mock_tcloud.cli:cli"
+
+[tool.setuptools]
+packages = ["mock_tcloud"]
\ No newline at end of file
diff --git a/vscode/extension/tests/utils.ts b/vscode/extension/tests/utils.ts
index 1a2c55e1..7b722c9e 100644
--- a/vscode/extension/tests/utils.ts
+++ b/vscode/extension/tests/utils.ts
@@ -2,6 +2,8 @@ import path from 'path'
 import fs from 'fs-extra'
 import os from 'os'
 import { _electron as electron, Page } from '@playwright/test'
+import { exec } from 'child_process'
+import { promisify } from 'util'
 
 // Absolute path to the VS Code executable you downloaded in step 1.
 export const VS_CODE_EXE = fs.readJsonSync(
@@ -18,6 +20,7 @@ export const SUSHI_SOURCE_PATH = path.join(
   'examples',
   'sushi',
 )
+export const REPO_ROOT = path.join(__dirname, '..', '..', '..')
 
 /**
  * Launch VS Code and return the window and a function to close the app.
@@ -86,3 +89,73 @@ export const clickExplorerTab = async (page: Page): Promise<void> => {
     await page.locator("text='Explorer'").waitFor({ state: 'visible' })
   }
 }
+
+const execAsync = promisify(exec)
+
+export interface PythonEnvironment {
+  pythonPath: string
+  pipPath: string
+}
+
+/**
+ * Create a virtual environment in the given directory.
+ * @param venvDir The directory to create the virtual environment in.
+ */
+export const createVirtualEnvironment = async (
+  venvDir: string,
+): Promise<PythonEnvironment> => {
+  const pythonCmd = process.platform === 'win32' ? 'python' : 'python3'
+  const { stderr } = await execAsync(`${pythonCmd} -m venv "${venvDir}"`)
+  if (stderr && !stderr.includes('WARNING')) {
+    throw new Error(`Failed to create venv: ${stderr}`)
+  }
+  // Get paths
+  const isWindows = process.platform === 'win32'
+  const binDir = path.join(venvDir, isWindows ? 'Scripts' : 'bin')
+  const pythonPath = path.join(binDir, isWindows ? 'python.exe' : 'python')
+  const pipPath = path.join(binDir, isWindows ? 'pip.exe' : 'pip')
+
+  return {
+    pythonPath,
+    pipPath,
+  }
+}
+
+/**
+ * Install packages in the given virtual environment.
+ * @param pythonDetails The Python environment to use.
+ * @param packagePaths The paths to the packages to install (string[]).
+ */
+export const pipInstall = async (
+  pythonDetails: PythonEnvironment,
+  packagePaths: string[],
+): Promise<void> => {
+  const { pipPath } = pythonDetails
+  const execString = `"${pipPath}" install -e "${packagePaths.join('" -e "')}"`
+  const { stderr } = await execAsync(execString)
+  if (stderr && !stderr.includes('WARNING') && !stderr.includes('notice')) {
+    throw new Error(`Failed to install package: ${stderr}`)
+  }
+}
+
+/**
+ * Open the lineage view in the given window.
+ */
+export const openLineageView = async (window: Page): Promise<void> => {
+  await window.keyboard.press(
+    process.platform === 'darwin' ? 'Meta+Shift+P' : 'Control+Shift+P',
+  )
+  await window.keyboard.type('Lineage: Focus On View')
+  await window.keyboard.press('Enter')
+}
+
+/**
+ * Restart the SQLMesh servers
+ */
+export const restartSqlmeshServers = async (window: Page): Promise<void> => {
+  await window.keyboard.press(
+    process.platform === 'darwin' ? 'Meta+Shift+P' : 'Control+Shift+P',
+  )
+  await window.keyboard.type('Restart SQLMesh servers')
+  await window.keyboard.press('Enter')
+}