From 0fec214a6a0b6329c69be9099095fc0eb32b0084 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 06:20:37 +0000
Subject: [PATCH 1/3] Read the timeout-cleanup cause on cancellation contexts

CancellationContext.from_dict() reads the optional top-level `cause`
(absent/"cancellation" ordinary, "timeout" for timeout cleanup, anything
else refused) and, with the timeout cause only, `timeout_kind` and
`work_deadline_at`. to_dict() writes them back exactly and leaves
ordinary contexts byte-identical.

A timeout request is raised at the authored call as
WorkflowTimeoutCleanupRequested, a WorkflowCancelled subclass, so existing
cancellation handlers run on timeout too. It stays BaseException-rooted
and is not asyncio.CancelledError.

Refs durable-workflow/workflow#863, durable-workflow/workflow#1129.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |   2 +
 docs/cooperative-cancellation-design.md       |  19 ++
 src/durable_workflow/__init__.py              |   2 +
 src/durable_workflow/cancellation.py          |  58 ++++-
 src/durable_workflow/errors.py                |  34 +++
 src/durable_workflow/workflow.py              |   5 +-
 .../cooperative-timeout-cleanup-cause.json    | 113 +++++++++
 tests/test_cancellation_cause.py              | 224 ++++++++++++++++++
 tests/test_replay_regression_corpus.py        |  17 +-
 9 files changed, 467 insertions(+), 7 deletions(-)
 create mode 100644 tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json
 create mode 100644 tests/test_cancellation_cause.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d7162027..6226a7c2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Update outcome reads check the request and answer. An empty, `.` or `..` workflow or update id, an unknown `wait_for` or a non-positive `wait_timeout_seconds` raises `ValueError` before any request, including discovery, so `..` can never resolve onto the workflow's describe route. A successful answer must name the requested `update_id`, carry a string `update_status` and, when it echoes a `workflow_id`, match it. `update_outcome()` returns a value only for a completed receipt. Any other answer raises the new `UnrecognizedUpdateOutcome` with `workflow_id`, `update_id` and `body`, and never returns a value. With `wait_timeout_seconds`, the HTTP timeout becomes that wait plus 30 seconds whenever that is longer than the client's `timeout`, so a long Server-side wait is not cut off.
 
+- 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`. A timeout context may also carry `timeout_kind` (`"execution"` or `"run"`) and `work_deadline_at`; both are refused without the timeout cause. `cause`, `timeout_kind`, `work_deadline_at` and `is_timeout` are readable on the context, and `to_dict()` writes them back exactly, 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. Children cancelled under a timeout request receive an ordinary cancellation.
+
 ### Changed
 
 - The `Client.update_workflow()` and `WorkflowHandle.update()` docstrings now say what they always did: they return the Server receipt, not the decoded result, and the Server answers at `wait_for="accepted"` when it is omitted. A receipt with `update_status: "accepted"` is a pending update even though its `result` is `None`.
diff --git a/docs/cooperative-cancellation-design.md b/docs/cooperative-cancellation-design.md
index cffec18f..abf62a97 100644
--- a/docs/cooperative-cancellation-design.md
+++ b/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 hard cleanup deadline.
+`timeout_kind` (`"execution"` or `"run"`) and `work_deadline_at` are set when
+the request carries them, and are refused without the timeout cause. A child
+cancelled under a timeout request receives an ordinary cancellation: the cause
+is not inherited. `to_dict()` writes `cause` back exactly as received 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
diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py
index 17366958..c5b50942 100644
--- a/src/durable_workflow/__init__.py
+++ b/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",
diff --git a/src/durable_workflow/cancellation.py b/src/durable_workflow/cancellation.py
index d0c90415..3d8e9655 100644
--- a/src/durable_workflow/cancellation.py
+++ b/src/durable_workflow/cancellation.py
@@ -68,6 +68,15 @@ def _text(snapshot: Mapping[str, Any], key: str) -> str:
     return value
 
 
+CANCELLATION_CAUSE = "cancellation"
+TIMEOUT_CAUSE = "timeout"
+_TIMEOUT_KINDS = ("execution", "run")
+
+
+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 +121,12 @@ 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)."""
+    timeout_kind: str | None = None
+    """``"execution"`` or ``"run"`` when a timeout-cleanup request names the deadline that passed."""
+    work_deadline_at: datetime | None = None
+    """The work deadline that passed, when a timeout-cleanup request carries it."""
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "requester", MappingProxyType(dict(self.requester)))
@@ -122,6 +137,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 +214,11 @@ 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, timeout_kind, work_deadline_at = _cause(snapshot, deadline)
         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, timeout_kind=timeout_kind, work_deadline_at=work_deadline_at,
         )
 
     def to_dict(self) -> dict[str, Any]:
@@ -213,16 +234,45 @@ 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.timeout_kind is not None:
+            snapshot["timeout_kind"] = self.timeout_kind
+        if self.work_deadline_at is not None:
+            snapshot["work_deadline_at"] = _format_timestamp(self.work_deadline_at)
         if self.scope_origin is not None:
             snapshot["scope_origin"] = self.scope_origin.to_dict()
             snapshot["scope_authority_deadline_at"] = snapshot["cleanup_deadline_at"]
         return snapshot
 
 
+def _cause(snapshot: Mapping[str, Any], deadline: datetime) -> tuple[str | None, str | None, datetime | None]:
+    """Read the optional top-level cause and the timeout fields only a timeout cause may carry."""
+    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")
+    timeout_kind = snapshot.get("timeout_kind")
+    work_deadline = snapshot.get("work_deadline_at")
+    if cause != TIMEOUT_CAUSE:
+        if timeout_kind is not None or work_deadline is not None:
+            raise ValueError("cancellation timeout fields require the timeout cause")
+        return cause, None, None
+    if timeout_kind is not None and timeout_kind not in _TIMEOUT_KINDS:
+        raise ValueError("cancellation timeout_kind must be execution or run")
+    work_deadline_at = None
+    if work_deadline is not None:
+        if not isinstance(work_deadline, str):
+            raise ValueError("cancellation timestamp requires an ISO date, time and timezone")
+        work_deadline_at = _timestamp(work_deadline)
+        if work_deadline_at >= deadline:
+            raise ValueError("cancellation work deadline must precede its cleanup deadline")
+    return cause, timeout_kind, work_deadline_at
+
+
 @dataclass(frozen=True)
 class ScopedCancellationLineage:
     """One immutable original scope address and its bounded cleanup budget."""
@@ -239,7 +289,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),
         }
 
 
diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py
index 94509023..49f075cc 100644
--- a/src/durable_workflow/errors.py
+++ b/src/durable_workflow/errors.py
@@ -970,6 +970,40 @@ 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 hard cleanup deadline, and ``context.timeout_kind`` and
+    ``context.work_deadline_at`` are set when the request carries them.
+    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.
 
diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py
index 8999efe7..1812ba70 100644
--- a/src/durable_workflow/workflow.py
+++ b/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,
             )
@@ -7141,7 +7142,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,
             ))
diff --git a/tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json b/tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json
new file mode 100644
index 00000000..7094e271
--- /dev/null
+++ b/tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json
@@ -0,0 +1,113 @@
+{
+  "$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",
+          "timeout_kind": "execution",
+          "work_deadline_at": "2026-11-13T00:01:00.000000Z"
+        }
+      }
+    },
+    {
+      "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",
+          "timeout_kind": "execution",
+          "work_deadline_at": "2026-11-13T00:01:00.000000Z"
+        }
+      }
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "complete_workflow",
+        "command_type": "CompleteWorkflow",
+        "result": [
+          "WorkflowTimeoutCleanupRequested",
+          "timeout",
+          "execution",
+          "req-1"
+        ]
+      }
+    ]
+  }
+}
diff --git a/tests/test_cancellation_cause.py b/tests/test_cancellation_cause.py
new file mode 100644
index 00000000..c2f8cbdc
--- /dev/null
+++ b/tests/test_cancellation_cause.py
@@ -0,0 +1,224 @@
+"""The optional top-level ``cause`` of a cancellation context (workflow#863).
+
+Vectors ported from Core's ``tests/Unit/V2/CancellationContextCauseTest.php``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import datetime, timezone
+from typing import Any
+
+import pytest
+
+from durable_workflow import (
+    CancellationContext,
+    ScopedCancellationContext,
+    WorkflowCancelled,
+    WorkflowTimeoutCleanupRequested,
+)
+from durable_workflow.errors import _cancellation_requested
+from durable_workflow.workflow import CompleteWorkflow, FailWorkflow, WorkflowContext, replay
+
+
+def snapshot() -> dict[str, Any]:
+    return {
+        "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": None,
+        "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"}],
+    }
+
+
+def timeout_snapshot(**fields: Any) -> dict[str, Any]:
+    return {**snapshot(), "cause": "timeout", **fields}
+
+
+def test_an_ordinary_context_is_unchanged_and_stays_a_cancellation() -> None:
+    context = CancellationContext.from_dict(snapshot())
+
+    assert context.cause is None
+    assert not context.is_timeout
+    assert context.to_dict() == snapshot()
+    assert json.dumps(context.to_dict()) == json.dumps(snapshot())
+    assert type(_cancellation_requested("cancelled", request_id="req-1", context=context)) is WorkflowCancelled
+
+
+def test_an_explicit_cancellation_cause_is_ordinary_and_round_trips() -> None:
+    context = CancellationContext.from_dict({**snapshot(), "cause": "cancellation"})
+
+    assert context.cause == "cancellation"
+    assert not context.is_timeout
+    assert context.to_dict() == {**snapshot(), "cause": "cancellation"}
+    assert type(_cancellation_requested("cancelled", request_id="req-1", context=context)) is WorkflowCancelled
+
+
+def test_a_null_cause_is_an_ordinary_cancellation() -> None:
+    context = CancellationContext.from_dict({**snapshot(), "cause": None})
+
+    assert context.cause is None
+    assert context == CancellationContext.from_dict(snapshot())
+
+
+def test_a_timeout_cause_round_trips_and_delivers_the_timeout_subtype() -> None:
+    context = CancellationContext.from_dict(timeout_snapshot())
+
+    assert context.cause == "timeout"
+    assert context.is_timeout
+    assert context.timeout_kind is None and context.work_deadline_at is None
+    assert context.to_dict() == timeout_snapshot()
+    assert list(context.to_dict()).index("cause") == list(context.to_dict()).index("lineage") + 1
+    exception = _cancellation_requested("cancelled", request_id="req-1", context=context)
+    assert isinstance(exception, WorkflowTimeoutCleanupRequested)
+    assert isinstance(exception, WorkflowCancelled)
+    assert exception.context is context
+    assert exception.request_id == "req-1"
+
+
+def test_timeout_fields_round_trip() -> None:
+    value = timeout_snapshot(timeout_kind="execution", work_deadline_at="2026-11-13T00:01:00.000000Z")
+    context = CancellationContext.from_dict(value)
+
+    assert context.timeout_kind == "execution"
+    assert context.work_deadline_at == datetime(2026, 11, 13, 0, 1, tzinfo=timezone.utc)
+    assert context.cleanup_deadline_at == datetime(2026, 11, 13, 0, 1, 10, tzinfo=timezone.utc)
+    assert context.to_dict() == value
+    assert CancellationContext.from_dict(context.to_dict()) == context
+
+
+def test_the_timeout_subtype_is_base_exception_rooted_and_not_task_cancellation() -> None:
+    assert issubclass(WorkflowTimeoutCleanupRequested, WorkflowCancelled)
+    assert issubclass(WorkflowTimeoutCleanupRequested, BaseException)
+    assert not issubclass(WorkflowTimeoutCleanupRequested, Exception)
+    assert not issubclass(WorkflowTimeoutCleanupRequested, asyncio.CancelledError)
+
+
+def test_a_descendant_is_cancelled_not_timed_out() -> None:
+    # Built the way Core's forDescendant builds it: the cause is not inherited.
+    descendant = timeout_snapshot()
+    del descendant["cause"]
+    descendant.update({
+        "request_id": "req-child", "parent_request_id": "req-1",
+        "lineage": [*descendant["lineage"], {
+            "request_id": "req-child", "workflow_instance_id": "child-instance", "workflow_run_id": "child-run",
+        }],
+    })
+    context = CancellationContext.from_dict(descendant)
+
+    assert not context.is_timeout
+    assert context.to_dict() == descendant
+    assert type(_cancellation_requested("cancelled", request_id="req-child", context=context)) is WorkflowCancelled
+
+
+@pytest.mark.parametrize("cause", ["deadline", "Timeout", "", 1, True, ["timeout"], {"cause": "timeout"}])
+def test_an_unknown_cause_is_refused(cause: Any) -> None:
+    with pytest.raises(ValueError, match="cause must be cancellation or timeout"):
+        CancellationContext.from_dict({**snapshot(), "cause": cause})
+
+
+@pytest.mark.parametrize("cause", [None, "cancellation"])
+@pytest.mark.parametrize("field,value", [
+    ("timeout_kind", "execution"), ("work_deadline_at", "2026-11-13T00:01:00.000000Z"),
+])
+def test_timeout_fields_require_the_timeout_cause(cause: str | None, field: str, value: str) -> None:
+    ordinary = snapshot() if cause is None else {**snapshot(), "cause": cause}
+    with pytest.raises(ValueError, match="timeout fields require the timeout cause"):
+        CancellationContext.from_dict({**ordinary, field: value})
+
+
+@pytest.mark.parametrize("field,value,message", [
+    ("timeout_kind", "workflow", "timeout_kind must be execution or run"),
+    ("timeout_kind", 1, "timeout_kind must be execution or run"),
+    ("work_deadline_at", "2026-11-13 00:01:00", "ISO date, time and timezone"),
+    ("work_deadline_at", 1, "ISO date, time and timezone"),
+    ("work_deadline_at", "2026-11-13T00:01:10.000000Z", "work deadline must precede its cleanup deadline"),
+    ("work_deadline_at", "2026-11-13T00:02:00.000000Z", "work deadline must precede its cleanup deadline"),
+])
+def test_invalid_timeout_fields_are_refused(field: str, value: Any, message: str) -> None:
+    with pytest.raises(ValueError, match=message):
+        CancellationContext.from_dict(timeout_snapshot(**{field: value}))
+
+
+def test_a_scoped_root_keeps_the_run_request_cause() -> None:
+    # Core's ScopedCancellationContext::fromRunContext keeps the run context's keys.
+    scoped = ScopedCancellationContext.from_dict({
+        "schema": "durable-workflow.scoped-cancellation-context/v1",
+        "root_context": timeout_snapshot(),
+        "lineage": [{
+            "request_id": "req-1", "workflow_instance_id": "instance-1", "workflow_run_id": "run-1",
+            "scope_id": "root", "cleanup_deadline_at": "2026-11-13T00:01:10.000000Z",
+        }],
+    })
+
+    assert scoped.root_context.is_timeout
+    assert scoped.to_dict()["root_context"] == timeout_snapshot()
+
+
+def request(context: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "event_type": "CooperativeCancellationRequested", "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": context,
+        },
+    }
+
+
+def delivery(context: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "event_type": "CooperativeCancellationDelivered",
+        "payload": {
+            "workflow_command_id": "req-1", "workflow_run_id": "run-1", "sequence": 1,
+            "call_kind": "timer", "cancellation": context,
+        },
+    }
+
+
+class ExistingCancellationHandler:
+    """Cancellation handling written before timeout cleanup existed."""
+
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        try:
+            yield ctx.start_timer(10)
+        except WorkflowCancelled as cancelled:
+            assert cancelled.context is ctx.cancellation_context
+            try:
+                ctx.throw_if_cancellation_requested()
+            except WorkflowCancelled as again:
+                assert type(again) is type(cancelled)
+            return {"type": type(cancelled).__name__, "cause": cancelled.context.cause}
+        return None
+
+
+@pytest.mark.parametrize("context,expected", [
+    (snapshot(), {"type": "WorkflowCancelled", "cause": None}),
+    (timeout_snapshot(), {"type": "WorkflowTimeoutCleanupRequested", "cause": "timeout"}),
+])
+def test_existing_cancellation_handlers_catch_the_timeout_at_the_authored_call(
+    context: dict[str, Any], expected: dict[str, Any],
+) -> None:
+    outcome = replay(ExistingCancellationHandler, [request(context), delivery(context)], [], run_id="run-1")
+
+    assert outcome.commands == [CompleteWorkflow(expected)]
+
+
+class Uncaught:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        yield ctx.start_timer(10)
+
+
+def test_an_uncaught_timeout_request_reports_its_type() -> None:
+    outcome = replay(Uncaught, [request(timeout_snapshot()), delivery(timeout_snapshot())], [], run_id="run-1")
+
+    assert len(outcome.commands) == 1 and isinstance(outcome.commands[0], FailWorkflow)
+    assert outcome.commands[0].exception_type == "WorkflowTimeoutCleanupRequested"
diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py
index 25a46679..8d663e1c 100644
--- a/tests/test_replay_regression_corpus.py
+++ b/tests/test_replay_regression_corpus.py
@@ -11,7 +11,7 @@
 import httpx
 import pytest
 
-from durable_workflow import Replayer, Worker, activity, serializer, workflow
+from durable_workflow import CancellationContext, Replayer, Worker, activity, serializer, workflow
 from durable_workflow.client import PROTOCOL_VERSION, Client, WorkflowExecution, WorkflowStreamAppendItem
 from durable_workflow.errors import (
     DurableWorkflowError,
@@ -309,6 +309,20 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         return "not cancelled"
 
 
+@workflow.defn(name="tests.replay.cooperative-timeout-cleanup-cause")
+class CooperativeTimeoutCleanupCauseWorkflow:
+    """Cancellation handling written before timeout cleanup also catches it."""
+
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        try:
+            yield ctx.start_timer(120)
+        except WorkflowCancelled as exc:
+            context = exc.context
+            assert isinstance(context, CancellationContext)
+            return [type(exc).__name__, context.cause, context.timeout_kind, exc.request_id]
+        return "not cancelled"
+
+
 @workflow.defn(name="tests.replay.child-policy-author")
 class ChildPolicyAuthorWorkflow:
     def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
@@ -407,6 +421,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None:
     ChildPolicyAuthorWorkflow,
     ColdReplacementSatisfiedConditionWorkflow,
     CooperativeReopenedConditionCleanupWorkflow,
+    CooperativeTimeoutCleanupCauseWorkflow,
     GoldenSagaCompensationWorkflow,
     GoldenSignalWaitWorkflow,
     GoldenSingleActivityWorkflow,

From 6c75557cb2b49a4b4245412140502751e4791de3 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 06:49:48 +0000
Subject: [PATCH 2/3] Keep only `cause` in the cancellation context

Core settled the timeout-cleanup context shape (workflow#1129 at
14a4b944): the context carries only the optional top-level `cause` plus
its existing cleanup_deadline_at. timeout_kind and work_deadline_at stay
on WorkflowTimeoutCleanupStarted, so the context no longer reads or writes
them. Port Core's scoped-descendant vector: a scoped child of a timeout
root has no `cause` and is an ordinary cancellation, while its scope
origin keeps the root's cause.

Refs durable-workflow/workflow#863, durable-workflow/workflow#1129.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  2 +-
 docs/cooperative-cancellation-design.md       | 12 +--
 src/durable_workflow/cancellation.py          | 38 +--------
 src/durable_workflow/errors.py                |  4 +-
 .../cooperative-timeout-cleanup-cause.json    |  9 +--
 tests/test_cancellation_cause.py              | 81 ++++++++++---------
 tests/test_replay_regression_corpus.py        |  2 +-
 7 files changed, 57 insertions(+), 91 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6226a7c2..ba9c78ac 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,7 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Update outcome reads check the request and answer. An empty, `.` or `..` workflow or update id, an unknown `wait_for` or a non-positive `wait_timeout_seconds` raises `ValueError` before any request, including discovery, so `..` can never resolve onto the workflow's describe route. A successful answer must name the requested `update_id`, carry a string `update_status` and, when it echoes a `workflow_id`, match it. `update_outcome()` returns a value only for a completed receipt. Any other answer raises the new `UnrecognizedUpdateOutcome` with `workflow_id`, `update_id` and `body`, and never returns a value. With `wait_timeout_seconds`, the HTTP timeout becomes that wait plus 30 seconds whenever that is longer than the client's `timeout`, so a long Server-side wait is not cut off.
 
-- 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`. A timeout context may also carry `timeout_kind` (`"execution"` or `"run"`) and `work_deadline_at`; both are refused without the timeout cause. `cause`, `timeout_kind`, `work_deadline_at` and `is_timeout` are readable on the context, and `to_dict()` writes them back exactly, 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. Children cancelled under a timeout request receive an ordinary cancellation.
+- 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`.
 
 ### Changed
 
diff --git a/docs/cooperative-cancellation-design.md b/docs/cooperative-cancellation-design.md
index abf62a97..4bd79ac5 100644
--- a/docs/cooperative-cancellation-design.md
+++ b/docs/cooperative-cancellation-design.md
@@ -45,12 +45,12 @@ work deadline through the same cooperative delivery. The context then has
 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 hard cleanup deadline.
-`timeout_kind` (`"execution"` or `"run"`) and `work_deadline_at` are set when
-the request carries them, and are refused without the timeout cause. A child
-cancelled under a timeout request receives an ordinary cancellation: the cause
-is not inherited. `to_dict()` writes `cause` back exactly as received and
-leaves ordinary contexts unchanged.
+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,
diff --git a/src/durable_workflow/cancellation.py b/src/durable_workflow/cancellation.py
index 3d8e9655..0e1ad2c3 100644
--- a/src/durable_workflow/cancellation.py
+++ b/src/durable_workflow/cancellation.py
@@ -70,7 +70,6 @@ def _text(snapshot: Mapping[str, Any], key: str) -> str:
 
 CANCELLATION_CAUSE = "cancellation"
 TIMEOUT_CAUSE = "timeout"
-_TIMEOUT_KINDS = ("execution", "run")
 
 
 def _format_timestamp(value: datetime) -> str:
@@ -123,10 +122,6 @@ class CancellationContext:
     scope_origin: ScopedCancellationContext | None = None
     cause: str | None = None
     """``"timeout"`` for a timeout-cleanup request, else ``None`` or ``"cancellation"`` (ordinary)."""
-    timeout_kind: str | None = None
-    """``"execution"`` or ``"run"`` when a timeout-cleanup request names the deadline that passed."""
-    work_deadline_at: datetime | None = None
-    """The work deadline that passed, when a timeout-cleanup request carries it."""
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "requester", MappingProxyType(dict(self.requester)))
@@ -214,11 +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, timeout_kind, work_deadline_at = _cause(snapshot, 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, cause=cause, timeout_kind=timeout_kind, work_deadline_at=work_deadline_at,
+            scope_origin=scope_origin, cause=cause,
         )
 
     def to_dict(self) -> dict[str, Any]:
@@ -240,39 +237,12 @@ def to_dict(self) -> dict[str, Any]:
         }
         if self.cause is not None:
             snapshot["cause"] = self.cause
-        if self.timeout_kind is not None:
-            snapshot["timeout_kind"] = self.timeout_kind
-        if self.work_deadline_at is not None:
-            snapshot["work_deadline_at"] = _format_timestamp(self.work_deadline_at)
         if self.scope_origin is not None:
             snapshot["scope_origin"] = self.scope_origin.to_dict()
             snapshot["scope_authority_deadline_at"] = snapshot["cleanup_deadline_at"]
         return snapshot
 
 
-def _cause(snapshot: Mapping[str, Any], deadline: datetime) -> tuple[str | None, str | None, datetime | None]:
-    """Read the optional top-level cause and the timeout fields only a timeout cause may carry."""
-    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")
-    timeout_kind = snapshot.get("timeout_kind")
-    work_deadline = snapshot.get("work_deadline_at")
-    if cause != TIMEOUT_CAUSE:
-        if timeout_kind is not None or work_deadline is not None:
-            raise ValueError("cancellation timeout fields require the timeout cause")
-        return cause, None, None
-    if timeout_kind is not None and timeout_kind not in _TIMEOUT_KINDS:
-        raise ValueError("cancellation timeout_kind must be execution or run")
-    work_deadline_at = None
-    if work_deadline is not None:
-        if not isinstance(work_deadline, str):
-            raise ValueError("cancellation timestamp requires an ISO date, time and timezone")
-        work_deadline_at = _timestamp(work_deadline)
-        if work_deadline_at >= deadline:
-            raise ValueError("cancellation work deadline must precede its cleanup deadline")
-    return cause, timeout_kind, work_deadline_at
-
-
 @dataclass(frozen=True)
 class ScopedCancellationLineage:
     """One immutable original scope address and its bounded cleanup budget."""
diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py
index 49f075cc..8d312846 100644
--- a/src/durable_workflow/errors.py
+++ b/src/durable_workflow/errors.py
@@ -983,9 +983,7 @@ class WorkflowTimeoutCleanupRequested(WorkflowCancelled):
     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 hard cleanup deadline, and ``context.timeout_kind`` and
-    ``context.work_deadline_at`` are set when the request carries them.
-    Experimental.
+    is the cleanup deadline. Experimental.
     """
 
     def __init__(
diff --git a/tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json b/tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json
index 7094e271..3d41f6d7 100644
--- a/tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json
+++ b/tests/fixtures/replay_regressions/cooperative-timeout-cleanup-cause.json
@@ -53,9 +53,7 @@
               "workflow_run_id": "run-1"
             }
           ],
-          "cause": "timeout",
-          "timeout_kind": "execution",
-          "work_deadline_at": "2026-11-13T00:01:00.000000Z"
+          "cause": "timeout"
         }
       }
     },
@@ -89,9 +87,7 @@
               "workflow_run_id": "run-1"
             }
           ],
-          "cause": "timeout",
-          "timeout_kind": "execution",
-          "work_deadline_at": "2026-11-13T00:01:00.000000Z"
+          "cause": "timeout"
         }
       }
     }
@@ -104,7 +100,6 @@
         "result": [
           "WorkflowTimeoutCleanupRequested",
           "timeout",
-          "execution",
           "req-1"
         ]
       }
diff --git a/tests/test_cancellation_cause.py b/tests/test_cancellation_cause.py
index c2f8cbdc..308d819f 100644
--- a/tests/test_cancellation_cause.py
+++ b/tests/test_cancellation_cause.py
@@ -7,7 +7,6 @@
 
 import asyncio
 import json
-from datetime import datetime, timezone
 from typing import Any
 
 import pytest
@@ -39,8 +38,8 @@ def snapshot() -> dict[str, Any]:
     }
 
 
-def timeout_snapshot(**fields: Any) -> dict[str, Any]:
-    return {**snapshot(), "cause": "timeout", **fields}
+def timeout_snapshot() -> dict[str, Any]:
+    return {**snapshot(), "cause": "timeout"}
 
 
 def test_an_ordinary_context_is_unchanged_and_stays_a_cancellation() -> None:
@@ -74,7 +73,6 @@ def test_a_timeout_cause_round_trips_and_delivers_the_timeout_subtype() -> None:
 
     assert context.cause == "timeout"
     assert context.is_timeout
-    assert context.timeout_kind is None and context.work_deadline_at is None
     assert context.to_dict() == timeout_snapshot()
     assert list(context.to_dict()).index("cause") == list(context.to_dict()).index("lineage") + 1
     exception = _cancellation_requested("cancelled", request_id="req-1", context=context)
@@ -84,17 +82,6 @@ def test_a_timeout_cause_round_trips_and_delivers_the_timeout_subtype() -> None:
     assert exception.request_id == "req-1"
 
 
-def test_timeout_fields_round_trip() -> None:
-    value = timeout_snapshot(timeout_kind="execution", work_deadline_at="2026-11-13T00:01:00.000000Z")
-    context = CancellationContext.from_dict(value)
-
-    assert context.timeout_kind == "execution"
-    assert context.work_deadline_at == datetime(2026, 11, 13, 0, 1, tzinfo=timezone.utc)
-    assert context.cleanup_deadline_at == datetime(2026, 11, 13, 0, 1, 10, tzinfo=timezone.utc)
-    assert context.to_dict() == value
-    assert CancellationContext.from_dict(context.to_dict()) == context
-
-
 def test_the_timeout_subtype_is_base_exception_rooted_and_not_task_cancellation() -> None:
     assert issubclass(WorkflowTimeoutCleanupRequested, WorkflowCancelled)
     assert issubclass(WorkflowTimeoutCleanupRequested, BaseException)
@@ -125,44 +112,60 @@ def test_an_unknown_cause_is_refused(cause: Any) -> None:
         CancellationContext.from_dict({**snapshot(), "cause": cause})
 
 
-@pytest.mark.parametrize("cause", [None, "cancellation"])
-@pytest.mark.parametrize("field,value", [
-    ("timeout_kind", "execution"), ("work_deadline_at", "2026-11-13T00:01:00.000000Z"),
-])
-def test_timeout_fields_require_the_timeout_cause(cause: str | None, field: str, value: str) -> None:
-    ordinary = snapshot() if cause is None else {**snapshot(), "cause": cause}
-    with pytest.raises(ValueError, match="timeout fields require the timeout cause"):
-        CancellationContext.from_dict({**ordinary, field: value})
-
-
-@pytest.mark.parametrize("field,value,message", [
-    ("timeout_kind", "workflow", "timeout_kind must be execution or run"),
-    ("timeout_kind", 1, "timeout_kind must be execution or run"),
-    ("work_deadline_at", "2026-11-13 00:01:00", "ISO date, time and timezone"),
-    ("work_deadline_at", 1, "ISO date, time and timezone"),
-    ("work_deadline_at", "2026-11-13T00:01:10.000000Z", "work deadline must precede its cleanup deadline"),
-    ("work_deadline_at", "2026-11-13T00:02:00.000000Z", "work deadline must precede its cleanup deadline"),
-])
-def test_invalid_timeout_fields_are_refused(field: str, value: Any, message: str) -> None:
-    with pytest.raises(ValueError, match=message):
-        CancellationContext.from_dict(timeout_snapshot(**{field: value}))
+def test_the_contract_carries_no_timeout_fields_in_the_context() -> None:
+    # The timeout kind and work deadline are recorded on WorkflowTimeoutCleanupStarted,
+    # not in the context. Like any other unknown key they are not read or written back.
+    context = CancellationContext.from_dict({
+        **timeout_snapshot(), "timeout_kind": "execution", "work_deadline_at": "2026-11-13T00:01:00.000000Z",
+    })
 
+    assert context.is_timeout
+    assert context.to_dict() == timeout_snapshot()
+    assert not hasattr(context, "timeout_kind") and not hasattr(context, "work_deadline_at")
 
-def test_a_scoped_root_keeps_the_run_request_cause() -> None:
+
+def scope_origin() -> dict[str, Any]:
     # Core's ScopedCancellationContext::fromRunContext keeps the run context's keys.
-    scoped = ScopedCancellationContext.from_dict({
+    return {
         "schema": "durable-workflow.scoped-cancellation-context/v1",
         "root_context": timeout_snapshot(),
         "lineage": [{
             "request_id": "req-1", "workflow_instance_id": "instance-1", "workflow_run_id": "run-1",
             "scope_id": "root", "cleanup_deadline_at": "2026-11-13T00:01:10.000000Z",
         }],
-    })
+    }
+
+
+def test_a_scoped_root_keeps_the_run_request_cause() -> None:
+    scoped = ScopedCancellationContext.from_dict(scope_origin())
 
     assert scoped.root_context.is_timeout
     assert scoped.to_dict()["root_context"] == timeout_snapshot()
 
 
+def test_a_scoped_descendant_of_a_timeout_is_cancelled_not_timed_out() -> None:
+    # Built the way Core's fromScopeContext builds it (workflow#1129 at 14a4b944):
+    # the root's cause is stripped, while scope_origin keeps the original root.
+    child = {key: value for key, value in timeout_snapshot().items() if key != "cause"}
+    child.update({
+        "schema": "durable-workflow.cancellation-context/v2",
+        "request_id": "req-child", "parent_request_id": "req-1",
+        "lineage": [*child["lineage"], {
+            "request_id": "req-child", "workflow_instance_id": "child-instance", "workflow_run_id": "child-run",
+        }],
+        "scope_origin": scope_origin(),
+        "scope_authority_deadline_at": "2026-11-13T00:01:10.000000Z",
+    })
+    context = CancellationContext.from_dict(child)
+
+    assert not context.is_timeout
+    assert context.scope_origin is not None and context.scope_origin.root_context.is_timeout
+    assert "cause" not in context.to_dict()
+    assert context.to_dict() == child
+    assert not CancellationContext.from_dict(context.to_dict()).is_timeout
+    assert type(_cancellation_requested("cancelled", request_id="req-child", context=context)) is WorkflowCancelled
+
+
 def request(context: dict[str, Any]) -> dict[str, Any]:
     return {
         "event_type": "CooperativeCancellationRequested", "recorded_at": "2026-11-13T00:01:00Z",
diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py
index 8d663e1c..8f019868 100644
--- a/tests/test_replay_regression_corpus.py
+++ b/tests/test_replay_regression_corpus.py
@@ -319,7 +319,7 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         except WorkflowCancelled as exc:
             context = exc.context
             assert isinstance(context, CancellationContext)
-            return [type(exc).__name__, context.cause, context.timeout_kind, exc.request_id]
+            return [type(exc).__name__, context.cause, exc.request_id]
         return "not cancelled"
 
 

From 280c5ffc080f6dafdaf78d04de461234a9acc309 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 09:49:58 +0000
Subject: [PATCH 3/3] Keep experimental timeout parsing in the unreleased
 changelog

---
 CHANGELOG.md | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index cff5b0b6..d7ed6570 100644
--- a/CHANGELOG.md
+++ b/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
@@ -23,8 +27,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Update outcome reads check the request and answer. An empty, `.` or `..` workflow or update id, an unknown `wait_for` or a non-positive `wait_timeout_seconds` raises `ValueError` before any request, including discovery, so `..` can never resolve onto the workflow's describe route. A successful answer must name the requested `update_id`, carry a string `update_status` and, when it echoes a `workflow_id`, match it. `update_outcome()` returns a value only for a completed receipt. Any other answer raises the new `UnrecognizedUpdateOutcome` with `workflow_id`, `update_id` and `body`, and never returns a value. With `wait_timeout_seconds`, the HTTP timeout becomes that wait plus 30 seconds whenever that is longer than the client's `timeout`, so a long Server-side wait is not cut off.
 
-- 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`.
-
 ### Changed
 
 - The `Client.update_workflow()` and `WorkflowHandle.update()` docstrings now say what they always did: they return the Server receipt, not the decoded result, and the Server answers at `wait_for="accepted"` when it is omitted. A receipt with `update_status: "accepted"` is a pending update even though its `result` is `None`.