Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry

[hkc-8010]

Avoid retrying the non-idempotent trigger Dag run POST through the generic execution API retry layer.

When the trigger POST raises an ambiguous transport-level httpx.RequestError, the client now probes the requested Dag run ID. If the run exists and reset_dag_run=False, it treats the trigger as successful. If the run is missing, it re-raises the original request error.

This preserves direct 409 Conflict behavior, including DAGRUN_ALREADY_EXISTS and the existing reset_dag_run=True clear path.

closes: #66905

Tests:

• uv run --project task-sdk ruff format task-sdk/src/airflow/sdk/api/client.py task-sdk/tests/task_sdk/api/test_client.py
• uv run --project task-sdk ruff check --fix task-sdk/src/airflow/sdk/api/client.py task-sdk/tests/task_sdk/api/test_client.py
• uv run --project task-sdk pytest task-sdk/tests/task_sdk/api/test_client.py::TestDagRunOperations -q
• prek run --files task-sdk/src/airflow/sdk/api/client.py task-sdk/tests/task_sdk/api/test_client.py
• breeze testing task-sdk-tests --python 3.10 task-sdk/tests/task_sdk/api/test_client.py::TestDagRunOperations
• breeze testing task-sdk-tests --python 3.10
• breeze ci selective-check --commit-ref HEAD --default-branch main --github-repository apache/airflow

---

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

• Yes (please specify the tool below)

Generated-by: OpenAI Codex following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[jason810496]

Thanks! The overall direction LGTM.

I agreed with not have auto-retry for the POST, but I feel we should check whether the DagRun existed before the actual POST call.

[jason810496]

Thanks for the update, would appreciate to harden the following points.
Please feel free to make as ready to review after fix, thanks!

[jason810496]

I marked it as draft to avoid adding this to 3.2.2 release before the follow-up fix.

[hkc-8010]

@jason810496 Addressed all four points — see commit 8a9259d:

• Pre-check and ambiguous-POST follow-up now both use self.get_count(dag_id=dag_id, run_ids=[run_id]) directly, removing _dag_run_exists and its ~40 lines of compat handling.
• _is_legacy_previous_dag_run_route_response was removed along with _dag_run_exists (it was only used there).
• Added a TODO at the clear() call site for the idempotency follow-up.

Would appreciate a re-review. Thanks!

[potiuk]

@hkc-8010 A few things need addressing before review — see our Pull Request quality criteria.

• ❌ CI fails: CI image checks / Static checks (and possibly other checks — see the Checks tab for the full list).

No rush.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @potiuk before posting

[hkc-8010]

@potiuk @jason810496 The earlier static-check failure appears to have been dependency lock drift after the branch fell behind main. I merged main into this branch, CI reran on the latest head, and all required checks are now green, including CI image checks / Static checks and Task SDK tests / task-sdk:P3.10 tests.

Would you mind taking another look when you get a chance? Thanks.

[jason810496]

After thoroughly review again. I think we might don't even need to introduce the extra check here, the client-side DagRunAlreadyExists error is acceptable in this case IMHO.

The trade-off here is that to achieve the exactly-once trigger for the execution API client we need to introduce extra get count round-trip to probe and make sure there isn't same DagRun before.

However, TriggerDagRunOperator might be critical path / hot path for some users, the additional probe here cause more pressure for the API-server / DB and increase the latency.

Additionally, we can safely rely on foreign key constraint of RDBMS to ensure there will only be exactly one DagRun on the server-side.

It's more like a design choice instead of straightforward patch.
I lean toward to keep it as-is instead of applying this patch.
WDYT? @amoghrajesh, @kaxil

[jason810496]

Still waiting for final decision on #66911 (review) comment, mark as request changes in case of accidental merge.