diff --git a/CHANGELOG.md b/CHANGELOG.md index 051bf08..d7ed657 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 diff --git a/docs/cooperative-cancellation-design.md b/docs/cooperative-cancellation-design.md index cffec18..4bd79ac 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 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 diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 1736695..c5b5094 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 d0c9041..0e1ad2c 100644 --- a/src/durable_workflow/cancellation.py +++ b/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), } diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 9450902..8d31284 100644 --- a/src/durable_workflow/errors.py +++ b/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. diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 373fbbc..0948e20 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, ) @@ -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, )) 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 0000000..3d41f6d --- /dev/null +++ b/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" + ] + } + ] + } +} diff --git a/tests/test_cancellation_cause.py b/tests/test_cancellation_cause.py new file mode 100644 index 0000000..308d819 --- /dev/null +++ b/tests/test_cancellation_cause.py @@ -0,0 +1,227 @@ +"""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 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() -> dict[str, Any]: + return {**snapshot(), "cause": "timeout"} + + +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.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_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}) + + +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 scope_origin() -> dict[str, Any]: + # Core's ScopedCancellationContext::fromRunContext keeps the run context's keys. + 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", + "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 2726a29..6b11249 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, @@ -310,6 +310,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, 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] @@ -408,6 +422,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: ChildPolicyAuthorWorkflow, ColdReplacementSatisfiedConditionWorkflow, CooperativeReopenedConditionCleanupWorkflow, + CooperativeTimeoutCleanupCauseWorkflow, GoldenSagaCompensationWorkflow, GoldenSignalWaitWorkflow, GoldenSingleActivityWorkflow,