diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index d20b751..b3b511d 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -2736,8 +2736,21 @@ async def execute_nexus_operation( ) context = f"{endpoint_name}/{service_name}/{operation_name}" + wire_body = request_body + transport = await self._runtime_external_payload_transport() + if transport is not None: + arguments_envelope = await self._externalize_runtime_payloads( + {"codec": request_body["payload_codec"], "blob": request_body["arguments"]}, + worker=False, + transport=transport, + uploaded={}, + slot=("arguments",), + ) + if "external_payload" in arguments_envelope: + wire_body = {**request_body, "arguments": arguments_envelope} + try: - data = await self._request("POST", path, json=request_body, context=context) + data = await self._request("POST", path, json=wire_body, context=context) except ServerError as exc: if exc.status != 409 or not isinstance(exc.body, dict): raise diff --git a/src/durable_workflow/nexus.py b/src/durable_workflow/nexus.py index 941e54d..76817b7 100644 --- a/src/durable_workflow/nexus.py +++ b/src/durable_workflow/nexus.py @@ -8,6 +8,7 @@ from dataclasses import dataclass, field from typing import Any +from . import serializer from .errors import NexusOperationFailed NEXUS_OPERATION_RESULT_SCHEMA = "durable-workflow.v2.sdk-python.nexus-operation-result" @@ -87,9 +88,10 @@ def nexus_request_payload( duplicate_start_policy: str | None = None, ) -> dict[str, Any]: """Build the server execute-operation request body.""" + codec = payload_codec or serializer.AVRO_CODEC return _clean_dict({ - "arguments": list(arguments), - "payload_codec": payload_codec, + "arguments": serializer.encode(list(arguments), codec=codec), + "payload_codec": codec, "mode_override": mode, "wait_for": wait_for, "wait_timeout_seconds": wait_timeout_seconds, diff --git a/tests/fixtures/codec_regressions/nexus-world-arguments-avro.json b/tests/fixtures/codec_regressions/nexus-world-arguments-avro.json new file mode 100644 index 0000000..bf2b780 --- /dev/null +++ b/tests/fixtures/codec_regressions/nexus-world-arguments-avro.json @@ -0,0 +1,21 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.codec-regression/v1", + "id": "nexus-world-arguments-avro", + "protocol": { + "codec": "avro", + "schema": "durable_workflow.protocol.Value", + "version": "1", + "fingerprint": "e2a33dff55802237" + }, + "bindings": ["python"], + "value": { + "type": "array", + "items": [{"type": "string", "value": "world"}] + }, + "framing": { + "encoding": "avro-single-object", + "wire_base64": "wwHioz3/VYAiNwwCCgp3b3JsZAA=" + }, + "failure_policy": {"operation": "round_trip", "error": null} +} diff --git a/tests/test_client.py b/tests/test_client.py index c2e67c6..2a77c36 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -289,7 +289,8 @@ async def test_execute_nexus_operation_records_request_and_evidence(self, client "/api/service-endpoints/greeter/services/shared/operations/greet/execute", ) body = mock.call_args.kwargs["json"] - assert body["arguments"] == ["Ada"] + assert body["payload_codec"] == "avro" + assert serializer.decode(body["arguments"], codec=body["payload_codec"]) == ["Ada"] assert body["mode_override"] == "sync" assert body["wait_for"] == "completed" assert body["idempotency_key"] == "idem-1" @@ -303,7 +304,7 @@ async def test_execute_nexus_operation_records_request_and_evidence(self, client assert recorded["caller_sdk_language"] == "sdk-python" assert recorded["service_sdk_language"] == "workflow-php" assert recorded["operation_name"] == "greet" - assert recorded["request_payload"]["arguments"] == ["Ada"] + assert serializer.decode(recorded["request_payload"]["arguments"], codec="avro") == ["Ada"] assert recorded["response_or_failure_surface"]["result"] == {"greeting": "hello Ada"} assert recorded["service_call_id"] == "svc-call-1" assert recorded["artifact_tuple"]["sdk-python"] == "0.4.95" diff --git a/tests/test_nexus_wire.py b/tests/test_nexus_wire.py new file mode 100644 index 0000000..72e5260 --- /dev/null +++ b/tests/test_nexus_wire.py @@ -0,0 +1,38 @@ +import json +from pathlib import Path + +import pytest + +from durable_workflow import serializer +from durable_workflow.nexus import nexus_request_payload, stable_nexus_idempotency_key + + +def test_nexus_request_matches_the_regression_corpus_wire(): + fixture = json.loads((Path(__file__).parent / "fixtures" / "codec_regressions" + / "nexus-world-arguments-avro.json").read_text()) + request = nexus_request_payload(arguments=["world"]) + assert request["arguments"] == fixture["framing"]["wire_base64"] + assert serializer.decode(request["arguments"], codec=request["payload_codec"]) == ["world"] + + +@pytest.mark.parametrize("arguments", [[], ["world"], [None, True, 42, {"nested": [b"\x00\xff", 1.5]}]]) +def test_nexus_arguments_use_the_portable_avro_payload(arguments): + request = nexus_request_payload(arguments=arguments) + assert request["payload_codec"] == "avro" + assert isinstance(request["arguments"], str) + assert serializer.decode(request["arguments"], codec="avro") == arguments + + +def test_nexus_refuses_json_as_a_workflow_payload_codec(): + with pytest.raises(ValueError, match="unsupported_payload_codec"): + nexus_request_payload(arguments=["world"], payload_codec="json") + + +def test_encoding_does_not_change_the_existing_workflow_call_identity(): + arguments = ["world", {"value": 42}] + identity = dict(workflow_id="caller", run_id="original-run", ordinal=0, + endpoint_name="endpoint", service_name="service", operation_name="greet") + key = stable_nexus_idempotency_key(**identity, arguments=arguments) + request = nexus_request_payload(arguments=arguments, idempotency_key=key) + decoded = serializer.decode(request["arguments"], codec=request["payload_codec"]) + assert request["idempotency_key"] == stable_nexus_idempotency_key(**identity, arguments=decoded) diff --git a/tests/test_runtime_external_payload_transport.py b/tests/test_runtime_external_payload_transport.py index 9983f7e..a2f2a44 100644 --- a/tests/test_runtime_external_payload_transport.py +++ b/tests/test_runtime_external_payload_transport.py @@ -309,6 +309,52 @@ def runtime_envelope(server: FakeRuntimePayloadServer, value: Any) -> dict[str, return {"codec": "avro", "external_payload": server.seed(blob)} +@pytest.mark.asyncio +@pytest.mark.parametrize("threshold_bytes", [1, 4096]) +async def test_nexus_arguments_offload_without_changing_recorded_arguments(threshold_bytes: int) -> None: + server = FakeRuntimePayloadServer(threshold_bytes=threshold_bytes) + path = "/api/service-endpoints/greeter/services/shared/operations/greet/execute" + server.responses[("POST", path)] = {"accepted": True, "id": "call-1", "status": "started"} + client = runtime_client(server) + arguments = ["hello" * 32, {"binary": b"\x00\xff"}] + try: + result = await client.execute_nexus_operation( + "greeter", "shared", "greet", arguments, caller_namespace="caller", + idempotency_key="original-call", + ) + finally: + await client.aclose() + + body = next(body for method, route, body in server.requests if method == "POST" and route == path) + if threshold_bytes == 1: + reference = body["arguments"]["external_payload"] + assert body["arguments"]["codec"] == "avro" + assert serializer.decode(server.payloads[reference["reference_id"]].decode(), codec="avro") == arguments + assert server.upload_count == 1 + assert "external_storage" not in json.dumps(body) + else: + assert serializer.decode(body["arguments"], codec="avro") == arguments + assert server.upload_count == 0 + assert body["caller_namespace"] == "caller" + assert body["idempotency_key"] == "original-call" + assert client.namespace == "billing" + assert serializer.decode(result.request_payload["arguments"], codec="avro") == arguments + assert result.request_payload["idempotency_key"] == "original-call" + + +@pytest.mark.asyncio +async def test_nexus_oversized_arguments_fail_before_upload_or_admission() -> None: + server = FakeRuntimePayloadServer(threshold_bytes=1, max_payload_bytes=32) + client = runtime_client(server) + try: + with pytest.raises(ExternalPayloadOversized): + await client.execute_nexus_operation("greeter", "shared", "greet", ["x" * 128]) + finally: + await client.aclose() + assert server.upload_count == 0 + assert not server.requests + + @pytest.mark.asyncio async def test_client_start_and_result_use_runtime_credentials_without_provider_setup() -> None: server = FakeRuntimePayloadServer()