Overall I like the idea both from a platform maintainer's perspective and a user experience perspective.
*1. Do these use cases justify an explicit immediate-emission API alongsideoutlets?* I think that would make sense as a configurable behavior. E.g. There's some `immediate` boolean parameter for this SDK interface. The flexibility here I think is important given the broad range of potential use-cases and differing levels of event scope. *2. Should it live on Asset, as in the PoC, or be a standalone SDK operation?* This is probably where I have the strongest opinion: I'd be very much in favor of this being a distinct, separate entity, and not defined on the Asset. Just for the purposes of my answer below, I'll refer to this hypothetical new SDK entity as a "*Signal*" (not a naming suggestion). My thinking is: (a) The Asset concept offers a different functionality and is conceptually unique from what this PoC is proposing (the "Signal" concept). Injecting this feature set into the Asset feels like it goes against the Unix design philosophy of modular, small components which do one thing well. (b) As an Airflow platform maintainer: I see a huge benefits for this proposal since this could extend well beyond the triggering operational dags use case. For example, this could effectively operate as a system-wide event publisher (something analogous to Trino's EventListener interface). There's an abundance of use cases for tracking the fully "Signal" history on an operational level. (c) As an Airflow user: This would be a much welcomed addition to further event-based expressiveness, and definitely could reduce the DAG code verbosity. Personally, I've found a number of situations where the TriggerDagRunOperator was the only tool for a particular situation, but I could have had a much simpler and easier-to-maintain DAG if there were some task sdk functionality as proposed. I also think users would welcome a dag-level/task-level "Signal" interface as that could have a more intuitive and auditable authN/authZ model than intra-dag resources such as assets. *3. What callback/notifier coverage and provenance should an initial supported version provide?* coverage: No strong opinion here, but it might make sense to start the coverage at the task level and work upwards towards larger granularity over time? provenance: At least the dag + dag-run. And if possible optional references to the task/TI, only when task information is applicable. *4. Should omitted partition values remain explicit-none, or should there be an opt-in way to inherit the current run's partition?* I'd lean towards a flexible design, so maybe an opt-in with a configurable (and nullable) default value. *5. Should idempotency support be required for an initial version, and how should it interact with retries and checkpointed processing?* If you're asking about internal idempotency for how the task SDK handles these signal emissions, then yes I'd lean towards idempotency-first design. I don't know how we could enforce that requirement on the user's dag / pipeline logic though. *6. Does this warrant an AIP, or would a focused design discussion and implementation be sufficient?* I don't have a strong opinion here, but I'd imagine this thread will lead to a clear answer for this question. On Wed, Oct 14, 2026 at 9:06 AM Hussein Awala <[email protected]> wrote: > 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 >
