Read the timeout-cleanup cause on cancellation contexts

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Timeout cleanup interruption (workflow#863, workflow#1129). `CancellationContext.from_dict()` reads the context's optional top-level `cause`: absent or `"cancellation"` is an ordinary cancellation, `"timeout"` marks a timeout-cleanup request (the run passed its work deadline and is now in cleanup), and any other value raises `ValueError`. The context carries no other timeout fields: its existing `cleanup_deadline_at` is the cleanup deadline, and the timeout kind and work deadline stay on the `WorkflowTimeoutCleanupStarted` event. `cause` and `is_timeout` are readable on the context, and `to_dict()` writes `cause` back exactly as received, including an explicit `"cancellation"`, leaving ordinary contexts byte-identical. A timeout request is raised at the authored call as the new experimental `WorkflowTimeoutCleanupRequested`, a subclass of `WorkflowCancelled`, so existing `except WorkflowCancelled:` handlers, such as saga compensation, run on timeout too. It stays `BaseException`-rooted and is not an `asyncio.CancelledError`. Test `cause`, the cross-SDK discriminator, to tell the two apart. `cause` is never inherited: a child cancelled under a timeout request, scoped or not, receives an ordinary cancellation without `cause`.
+
 ## [2.17.1] - 2026-11-13
 
 ### Fixed

docs/cooperative-cancellation-design.md

@@ -33,6 +33,25 @@ metadata is limited to caller type, ID and label. It is a read-only mapping and
 lineage is a tuple of immutable `CancellationLineage` entries. Timestamp values
 are immutable UTC dates. `to_dict()` returns a detached portable snapshot.
 
+### Timeout cleanup cause
+
+A run that opts in to timeout cleanup (workflow issue 863) is interrupted at its
+work deadline through the same cooperative delivery. The context then has
+`cause == "timeout"` (`is_timeout` is true), and the delivered exception is
+`WorkflowTimeoutCleanupRequested`, a subclass of `WorkflowCancelled`. Existing
+`except WorkflowCancelled:` handlers therefore run on timeout too. Like
+`WorkflowCancelled`, it derives from `BaseException`, never from
+`asyncio.CancelledError`. Test `cause`, the cross-SDK discriminator, rather than
+the type, to tell the two apart. The run still ends timed out.
+
+`cause` is `None` (absent) or `"cancellation"` for an ordinary request; any
+other value is refused. `cleanup_deadline_at` is the cleanup deadline. The
+context carries no other timeout fields: the timeout kind and work deadline are
+recorded on `WorkflowTimeoutCleanupStarted`. `cause` is never inherited: a child
+cancelled under a timeout request, scoped or not, receives an ordinary
+cancellation. `to_dict()` writes `cause` back exactly as received, including an
+explicit `"cancellation"`, and leaves ordinary contexts unchanged.
+
 A child preserves the root request time and budget even when its local request
 arrives later. Canonical request history supplies the context on cold replay,
 while transport observation retains the opaque refresh route. The parser rejects

src/durable_workflow/__init__.py

@@ -119,6 +119,7 @@
     WorkflowTaskCompletionTooLarge,
     WorkflowTerminated,
     WorkflowTimedOut,
+    WorkflowTimeoutCleanupRequested,
 )
 from .external_storage import (
     EXTERNAL_PAYLOAD_REFERENCE_SCHEMA,
@@ -428,6 +429,7 @@
     "WorkflowTaskCompletionTooLarge",
     "WorkflowTerminated",
     "WorkflowTimedOut",
+    "WorkflowTimeoutCleanupRequested",
     "EXTERNAL_TASK_INPUT_CONTRACT_SCHEMA",
     "EXTERNAL_TASK_INPUT_MEDIA_TYPE",
     "EXTERNAL_TASK_INPUT_SCHEMA",

src/durable_workflow/cancellation.py

@@ -68,6 +68,14 @@ def _text(snapshot: Mapping[str, Any], key: str) -> str:
     return value
 
 
+CANCELLATION_CAUSE = "cancellation"
+TIMEOUT_CAUSE = "timeout"
+
+
+def _format_timestamp(value: datetime) -> str:
+    return value.isoformat(timespec="microseconds").replace("+00:00", "Z")
+
+
 def _timestamp(value: str) -> datetime:
     if not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})", value):
         raise ValueError("cancellation timestamp requires an ISO date, time and timezone")
@@ -112,6 +120,8 @@ class CancellationContext:
     lineage: tuple[CancellationLineage, ...]
     _replay_clock: Callable[[], datetime] | None = field(default=None, repr=False, compare=False)
     scope_origin: ScopedCancellationContext | None = None
+    cause: str | None = None
+    """``"timeout"`` for a timeout-cleanup request, else ``None`` or ``"cancellation"`` (ordinary)."""
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "requester", MappingProxyType(dict(self.requester)))
@@ -122,6 +132,11 @@ def deadline(self) -> datetime:
         """The original immutable cleanup deadline."""
         return self.cleanup_deadline_at
 
+    @property
+    def is_timeout(self) -> bool:
+        """Whether the run's work deadline passed and this request opens its timeout cleanup."""
+        return self.cause == TIMEOUT_CAUSE
+
     def remaining(self) -> float:
         """Seconds left at the consumed replay boundary, clamped to zero.
 
@@ -194,10 +209,13 @@ def from_dict(cls, snapshot: Mapping[str, Any]) -> CancellationContext:
             raise ValueError("cancellation deadline must follow the original request")
         if scope_origin is not None:
             _assert_scope_origin(snapshot, scope_origin, normalized, normalized_requester, requested_at, deadline)
+        cause = snapshot.get("cause")
+        if cause is not None and cause not in (CANCELLATION_CAUSE, TIMEOUT_CAUSE):
+            raise ValueError("cancellation cause must be cancellation or timeout")
         return cls(
             request_id, root_request_id, root_instance_id, root_run_id, parent_request_id,
             reason, normalized_requester, _text(snapshot, "source"), requested_at, deadline, tuple(normalized),
-            scope_origin=scope_origin,
+            scope_origin=scope_origin, cause=cause,
         )
 
     def to_dict(self) -> dict[str, Any]:
@@ -213,10 +231,12 @@ def to_dict(self) -> dict[str, Any]:
             "reason": self.reason,
             "requester": dict(self.requester),
             "source": self.source,
-            "requested_at": self.requested_at.isoformat(timespec="microseconds").replace("+00:00", "Z"),
-            "cleanup_deadline_at": self.cleanup_deadline_at.isoformat(timespec="microseconds").replace("+00:00", "Z"),
+            "requested_at": _format_timestamp(self.requested_at),
+            "cleanup_deadline_at": _format_timestamp(self.cleanup_deadline_at),
             "lineage": [entry.to_dict() for entry in self.lineage],
         }
+        if self.cause is not None:
+            snapshot["cause"] = self.cause
         if self.scope_origin is not None:
             snapshot["scope_origin"] = self.scope_origin.to_dict()
             snapshot["scope_authority_deadline_at"] = snapshot["cleanup_deadline_at"]
@@ -239,7 +259,7 @@ def to_dict(self) -> dict[str, str]:
             "workflow_instance_id": self.workflow_instance_id,
             "workflow_run_id": self.workflow_run_id,
             "scope_id": self.scope_id,
-            "cleanup_deadline_at": self.cleanup_deadline_at.isoformat(timespec="microseconds").replace("+00:00", "Z"),
+            "cleanup_deadline_at": _format_timestamp(self.cleanup_deadline_at),
         }
 
 

src/durable_workflow/errors.py

@@ -970,6 +970,38 @@ def __init__(
         self.context = context
 
 
+class WorkflowTimeoutCleanupRequested(WorkflowCancelled):
+    """The run's work deadline passed and its timeout cleanup has started.
+
+    Raised at the authored call, like a cooperative cancellation request, when a
+    run that opted in to timeout cleanup reaches its work deadline. It is a
+    :class:`WorkflowCancelled`, so existing ``except WorkflowCancelled:``
+    handlers (saga compensation, cleanup) run unchanged. The run still ends
+    timed out, whatever the cleanup returns.
+
+    Like :class:`WorkflowCancelled` it is :class:`BaseException`-rooted, and it
+    is not an :class:`asyncio.CancelledError`. ``context.cause`` is
+    ``"timeout"``; test ``cause`` rather than the type when telling the two
+    apart, since ``cause`` is the cross-SDK discriminator. ``context.cleanup_deadline_at``
+    is the cleanup deadline. Experimental.
+    """
+
+    def __init__(
+        self, message: str = "workflow work deadline passed: timeout cleanup requested", *,
+        request_id: str | None = None, context: CancellationContext | ScopedCancellationContext | None = None,
+    ) -> None:
+        super().__init__(message, request_id=request_id, context=context)
+
+
+def _cancellation_requested(
+    message: str, *, request_id: str | None, context: CancellationContext | ScopedCancellationContext | None,
+) -> WorkflowCancelled:
+    """The exception for a delivered run request: the timeout subtype when timeout cleanup requested it."""
+    if isinstance(context, CancellationContext) and context.is_timeout:
+        return WorkflowTimeoutCleanupRequested(request_id=request_id, context=context)
+    return WorkflowCancelled(message, request_id=request_id, context=context)
+
+
 class ActivityCancelled(BaseException):
     """An in-flight activity was cancelled.
 

src/durable_workflow/workflow.py

@@ -74,6 +74,7 @@
     WorkflowCancelled,
     WorkflowEntryArgumentsMismatchError,
     WorkflowPayloadDecodeError,
+    _cancellation_requested,
 )
 from .external_storage import ExternalPayloadCache, ExternalStorageDriver
 from .nexus import NexusOperationResult, stable_nexus_idempotency_key
@@ -2359,7 +2360,7 @@ def is_cancellation_requested(self) -> bool:
     def throw_if_cancellation_requested(self) -> None:
         """Raise :class:`WorkflowCancelled` at an explicit safe point."""
         if self._cancel_requested and self._cancellation_shield_depth == 0:
-            raise WorkflowCancelled(
+            raise _cancellation_requested(
                 "workflow cancellation was requested", request_id=self._cancellation_request_id,
                 context=self._cancellation_context,
             )
@@ -7161,7 +7162,7 @@ def _consume_cancellation(command: Any, boundary: CancellationDelivery) -> _Repl
             authored_sequence += 1
         try:
             _refresh_replay_logging()
-            advanced_cmd = _resume_business(gen.throw, WorkflowCancelled(
+            advanced_cmd = _resume_business(gen.throw, _cancellation_requested(
                 "workflow cancellation was requested", request_id=boundary.request_id,
                 context=ctx._cancellation_context,
             ))

tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json

@@ -0,0 +1,108 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "cooperative-timeout-cleanup-cause",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.cooperative-timeout-cleanup-cause",
+    "input": [],
+    "payload_codec": "avro",
+    "run_id": "run-1"
+  },
+  "history": [
+    {
+      "event_type": "TimerScheduled",
+      "payload": {
+        "sequence": 1,
+        "timer_kind": "durable_timer",
+        "delay_seconds": 120
+      }
+    },
+    {
+      "event_type": "CooperativeCancellationRequested",
+      "workflow_command_id": "req-1",
+      "recorded_at": "2026-11-13T00:01:00Z",
+      "payload": {
+        "workflow_command_id": "req-1",
+        "workflow_instance_id": "instance-1",
+        "workflow_run_id": "run-1",
+        "reason": "work deadline passed",
+        "cleanup_deadline_at": "2026-11-13T00:01:10.000000Z",
+        "cancellation": {
+          "schema": "durable-workflow.cancellation-context/v1",
+          "request_id": "req-1",
+          "root_request_id": "req-1",
+          "root_workflow_instance_id": "instance-1",
+          "root_workflow_run_id": "run-1",
+          "parent_request_id": null,
+          "reason": "work deadline passed",
+          "requester": {
+            "type": "system",
+            "id": "timeout_cleanup"
+          },
+          "source": "system",
+          "requested_at": "2026-11-13T00:01:00.000000Z",
+          "cleanup_deadline_at": "2026-11-13T00:01:10.000000Z",
+          "lineage": [
+            {
+              "request_id": "req-1",
+              "workflow_instance_id": "instance-1",
+              "workflow_run_id": "run-1"
+            }
+          ],
+          "cause": "timeout"
+        }
+      }
+    },
+    {
+      "event_type": "CooperativeCancellationDelivered",
+      "workflow_command_id": "req-1",
+      "payload": {
+        "workflow_command_id": "req-1",
+        "workflow_run_id": "run-1",
+        "sequence": 1,
+        "call_kind": "timer",
+        "cancellation": {
+          "schema": "durable-workflow.cancellation-context/v1",
+          "request_id": "req-1",
+          "root_request_id": "req-1",
+          "root_workflow_instance_id": "instance-1",
+          "root_workflow_run_id": "run-1",
+          "parent_request_id": null,
+          "reason": "work deadline passed",
+          "requester": {
+            "type": "system",
+            "id": "timeout_cleanup"
+          },
+          "source": "system",
+          "requested_at": "2026-11-13T00:01:00.000000Z",
+          "cleanup_deadline_at": "2026-11-13T00:01:10.000000Z",
+          "lineage": [
+            {
+              "request_id": "req-1",
+              "workflow_instance_id": "instance-1",
+              "workflow_run_id": "run-1"
+            }
+          ],
+          "cause": "timeout"
+        }
+      }
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "complete_workflow",
+        "command_type": "CompleteWorkflow",
+        "result": [
+          "WorkflowTimeoutCleanupRequested",
+          "timeout",
+          "req-1"
+        ]
+      }
+    ]
+  }
+}