Expose acknowledged activity heartbeat checkpoints

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

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:

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",

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

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 @@
     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 @@
         """``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.
 

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
 

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."""
 

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")
@@ -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,
         )