#334: Python SDK defaults to Avro codec for all new payloads

serializer.encode() and envelope() default to avro instead of json.
client.start_workflow() sends avro-encoded input envelopes.
client.complete_activity_task() defaults to avro codec.
worker unsupported-codec fallback changed from json to avro.
Tests updated to expect avro-encoded blobs in outgoing payloads.

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -356,7 +356,7 @@ async def start_workflow(
             "workflow_id": workflow_id,
             "workflow_type": workflow_type,
             "task_queue": task_queue,
-            "input": {"codec": "json", "blob": encoded_input},
+            "input": serializer.envelope(input if input is not None else []),
             "execution_timeout_seconds": execution_timeout_seconds,
             "run_timeout_seconds": run_timeout_seconds,
         }
@@ -827,7 +827,7 @@ async def complete_activity_task(
         activity_attempt_id: str,
         lease_owner: str,
         result: Any,
-        codec: str = serializer.JSON_CODEC,
+        codec: str = serializer.AVRO_CODEC,
     ) -> Any:
         body: dict[str, Any] = {
             "activity_attempt_id": activity_attempt_id,

src/durable_workflow/serializer.py

@@ -24,7 +24,7 @@
 SUPPORTED_CODECS = (JSON_CODEC, AVRO_CODEC)
 
 
-def encode(value: Any, codec: str = JSON_CODEC) -> str:
+def encode(value: Any, codec: str = AVRO_CODEC) -> str:
     """Encode a Python value as a payload blob for *codec*.
 
     Raises ``ValueError`` for unknown codecs and
@@ -40,7 +40,7 @@ def encode(value: Any, codec: str = JSON_CODEC) -> str:
     )
 
 
-def envelope(value: Any, codec: str = JSON_CODEC) -> dict[str, str]:
+def envelope(value: Any, codec: str = AVRO_CODEC) -> dict[str, str]:
     """Wrap a value in a ``{codec, blob}`` payload envelope."""
     return {"codec": codec, "blob": encode(value, codec=codec)}
 

src/durable_workflow/worker.py

@@ -230,7 +230,7 @@ async def _run_activity_task(self, task: dict[str, Any]) -> None:
         attempt_number: int = task.get("attempt_number", 1)
         raw_args = task.get("arguments")
         inbound_codec = task.get("payload_codec") or serializer.JSON_CODEC
-        result_codec = inbound_codec if inbound_codec in serializer.SUPPORTED_CODECS else serializer.JSON_CODEC
+        result_codec = inbound_codec if inbound_codec in serializer.SUPPORTED_CODECS else serializer.AVRO_CODEC
         try:
             args = serializer.decode_envelope(raw_args, codec=inbound_codec) or []
         except AvroNotInstalledError as e:

tests/test_client.py

@@ -16,6 +16,7 @@
     WorkflowAlreadyStarted,
     WorkflowNotFound,
 )
+from durable_workflow import serializer
 
 
 def _mock_response(status: int = 200, json_data: dict | None = None, text: str = "") -> httpx.Response:
@@ -74,9 +75,9 @@ async def test_success(self, client: Client) -> None:
             call_args = mock.call_args
             body = call_args.kwargs.get("json") or call_args[1].get("json")
             assert body["workflow_type"] == "greeter"
-            assert body["input"]["codec"] == "json"
+            assert body["input"]["codec"] == "avro"
             import json as _json
-            assert _json.loads(body["input"]["blob"]) == ["hello"]
+            assert serializer.decode(body["input"]["blob"], codec="avro") == ["hello"]
 
     @pytest.mark.asyncio
     async def test_duplicate_raises(self, client: Client) -> None:
@@ -142,8 +143,8 @@ async def test_signal(self, client: Client) -> None:
             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"] == "json"
-            assert json.loads(body["input"]["blob"]) == ["data"]
+            assert body["input"]["codec"] == "avro"
+            assert serializer.decode(body["input"]["blob"], codec="avro") == ["data"]
 
 
 class TestCancelWorkflow:
@@ -303,8 +304,8 @@ async def test_update(self, client: Client) -> None:
             call_args = mock.call_args
             assert "/update/my-update" in call_args[0][1]
             body = call_args.kwargs.get("json") or call_args[1].get("json")
-            assert body["input"]["codec"] == "json"
-            assert json.loads(body["input"]["blob"]) == ["data"]
+            assert body["input"]["codec"] == "avro"
+            assert serializer.decode(body["input"]["blob"], codec="avro") == ["data"]
             assert body["wait_for"] == "completed"
             assert body["wait_timeout_seconds"] == 10
 

tests/test_replay.py

@@ -15,6 +15,7 @@
     WorkflowContext,
     replay,
 )
+from durable_workflow import serializer
 
 
 @workflow.defn(name="simple-return")
@@ -169,8 +170,8 @@ def test_server_command(self) -> None:
         cmd = CompleteWorkflow(result={"key": "val"})
         server_cmd = cmd.to_server_command("q")
         assert server_cmd["type"] == "complete_workflow"
-        assert server_cmd["result"]["codec"] == "json"
-        assert '"key"' in server_cmd["result"]["blob"]
+        assert server_cmd["result"]["codec"] == "avro"
+        assert serializer.decode(server_cmd["result"]["blob"], codec="avro") == {"key": "val"}
 
 
 @workflow.defn(name="continue-as-new-wf")
@@ -273,8 +274,8 @@ def test_server_command_shape(self) -> None:
         cmd = RecordSideEffect(result={"key": "val"})
         sc = cmd.to_server_command("q")
         assert sc["type"] == "record_side_effect"
-        assert sc["result"]["codec"] == "json"
-        assert '"key"' in sc["result"]["blob"]
+        assert sc["result"]["codec"] == "avro"
+        assert serializer.decode(sc["result"]["blob"], codec="avro") == {"key": "val"}
 
 
 class TestWorkflowContext:

tests/test_schedules.py

@@ -21,6 +21,7 @@
     ScheduleAlreadyExists,
     ScheduleNotFound,
 )
+from durable_workflow import serializer
 
 
 def _mock_response(status: int = 200, json_data: dict | None = None) -> httpx.Response:
@@ -71,7 +72,7 @@ def test_full(self) -> None:
         d = action.to_dict()
         assert d["workflow_type"] == "greeter"
         assert d["task_queue"] == "q1"
-        assert d["input"] == {"codec": "json", "blob": '["hello"]'}
+        assert d["input"]["codec"] == "avro"
         assert d["execution_timeout_seconds"] == 3600
         assert d["run_timeout_seconds"] == 600
 
@@ -108,8 +109,8 @@ async def test_action_input_uses_codec_envelope(self, client: Client) -> None:
             )
             body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
             action_input = body["action"]["input"]
-            assert action_input["codec"] == "json"
-            assert action_input["blob"] == '["Alice",42]'
+            assert action_input["codec"] == "avro"
+            assert serializer.decode(action_input["blob"], codec="avro") == ["Alice", 42]
 
     @pytest.mark.asyncio
     async def test_minimal(self, client: Client) -> None:

tests/test_serializer.py

@@ -17,18 +17,18 @@
 
 class TestEncode:
     def test_list(self) -> None:
-        assert serializer.encode(["a", 1, True]) == '["a",1,true]'
+        assert serializer.encode(["a", 1, True], codec="json") == '["a",1,true]'
 
     def test_dict(self) -> None:
-        assert serializer.encode({"k": "v"}) == '{"k":"v"}'
+        assert serializer.encode({"k": "v"}, codec="json") == '{"k":"v"}'
 
     def test_none(self) -> None:
-        assert serializer.encode(None) == "null"
+        assert serializer.encode(None, codec="json") == "null"
 
 
 class TestDecode:
     def test_roundtrip_list(self) -> None:
-        assert serializer.decode(serializer.encode(["a", 1, True])) == ["a", 1, True]
+        assert serializer.decode(serializer.encode(["a", 1, True], codec="json")) == ["a", 1, True]
 
     def test_none_blob(self) -> None:
         assert serializer.decode(None) is None
@@ -79,12 +79,13 @@ def test_empty_string_passthrough(self) -> None:
 class TestEnvelope:
     def test_structure(self) -> None:
         env = serializer.envelope(["a", 1])
-        assert env["codec"] == "json"
-        assert env["blob"] == '["a",1]'
+        assert env["codec"] == "avro"
+        assert env["blob"] == serializer.encode(["a", 1], codec="avro")
 
     def test_none_value(self) -> None:
         env = serializer.envelope(None)
-        assert env == {"codec": "json", "blob": "null"}
+        assert env["codec"] == "avro"
+        assert serializer.decode(env["blob"], codec="avro") is None
 
 
 # Fixture blobs were generated by the PHP workflow package's