Hi everyone, I'd like to discuss a native Task SDK capability for explicitly emitting asset events from running tasks, task- and Dag-level callbacks, and notifiers used at either level.
I have put together a small proof of concept in draft PR #74726 <https://github.com/apache/airflow/pull/74726>. The goal is to get feedback on the use cases, API, and semantics before pursuing a full implementation or an AIP. The central idea is to let authors publish asset availability when it becomes true, rather than requiring every publication to coincide with successful completion of a task. Why an explicit emission API? Outlets are a good fit for “this task succeeded, so publish its asset updates.” I do not propose changing that behavior. However, task completion is not always the right publication boundary. A task can finish several independently durable units of work while still having more to do. Likewise, a callback can represent a workflow-level outcome that should not be attributed to a particular task. There are two main use cases I would like to explore. 1. Incremental publication from a running task Consider an ingestion task that processes a large backlog in batches and uses task state/checkpoint storage to track its progress. Once a batch is committed to storage, another Dag could start processing it immediately, while the ingestion task continues with subsequent batches: Producer Dag: ingestion task remains running Commit batch 41 to storage Emit an event for partition batch-41 Record progress in the processing checkpoint Commit batch 42 to storage Emit an event for partition batch-42 Continue processing batch 43 ... Consumer Dag batch-41 event -> validate and transform batch 41 batch-42 event -> validate and transform batch 42 If the producer later fails on batch 43, batches 41 and 42 are still valid, available outputs. Their publication should not necessarily depend on the outcome of the remaining work in that task. This complements task state: state records the producer's progress; events communicate availability to other workflows. It also allows users to choose when, how often, and for which outputs to emit events based on runtime decisions. Splitting the work into smaller tasks remains a good option. But it should not be necessary solely to publish an event, particularly when processing naturally shares a connection, session, or library-managed batch loop. This example assumes appropriate partition scheduling: ordinary asset scheduling can coalesce events, so one emission does not automatically mean one consumer Dag run. Batch-specific partition keys make the intended units of downstream work explicit. 2. Publication from callbacks and notifiers The same capability should be usable from supported task- and Dag-level callback contexts, not only failure callbacks. For example, a success callback could publish a workflow-level completion signal, or a retry callback could initiate an operational check. Notifiers should be able to use the same native operation; an asset notifier could be a thin adapter rather than a separate REST/authentication implementation. An important use case is triggering operational Dags: - Investigating a failure, potentially with an AI-assisted workflow. - Cleaning up temporary resources after completion. - Running post-processing validation. - Initiating reconciliation or other follow-up work. The callback or notifier would publish the signal and remain lightweight. The consumer Dag would own the operational work, with Airflow's scheduling, retries, logs, and visibility. AI-assisted investigation is just one example, not a dependency or the defining purpose of this proposal. Dag-level publication should not require a dummy TaskInstance. A representative TI in a callback context is not necessarily the producer of a workflow-level event, and some callback contexts have no TI at all. Such an event should retain source Dag/run provenance without inventing a task source. One question for the community is whether workflow-outcome signals fit ordinary assets, need additional conventions, or would be better represented by another abstraction. I would like to discuss that alongside the data-availability use case rather than assume the answer. Proposed API and behavior The PoC uses an instance method on an already registered asset: receipt = batch_ready.emit( partition_key="batch-0042", extra={"batch_id": "0042", "row_count": 10000}, ) The method name and API placement are open for discussion. The current semantics are: - Immediate publication: the event is persisted when called, independently of eventual task/Dag success. A later failure does not retract it. - No state transition: emitting an event does not mark the task or Dag successful. - Existing outlets are unchanged: explicit emission complements their success-only behavior. Using both can produce separate events. - Explicit partitions: partition_key and an optional timezone-aware partition_date are supported. A date requires a key. Omitted values do not implicitly inherit the producer run's partition in this PoC. - Runtime-only: emission during ordinary Dag parsing is rejected. - A small receipt: the return value contains the event ID and server timestamp; it acknowledges persistence, not downstream execution. The operation follows the existing SDK boundary: typed supervisor communication, the Execution API, and the existing asset manager. User task/callback code does not open database sessions or manage separate REST credentials. What the PoC does (and does not) cover: The current implementation exercises running Python tasks and Python Dag lifecycle callbacks through the existing in-process Dag processor transport. It includes partition-aware examples and tests, including a real supervised callback subprocess whose context has no TI. Local testing also exercised a task that emits two partitions before deliberately failing, and a Dag callback that emits a partition after failure. Their consumer runs succeeded. These examples demonstrate independence from successful completion; they are not intended to limit the API to failure handling. The broader callback/notifier coverage described above is a goal, not something the PoC fully implements. It does not yet include a notifier adapter, every task/executor callback path, remote/API-backed Dag processor credentials, or an emission API for other language SDKs. It also excludes automatic asset registration, aliases, and async emission. There is no exactly-once guarantee. Automatic HTTP retries are disabled for the event POST, but a lost acknowledgement may follow a committed event, and caller/task retries can still create duplicates. External data writes, event emission, and checkpoint updates are not automatically atomic; their ordering and deduplication need deliberate handling. The batch/checkpoint example illustrates the motivation, not a transaction protocol implemented by this PoC. Production callback capabilities, publication authorization, and idempotency need further design. Operational consumers would also need safe-to-retry behavior and protection against feedback loops. Event metadata should contain useful references, not secrets or entire logs. The PR records the focused test and local acceptance results, as well as outstanding validation. It remains a discussion draft, not a merge-ready API. Feedback requested 1. Do these use cases justify an explicit immediate-emission API alongside outlets? 2. Should it live on Asset, as in the PoC, or be a standalone SDK operation? 3. What callback/notifier coverage and provenance should an initial supported version provide? 4. Should omitted partition values remain explicit-none, or should there be an opt-in way to inherit the current run's partition? 5. Should idempotency support be required for an initial version, and how should it interact with retries and checkpointed processing? 6. Does this warrant an AIP, or would a focused design discussion and implementation be sufficient? For related work, #67637 <https://github.com/apache/airflow/pull/67637> explores success-only Dag-level outlets. This PoC explores a different publication boundary: an explicit runtime operation independent of task/Dag success. I'd appreciate feedback on both the data-processing and operational-workflow motivations, and on the smallest useful scope to pursue next. Thanks, Hussein
