hussein-awala opened a new pull request, #74726:
URL: https://github.com/apache/airflow/pull/74726
## Discussion-only PoC
This draft explores a native Task SDK API for **publishing asset
availability when it becomes true**, rather than coupling every publication to
successful completion of a task.
The proposed interface is `asset.emit(extra=..., partition_key=...,
partition_date=...)`. This is a proof of concept for discussing ergonomics,
semantics, runtime coverage, and whether a full implementation needs an AIP.
**It is not ready to merge as a supported production API.**
## Motivation
### 1. Publication from callbacks and notifiers
Asset-event publication should be available to task-level and Dag-level
callbacks, and to notifiers invoked at either level. This is not limited to
failure callbacks: completion, success, retry, and other applicable lifecycle
callbacks can have useful follow-up work.
A native API would let these entry points share transport and provenance
handling rather than each implementing REST endpoint discovery and
authentication. A notifier could be a thin adapter over that API.
**Operational workflows are one use case:** a callback or notifier could
publish a signal consumed by a dedicated operational Dag to investigate a
failure (including AI-assisted investigation), clean up temporary resources
after completion, run validation, or initiate reconciliation. The callback
stays lightweight; the consumer Dag owns scheduling, retries, logs, and
visibility. AI is an example, not a dependency or the defining purpose of this
feature.
These workflows should be explicitly opted in. Cleanup needs to be safe to
retry, investigation workflows should avoid triggering themselves, and event
metadata should contain references to relevant runs/resources rather than
secrets or entire logs. Whether workflow-outcome signals should use ordinary
assets or a distinct abstraction is also worth discussing.
### 2. More flexible publication from a running task
Outlets express **“publish this update when the task succeeds.”** An
immediate API expresses **“this specific output is ready now.”** Both are
useful; this proposal does not change existing outlet semantics.
For example, an ingestion task may process a large backlog in batches, using
task state/checkpoint storage to track progress:
```text
Producer Dag: ingestion task remains running
Commit batch 41 to storage
Emit readiness for partition batch-41
Record progress in the processing checkpoint
Commit batch 42 to storage
Emit readiness for partition batch-42
Continue processing batch 43 ...
Consumer Dag
batch-41 readiness -> validate/transform batch 41
batch-42 readiness -> validate/transform batch 42
```
The consumer can process the first committed batches while the producer
continues. A later failure on batch 43 should not necessarily hide already
durable outputs from batches 41 and 42.
Task state records the producer's progress; asset events communicate
availability to other workflows. This also avoids introducing artificial task
boundaries solely to publish an event when work naturally shares a connection,
session, or internal batch loop.
The batch example above illustrates the motivation; **this PoC does not
implement a checkpointed ingestion operator or change state-store semantics**.
External writes, emission, and checkpoint updates are not automatically atomic.
Also, ordinary asset scheduling can coalesce events: distinct batch keys and
appropriate partition timetable semantics are needed when batches should be
processed separately. One `emit()` call does not guarantee one consumer Dag run.
## Proposed API and semantics
```python
receipt = asset.emit(
extra={"rows": 100},
partition_key="2026-10-01",
partition_date=pendulum.datetime(2026, 10, 1, tz="UTC"),
)
print(receipt.id, receipt.timestamp)
```
- Events are persisted immediately, independently of the producer's eventual
outcome.
- Emitting does not mark the source task or Dag successful, and later
failure does not retract the event.
- Existing success-only outlets are unchanged. Using both mechanisms can
create separate events.
- The asset must already be registered and is matched by both name and URI.
- Partition arguments are explicit; omitted values do not inherit the
producer run's partition. A date requires a key and a timezone-aware datetime.
- The event stores its partition key. The date is passed through existing
partition scheduling/mapper logic; this does not introduce an asset-event date
column.
- The receipt acknowledges persistence, not downstream execution.
- Ordinary parse-time emission is rejected.
## What this PoC implements
The current implementation exercises **running Python tasks and Python Dag
lifecycle callbacks through the existing in-process Dag processor transport**.
The broader goal includes task callbacks and task-/Dag-level notifiers;
**notifier adapters and coverage of all callback execution paths are not
implemented here**.
The request path is:
`Asset.emit()` -> typed supervisor message -> Execution API client -> server
route -> existing asset manager.
Task provenance comes from the authenticated task attempt. Dag callback
provenance comes from the trusted parent's actual callback request batch,
without treating a representative TI as the producer. Dag-sourced events retain
Dag/run fields and leave task/TI/map-index fields null. User callback code does
not access the metadata database.
Included examples in
`airflow-core/src/airflow/example_dags/example_asset_emission.py`:
- A task publishes two dated partitions, then deliberately fails.
- A separate failing Dag publishes a partition from its Dag callback without
reading a TI.
- Two partition-aware consumers display partition/date and provenance.
The failure examples demonstrate independence from successful completion;
they are not intended to restrict the API to failure handling.
Local acceptance also exposed a public REST/UI mismatch: the database and
SDK supported a null source map index, while the public response required an
integer. This draft aligns that response and generated clients, links Dag-only
sources to their Dag run rather than `/tasks/null`, and surfaces event-fetch
errors in the asset/run pages.
## Deliberate limits and open questions
- Remote/API-backed Dag processor credentials are **not** admitted to the
emission endpoint. Production callback support needs an explicit
identity/capability design, not relaxed authentication or a fabricated TI.
- No automatic asset creation, aliases, async API, triggerer support,
executor callback workers, or other-language public SDK implementation.
Generated supervisor wire schemas/types are synchronized for the language SDKs;
that synchronization is not implementation of an emission API in those SDKs.
- No durable callback-delivery or exactly-once guarantee. Automatic HTTP
retries are disabled for the event POST, but lost acknowledgements and
caller/task retries can still produce duplicates.
- Existing asset/team filtering is reused. Publication authorization and
per-emission consumer controls need further review.
- API placement (`Asset.emit()` versus a standalone SDK function), partition
inheritance, idempotency, callback coverage, notifier design, and whether this
warrants an AIP remain discussion points.
Related: #67637 explores success-only Dag-level outlets. This PoC explores a
different publication boundary: an explicit runtime operation independent of
task/Dag success.
## Validation
- On the rebased publication branch, **186 SDK/supervisor tests and 268
Core/API/partition/callback tests passed** in Breeze with Python 3.12 / SQLite.
- The callback integration test launches a real supervised parser subprocess
with no TI in its callback context.
- Local scheduler/UI acceptance: both deliberate producer failures produced
their events; all three partitioned consumer runs succeeded. The previously
failing asset-event and upstream-event API requests now return 200 with the
existing callback event, without rewriting its provenance.
- Eight focused UI tests, TypeScript checking, frontend lint, and the
production UI build passed on the tested PoC worktree.
- Fast and manual static checks were run for the publication worktree.
`mypy-airflow-core` reports two httpx/httpx2 override errors in the unchanged
`dag_processing/api_client.py` (lines 211 and 329); **the same two errors
reproduce on clean upstream `f31684af85`**. That hook is explicitly excluded
from the draft commit, not treated as passing. Other hooks and commit signing
remain enabled.
- **The full core/provider/SDK/CLI/UI/API-client/docs matrix reported by
`breeze verify` has not been run. Full CI, cross-database validation, and
production hardening remain pending.** This is intentionally a discussion
draft, not a claim of merge readiness.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — OpenAI coding assistant in Pi
Generated-by: OpenAI coding assistant in Pi following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
Drafted-by: OpenAI coding assistant in Pi (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
