Report persisted workflow deadlines as typed terminal timeouts

src/durable_workflow/__init__.py

@@ -95,6 +95,7 @@
     WorkflowNotFound,
     WorkflowPayloadDecodeError,
     WorkflowTerminated,
+    WorkflowTimedOut,
 )
 from .external_storage import (
     EXTERNAL_PAYLOAD_REFERENCE_SCHEMA,
@@ -374,6 +375,7 @@
     "WorkflowFailed",
     "WorkflowNotFound",
     "WorkflowTerminated",
+    "WorkflowTimedOut",
     "EXTERNAL_TASK_INPUT_CONTRACT_SCHEMA",
     "EXTERNAL_TASK_INPUT_MEDIA_TYPE",
     "EXTERNAL_TASK_INPUT_SCHEMA",

src/durable_workflow/client.py

@@ -46,6 +46,7 @@
     WorkflowCancelled,
     WorkflowFailed,
     WorkflowTerminated,
+    WorkflowTimedOut,
     _raise_for_status,
 )
 from .external_storage import (
@@ -4141,8 +4142,9 @@ async def get_result(
         Raises :class:`~durable_workflow.errors.WorkflowFailed`,
         :class:`~durable_workflow.errors.WorkflowCancelled`, or
         :class:`~durable_workflow.errors.WorkflowTerminated` if the workflow
-        ended in a non-success state, or :class:`TimeoutError` if ``timeout``
-        seconds elapse before the workflow terminates.
+        ended in a non-success state. A persisted execution or run deadline
+        raises :class:`~durable_workflow.errors.WorkflowTimedOut`; a caller's
+        :class:`TimeoutError` means ``timeout`` seconds elapsed while waiting.
         """
         deadline = asyncio.get_running_loop().time() + timeout
         while True:
@@ -4177,6 +4179,8 @@ async def get_result(
                         raise WorkflowCancelled(
                             payload.get("reason", "workflow was cancelled")
                         )
+                    if etype == "WorkflowTimedOut":
+                        raise WorkflowTimedOut()
                 return None
             if asyncio.get_running_loop().time() > deadline:
                 raise TimeoutError(

src/durable_workflow/errors.py

@@ -560,6 +560,17 @@ def __init__(self, message: str = "workflow was terminated") -> None:
         super().__init__(message)
 
 
+class WorkflowTimedOut(DurableWorkflowError):
+    """A persisted workflow execution or run deadline expired.
+
+    Unlike a caller's :class:`TimeoutError` while polling, this is a terminal
+    workflow outcome recorded by the runtime.
+    """
+
+    def __init__(self, message: str = "workflow execution timed out") -> None:
+        super().__init__(message)
+
+
 class SagaCompensationFailed(DurableWorkflowError):
     """A saga compensation failed after an earlier workflow failure.
 

tests/test_workflow_result_timeout.py

@@ -0,0 +1,61 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+from durable_workflow import Client, WorkflowTimedOut
+from durable_workflow.client import WorkflowExecution, WorkflowHandle
+from durable_workflow.errors import WorkflowCancelled, WorkflowFailed, WorkflowTerminated
+
+
+def result_client(status: str, event_type: str, payload: dict) -> tuple[Client, WorkflowHandle]:
+    client = Client("https://unused.example")
+    client.describe_workflow = AsyncMock(
+        return_value=WorkflowExecution(
+            workflow_id="order", run_id="current-run", workflow_type="order", status=status,
+        )
+    )
+    client.get_history = AsyncMock(
+        return_value={"events": [{"event_type": event_type, "payload": payload}]}
+    )
+    return client, WorkflowHandle(client, workflow_id="order", run_id="selected-run", workflow_type="order")
+
+
+@pytest.mark.parametrize("kind", ["execution_timeout", "run_timeout"])
+async def test_persisted_deadline_raises_typed_timeout_for_selected_history(kind: str) -> None:
+    client, handle = result_client(
+        "failed", "WorkflowTimedOut", {"timeout_kind": kind, "deadline_at": "2026-01-01T00:00:00Z"},
+    )
+    try:
+        with pytest.raises(WorkflowTimedOut, match="workflow execution timed out"):
+            await client.get_result(handle, timeout=0)
+        client.get_history.assert_awaited_once_with("order", "selected-run")
+    finally:
+        await client.aclose()
+
+
+async def test_polling_timeout_is_not_a_terminal_workflow_timeout() -> None:
+    client, handle = result_client("waiting", "ConditionWaitOpened", {})
+    try:
+        with pytest.raises(TimeoutError, match="not terminal") as caught:
+            await client.get_result(handle, timeout=0)
+        assert not isinstance(caught.value, WorkflowTimedOut)
+        client.get_history.assert_not_awaited()
+    finally:
+        await client.aclose()
+
+
+@pytest.mark.parametrize(
+    ("status", "event_type", "exception"),
+    [
+        ("failed", "WorkflowFailed", WorkflowFailed),
+        ("cancelled", "WorkflowCancelled", WorkflowCancelled),
+        ("terminated", "WorkflowTerminated", WorkflowTerminated),
+    ],
+)
+async def test_other_terminal_failures_preserve_their_exception(status, event_type, exception) -> None:
+    client, handle = result_client(status, event_type, {"message": "stopped", "reason": "stopped"})
+    try:
+        with pytest.raises(exception, match="stopped"):
+            await client.get_result(handle, timeout=0)
+    finally:
+        await client.aclose()