Expose logical execution identity in ActivityInfo

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`. 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.
+
 - Lazy `Client.iter_workflows`, `Client.iter_history` and `WorkflowHandle.iter_history` follow opaque page tokens while retaining only the current page and bounded cursor bookkeeping. They preserve run/filter/namespace identity, follow empty continued pages, stop on early close, and surface HTTP, cancellation and invalid/cyclic cursor errors. Existing one-page methods keep their contracts. (durable-workflow/.github#247)
 
 ## [2.15.0] - 2026-11-04

docs/sdk-reference.md

@@ -226,6 +226,53 @@ 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.
+
+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
+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

src/durable_workflow/activity.py

@@ -82,13 +82,17 @@ 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 {}
         object.__setattr__(self, "retry_info", ActivityRetryInfo.from_task(self.attempt_number, task))
         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:

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,

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()

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()}
 

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,

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()

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,