Expose activity retry context and pin originating-run signals

.github/workflows/ci.yml

@@ -169,7 +169,8 @@ jobs:
         shell: bash
         run: |
           if [[ -f "$RUNNER_TEMP/sdk-unit-tests.log" ]]; then
-            phase_unit_tail="$(tail -c 12000 "$RUNNER_TEMP/sdk-unit-tests.log")"
+            # Check annotations truncate at 4096 bytes; keep the final summary.
+            phase_unit_tail="$(tail -c 3000 "$RUNNER_TEMP/sdk-unit-tests.log")"
             phase_unit_tail="${phase_unit_tail//'%'/'%25'}"
             phase_unit_tail="${phase_unit_tail//$'\r'/'%0D'}"
             phase_unit_tail="${phase_unit_tail//$'\n'/'%0A'}"
@@ -267,11 +268,38 @@ jobs:
         working-directory: sdk-python
         run: python scripts/ci/configure-integration-endpoint.py
       - name: Run integration tests
+        id: integration_tests
         shell: bash
         working-directory: sdk-python
         env:
           DURABLE_WORKFLOW_AUTH_TOKEN: test-token
         run: pytest tests/integration/ -v --capture=tee-sys --junitxml=integration-results.xml 2>&1 | tee integration-scenarios.log
+      - name: Report bounded integration failure identities
+        if: ${{ always() && steps.integration_tests.outcome == 'failure' }}
+        working-directory: sdk-python
+        run: |
+          python - <<'PY'
+          from pathlib import Path
+          import re
+          import xml.etree.ElementTree as ET
+
+          report = Path("integration-results.xml")
+          if not report.is_file():
+              print("::error title=Integration diagnostics::JUnit report unavailable")
+              raise SystemExit(0)
+          failed = 0
+          for case in ET.parse(report).getroot().iter("testcase"):
+              if case.find("failure") is None and case.find("error") is None:
+                  continue
+              failed += 1
+              if failed > 20:
+                  continue
+              identity = case.get("classname", "") + "." + case.get("name", "").partition("[")[0]
+              if len(identity) > 256 or re.fullmatch(r"[A-Za-z0-9_.]+", identity) is None:
+                  identity = "unidentified_integration_case"
+              print(f"::error title=Integration test failure::{identity}")
+          print(f"::notice title=Integration summary::failed_test_cases={failed}")
+          PY
       - name: Retain connected scenario results
         if: ${{ always() && github.server_url == 'https://github.com' }}
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a  # v7

CHANGELOG.md

@@ -6,6 +6,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.12.0] - 2026-10-27
+
+### Added
+
+- Expose immutable received retry policy and original attempt metadata through
+  `ActivityInfo.retry_info`, with explicit unknown, legacy-null and unlimited
+  states and a last-configured-attempt check.
+- Expose nullable original workflow/run IDs in remote, inline and prepared
+  activity contexts without substituting task or activity-attempt IDs.
+- Add an optional `run_id` to async and sync signal calls, with canonical
+  activity guidance for stable business-event keys and delivery reconciliation.
+
+### Fixed
+
+- Preserve the bound run when `WorkflowHandle.signal` sends a signal, instead
+  of silently targeting whichever run is current.
+
 ## [2.11.2] - 2026-10-25
 
 ### Fixed

docs/reference/activity.md

@@ -7,6 +7,68 @@ default. They can call `context().heartbeat_sync(details)` and inspect
 See [activity execution and shutdown](worker.md#activity-execution) 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.
+
+```python
+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](https://github.com/durable-workflow/.github/issues/216).
+
 ## Heartbeats during one blocking call
 
 Capture `ctx = activity.context()` inside the activity before starting a helper

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.11.2"
+version = "2.12.0"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -76,8 +76,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.11.2"
-registry-version = "2.11.2"
+product-train = "2.12.0"
+registry-version = "2.12.0"
 supported-server-versions = "2.8.0"
 worker-protocol-version = "1.20"
 control-plane-version = "2"

src/durable_workflow/__init__.py

@@ -7,7 +7,7 @@
     __version__ = "0.0.0+unknown"
 
 from . import activity, sync, testing, workflow
-from .activity import ActivityContext, ActivityInfo
+from .activity import ActivityContext, ActivityInfo, ActivityRetryInfo
 from .auth_composition import (
     AUTH_COMPOSITION_CONTRACT_SCHEMA,
     AUTH_COMPOSITION_CONTRACT_VERSION,
@@ -248,6 +248,7 @@
     "ActivityContext",
     "ActivityHandler",
     "ActivityInfo",
+    "ActivityRetryInfo",
     "ActivityInterceptorContext",
     "ActivityRetryPolicy",
     "BridgeAdapterOutcome",

src/durable_workflow/activity.py

@@ -3,8 +3,9 @@
 
 import asyncio
 import contextvars
-from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
+import json
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import InitVar, dataclass, field
 from typing import TYPE_CHECKING, Any
 
 from ._activity_lease import current_activity_authority
@@ -20,6 +21,53 @@
 )
 
 
+@dataclass(frozen=True, init=False)
+class ActivityRetryInfo:
+    """Original attempt and immutable received retry snapshot, without defaults."""
+
+    attempt_number: int
+    policy_provided: bool
+    # Immutable JSON also survives the supervised callback's spawn serialization.
+    _policy_json: str = field(repr=False)
+
+    def __init__(self, attempt_number: int, policy_provided: bool = False, policy: Any = None) -> None:
+        object.__setattr__(self, "attempt_number", attempt_number)
+        object.__setattr__(self, "policy_provided", policy_provided)
+        object.__setattr__(self, "_policy_json", json.dumps(policy))
+
+    @classmethod
+    def from_task(cls, attempt_number: int, task: Mapping[str, Any]) -> ActivityRetryInfo:
+        return cls(attempt_number, "retry_policy" in task, task.get("retry_policy"))
+
+    def policy(self) -> Any:
+        """Return a fresh copy, including unknown fields and explicit legacy null."""
+        return json.loads(self._policy_json)
+
+    def max_attempts(self) -> int | None:
+        """A positive configured limit. An unknown limit does not mean unlimited."""
+        policy = self.policy()
+        maximum = policy.get("max_attempts") if isinstance(policy, dict) else None
+        return maximum if type(maximum) is int and maximum > 0 else None
+
+    def is_unlimited(self) -> bool:
+        """Only an explicit max_attempts:null inside a policy means unlimited."""
+        policy = self.policy()
+        return isinstance(policy, dict) and "max_attempts" in policy and policy["max_attempts"] is None
+
+    def is_last_configured_attempt(self) -> bool | None:
+        """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.
+        """
+        if self.attempt_number < 1:
+            return None
+        maximum = self.max_attempts()
+        if maximum is not None:
+            return self.attempt_number >= maximum
+        return False if self.is_unlimited() else None
+
+
 @dataclass(frozen=True)
 class ActivityInfo:
     """Metadata attached to the currently running activity attempt."""
@@ -30,6 +78,17 @@ class ActivityInfo:
     attempt_number: int
     task_queue: str
     worker_id: str
+    task_metadata: InitVar[Mapping[str, Any] | None] = None
+    retry_info: ActivityRetryInfo = field(init=False)
+    workflow_id: str | None = field(init=False)
+    run_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)
 
 
 class ActivityContext:

src/durable_workflow/client.py

@@ -1386,6 +1386,10 @@ async def signal(
         request_id: str | None = None,
     ) -> dict[str, Any]:
         """Deliver an external signal to this workflow. See :meth:`Client.signal_workflow`."""
+        if self.run_id is not None:
+            return await self._client.signal_workflow(
+                self.workflow_id, signal_name, args=args, request_id=request_id, run_id=self.run_id,
+            )
         return await self._client.signal_workflow(
             self.workflow_id, signal_name, args=args, request_id=request_id,
         )
@@ -4461,6 +4465,7 @@ async def signal_workflow(
         *,
         args: list[Any] | None = None,
         request_id: str | None = None,
+        run_id: str | None = None,
     ) -> dict[str, Any]:
         """Deliver an external signal to a running workflow.
 
@@ -4469,12 +4474,18 @@ async def signal_workflow(
         observe the signal. See the main docs for how to declare the allowed
         signal names on a workflow type.
 
+        Pass ``run_id`` to target the original run. Omitting it targets the
+        current run when the Server receives the request. A run-pinned
+        :class:`WorkflowHandle` forwards its run identity automatically.
+
         ``request_id`` is the caller command key (1-255 opaque, case-sensitive
-        bytes) that makes a retry of this signal safe. When omitted the client
+        bytes). When omitted the client
         generates ``sdk-python-signal-<uuid>`` once per call, so the transport
         retry policy resends the same key. Resending the same key with the same
         signal name, arguments and run target returns the original receipt
-        instead of recording a second signal.
+        only when the runtime supports and applies command-key retention.
+        A runtime without that retention can record repeated signals. Keep
+        business-event processing idempotent and reconcile uncertain delivery.
 
         Returns the Server's admission receipt. ``request_id`` is always
         present (the key that was sent); a Server that retains command keys
@@ -4486,19 +4497,26 @@ async def signal_workflow(
         in its ``details``. Every exception raised after the key was chosen,
         including transport errors, carries it as ``request_id``.
         """
+        if run_id is not None and (not isinstance(run_id, str) or not run_id):
+            raise ValueError("run_id must be a non-empty string when provided")
+        path = f"/workflows/{quote(workflow_id, safe='')}"
+        if run_id is not None:
+            path += f"/runs/{quote(run_id, safe='')}"
+        path += f"/signal/{quote(signal_name, safe='')}"
         key = _command_request_id(request_id, "signal")
         body: dict[str, Any] = {}
         if args:
             body["input"] = self._payload_envelope(
                 args,
                 kind="signal",
                 workflow_id=workflow_id,
+                run_id=run_id,
                 signal_name=signal_name,
             )
         body["request_id"] = key
         try:
             data = await self._request(
-                "POST", f"/workflows/{workflow_id}/signal/{signal_name}",
+                "POST", path,
                 json=self._with_observability_context(body), context=workflow_id,
             )
         except Exception as exc:

src/durable_workflow/sync.py

@@ -535,9 +535,10 @@ def signal_workflow(
         *,
         args: list[Any] | None = None,
         request_id: str | None = None,
+        run_id: str | None = None,
     ) -> dict[str, Any]:
         result: dict[str, Any] = self._runner.run(
-            self._async.signal_workflow(workflow_id, signal_name, args=args, request_id=request_id)
+            self._async.signal_workflow(workflow_id, signal_name, args=args, request_id=request_id, run_id=run_id)
         )
         return result
 

src/durable_workflow/worker.py

@@ -2105,6 +2105,7 @@ 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, "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,
@@ -2462,6 +2463,7 @@ async def heartbeat(
                     attempt_number=attempt_number,
                     task_queue=self.task_queue,
                     worker_id=self.worker_id,
+                    task_metadata={"workflow_id": task.get("workflow_id"), "run_id": task.get("run_id")},
                 )
                 _set_context(ActivityContext(info=info, client=self.client, heartbeat_callback=heartbeat))
                 try:
@@ -3196,6 +3198,7 @@ async def _run_activity_task_body(self, task: dict[str, Any]) -> str:
                 attempt_number=attempt_number,
                 task_queue=self.task_queue,
                 worker_id=self.worker_id,
+                task_metadata=task,
             ),
             client=self.client,
         )