From 6c0ffdb6b3be359686eddc4e76d022e1e63f0628 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 17:08:29 +0000 Subject: [PATCH] Generate and reuse durable request ids for signals and updates Part of durable-workflow/.github#216 (safe signal/update retries after a lost response). Work in progress; see the draft PR body for the remaining steps. Co-Authored-By: Claude Fable 5.1 --- src/durable_workflow/client.py | 138 +++++++++- src/durable_workflow/errors.py | 79 +++++- src/durable_workflow/retry_policy.py | 24 +- src/durable_workflow/sync.py | 27 +- .../workflow_updates_conformance.py | 74 +++++- tests/test_client.py | 249 +++++++++++++++++- tests/test_errors.py | 73 +++++ tests/test_retry_policy.py | 12 + tests/test_sync.py | 28 +- tests/test_workflow_updates_conformance.py | 11 +- 10 files changed, 674 insertions(+), 41 deletions(-) diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 442737ad..c8e9215c 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -19,6 +19,7 @@ from __future__ import annotations import asyncio +import contextlib import copy import hashlib import json as json_module @@ -47,6 +48,7 @@ from ._cooperative_cancellation import CancellationDelivery, CancellationRequest from ._query_task_authority import QueryTaskAuthority, current_query_authority from .errors import ( + REQUEST_ID_CONFLICT, ExternalPayloadIntegrityMismatch, ExternalPayloadOversized, ExternalPayloadUnavailable, @@ -55,6 +57,8 @@ RuntimeCapabilityUnsupported, RuntimeDiscoveryUnavailable, ServerError, + SignalFailed, + UpdateFailed, WorkflowCancelled, WorkflowFailed, WorkflowTerminated, @@ -239,6 +243,49 @@ def _normalize_base_url(base_url: str) -> str: return base_url.rstrip("/") +def _command_request_id(request_id: str | None, command: str) -> str: + """Return the caller command key for one logical signal or update call. + + The key is the body field ``request_id`` (1-255 opaque, case-sensitive + bytes). A caller-supplied key is sent verbatim; otherwise one + ``sdk-python--`` key is generated per logical call so + the transport retry policy resends the identical key. + """ + return request_id or f"sdk-python-{command}-{uuid.uuid4().hex}" + + +def _attach_request_id(exc: BaseException, request_id: str) -> None: + """Record the sent command key on an exception as ``exc.request_id``. + + Typed SDK failures declare the attribute; transport errors raised by + ``httpx`` receive it dynamically so a manual retry can reuse the key. + """ + with contextlib.suppress(AttributeError, TypeError): + exc.request_id = request_id # type: ignore[attr-defined] + + +def _command_request_failure(exc: Exception, *, command: str, request_id: str) -> Exception: + """Return the exception a failed signal or update call should raise. + + A ``request_id_conflict`` refusal that reached the caller as a generic + :class:`ServerError` becomes the command's typed failure. Every other + exception is returned unchanged, annotated with the key that was sent. + """ + if isinstance(exc, ServerError) and exc.reason() == REQUEST_ID_CONFLICT: + message = exc.body.get("message") if isinstance(exc.body, dict) else None + if not isinstance(message, str) or not message: + message = f"{command} request id already retained with a different intent" + if command == "update" and isinstance(exc.body, dict): + return UpdateFailed( + message, status=exc.status, body=exc.body, reason=REQUEST_ID_CONFLICT, request_id=request_id, + ) + return SignalFailed( + message, reason=REQUEST_ID_CONFLICT, status=exc.status, body=exc.body, request_id=request_id, + ) + _attach_request_id(exc, request_id) + return exc + + def _route_for_metrics(path: str) -> str: clean_path = path.split("?", 1)[0] parts = [part for part in clean_path.strip("/").split("/") if part] @@ -1346,9 +1393,17 @@ async def describe_run(self, run_id: str | None = None) -> WorkflowRun: raise ValueError("run_id is required to describe a workflow run from a handle") return await self._client.describe_workflow_run(self.workflow_id, selected_run_id) - async def signal(self, signal_name: str, args: list[Any] | None = None) -> None: + async def signal( + self, + signal_name: str, + args: list[Any] | None = None, + *, + request_id: str | None = None, + ) -> dict[str, Any]: """Deliver an external signal to this workflow. See :meth:`Client.signal_workflow`.""" - await self._client.signal_workflow(self.workflow_id, signal_name, args=args) + return await self._client.signal_workflow( + self.workflow_id, signal_name, args=args, request_id=request_id, + ) async def append_message( self, @@ -4303,15 +4358,38 @@ async def send_webhook_bridge_event( return BridgeAdapterOutcome.from_dict(data) async def signal_workflow( - self, workflow_id: str, signal_name: str, *, args: list[Any] | None = None - ) -> None: + self, + workflow_id: str, + signal_name: str, + *, + args: list[Any] | None = None, + request_id: str | None = None, + ) -> dict[str, Any]: """Deliver an external signal to a running workflow. Signals are fire-and-forget: the server records the signal in durable history and returns immediately. They do not wait for the workflow to observe the signal. See the main docs for how to declare the allowed signal names on a workflow type. + + ``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 + 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. + + Returns the Server's admission receipt. ``request_id`` is always + present (the key that was sent); a Server that retains command keys + also reports ``retained`` (``True`` when an earlier admission was + returned) and ``intent_verified``. Raises + :class:`~durable_workflow.errors.SignalFailed` with + ``reason == "request_id_conflict"`` when the key was already retained + for a signal with a different intent; the retained command is described + in its ``details``. Every exception raised after the key was chosen, + including transport errors, carries it as ``request_id``. """ + key = _command_request_id(request_id, "signal") body: dict[str, Any] = {} if args: body["input"] = self._payload_envelope( @@ -4320,7 +4398,19 @@ async def signal_workflow( workflow_id=workflow_id, signal_name=signal_name, ) - await self._request("POST", f"/workflows/{workflow_id}/signal/{signal_name}", json=body, context=workflow_id) + body["request_id"] = key + try: + data = await self._request( + "POST", f"/workflows/{workflow_id}/signal/{signal_name}", json=body, context=workflow_id, + ) + except Exception as exc: + failure = _command_request_failure(exc, command="signal", request_id=key) + if failure is exc: + raise + raise failure from exc + receipt: dict[str, Any] = data if isinstance(data, dict) else {} + receipt.setdefault("request_id", key) + return receipt async def append_message_stream( self, @@ -4496,15 +4586,28 @@ async def update_workflow( Validator rejection and validation-routing failures remain typed and do not silently fall back to post-accept handler failure. - ``request_id`` lets the caller deduplicate retries. Raises - :class:`~durable_workflow.errors.UpdateRejected` when the workflow's - validator rejects the update, or + ``request_id`` is the caller command key (1-255 opaque, case-sensitive + bytes) that makes a retry of this update safe. When omitted the client + generates ``sdk-python-update-`` once per call, so the transport + retry policy resends the same key. Resending the same key with the same + update name, arguments and run target returns the original receipt + (``retained`` is ``True`` on a Server that reports it) instead of + admitting a second update. The returned response always carries the + ``request_id`` that was sent. + + Raises :class:`~durable_workflow.errors.UpdateRejected` when the + workflow's validator rejects the update, or :class:`~durable_workflow.errors.UpdateValidationFailed` when the declared validation boundary cannot be enforced. An accepted update whose handler fails raises :class:`~durable_workflow.errors.UpdateFailed` - with the failure message, Server response and durable identities. + with the failure message, Server response and durable identities; the + same type with ``reason == "request_id_conflict"`` reports a key that + was already retained for an update with a different intent, described + in its ``details``. Every exception raised after the key was chosen, + including transport errors, carries it as ``request_id``. """ await self._require_update_wait_stage(wait_for or "accepted") + key = _command_request_id(request_id, "update") body: dict[str, Any] = {} if args: body["input"] = self._payload_envelope( @@ -4517,10 +4620,19 @@ async def update_workflow( body["wait_for"] = wait_for if wait_timeout_seconds is not None: body["wait_timeout_seconds"] = wait_timeout_seconds - body["request_id"] = request_id or f"sdk-python-update-{uuid.uuid4().hex}" - return await self._request( - "POST", f"/workflows/{workflow_id}/update/{update_name}", json=body, context=workflow_id - ) + body["request_id"] = key + try: + data = await self._request( + "POST", f"/workflows/{workflow_id}/update/{update_name}", json=body, context=workflow_id + ) + except Exception as exc: + failure = _command_request_failure(exc, command="update", request_id=key) + if failure is exc: + raise + raise failure from exc + if isinstance(data, dict): + data.setdefault("request_id", key) + return data async def get_result( self, diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 724d29e4..7b45aaf8 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -412,8 +412,45 @@ def validation_errors(self) -> dict[str, Any] | None: return None +REQUEST_ID_CONFLICT = "request_id_conflict" +"""Reason reported when a command key is reused with a different intent.""" + +_REQUEST_ID_CONFLICT_FIELDS = ( + "request_id", + "retained_command_id", + "retained_command_type", + "retained_outcome", + "conflict", +) + + +def _body_string(body: object, name: str) -> str | None: + if isinstance(body, dict): + value = body.get(name) + if isinstance(value, str) and value: + return value + return None + + +def _request_id_conflict_details(body: object) -> dict[str, Any]: + """Return the retained-command fields of a ``request_id_conflict`` response body.""" + if not isinstance(body, dict) or body.get("reason") != REQUEST_ID_CONFLICT: + return {} + return {name: body[name] for name in _REQUEST_ID_CONFLICT_FIELDS if name in body} + + class SignalFailed(DurableWorkflowError): - """A workflow signal was rejected before it could be delivered.""" + """A workflow signal was rejected before it could be delivered. + + :attr:`request_id` is the caller command key the client sent with the + signal (caller-supplied or SDK-generated), so the application can retry + the same logical signal with the same key. :attr:`reason` is the Server's + machine-readable refusal reason; ``request_id_conflict`` means the key was + already retained for a signal with a different intent, and + :attr:`details` then carries ``retained_command_id``, + ``retained_command_type``, ``retained_outcome`` and the mismatched + ``conflict`` parts. + """ def __init__( self, @@ -422,11 +459,15 @@ def __init__( reason: str | None = None, status: int | None = None, body: object | None = None, + request_id: str | None = None, + details: dict[str, Any] | None = None, ) -> None: super().__init__(message) self.reason = reason self.status = status self.body = body + self.request_id = request_id if request_id is not None else _body_string(body, "request_id") + self.details: dict[str, Any] = dict(details) if details is not None else _request_id_conflict_details(body) @property def validation_errors(self) -> dict[str, Any] | None: @@ -508,17 +549,37 @@ def __init__( class UpdateFailed(InvalidArgument): - """An accepted workflow update failed during handler execution. + """An accepted workflow update failed, or its command key was refused. ``body`` preserves the Server response, including its durable failure identity. This inherits from :class:`InvalidArgument` to preserve catches written for earlier SDK versions that mapped all HTTP 422 responses there. + + :attr:`request_id` is the caller command key the client sent with the + update (caller-supplied or SDK-generated). :attr:`reason` is the Server's + machine-readable reason when it returned one; ``request_id_conflict`` + means the key was already retained for an update with a different intent, + and :attr:`details` then carries ``retained_command_id``, + ``retained_command_type``, ``retained_outcome`` and the mismatched + ``conflict`` parts. """ - def __init__(self, message: str, *, status: int, body: dict[str, Any]) -> None: + def __init__( + self, + message: str, + *, + status: int, + body: dict[str, Any], + reason: str | None = None, + request_id: str | None = None, + details: dict[str, Any] | None = None, + ) -> None: super().__init__(message) self.status = status self.body = body + self.reason = reason if reason is not None else self._identity(body, "reason") + self.request_id = request_id if request_id is not None else self._identity(body, "request_id") + self.details: dict[str, Any] = dict(details) if details is not None else _request_id_conflict_details(body) self.workflow_id = self._identity(body, "workflow_id") self.run_id = self._identity(body, "run_id") self.update_id = self._identity(body, "update_id") @@ -909,6 +970,18 @@ def update_validation_failed(default: str) -> UpdateValidationFailed: raise UpdateRejected(message or "update rejected", body=body) if reason in update_validation_failure_reasons: raise update_validation_failed("update validation failed") + if reason == REQUEST_ID_CONFLICT: + # The retained command type names the key scope; the Server also + # labels the control-plane operation. Either selects the typed failure. + command_type = _body_string(body, "retained_command_type") or operation + if command_type == "signal": + raise signal_failed("signal request id already retained with a different intent") + if command_type == "update" and isinstance(body, dict): + raise UpdateFailed( + message or "update request id already retained with a different intent", + status=status, + body=body, + ) raise ServerError(status, body) if status == 504: diff --git a/src/durable_workflow/retry_policy.py b/src/durable_workflow/retry_policy.py index dd3616a4..aabb4a88 100644 --- a/src/durable_workflow/retry_policy.py +++ b/src/durable_workflow/retry_policy.py @@ -3,7 +3,8 @@ .. warning:: :class:`TransportRetryPolicy` covers **only client-side HTTP retries** for - transient transport errors (connection failures, timeouts, 5xx responses, + transient transport errors (connection failures, timeouts, a server or + proxy that disconnects before sending a response, 5xx responses, 429 rate-limiting). It is **not** the activity retry policy. Activity-level retry and timeout configuration lives on :class:`durable_workflow.workflow.ActivityRetryPolicy` and is passed to @@ -201,9 +202,16 @@ class TransportRetryPolicy: """ Retry policy for transient HTTP transport errors. - Retries requests that fail with transient errors (connection errors, - timeouts, 5xx server errors, 429 rate limit). Does not retry client - errors (4xx except 429). + Retries requests that fail with transient errors: ``httpx.NetworkError`` + (including ``ConnectError``), ``httpx.TimeoutException``, + ``httpx.RemoteProtocolError`` (the server or a proxy closed the connection + before sending a response), 5xx server errors and 429 rate limit. Does not + retry client errors (4xx except 429) or ``httpx.LocalProtocolError``. + + A retried request is resent byte-identical. Signals and updates carry a + durable ``request_id`` per logical call, so a retry after a lost response + is answered from the retained receipt instead of admitting a second + command. This policy runs inside :class:`~durable_workflow.Client` around HTTP requests. It does not retry workflow runs, workflow tasks, activity @@ -228,8 +236,12 @@ def should_retry(self, exc: Exception, attempt: int) -> bool: if attempt >= self.max_attempts: return False - # Retry connection errors and timeouts - if isinstance(exc, httpx.ConnectError | httpx.TimeoutException | httpx.NetworkError): + # Retry connection errors, timeouts and a peer that disconnected before + # answering (a lost response is recovered through the request's identity). + if isinstance( + exc, + httpx.ConnectError | httpx.TimeoutException | httpx.NetworkError | httpx.RemoteProtocolError, + ): return True # Retry 5xx server errors and 429 rate limit diff --git a/src/durable_workflow/sync.py b/src/durable_workflow/sync.py index 50b73343..da48428a 100644 --- a/src/durable_workflow/sync.py +++ b/src/durable_workflow/sync.py @@ -118,8 +118,17 @@ def describe_run(self, run_id: str | None = None) -> WorkflowRun: result: WorkflowRun = self._runner.run(self._handle.describe_run(run_id)) return result - def signal(self, signal_name: str, args: list[Any] | None = None) -> None: - self._runner.run(self._handle.signal(signal_name, args=args)) + def signal( + self, + signal_name: str, + args: list[Any] | None = None, + *, + request_id: str | None = None, + ) -> dict[str, Any]: + result: dict[str, Any] = self._runner.run( + self._handle.signal(signal_name, args=args, request_id=request_id) + ) + return result def query(self, query_name: str, args: list[Any] | None = None) -> Any: return self._runner.run(self._handle.query(query_name, args=args)) @@ -513,8 +522,18 @@ def describe_workflow_run(self, workflow_id: str, run_id: str) -> WorkflowRun: result: WorkflowRun = self._runner.run(self._async.describe_workflow_run(workflow_id, run_id)) return result - def signal_workflow(self, workflow_id: str, signal_name: str, *, args: list[Any] | None = None) -> None: - self._runner.run(self._async.signal_workflow(workflow_id, signal_name, args=args)) + def signal_workflow( + self, + workflow_id: str, + signal_name: str, + *, + args: list[Any] | None = None, + request_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) + ) + return result def query_workflow(self, workflow_id: str, query_name: str, *, args: list[Any] | None = None) -> Any: return self._runner.run(self._async.query_workflow(workflow_id, query_name, args=args)) diff --git a/src/durable_workflow/workflow_updates_conformance.py b/src/durable_workflow/workflow_updates_conformance.py index 5617cada..6ce139e7 100644 --- a/src/durable_workflow/workflow_updates_conformance.py +++ b/src/durable_workflow/workflow_updates_conformance.py @@ -24,7 +24,7 @@ from . import serializer, workflow from .client import Client -from .errors import InvalidArgument, UpdateRejected +from .errors import InvalidArgument, UpdateFailed, UpdateRejected from .workflow import CompleteUpdate, FailUpdate, WorkflowContext, apply_update, validate_update SCHEMA = "durable-workflow.v2.workflow-updates.python-sdk-sidecar" @@ -392,12 +392,20 @@ async def run_surface_probe() -> ProbeRecorder: _run_worker_surface(recorder) - duplicate_response = { + duplicate_receipt = { "outcome": "accepted", "update_status": "accepted", "update_id": "upd-duplicate", "request_id": "duplicate-key", + "intent_verified": True, } + # Intent admitted under "duplicate-key": the same key with the same decoded + # arguments returns this receipt as retained; a different intent is refused. + retained_duplicate_intent: list[Any] = [] + + def _decoded_input(body: Mapping[str, Any]) -> Any: + input_payload = body.get("input") + return serializer.decode_envelope(dict(input_payload)) if isinstance(input_payload, Mapping) else [] def handler(request: httpx.Request) -> httpx.Response: body = _request_body(request) @@ -426,7 +434,26 @@ def handler(request: httpx.Request) -> httpx.Response: if request.method == "POST" and path == f"/api/workflows/{WORKFLOW_ID}/update/accept": if body.get("request_id") == "duplicate-key": - return _response(request, 200, duplicate_response) + intent = _decoded_input(body) + if not retained_duplicate_intent: + retained_duplicate_intent.append(intent) + return _response(request, 202, {**duplicate_receipt, "retained": False}) + if retained_duplicate_intent[0] == intent: + return _response(request, 202, {**duplicate_receipt, "retained": True}) + return _response( + request, + 409, + { + "reason": "request_id_conflict", + "message": "request id duplicate-key is retained for a different update intent", + "request_id": "duplicate-key", + "retained_command_id": "cmd-duplicate", + "retained_command_type": "update", + "retained_outcome": "accepted", + "conflict": ["arguments"], + "control_plane": {"operation": "update"}, + }, + ) if body.get("request_id") == "terminal-key": return _response( request, @@ -567,21 +594,48 @@ def handler(request: httpx.Request) -> httpx.Response: duplicate_first = await client.update_workflow( WORKFLOW_ID, "accept", - args=[True, "duplicate-first"], + args=[True, "duplicate"], wait_for="accepted", request_id="duplicate-key", ) - duplicate_second = await client.update_workflow( + duplicate_retry = await client.update_workflow( WORKFLOW_ID, "accept", - args=[True, "duplicate-second"], + args=[True, "duplicate"], wait_for="accepted", request_id="duplicate-key", ) recorder.check( - duplicate_first.get("update_id") == duplicate_second.get("update_id") == "upd-duplicate", - "duplicate request id must produce a stable duplicate update id in the client surface", + duplicate_first.get("update_id") == duplicate_retry.get("update_id") == "upd-duplicate", + "same request id with the same intent must return the original update id in the client surface", ) + recorder.check( + duplicate_first.get("retained") is False and duplicate_retry.get("retained") is True, + "a keyed retry must return the original receipt with retained true when the field is present", + ) + recorder.check( + duplicate_first.get("request_id") == duplicate_retry.get("request_id") == "duplicate-key", + "update receipts must carry the request id that was sent", + ) + try: + await client.update_workflow( + WORKFLOW_ID, + "accept", + args=[True, "duplicate-different-intent"], + wait_for="accepted", + request_id="duplicate-key", + ) + recorder.check(False, "same request id with a different intent must raise a request_id_conflict refusal") + except UpdateFailed as exc: + recorder.check( + exc.reason == "request_id_conflict" + and exc.request_id == "duplicate-key" + and exc.details.get("retained_command_id") == "cmd-duplicate" + and exc.details.get("retained_command_type") == "update" + and exc.details.get("conflict") == ["arguments"], + "request id conflict must expose the reason, the sent key and the retained command", + ) + recorder.typed_error("duplicate_request_idempotency", exc.__class__.__name__, exc.reason or str(exc)) recorder.mark("duplicate_request_idempotency") try: @@ -638,6 +692,10 @@ def handler(request: httpx.Request) -> httpx.Response: request_id = _request_body(request_obj).get("request_id") if isinstance(request_id, str) and request_id: key = request_id + if response_payload.get("retained") is True: + key = f"{request_id}:retained" + elif response_payload.get("reason") == "request_id_conflict": + key = f"{request_id}:conflict" recorder.client_requests[key] = _summarize_request(request_obj, response_payload) finally: await client.aclose() diff --git a/tests/test_client.py b/tests/test_client.py index bceeb32e..59e94000 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -2,7 +2,9 @@ import json import logging +from collections.abc import Callable from pathlib import Path +from typing import Any from unittest.mock import AsyncMock, patch from urllib.parse import parse_qs, urlsplit @@ -35,6 +37,7 @@ WorkflowAlreadyStarted, WorkflowNotFound, ) +from durable_workflow.retry_policy import TransportRetryPolicy SYNTHETIC_EXTERNAL_CAPABILITY_MANIFEST = { "local_activities": { @@ -870,12 +873,13 @@ class TestSignalWorkflow: async def test_signal(self, client: Client) -> None: resp = _mock_response(200, {"ok": True}) with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock: - await client.signal_workflow("wf-1", "my-signal", args=["data"]) + receipt = await client.signal_workflow("wf-1", "my-signal", args=["data"]) call_args = mock.call_args assert "/signal/my-signal" in call_args[0][1] body = call_args.kwargs.get("json") or call_args[1].get("json") assert body["input"]["codec"] == "avro" assert serializer.decode(body["input"]["blob"], codec="avro") == ["data"] + assert receipt == {"ok": True, "request_id": body["request_id"]} @pytest.mark.asyncio async def test_signal_request_matches_polyglot_fixture(self, client: Client) -> None: @@ -3096,6 +3100,243 @@ async def test_heartbeat(self, client: Client) -> None: assert result["cancel_requested"] is False +class TestCommandRequestIds: + """Caller command keys on signal and update calls (shared issue #216).""" + + _CONFLICT_FIELDS = ("request_id", "retained_command_id", "retained_command_type", "retained_outcome", "conflict") + + @staticmethod + async def _transport_client( + handler: Callable[[httpx.Request], httpx.Response], *, max_attempts: int = 3, + ) -> Client: + client = Client( + "http://localhost:8080", + token="test-token", + namespace="ns1", + retry_policy=TransportRetryPolicy(max_attempts=max_attempts, initial_backoff_seconds=0, jitter=False), + ) + client._cluster_info = _supported_runtime_info() + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler)) + return client + + @staticmethod + async def _send(client: Client, command: str, **kwargs: Any) -> dict[str, Any]: + if command == "signal": + return await client.signal_workflow("wf-1", "approve", args=["x"], **kwargs) + result: dict[str, Any] = await client.update_workflow( + "wf-1", "adjust", args=["x"], wait_for="accepted", **kwargs, + ) + return result + + @pytest.mark.asyncio + async def test_signal_generates_request_id_once_per_call_and_returns_it(self, client: Client) -> None: + resp = _mock_response(202, {"outcome": "received", "command_id": "cmd-1"}) + with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock: + first = await client.signal_workflow("wf-1", "approve") + second = await client.signal_workflow("wf-1", "approve") + + keys = [call.kwargs["json"]["request_id"] for call in mock.await_args_list] + for key in keys: + assert key.startswith("sdk-python-signal-") + assert len(key) == len("sdk-python-signal-") + 32 + assert keys[0] != keys[1] + assert first == {"outcome": "received", "command_id": "cmd-1", "request_id": keys[0]} + assert second["request_id"] == keys[1] + + @pytest.mark.asyncio + async def test_explicit_request_id_is_sent_verbatim(self, client: Client) -> None: + key = "Order-42:Approve" + resp = _mock_response(202, {"outcome": "received"}) + with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock: + signal = await client.signal_workflow("wf-1", "approve", args=["x"], request_id=key) + update = await client.update_workflow("wf-1", "adjust", args=["x"], request_id=key) + + assert [call.kwargs["json"]["request_id"] for call in mock.await_args_list] == [key, key] + assert signal["request_id"] == key + assert update["request_id"] == key + + @pytest.mark.parametrize("command", ["signal", "update"]) + @pytest.mark.asyncio + async def test_transport_retry_resends_identical_body_and_request_id(self, command: str) -> None: + attempts: list[bytes] = [] + + def handler(request: httpx.Request) -> httpx.Response: + attempts.append(request.content) + if len(attempts) == 1: + raise httpx.ConnectError("connection reset", request=request) + sent = json.loads(request.content) + return httpx.Response(202, json={"outcome": "received", "request_id": sent["request_id"]}) + + client = await self._transport_client(handler) + try: + receipt = await self._send(client, command) + finally: + await client.aclose() + + assert len(attempts) == 2 + assert attempts[0] == attempts[1] + key = json.loads(attempts[0])["request_id"] + assert key.startswith(f"sdk-python-{command}-") + assert receipt["request_id"] == key + + @pytest.mark.asyncio + async def test_signal_retries_after_remote_disconnect_with_same_request_id(self) -> None: + # A proxy that closes the socket after the Server already admitted the + # signal surfaces as RemoteProtocolError; the retry carries the same key + # and the Server answers it from the retained receipt. + attempts: list[bytes] = [] + + def handler(request: httpx.Request) -> httpx.Response: + attempts.append(request.content) + if len(attempts) == 1: + raise httpx.RemoteProtocolError("Server disconnected without sending a response.", request=request) + sent = json.loads(request.content) + return httpx.Response(202, json={ + "outcome": "received", "command_id": "cmd-1", "request_id": sent["request_id"], + "retained": True, "intent_verified": True, + }) + + client = await self._transport_client(handler) + try: + receipt = await client.signal_workflow("wf-1", "approve", args=["x"]) + finally: + await client.aclose() + + assert len(attempts) == 2 + assert attempts[0] == attempts[1] + key = json.loads(attempts[0])["request_id"] + assert key.startswith("sdk-python-signal-") + assert receipt == { + "outcome": "received", "command_id": "cmd-1", "request_id": key, + "retained": True, "intent_verified": True, + } + + @pytest.mark.parametrize("command", ["signal", "update"]) + @pytest.mark.asyncio + async def test_request_id_conflict_raises_typed_failure(self, client: Client, command: str) -> None: + conflict = { + "reason": "request_id_conflict", + "message": "request id order-42 is retained for a different intent", + "request_id": "order-42", + "retained_command_id": "cmd-7", + "retained_command_type": command, + "retained_outcome": "received" if command == "signal" else "accepted", + "conflict": ["arguments"], + } + expected = SignalFailed if command == "signal" else UpdateFailed + with ( + patch.object( + client._http, "request", new_callable=AsyncMock, return_value=_mock_response(409, conflict), + ) as mock, + pytest.raises(expected) as excinfo, + ): + await self._send(client, command, request_id="order-42") + + error = excinfo.value + assert mock.await_count == 1 + assert error.reason == "request_id_conflict" + assert error.request_id == "order-42" + assert error.status == 409 + assert error.body == conflict + assert error.details == {name: conflict[name] for name in self._CONFLICT_FIELDS} + + @pytest.mark.parametrize("command", ["signal", "update"]) + @pytest.mark.asyncio + async def test_request_id_conflict_without_command_type_is_mapped_by_the_call_site( + self, client: Client, command: str, + ) -> None: + conflict = { + "reason": "request_id_conflict", + "message": "retained", + "request_id": "order-42", + "retained_command_id": "cmd-7", + } + expected = SignalFailed if command == "signal" else UpdateFailed + with ( + patch.object( + client._http, "request", new_callable=AsyncMock, return_value=_mock_response(409, conflict), + ), + pytest.raises(expected) as excinfo, + ): + await self._send(client, command, request_id="order-42") + + error = excinfo.value + assert error.reason == "request_id_conflict" + assert error.request_id == "order-42" + assert error.status == 409 + assert error.details == {"request_id": "order-42", "retained_command_id": "cmd-7"} + assert isinstance(error.__cause__, ServerError) + + @pytest.mark.parametrize("command", ["signal", "update"]) + @pytest.mark.asyncio + async def test_transport_failure_carries_request_id(self, command: str) -> None: + attempts: list[bytes] = [] + + def handler(request: httpx.Request) -> httpx.Response: + attempts.append(request.content) + raise httpx.ConnectError("connection refused", request=request) + + client = await self._transport_client(handler, max_attempts=2) + try: + with pytest.raises(httpx.ConnectError) as excinfo: + await self._send(client, command) + finally: + await client.aclose() + + assert len(attempts) == 2 + key = json.loads(attempts[0])["request_id"] + assert key.startswith(f"sdk-python-{command}-") + assert json.loads(attempts[1])["request_id"] == key + assert excinfo.value.request_id == key + + @pytest.mark.parametrize("command", ["signal", "update"]) + @pytest.mark.asyncio + async def test_retained_receipt_is_returned_unchanged(self, client: Client, command: str) -> None: + retained = { + "outcome": "received" if command == "signal" else "accepted", + "command_id": "cmd-7", + "request_id": "order-42", + "retained": True, + "intent_verified": True, + } + if command == "update": + retained.update({"update_status": "accepted", "update_id": "upd-7"}) + with patch.object( + client._http, "request", new_callable=AsyncMock, return_value=_mock_response(202, retained), + ): + receipt = await self._send(client, command, request_id="order-42") + + assert receipt == retained + + @pytest.mark.asyncio + async def test_signal_and_update_do_not_send_x_request_id_header(self) -> None: + seen: list[httpx.Headers] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.headers) + return httpx.Response(202, json={"outcome": "received"}) + + client = await self._transport_client(handler) + try: + await self._send(client, "signal", request_id="order-42") + await self._send(client, "update", request_id="order-42") + finally: + await client.aclose() + + assert len(seen) == 2 + assert all("x-request-id" not in headers for headers in seen) + + @pytest.mark.asyncio + async def test_handle_signal_forwards_request_id_and_returns_receipt(self, client: Client) -> None: + receipt = {"outcome": "received", "request_id": "order-42"} + client.signal_workflow = AsyncMock(return_value=receipt) + handle = WorkflowHandle(client, workflow_id="wf-1") + + assert await handle.signal("approve", ["x"], request_id="order-42") == receipt + client.signal_workflow.assert_awaited_once_with("wf-1", "approve", args=["x"], request_id="order-42") + + class TestUpdateWorkflow: @pytest.mark.asyncio async def test_failed_update_retries_preserve_failure_identity(self, client: Client) -> None: @@ -3199,18 +3440,20 @@ async def test_update_rejected(self, client: Client) -> None: async def test_update_no_args(self, client: Client) -> None: resp = _mock_response(200, {"outcome": "accepted"}) with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock: - await client.update_workflow("wf-1", "my-update") + result = await client.update_workflow("wf-1", "my-update") body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json") assert "input" not in body assert body["request_id"].startswith("sdk-python-update-") + assert result == {"outcome": "accepted", "request_id": body["request_id"]} @pytest.mark.asyncio async def test_update_with_request_id(self, client: Client) -> None: resp = _mock_response(200, {"outcome": "accepted"}) with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock: - await client.update_workflow("wf-1", "my-update", request_id="req-123") + result = await client.update_workflow("wf-1", "my-update", request_id="req-123") body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json") assert body["request_id"] == "req-123" + assert result["request_id"] == "req-123" @pytest.mark.asyncio async def test_update_request_matches_polyglot_fixture(self, client: Client) -> None: diff --git a/tests/test_errors.py b/tests/test_errors.py index 2cbc0801..cbcb4b8a 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -80,6 +80,79 @@ def test_409_terminal_signal_rejected(self) -> None: assert exc_info.value.reason == "run_not_active" assert exc_info.value.status == 409 + def test_409_request_id_conflict_maps_to_signal_failed(self) -> None: + body = { + "reason": "request_id_conflict", + "message": "request id order-42 is retained for a different intent", + "request_id": "order-42", + "retained_command_id": "cmd-7", + "retained_command_type": "signal", + "retained_outcome": "received", + "conflict": ["name", "arguments"], + } + with pytest.raises(SignalFailed) as exc_info: + _raise_for_status(409, body) + error = exc_info.value + assert error.reason == "request_id_conflict" + assert error.status == 409 + assert error.request_id == "order-42" + assert error.body == body + assert error.details == { + "request_id": "order-42", + "retained_command_id": "cmd-7", + "retained_command_type": "signal", + "retained_outcome": "received", + "conflict": ["name", "arguments"], + } + + def test_409_request_id_conflict_maps_to_update_failed(self) -> None: + body = { + "reason": "request_id_conflict", + "message": "request id order-42 is retained for a different intent", + "request_id": "order-42", + "retained_command_id": "cmd-7", + "retained_command_type": "update", + "retained_outcome": "accepted", + "conflict": ["arguments"], + } + with pytest.raises(UpdateFailed) as exc_info: + _raise_for_status(409, body) + error = exc_info.value + assert error.reason == "request_id_conflict" + assert error.status == 409 + assert error.request_id == "order-42" + assert error.body == body + assert error.details["retained_command_id"] == "cmd-7" + assert error.details["conflict"] == ["arguments"] + assert error.update_id is None + + def test_409_request_id_conflict_falls_back_to_control_plane_operation(self) -> None: + body = {"reason": "request_id_conflict", "message": "retained", "control_plane": {"operation": "update"}} + with pytest.raises(UpdateFailed) as exc_info: + _raise_for_status(409, body) + assert exc_info.value.reason == "request_id_conflict" + assert exc_info.value.request_id is None + assert exc_info.value.details == {} + + def test_409_request_id_conflict_without_command_type_stays_server_error(self) -> None: + with pytest.raises(ServerError) as exc_info: + _raise_for_status(409, {"reason": "request_id_conflict", "message": "retained"}) + assert exc_info.value.reason() == "request_id_conflict" + + def test_signal_and_update_failed_constructors_stay_compatible(self) -> None: + signal = SignalFailed("signal rejected") + assert (signal.reason, signal.status, signal.body) == (None, None, None) + assert (signal.request_id, signal.details) == (None, {}) + + update = UpdateFailed("handler failed", status=422, body={"update_status": "failed", "update_id": "upd-1"}) + assert (update.reason, update.request_id, update.details, update.update_id) == (None, None, {}, "upd-1") + + keyed = UpdateFailed("handler failed", status=422, body={"reason": "update_failed", "request_id": "key-1"}) + assert (keyed.reason, keyed.request_id, keyed.details) == ("update_failed", "key-1", {}) + + explicit = SignalFailed("conflict", reason="request_id_conflict", request_id="key-2", details={"conflict": []}) + assert (explicit.request_id, explicit.details) == ("key-2", {"conflict": []}) + def test_409_terminal_query_rejected(self) -> None: with pytest.raises(QueryFailed) as exc_info: _raise_for_status( diff --git a/tests/test_retry_policy.py b/tests/test_retry_policy.py index 5fe9e6e0..13096d1d 100644 --- a/tests/test_retry_policy.py +++ b/tests/test_retry_policy.py @@ -28,6 +28,18 @@ def test_should_retry_network_error(self) -> None: exc = httpx.NetworkError("network error") assert policy.should_retry(exc, attempt=0) is True + def test_should_retry_remote_disconnect_before_response(self) -> None: + policy = RetryPolicy(max_attempts=3) + exc = httpx.RemoteProtocolError("Server disconnected without sending a response.") + assert policy.should_retry(exc, attempt=0) is True + assert policy.should_retry(exc, attempt=2) is True + assert policy.should_retry(exc, attempt=3) is False # max_attempts reached + + def test_should_not_retry_local_protocol_error(self) -> None: + policy = RetryPolicy(max_attempts=3) + exc = httpx.LocalProtocolError("malformed request") + assert policy.should_retry(exc, attempt=0) is False + def test_should_retry_5xx_server_error(self) -> None: policy = RetryPolicy(max_attempts=3) response = httpx.Response(status_code=500, request=httpx.Request("GET", "http://test")) diff --git a/tests/test_sync.py b/tests/test_sync.py index 60dcebdb..34e2e35a 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -223,9 +223,31 @@ def test_describe(self) -> None: class TestSyncClientSignal: def test_signal(self) -> None: client = Client("http://localhost:8080") - resp = _mock_response(200, {"ok": True}) - with patch.object(client._async._http, "request", new_callable=AsyncMock, return_value=resp): - client.signal_workflow("wf-1", "my-signal", args=["data"]) + resp = _mock_response(202, {"outcome": "received"}) + with patch.object(client._async._http, "request", new_callable=AsyncMock, return_value=resp) as request: + receipt = client.signal_workflow("wf-1", "my-signal", args=["data"]) + key = request.call_args.kwargs["json"]["request_id"] + assert key.startswith("sdk-python-signal-") + assert receipt == {"outcome": "received", "request_id": key} + + def test_signal_passes_request_id_through(self) -> None: + client = Client("http://localhost:8080") + retained = {"outcome": "received", "request_id": "sync-key", "retained": True} + with patch.object( + client._async._http, "request", new_callable=AsyncMock, return_value=_mock_response(202, retained), + ) as request: + receipt = client.signal_workflow("wf-1", "my-signal", request_id="sync-key") + assert request.call_args.kwargs["json"]["request_id"] == "sync-key" + assert receipt == retained + + def test_handle_signal_passes_request_id_through(self) -> None: + async_handle = WorkflowHandle(AsyncMock(), workflow_id="wf-1", run_id="r1", workflow_type="greeter") + receipt = {"outcome": "received", "request_id": "sync-key"} + async_handle.signal = AsyncMock(return_value=receipt) # type: ignore[method-assign] + with Client("http://localhost:8080") as client: + handle = SyncWorkflowHandle(async_handle, client._runner) + assert handle.signal("approve", ["x"], request_id="sync-key") == receipt + async_handle.signal.assert_awaited_once_with("approve", args=["x"], request_id="sync-key") class TestSyncClientCancel: diff --git a/tests/test_workflow_updates_conformance.py b/tests/test_workflow_updates_conformance.py index 371f0350..2392f77b 100644 --- a/tests/test_workflow_updates_conformance.py +++ b/tests/test_workflow_updates_conformance.py @@ -38,6 +38,15 @@ def test_workflow_updates_surface_probe_records_required_python_cells() -> None: assert outputs["python_worker_update_handler"]["observations"]["failed"]["command_type"] == "fail_update" assert "info" in outputs["python_client_update_request"]["requests"] assert "accepted-key" in outputs["python_client_update_request"]["requests"] - assert "duplicate-key" in outputs["python_client_update_request"]["requests"] + requests = outputs["python_client_update_request"]["requests"] + assert requests["duplicate-key"]["response"]["retained"] is False + assert requests["duplicate-key:retained"]["response"]["retained"] is True + first_receipt = requests["duplicate-key"]["response"] + assert requests["duplicate-key:retained"]["response"]["update_id"] == first_receipt["update_id"] + assert requests["duplicate-key:conflict"]["response"]["reason"] == "request_id_conflict" + assert any( + error["cell"] == "duplicate_request_idempotency" and error["refusal_reason"] == "request_id_conflict" + for error in outputs["typed_errors"] + ) assert any(error["cell"] == "unknown_update_refusal" for error in outputs["typed_errors"]) assert any(error["cell"] == "terminal_workflow_update_behavior" for error in outputs["typed_errors"])