TD-P012 (#370): fail deterministically on codec-decode errors at task boundary

Previously a worker task whose inbound payload could not be decoded left
the lease to time out instead of reporting a structured failure. The
three paths that bypassed fail_{workflow,activity}_task were:

1. Activity argument decode — decode_envelope() was called outside any
   try/except, so AvroNotInstalledError (ImportError subclass) or
   malformed-payload ValueError bubbled to the dispatcher's log-and-drop
   catch block.
2. Workflow start-input decode — try/except caught ValueError only;
   AvroNotInstalledError slipped past because ImportError is not a
   ValueError.
3. Replay history-result decode — serializer.decode() was called with no
   codec override, so Avro-encoded activity results were read as JSON.
   Any decode failure propagated to the dispatcher.

Changes:
- workflow.replay() accepts payload_codec and threads it through the
  ActivityCompleted / SideEffectRecorded / ChildRunCompleted result
  decodes so Avro-pinned runs replay correctly.
- Worker wraps each decode site in targeted try/except for
  AvroNotInstalledError and (ValueError, TypeError); reports fail_*
  with a clear codec/install hint, failure_type, and stack trace. Activity
  decode failures are marked non_retryable — bad bytes won't become good
  on retry.

Regression tests cover:
- activity with invalid JSON bytes -> fail_activity_task, non_retryable
- activity with invalid Avro bytes -> fail_activity_task, non_retryable
- activity with avro extra missing (monkeypatched) -> fail_activity_task
- workflow with invalid JSON start -> fail_workflow_task
- workflow with avro extra missing on start -> fail_workflow_task
- workflow replay with avro extra missing on history -> fail_workflow_task

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: durable-workflow-ops

src/durable_workflow/worker.py

@@ -11,7 +11,7 @@
 from . import serializer
 from .activity import ActivityContext, ActivityInfo, _set_context
 from .client import Client
-from .errors import ActivityCancelled, NonRetryableError
+from .errors import ActivityCancelled, AvroNotInstalledError, NonRetryableError
 from .workflow import replay
 
 log = logging.getLogger("durable_workflow.worker")
@@ -120,8 +120,40 @@ async def _run_workflow_task(self, task: dict[str, Any]) -> list[dict[str, Any]]
             decoded = serializer.decode_envelope(raw_args, codec=codec)
             if decoded is not None:
                 start_input = decoded if isinstance(decoded, list) else [decoded]
-        except ValueError as e:
-            log.warning("task %s start input unreadable (%s); falling back to [].", task_id, e)
+        except AvroNotInstalledError as e:
+            log.exception("task %s start input Avro decode failed (extra not installed)", task_id)
+            try:
+                await self.client.fail_workflow_task(
+                    task_id=task_id,
+                    lease_owner=self.worker_id,
+                    workflow_task_attempt=attempt,
+                    message=(
+                        f"cannot decode workflow start input with codec 'avro': {e}. "
+                        f"Install the avro extra: pip install 'durable-workflow[avro]'."
+                    ),
+                    failure_type=type(e).__name__,
+                    stack_trace=traceback.format_exc(),
+                )
+            except Exception as fe:
+                log.warning("failed to report Avro-missing start input failure: %s", fe)
+            return None
+        except (ValueError, TypeError) as e:
+            log.exception("task %s start input decode failed (codec=%r)", task_id, codec)
+            try:
+                await self.client.fail_workflow_task(
+                    task_id=task_id,
+                    lease_owner=self.worker_id,
+                    workflow_task_attempt=attempt,
+                    message=(
+                        f"cannot decode workflow start input with codec {codec!r}: {e}. "
+                        f"Verify the start input bytes match the declared codec and writer schema."
+                    ),
+                    failure_type=type(e).__name__,
+                    stack_trace=traceback.format_exc(),
+                )
+            except Exception as fe:
+                log.warning("failed to report start input decode failure: %s", fe)
+            return None
 
         run_id: str = task.get("run_id", "")
 
@@ -140,7 +172,24 @@ async def _run_workflow_task(self, task: dict[str, Any]) -> list[dict[str, Any]]
             return None
 
         try:
-            outcome = replay(cls, history, start_input, run_id=run_id)
+            outcome = replay(cls, history, start_input, run_id=run_id, payload_codec=codec)
+        except AvroNotInstalledError as e:
+            log.exception("replay failed: Avro extra not installed")
+            try:
+                await self.client.fail_workflow_task(
+                    task_id=task_id,
+                    lease_owner=self.worker_id,
+                    workflow_task_attempt=attempt,
+                    message=(
+                        f"cannot replay workflow history with codec 'avro': {e}. "
+                        f"Install the avro extra: pip install 'durable-workflow[avro]'."
+                    ),
+                    failure_type=type(e).__name__,
+                    stack_trace=traceback.format_exc(),
+                )
+            except Exception as fe:
+                log.warning("failed to report replay Avro-missing failure: %s", fe)
+            return None
         except Exception as e:
             log.exception("replay failed")
             try:
@@ -182,7 +231,44 @@ async def _run_activity_task(self, task: dict[str, Any]) -> None:
         raw_args = task.get("arguments")
         inbound_codec = task.get("payload_codec") or serializer.JSON_CODEC
         result_codec = inbound_codec if inbound_codec in serializer.SUPPORTED_CODECS else serializer.JSON_CODEC
-        args = serializer.decode_envelope(raw_args, codec=inbound_codec) or []
+        try:
+            args = serializer.decode_envelope(raw_args, codec=inbound_codec) or []
+        except AvroNotInstalledError as e:
+            log.exception("activity %s arguments Avro decode failed (extra not installed)", task_id)
+            try:
+                await self.client.fail_activity_task(
+                    task_id=task_id,
+                    activity_attempt_id=attempt_id,
+                    lease_owner=self.worker_id,
+                    message=(
+                        f"cannot decode activity arguments with codec 'avro': {e}. "
+                        f"Install the avro extra: pip install 'durable-workflow[avro]'."
+                    ),
+                    failure_type=type(e).__name__,
+                    stack_trace=traceback.format_exc(),
+                    non_retryable=True,
+                )
+            except Exception as fe:
+                log.warning("failed to report Avro-missing activity decode failure: %s", fe)
+            return
+        except (ValueError, TypeError) as e:
+            log.exception("activity %s arguments decode failed (codec=%r)", task_id, inbound_codec)
+            try:
+                await self.client.fail_activity_task(
+                    task_id=task_id,
+                    activity_attempt_id=attempt_id,
+                    lease_owner=self.worker_id,
+                    message=(
+                        f"cannot decode activity arguments with codec {inbound_codec!r}: {e}. "
+                        f"Verify the argument bytes match the declared codec and writer schema."
+                    ),
+                    failure_type=type(e).__name__,
+                    stack_trace=traceback.format_exc(),
+                    non_retryable=True,
+                )
+            except Exception as fe:
+                log.warning("failed to report activity decode failure: %s", fe)
+            return
         if not isinstance(args, list):
             args = [args]
 

src/durable_workflow/workflow.py

@@ -279,6 +279,7 @@ def replay(
     start_input: list[Any],
     *,
     run_id: str = "",
+    payload_codec: str | None = None,
 ) -> ReplayOutcome:
     events = list(history_events)
 
@@ -300,14 +301,14 @@ def replay(
         etype = ev.get("event_type")
         payload = ev.get("payload") or {}
         if etype in ("ActivityCompleted", "activity_completed"):
-            resolved_results.append(serializer.decode(payload.get("result")))
+            resolved_results.append(serializer.decode(payload.get("result"), codec=payload_codec))
         elif etype in ("TimerFired", "timer_fired"):
             resolved_results.append(None)
         elif etype in (
             "SideEffectRecorded", "side_effect_recorded",
             "ChildRunCompleted", "child_run_completed",
         ):
-            resolved_results.append(serializer.decode(payload.get("result")))
+            resolved_results.append(serializer.decode(payload.get("result"), codec=payload_codec))
         elif etype in ("ChildRunFailed", "child_run_failed"):
             resolved_results.append(ChildWorkflowFailed(
                 payload.get("message", "child workflow failed")

tests/test_worker.py

@@ -292,6 +292,166 @@ async def test_workflow_with_envelope_arguments(self, mock_client: AsyncMock) ->
         mock_client.complete_workflow_task.assert_called_once()
 
 
+class TestCodecDecodeFailures:
+    """TD-P012 / #370 regression: codec decode failures at the task boundary
+    must turn into a deterministic fail_{workflow,activity}_task call so the
+    lease does not sit until timeout."""
+
+    @pytest.mark.asyncio
+    async def test_activity_json_decode_failure_fails_task(self, mock_client: AsyncMock) -> None:
+        worker = Worker(mock_client, task_queue="q1", workflows=[], activities=[echo_activity])
+        task = {
+            "task_id": "at-bad-json",
+            "activity_attempt_id": "aa-bad-json",
+            "activity_type": "test-act",
+            "arguments": "{not valid json",
+            "payload_codec": "json",
+        }
+        await worker._run_activity_task(task)
+        mock_client.fail_activity_task.assert_called_once()
+        call_kwargs = mock_client.fail_activity_task.call_args.kwargs
+        assert "decode" in call_kwargs["message"].lower()
+        assert "json" in call_kwargs["message"]
+        assert call_kwargs["non_retryable"] is True
+        mock_client.complete_activity_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_activity_avro_decode_failure_fails_task(self, mock_client: AsyncMock) -> None:
+        pytest.importorskip("avro", reason="avro extra not installed")
+        worker = Worker(mock_client, task_queue="q1", workflows=[], activities=[echo_activity])
+        task = {
+            "task_id": "at-bad-avro",
+            "activity_attempt_id": "aa-bad-avro",
+            "activity_type": "test-act",
+            "arguments": "!!!not-valid-base64!!!",
+            "payload_codec": "avro",
+        }
+        await worker._run_activity_task(task)
+        mock_client.fail_activity_task.assert_called_once()
+        call_kwargs = mock_client.fail_activity_task.call_args.kwargs
+        assert "decode" in call_kwargs["message"].lower()
+        assert "avro" in call_kwargs["message"]
+        assert call_kwargs["non_retryable"] is True
+        mock_client.complete_activity_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_activity_avro_missing_extra_fails_task(
+        self, mock_client: AsyncMock, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from durable_workflow import _avro
+        from durable_workflow.errors import AvroNotInstalledError
+
+        def _raise_missing(_blob: str) -> None:
+            raise AvroNotInstalledError(
+                "The 'avro' package is required to encode/decode payloads with the 'avro' "
+                "codec. Install with: pip install 'durable-workflow[avro]'"
+            )
+
+        monkeypatch.setattr(_avro, "decode", _raise_missing)
+
+        worker = Worker(mock_client, task_queue="q1", workflows=[], activities=[echo_activity])
+        task = {
+            "task_id": "at-no-avro",
+            "activity_attempt_id": "aa-no-avro",
+            "activity_type": "test-act",
+            "arguments": "anything",
+            "payload_codec": "avro",
+        }
+        await worker._run_activity_task(task)
+        mock_client.fail_activity_task.assert_called_once()
+        call_kwargs = mock_client.fail_activity_task.call_args.kwargs
+        assert "avro extra" in call_kwargs["message"].lower() or "pip install" in call_kwargs["message"]
+        assert call_kwargs["failure_type"] == "AvroNotInstalledError"
+        assert call_kwargs["non_retryable"] is True
+        mock_client.complete_activity_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_workflow_json_decode_failure_fails_task(self, mock_client: AsyncMock) -> None:
+        worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
+        task = {
+            "task_id": "t-bad-json",
+            "workflow_type": "test-wf",
+            "workflow_task_attempt": 1,
+            "history_events": [],
+            "arguments": "{not valid json",
+            "payload_codec": "json",
+        }
+        await worker._run_workflow_task(task)
+        mock_client.fail_workflow_task.assert_called_once()
+        call_kwargs = mock_client.fail_workflow_task.call_args.kwargs
+        assert "decode" in call_kwargs["message"].lower()
+        assert "json" in call_kwargs["message"]
+        mock_client.complete_workflow_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_workflow_avro_missing_extra_fails_task(
+        self, mock_client: AsyncMock, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from durable_workflow import _avro
+        from durable_workflow.errors import AvroNotInstalledError
+
+        def _raise_missing(_blob: str) -> None:
+            raise AvroNotInstalledError(
+                "The 'avro' package is required to encode/decode payloads with the 'avro' "
+                "codec. Install with: pip install 'durable-workflow[avro]'"
+            )
+
+        monkeypatch.setattr(_avro, "decode", _raise_missing)
+
+        worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
+        task = {
+            "task_id": "t-no-avro",
+            "workflow_type": "test-wf",
+            "workflow_task_attempt": 1,
+            "history_events": [],
+            "arguments": "anything",
+            "payload_codec": "avro",
+        }
+        await worker._run_workflow_task(task)
+        mock_client.fail_workflow_task.assert_called_once()
+        call_kwargs = mock_client.fail_workflow_task.call_args.kwargs
+        assert "avro extra" in call_kwargs["message"].lower() or "pip install" in call_kwargs["message"]
+        assert call_kwargs["failure_type"] == "AvroNotInstalledError"
+        mock_client.complete_workflow_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_workflow_replay_avro_missing_extra_fails_task(
+        self, mock_client: AsyncMock, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Avro-encoded history result that cannot be decoded (extra missing)
+        surfaces as fail_workflow_task, not an unhandled dispatcher exception."""
+        from durable_workflow import _avro
+        from durable_workflow.errors import AvroNotInstalledError
+
+        def _raise_missing(_blob: str) -> None:
+            raise AvroNotInstalledError(
+                "The 'avro' package is required to encode/decode payloads with the 'avro' "
+                "codec. Install with: pip install 'durable-workflow[avro]'"
+            )
+
+        monkeypatch.setattr(_avro, "decode", _raise_missing)
+
+        worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
+        # JSON envelope for start args bypasses the Avro path so the replay
+        # decode of history result (under the run's avro codec) is the site
+        # that triggers AvroNotInstalledError.
+        task = {
+            "task_id": "t-replay-no-avro",
+            "workflow_type": "test-wf",
+            "workflow_task_attempt": 1,
+            "history_events": [
+                {"event_type": "ActivityCompleted", "payload": {"result": "anything"}},
+            ],
+            "arguments": {"codec": "json", "blob": '["hello"]'},
+            "payload_codec": "avro",
+        }
+        await worker._run_workflow_task(task)
+        mock_client.fail_workflow_task.assert_called_once()
+        call_kwargs = mock_client.fail_workflow_task.call_args.kwargs
+        assert call_kwargs["failure_type"] == "AvroNotInstalledError"
+        mock_client.complete_workflow_task.assert_not_called()
+
+
 class TestWorkerStop:
     @pytest.mark.asyncio
     async def test_stop_sets_event(self, mock_client: AsyncMock) -> None: