Use event codec for replayed activity results

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -351,7 +351,6 @@ async def start_workflow(
         memo: dict[str, Any] | None = None,
         search_attributes: dict[str, Any] | None = None,
     ) -> WorkflowHandle:
-        encoded_input = serializer.encode(input if input is not None else [])
         body: dict[str, Any] = {
             "workflow_id": workflow_id,
             "workflow_type": workflow_type,

src/durable_workflow/workflow.py

@@ -273,6 +273,11 @@ class ReplayOutcome:
     commands: list[Command]
 
 
+def _decode_history_result(payload: dict[str, Any], fallback_codec: str | None) -> Any:
+    codec = payload.get("payload_codec") or fallback_codec
+    return serializer.decode_envelope(payload.get("result"), codec=codec)
+
+
 def replay(
     workflow_cls: type,
     history_events: Iterable[dict[str, Any]],
@@ -301,14 +306,14 @@ def replay(
         etype = ev.get("event_type")
         payload = ev.get("payload") or {}
         if etype in ("ActivityCompleted", "activity_completed"):
-            resolved_results.append(serializer.decode(payload.get("result"), codec=payload_codec))
+            resolved_results.append(_decode_history_result(payload, payload_codec))
         elif etype in ("TimerFired", "timer_fired"):
             resolved_results.append(None)
         elif etype in (
             "SideEffectRecorded", "side_effect_recorded",
             "ChildRunCompleted", "child_run_completed",
         ):
-            resolved_results.append(serializer.decode(payload.get("result"), codec=payload_codec))
+            resolved_results.append(_decode_history_result(payload, payload_codec))
         elif etype in ("ChildRunFailed", "child_run_failed"):
             resolved_results.append(ChildWorkflowFailed(
                 payload.get("message", "child workflow failed")

tests/integration/test_polyglot.py

@@ -169,20 +169,22 @@ async def test_python_workflow_calls_php_activity(server_url: str, server_token:
         server_cmd2 = cmd2.to_server_command(task_queue)
 
         # Debug: inspect command before asserting
-        print(f"\n=== Replay outcome ===")
+        print("\n=== Replay outcome ===")
         print(f"Command type: {server_cmd2.get('type')}")
         print(f"Full command: {server_cmd2}")
         if server_cmd2.get("type") == "fail_workflow":
             print(f"Failure message: {server_cmd2.get('message', 'N/A')}")
             print(f"Failure details: {server_cmd2.get('details', 'N/A')}")
             import json
             print(f"History events count: {len(history2)}")
-            print(f"Last few history events:")
+            print("Last few history events:")
             for evt in history2[-3:]:
                 print(f"  - {evt.get('event_type')}: {json.dumps(evt, indent=2)[:200]}")
 
-        assert server_cmd2["type"] == "complete_workflow", \
-            f"Expected complete_workflow but got {server_cmd2['type']}: {server_cmd2.get('message', 'no error message')}"
+        assert server_cmd2["type"] == "complete_workflow", (
+            f"Expected complete_workflow but got {server_cmd2['type']}: "
+            f"{server_cmd2.get('message', 'no error message')}"
+        )
 
         # Verify workflow result includes PHP activity output
         workflow_result = server_cmd2.get("result", {})

tests/test_client.py

@@ -6,6 +6,7 @@
 import httpx
 import pytest
 
+from durable_workflow import serializer
 from durable_workflow.client import CONTROL_PLANE_VERSION, PROTOCOL_VERSION, Client, WorkflowExecution, WorkflowHandle
 from durable_workflow.errors import (
     InvalidArgument,
@@ -16,7 +17,6 @@
     WorkflowAlreadyStarted,
     WorkflowNotFound,
 )
-from durable_workflow import serializer
 
 
 def _mock_response(status: int = 200, json_data: dict | None = None, text: str = "") -> httpx.Response:
@@ -76,7 +76,6 @@ async def test_success(self, client: Client) -> None:
             body = call_args.kwargs.get("json") or call_args[1].get("json")
             assert body["workflow_type"] == "greeter"
             assert body["input"]["codec"] == "avro"
-            import json as _json
             assert serializer.decode(body["input"]["blob"], codec="avro") == ["hello"]
 
     @pytest.mark.asyncio

tests/test_replay.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from durable_workflow import workflow
+from durable_workflow import serializer, workflow
 from durable_workflow.errors import ChildWorkflowFailed
 from durable_workflow.workflow import (
     CompleteWorkflow,
@@ -15,7 +15,6 @@
     WorkflowContext,
     replay,
 )
-from durable_workflow import serializer
 
 
 @workflow.defn(name="simple-return")
@@ -80,6 +79,22 @@ def test_completed_activity_triggers_completion(self) -> None:
         assert isinstance(cmd, CompleteWorkflow)
         assert cmd.result == {"greeting": "hello, world"}
 
+    def test_completed_activity_uses_event_payload_codec(self) -> None:
+        history = [
+            {
+                "event_type": "ActivityCompleted",
+                "payload": {
+                    "result": serializer.encode("hello, avro", codec="avro"),
+                    "payload_codec": "avro",
+                },
+            },
+        ]
+        outcome = replay(OneActivity, history, ["world"])
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, CompleteWorkflow)
+        assert cmd.result == {"greeting": "hello, avro"}
+
     def test_server_command_shape(self) -> None:
         outcome = replay(OneActivity, [], ["world"])
         cmd = outcome.commands[0]

tests/test_schedules.py

@@ -6,6 +6,7 @@
 import httpx
 import pytest
 
+from durable_workflow import serializer
 from durable_workflow.client import (
     Client,
     ScheduleAction,
@@ -21,7 +22,6 @@
     ScheduleAlreadyExists,
     ScheduleNotFound,
 )
-from durable_workflow import serializer
 
 
 def _mock_response(status: int = 200, json_data: dict | None = None) -> httpx.Response: