Activity¶

Synchronous activities execute in the worker's bounded remote thread pool by default. They can call context().heartbeat_sync(details) and inspect context().is_cancelled; a heartbeat reporting cancellation raises ActivityCancelled. Async activities use await context().heartbeat(details). See activity execution and shutdown for thread capacity, cancellation and process-isolation limits.

Original attempt and received retry policy¶

Requires Python SDK 2.12.0 or later. Inside an activity, capture info = activity.context().info. Its attempt_number is the original attempt number. info.retry_info is an immutable snapshot of that number and the retry policy received with the task. retry_info.policy() returns a fresh copy, including unknown policy fields. Changing that copy does not change the task or snapshot.

Received task metadata	`policy_provided`	`max_attempts()`	`is_unlimited()`	`is_last_configured_attempt()`
No `retry_policy` field	`False`	`None`	`False`	`None`
Root `retry_policy: null`	`True`	`None`	`False`	`None`
Policy object with `max_attempts: null`	`True`	`None`	`True`	`False`
Policy object with positive integer `max_attempts: N`	`True`	`N`	`False`	Whether the original attempt is at least `N`

An absent, malformed or unknown limit stays unknown. The SDK does not infer a policy from Server defaults. A last-configured-attempt result describes the received budget. Cancellation, deadlines, workflow closure and non-retryable failure can still end execution before that limit.

Ordinary remote and supervised callbacks receive the same original snapshot. Inline and prepared local callbacks have their own attempt context and do not inherit the parent's remote retry policy.

Signal the originating run¶

info.workflow_id and info.run_id are nullable, original identities from the received task. Inline and prepared callbacks inherit only known parent identities. Task IDs and activity-attempt IDs are never substituted for missing workflow/run identities. A standalone or older task can have no workflow identity; handle that case explicitly.

Call this helper from an async activity, passing an existing application Client authorized to signal the run. The handle pins the original run, so a later continue-as-new run cannot receive the signal accidentally.

from durable_workflow import Client, activity


async def signal_originating_run(client: Client, business_event_id: str) -> dict:
    info = activity.context().info
    if info.workflow_id is None or info.run_id is None:
        raise RuntimeError("This activity has no known originating workflow run")
    handle = client.get_workflow_handle(info.workflow_id, run_id=info.run_id)
    return await handle.signal(
        "payment_recorded",
        [business_event_id],
        request_id=f"payment-recorded:{business_event_id}",
    )

Use a stable business-event identity across activity retries. An attempt number, task ID or random value generated on every retry identifies a different intent. Keep the event ID within the Server's command-key length limit. A stable key does not by itself guarantee signal deduplication: the currently published Server does not retain these keys for every signal path. Make business-event processing idempotent and reconcile uncertain delivery rather than assuming that one key means one recorded signal. Track Server receipt-retention support in the portable command-key requirement.

Heartbeats during one blocking call¶

Capture ctx = activity.context() inside the activity before starting a helper thread. A newly created thread does not automatically have the activity's context. Calling activity.context() there can raise "called outside of an activity execution". Pass the captured context instead. Its ctx.heartbeat_sync(details) schedules the heartbeat on the owning worker loop with the original task, attempt and worker identity. It can be called from a helper thread, but must never run on that worker loop. Async activity code uses await ctx.heartbeat(details).

The complete threaded-heartbeat example provides call_with_heartbeats(work). Copy it into your application and use it from an ordinary synchronous activity:

from durable_workflow import activity
from my_app.exports import export_service
from my_app.threaded_heartbeats import call_with_heartbeats


@activity.defn(name="wait_for_export")
def wait_for_export(export_id: str) -> dict:
    return call_with_heartbeats(
        lambda cancelled: export_service.wait(
            export_id, timeout=120, cancel_event=cancelled,
        ),
        interval=5,
    )

export_service is your application's client, not a Durable Workflow API. It must implement the finite timeout and observe cancel_event promptly, returning or raising when it is set. Use the equivalent timeout and cancellation mechanism from your own library. If it cannot cooperate, give it a finite I/O deadline; the helper cannot force a blocked thread to exit. A worker process may need separate isolation for a call that cannot be bounded.

The helper records the original exception, sets the cancellation event, and stops heartbeating when cancellation, lease-authority loss or a transport error occurs. After the work unwinds, it joins the helper and re-raises that exception before a result can be returned. In particular, do not swallow ActivityCancelled or ActivityLeaseAuthorityLost, replace them with an application error, or return a successful result after either. ActivityCancelled inherits from BaseException, so except Exception alone does not catch it. Ordinary return and work exceptions also stop and join the helper.

Keep the client and its worker event loop alive until the helper has joined. An in-flight heartbeat still needs its bounded transport request to finish. Use finite client timeouts and bounded retry policy as well as a deadline on the business operation. A helper reports liveness even if that operation hangs, so its heartbeat is not proof of progress. Choose the interval using the configured heartbeat timeout and leave room for request, scheduling and retry delays. Five seconds is an example, not a guaranteed safe cadence.

activity ¶

Activity decorator, registry, and execution context.

ActivityRetryInfo `dataclass` ¶

ActivityRetryInfo(attempt_number, policy_provided=False, policy=None)

Original attempt and immutable received retry snapshot, without defaults.

policy ¶

policy()

Return a fresh copy, including unknown fields and explicit legacy null.

max_attempts ¶

max_attempts()

A positive configured limit. An unknown limit does not mean unlimited.

is_unlimited ¶

is_unlimited()

Only an explicit max_attempts:null inside a policy means unlimited.

is_last_configured_attempt ¶

is_last_configured_attempt()

Check the budget, not whether another execution will actually occur.

Cancellation, non-retryable failure, workflow closure and deadlines can end execution before the configured budget is exhausted.

ActivityInfo `dataclass` ¶

ActivityInfo(task_id, activity_type, activity_attempt_id, attempt_number, task_queue, worker_id, task_metadata=None)

Metadata attached to the currently running activity attempt.

ActivityContext ¶

ActivityContext(*, info, client, heartbeat_callback=None)

Per-attempt activity context exposed by durable_workflow.activity.context.

info `property` ¶

info

Metadata for the currently running activity attempt.

is_cancelled `property` ¶

is_cancelled

True once the server has signalled that the owning workflow cancelled this activity.

heartbeat_sync ¶

heartbeat_sync(details=None)

Heartbeat from a synchronous activity thread and check attempt authority.

The request runs on the worker's event loop using the original attempt and client. Async handlers must use await heartbeat() instead.

heartbeat `async` ¶

heartbeat(details=None)

Report progress and check cancellation and continued attempt authority.

Long-running activities should call heartbeat() periodically so the server can distinguish a slow-but-alive attempt from a dead worker. Optional details are attached to the heartbeat and surface as the activity's last-known progress on failure.

Raises ActivityCancelled when the owning workflow has requested cancellation, so the activity can exit cleanly at its next natural break point.

Raises ActivityLeaseAuthorityLost with the original Server reason when the attempt cannot continue, including an elapsed deadline, closed attempt or closed run.

context ¶

context()

Return the current activity attempt context.

defn ¶

defn(*, name, typed=False)

Register a callable as an activity type under a language-neutral name.

With typed=True, positional arguments are decoded to the callable's parameter annotations: dataclasses, Enum (including IntEnum and StrEnum), Optional[T], list[T], dict[str, T] and the scalars str, int, float, bool and bytes. Keys a dataclass does not declare are ignored and missing fields take their defaults. A value that does not match fails the activity without retries with TypedValueError. An annotation outside this list raises TypeError here, when the activity is declared. Untyped activities keep receiving plain values.

registry ¶

registry()

Return a copy of activity callables registered in this process.