Add lazy workflow list and run history iterators

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Lazy `Client.iter_workflows`, `Client.iter_history` and `WorkflowHandle.iter_history` follow opaque page tokens while retaining only the current page and bounded cursor bookkeeping. They preserve run/filter/namespace identity, follow empty continued pages, stop on early close, and surface HTTP, cancellation and invalid/cyclic cursor errors. Existing one-page methods keep their contracts. (durable-workflow/.github#247)
+
 ## [2.15.0] - 2026-11-04
 
 ### Added

docs/sdk-reference.md

@@ -957,6 +957,38 @@ activities, for example `OrderInput(order_id=UUID(data["order_id"]), ...)`.
 Adapter output is part of the durable history contract, so changing that shape
 is a workflow compatibility change.
 
+## Whole run histories and workflow lists (unreleased)
+
+`Client.get_history(...)` and `Client.list_workflows(...)` each return one
+Server page. Use the lazy iterators to visit all pages without collecting
+them in memory:
+
+```python
+async for execution in client.iter_workflows(workflow_type="invoice", page_size=100):
+    print(execution.workflow_id)
+
+async for event in client.iter_history(workflow_id, run_id, page_size=100):
+    print(event["event_type"])
+```
+
+`WorkflowHandle.iter_history(...)` offers the same traversal when the handle
+has an explicit `run_id`. List iterators yield `WorkflowExecution` values;
+history iterators yield the event dictionaries from `get_history` in Server
+order. Filters, run identity, namespace and page size stay fixed. A supplied
+`next_page_token` resumes at that opaque continuation; do not parse or alter it.
+
+The next page is fetched only after consuming the current page. Stop or close
+the iterator to avoid later requests. Empty pages with a continuation are
+followed, while a missing or empty token ends traversal. HTTP errors,
+cancellation and malformed or cyclic continuations propagate, possibly after
+earlier items have been yielded. A changing namespace is refused before the
+next request. This traverses the Server's cursor results; it does not create a
+snapshot of a concurrently changing workflow list.
+
+These helpers are unreleased source tracked in
+[the shared paging issue](https://github.com/durable-workflow/.github/issues/247).
+Published-package and actual multi-page acceptance remain separate.
+
 ## Typed activity values
 
 Typed activity arguments and activity `result_type` are available from SDK

src/durable_workflow/client.py

@@ -378,6 +378,33 @@ class _RuntimeExternalPayloadTransport:
     prepared_completion_context: bool = False
 
 
+class _PageTokens:
+    """Detect cyclic opaque continuations with constant-space checkpoints."""
+
+    def __init__(self, token: str | None) -> None:
+        self.current = token
+        self._checkpoint = token
+        self._distance = 0
+        self._power = 1
+
+    def advance(self, token: Any) -> bool:
+        if token is None or token == "":
+            self.current = None
+            return False
+        if not isinstance(token, str):
+            raise RuntimeError("pagination returned an invalid page token")
+        if token == self.current or token == self._checkpoint:
+            raise RuntimeError("pagination repeated a page token")
+        self.current = token
+        self._distance += 1
+        # Brent's cycle detector retains one checkpoint, not every old token.
+        if self._distance == self._power:
+            self._checkpoint = token
+            self._power *= 2
+            self._distance = 0
+        return True
+
+
 @dataclass
 class WorkflowExecution:
     """Current server view of one workflow execution."""
@@ -1376,6 +1403,17 @@ async def export_history(self) -> Any:
             raise ValueError("run_id is required to export workflow history from a handle")
         return await self._client.export_history(self.workflow_id, self.run_id)
 
+    async def iter_history(
+        self, *, page_size: int | None = None, next_page_token: str | None = None,
+    ) -> AsyncIterator[dict[str, Any]]:
+        """Yield this explicit run's history. See :meth:`Client.iter_history`."""
+        if self.run_id is None:
+            raise ValueError("run_id is required to iterate workflow history from a handle")
+        async for event in self._client.iter_history(
+            self.workflow_id, self.run_id, page_size=page_size, next_page_token=next_page_token,
+        ):
+            yield event
+
     async def list_runs(self) -> WorkflowRunList:
         """List all runs in this workflow execution chain. See :meth:`Client.list_workflow_runs`."""
         return await self._client.list_workflow_runs(self.workflow_id)
@@ -3871,6 +3909,39 @@ async def list_workflows(
             next_page_token=data.get("next_page_token"),
         )
 
+    async def iter_workflows(
+        self,
+        *,
+        workflow_type: str | None = None,
+        status: str | None = None,
+        query: str | None = None,
+        page_size: int | None = None,
+        next_page_token: str | None = None,
+    ) -> AsyncIterator[WorkflowExecution]:
+        """Yield workflow list results lazily, retaining no previous pages.
+
+        Filters and namespace stay fixed across pages. An empty page with a
+        continuation is followed; missing/empty continuation ends iteration.
+        Cancellation, malformed/cyclic continuations and HTTP errors propagate,
+        possibly after earlier results have been yielded. This is a traversal
+        of Server pages, not a snapshot of concurrently changing workflows.
+        """
+        tokens = _PageTokens(next_page_token)
+        namespace = self.namespace
+        while True:
+            if self.namespace != namespace:
+                raise RuntimeError("client namespace changed during pagination")
+            page = await self.list_workflows(
+                workflow_type=workflow_type, status=status, query=query, page_size=page_size,
+                next_page_token=tokens.current,
+            )
+            more = tokens.advance(page.next_page_token)
+            for execution in page.executions:
+                yield execution
+            del page
+            if not more:
+                return
+
     async def get_history(
         self,
         workflow_id: str,
@@ -3893,6 +3964,39 @@ async def get_history(
             "GET", path, context=workflow_id
         )
 
+    async def iter_history(
+        self,
+        workflow_id: str,
+        run_id: str,
+        *,
+        page_size: int | None = None,
+        next_page_token: str | None = None,
+    ) -> AsyncIterator[dict[str, Any]]:
+        """Yield run-history events in Server order, fetching one page at a time.
+
+        Workflow/run/namespace identity and page size stay fixed. Stopping the
+        iterator fetches no subsequent page. Empty continued pages are followed;
+        malformed/cyclic continuations and HTTP errors propagate rather than
+        appearing as a complete history. Already yielded events are not undone.
+        """
+        tokens = _PageTokens(next_page_token)
+        namespace = self.namespace
+        while True:
+            if self.namespace != namespace:
+                raise RuntimeError("client namespace changed during pagination")
+            page = await self.get_history(
+                workflow_id, run_id, page_size=page_size, next_page_token=tokens.current,
+            )
+            if (not isinstance(page, dict) or not isinstance(page.get("events"), list)
+                    or any(not isinstance(event, dict) for event in page["events"])):
+                raise RuntimeError("history pagination returned invalid events")
+            more = tokens.advance(page.get("next_page_token"))
+            for event in page["events"]:
+                yield event
+            del page
+            if not more:
+                return
+
     async def export_history(self, workflow_id: str, run_id: str) -> Any:
         """Export one workflow run history as a replay bundle."""
         return await self._request(

tests/test_client_iterators.py

@@ -0,0 +1,207 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator, Callable
+from typing import Any
+
+import httpx
+import pytest
+
+from durable_workflow.client import Client
+from durable_workflow.errors import ServerError
+from durable_workflow.retry_policy import TransportRetryPolicy
+
+
+def paged_client(handler: Callable[[httpx.Request], Any]) -> Client:
+    client = Client("https://runtime.example", control_token="synthetic-control", namespace="tenant-a",
+                    retry_policy=TransportRetryPolicy(max_attempts=1))
+    client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler))
+    return client
+
+
+def page(kind: str, values: list[int], token: Any = None) -> dict[str, Any]:
+    if kind == "workflows":
+        items = [{"workflow_id": str(value), "run_id": "run", "workflow_type": "invoice"} for value in values]
+        return {"workflows": items, "next_page_token": token}
+    return {"events": [{"sequence": value, "event_type": "CustomEvent"} for value in values],
+            "next_page_token": token}
+
+
+def iterate(client: Client, kind: str, *, token: str | None = None) -> AsyncIterator[Any]:
+    if kind == "workflows":
+        return client.iter_workflows(workflow_type="invoice", status="running", query="customer = 'fixed'",
+                                     page_size=2, next_page_token=token)
+    return client.iter_history("wf-fixed", "run-fixed", page_size=2, next_page_token=token)
+
+
+def value_of(kind: str, item: Any) -> int:
+    return int(item.workflow_id) if kind == "workflows" else int(item["sequence"])
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+async def test_lazy_pages_preserve_identity_filters_and_opaque_tokens(kind: str) -> None:
+    requests: list[httpx.Request] = []
+    pages = [page(kind, [1, 2], "opaque+/="), page(kind, [], "tail"), page(kind, [3])]
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json=pages[len(requests) - 1])
+
+    async with paged_client(respond) as client:
+        iterator = iterate(client, kind)
+        assert not requests
+        assert value_of(kind, await anext(iterator)) == 1
+        assert len(requests) == 1
+        assert value_of(kind, await anext(iterator)) == 2
+        assert len(requests) == 1
+        assert value_of(kind, await anext(iterator)) == 3
+        assert len(requests) == 3  # The empty continued page is followed.
+        with pytest.raises(StopAsyncIteration):
+            await anext(iterator)
+
+    assert [r.url.params.get("next_page_token") for r in requests] == [None, "opaque+/=", "tail"]
+    assert all(r.headers["X-Namespace"] == "tenant-a" for r in requests)
+    assert all(r.url.params["page_size"] == "2" for r in requests)
+    assert "opaque%2B%2F%3D" in str(requests[1].url)
+    if kind == "history":
+        assert {r.url.path for r in requests} == {"/api/workflows/wf-fixed/runs/run-fixed/history"}
+    else:
+        for request in requests:
+            assert request.url.path == "/api/workflows"
+            assert request.url.params["workflow_type"] == "invoice"
+            assert request.url.params["status"] == "running"
+            assert request.url.params["query"] == "customer = 'fixed'"
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+async def test_closing_early_does_not_fetch_the_next_page(kind: str) -> None:
+    requests: list[httpx.Request] = []
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json=page(kind, [1, 2], "next"))
+
+    async with paged_client(respond) as client:
+        iterator = iterate(client, kind)
+        assert value_of(kind, await anext(iterator)) == 1
+        await iterator.aclose()  # type: ignore[attr-defined]
+        assert len(requests) == 1
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+@pytest.mark.parametrize("final_token", [None, ""])
+async def test_empty_final_page_terminates_without_another_request(kind: str, final_token: str | None) -> None:
+    async with paged_client(lambda _: httpx.Response(200, json=page(kind, [], final_token))) as client:
+        assert [item async for item in iterate(client, kind)] == []
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+async def test_missing_continuation_is_a_final_page(kind: str) -> None:
+    body = page(kind, [9])
+    del body["next_page_token"]
+    async with paged_client(lambda _: httpx.Response(200, json=body)) as client:
+        items = [item async for item in iterate(client, kind)]
+    assert [value_of(kind, item) for item in items] == [9]
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+async def test_resume_token_is_sent_without_normalization(kind: str) -> None:
+    requests: list[httpx.Request] = []
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json=page(kind, [7]))
+
+    async with paged_client(respond) as client:
+        items = [item async for item in iterate(client, kind, token=" resume/+? ")]
+    assert [value_of(kind, item) for item in items] == [7]
+    assert requests[0].url.params["next_page_token"] == " resume/+? "
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+@pytest.mark.parametrize("tokens", [["same"], ["a", "b"], ["a", "b", "c"]])
+async def test_cyclic_continuations_fail_instead_of_looping(kind: str, tokens: list[str]) -> None:
+    calls = 0
+
+    def respond(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        token = tokens[calls % len(tokens)]
+        calls += 1
+        assert calls <= 8, "cyclic pagination did not terminate"
+        return httpx.Response(200, json=page(kind, [], token))
+
+    async with paged_client(respond) as client:
+        with pytest.raises(RuntimeError, match="repeated a page token"):
+            _ = [item async for item in iterate(client, kind)]
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+@pytest.mark.parametrize("token", [False, 0, 1, [], {}])
+async def test_malformed_continuations_are_not_silent_end_of_data(kind: str, token: Any) -> None:
+    async with paged_client(lambda _: httpx.Response(200, json=page(kind, [1], token))) as client:
+        with pytest.raises(RuntimeError, match="invalid page token"):
+            _ = [item async for item in iterate(client, kind)]
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+async def test_namespace_change_cannot_reuse_a_cursor_in_another_namespace(kind: str) -> None:
+    calls = 0
+
+    def respond(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        return httpx.Response(200, json=page(kind, [1], "next"))
+
+    async with paged_client(respond) as client:
+        iterator = iterate(client, kind)
+        await anext(iterator)
+        client.namespace = "another-tenant"
+        with pytest.raises(RuntimeError, match="namespace changed"):
+            await anext(iterator)
+        assert calls == 1
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+@pytest.mark.parametrize("cancelled", [False, True], ids=["http-error", "cancellation"])
+async def test_later_page_error_or_cancellation_propagates(kind: str, cancelled: bool) -> None:
+    calls = 0
+
+    async def respond(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return httpx.Response(200, json=page(kind, [1], "next"))
+        if cancelled:
+            raise asyncio.CancelledError()
+        return httpx.Response(503, json={"message": "temporarily unavailable"})
+
+    async with paged_client(respond) as client:
+        iterator = iterate(client, kind)
+        assert value_of(kind, await anext(iterator)) == 1
+        with pytest.raises(asyncio.CancelledError if cancelled else ServerError):
+            await anext(iterator)
+        assert calls == 2
+
+
+@pytest.mark.parametrize("events", [None, {}, [None], ["bad"]])
+async def test_history_requires_an_event_array(events: Any) -> None:
+    async with paged_client(lambda _: httpx.Response(200, json={"events": events})) as client:
+        with pytest.raises(RuntimeError, match="invalid events"):
+            _ = [event async for event in client.iter_history("workflow", "run")]
+
+
+async def test_handle_iteration_requires_an_explicit_run_and_uses_it() -> None:
+    requests: list[httpx.Request] = []
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json=page("history", [1]))
+
+    async with paged_client(respond) as client:
+        missing = client.get_workflow_handle("wf")
+        with pytest.raises(ValueError, match="run_id is required"):
+            _ = [event async for event in missing.iter_history()]
+        assert not requests
+        handle = client.get_workflow_handle("wf", run_id="original-run")
+        assert [event["sequence"] async for event in handle.iter_history()] == [1]
+    assert requests[0].url.path == "/api/workflows/wf/runs/original-run/history"