Fix seed PR backfill interval overrides

Signed-off-by: eakmanrq <6326532+eakmanrq@users.noreply.github.com>

Author: eakmanrq

sqlmesh/core/context.py

@@ -1666,6 +1666,7 @@ def plan_builder(
             # This ensures that no models outside the impacted sub-DAG(s) will be backfilled unexpectedly.
             backfill_models = modified_model_names or None
 
+        plan_execution_time = execution_time or now()
         max_interval_end_per_model = None
         default_start, default_end = None, None
         if not run:
@@ -1680,17 +1681,31 @@ def plan_builder(
                 max_interval_end_per_model,
                 backfill_models,
                 modified_model_names,
-                execution_time or now(),
+                plan_execution_time,
             )
 
+            if (
+                start
+                and default_end
+                and to_datetime(start, relative_base=to_datetime(plan_execution_time))
+                > to_datetime(default_end)
+            ):
+                # If the requested start is newer than prod's latest interval end, fall back to execution time
+                # instead of forcing an invalid [start, default_end] range.
+                default_start, default_end = None, None
+
             # Refresh snapshot intervals to ensure that they are up to date with values reflected in the max_interval_end_per_model.
             self.state_sync.refresh_snapshot_intervals(context_diff.snapshots.values())
+            max_interval_end_per_model = self._filter_stale_end_overrides(
+                max_interval_end_per_model,
+                context_diff.snapshots_by_name,
+            )
 
         start_override_per_model = self._calculate_start_override_per_model(
             min_intervals,
             start or default_start,
             end or default_end,
-            execution_time or now(),
+            plan_execution_time,
             backfill_models,
             snapshots,
             max_interval_end_per_model,
@@ -3181,6 +3196,20 @@ def _get_max_interval_end_per_model(
             ).items()
         }
 
+    @staticmethod
+    def _filter_stale_end_overrides(
+        max_interval_end_per_model: t.Dict[str, datetime],
+        snapshots_by_name: t.Dict[str, Snapshot],
+    ) -> t.Dict[str, datetime]:
+        # Drop stale interval ends for snapshots whose new versions have no intervals yet. Otherwise the old
+        # prod end is reused as an end_override, causing missing_intervals() to skip the new snapshot entirely
+        # when the requested start is newer than that stale end.
+        return {
+            model_fqn: end
+            for model_fqn, end in max_interval_end_per_model.items()
+            if model_fqn not in snapshots_by_name or snapshots_by_name[model_fqn].intervals
+        }
+
     @staticmethod
     def _get_models_for_interval_end(
         snapshots: t.Dict[str, Snapshot], backfill_models: t.Set[str]

tests/core/test_context.py

@@ -1223,6 +1223,42 @@ def test_plan_seed_model_excluded_from_default_end(copy_to_temp_path: t.Callable
         context.close()
 
 
+@pytest.mark.slow
+def test_seed_model_pr_plan_with_stale_prod_intervals(copy_to_temp_path: t.Callable):
+    path = copy_to_temp_path("examples/sushi")
+
+    with time_machine.travel("2024-06-01 00:00:00 UTC"):
+        context = Context(paths=path, gateway="duckdb_persistent")
+        context.plan("prod", no_prompts=True, auto_apply=True)
+        context.close()
+
+    with time_machine.travel("2026-04-13 00:00:00 UTC"):
+        context = Context(paths=path, gateway="duckdb_persistent")
+
+        model = context.get_model("sushi.waiter_names").copy()
+        model.seed.content += "10,Trey\n"
+        context.upsert_model(model)
+
+        plan = context.plan("dev", start="2 months ago", no_prompts=True)
+        missing_interval_names = {si.snapshot_id.name for si in plan.missing_intervals}
+
+        assert plan.user_provided_flags == {"start": "2 months ago"}
+        assert plan.provided_end is None
+        assert to_timestamp(plan.start) == to_timestamp("2026-02-13")
+        assert to_timestamp(plan.end) == to_timestamp("2026-04-13")
+        assert any("waiter_names" in name for name in missing_interval_names)
+        assert any("waiter_as_customer_by_day" in name for name in missing_interval_names)
+
+        context.apply(plan)
+
+        promoted_snapshot_names = {
+            snapshot.name for snapshot in context.state_sync.get_environment("dev").promoted_snapshots
+        }
+        assert any("waiter_names" in name for name in promoted_snapshot_names)
+        assert any("waiter_as_customer_by_day" in name for name in promoted_snapshot_names)
+        context.close()
+
+
 @pytest.mark.slow
 def test_schema_error_no_default(sushi_context_pre_scheduling) -> None:
     context = sushi_context_pre_scheduling