[hussein-awala]

Adds AssetEventSensor to the standard provider for Dags that need to wait for asset events between tasks. Basic use requires Airflow 3.2+. The partition_key, partition_key_regexp_pattern, and extra filters require Airflow 3.4+.

Pass an Asset, Asset.ref(...), or AssetAlias through asset=. The sensor declares the target as an inlet and reads events through the SDK inlet accessor.

Supply at least one query filter: after, before, partition_key, partition_key_regexp_pattern, or non-empty extra. Time bounds are optional when another filter is supplied. The sensor checks this requirement at construction and after template rendering, before it queries events.

The sensor supports ordering and query limits. expected_count and count_policy control the required count. An optional process_result callable transforms the events before counting. Successful sensors return the processed events through XCom.

after and before accept datetime values, ISO strings, Jinja templates, template callables, and any object with resolve(context). The built-in RunStart, RunAfter, LogicalDate, DataIntervalStart, and DataIntervalEnd helpers accept a signed offset. They use Airflow's existing template rendering. Custom resolvers need no inheritance, dataclass, or serialization methods.

The sensor guide compares the bounds in a table, including their behavior when tasks are cleared. It explains inclusive boundaries, timetable differences, and stable custom representations for Dag serialization. The PR also includes an example Dag and tests for custom resolvers, mapped sensors, and rescheduling.

Testing:

• All 152 sensor tests passed in Breeze after merging main at 2c560aad52.
• With released Airflow 3.2.2 and Task SDK 1.2.2, 139 tests passed. The 13 tests for Airflow 3.4+ filters were skipped.
• Fast and manual static checks passed, including provider mypy.
• The standard provider documentation built successfully in Breeze with the new bounds and comparison table.
• On the previous revision, broader checks from breeze verify passed: core database and non-database tests; provider database tests with Postgres; provider non-database tests; and documentation builds. The provider checks covered common.ai, common.compat, google, openlineage, and standard.
• On an earlier revision, live producer and consumer Dags exercised six sensor configurations. All eight consumer tasks succeeded and checked their XCom results. That live UI test has not been repeated for the new time-bound helpers.

Limitations:

• Regexp filtering requires a positive [api] regexp_query_timeout. The default value, 0, disables regexp queries.
• Mapped sensors need an explicit partial(inlets=[...]) declaration for the asset to appear in lineage. The runtime inlet workaround links to the core fix in Lazy-resolve inlet refs and refresh asset context #75751.
• The earlier live inlet UI test showed consuming tasks for concrete assets. Name references, URI references, and aliases executed successfully but lacked those UI links under the tested core behavior.
• The sensor supports poke and reschedule modes. It does not support deferrable execution.
• Custom bounds must return stable values after rescheduling. Custom objects also need a deterministic __repr__ for mapped arguments and older core serialization paths.

---

Was generative AI tooling used to co-author this PR?

• Yes

Generated-by: Cursor (Claude Opus 4.8) and Codex (GPT-6) following the guidelines

[Copilot]

Pull request overview

Adds a new AssetEventSensor and AssetEventTrigger to the Standard provider to wait for asset events matching filter criteria until an expected count is reached (including a deferrable mode that runs optional process_result processing in the triggerer). This integrates with the Task SDK’s InletEventsAccessor-based asset-event querying and is version-gated to Airflow 3.4+.

Changes:

• Introduce AssetEventSensor + AssetEventTrigger with support for filters (after/before, ordering/limit, partition_key, partition_key_regexp_pattern, extra) and expected_count semantics.
• Add provider/version compatibility flags (AIRFLOW_V_3_4_PLUS) and wire new modules into provider metadata.
• Add unit tests (including DB-backed harness via a SUPERVISOR_COMMS stand-in) for sensor/trigger behavior and deferrable flow.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
providers/standard/src/airflow/providers/standard/sensors/asset.py	New AssetEventSensor implementation (sync + deferrable), filter forwarding, process_result handling, and execute_complete.
providers/standard/src/airflow/providers/standard/triggers/asset.py	New AssetEventTrigger implementation + shared helpers for fetch/count/serialization.
providers/standard/src/airflow/providers/standard/exceptions.py	Add sensor-specific exception types for trigger event validation.
providers/standard/src/airflow/providers/standard/version_compat.py	Add AIRFLOW_V_3_4_PLUS feature gate for the provider.
providers/standard/src/airflow/providers/standard/get_provider_info.py	Register new sensor/trigger modules in provider info.
providers/standard/provider.yaml	Register new sensor/trigger modules in provider YAML metadata.
providers/standard/tests/unit/standard/conftest.py	Add DB-backed fixtures and SUPERVISOR_COMMS stand-in to exercise real SQL/filtering paths in unit tests.
providers/standard/tests/unit/standard/sensors/test_asset.py	Add sensor tests (poke semantics, process_result, deferrable short-circuit/defer behavior, filter forwarding).
providers/standard/tests/unit/standard/triggers/test_asset.py	Add trigger tests (serialization, polling loop, process_result import/behavior, DB-backed run).
devel-common/src/tests_common/test_utils/version_compat.py	Add AIRFLOW_V_3_4_PLUS for tests to skip appropriately.

[Lee-W]

Thanks! I'll take a deeper look, but I'd first like to understand why we need a sensor like this.

I thought we could always use a custom PartitionMapper to achieve something similar. Are there any use cases I'm missing? Thanks!

[hussein-awala]

Thanks! I'll take a deeper look, but I'd first like to understand why we need a sensor like this.

I thought we could always use a custom PartitionMapper to achieve something similar. Are there any use cases I'm missing? Thanks!

@Lee-W Good point, but they're at different layers. PartitionMapper is scheduler-side: it decides whether/how a run is created from partition-aligned events, and its only count gate (MinimumCount/WaitForAll) works over a declared partition-key window on a PartitionedAssetTimetable.

AssetEventSensor is a task-level primitive: it waits inside an already-running Dag (cron/manual/any trigger) until events matching arbitrary filters reach a count, then passes them downstream via XCom. Two things a mapper can't do:

1. Mid-Dag gating in a non-asset-scheduled Dag (e.g. wait for ≥3 events on today's partition before continuing).
2. Filter on extra/time-range and reuse the payload (e.g. wait for N events with extra={"status":"validated"}, then process_result them for downstream tasks), mappers only see partition_key.

We can say it's a wrapper for the inlet_events but with the sensor features (wait for the events, max wait, soft fail, ...)

So complementary, not redundant. Can add this to the docs if useful.

[Lee-W]

Thanks for answering in details. But I'm still not convinced this should be a task-level thing.

1. Why couldn't we just split them into two Dags and let the downstream one be asset-scheduled? If a Dag needs both a time and an asset schedule, we also have AssetOrTimeSchedule. And for the "wait for N" cases, we have RollupMapper with a WaitPolicy (MinimumCount / WaitForAll).
2. And why do we need extra here? partition_key is the field we designed to be the discriminator — it's the indexed one, while extra is free-form metadata the scheduler doesn't index. If a producer needs to distinguish events by something, that something belongs in the partition key, and a custom PartitionMapper can decode it into whatever shape the downstream Dag needs.

But my main concern is that this dependency is not known by Airflow. We can't schedule on it, and it doesn't show up in the lineage. The whole point of assets is that declaring a data dependency makes it visible to the platform. That works against what the feature is for.

[nathadfield]

@Lee-W We hit this exact tension migrating some pipelines to partitioned assets, a concrete data point.

Take a simple linear DAG: task A runs, then task B needs an asset produced by a different DAG before it can proceed, then task C runs. Today that's A -> ExternalTaskSensor -> B -> C, and A runs immediately regardless of whether the other DAG's data is ready. Forcing that dependency onto the DAG's own schedule would delay A too, a real regression, since A never needed it.

Splitting A into its own DAG doesn't help either: A, B, and C are one logical unit of work that should be run, observed, and retried together, and splitting them apart only serves the scheduler's DAG-level-only asset constraint, not any real independence between them. A task-level sensor between A and B is the right fit, and this shape isn't rare.

On the visibility concern: could the sensor also set inlets for the asset(s) it waits on? inlets/outlets already drive the lineage graph independent of scheduling, so the dependency stays visible to the platform without tying it to what triggers the DAG run.

[Leondon9]

I opened the original discussion behind this (#67375), so let me add the concrete case, since the "why task-level?" question is the crux.

The workload is a time-scheduled batch consumer (@daily, etc.) whose SQL is written against its own data_interval_start / _end. Mid-run it needs to wait for an upstream asset partition (e.g. orders / dt=<interval>) produced by a different Dag on its own cadence.

Why the existing mechanisms don't quite cover it:

• Asset-schedule the consumer / AssetOrTimeSchedule: asset-triggered runs get a degenerate data interval — DataInterval.exact(run_after), pinned to the trigger instant — not the window the consumer's SQL is built on. Moving the wait onto the schedule changes the run's contract, even though the consumer still wants its own time interval.
• RollupMapper + WaitPolicy: scheduler-side — it gates whether a run is created from partition-aligned events. It can't gate between tasks in an already-running, time-scheduled Dag, which is where the wait belongs here.

On the visibility concern (the strongest objection, I think): I don't think satisfying it requires scheduling on the asset. Airflow already separates lineage from scheduling — inlets/outlets draw the dependency graph independently of what triggers a run. A task-level sensor can declare the awaited asset as an inlet, so the dependency stays visible to the platform (asset graph, lineage, OpenLineage) without coupling the run to asset-triggering. That keeps both: the consumer keeps its time schedule and data interval, and the data dependency stays first-class.

So I'd argue this is a genuine gap for time-partitioned batch consumers — the alternatives either change the run's contract or act at the wrong layer — and a task-level sensor is complementary to asset scheduling, not a replacement for it.

[uranusjr]

There are multiple issues with deferrable support. I’d suggest dropping this for now and only implement a synchronous sensor in this PR. After that’s merged, we can

1. Introduce triggerer changes needed to support asset event fetching.
2. Refactor InletEventsAccessor internals to ready it for an async interface.
3. Add the trigger to depend on the above changes.
4. Add deferrable=True to the sensor that only works on Airflow 3.4+.

[Leondon9]

This sequencing makes sense. FWIW step 1 — the triggerer changes for asset-event fetching — is already implemented in #67941, which independently hit the exact issue you flagged: GetAssetEventByAsset / GetAssetEventByAssetAlias aren't handled in the triggerer, and a mocked SUPERVISOR_COMMS hides it. That PR adds them to the ToTriggerSupervisor union + dispatch, with a regression test driving the real supervisor path (it fails if the union entries are removed).

Its trigger also takes the shape you're describing: natively async (await SUPERVISOR_COMMS.asend(GetAssetEventByAsset(...))), no sync_to_async and no InletEventsAccessor — so for this narrower partition-only case it doesn't need the accessor async-refactor (that refactor is still separately useful if the generic sensor routes through the accessor).

Happy for #67941 to serve as that step-1 prerequisite, or for the relevant commits to be split/reviewed however fits the plan — whatever keeps this moving.

[hussein-awala]

There are multiple issues with deferrable support. I’d suggest dropping this for now and only implement a synchronous sensor in this PR. After that’s merged, we can

1. Introduce triggerer changes needed to support asset event fetching.
2. Refactor InletEventsAccessor internals to ready it for an async interface.
3. Add the trigger to depend on the above changes.
4. Add deferrable=True to the sensor that only works on Airflow 3.4+.

@uranusjr Thanks, that's a fair catch on the triggerer,GetAssetEventByAsset / GetAssetEventByAssetAlias aren't available there, so the deferrable path couldn't have worked in practice, and my DB-backed tests hid it because the harness stands in for SUPERVISOR_COMMS and runs the execution-API SQL directly.

I've dropped deferrable support from this PR: AssetEventTrigger and its tests are gone, the shared fetch/count/serialize helpers moved into the sensor module, and the trigger is deregistered from provider.yaml. What's left is the synchronous sensor only. That also resolves your point on the example DAG — process_result=latest_event_only is fine now that it only ever runs on the worker.

[uranusjr]

Also, the lineage point raised above makes sense to me; we should resolve that before moving forward. Let’s have the sensor declare the awaited asset as an inlet and read context["inlet_events"][asset]. That makes the dependency visible without coupling it to scheduling, and drops the InletEventsAccessor usage.

[Leondon9]

I raised this use case in #67375, and I've got the deferrable path working in #67941 already — happy to pick up the follow-up (steps 1/3/4) once the sync sensor lands.

It's in the shape you described: triggerer support for GetAssetEventByAsset/ByAssetAlias (with a test on the real supervisor path, which is where the mocked-comms gap hides), a natively-async trigger with no InletEventsAccessor, and deferrable gated to 3.4+. Will rebase onto the merged sensor and reshape however fits the sequencing.

[nathadfield]

Hey @hussein-awala, @uranusjr & @Leondon9! I'm keen to see this move forward asap and wondered if any more progress has been made?

I'm more than happy to assist if I can in any way.

[Leondon9]

Thanks for keeping this alive, @nathadfield. The path is clear from @uranusjr's review — a synchronous AssetEventSensor first (drop deferrable, use the inlet / context["inlet_events"][asset] approach for lineage), with the triggerer/deferrable work as a follow-up.
@hussein-awala , is there anything blocking the sync-only rework, or anything we can do to help get it over the line? Happy to review, test, or take a pass at the changes if you're short on time — I'd just really like to see this land.

[hussein-awala]

I have limited availability to test my changes, but I’ll have everything ready by the end of next week.

I’ll convert the PR to draft since I’ll be pushing some untested changes, and I’ll mark it as ready once the changes have been tested and reviewed.

[hussein-awala]

@uranusjr @cmarteepants, could you review the run-relative bounds in this update? after and before now accept any object with resolve(context). The built-in helpers cover run start, logical date, and data interval start and end. Each helper accepts an optional lookback.

The implementation uses Airflow’s existing template rendering. Custom resolvers need only resolve(context). Inheritance is optional.

Please check the resolver interface, anchor semantics, and behavior after rescheduling.

[uranusjr]

Template callables are only evaluated from 3.3.0 so we’ll need some additional function gating below and notes in documentation.

[hussein-awala]

I checked the 3.2.0 source. It already evaluates template callables with context and jinja_env.
I also tested this with released Airflow 3.2.2 and Task SDK 1.2.2. The callable tests pass for both after and before, so I kept support on 3.2+.

[uranusjr]

This relies on resolve() duck-typing, but it’s never documented as a public interface. It’s fine in the standard provider, but we’ll need some tests around it in the provider so a future Airflow changes get compat coverage.

[hussein-awala]

I extended the provider tests to cover plain objects and protocol subclasses for both after and before. They exercise template rendering and polling with datetime and ISO string results. The tests also cover mapped custom resolvers. These cases pass with released Airflow 3.2.2.

[uranusjr]

This is both the lookup key and the error label. RunStart fakes it and overrides _get_timestamp anyway. It would be simpler IMO to make _get_timestamp the single abstract hook with a separate label, which also removes the cast at L81 too.

The built-ins serialize as their repr. This format is specific to Python so other languages would have trouble reproducing it. It would be better to either add and explicit serialize().

[hussein-awala]

I made _get_timestamp() the single abstract hook and added a separate error label. This removes the lookup cast.
The built-in helpers now provide serialize() with anchor and offset_seconds fields. Mapped arguments still use the core serializer’s representation fallback, so the helpers retain a deterministic __repr__.

[uranusjr]

Do we have a reference (issue/link) to be used here?

[hussein-awala]

I added a link to #75751 at the workaround. The comment states when we can remove it.

[uranusjr]

This misses _ParameterBound without __repr__ serializes ugly in partial_kwargs and literal expand values, and every parse writes a new Dag version because the repr format contains an object id.

Document that mapped custom resolvers need a deterministic repr, and add a test that serializes the mapped operator twice and compares. The real fix (a qualname fallback in BaseSerialization) is a separate core PR.

[hussein-awala]

I documented the need for a deterministic __repr__ that includes the resolver’s configuration.
The new tests compare serialization of two independently created, equivalent Dags. They cover partial and expand for every built-in helper and a custom resolver. They also check that different configurations produce different serialized values.
The general fallback change remains separate core work, as suggested.

[uranusjr]

I think DataIntervalEnd() is better for a time-scheduled dag. Either change this, or switch to an asset-triggered dag. (Or maybe we can have two dags here?)

[hussein-awala]

I changed the daily example to after=DataIntervalEnd().

[uranusjr]

This contradicts the warning about having no bounds would fetch the entire history. (I wonder if we should reject this by default and require the user to pass at least one bound, or set e.g. limit=None explicitly.)

[hussein-awala]

I added after=DataIntervalEnd() to this example.

The sensor now requires at least one query filter: after, before, partition_key, partition_key_regexp_pattern, or non-empty extra. Time bounds remain optional when another query filter is present.
The sensor checks this at construction and before polling, after template rendering. An empty extra dictionary does not qualify. Neither limit nor process_result alone satisfies the requirement.

[uranusjr]

This was written for a templated example. The example now uses RunStart, so the run_after sentence no longer applies. Rewrite to cover what users will hit.

Something like for asset-triggered runs, the logical date and data interval are None, so LogicalDate() and DataIntervalStart/End() raise, so RunStart() should be used instead.

[hussein-awala]

I replaced the obsolete note. The guide now explains that asset-triggered runs have no logical date or data interval. Bounds that require those timestamps raise ValueError.
It also compares RunStart() with the new RunAfter() helper. RunAfter() keeps a fixed boundary after clearing, but its inclusive lower bound can match a triggering event.

[uranusjr]

True for reschedules, not for clears. Clearing a finished run resets start_date, so the window moves forward, and a cleared sensor can wait forever for events the original run already saw.

[hussein-awala]

I corrected the guide to distinguish clearing from rescheduling. Clearing a finished run gives it a new actual start time when it starts again. RunStart() therefore moves forward.
The table contrasts this with the fixed RunAfter(), logical date, and interval bounds. The tests check each helper’s response to changed run metadata. They also check that reconstructed sensors keep their bounds after rescheduling.

[uranusjr]

On before, this moves the upper bound earlier, which reads oddly. A signed offset reads naturally for both bounds. (It’s awkward to change it though)

[hussein-awala]

I replaced lookback with a signed offset. Each helper adds the offset to its timestamp. Negative values move the bound earlier, and positive values move it later.
The docs show examples for both bounds. The tests cover zero, positive, and negative offsets.

[uranusjr]

Probably not needed since we already said inheritance is optional above?

[hussein-awala]

I removed the repeated sentence. The interface description states once that inheritance is optional.

[cmarteepants]

@hussein-awala nice, thanks for adding the bounds! RunStart(lookback=...) reads well for the common case. Building on TP's review this morning:

1. No bounds by default: +1 to requiring an explicit choice, either at least one bound or an explicit opt-in to the full history.

2. Clearing an asset-triggered run: I didn't realize this, but yes as TP pointed out, clearing a finished run resets start_date, so a RunStart() window moves forward. I don't think that's wrong, just a tradeoff. Users clear a run either to pick up new data (after an upstream fix, for example) or to redo the same work (after fixing a downstream bug, for example).

   RunStart() already handles the former. For the latter, scheduled runs can use LogicalDate(), which doesn't change on clear, but asset-triggered runs have no logical date. A RunAfter() bound would fill that gap, as long as it avoids the triggering-event trap from my first review (run_after is the newest triggering event's timestamp, and after is inclusive).

   Authors have to pick up front, but at least a wrong guess with RunStart() times out loudly while a fixed bound will pass silently.

3. Docs: we should spell out these tradeoffs. Something like:

   Bound	Available on	On clear	Watch out for
   RunStart()	All runs	Moves to the clear time (finished runs only), so the rerun waits for new events	With lookback, events from before the run can satisfy it immediately
   LogicalDate()	Scheduled runs only; raises on asset-triggered runs	Fixed	Meaning depends on the timetable: the trigger time for CronTriggerTimetable (the default for cron strings), the interval start for CronDataIntervalTimetable
   DataIntervalStart() / DataIntervalEnd()	Scheduled runs only; raises on asset-triggered runs	Fixed	Zero-length on CronTriggerTimetable, so both equal the logical date
   RunAfter() (proposed)	All runs	Fixed	On asset-triggered runs, equals the triggering event's timestamp

   My earlier pass had a docs section covering this. Once we settle on the approach, I'm happy to take a pass and add it.

[hussein-awala]

Thank you both for the review!

I believe it's better to keep time bounds optional. Users can filter by partition, for example to get the latest event for a partition regardless of its creation date. I do not think we should require start or end bounds for this use case.

I will document each bound’s use cases and tradeoffs, as Constance suggested, including its behavior when tasks are cleared.

I will go through TP’s comments and ping him once I have addressed them.

[hussein-awala]

@cmarteepants Thank you for the suggestions. I added the comparison table with availability, behavior after clearing, and cautions for each bound.

The sensor now requires at least one query filter. Time bounds remain optional when another query filter is present.

I also added RunAfter(), which uses dag_run.run_after and stays fixed after clearing. It keeps the same inclusive semantics as the other helpers. The docs explain that a triggering event can satisfy after=RunAfter().

To exclude events exactly at that timestamp, users can set after=RunAfter(offset=timedelta(microseconds=1)). This does not exclude later events from an earlier execution. Fixed bounds also do not freeze the event set.