Encode Nexus arguments with the portable Avro codec

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

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,

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}
+}

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"

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)

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()