fix(task-sdk): handle missing DAG in TriggerDagRunOperator gracefully

[YoannAbriel]

Problem

TriggerDagRunOperator crashes with UnboundLocalError: cannot access local variable 'state' when targeting a non-existent DAG. The API server returns 404, which SUPERVISOR_COMMS.send() re-raises as ServerResponseError — but this propagates past all exception handlers in run() into the finally block where state was declared but never assigned.

Root Cause

Two issues:

1. state: TaskInstanceState in run() is declared without a default value. Any unhandled exception reaching the finally block causes UnboundLocalError when accessing state.value for stats.

2. The DagRunOperations.trigger() client method only handles 409 (conflict) — a 404 from the API server is re-raised as ServerResponseError, which isn't caught by _handle_trigger_dag_run.

Fix

• Initialize state = TaskInstanceState.FAILED as a safe default so the finally block always works, regardless of what exception propagates.
• Add DAG_NOT_FOUND to ErrorType enum.
• Handle 404 in DagRunOperations.trigger(), returning ErrorResponse(error=ErrorType.DAG_NOT_FOUND).
• Handle DAG_NOT_FOUND in _handle_trigger_dag_run, failing the task gracefully instead of crashing.

Tests added for both the client-side 404 handling and the task runner DAG-not-found path.

Closes: #63089

---

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

• Yes — Claude Code

Generated-by: Claude Code 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.

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

[Copilot]

New test introduces an unspecced mock.MagicMock() for log. Airflow’s testing guidelines prefer mocks with spec/autospec to avoid silently accepting unexpected attributes; consider using a real structlog.get_logger(...) or mock.Mock(spec=...) for the logger instead.

[Copilot]

The new 404 handling logs only a generic "Dag not found" message and doesn’t include available response context (e.detail, status_code). Consider logging those fields (and/or returning them in ErrorResponse.detail) for easier debugging, consistent with other client methods’ 404 handling.

Suggested change

	log.error("Dag not found.", dag_id=dag_id)
	return ErrorResponse(error=ErrorType.DAG_NOT_FOUND)
	log.error(
	"Dag not found.",
	dag_id=dag_id,
	status_code=e.response.status_code,
	detail=e.detail,
	)
	return ErrorResponse(error=ErrorType.DAG_NOT_FOUND, detail=e.detail)

[potiuk]

@YoannAbriel — Removing the ready for maintainer review label and converting back to draft. The branch now has merge conflicts with main that surfaced after the label was added.

The label's contract is that the PR is ready for maintainer review — a regression like this means the PR temporarily isn't. Rebase your branch onto the latest main, resolve conflicts, then mark "Ready for review" again to re-enter the queue.

git fetch upstream main && git rebase upstream/main, resolve, git push --force-with-lease. See the working-with-git docs.

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.

[potiuk]

@YoannAbriel This draft PR has had no activity for 2 weeks. Closing to keep the queue clean.

You are welcome to reopen and continue when you're ready. If you'd like to pick it back up, please rebase onto the current main branch first.

---

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.