From c286f19a81b87599ce1fb9225fb66a8b914d901d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:53:10 -0600 Subject: [PATCH 1/3] Expose received logical activity identity in callback context --- CHANGELOG.md | 2 + docs/sdk-reference.md | 40 ++++++++++ src/durable_workflow/activity.py | 4 + src/durable_workflow/worker.py | 3 +- tests/test_activity_identity.py | 79 ++++++++++++++++++++ tests/test_activity_retry_info.py | 3 +- tests/test_cooperative_remote_worker.py | 3 +- tests/test_prepared_local_activity_worker.py | 6 +- tests/test_worker.py | 7 +- 9 files changed, 140 insertions(+), 7 deletions(-) create mode 100644 tests/test_activity_identity.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bbf7d57..ea2fb440 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Unreleased `ActivityInfo.activity_execution_id` exposes the received logical activity identity without task/attempt/session fallbacks. Prepared local callbacks receive their validated admission ID; inline local callbacks and older claims without an ID expose `None`. Shared identity acceptance remains tracked in durable-workflow/.github#211. + - Worker schedule-to-start histogram `durable_workflow_worker_task_schedule_to_start_seconds` for workflow and activity dispatches, tagged only with `task_kind` and `task_queue`. It requires Server's `claimed_task_ready_age` capability and a valid integer `ready_age_ms`, records Server-computed milliseconds as seconds, preserves real zero and omits missing/invalid samples. The shared metrics contract is pinned at revision 2. Published-Server and installed-package acceptance remains pending in #188. - Condition wait opens carry a deterministic `python:condition-wait:{ordinal}` occurrence ID for each authored call, including selection members. False-wake reopens preserve the first open's ID and remaining timeout; legacy chains without an ID keep that absence even when later physical opens have an ID. Separate calls and grace waits receive new ordinals. These identities enable unresolved-open deduplication on supporting runtimes. (#259) diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index c746b5a2..ee0b69fe 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -226,6 +226,46 @@ receipt = yield ctx.start_child_workflow( ) ``` +## Activity execution identity (unreleased) + +Inside an activity, `activity.context().info.activity_execution_id` exposes the +logical execution ID received from Server. `workflow_id` and `run_id` identify +its owning workflow and run. The execution ID is the idempotency key for the +logical call across retries. `task_id` and `activity_attempt_id` identify an +attempt and must not replace it. + +Use the received key when calling an external service that supports idempotency: + +```python +from durable_workflow import activity +from durable_workflow.errors import NonRetryableError + + +def current_idempotency_key() -> str: + execution_id = activity.context().info.activity_execution_id + if execution_id is None: + raise NonRetryableError("this operation requires a Server activity execution ID") + return execution_id +``` + +Pass that key to the external service's idempotency mechanism. Keep different +external operations within one activity distinguishable using your own stable +operation suffix. Idempotency still depends on the external service honoring +the key. + +The accessor is read-only. Missing, invalid or blank execution IDs are `None`, +including on older claims, and are never derived from task, attempt or session +IDs. Nonblank received strings are preserved exactly. Session-routed activities +use their ordinary activity claim identity. Prepared local activities expose the +validated admission receipt's execution ID. Inline local activities have no +admitted execution ID and expose `None`, while retaining their parent workflow +and run IDs. Server-free mocks do not create Server activity identity. + +This source addition is unreleased and is outside the current typed-value +release candidate. Published-package retry identity and execution-guarantees +guide acceptance remain tracked in +[the shared identity issue](https://github.com/durable-workflow/.github/issues/211). + ## Local activities Use a local activity for short work that should run in the workflow worker diff --git a/src/durable_workflow/activity.py b/src/durable_workflow/activity.py index 27bf7588..ebc810df 100644 --- a/src/durable_workflow/activity.py +++ b/src/durable_workflow/activity.py @@ -82,6 +82,7 @@ class ActivityInfo: retry_info: ActivityRetryInfo = field(init=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: task = task_metadata if task_metadata is not None else {} @@ -89,6 +90,9 @@ def __post_init__(self, task_metadata: Mapping[str, Any] | None) -> None: for name in ("workflow_id", "run_id"): value = task.get(name) object.__setattr__(self, name, value if isinstance(value, str) and value else None) + execution_id = task.get("activity_execution_id") + object.__setattr__(self, "activity_execution_id", + execution_id if isinstance(execution_id, str) and execution_id.strip() else None) class ActivityContext: diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index cc4e3edb..b3dffe15 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -2145,7 +2145,8 @@ def observe(value: Mapping[str, Any]) -> None: handler, tuple(call.command.arguments), ActivityInfo( task_id, call.command.activity_type, attempt.attempt_id, attempt.attempt_number, self.task_queue, self.worker_id, - task_metadata={"workflow_id": task.get("workflow_id"), "run_id": task.get("run_id")}, + task_metadata={"workflow_id": task.get("workflow_id"), "run_id": task.get("run_id"), + "activity_execution_id": attempt.execution_id}, ), {**task, "activity_attempt_id": attempt.attempt_id}, self.interceptors, namespace=getattr(self.client, "namespace", None), execution_context=telemetry.capture() if telemetry is not None else None, diff --git a/tests/test_activity_identity.py b/tests/test_activity_identity.py new file mode 100644 index 00000000..809c019b --- /dev/null +++ b/tests/test_activity_identity.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import pickle +from dataclasses import FrozenInstanceError +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import ActivityInfo, Worker, activity, serializer +from durable_workflow.client import Client + + +@pytest.mark.parametrize(("received", "expected"), [ + ("execution-1", "execution-1"), ("0", "0"), (" opaque ", " opaque "), + (None, None), ("", None), (" \t\n", None), (False, None), (0, None), (1, None), + ([], None), ({"id": "execution-1"}, None), +]) +def test_execution_identity_preserves_only_received_nonblank_strings(received: Any, expected: str | None) -> None: + info = ActivityInfo("task", "activity", "attempt", 1, "queue", "owner", task_metadata={ + "activity_execution_id": received, "worker_session": {"session_id": "not-an-execution"}, + }) + assert info.activity_execution_id == expected + + +def test_legacy_constructor_does_not_invent_logical_identity() -> None: + info = ActivityInfo("task", "activity", "attempt", 1, "queue", "owner") + assert info.activity_execution_id is None + assert info.workflow_id is None + assert info.run_id is None + + +def test_received_identity_is_immutable_and_survives_callback_serialization() -> None: + task = {"activity_execution_id": "original", "workflow_id": "workflow", "run_id": "run"} + info = ActivityInfo("task", "activity", "attempt", 1, "queue", "owner", task_metadata=task) + task["activity_execution_id"] = "replacement" + restored = pickle.loads(pickle.dumps(info)) + assert restored.activity_execution_id == "original" + assert restored.workflow_id == "workflow" + assert restored.run_id == "run" + with pytest.raises(FrozenInstanceError): + info.activity_execution_id = "replacement" # type: ignore[misc] + + +@pytest.mark.parametrize("session_routed", [False, True]) +async def test_received_retry_keeps_execution_identity_and_distinct_attempts(session_routed: bool) -> None: + client = AsyncMock(spec=Client) + seen: list[ActivityInfo] = [] + + @activity.defn(name="identity.retry") + async def inspect_identity() -> str | None: + info = activity.context().info + seen.append(info) + if info.activity_execution_id == "execution-1" and info.attempt_number == 1: + raise ValueError("retry this received attempt") + return info.activity_execution_id + + worker = Worker(client, task_queue="queue", activities=[inspect_identity]) + identities = [("execution-1", 1), ("execution-1", 2), ("execution-2", 1), (None, 1)] + try: + for index, (execution, attempt) in enumerate(identities): + task = { + "task_id": f"task-{index}", "activity_attempt_id": f"attempt-{index}", + "attempt_number": attempt, "activity_type": "identity.retry", + "arguments": serializer.envelope([]), "payload_codec": "avro", + "workflow_id": "workflow", "run_id": "run", + **({"activity_execution_id": execution} if execution is not None else {}), + **({"worker_session": {"session_id": "session-1"}} if session_routed else {}), + } + outcome = await worker._run_activity_task(task) + assert outcome == ("failed" if index == 0 else "completed") + assert [info.activity_execution_id for info in seen] == ["execution-1", "execution-1", "execution-2", None] + assert len({info.activity_attempt_id for info in seen}) == 4 + assert {info.workflow_id for info in seen} == {"workflow"} + assert {info.run_id for info in seen} == {"run"} + client.fail_activity_task.assert_awaited_once() + assert client.complete_activity_task.await_count == 3 + finally: + await worker.stop() diff --git a/tests/test_activity_retry_info.py b/tests/test_activity_retry_info.py index cda0e585..20875106 100644 --- a/tests/test_activity_retry_info.py +++ b/tests/test_activity_retry_info.py @@ -13,7 +13,8 @@ def inspect_activity_context(_marker: str | None = None) -> dict[str, Any]: info = activity.context().info retry = info.retry_info - return {"workflow_id": info.workflow_id, "run_id": info.run_id, "attempt": retry.attempt_number, + return {"workflow_id": info.workflow_id, "run_id": info.run_id, + "activity_execution_id": info.activity_execution_id, "attempt": retry.attempt_number, "policy_provided": retry.policy_provided, "policy": retry.policy(), "maximum": retry.max_attempts(), "unlimited": retry.is_unlimited(), "last": retry.is_last_configured_attempt()} diff --git a/tests/test_cooperative_remote_worker.py b/tests/test_cooperative_remote_worker.py index ad8f4c88..254c7562 100644 --- a/tests/test_cooperative_remote_worker.py +++ b/tests/test_cooperative_remote_worker.py @@ -212,12 +212,13 @@ async def test_supervised_worker_preserves_received_context(owner: Any, configur worker.activities["remote"] = inspect_activity_context work = task() if configured: - work.update(attempt_number=3, workflow_id="0", run_id="017", + work.update(attempt_number=3, workflow_id="0", run_id="017", activity_execution_id="execution-017", retry_policy={"max_attempts": 3, "future": [True, None]}) assert await worker._run_activity_task(work) == "completed" client.fail_activity_task.assert_not_awaited() assert client.complete_activity_task.await_args.kwargs["result"] == { "workflow_id": "0" if configured else None, "run_id": "017" if configured else None, + "activity_execution_id": "execution-017" if configured else None, "attempt": 3 if configured else 1, "policy_provided": configured, "policy": {"max_attempts": 3, "future": [True, None]} if configured else None, "maximum": 3 if configured else None, "unlimited": False, "last": True if configured else None, diff --git a/tests/test_prepared_local_activity_worker.py b/tests/test_prepared_local_activity_worker.py index e774e66f..343bcb57 100644 --- a/tests/test_prepared_local_activity_worker.py +++ b/tests/test_prepared_local_activity_worker.py @@ -416,18 +416,20 @@ async def test_worker_runs_only_an_admitted_process_then_replays_canonical_resul await wait_for_exit(int((tmp_path / "callback").read_text())) -async def test_prepared_local_activity_preserves_parent_context_only( +async def test_prepared_local_activity_preserves_admitted_execution_and_parent_context( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: server = PreparedServer() worker = await server.worker(monkeypatch, handler=inspect_activity_context) task = server.task(tmp_path / "unused") task["retry_policy"] = {"max_attempts": 9} + task["activity_execution_id"] = "unrelated-parent-metadata" try: commands = await worker._run_workflow_task(task) assert commands is not None and [command["type"] for command in commands] == ["complete_workflow"] assert serializer.decode_envelope(commands[0]["result"]) == { - "workflow_id": "workflow-1", "run_id": "run-1", "attempt": 1, "policy_provided": False, + "workflow_id": "workflow-1", "run_id": "run-1", "activity_execution_id": "execution", + "attempt": 1, "policy_provided": False, "policy": None, "maximum": None, "unlimited": False, "last": None, } server.client.fail_workflow_task.assert_not_awaited() diff --git a/tests/test_worker.py b/tests/test_worker.py index b795a3dd..281af796 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -1353,12 +1353,14 @@ async def test_inline_local_activity_preserves_parent_context_only(self, mock_cl commands = await worker._run_workflow_task({ "task_id": "local-task", "workflow_type": "local-worker-wf", "workflow_task_attempt": 1, "workflow_id": "0", "run_id": "017", "retry_policy": {"max_attempts": 9}, + "activity_execution_id": "unrelated-parent-metadata", "history_events": [], "arguments": serializer.envelope(["ignored"]), "payload_codec": "avro", }) assert commands is not None assert [command["type"] for command in commands] == ["record_local_activity", "complete_workflow"] assert serializer.decode_envelope(commands[-1]["result"]) == { - "workflow_id": "0", "run_id": "017", "attempt": 1, "policy_provided": False, + "workflow_id": "0", "run_id": "017", "activity_execution_id": None, + "attempt": 1, "policy_provided": False, "policy": None, "maximum": None, "unlimited": False, "last": None, } mock_client.fail_workflow_task.assert_not_awaited() @@ -3341,7 +3343,7 @@ class TestActivityTaskExecution: @pytest.mark.parametrize("configured", [False, True]) async def test_ordinary_worker_preserves_received_context(self, mock_client: AsyncMock, configured: bool) -> None: worker = Worker(mock_client, task_queue="q1", activities=[inspect_activity_context]) - metadata = {"attempt_number": 3, "workflow_id": "0", "run_id": "017", + metadata = {"attempt_number": 3, "workflow_id": "0", "run_id": "017", "activity_execution_id": "execution-017", "retry_policy": {"max_attempts": 3, "future": [True, None]}} if configured else {} task = {"task_id": "metadata-task", "activity_attempt_id": "metadata-attempt", "activity_type": "context.metadata", "arguments": serializer.envelope([]), "payload_codec": "avro", @@ -3350,6 +3352,7 @@ async def test_ordinary_worker_preserves_received_context(self, mock_client: Asy mock_client.fail_activity_task.assert_not_awaited() assert mock_client.complete_activity_task.await_args.kwargs["result"] == { "workflow_id": "0" if configured else None, "run_id": "017" if configured else None, + "activity_execution_id": "execution-017" if configured else None, "attempt": 3 if configured else 1, "policy_provided": configured, "policy": {"max_attempts": 3, "future": [True, None]} if configured else None, "maximum": 3 if configured else None, "unlimited": False, "last": True if configured else None, From eb73b5f86729ea2e7638cb07ed30eb21d10157e6 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:14:07 -0600 Subject: [PATCH 2/3] Clarify that activity identity is idempotency metadata --- CHANGELOG.md | 2 +- docs/sdk-reference.md | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea2fb440..6a1a0dbb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Unreleased `ActivityInfo.activity_execution_id` exposes the received logical activity identity without task/attempt/session fallbacks. Prepared local callbacks receive their validated admission ID; inline local callbacks and older claims without an ID expose `None`. Shared identity acceptance remains tracked in durable-workflow/.github#211. +- Unreleased `ActivityInfo.activity_execution_id` exposes the received logical activity identity without task/attempt/session fallbacks. Prepared local callbacks receive their validated admission ID; inline local callbacks and older claims without an ID expose `None`. These IDs are public idempotency metadata, not live-claim or external-effect authority, and replace no credentials, heartbeat/current-attempt checks, per-effect suffixes or outcome reconciliation. Shared identity acceptance remains tracked in durable-workflow/.github#211. - Worker schedule-to-start histogram `durable_workflow_worker_task_schedule_to_start_seconds` for workflow and activity dispatches, tagged only with `task_kind` and `task_queue`. It requires Server's `claimed_task_ready_age` capability and a valid integer `ready_age_ms`, records Server-computed milliseconds as seconds, preserves real zero and omits missing/invalid samples. The shared metrics contract is pinned at revision 2. Published-Server and installed-package acceptance remains pending in #188. diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index ee0b69fe..d939d084 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -253,6 +253,13 @@ external operations within one activity distinguishable using your own stable operation suffix. Idempotency still depends on the external service honoring the key. +An execution ID is public idempotency metadata. Application code and tests can +construct, copy or deserialize an `ActivityInfo` with matching IDs; this does +not prove ownership of a live Server claim or authenticate or authorize an +external effect. Keep the external service's credentials, current-attempt and +heartbeat authority checks, per-effect key suffixes, and outcome +compare-and-swap or reconciliation. The ID replaces none of those controls. + The accessor is read-only. Missing, invalid or blank execution IDs are `None`, including on older claims, and are never derived from task, attempt or session IDs. Nonblank received strings are preserved exactly. Session-routed activities From db7e1059316e794df66e33a63dc434a8120c020a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:15:05 -0600 Subject: [PATCH 3/3] Keep activity identity outside the frozen 2.15 changelog --- CHANGELOG.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c4bd312e..571ad279 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,12 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -## [2.15.0] - 2026-11-04 - ### Added - Unreleased `ActivityInfo.activity_execution_id` exposes the received logical activity identity without task/attempt/session fallbacks. Prepared local callbacks receive their validated admission ID; inline local callbacks and older claims without an ID expose `None`. These IDs are public idempotency metadata, not live-claim or external-effect authority, and replace no credentials, heartbeat/current-attempt checks, per-effect suffixes or outcome reconciliation. Shared identity acceptance remains tracked in durable-workflow/.github#211. +## [2.15.0] - 2026-11-04 + +### Added + - Worker schedule-to-start histogram `durable_workflow_worker_task_schedule_to_start_seconds` for workflow and activity dispatches, tagged only with `task_kind` and `task_queue`. It requires Server's `claimed_task_ready_age` capability and a valid integer `ready_age_ms`, records Server-computed milliseconds as seconds, preserves real zero and omits missing/invalid samples. The shared metrics contract is pinned at revision 2. Published-Server and installed-package acceptance remains pending in #188. - Condition wait opens carry a deterministic `python:condition-wait:{ordinal}` occurrence ID for each authored call, including selection members. False-wake reopens preserve the first open's ID and remaining timeout; legacy chains without an ID keep that absence even when later physical opens have an ID. Separate calls and grace waits receive new ordinals. These identities enable unresolved-open deduplication on supporting runtimes. (#259)