Prepare frozen 2.15 source successor with the accepted batch fix

src/durable_workflow/workflow.py

@@ -1856,7 +1856,7 @@ def commands_to_server_commands(
             envelope_jobs.append((
                 len(server_commands) - 1,
                 "result",
-                command.result,
+                to_payload_value(command.result),
                 _payload_warning_context(
                     warning_context,
                     kind="workflow_result",
@@ -1870,7 +1870,7 @@ def commands_to_server_commands(
             envelope_jobs.append((
                 len(server_commands) - 1,
                 "result",
-                command.result,
+                to_payload_value(command.result),
                 _payload_warning_context(
                     warning_context,
                     kind="update_result",
@@ -1888,7 +1888,7 @@ def commands_to_server_commands(
             envelope_jobs.append((
                 len(server_commands) - 1,
                 "arguments",
-                command.arguments,
+                to_payload_values(list(command.arguments)),
                 _payload_warning_context(
                     warning_context,
                     kind="continue_as_new_input",

tests/fixtures/replay_regressions/typed-terminal-result-batched-encoding.json

@@ -0,0 +1,55 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "typed-terminal-result-batched-encoding",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.typed-terminal-result",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-05T00:00:00+00:00",
+      "payload": {},
+      "sequence": 1
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "recorded_at": "2026-11-05T00:00:01+00:00",
+      "payload": {
+        "sequence": 1,
+        "activity_type": "tests.typed.echo"
+      },
+      "sequence": 2
+    },
+    {
+      "event_type": "ActivityCompleted",
+      "recorded_at": "2026-11-05T00:00:02+00:00",
+      "payload": {
+        "sequence": 1,
+        "activity_type": "tests.typed.echo",
+        "result": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNw4GCG5hbWUKBkFkYQh0aWVyCghnb2xkCHRhZ3MMAgoGdmlwAAA="
+        },
+        "payload_codec": "avro"
+      },
+      "sequence": 3
+    }
+  ],
+  "expected": {
+    "type": "complete_workflow",
+    "result": {
+      "name": "Ada",
+      "tier": "gold",
+      "tags": [
+        "vip"
+      ]
+    }
+  }
+}

tests/test_replay_regression_corpus.py

@@ -58,7 +58,7 @@
     RedriveVisibilityWorkflow,
 )
 from tests.test_typed_values import TypedActivityResultsWorkflow
-from tests.test_typed_workflow_values import TypedOnboarding
+from tests.test_typed_workflow_values import TypedOnboarding, TypedTerminalResult
 from tests.test_update_signal_condition_replay import (
     PostConditionReceiversWorkflow,
     UpdateSignalConditionTimerWorkflow,
@@ -426,6 +426,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None:
     OrphanedDuplicateDecision,
     TypedActivityResultsWorkflow,
     TypedOnboarding,
+    TypedTerminalResult,
     FingerprintApproval,
     FingerprintGenexpr,
     FingerprintTerminalDuplicate,

tests/test_typed_workflow_values.py

@@ -12,6 +12,7 @@
 import pytest
 
 from durable_workflow import TypedValueError, serializer, workflow
+from durable_workflow.external_storage import LocalFilesystemExternalStorage
 from durable_workflow.workflow import (
     CompleteUpdate,
     CompleteWorkflow,
@@ -20,6 +21,7 @@
     StartChildWorkflow,
     WorkflowContext,
     apply_update,
+    commands_to_server_commands,
     query_state,
     replay,
 )
@@ -49,6 +51,12 @@ class Approval:
 ADA_NEUTRAL = {"name": "Ada", "tier": "gold", "tags": ["vip"]}
 
 
+@workflow.defn(name="tests.replay.typed-terminal-result")
+class TypedTerminalResult:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        return (yield ctx.schedule_activity("tests.typed.echo", [], result_type=Customer))
+
+
 @workflow.defn(name="tests.replay.typed-workflow-values", typed=True)
 class TypedOnboarding:
     """Typed run, signal, update and query arguments, and a typed child result."""
@@ -168,6 +176,39 @@ def test_child_arguments_and_terminal_values_travel_as_neutral_bytes() -> None:
     assert CompleteWorkflow(ADA).result is ADA
 
 
+@pytest.mark.parametrize("external", [False, True])
+@pytest.mark.parametrize(("command", "key", "neutral"), [
+    (CompleteWorkflow(ADA), "result", ADA_NEUTRAL),
+    (CompleteUpdate("typed-update", {"customer": ADA, "tier": Tier.GOLD}),
+     "result", {"customer": ADA_NEUTRAL, "tier": "gold"}),
+    (ContinueAsNew(arguments=[ADA, Tier.GOLD]), "arguments", [ADA_NEUTRAL, "gold"]),
+    (CompleteWorkflow(ADA_NEUTRAL), "result", ADA_NEUTRAL),
+    (CompleteUpdate("plain-update", None), "result", None),
+    (ContinueAsNew(arguments=[ADA_NEUTRAL, "gold"]), "arguments", [ADA_NEUTRAL, "gold"]),
+])
+def test_worker_batch_preserves_typed_terminal_wire_contract(
+    tmp_path: Path, external: bool, command: Any, key: str, neutral: Any,
+) -> None:
+    storage = LocalFilesystemExternalStorage(tmp_path) if external else None
+    options = {"size_warning": None, "external_storage": storage,
+               "external_storage_threshold_bytes": 1 if external else None}
+    direct = command.to_server_command("q", **options)
+    authored = command.arguments if isinstance(command, ContinueAsNew) else command.result
+    before = CompleteWorkflow({"plain": [True, 0, b"bytes"]})
+    after = CompleteUpdate("tail", [None, 1.5])
+    batch = commands_to_server_commands([before, command, after], "q", **options)
+    assert len(batch) == 3
+    assert serializer.decode_envelope(batch[0]["result"], external_storage=storage) == before.result
+    assert serializer.decode_envelope(batch[2]["result"], external_storage=storage) == after.result
+    assert serializer.decode_envelope(batch[1][key], external_storage=storage) == neutral
+    assert {k: v for k, v in batch[1].items() if k != key} == {k: v for k, v in direct.items() if k != key}
+    if external:
+        assert "external_storage" in batch[1][key]
+    else:
+        assert batch[1] == direct
+    assert (command.arguments if isinstance(command, ContinueAsNew) else command.result) is authored
+
+
 def test_a_child_result_type_mismatch_is_raised_at_the_yield() -> None:
     @workflow.defn(name="tests.typed.child-mismatch")
     class ChildMismatch: