Fix depends_on_past for manual/asset runs when logical_date is NULL

airflow-core/src/airflow/models/dag.py

@@ -252,14 +252,15 @@ def get_last_dagrun(dag_id: str, session: Session, include_manually_triggered: b
     """
     Return the last dag run for a dag, None if there was none.
 
-    Last dag run can be any type of run e.g. scheduled or backfilled.
-    Overridden DagRuns are ignored.
+    Last dag run can be any type of run (e.g. scheduled, manual, asset-triggered).
+    Overridden DagRuns are ignored (AIP-39).
     """
     DR = DagRun
-    query = select(DR).where(DR.dag_id == dag_id, DR.logical_date.is_not(None))
+    query = select(DR).where(DR.dag_id == dag_id)
     if not include_manually_triggered:
         query = query.where(DR.run_type != DagRunType.MANUAL)
-    query = query.order_by(DR.logical_date.desc())
+    # Order by logical_date if available, otherwise fallback to run_after (AIP-39)
+    query = query.order_by(func.coalesce(DR.logical_date, DR.run_after).desc())
     return session.scalar(query.limit(1))
 
 

airflow-core/src/airflow/models/dagrun.py

@@ -1038,21 +1038,32 @@ def get_previous_dagrun(
         dag_run: DagRun, state: DagRunState | None = None, *, session: Session = NEW_SESSION
     ) -> DagRun | None:
         """
-        Return the previous DagRun, if there is one.
+        Return the previous DagRun, if there is one (AIP-39).
 
         :param dag_run: the dag run
         :param session: SQLAlchemy ORM Session
         :param state: the dag run state
         """
-        if not dag_run or dag_run.logical_date is None:
+        if not dag_run:
             return None
-        filters = [
-            DagRun.dag_id == dag_run.dag_id,
-            DagRun.logical_date < dag_run.logical_date,
-        ]
+
+        filters = [DagRun.dag_id == dag_run.dag_id]
         if state is not None:
             filters.append(DagRun.state == state)
-        return session.scalar(select(DagRun).where(*filters).order_by(DagRun.logical_date.desc()).limit(1))
+
+        # Use (run_after, id) to correctly order runs when run_after values are equal
+        # (e.g. two manual runs triggered at the same time, or a scheduled run whose
+        # run_after equals the next run's run_after).  id is a monotonically-increasing
+        # surrogate key so it gives a stable, deterministic tiebreak.
+        filters.append(
+            or_(
+                DagRun.run_after < dag_run.run_after,
+                and_(DagRun.run_after == dag_run.run_after, DagRun.id < dag_run.id),
+            )
+        )
+        return session.scalar(
+            select(DagRun).where(*filters).order_by(DagRun.run_after.desc(), DagRun.id.desc()).limit(1)
+        )
 
     @staticmethod
     @provide_session

airflow-core/src/airflow/ti_deps/deps/prev_dagrun_dep.py

@@ -19,7 +19,7 @@
 
 from typing import TYPE_CHECKING
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, literal, or_, select
 
 from airflow.models.backfill import BackfillDagRun
 from airflow.models.dagrun import DagRun
@@ -75,12 +75,30 @@ def _has_any_prior_tis(ti: TI, *, session: Session) -> bool:
 
         This function exists for easy mocking in tests.
         """
-        query = exists_query(
-            TI.dag_id == ti.dag_id,
-            TI.task_id == ti.task_id,
-            TI.logical_date < ti.logical_date,
-            session=session,
-        )
+        if ti.logical_date is not None:
+            query = exists_query(
+                TI.dag_id == ti.dag_id,
+                TI.task_id == ti.task_id,
+                TI.logical_date < ti.logical_date,
+                session=session,
+            )
+        else:
+            # Fallback to run_after for manual/asset runs (AIP-39)
+            dr = ti.get_dagrun(session=session)
+            query = (
+                session.scalar(
+                    select(literal(1))
+                    .select_from(TI)
+                    .join(DagRun, TI.run_id == DagRun.run_id)
+                    .where(
+                        TI.dag_id == ti.dag_id,
+                        TI.task_id == ti.task_id,
+                        DagRun.run_after < dr.run_after,
+                    )
+                    .limit(1)
+                )
+                is not None
+            )
         return query
 
     @staticmethod

airflow-core/tests/unit/ti_deps/deps/test_prev_dagrun_dep.py

@@ -96,6 +96,50 @@ def test_first_task_run_of_new_task(self, testing_dag_bundle):
             assert dep.is_met(ti=ti, dep_context=dep_context)
             mock_has_any_prior_tis.assert_called_once_with(ti, session=ANY)
 
+    def test_prev_dagrun_with_manual_run_logical_date_null(self, session):
+        dag = DAG(
+            "test_depends_on_past_with_null_logical_date",
+            schedule=None,
+            start_date=START_DATE,
+        )
+        task = BaseOperator(
+            task_id="test_task",
+            dag=dag,
+            depends_on_past=True,
+            start_date=START_DATE,
+            wait_for_downstream=False,
+        )
+        scheduler_dag = sync_dag_to_db(dag, session=session)
+
+        first_run_after = convert_to_utc(datetime(2016, 1, 1))
+        first_dr = scheduler_dag.create_dagrun(
+            run_id="manual__2016-01-01T00:00:00+00:00",
+            state=DagRunState.RUNNING,
+            logical_date=None,
+            run_type=DagRunType.MANUAL,
+            data_interval=None,
+            run_after=first_run_after,
+            triggered_by=DagRunTriggeredByType.TEST,
+        )
+        first_ti = first_dr.get_task_instance(task.task_id, session=session)
+        first_ti.set_state(TaskInstanceState.FAILED, session=session)
+
+        second_run_after = convert_to_utc(datetime(2016, 1, 2))
+        second_dr = scheduler_dag.create_dagrun(
+            run_id="manual__2016-01-02T00:00:00+00:00",
+            state=DagRunState.RUNNING,
+            logical_date=None,
+            run_type=DagRunType.MANUAL,
+            data_interval=None,
+            run_after=second_run_after,
+            triggered_by=DagRunTriggeredByType.TEST,
+        )
+        second_ti = second_dr.get_task_instance(task.task_id, session=session)
+        second_ti.task = task
+
+        dep_context = DepContext(ignore_depends_on_past=False)
+        assert not PrevDagrunDep().is_met(ti=second_ti, dep_context=dep_context)
+
 
 @pytest.mark.parametrize(
     "kwargs",