fix(api): scope NULL branch in NullableDatetimeRangeFilter for start_date lower bounds [fj4WqyCCw3C5ShR1RfB7MoBPTpkRrBFYP1uT35g3MvT]

airflow-core/src/airflow/api_fastapi/common/parameters.py

@@ -1197,8 +1197,16 @@ class NullableDatetimeRangeFilter(RangeFilter):
     started/ended task will eventually satisfy any past lower bound. For upper bounds the NULL branch
     is ``col IS NULL AND now() <= x``, preserving the COALESCE(col, now()) semantics without the
     function-wrap index penalty.
+
+    When *extra_null_condition* is provided (e.g. ``end_date.is_(None)`` for ``start_date`` lower
+    bounds), the NULL branch also requires that condition — preventing terminal rows whose column
+    happens to be ``NULL`` from matching every lower-bound filter indefinitely.
     """
 
+    def __init__(self, range_val, attr, extra_null_condition=None):
+        super().__init__(range_val, attr)
+        self.extra_null_condition = extra_null_condition
+
     def to_orm(self, select: Select) -> Select:
         if self.skip_none is False:
             raise ValueError(f"Cannot set 'skip_none' to False on a {type(self)}")
@@ -1208,10 +1216,16 @@ def to_orm(self, select: Select) -> Select:
 
         if self.value.lower_bound_gte:
             x = self.value.lower_bound_gte
-            select = select.where(or_(self.attribute >= x, self.attribute.is_(None)))
+            null_condition = self.attribute.is_(None)
+            if self.extra_null_condition is not None:
+                null_condition = and_(null_condition, self.extra_null_condition)
+            select = select.where(or_(self.attribute >= x, null_condition))
         if self.value.lower_bound_gt:
             x = self.value.lower_bound_gt
-            select = select.where(or_(self.attribute > x, self.attribute.is_(None)))
+            null_condition = self.attribute.is_(None)
+            if self.extra_null_condition is not None:
+                null_condition = and_(null_condition, self.extra_null_condition)
+            select = select.where(or_(self.attribute > x, null_condition))
         if self.value.upper_bound_lte:
             x = self.value.upper_bound_lte
             select = select.where(or_(self.attribute <= x, and_(self.attribute.is_(None), func.now() <= x)))
@@ -1238,7 +1252,12 @@ def depends_datetime(
             upper_bound_lte=upper_bound_lte,
             upper_bound_lt=upper_bound_lt,
         )
-        if filter_name in ("start_date", "end_date"):
+        if filter_name == "start_date":
+            return NullableDatetimeRangeFilter(
+                range_val, attr,
+                extra_null_condition=getattr(model, "end_date").is_(None),
+            )
+        if filter_name == "end_date":
             return NullableDatetimeRangeFilter(range_val, attr)
         return RangeFilter(range_val, attr)
 

airflow-core/tests/unit/api_fastapi/common/test_parameters.py

@@ -520,11 +520,12 @@ def test_other_column_returns_plain_filter(self):
         assert type(rf) is RangeFilter
 
     def test_lower_bound_does_not_include_now(self):
-        """NULL branch on lower bounds passes unconditionally — no now() call."""
+        """NULL branch on lower bounds includes end_date IS NULL for start_date — no now() call."""
         bound = datetime(2026, 5, 3, 12, 0, 0, tzinfo=timezone.utc)
-        rf = _make_datetime_filter("start_date", lower_bound_gte=bound)
-        sql = _compile(rf.to_orm(select(TaskInstance)))
-        assert "is null" in sql
+        rf = _make_datetime_filter("start_date", model=DagRun, lower_bound_gte=bound)
+        sql = _compile(rf.to_orm(select(DagRun)))
+        # The NULL branch should require end_date IS NULL alongside start_date IS NULL
+        assert "end_date" in sql and "is null" in sql.lower()
         assert "now()" not in sql
         assert "coalesce" not in sql
 
@@ -544,6 +545,17 @@ def test_no_coalesce_for_start_date(self):
         assert "coalesce" not in sql
 
 
+
+    def test_start_date_lower_bound_excludes_terminal_runs(self):
+        """Terminal dag runs with NULL start_date should NOT match start_date_gte."""
+        bound = datetime(2026, 5, 3, 12, 0, 0, tzinfo=timezone.utc)
+        rf = _make_datetime_filter("start_date", model=DagRun, lower_bound_gte=bound)
+        sql = _compile(rf.to_orm(select(DagRun)))
+        assert "end_date is not null" not in sql  # we want end_date IS NULL, not IS NOT NULL
+        assert "end_date" in sql and "is null" in sql.lower()
+        assert "start_date" in sql and "is null" in sql.lower()
+        assert "start_date" in sql and ">=" in sql
+
 class TestRegexParamFactory:
     """The regexp filter dependency must apply the query timeout itself (callers can't forget)."""