[mrafayaleem]

Ports the Affirm fork patches from 2.5.1+affirm4 onto upstream Apache Airflow 2.9.3, for BATCH-4364. Plan: Airflow 2.5.1 → 2.9.3 Upgrade Plan §5.1–5.2.

Base branch: 2.9.3-upstream is upstream Apache Airflow 2.9.3, unmodified (81845de). So this diff is exactly the Affirm patch set: 2 commits, 13 files. Approval leads to the 2.9.3+affirm1 tag (BATCH-4365 builds the wheel from it). Nothing merges into master, per plan §5.2.

Affirm changes

#	Change	Action
A1	cancel_dag_run / clear_dag_run experimental API modules, endpoints, api_client / local_client methods	Ported
A2	cancel_task_instances() in models/taskinstance.py	Ported; from airflow.jobs.job import Job (the 2.9 airflow.jobs package is empty)
A3	cancel_dag_run: State.finished() → State.finished	Fixed; it's a frozenset, so the old call raised TypeError
A4	trigger_dag writes an api_trigger_dag audit Log row and defaults execution_date to now	Ported as is
A5	get_dag_runs(**kwargs) + dag_runs dagrun_id / execution_date_gte filters	Ported
A6	DagModel.set_is_paused → [affirm] toggle_dag_hook	Ported (see fixes below)
A7	Footer version link → the fork	Ported
A8–A10	ti_count guard + test, RELEASE_NOTES.rst edits	Dropped (A8 upstream since 2.5.3, apache#30084)
A11	__version__ = "2.9.3+affirm1" in airflow/__init__.py (hatchling; no setup.py)	Done

Also dropped: an unused send_email import in models/dag.py.

Fixes found while porting (beyond the plan)

• A6: models/dag.py imports the config as airflow_conf in 2.9, so the old conf.getimport(...) raised NameError on every pause/unpause. It also passes fallback=None now, so installs without an [affirm] section skip the hook instead of raising. Prod sets the key, so behaviour there is unchanged.
• cancel_dag_run: the run and its TIs were loaded in a different session from the one used for writes. session.merge(ti) then cascaded the stale QUEUED DagRun back over the FAILED state that cancel_task_instances had just set. It now loads both in its own session. This was hidden on 2.5.1 because A3's TypeError fired first.

Tests

tests/www/api/experimental/: 34 passed (Python 3.9.18, constraints-2.9.3/constraints-3.9.txt).

• New test_cancel_clear_dag_run_endpoints.py covers cancel and clear, including RUNNING TI → SHUTDOWN with its job shut down, and the A3 finished-run regression.
• New dag_runs tests for the dagrun_id and execution_date_gte filters, plus the bad-date 400.
• New test that toggle_dag_hook fires on pause.
• test_trigger_dag now asserts the audit row. I removed its zero-microsecond assertion, because the fork sets execution_date before replace_microseconds is derived (unchanged from 2.5.1+affirm4).

Related upstream tests for DAG pause, trigger and delete, and tests/api/common/: 17 passed.

Downstream validation: the airflow-att unit suite was run against upstream 2.9.3 with this PR's airflow/ patches applied (DT pins from plan §6.1): 969/974 with Affirm/all-the-things#236129. The 5 remaining failures are AffirmDAG start_date cases unrelated to these patches.

CI: this fork only has upstream Apache's workflows. Their Build Info job never gets a runner here (it was cancelled after 24h without running), so the results above are from local runs.

Open items

• Does /api/experimental/dags/<id>/cancel_dag_run have a live caller? In all-the-things, platform/batch/live/.../task_scheduler_client.py:196 defines cancel_dagrun(), but only its unit test calls it. Callers outside that repo still need confirmation. Today the endpoint always returns 500 (A3), so any caller would start seeing successful cancels.
• After approval, tag 2.9.3+affirm1. That tag unblocks BATCH-4365, which builds and publishes the wheel.

🤖 Generated with Claude Code