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

Author: durable-workflow-ops

CHANGELOG.md

@@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `NamespaceDescription.deleted`.
 
 ### Fixed
+- 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.
 - Python parent workflows now decode successful child workflow completions from
   the server's documented `ChildRunCompleted.output` history payload, while
   still accepting the older `result` alias. This prevents completed child

src/durable_workflow/worker.py

@@ -383,6 +383,29 @@ def _query_history_with_export_signal_arguments(
     return enriched if changed else history
 
 
+def _query_history_events(
+    history: Any,
+    history_export: Any,
+    *,
+    default_codec: str | None,
+) -> Any:
+    if isinstance(history, list):
+        events = history
+    else:
+        events = []
+
+    if isinstance(history_export, Mapping):
+        export_events = history_export.get("history_events")
+        if isinstance(export_events, list) and len(export_events) > len(events):
+            events = export_events
+
+    return _query_history_with_export_signal_arguments(
+        events,
+        history_export,
+        default_codec=default_codec,
+    )
+
+
 def _callable_fingerprint_payload(value: object) -> str:
     if isinstance(value, staticmethod | classmethod):
         value = value.__func__
@@ -1289,7 +1312,7 @@ async def _run_query_task_core(self, task: dict[str, Any], *, client: Client | N
             return "failed"
 
         result_codec = _command_payload_codec(codec)
-        history = _query_history_with_export_signal_arguments(
+        history = _query_history_events(
             task.get("history_events", []),
             task.get("history_export"),
             default_codec=codec,

tests/test_worker.py

@@ -108,6 +108,20 @@ def run(self, ctx):  # type: ignore[no-untyped-def]
         return self.status
 
 
+@workflow.defn(name="activity-query-wf")
+class ActivityQueryWorkflow:
+    def __init__(self) -> None:
+        self.activity_result: str | None = None
+
+    @workflow.query("state")
+    def state(self) -> dict[str, str | None]:
+        return {"activity_result": self.activity_result}
+
+    def run(self, ctx):  # type: ignore[no-untyped-def]
+        self.activity_result = yield ctx.schedule_activity("load", [])
+        return self.activity_result
+
+
 @workflow.defn(name="query-state-unavailable-wf")
 class QueryStateUnavailableWorkflow:
     @workflow.query("status")
@@ -1207,6 +1221,54 @@ async def test_query_task_replays_signal_arguments_from_history_export(
         )
         mock_client.fail_query_task.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_query_task_uses_export_history_when_inline_history_is_empty(
+        self, mock_client: AsyncMock
+    ) -> None:
+        worker = Worker(mock_client, task_queue="q1", workflows=[ActivityQueryWorkflow], activities=[])
+        activity_result = serializer.envelope("loaded", codec="json")
+        task = {
+            "query_task_id": "qt-export-history",
+            "query_task_attempt": 1,
+            "workflow_type": "activity-query-wf",
+            "workflow_id": "wf-export-history",
+            "run_id": "run-export-history",
+            "query_name": "state",
+            "history_events": [],
+            "history_export": {
+                "payloads": {"codec": "json"},
+                "history_events": [
+                    {
+                        "type": "ActivityCompleted",
+                        "payload": {
+                            "sequence": 1,
+                            "activity_type": "load",
+                            "payload_codec": "json",
+                            "result": activity_result,
+                        },
+                    }
+                ],
+            },
+            "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="qt-export-history",
+            lease_owner=worker.worker_id,
+            query_task_attempt=1,
+            result={"activity_result": "loaded"},
+            codec="json",
+            workflow_id="wf-export-history",
+            run_id="run-export-history",
+            query_name="state",
+        )
+        mock_client.fail_query_task.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_query_task_replays_repeated_condition_wait_signal_arguments(
         self, mock_client: AsyncMock