diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a360115..f7686c9 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ffaa15..79824c1 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/reference/activity.md b/docs/reference/activity.md index 80460bc..c27220b 100644 --- a/docs/reference/activity.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index d9bddd0..b8d8922 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 10ee58e..dde9398 100644 --- a/src/durable_workflow/__init__.py +++ b/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", diff --git a/src/durable_workflow/activity.py b/src/durable_workflow/activity.py index 3377d00..5334e4f 100644 --- a/src/durable_workflow/activity.py +++ b/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: diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 4df0b6f..abb6066 100644 --- a/src/durable_workflow/client.py +++ b/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-`` 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,6 +4497,12 @@ 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: @@ -4493,12 +4510,13 @@ async def signal_workflow( 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: diff --git a/src/durable_workflow/sync.py b/src/durable_workflow/sync.py index 499aef6..9c37e80 100644 --- a/src/durable_workflow/sync.py +++ b/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 diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 28ee9a4..10d4294 100644 --- a/src/durable_workflow/worker.py +++ b/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, ) diff --git a/tests/test_activity_retry_info.py b/tests/test_activity_retry_info.py new file mode 100644 index 0000000..cda0e58 --- /dev/null +++ b/tests/test_activity_retry_info.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import pickle +from dataclasses import FrozenInstanceError +from typing import Any + +import pytest + +from durable_workflow import ActivityInfo, ActivityRetryInfo, activity + + +@activity.defn(name="context.metadata") +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, + "policy_provided": retry.policy_provided, "policy": retry.policy(), "maximum": retry.max_attempts(), + "unlimited": retry.is_unlimited(), "last": retry.is_last_configured_attempt()} + + +@pytest.mark.parametrize(("attempt", "task", "maximum", "unlimited", "last"), [ + (1, {"retry_policy": {"max_attempts": 3}}, 3, False, False), + (3, {"retry_policy": {"max_attempts": 3}}, 3, False, True), + (4, {"retry_policy": {"max_attempts": 3}}, 3, False, True), + (1, {"retry_policy": {"max_attempts": 1}}, 1, False, True), + (1, {}, None, False, None), + (1, {"retry_policy": None}, None, False, None), + (99, {"retry_policy": {"max_attempts": None}}, None, True, False), + (1, {"retry_policy": {"future": [True, 2.0]}}, None, False, None), + (1, {"retry_policy": {"max_attempts": "3"}}, None, False, None), + (1, {"retry_policy": {"max_attempts": True}}, None, False, None), + (1, {"retry_policy": {"max_attempts": 3.0}}, None, False, None), + (1, {"retry_policy": {"max_attempts": 0}}, None, False, None), + (1, {"retry_policy": "legacy"}, None, False, None), + (0, {"retry_policy": {"max_attempts": 1}}, 1, False, None), +]) +def test_received_budget_preserves_unknown_and_unlimited( + attempt: int, task: dict[str, Any], maximum: int | None, unlimited: bool, last: bool | None, +) -> None: + retry = ActivityRetryInfo.from_task(attempt, task) + assert retry.attempt_number == attempt + assert retry.policy_provided is ("retry_policy" in task) + assert retry.policy() == task.get("retry_policy") + assert retry.max_attempts() == maximum + assert retry.is_unlimited() is unlimited + assert retry.is_last_configured_attempt() is last + + +def test_snapshot_copies_and_original_identity_survive_spawn_serialization() -> None: + task = {"workflow_id": "0", "run_id": "017", "retry_policy": {"max_attempts": 3, "future": [True, 2.0, None]}} + info = ActivityInfo("task", "activity", "attempt", 3, "queue", "owner", task_metadata=task) + task["retry_policy"]["max_attempts"] = 10 + task["workflow_id"] = "replacement" + copy = info.retry_info.policy() + copy["future"][0] = False + assert info.retry_info.policy() == {"max_attempts": 3, "future": [True, 2.0, None]} + assert info.workflow_id == "0" and info.run_id == "017" + assert info.retry_info.is_last_configured_attempt() is True + restored = pickle.loads(pickle.dumps(info)) + assert restored == info + assert restored.retry_info.policy() == info.retry_info.policy() + with pytest.raises(FrozenInstanceError): + info.workflow_id = "replacement" # type: ignore[misc] + with pytest.raises(FrozenInstanceError): + info.retry_info.attempt_number = 100 # type: ignore[misc] + + +def test_legacy_constructor_has_no_invented_received_policy_or_identity() -> None: + info = ActivityInfo("task", "activity", "attempt", 2, "queue", "owner") + assert info.retry_info.attempt_number == 2 + assert info.retry_info.policy_provided is False + assert info.retry_info.is_last_configured_attempt() is None + assert info.workflow_id is None and info.run_id is None + + +@pytest.mark.parametrize("task", [ + {}, {"workflow_id": None, "run_id": None}, {"workflow_id": "", "run_id": 42}, + {"workflow_instance_id": "internal", "workflow_run_id": "internal-run"}, +]) +def test_public_claim_identity_stays_unknown_when_missing_or_invalid(task: dict[str, Any]) -> None: + info = ActivityInfo("task", "activity", "attempt", 1, "queue", "owner", task_metadata=task) + assert info.workflow_id is None and info.run_id is None diff --git a/tests/test_client.py b/tests/test_client.py index 59e9400..f9e5538 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -869,6 +869,41 @@ async def test_maintenance_delegates_to_client(self, client: Client) -> None: class TestSignalWorkflow: + @pytest.mark.asyncio + @pytest.mark.parametrize("use_handle", [False, True]) + async def test_signal_keeps_original_run_and_business_key(self, client: Client, use_handle: bool) -> None: + seen = [] + + def receive(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(202, json={"outcome": "received"}) + + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(receive)) + async with client: + if use_handle: + handle = client.get_workflow_handle("invoice-42", run_id="original-run") + await handle.signal("payment_recorded", ["payment-17"], request_id="payment:17") + else: + await client.signal_workflow("invoice-42", "payment_recorded", args=["payment-17"], + run_id="original-run", request_id="payment:17") + assert len(seen) == 1 + assert seen[0].url.path == "/api/workflows/invoice-42/runs/original-run/signal/payment_recorded" + body = json.loads(seen[0].content) + assert body["request_id"] == "payment:17" + assert serializer.decode_envelope(body["input"]) == ["payment-17"] + + @pytest.mark.asyncio + @pytest.mark.parametrize("run_id", ["", False, 1]) + async def test_invalid_run_pin_never_falls_back_to_current(self, client: Client, run_id: Any) -> None: + with ( + patch.object(client._http, "request", new_callable=AsyncMock) as request, + pytest.raises(ValueError, match="run_id"), + ): + await client.signal_workflow("invoice-42", "payment_recorded", run_id=run_id) + request.assert_not_awaited() + await client.aclose() + @pytest.mark.asyncio async def test_signal(self, client: Client) -> None: resp = _mock_response(200, {"ok": True}) diff --git a/tests/test_cooperative_remote_worker.py b/tests/test_cooperative_remote_worker.py index 82f6e72..ad8f4c8 100644 --- a/tests/test_cooperative_remote_worker.py +++ b/tests/test_cooperative_remote_worker.py @@ -19,6 +19,7 @@ from durable_workflow.retry_policy import TransportRetryPolicy from durable_workflow.worker import Worker from tests.test_activity_process import wait_for_exit, wait_for_file +from tests.test_activity_retry_info import inspect_activity_context from tests.test_worker import compatible_cluster_info @@ -205,6 +206,25 @@ async def heartbeat(**kwargs: Any) -> dict[str, Any]: client.fail_activity_task.assert_not_awaited() +@pytest.mark.parametrize("configured", [False, True]) +async def test_supervised_worker_preserves_received_context(owner: Any, configured: bool) -> None: + worker, client = owner + worker.activities["remote"] = inspect_activity_context + work = task() + if configured: + work.update(attempt_number=3, workflow_id="0", run_id="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, + "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, + } + assert not worker._remote_activity_processes + + @pytest.mark.parametrize("field", ["lease", "heartbeat", "start_to_close", "schedule_to_close", "session", "malformed"]) async def test_invalid_or_elapsed_observed_bounds_prevent_callback_start(owner: Any, field: str) -> None: worker, client = owner diff --git a/tests/test_prepared_local_activity_worker.py b/tests/test_prepared_local_activity_worker.py index 906d20f..e774e66 100644 --- a/tests/test_prepared_local_activity_worker.py +++ b/tests/test_prepared_local_activity_worker.py @@ -20,6 +20,7 @@ from durable_workflow.worker import Worker from durable_workflow.workflow import LocalActivityExecutionAborted, replay from tests.test_activity_process import blocked_without_python_progress, wait_for_exit, wait_for_file, wait_for_release +from tests.test_activity_retry_info import inspect_activity_context from tests.test_cancellation_context import delivery, request, snapshot from tests.test_replay_regression_corpus import PreparedLocalColdResultsWorkflow from tests.test_worker import compatible_cluster_info @@ -415,6 +416,26 @@ 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( + 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} + 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, + "policy": None, "maximum": None, "unlimited": False, "last": None, + } + server.client.fail_workflow_task.assert_not_awaited() + assert not worker._prepared_local_activity_processes + finally: + await worker.stop() + + async def test_bad_admission_never_spawns_and_never_completes_or_fails_a_claim( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: diff --git a/tests/test_sync.py b/tests/test_sync.py index 34e2e35..2d7309f 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -221,6 +221,14 @@ def test_describe(self) -> None: class TestSyncClientSignal: + def test_signal_pins_the_original_run(self) -> None: + with Client("http://localhost:8080") as client: + with patch.object(client._async._http, "request", new_callable=AsyncMock, + return_value=_mock_response(202, {"outcome": "received"})) as request: + client.signal_workflow("invoice-42", "payment_recorded", run_id="original-run", request_id="payment:17") + assert request.call_args.args[1] == "/api/workflows/invoice-42/runs/original-run/signal/payment_recorded" + assert request.call_args.kwargs["json"]["request_id"] == "payment:17" + def test_signal(self) -> None: client = Client("http://localhost:8080") resp = _mock_response(202, {"outcome": "received"}) diff --git a/tests/test_worker.py b/tests/test_worker.py index 63fadf6..141df05 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -48,6 +48,7 @@ _should_fail_workflow_task_after_completion_error, ) from tests.query_task_helpers import leased_query_task +from tests.test_activity_retry_info import inspect_activity_context class TypedCancelFlightError(Exception): @@ -1341,6 +1342,27 @@ async def test_local_activity_runs_in_workflow_worker_and_records_heartbeats( assert [command["type"] for command in replayed] == ["complete_workflow"] mock_client.heartbeat_workflow_task.assert_not_awaited() + @pytest.mark.asyncio + async def test_inline_local_activity_preserves_parent_context_only(self, mock_client: AsyncMock) -> None: + mock_client.heartbeat_workflow_task.return_value = { + "task_id": "local-task", "lease_owner": "local-worker", "workflow_task_attempt": 1, "renewed": True, + } + worker = Worker(mock_client, task_queue="q1", worker_id="local-worker", + workflows=[LocalWorkerWorkflow], activities=[local_echo]) + worker.activities["local.echo"] = inspect_activity_context + 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}, + "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, + "policy": None, "maximum": None, "unlimited": False, "last": None, + } + mock_client.fail_workflow_task.assert_not_awaited() + @pytest.mark.asyncio async def test_local_activity_lease_refusal_never_completes_task(self, mock_client: AsyncMock) -> None: mock_client.heartbeat_workflow_task.return_value = { @@ -3316,6 +3338,23 @@ async def test_cancellation_before_dispatch_starts_releases_admitted_capacity(se 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", + "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", + **metadata} + assert await worker._run_activity_task(task) == "completed" + 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, + "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, + } + @pytest.mark.asyncio async def test_sync_activity(self, mock_client: AsyncMock) -> None: worker = Worker(mock_client, task_queue="q1", workflows=[], activities=[echo_activity])