AIP-103: Passing Asset name/uri and AssetStateAccessor through to BaseEventTrigger

airflow-core/src/airflow/executors/workloads/trigger.py

@@ -46,3 +46,6 @@ class RunTrigger(BaseModel):
     dag_run_data: dict | None = (
         None  # Serialized DagRun data in dict format so it can be deserialized in trigger subprocess.
     )
+
+    # name: uri of all "watched" Assets
+    watched_assets: dict[str, str] | None = None  # Set for BaseEventTrigger asset watchers only

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -737,11 +737,18 @@ def _create_workload(
         render_log_fname: Callable[..., str],
         session: Session,
     ) -> workloads.RunTrigger | None:
+        # Pass the "watched" Assets through for downstream use in BaseEventTrigger
         if trigger.task_instance is None:
+            watched_assets: dict[str, str] | None = None
+
+            if trigger.asset_watchers:
+                watched_assets = {a.name: a.uri for a in trigger.assets}
+
             return workloads.RunTrigger(
                 id=trigger.id,
                 classpath=trigger.classpath,
                 encrypted_kwargs=trigger.encrypted_kwargs,
+                watched_assets=watched_assets,
             )
 
         if not trigger.task_instance.dag_version_id:
@@ -1262,6 +1269,16 @@ async def create_triggers(self):
             trigger_instance.triggerer_job_id = self.job_id
             trigger_instance.timeout_after = workload.timeout_after
 
+            if isinstance(trigger_instance, BaseEventTrigger) and workload.watched_assets:
+                # Reconstruct AssetStateAccessors from watched_assets
+                from airflow.sdk.definitions.asset import Asset
+                from airflow.sdk.execution_time.context import AssetStateAccessors
+
+                # Potentially address Asset vs. AssetRef, AssetUriRef, etc.
+                trigger_instance.asset_state = AssetStateAccessors(
+                    inlets=[Asset(name=name, uri=uri) for name, uri in workload.watched_assets.items()]
+                )
+
             self.triggers[trigger_id] = {
                 "task": asyncio.create_task(
                     self.run_trigger(trigger_id, trigger_instance, workload.timeout_after, context),

airflow-core/src/airflow/triggers/base.py

@@ -33,6 +33,7 @@
 )
 
 from airflow.sdk.definitions._internal.templater import Templater
+from airflow.sdk.execution_time.context import AssetStateAccessors
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.state import TaskInstanceState
 
@@ -297,6 +298,12 @@ class BaseEventTrigger(BaseTrigger):
 
     supports_triggerer_queue: bool = False
 
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        # Injected by the triggerer before run() is called; mirrors how trigger_id is set
+        self.asset_state: AssetStateAccessors | None = None
+
     @staticmethod
     def hash(classpath: str, kwargs: dict[str, Any]) -> int:
         """