Validate trigger events in Anthropic deferrable tasks

[takayoshi-makabe]

The Anthropic provider's three deferrable resume paths — AnthropicBatchOperator.execute_complete, AnthropicBatchSensor.execute_complete and AnthropicAgentSessionOperator.execute_complete — used the trigger event without validating it:

• event=None: happens when a deferred task resumes without a payload — a custom or subclassed trigger that yields TriggerEvent(None), or an event payload lost between the triggerer and the worker. Crashed with an opaque
  TypeError: 'NoneType' object is not subscriptable.
• Unrecognised status: happens on triggerer/worker version skew (a task defers under one provider version and resumes under another whose trigger emits a different status vocabulary) or with a custom trigger. These methods branch deny-list style (timeout/error raise, everything else succeeds), so an unknown status fell through into the success path — the task was marked SUCCESS even though the Anthropic batch/session outcome was unknown.

This adds a validate_execute_complete_event() helper applied at the top of all three methods. The None check mirrors the amazon provider's helper of the same name (https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/utils/__init__.py#L70-L75).
The status check plays the role amazon's call sites cover with their if validated_event["status"] != "success": raise allow-list branching (https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py#L207-L209)

Note: event validation across other providers is inconsistent — databricks has a required-keys helper (validate_trigger_event), while e.g. google and dbt.cloud access the event unvalidated with the same deny-list branching. This PR only fixes the Anthropic provider.

---

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

• Yes (please specify the tool below)

Claude Code

---

• 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.

[Vamsi-klu]

Could the helper also validate the keys each resume path needs, not just status? An event like {"status": "success"} passes validate_execute_complete_event() but then the operator/sensor hits KeyError on session_id, batch_id, or request_counts. It would be cleaner for AnthropicTriggerEventError to cover malformed allowed-status events too, with tests for missing required keys per execute_complete path.

---

Drafted-by: Codex (GPT-5); reviewed by @Vamsi-klu before posting