Warn when a deadline alert cannot be scheduled for a Dag run

airflow-core/src/airflow/serialization/definitions/dag.py

@@ -789,6 +789,15 @@ def _process_dagrun_deadline_alerts(
                         "deadline_alerts.deadline_created",
                         tags=prune_dict({"dag_id": self.dag_id, "team_name": team_name}),
                     )
+                else:
+                    # A reference returns None when the Dag run attribute it reads is unset (e.g.
+                    # logical_date on a run triggered without one); without this the alert never fires.
+                    log.warning(
+                        "deadline alert skipped, reference produced no deadline time",
+                        dag_id=self.dag_id,
+                        run_id=orm_dagrun.run_id,
+                        reference=type(deserialized_deadline_alert.reference).__name__,
+                    )
 
     @provide_session
     def set_task_instance_state(

airflow-core/tests/unit/models/test_dag.py

@@ -2448,6 +2448,39 @@ def test_dagrun_deadline(self, reference_type, reference_column, testing_dag_bun
         assert len(dr.deadlines) == 1
         assert dr.deadlines[0].deadline_time == getattr(dr, reference_column, DEFAULT_DATE) + interval
 
+    @pytest.mark.need_serialized_dag
+    def test_dagrun_deadline_skipped_when_reference_has_no_value(self, testing_dag_bundle, session, caplog):
+        dag = DAG(
+            dag_id="test_deadline_without_logical_date",
+            schedule=None,
+            deadline=DeadlineAlert(
+                reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
+                interval=datetime.timedelta(hours=1),
+                callback=AsyncCallback(empty_callback_for_deadline),
+            ),
+        )
+
+        scheduler_dag = sync_dag_to_db(dag, session=session)
+
+        dr = scheduler_dag.create_dagrun(
+            run_id="test_dagrun_deadline_without_logical_date",
+            run_type=DagRunType.MANUAL,
+            state=State.QUEUED,
+            logical_date=None,
+            run_after=TEST_DATE,
+            triggered_by=DagRunTriggeredByType.TEST,
+        )
+        session.flush()
+        dr = session.merge(dr)
+
+        assert dr.deadlines == []
+        assert {
+            "event": "deadline alert skipped, reference produced no deadline time",
+            "dag_id": "test_deadline_without_logical_date",
+            "run_id": "test_dagrun_deadline_without_logical_date",
+            "reference": "DagRunLogicalDateDeadline",
+        } in caplog
+
     def test_dag_with_multiple_deadlines(self, testing_dag_bundle, session):
         """Test that a Dag with multiple deadlines stores all deadlines and persists on re-serialization."""
         deadlines = [