Fix #216 (result key) and #207 (history pagination)

- Client.get_result() now reads payload.output instead of payload.result for WorkflowCompleted events, with fallback to result for backwards compatibility
- Worker now paginates history when next_history_page_token is present, preventing silent replay failures for large workflows
- Added Client.workflow_task_history() method to fetch additional history pages

Resolves: #216, #207
Tests: 204/204 unit tests pass, ruff/mypy clean

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -477,7 +477,7 @@ async def get_result(
                     etype = ev.get("event_type")
                     payload = ev.get("payload") or {}
                     if etype in ("WorkflowCompleted", "workflow_completed"):
-                        return serializer.decode(payload.get("result"))
+                        return serializer.decode(payload.get("output") or payload.get("result"))
                     if etype in ("WorkflowFailed", "workflow_failed"):
                         raise WorkflowFailed(
                             payload.get("message", "workflow failed"),
@@ -750,6 +750,23 @@ async def fail_workflow_task(
             "POST", f"/worker/workflow-tasks/{task_id}/fail", worker=True, json=body
         )
 
+    async def workflow_task_history(
+        self,
+        *,
+        task_id: str,
+        page_token: str,
+        lease_owner: str,
+        workflow_task_attempt: int,
+    ) -> Any:
+        body: dict[str, Any] = {
+            "page_token": page_token,
+            "lease_owner": lease_owner,
+            "workflow_task_attempt": workflow_task_attempt,
+        }
+        return await self._request(
+            "POST", f"/worker/workflow-tasks/{task_id}/history", worker=True, json=body
+        )
+
     async def poll_activity_task(
         self, *, worker_id: str, task_queue: str, timeout: float = 35.0
     ) -> Any:

src/durable_workflow/worker.py

@@ -69,6 +69,23 @@ async def _run_workflow_task(self, task: dict[str, Any]) -> list[dict[str, Any]]
         wf_type: str = task.get("workflow_type", "")
         history = task.get("history_events", [])
 
+        # Paginate history if needed
+        next_page_token = task.get("next_history_page_token")
+        while next_page_token:
+            try:
+                page_data = await self.client.workflow_task_history(
+                    task_id=task_id,
+                    page_token=next_page_token,
+                    lease_owner=self.worker_id,
+                    workflow_task_attempt=attempt,
+                )
+                if page_data and page_data.get("history_events"):
+                    history.extend(page_data["history_events"])
+                next_page_token = page_data.get("next_history_page_token") if page_data else None
+            except Exception as e:
+                log.warning("failed to fetch history page for task %s: %s", task_id, e)
+                break
+
         start_input: list[Any] = []
         codec = task.get("payload_codec")
         raw_args = task.get("arguments")