From 149fffe3174906dd893a83c95e115cda9cf72e66 Mon Sep 17 00:00:00 2001
From: Vincent <97131062+vincbeck@users.noreply.github.com>
Date: Wed, 17 Jun 2026 12:32:58 -0400
Subject: [PATCH] [v3-3-test] Fix triggerer crash when trigger subclass does
 not call `super().__init__()` (#68636) (cherry picked from commit
 fa273d83075f7c78843c67cb7ac550f12026ec19)

Co-authored-by: Vincent <97131062+vincbeck@users.noreply.github.com>
---
 airflow-core/src/airflow/triggers/base.py     |  4 ++--
 .../tests/unit/triggers/test_base_trigger.py  | 19 +++++++++++++++++++
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/airflow-core/src/airflow/triggers/base.py b/airflow-core/src/airflow/triggers/base.py
index 6f6061e3366f8..f587fd8de3ff4 100644
--- a/airflow-core/src/airflow/triggers/base.py
+++ b/airflow-core/src/airflow/triggers/base.py
@@ -101,8 +101,8 @@ def task(self) -> Operator | None:
         return None
 
     @property
-    def task_instance(self) -> TaskInstance:
-        return self._task_instance
+    def task_instance(self) -> TaskInstance | None:
+        return getattr(self, "_task_instance", None)
 
     @task_instance.setter
     def task_instance(self, value: TaskInstance | None) -> None:
diff --git a/airflow-core/tests/unit/triggers/test_base_trigger.py b/airflow-core/tests/unit/triggers/test_base_trigger.py
index 5f32cb23bfe41..429e54615293a 100644
--- a/airflow-core/tests/unit/triggers/test_base_trigger.py
+++ b/airflow-core/tests/unit/triggers/test_base_trigger.py
@@ -140,6 +140,25 @@ def test_render_template_fields_empty_when_no_trigger_kwargs(create_task_instanc
     assert trigger.name == "Hello {{ name }}"
 
 
+class _TriggerWithoutSuperInit(BaseTrigger):
+    """A trigger that does not call super().__init__() — simulates third-party triggers."""
+
+    def __init__(self, queue_url: str):
+        self.queue_url = queue_url
+
+    def serialize(self):
+        return (f"{type(self).__module__}.{type(self).__qualname__}", {"queue_url": self.queue_url})
+
+    async def run(self):
+        yield TriggerEvent({"queue_url": self.queue_url})
+
+
+def test_task_instance_property_works_without_super_init():
+    """task_instance property must return None when subclass skips super().__init__()."""
+    trigger = _TriggerWithoutSuperInit(queue_url="https://sqs.example.com/queue")
+    assert trigger.task_instance is None
+
+
 class _PlainEventTrigger(BaseEventTrigger):
     """A BaseEventTrigger that does not opt into shared streams."""