[cross-repo from server#389] Conformance blocker: complete replay parity coverage beyond smoke (#158)

Author: durable-workflow-ops

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `NamespaceDescription.deleted`.
 
 ### Fixed
+- Worker query tasks now treat null or empty compact history payload fields as
+  missing when a history export carries the durable activity result or signal
+  arguments, so cold replay after worker restart does not answer queries from
+  an unmodified workflow instance.
 - Worker query tasks now replay from the bundled history export when the
   inline task history is empty or truncated, so cold query replay reconstructs
   activity-derived state instead of answering from a fresh workflow instance.

src/durable_workflow/worker.py

@@ -123,6 +123,10 @@ def _string_or_none(value: Any) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _is_missing_payload_placeholder(value: Any) -> bool:
+    return value is None or (isinstance(value, str) and value == "")
+
+
 def _is_final_query_task_rejection(error: BaseException) -> bool:
     return (
         isinstance(error, ServerError)
@@ -225,7 +229,7 @@ def _activity_result_by_sequence_from_export(
             sequence = int(sequence)
         if not isinstance(sequence, int):
             continue
-        if raw_activity.get("result") is None:
+        if _is_missing_payload_placeholder(raw_activity.get("result")):
             continue
         results.setdefault(sequence, raw_activity)
 
@@ -311,13 +315,22 @@ def _query_history_with_export_signal_arguments(
             activity = activity_results_by_sequence.get(sequence) if isinstance(sequence, int) else None
             if activity is not None:
                 payload_changed = False
-                if "result" not in payload and activity.get("result") is not None:
+                if (
+                    _is_missing_payload_placeholder(payload.get("result"))
+                    and not _is_missing_payload_placeholder(activity.get("result"))
+                ):
                     payload["result"] = activity["result"]
                     payload_changed = True
-                if "payload_codec" not in payload and isinstance(activity.get("payload_codec"), str):
+                if not (
+                    isinstance(payload.get("payload_codec"), str)
+                    and payload.get("payload_codec")
+                ) and isinstance(activity.get("payload_codec"), str):
                     payload["payload_codec"] = activity["payload_codec"]
                     payload_changed = True
-                if "activity_type" not in payload and isinstance(activity.get("activity_type"), str):
+                if not (
+                    isinstance(payload.get("activity_type"), str)
+                    and payload.get("activity_type")
+                ) and isinstance(activity.get("activity_type"), str):
                     payload["activity_type"] = activity["activity_type"]
                     payload_changed = True
                 if payload_changed:
@@ -358,18 +371,28 @@ def _query_history_with_export_signal_arguments(
         payload = dict(raw_payload)
         payload_changed = False
         workflow_sequence = signal.get("workflow_sequence")
-        if isinstance(workflow_sequence, int):
-            payload.setdefault("workflow_sequence", workflow_sequence)
+        if isinstance(workflow_sequence, int) and not isinstance(payload.get("workflow_sequence"), int):
+            payload["workflow_sequence"] = workflow_sequence
             payload_changed = True
-        elif isinstance(workflow_sequence, str) and workflow_sequence.isdigit():
-            payload.setdefault("workflow_sequence", int(workflow_sequence))
+        elif (
+            isinstance(workflow_sequence, str)
+            and workflow_sequence.isdigit()
+            and not isinstance(payload.get("workflow_sequence"), int)
+        ):
+            payload["workflow_sequence"] = int(workflow_sequence)
             payload_changed = True
 
         envelope = _signal_arguments_envelope_from_export(signal, default_codec=signal_default_codec)
         if envelope is not None:
-            payload.setdefault("arguments", envelope)
-            payload.setdefault("payload_codec", envelope.get("codec"))
-            payload_changed = True
+            if _is_missing_payload_placeholder(payload.get("arguments")):
+                payload["arguments"] = envelope
+                payload_changed = True
+            if not (
+                isinstance(payload.get("payload_codec"), str)
+                and payload.get("payload_codec")
+            ):
+                payload["payload_codec"] = envelope.get("codec")
+                payload_changed = True
 
         if not payload_changed:
             enriched.append(raw_event)

tests/test_worker.py

@@ -1467,6 +1467,106 @@ async def test_query_task_replays_history_from_export_after_worker_restart(
         )
         mock_client.fail_query_task.assert_not_called()
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("placeholder", "query_task_id", "workflow_id", "run_id"),
+        [
+            (None, "qt-null-placeholders", "wf-null-placeholders", "run-null-placeholders"),
+            ("", "qt-empty-placeholders", "wf-empty-placeholders", "run-empty-placeholders"),
+        ],
+    )
+    async def test_query_task_replaces_missing_history_payload_placeholders_from_export(
+        self,
+        mock_client: AsyncMock,
+        placeholder: object,
+        query_task_id: str,
+        workflow_id: str,
+        run_id: str,
+    ) -> None:
+        worker = Worker(mock_client, task_queue="q1", workflows=[ReplayQuerySnapshotWorkflow], activities=[])
+        approval_arguments = serializer.encode(["alice"], codec="json")
+        task = {
+            "query_task_id": query_task_id,
+            "query_task_attempt": 1,
+            "workflow_type": "replay-query-snapshot-wf",
+            "workflow_id": workflow_id,
+            "run_id": run_id,
+            "query_name": "state",
+            "history_events": [
+                {
+                    "event_type": "ActivityCompleted",
+                    "payload": {
+                        "sequence": 1,
+                        "activity_type": "",
+                        "payload_codec": None,
+                        "result": placeholder,
+                    },
+                },
+                {
+                    "event_type": "ConditionWaitOpened",
+                    "payload": {
+                        "sequence": 2,
+                        "condition_wait_id": "wait-approval",
+                        "condition_key": "approval",
+                    },
+                },
+                {
+                    "event_type": "SignalReceived",
+                    "payload": {
+                        "signal_id": "sig-approve",
+                        "workflow_command_id": "cmd-approve",
+                        "signal_name": "approve",
+                        "workflow_sequence": None,
+                        "payload_codec": "",
+                        "arguments": placeholder,
+                    },
+                },
+            ],
+            "history_export": {
+                "payloads": {"codec": "json"},
+                "activities": [
+                    {
+                        "sequence": 1,
+                        "activity_type": "load-state",
+                        "payload_codec": "json",
+                        "result": serializer.encode("loaded", codec="json"),
+                    },
+                ],
+                "signals": [
+                    {
+                        "id": "sig-approve",
+                        "command_id": "cmd-approve",
+                        "name": "approve",
+                        "workflow_sequence": 2,
+                        "payload_codec": "json",
+                        "arguments": approval_arguments,
+                    },
+                ],
+            },
+            "workflow_arguments": serializer.envelope([], codec="json"),
+            "query_arguments": serializer.envelope([], codec="json"),
+            "payload_codec": "json",
+        }
+
+        outcome = await worker._run_query_task(task)
+
+        assert outcome == "completed"
+        mock_client.complete_query_task.assert_awaited_once_with(
+            query_task_id=query_task_id,
+            lease_owner=worker.worker_id,
+            query_task_attempt=1,
+            result={
+                "activity_result": "loaded",
+                "approved_by": "alice",
+                "finished": True,
+            },
+            codec="json",
+            workflow_id=workflow_id,
+            run_id=run_id,
+            query_name="state",
+        )
+        mock_client.fail_query_task.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_query_task_enriches_compact_activity_completion_from_export(
         self, mock_client: AsyncMock