Validate trigger events in Anthropic deferrable tasks

providers/anthropic/src/airflow/providers/anthropic/exceptions.py

@@ -29,6 +29,10 @@ class AnthropicBatchTimeout(AnthropicError):
     """Raised when an Anthropic Message Batch does not reach a terminal status in time."""
 
 
+class AnthropicTriggerEventError(AnthropicError):
+    """Raised when a deferred task resumes with a missing or malformed trigger event."""
+
+
 class AnthropicAgentSessionError(AnthropicError):
     """Raised when a Managed Agents session terminates or fails."""
 

providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py

@@ -38,6 +38,7 @@
     AnthropicBatchJobError,
     AnthropicBatchTimeout,
     AnthropicError,
+    AnthropicTriggerEventError,
 )
 from airflow.providers.common.compat.sdk import AirflowSkipException, BaseHook
 
@@ -149,6 +150,29 @@ def evaluate_session_state(
     return False, None, False
 
 
+#: Statuses the provider's triggers emit in their terminal event.
+TRIGGER_EVENT_STATUSES = frozenset({"success", "error", "timeout"})
+
+
+def validate_execute_complete_event(event: dict[str, Any] | None = None) -> dict[str, Any]:
+    """
+    Validate the event a deferred task resumes with, returning it if well-formed.
+
+    The event crosses the triggerer/worker boundary through the metadata DB, so a
+    resuming task can receive ``None`` or a status its handlers do not recognize
+    (version skew, a custom trigger). Both must fail loudly: the ``execute_complete``
+    handlers raise on ``timeout``/``error`` and treat everything else as success, so
+    an unrecognized status would otherwise silently succeed.
+    """
+    if event is None:
+        raise AnthropicTriggerEventError("Trigger error: event is None")
+    if event.get("status") not in TRIGGER_EVENT_STATUSES:
+        raise AnthropicTriggerEventError(
+            f"Unexpected trigger event status {event.get('status')!r}: {event!r}"
+        )
+    return event
+
+
 def evaluate_batch_counts(
     *,
     batch_id: str | None,

providers/anthropic/src/airflow/providers/anthropic/operators/agent.py

@@ -23,7 +23,7 @@
 from typing import TYPE_CHECKING, Any
 
 from airflow.providers.anthropic.exceptions import AnthropicAgentSessionError, AnthropicAgentSessionTimeout
-from airflow.providers.anthropic.hooks.anthropic import AnthropicHook
+from airflow.providers.anthropic.hooks.anthropic import AnthropicHook, validate_execute_complete_event
 from airflow.providers.anthropic.triggers.agent import AnthropicAgentSessionTrigger
 from airflow.providers.common.compat.sdk import BaseOperator, conf
 
@@ -191,8 +191,7 @@ def execute(self, context: Context) -> str | None:
         return session.id
 
     def execute_complete(self, context: Context, event: Any = None) -> str:
-        if not event:
-            raise AnthropicAgentSessionError("Trigger resumed without an event payload.")
+        event = validate_execute_complete_event(event)
         # The deferred task is a fresh instance; restore the session id from the event.
         self.session_id = event["session_id"]
         status = event["status"]

providers/anthropic/src/airflow/providers/anthropic/operators/batch.py

@@ -23,7 +23,11 @@
 from typing import TYPE_CHECKING, Any
 
 from airflow.providers.anthropic.exceptions import AnthropicBatchJobError, AnthropicBatchTimeout
-from airflow.providers.anthropic.hooks.anthropic import AnthropicHook, evaluate_batch_counts
+from airflow.providers.anthropic.hooks.anthropic import (
+    AnthropicHook,
+    evaluate_batch_counts,
+    validate_execute_complete_event,
+)
 from airflow.providers.anthropic.triggers.batch import AnthropicBatchTrigger
 from airflow.providers.common.compat.sdk import BaseOperator, conf
 
@@ -149,6 +153,7 @@ def execute_complete(self, context: Context, event: Any = None) -> str:
         The deferred task is a fresh instance, so the batch ID is read from the event,
         not ``self.batch_id``.
         """
+        event = validate_execute_complete_event(event)
         self.batch_id = event["batch_id"]
         status = event["status"]
         if status == "timeout":

providers/anthropic/src/airflow/providers/anthropic/sensors/batch.py

@@ -23,7 +23,12 @@
 from typing import TYPE_CHECKING, Any
 
 from airflow.providers.anthropic.exceptions import AnthropicBatchJobError, AnthropicBatchTimeout
-from airflow.providers.anthropic.hooks.anthropic import AnthropicHook, BatchStatus, evaluate_batch_counts
+from airflow.providers.anthropic.hooks.anthropic import (
+    AnthropicHook,
+    BatchStatus,
+    evaluate_batch_counts,
+    validate_execute_complete_event,
+)
 from airflow.providers.anthropic.triggers.batch import AnthropicBatchTrigger
 from airflow.providers.common.compat.sdk import BaseSensorOperator, conf
 
@@ -107,6 +112,7 @@ def execute(self, context: Context) -> None:
         super().execute(context)
 
     def execute_complete(self, context: Context, event: Any = None) -> None:
+        event = validate_execute_complete_event(event)
         status = event["status"]
         if status == "timeout":
             raise AnthropicBatchTimeout(event["message"])

providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py

@@ -25,6 +25,7 @@
     AnthropicAgentSessionTimeout,
     AnthropicBatchTimeout,
     AnthropicError,
+    AnthropicTriggerEventError,
 )
 from airflow.providers.anthropic.hooks.anthropic import (
     DEFAULT_MODEL,
@@ -33,6 +34,7 @@
     BatchStatus,
     SessionStatus,
     evaluate_session_state,
+    validate_execute_complete_event,
 )
 
 pytest.importorskip("anthropic")
@@ -76,6 +78,33 @@ def test_is_terminal(self, status, expected):
         assert SessionStatus.is_terminal(status) is expected
 
 
+class TestValidateTriggerEvent:
+    @pytest.mark.parametrize(
+        ("event", "match"),
+        [
+            pytest.param(None, "event is None", id="none"),
+            pytest.param({}, "Unexpected trigger event status None", id="missing-status"),
+            pytest.param(
+                {"status": "ended", "batch_id": "b"}, "Unexpected trigger event status", id="unknown-status"
+            ),
+        ],
+    )
+    def test_invalid_event_raises(self, event, match):
+        with pytest.raises(AnthropicTriggerEventError, match=match):
+            validate_execute_complete_event(event)
+
+    @pytest.mark.parametrize(
+        "event",
+        [
+            pytest.param({"status": "success", "batch_id": "b"}, id="success"),
+            pytest.param({"status": "error", "batch_id": "b", "message": "boom"}, id="error"),
+            pytest.param({"status": "timeout", "batch_id": "b", "message": "slow"}, id="timeout"),
+        ],
+    )
+    def test_valid_event_is_returned(self, event):
+        assert validate_execute_complete_event(event) is event
+
+
 def _session(status, outcome_results=None):
     s = mock.MagicMock()
     s.status = status

providers/anthropic/tests/unit/anthropic/operators/test_agent.py

@@ -21,7 +21,11 @@
 import pytest
 
 from airflow.exceptions import TaskDeferred
-from airflow.providers.anthropic.exceptions import AnthropicAgentSessionError, AnthropicAgentSessionTimeout
+from airflow.providers.anthropic.exceptions import (
+    AnthropicAgentSessionError,
+    AnthropicAgentSessionTimeout,
+    AnthropicTriggerEventError,
+)
 from airflow.providers.anthropic.hooks.anthropic import AnthropicHook
 from airflow.providers.anthropic.operators.agent import AnthropicAgentSessionOperator
 from airflow.providers.anthropic.triggers.agent import AnthropicAgentSessionTrigger
@@ -201,10 +205,21 @@ def test_timeout_archives_and_raises(self, mock_hook_prop):
             op.execute_complete({}, {"status": "timeout", "session_id": "s", "message": "slow"})
         hook.archive_session.assert_called_once_with("s")
 
-    def test_none_event_raises(self):
+    @pytest.mark.parametrize(
+        ("event", "match"),
+        [
+            pytest.param(None, "event is None", id="none"),
+            pytest.param(
+                {"status": "rescheduling", "session_id": "s"},
+                "Unexpected trigger event status",
+                id="unknown-status",
+            ),
+        ],
+    )
+    def test_invalid_event_raises(self, event, match):
         op = AnthropicAgentSessionOperator(task_id="a", agent_id="ag", environment_id="env", message="hi")
-        with pytest.raises(AnthropicAgentSessionError, match="without an event"):
-            op.execute_complete({}, None)
+        with pytest.raises(AnthropicTriggerEventError, match=match):
+            op.execute_complete({}, event)
 
 
 class TestOnKill:

providers/anthropic/tests/unit/anthropic/operators/test_batch.py

@@ -21,7 +21,11 @@
 import pytest
 
 from airflow.exceptions import TaskDeferred
-from airflow.providers.anthropic.exceptions import AnthropicBatchJobError, AnthropicBatchTimeout
+from airflow.providers.anthropic.exceptions import (
+    AnthropicBatchJobError,
+    AnthropicBatchTimeout,
+    AnthropicTriggerEventError,
+)
 from airflow.providers.anthropic.hooks.anthropic import AnthropicHook
 from airflow.providers.anthropic.operators.batch import AnthropicBatchOperator
 from airflow.providers.anthropic.triggers.batch import AnthropicBatchTrigger
@@ -188,6 +192,18 @@ def test_partial_error_fails_when_strict(self):
         with pytest.raises(AnthropicBatchJobError, match="failed request"):
             op.execute_complete(_context(), event)
 
+    @pytest.mark.parametrize(
+        "event",
+        [
+            pytest.param(None, id="none"),
+            pytest.param({"status": "ended", "batch_id": "b"}, id="unknown-status"),
+        ],
+    )
+    def test_invalid_event_raises_instead_of_succeeding(self, event):
+        op = AnthropicBatchOperator(task_id="t", requests=REQUESTS)
+        with pytest.raises(AnthropicTriggerEventError):
+            op.execute_complete(_context(), event)
+
 
 class TestOnKill:
     @mock.patch.object(AnthropicBatchOperator, "hook", new_callable=mock.PropertyMock)

providers/anthropic/tests/unit/anthropic/sensors/test_batch.py

@@ -21,7 +21,11 @@
 import pytest
 
 from airflow.exceptions import TaskDeferred
-from airflow.providers.anthropic.exceptions import AnthropicBatchJobError, AnthropicBatchTimeout
+from airflow.providers.anthropic.exceptions import (
+    AnthropicBatchJobError,
+    AnthropicBatchTimeout,
+    AnthropicTriggerEventError,
+)
 from airflow.providers.anthropic.hooks.anthropic import AnthropicHook
 from airflow.providers.anthropic.sensors.batch import AnthropicBatchSensor
 from airflow.providers.anthropic.triggers.batch import AnthropicBatchTrigger
@@ -102,3 +106,15 @@ def test_execute_complete_success(self):
         sensor = AnthropicBatchSensor(task_id="s", batch_id="b1")
         event = {"status": "success", "batch_id": "b1", "request_counts": {"succeeded": 2}}
         assert sensor.execute_complete({}, event) is None
+
+    @pytest.mark.parametrize(
+        "event",
+        [
+            pytest.param(None, id="none"),
+            pytest.param({"status": "ended", "batch_id": "b1"}, id="unknown-status"),
+        ],
+    )
+    def test_execute_complete_invalid_event_raises_instead_of_succeeding(self, event):
+        sensor = AnthropicBatchSensor(task_id="s", batch_id="b1")
+        with pytest.raises(AnthropicTriggerEventError):
+            sensor.execute_complete({}, event)