Fix blank failure reports and prepare Python SDK 2.12.3

CHANGELOG.md

@@ -6,6 +6,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.12.3] - 2026-11-02
+
+### Fixed
+
+- Report exceptions with empty or whitespace-only messages using meaningful
+  fallback text, so workflow and activity failures reach callers instead of
+  being rejected as invalid failure reports. Preserve the exception type,
+  stack trace and task identity in ordinary and supervised activity workers.
+- Apply the same blank-message handling to invocable activities, queries and
+  update validators while keeping nonblank diagnostic text unchanged.
+
 ## [2.12.1] - 2026-10-29
 
 ### Fixed

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.12.2"
+version = "2.12.3"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -79,8 +79,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.12.2"
-registry-version = "2.12.2"
+product-train = "2.12.3"
+registry-version = "2.12.3"
 supported-server-versions = "2.8.0"
 worker-protocol-version = "1.20"
 control-plane-version = "2"

src/durable_workflow/_activity_process.py

@@ -24,6 +24,7 @@
 from pathlib import Path
 from typing import Any, cast
 
+from ._failure_reporting import exception_message
 from .activity import ActivityContext, ActivityInfo, _set_context
 from .errors import ActivityCancelled, NonRetryableError
 from .interceptors import ActivityInterceptorContext, WorkerInterceptor
@@ -139,7 +140,7 @@ async def intercepted(
         except BaseException as error:
             code = getattr(error, "code", None)
             _write_payload(directory, "outcome", CallbackOutcome(failure=CallbackFailure(
-                message=str(error), failure_type=type(error).__name__,
+                message=exception_message(error), failure_type=type(error).__name__,
                 failure_class=f"{type(error).__module__}.{type(error).__qualname__}",
                 failure_code=code if isinstance(code, int) and not isinstance(code, bool) else None,
                 stack_trace=traceback.format_exc(), non_retryable=isinstance(error, NonRetryableError),

src/durable_workflow/_failure_reporting.py

@@ -26,6 +26,12 @@
 })
 
 
+def exception_message(error: BaseException, *, fallback: str | None = None) -> str:
+    """Keep useful diagnostic text; blank exceptions still need a valid report."""
+    message = str(error)
+    return message if message.strip() else fallback or type(error).__name__
+
+
 def encoded_json(value: Any) -> bytes:
     """Use HTTPX's actual JSON settings, including escaping and NaN refusal."""
     return httpx.Request("POST", "https://failure.invalid", json=value).content

src/durable_workflow/invocable.py

@@ -12,6 +12,7 @@
 from typing import Any, cast
 
 from . import serializer
+from ._failure_reporting import exception_message
 from .errors import ActivityCancelled, NonRetryableError
 from .external_storage import ExternalPayloadCache, ExternalStorageDriver
 from .external_task_input import ExternalTaskInput, parse_external_task_input
@@ -105,7 +106,7 @@ async def handle(self, envelope: Mapping[str, Any]) -> dict[str, Any]:
                 started,
                 kind="application",
                 classification="application_error",
-                message=str(exc),
+                message=exception_message(exc),
                 failure_type=type(exc).__name__,
                 stack_trace=traceback.format_exc(),
                 retryable=False,
@@ -116,7 +117,7 @@ async def handle(self, envelope: Mapping[str, Any]) -> dict[str, Any]:
                 started,
                 kind="timeout",
                 classification="deadline_exceeded",
-                message=str(exc) or "Invocable activity handler exceeded its deadline.",
+                message=exception_message(exc, fallback="Invocable activity handler exceeded its deadline."),
                 failure_type=type(exc).__name__,
                 stack_trace=traceback.format_exc(),
                 timeout_type="deadline_exceeded",
@@ -140,7 +141,7 @@ async def handle(self, envelope: Mapping[str, Any]) -> dict[str, Any]:
                 started,
                 kind="application",
                 classification="application_error",
-                message=str(exc),
+                message=exception_message(exc),
                 failure_type=type(exc).__name__,
                 stack_trace=traceback.format_exc(),
                 retryable=True,

src/durable_workflow/worker.py

@@ -44,7 +44,7 @@
 from ._activity_process import CallbackFailure, CallbackInvocation, CallbackProcessLost, SupervisedCallback
 from ._cancellation_scope_history import CancellationScopeBudget, CancellationScopeDeliveryReceipt
 from ._cooperative_cancellation import CancellationRequest, read_cancellation_history
-from ._failure_reporting import REQUEST_BYTES, normalize_local_failure
+from ._failure_reporting import REQUEST_BYTES, exception_message, normalize_local_failure
 from ._prepared_local_activity import PreparedAttempt, PreparedCancellationObserved, PreparedLocalRunner
 from ._query_task_authority import QueryTaskAuthority, current_query_authority
 from ._sticky_workflow_cache import CacheKey, StickyWorkflowCache, complete_history
@@ -3240,7 +3240,7 @@ async def _run_activity_task_body(self, task: dict[str, Any]) -> str:
         except NonRetryableError as e:
             log.exception("activity failed (non-retryable)")
             await self._report_activity_task_failure(
-                task_id, attempt_id, "activity failure", message=str(e), failure_type=type(e).__name__,
+                task_id, attempt_id, "activity failure", message=exception_message(e), failure_type=type(e).__name__,
                 failure_class=_exception_class_name(e), failure_code=_exception_code(e),
                 stack_trace=traceback.format_exc(), non_retryable=True,
             )
@@ -3271,7 +3271,8 @@ async def _handle_activity_exception(self, task_id: str, attempt_id: str, error:
             raise error
         log.error("activity failed", exc_info=(type(error), error, error.__traceback__))
         await self._report_activity_task_failure(
-            task_id, attempt_id, "activity failure", message=str(error), failure_type=type(error).__name__,
+            task_id, attempt_id, "activity failure", message=exception_message(error),
+            failure_type=type(error).__name__,
             failure_class=_exception_class_name(error), failure_code=_exception_code(error),
             stack_trace="".join(traceback.format_exception(type(error), error, error.__traceback__)),
         )
@@ -3573,7 +3574,7 @@ async def execute() -> Any:
             )
             return "failed"
         except QueryFailed as e:
-            message = str(e)
+            message = exception_message(e)
             if "unknown query" in message:
                 reason = "rejected_unknown_query"
             elif message.startswith("workflow replay failed before query:"):
@@ -3594,7 +3595,7 @@ async def execute() -> Any:
             await self._fail_query_task(
                 query_task_id,
                 attempt,
-                str(e) or "query execution failed",
+                exception_message(e, fallback="query execution failed"),
                 reason="query_rejected",
                 failure_type=type(e).__name__,
                 stack_trace=traceback.format_exc(),
@@ -3631,7 +3632,7 @@ async def execute() -> Any:
             await self._fail_query_task(
                 query_task_id,
                 attempt,
-                str(e) or "query result completion was rejected by the server",
+                exception_message(e, fallback="query result completion was rejected by the server"),
                 reason=server_reason if server_reason else "query_result_completion_failed",
                 failure_type=type(e).__name__,
                 stack_trace=traceback.format_exc(),
@@ -3667,7 +3668,7 @@ async def execute() -> Any:
             await self._fail_query_task(
                 query_task_id,
                 attempt,
-                str(e) or "query result completion failed",
+                exception_message(e, fallback="query result completion failed"),
                 reason="query_result_completion_failed",
                 failure_type=type(e).__name__,
                 stack_trace=traceback.format_exc(),
@@ -4075,7 +4076,7 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str:
             return await self._reject_update_validation_task(
                 task_id,
                 attempt,
-                str(exc) or "update validator rejected the request",
+                exception_message(exc, fallback="update validator rejected the request"),
                 reason="update_validator_rejected",
                 failure_type=exc.exception_type or type(exc).__name__,
                 validation_errors=exc.validation_errors,
@@ -4085,7 +4086,7 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str:
             return await self._reject_update_validation_task(
                 task_id,
                 attempt,
-                str(exc),
+                exception_message(exc),
                 reason="update_validation_worker_failed",
                 failure_type=type(exc).__name__,
                 stack_trace=traceback.format_exc(),
@@ -4094,7 +4095,7 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str:
             return await self._reject_update_validation_task(
                 task_id,
                 attempt,
-                str(exc) or "update validator rejected the request",
+                exception_message(exc, fallback="update validator rejected the request"),
                 reason="update_validator_rejected",
                 failure_type=type(exc).__name__,
                 stack_trace=traceback.format_exc(),

src/durable_workflow/workflow.py

@@ -43,6 +43,7 @@
     ScopeBoundary,
 )
 from ._cooperative_cancellation import CancellationDelivery, read_cancellation_history
+from ._failure_reporting import exception_message
 from .cancellation import (
     CancellationContext,
     CancellationPolicy,
@@ -3292,7 +3293,7 @@ def invoke() -> Any:
             raise
         except Exception as exc:
             state.outcome.close()
-            raise QueryFailed(str(exc) or f"query {query_name!r} failed") from exc
+            raise QueryFailed(exception_message(exc, fallback=f"query {query_name!r} failed")) from exc
         except BaseException:
             state.outcome.close()
             raise
@@ -3594,7 +3595,7 @@ def _invoke_update_validator(
     except Exception as exc:
         errors = exc.errors if isinstance(exc, InvalidArgument) else None
         raise UpdateRejected(
-            str(exc) or f"update validator {update_name!r} rejected the request",
+            exception_message(exc, fallback=f"update validator {update_name!r} rejected the request"),
             exception_type=type(exc).__name__,
             validation_errors=errors,
         ) from exc
@@ -3614,7 +3615,7 @@ def _accepted_update_payload(
 
 
 def _fail_update_from_exception(update_id: str, prefix: str, exc: Exception) -> FailUpdate:
-    message = str(exc) or type(exc).__name__
+    message = exception_message(exc)
     return FailUpdate(
         update_id=update_id,
         message=f"{prefix}: {message}",
@@ -3628,7 +3629,7 @@ def _exception_class_name(exc: BaseException) -> str:
 
 
 def _fail_workflow_from_exception(exc: BaseException, *, prefix: str | None = None) -> FailWorkflow:
-    message = str(exc) or type(exc).__name__
+    message = exception_message(exc)
     if prefix:
         message = f"{prefix}: {message}"
 
@@ -3645,12 +3646,12 @@ def _fail_workflow_from_exception(exc: BaseException, *, prefix: str | None = No
             "initiating_failure": {
                 "type": type(initiating).__name__,
                 "class": _exception_class_name(initiating),
-                "message": str(initiating) or type(initiating).__name__,
+                "message": exception_message(initiating),
             },
             "compensation_failure": {
                 "type": type(compensation).__name__,
                 "class": _exception_class_name(compensation),
-                "message": str(compensation) or type(compensation).__name__,
+                "message": exception_message(compensation),
                 "activity_type": exc.compensation_activity_type,
                 "registration_order": exc.compensation_registration_order,
             },

tests/fixtures/replay_regressions/blank-workflow-failure-message.json

@@ -0,0 +1,15 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "blank-workflow-failure-message",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.replay.blank-failure", "input": [" \t\n"], "payload_codec": "avro"},
+  "history": [{"event_type": "WorkflowStarted", "payload": {}}],
+  "expected": {
+    "type": "fail_workflow",
+    "message": "AssertionError",
+    "exception_type": "AssertionError",
+    "exception_class": "builtins.AssertionError"
+  }
+}

tests/integration/test_empty_callback_failure.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from typing import Any
+
+import httpx
+import pytest
+
+from durable_workflow import Client, Worker, activity
+from durable_workflow.errors import WorkflowFailed
+from tests.worker_observation import stop_and_drain
+
+
+@activity.defn(name="tests.empty-failure")
+async def empty_failure(message: str) -> None:
+    raise AssertionError(message)
+
+
+@pytest.mark.parametrize("supervised", [False, True])
+@pytest.mark.parametrize("message", ["", " \t\n"])
+async def test_blank_callback_failure_reaches_a_terminal_caller(
+    server_url: str, server_token: str, supervised: bool, message: str,
+) -> None:
+    identity = "empty-failure-" + uuid.uuid4().hex[:12]
+    replies: list[tuple[int, dict[str, Any], dict[str, Any]]] = []
+    async with Client(server_url, worker_token=server_token, control_token=server_token) as client:
+        async def record(response: httpx.Response) -> None:
+            if response.request.url.path.endswith("/fail"):
+                await response.aread()
+                replies.append((response.status_code, json.loads(response.request.content), response.json()))
+
+        client._http.event_hooks["response"].append(record)
+        worker = Worker(client, task_queue=identity, worker_id=identity, activities=[empty_failure],
+                        capabilities=["cooperative_cancellation"] if supervised else [])
+        runner = asyncio.create_task(worker.run())
+        try:
+            handle = await client.start_activity(activity_type="tests.empty-failure", task_queue=identity,
+                activity_id=identity, input=[message], retry_policy={"max_attempts": 1})
+            with pytest.raises(WorkflowFailed):
+                await handle.result(timeout=15, poll_interval=.05)
+            assert (await handle.describe()).status == "failed"
+            assert len(replies) == 1
+            status, submitted, accepted = replies[0]
+            assert status == 200
+            assert accepted["recorded"] is True
+            assert submitted["lease_owner"] == identity
+            assert submitted["activity_attempt_id"]
+            assert submitted["failure"]["message"] == "AssertionError"
+            assert submitted["failure"]["type"] == "AssertionError"
+            assert "empty_failure" in submitted["failure"]["stack_trace"]
+        finally:
+            await stop_and_drain([worker], [runner])