From 2fd389c058822705507077b137ea8f0ca4c8189e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:21:10 +0000 Subject: [PATCH 1/2] Expose acknowledged activity heartbeat checkpoints --- CHANGELOG.md | 4 + docs/sdk-reference.md | 26 ++ src/durable_workflow/__init__.py | 4 + src/durable_workflow/_activity_checkpoint.py | 131 ++++++++++ src/durable_workflow/activity.py | 32 ++- src/durable_workflow/client.py | 3 +- src/durable_workflow/errors.py | 7 + src/durable_workflow/worker.py | 2 + tests/test_activity_checkpoint.py | 249 +++++++++++++++++++ tests/test_worker.py | 4 + 10 files changed, 460 insertions(+), 2 deletions(-) create mode 100644 src/durable_workflow/_activity_checkpoint.py create mode 100644 tests/test_activity_checkpoint.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f796e7..1daf9ef 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 + +- Read-only acknowledged heartbeat checkpoint accessors distinguish unsupported claims from an explicitly empty checkpoint. Received details, contract version and acknowledgement time remain immutable across heartbeat outcomes. This source addition still requires package publication and installed replacement qualification. (durable-workflow/.github#218) + ## [2.18.0] - 2026-11-14 ### Added diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 3bb9f59..9bf0eff 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -1390,6 +1390,32 @@ supervision. The SDK cannot forcibly stop a running native thread or undo its external effects. Keep side effects idempotent and heartbeat at useful progress boundaries. +### Acknowledged heartbeat checkpoints (unreleased) + +The source API adds `activity.context().last_heartbeat_checkpoint()`, +`last_heartbeat_details(result_type=...)`, `last_heartbeat_at()` and +`heartbeat_resume_version()`. Package publication and installed cold-replacement +qualification are pending under [shared #218](https://github.com/durable-workflow/.github/issues/218). + +A participating claim must advertise `activity_heartbeat_resume` version 1 and +carry both checkpoint fields for its original execution, attempt, run and +namespace. `None` means this supported claim explicitly has no checkpoint. +Older or nonparticipating claims, inline local activities and Server-free tests +raise `ActivityHeartbeatResumeUnavailable`; they do not invent empty state. + +The checkpoint preserves received details and the original RFC3339 +acknowledgement timestamp. Details are returned as independent copies, with +optional decoding to the SDK's supported annotated types. Its immutable +`contract_version` is the negotiated format version, not a write counter. +Server v1 supplies no per-checkpoint sequence number. + +This is the snapshot received at claim time. A successful, rejected or uncertain +heartbeat cannot change it. After a lost response, a later replacement claim +may contain the committed checkpoint even though the earlier caller could not +confirm acknowledgement. A checkpoint does not atomically commit an external +business effect. Keep stable idempotency keys, provider reconciliation and +external fencing where the effect requires them. + ## Metrics Pass a recorder to `Client(metrics=...)` or `Worker(metrics=...)` to collect request, poll, and task metrics. The SDK ships a no-op default, an `InMemoryMetrics` recorder for tests or custom exporter loops, and `PrometheusMetrics` for deployments that install the optional extra: diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index c5b5094..7aa5d6f 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -7,6 +7,7 @@ __version__ = "0.0.0+unknown" from . import activity, sync, testing, workflow +from ._activity_checkpoint import ActivityHeartbeatCheckpoint from .activity import ActivityContext, ActivityInfo, ActivityRetryInfo from .auth_composition import ( AUTH_COMPOSITION_CONTRACT_SCHEMA, @@ -70,6 +71,7 @@ from .errors import ( ActivityCancelled, ActivityFailed, + ActivityHeartbeatResumeUnavailable, ChildWorkflowCancelled, ChildWorkflowFailed, ChildWorkflowTerminated, @@ -260,6 +262,8 @@ "ActivityHandler", "ActivityInfo", "ActivityRetryInfo", + "ActivityHeartbeatCheckpoint", + "ActivityHeartbeatResumeUnavailable", "ActivityInterceptorContext", "ActivityRetryPolicy", "BridgeAdapterOutcome", diff --git a/src/durable_workflow/_activity_checkpoint.py b/src/durable_workflow/_activity_checkpoint.py new file mode 100644 index 0000000..9cbb59d --- /dev/null +++ b/src/durable_workflow/_activity_checkpoint.py @@ -0,0 +1,131 @@ +"""Immutable heartbeat-resume metadata captured from one negotiated poll.""" + +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from ._typed_values import decode_payload_value, validate_annotation +from .errors import ActivityHeartbeatResumeUnavailable + +_IDENTITY = ("task_id", "activity_execution_id", "activity_attempt_id", "workflow_id", "run_id", "namespace") +_TIMESTAMP = re.compile( + r"[0-9]{4}-[0-9]{2}-[0-9]{2}T(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]" + r"(?:\.[0-9]{1,9})?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])" +) + + +def _invalid() -> ValueError: + return ValueError("activity_heartbeat_resume v1 claim has an invalid checkpoint") + + +@dataclass(frozen=True, init=False) +class ActivityHeartbeatCheckpoint: + """The acknowledged snapshot received with this claim, never a pending write. + + ``contract_version`` describes the negotiated wire contract. Server v1 + supplies no checkpoint write sequence. Acknowledgement does not commit an + external business effect atomically. + """ + + acknowledged_at: str + contract_version: int = field(default=1, init=False) + _details_json: str = field(repr=False) + + def __init__(self, details: Any, acknowledged_at: str) -> None: + if details is None or not isinstance(acknowledged_at, str) or not _TIMESTAMP.fullmatch(acknowledged_at): + raise _invalid() + try: + # Python 3.10's parser accepts fewer fractional forms. Validate the + # calendar/offset separately and retain the original precision. + calendar = re.sub(r"\.\d+(?=Z|[+-])", "", acknowledged_at) + datetime.fromisoformat(calendar.replace("Z", "+00:00")) + encoded = json.dumps(details, allow_nan=False) + except (ValueError, TypeError, OverflowError, RecursionError): + raise _invalid() from None + object.__setattr__(self, "acknowledged_at", acknowledged_at) + object.__setattr__(self, "_details_json", encoded) + + def details(self, *, result_type: Any = None) -> Any: + """Return an independent copy, optionally decoded using SDK typed values.""" + value = json.loads(self._details_json) + if result_type is None: + return value + validate_annotation(result_type, "checkpoint") + return decode_payload_value(value, result_type, "checkpoint") + + +@dataclass(frozen=True) +class _HeartbeatResume: + available: bool = False + checkpoint: ActivityHeartbeatCheckpoint | None = None + identity: tuple[Any, ...] = () + + def get(self) -> ActivityHeartbeatCheckpoint | None: + if not self.available: + raise ActivityHeartbeatResumeUnavailable() + return self.checkpoint + + +class _ActivityTask(dict[str, Any]): + """Keep poll metadata outside the task's unchanged public wire dictionary.""" + + def __init__(self, task: Mapping[str, Any], resume: _HeartbeatResume) -> None: + super().__init__(task) + self.heartbeat_resume = resume + + +def _identity(task: Mapping[str, Any]) -> tuple[Any, ...]: + values = tuple(task.get(name) for name in _IDENTITY) + attempt = task.get("attempt_number") + if (any(not isinstance(value, str) or not value.strip() for value in values) + or type(attempt) is not int or attempt < 1): + raise _invalid() + return (*values, attempt, task.get("lease_owner"), task.get("activity_type"), task.get("task_queue")) + + +def capture_poll(data: Mapping[str, Any], namespace: str, worker_id: str, task_queue: str | None = None) -> Any: + task = data.get("task") + capabilities = data.get("server_capabilities") + capability = capabilities.get("activity_heartbeat_resume") if isinstance(capabilities, Mapping) else None + if (not isinstance(task, dict) or not isinstance(capability, Mapping) + or capability.get("supported") is not True or type(capability.get("version")) is not int + or capability["version"] != 1): + return task + if "last_heartbeat_details" not in task and "last_heartbeat_at" not in task: + return task + identity = _identity(task) + if task["namespace"] != namespace or task.get("lease_owner") != worker_id: + raise _invalid() + if task_queue is not None and "task_queue" in task and task["task_queue"] != task_queue: + raise _invalid() + if "last_heartbeat_details" not in task or "last_heartbeat_at" not in task: + raise _invalid() + details, at = task["last_heartbeat_details"], task["last_heartbeat_at"] + if details is None and at is None: + checkpoint = None + elif details is not None and task["attempt_number"] > 1: + checkpoint = ActivityHeartbeatCheckpoint(details, at) + else: + raise _invalid() + return _ActivityTask(task, _HeartbeatResume(True, checkpoint, identity)) + + +def claim_resume( + task: Mapping[str, Any], namespace: str | None, task_id: str, attempt_id: str, attempt: int, worker_id: str, + activity_type: str, task_queue: str, +) -> _HeartbeatResume: + if not isinstance(task, _ActivityTask): + return _HeartbeatResume() + resume = task.heartbeat_resume + if (resume.identity != _identity(task) or task.get("namespace") != namespace + or task.get("task_id") != task_id or task.get("activity_attempt_id") != attempt_id + or type(attempt) is not int or task.get("attempt_number") != attempt + or task.get("lease_owner") != worker_id or task.get("activity_type") != activity_type + or "task_queue" in task and task["task_queue"] != task_queue): + raise _invalid() + return resume diff --git a/src/durable_workflow/activity.py b/src/durable_workflow/activity.py index ebc810d..4719cd7 100644 --- a/src/durable_workflow/activity.py +++ b/src/durable_workflow/activity.py @@ -8,6 +8,7 @@ from dataclasses import InitVar, dataclass, field from typing import TYPE_CHECKING, Any +from ._activity_checkpoint import ActivityHeartbeatCheckpoint, _HeartbeatResume, claim_resume from ._activity_lease import current_activity_authority from .errors import ActivityCancelled, ActivityLeaseAuthorityLost @@ -79,14 +80,20 @@ class ActivityInfo: task_queue: str worker_id: str task_metadata: InitVar[Mapping[str, Any] | None] = None + task_namespace: InitVar[str | None] = None retry_info: ActivityRetryInfo = field(init=False) + _heartbeat_resume: _HeartbeatResume = field(init=False, repr=False) workflow_id: str | None = field(init=False) run_id: str | None = field(init=False) activity_execution_id: str | None = field(init=False) - def __post_init__(self, task_metadata: Mapping[str, Any] | None) -> None: + def __post_init__(self, task_metadata: Mapping[str, Any] | None, task_namespace: str | None) -> None: task = task_metadata if task_metadata is not None else {} object.__setattr__(self, "retry_info", ActivityRetryInfo.from_task(self.attempt_number, task)) + object.__setattr__(self, "_heartbeat_resume", claim_resume( + task, task_namespace, self.task_id, self.activity_attempt_id, self.attempt_number, self.worker_id, + self.activity_type, self.task_queue, + )) for name in ("workflow_id", "run_id"): value = task.get(name) object.__setattr__(self, name, value if isinstance(value, str) and value else None) @@ -125,6 +132,29 @@ def is_cancelled(self) -> bool: """``True`` once the server has signalled that the owning workflow cancelled this activity.""" return self._cancel_requested + def heartbeat_resume_version(self) -> int | None: + """The received contract version, or ``None`` when unavailable.""" + return 1 if self._info._heartbeat_resume.available else None + + def last_heartbeat_checkpoint(self) -> ActivityHeartbeatCheckpoint | None: + """Return the received snapshot, or ``None`` for supported empty state. + + Raises ActivityHeartbeatResumeUnavailable for older or nonparticipating + claims, inline local activities and Server-free tests. Heartbeat writes + never change this snapshot, including rejected or uncertain writes. + """ + return self._info._heartbeat_resume.get() + + def last_heartbeat_details(self, *, result_type: Any = None) -> Any: + """Read acknowledged details, optionally decoded to an application type.""" + checkpoint = self.last_heartbeat_checkpoint() + return checkpoint.details(result_type=result_type) if checkpoint is not None else None + + def last_heartbeat_at(self) -> str | None: + """The original RFC3339 acknowledgement time without losing precision.""" + checkpoint = self.last_heartbeat_checkpoint() + return checkpoint.acknowledged_at if checkpoint is not None else None + def heartbeat_sync(self, details: dict[str, Any] | None = None) -> None: """Heartbeat from a synchronous activity thread and check attempt authority. diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 294c4c7..0da4edd 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -41,6 +41,7 @@ import httpx from . import serializer +from ._activity_checkpoint import capture_poll as _capture_activity_checkpoint from ._activity_lease import ActivityLeaseAuthority, current_activity_authority from ._cancellation_scope import CancellationScopeOpenReceipt, scope_identity from ._cancellation_scope_history import ( @@ -6782,7 +6783,7 @@ async def poll_activity_task( except httpx.TimeoutException: continue - return (data or {}).get("task") + return _capture_activity_checkpoint(data or {}, self.namespace, worker_id, task_queue) return None diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 8d31284..66d3dd4 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -27,6 +27,13 @@ class DurableWorkflowError(Exception): """Base class for every exception raised by the SDK.""" +class ActivityHeartbeatResumeUnavailable(DurableWorkflowError): + """This claim did not supply the supported heartbeat-resume contract.""" + + def __init__(self) -> None: + super().__init__("activity_heartbeat_resume v1 is unavailable for this claim") + + class ReplayClockUnavailable(DurableWorkflowError): """The current execution has no proved business replay-time boundary.""" diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 1c17f49..defb8f7 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -1531,6 +1531,7 @@ async def _register(self) -> None: if any(contract["updates"] for contract in self.workflow_command_contracts.values()): capabilities.append(WORKFLOW_UPDATES_CAPABILITY) capabilities.append(MESSAGE_STREAMS_CAPABILITY) + capabilities.append("activity_heartbeat_resume") capabilities.extend(self.capabilities) if self._nexus_cooperative_cancellation_supported and "nexus_cooperative_cancellation" not in capabilities: capabilities.append("nexus_cooperative_cancellation") @@ -3416,6 +3417,7 @@ async def _run_activity_task_body(self, task: dict[str, Any]) -> str: task_queue=self.task_queue, worker_id=self.worker_id, task_metadata=task, + task_namespace=getattr(self.client, "namespace", None), ), client=self.client, ) diff --git a/tests/test_activity_checkpoint.py b/tests/test_activity_checkpoint.py new file mode 100644 index 0000000..9ed8c05 --- /dev/null +++ b/tests/test_activity_checkpoint.py @@ -0,0 +1,249 @@ +"""Negotiated poll-to-handler controls, without a real Server.""" + +from __future__ import annotations + +import copy +import json +import pickle +from dataclasses import FrozenInstanceError, dataclass +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest + +from durable_workflow import ( + ActivityContext, + ActivityHeartbeatResumeUnavailable, + ActivityInfo, + Client, + TypedValueError, + Worker, + activity, + serializer, +) +from durable_workflow._activity_checkpoint import capture_poll +from durable_workflow._activity_process import CallbackInvocation, SupervisedCallback + +AT = "2026-11-14T09:00:00.123456789Z" + + +def poll(*, attempt: int = 2, details: Any = None, at: Any = None) -> dict[str, Any]: + return { + "task": { + "task_id": "task", "activity_execution_id": "execution", "activity_attempt_id": "attempt", + "workflow_id": "workflow", "run_id": "run", "namespace": "default", "attempt_number": attempt, + "activity_type": "checkpoint.inspect", "lease_owner": "worker", "task_queue": "queue", + "arguments": serializer.envelope([]), "payload_codec": "avro", + "retry_policy": {"max_attempts": 3, "backoff_seconds": [1, 2]}, + "last_heartbeat_details": details, "last_heartbeat_at": at, + }, + "server_capabilities": {"activity_heartbeat_resume": {"supported": True, "version": 1}}, + } + + +def context(wire: dict[str, Any]) -> ActivityContext: + task = capture_poll(wire, "default", "worker") + return ActivityContext(info=ActivityInfo( + task["task_id"], "checkpoint.inspect", task["activity_attempt_id"], task["attempt_number"], "queue", "worker", + task_metadata=task, task_namespace="default", + ), client=AsyncMock(spec=Client)) + + +@pytest.mark.parametrize("capability", [ + None, False, {"supported": False, "version": 1}, {"supported": True, "version": 2}, + {"supported": True, "version": "1"}, {"supported": True, "version": True}, {"supported": 1, "version": 1}, +]) +def test_unavailable_is_distinct_from_supported_empty(capability: Any) -> None: + wire = poll() + wire["server_capabilities"]["activity_heartbeat_resume"] = capability + ctx = context(wire) + assert ctx.heartbeat_resume_version() is None + for accessor in (ctx.last_heartbeat_checkpoint, ctx.last_heartbeat_details, ctx.last_heartbeat_at): + with pytest.raises(ActivityHeartbeatResumeUnavailable): + accessor() + + +def test_omitted_fields_and_legacy_context_do_not_invent_empty_state() -> None: + wire = poll() + del wire["task"]["last_heartbeat_details"], wire["task"]["last_heartbeat_at"] + with pytest.raises(ActivityHeartbeatResumeUnavailable): + context(wire).last_heartbeat_checkpoint() + info = ActivityInfo("task", "activity", "attempt", 1, "queue", "worker") + with pytest.raises(ActivityHeartbeatResumeUnavailable): + ActivityContext(info=info, client=AsyncMock(spec=Client)).last_heartbeat_details() + + +@pytest.mark.parametrize("attempt", [1, 2]) +def test_explicit_empty_claim_is_supported_on_first_and_later_attempts(attempt: int) -> None: + ctx = context(poll(attempt=attempt)) + assert ctx.heartbeat_resume_version() == 1 + assert ctx.last_heartbeat_checkpoint() is None + assert ctx.last_heartbeat_details() is None + assert ctx.last_heartbeat_at() is None + + +@dataclass +class Progress: + cursor: int + flags: list[bool | None] + + +def test_snapshot_is_typed_independent_private_and_spawn_serializable() -> None: + details = {"cursor": 2**63 - 1, "flags": [True, None, False], "secret": "private-checkpoint"} + wire = poll(details=details, at=AT) + ctx = context(wire) + details["cursor"] = 0 + first = ctx.last_heartbeat_details() + first["flags"].clear() + assert ctx.last_heartbeat_details(result_type=Progress) == Progress(2**63 - 1, [True, None, False]) + assert ctx.last_heartbeat_at() == AT + checkpoint = ctx.last_heartbeat_checkpoint() + assert checkpoint is not None + assert checkpoint.contract_version == 1 + assert "private-checkpoint" not in repr(checkpoint) + with pytest.raises(FrozenInstanceError): + checkpoint.acknowledged_at = "replacement" # type: ignore[misc] + restored = ActivityContext(info=pickle.loads(pickle.dumps(ctx.info)), client=AsyncMock(spec=Client)) + assert restored.last_heartbeat_details() == ctx.last_heartbeat_details() + assert restored.last_heartbeat_at() == AT + with pytest.raises(TypedValueError): + ctx.last_heartbeat_details(result_type=list[int]) + + +@pytest.mark.parametrize("empty", [False, True]) +@pytest.mark.parametrize(("field", "value"), [ + ("task_id", ""), ("activity_execution_id", None), ("activity_execution_id", " "), + ("activity_attempt_id", 3), ("workflow_id", ""), ("run_id", None), ("namespace", "other"), + ("lease_owner", "other"), ("attempt_number", 0), ("attempt_number", -1), + ("attempt_number", True), ("attempt_number", "2"), ("attempt_number", 2.0), +]) +def test_even_empty_checkpoint_requires_the_original_claim_identity(empty: bool, field: str, value: Any) -> None: + wire = poll(details=None if empty else {"cursor": 1}, at=None if empty else AT) + wire["task"][field] = value + with pytest.raises(ValueError, match="invalid checkpoint"): + context(wire) + + +@pytest.mark.parametrize(("details", "at", "attempt"), [ + ({}, AT, 1), ({}, None, 2), (None, AT, 2), ({}, "not-a-time", 2), + ({}, "2026-11-14T09:00:00", 2), ({}, "2026-02-30T09:00:00Z", 2), + ({}, "2026-11-14T09:00:00+99:00", 2), ({"value": float("nan")}, AT, 2), + ({}, "2026-11-14T09:00:00+01:75", 2), ({}, "2026-11-14T09:00:00.1234567890Z", 2), + ({}, "2026-11-14T09:00:00.١٢٣Z", 2), +]) +def test_invalid_pairs_and_timestamps_refuse_without_exposing_details(details: Any, at: Any, attempt: int) -> None: + with pytest.raises(ValueError, match="^activity_heartbeat_resume v1 claim has an invalid checkpoint$"): + context(poll(attempt=attempt, details=details, at=at)) + + +@pytest.mark.parametrize("field", ["last_heartbeat_details", "last_heartbeat_at"]) +def test_one_missing_checkpoint_field_refuses(field: str) -> None: + wire = poll() + del wire["task"][field] + with pytest.raises(ValueError, match="invalid checkpoint"): + context(wire) + + +@pytest.mark.parametrize("field", [ + "task_id", "activity_execution_id", "activity_attempt_id", "workflow_id", "run_id", "activity_type", "task_queue", +]) +def test_task_mutation_cannot_rebind_the_captured_snapshot(field: str) -> None: + task = capture_poll(poll(details={"cursor": 1}, at=AT), "default", "worker") + task[field] = "replacement" + with pytest.raises(ValueError, match="invalid checkpoint"): + ActivityInfo(task["task_id"], "activity", task["activity_attempt_id"], 2, "queue", "worker", + task_metadata=task, task_namespace="default") + + +@pytest.mark.parametrize("outcome", ["success", "refused", "uncertain"]) +async def test_heartbeat_outcomes_never_advance_the_received_checkpoint(outcome: str) -> None: + ctx = context(poll(details={"cursor": 1}, at=AT)) + if outcome == "uncertain": + ctx._client.heartbeat_activity_task.side_effect = httpx.ReadTimeout("response lost") # type: ignore[attr-defined] + with pytest.raises(httpx.ReadTimeout): + await ctx.heartbeat({"cursor": 2}) + else: + ctx._client.heartbeat_activity_task.return_value = { # type: ignore[attr-defined] + "can_continue": outcome == "success", "last_heartbeat_details": {"cursor": 2}, + "last_heartbeat_at": "2026-11-14T10:00:00Z", "reason": "stale_attempt", + } + if outcome == "success": + await ctx.heartbeat({"cursor": 2}) + else: + from durable_workflow.errors import ActivityLeaseAuthorityLost + with pytest.raises(ActivityLeaseAuthorityLost): + await ctx.heartbeat({"cursor": 2}) + assert ctx.last_heartbeat_details() == {"cursor": 1} + assert ctx.last_heartbeat_at() == AT + + +async def test_real_client_poll_delivers_snapshot_to_handler_without_changing_task_wire() -> None: + wire = poll(details={"cursor": 7}, at=AT) + seen: list[Any] = [] + requests: list[str] = [] + + async def respond(request: httpx.Request) -> httpx.Response: + requests.append(request.url.path) + if request.url.path.endswith("/poll"): + return httpx.Response(200, json=wire) + assert request.url.path.endswith("/complete") + return httpx.Response(200, json={"recorded": True}) + + @activity.defn(name="checkpoint.inspect") + async def inspect_checkpoint() -> None: + ctx = activity.context() + seen.append((ctx.last_heartbeat_details(), ctx.last_heartbeat_at(), ctx.info.retry_info.policy())) + + async with Client("http://checkpoint.invalid", namespace="default") as client: + await client._http.aclose() + client._http = httpx.AsyncClient(transport=httpx.MockTransport(respond), base_url="http://checkpoint.invalid") + client._cluster_info = {"worker_protocol": {"server_capabilities": {}}} + worker = Worker(client, worker_id="worker", task_queue="queue", workflows=[], activities=[inspect_checkpoint]) + try: + task = await client.poll_activity_task(worker_id="worker", task_queue="queue") + assert task == wire["task"] + assert json.loads(json.dumps(task)) == wire["task"] + original = copy.deepcopy(wire) + assert await worker._run_activity_task(task) == "completed" + assert wire == original + finally: + await worker.stop() + assert seen == [({"cursor": 7}, AT, {"max_attempts": 3, "backoff_seconds": [1, 2]})] + assert len(requests) == 2 + + +def checkpoint_in_cold_callback() -> dict[str, Any]: + ctx = activity.context() + return {"details": ctx.last_heartbeat_details(), "at": ctx.last_heartbeat_at(), + "version": ctx.heartbeat_resume_version(), "attempt": ctx.info.attempt_number} + + +async def test_supervised_callback_receives_the_same_immutable_snapshot() -> None: + wire = poll(details={"cursor": 3}, at=AT) + ctx = context(wire) + callback = SupervisedCallback(CallbackInvocation( + handler=checkpoint_in_cold_callback, args=(), info=ctx.info, + task=capture_poll(wire, "default", "worker"), interceptors=(), + )) + + async def no_heartbeat(_details: Any) -> None: + raise AssertionError("checkpoint read emitted a heartbeat") + + try: + await callback.start() + result = await callback.result(no_heartbeat) + assert result.failure is None + assert result.value == {"details": {"cursor": 3}, "at": AT, "version": 1, "attempt": 2} + finally: + await callback.close() + + +def test_namespace_or_queue_change_cannot_reuse_a_received_checkpoint() -> None: + wire = poll(details={}, at=AT) + with pytest.raises(ValueError, match="invalid checkpoint"): + capture_poll(wire, "default", "worker", "other-queue") + task = capture_poll(wire, "default", "worker", "queue") + with pytest.raises(ValueError, match="invalid checkpoint"): + ActivityInfo("task", "checkpoint.inspect", "attempt", 2, "queue", "worker", + task_metadata=task, task_namespace="other-namespace") diff --git a/tests/test_worker.py b/tests/test_worker.py index d8d575f..dcd6584 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -437,6 +437,7 @@ async def test_register(self, mock_client: AsyncMock) -> None: "workflow_entry_arguments_validation", "query_tasks", MESSAGE_STREAMS_CAPABILITY, + "activity_heartbeat_resume", "worker_sessions", ], "capability_manifest": { @@ -586,6 +587,7 @@ def run(self, ctx): # type: ignore[no-untyped-def] "query_tasks", WORKFLOW_UPDATES_CAPABILITY, MESSAGE_STREAMS_CAPABILITY, + "activity_heartbeat_resume", "worker_sessions", ] @@ -675,6 +677,7 @@ def run(self, ctx): # type: ignore[no-untyped-def] UPDATE_VALIDATION_TASKS_CAPABILITY, WORKFLOW_UPDATES_CAPABILITY, MESSAGE_STREAMS_CAPABILITY, + "activity_heartbeat_resume", "worker_sessions", ] @@ -783,6 +786,7 @@ async def test_register_keeps_baseline_capabilities_when_server_does_not_support "local_activities", "workflow_entry_arguments_validation", MESSAGE_STREAMS_CAPABILITY, + "activity_heartbeat_resume", "worker_sessions", ] From 97b07b0458ece801d199f774738ee49a29215620 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:25:44 +0000 Subject: [PATCH 2/2] Declare heartbeat resume in base worker capabilities --- src/durable_workflow/worker.py | 4 ++-- tests/test_worker.py | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index defb8f7..b0dbc44 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -151,7 +151,7 @@ DURABLE_SELECTION_CAPABILITY = "durable_selection" LOCAL_ACTIVITIES_CAPABILITY = "local_activities" -# Command capabilities implemented by every high-level Python worker. Keep this +# Base capabilities implemented by every high-level Python worker. Keep this # advertisement explicit: worker capabilities are not inferred from runtime, # SDK, or protocol-version metadata. _WORKFLOW_COMMAND_CAPABILITIES = ( @@ -160,6 +160,7 @@ DURABLE_SELECTION_CAPABILITY, LOCAL_ACTIVITIES_CAPABILITY, "workflow_entry_arguments_validation", + "activity_heartbeat_resume", ) _WORKFLOW_WORK_TASK_KINDS = ("workflow", "update_validation") @@ -1531,7 +1532,6 @@ async def _register(self) -> None: if any(contract["updates"] for contract in self.workflow_command_contracts.values()): capabilities.append(WORKFLOW_UPDATES_CAPABILITY) capabilities.append(MESSAGE_STREAMS_CAPABILITY) - capabilities.append("activity_heartbeat_resume") capabilities.extend(self.capabilities) if self._nexus_cooperative_cancellation_supported and "nexus_cooperative_cancellation" not in capabilities: capabilities.append("nexus_cooperative_cancellation") diff --git a/tests/test_worker.py b/tests/test_worker.py index dcd6584..6c2959b 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -435,9 +435,9 @@ async def test_register(self, mock_client: AsyncMock) -> None: DURABLE_SELECTION_CAPABILITY, "local_activities", "workflow_entry_arguments_validation", + "activity_heartbeat_resume", "query_tasks", MESSAGE_STREAMS_CAPABILITY, - "activity_heartbeat_resume", "worker_sessions", ], "capability_manifest": { @@ -584,10 +584,10 @@ def run(self, ctx): # type: ignore[no-untyped-def] DURABLE_SELECTION_CAPABILITY, "local_activities", "workflow_entry_arguments_validation", + "activity_heartbeat_resume", "query_tasks", WORKFLOW_UPDATES_CAPABILITY, MESSAGE_STREAMS_CAPABILITY, - "activity_heartbeat_resume", "worker_sessions", ] @@ -673,11 +673,11 @@ def run(self, ctx): # type: ignore[no-untyped-def] DURABLE_SELECTION_CAPABILITY, "local_activities", "workflow_entry_arguments_validation", + "activity_heartbeat_resume", "query_tasks", UPDATE_VALIDATION_TASKS_CAPABILITY, WORKFLOW_UPDATES_CAPABILITY, MESSAGE_STREAMS_CAPABILITY, - "activity_heartbeat_resume", "worker_sessions", ] @@ -785,8 +785,8 @@ async def test_register_keeps_baseline_capabilities_when_server_does_not_support DURABLE_SELECTION_CAPABILITY, "local_activities", "workflow_entry_arguments_validation", - MESSAGE_STREAMS_CAPABILITY, "activity_heartbeat_resume", + MESSAGE_STREAMS_CAPABILITY, "worker_sessions", ]