Nexus: add workflow-safe Python service-call caller API (#220)

Author: durable-workflow-ops

README.md

@@ -124,6 +124,36 @@ receipt = yield ctx.start_child_workflow(
 )
 ```
 
+## Nexus service calls
+
+Workflow code can call a registered Nexus service operation through
+`WorkflowContext.call_nexus_service(...)`. The worker executes the service
+operation through the service-catalog API, records the response or typed
+failure as a durable side-effect marker, and resumes replay from that marker
+on subsequent workflow tasks.
+
+```python
+from durable_workflow import NexusOperationFailed
+
+try:
+    result = yield ctx.call_nexus_service(
+        "greeter",
+        "shared",
+        "greet",
+        ["Ada"],
+        service_sdk_language="workflow-php",
+    )
+    print(result.service_call_id, result.result)
+except NexusOperationFailed as exc:
+    print(exc.service_call_id, exc.service_error_type, exc.typed_error_message)
+```
+
+The SDK assigns a deterministic idempotency key when one is not provided and
+attaches the caller workflow instance id, caller run id, `sdk-python` caller
+language, target service language, operation name, request payload,
+service-call id, response or failure surface, and optional artifact metadata
+to the recorded result.
+
 ## Workflow signals, queries, and updates
 
 Signals mutate workflow state during replay:

docs/reference/client.md

@@ -1,3 +1,9 @@
 # Client
 
+`Client.execute_nexus_operation(...)` calls a registered Nexus service
+operation through the service-catalog API and returns `NexusOperationResult`.
+Workflow authors should normally yield `ctx.call_nexus_service(...)`; the
+worker uses this client method to execute once and record the durable result
+or typed service failure.
+
 ::: durable_workflow.client

docs/reference/workflow.md

@@ -5,4 +5,9 @@ Workflow retry and timeout settings are durable command budgets. Use
 `ChildWorkflowRetryPolicy` on `ctx.start_child_workflow(...)` for child workflow
 attempts. Use `TransportRetryPolicy` only for client HTTP retries.
 
+Use `ctx.call_nexus_service(...)` from workflow code to start a Nexus service
+operation durably. The Python worker records the accepted response or typed
+service failure as a side-effect marker, then replay resumes with
+`NexusOperationResult` or raises `NexusOperationFailed` at the yield point.
+
 ::: durable_workflow.workflow

src/durable_workflow/__init__.py

@@ -60,6 +60,7 @@
     DurableWorkflowError,
     InvalidArgument,
     NamespaceNotFound,
+    NexusOperationFailed,
     NonDeterministicReplayError,
     NonRetryableError,
     QueryFailed,
@@ -76,6 +77,12 @@
     WorkflowPayloadDecodeError,
     WorkflowTerminated,
 )
+from .nexus import (
+    NEXUS_CALLER_SDK_LANGUAGE,
+    NEXUS_OPERATION_RESULT_SCHEMA,
+    NEXUS_OPERATION_RESULT_VERSION,
+    NexusOperationResult,
+)
 from .external_storage import (
     EXTERNAL_PAYLOAD_REFERENCE_SCHEMA,
     AzureBlobExternalStorage,
@@ -173,6 +180,7 @@
     ActivityRetryPolicy,
     ChildWorkflowRetryPolicy,
     ContinueAsNew,
+    NexusServiceCall,
     Replayer,
     ReplayOutcome,
     StartChildWorkflow,
@@ -197,6 +205,9 @@
     "ContinueAsNew",
     "NamespaceDescription",
     "NamespaceList",
+    "NEXUS_CALLER_SDK_LANGUAGE",
+    "NEXUS_OPERATION_RESULT_SCHEMA",
+    "NEXUS_OPERATION_RESULT_VERSION",
     "AUTH_COMPOSITION_CONTRACT_SCHEMA",
     "AUTH_COMPOSITION_CONTRACT_VERSION",
     "AUTH_COMPOSITION_REQUIRED_EFFECTIVE_CONFIG_FIELDS",
@@ -275,6 +286,9 @@
     "LocalFilesystemExternalStorage",
     "MetricsRecorder",
     "NamespaceNotFound",
+    "NexusOperationFailed",
+    "NexusOperationResult",
+    "NexusServiceCall",
     "NonDeterministicReplayError",
     "NoopMetrics",
     "QueryFailed",

src/durable_workflow/client.py

@@ -24,7 +24,7 @@
 import time
 import uuid
 import warnings
-from collections.abc import AsyncIterator, Callable
+from collections.abc import AsyncIterator, Callable, Mapping, Sequence
 from dataclasses import dataclass
 from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as _pkg_version
@@ -43,6 +43,7 @@
 )
 from .external_storage import ExternalPayloadCache, ExternalPayloadStoragePolicy, ExternalStorageDriver
 from .metrics import CLIENT_REQUEST_DURATION_SECONDS, CLIENT_REQUESTS, NOOP_METRICS, MetricsRecorder
+from .nexus import NexusOperationResult, nexus_request_payload
 from .retry_policy import TransportRetryPolicy
 
 PROTOCOL_VERSION = "1.1"
@@ -1563,6 +1564,112 @@ def get_workflow_handle(
         """
         return WorkflowHandle(self, workflow_id=workflow_id, run_id=run_id, workflow_type=workflow_type)
 
+    async def execute_nexus_operation(
+        self,
+        endpoint_name: str,
+        service_name: str,
+        operation_name: str,
+        arguments: Sequence[Any] | None = None,
+        *,
+        mode: str | None = None,
+        wait_for: str | None = None,
+        wait_timeout_seconds: int | None = None,
+        idempotency_key: str | None = None,
+        payload_codec: str | None = None,
+        caller_namespace: str | None = None,
+        caller_workflow_instance_id: str | None = None,
+        caller_workflow_run_id: str | None = None,
+        service_sdk_language: str | None = None,
+        artifact_tuple: Mapping[str, Any] | None = None,
+        published_artifact_worker_execution: bool | None = None,
+        target_workflow_instance_id: str | None = None,
+        target_workflow_run_id: str | None = None,
+        connection: str | None = None,
+        queue: str | None = None,
+        business_key: str | None = None,
+        labels: Mapping[str, Any] | None = None,
+        memo: Mapping[str, Any] | None = None,
+        search_attributes: Mapping[str, Any] | None = None,
+        duplicate_start_policy: str | None = None,
+        raise_on_failure: bool = True,
+    ) -> NexusOperationResult:
+        """Execute a Nexus service operation through the service-catalog API.
+
+        Workflow code should normally yield
+        :meth:`durable_workflow.workflow.WorkflowContext.call_nexus_service`;
+        the worker uses this client method to make the single durable
+        execute request and records the returned service-call surface into
+        workflow history.
+        """
+        request_body = nexus_request_payload(
+            arguments=list(arguments) if arguments is not None else [],
+            payload_codec=payload_codec,
+            mode=mode,
+            wait_for=wait_for,
+            wait_timeout_seconds=wait_timeout_seconds,
+            idempotency_key=idempotency_key,
+            caller_namespace=caller_namespace or self.namespace,
+            caller_workflow_instance_id=caller_workflow_instance_id,
+            caller_workflow_run_id=caller_workflow_run_id,
+            target_workflow_instance_id=target_workflow_instance_id,
+            target_workflow_run_id=target_workflow_run_id,
+            connection=connection,
+            queue=queue,
+            business_key=business_key,
+            labels=labels,
+            memo=memo,
+            search_attributes=search_attributes,
+            duplicate_start_policy=duplicate_start_policy,
+        )
+        path = (
+            f"/service-endpoints/{quote(endpoint_name, safe='')}"
+            f"/services/{quote(service_name, safe='')}"
+            f"/operations/{quote(operation_name, safe='')}/execute"
+        )
+        context = f"{endpoint_name}/{service_name}/{operation_name}"
+
+        try:
+            data = await self._request("POST", path, json=request_body, context=context)
+        except ServerError as exc:
+            if exc.status != 409 or not isinstance(exc.body, dict):
+                raise
+            result = NexusOperationResult.from_response(
+                exc.body,
+                endpoint_name=endpoint_name,
+                service_name=service_name,
+                operation_name=operation_name,
+                request_payload=request_body,
+                service_sdk_language=service_sdk_language,
+                artifact_tuple=artifact_tuple,
+                published_artifact_worker_execution=published_artifact_worker_execution,
+            )
+            if raise_on_failure:
+                raise result.to_failure() from exc
+            return result
+
+        if not isinstance(data, dict):
+            raise ServerError(
+                200,
+                {
+                    "reason": "invalid_nexus_operation_response",
+                    "message": f"expected JSON object, got {type(data).__name__}",
+                },
+            )
+
+        result = NexusOperationResult.from_response(
+            data,
+            endpoint_name=endpoint_name,
+            service_name=service_name,
+            operation_name=operation_name,
+            request_payload=request_body,
+            service_sdk_language=service_sdk_language,
+            artifact_tuple=artifact_tuple,
+            published_artifact_worker_execution=published_artifact_worker_execution,
+        )
+        if raise_on_failure and result.is_failure:
+            raise result.to_failure()
+        return result
+
     # ── Health ─────────────────────────────────────────────────────────
     async def health(self) -> dict[str, Any]:
         """Call the server's ``/health`` endpoint and return the JSON response."""

src/durable_workflow/errors.py

@@ -44,6 +44,39 @@ def reason(self) -> str | None:
         return None
 
 
+class NexusOperationFailed(DurableWorkflowError):
+    """A Nexus service operation completed with a typed service failure."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        service_call_id: str | None = None,
+        endpoint_name: str | None = None,
+        service_name: str | None = None,
+        operation_name: str | None = None,
+        status: str | None = None,
+        outcome: str | None = None,
+        reason: str | None = None,
+        service_error_type: str | None = None,
+        caller_observed_error_type: str | None = None,
+        typed_error_message: str | None = None,
+        body: object | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.service_call_id = service_call_id
+        self.endpoint_name = endpoint_name
+        self.service_name = service_name
+        self.operation_name = operation_name
+        self.status = status
+        self.outcome = outcome
+        self.reason = reason
+        self.service_error_type = service_error_type
+        self.caller_observed_error_type = caller_observed_error_type
+        self.typed_error_message = typed_error_message
+        self.body = body
+
+
 class WorkflowFailed(DurableWorkflowError):
     """A workflow finished in the ``failed`` state.