Qualify published whole-history and workflow-list pagination

.github/workflows/ci.yml

@@ -27,6 +27,10 @@ on:
         description: Run the reviewed published SDK/native timer-redrive tuple
         type: boolean
         default: false
+      published_pagination:
+        description: Read real history and filtered workflow-list pages with the published SDK
+        type: boolean
+        default: false
       activity_retry_cap:
         description: Run the reviewed installed scalar local retry-cap case
         type: boolean
@@ -45,6 +49,31 @@ concurrency:
   cancel-in-progress: ${{ github.event_name == 'pull_request' }}
 
 jobs:
+  published-pagination:
+    if: github.event_name == 'workflow_dispatch' && inputs.published_pagination
+    needs: [action-policy, lint]
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803  # v6
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1  # v6
+        with:
+          python-version: "3.12.16"
+      - name: Bootstrap patched pip trust
+        uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7  # clean-trust
+        with:
+          clients: pip,python
+      - name: Read the published pagination fixture
+        run: python scripts/ci/published-activity-acceptance.py --scenario pagination
+      - name: Retain published tuple and page evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a  # v7
+        with:
+          name: published-pagination-${{ github.sha }}
+          path: published-activity-results/
+          retention-days: 7
   activity-retry-cap:
     if: github.event_name == 'workflow_dispatch' && inputs.activity_retry_cap
     needs: [action-policy, lint]

scripts/ci/activity_acceptance_consumer.py

@@ -25,6 +25,7 @@
 CAPTURED_VERIFIER = ""
 CAPTURED_ENTRY = ""
 CASE_FILES = {
+    "pagination": {"test_published_pagination.py"},
     "activity-identity": {"test_activity_identity.py"},
     "timer-redrive": {"test_native_timer_redrive.py", "redrive_timer_clock_workflow.py"},
     "retry-cap": {"test_activity_retry_cap.py"},
@@ -251,6 +252,7 @@ def execute_case(verifier, python, packet, before, environment, admission, insta
     test_name = {
         "timer-redrive": "test_native_timer_redrive.py", "retry-cap": "test_activity_retry_cap.py",
         "activity-identity": "test_activity_identity.py",
+        "pagination": "test_published_pagination.py",
     }[packet["scenario"]]
     environment.update(
         {
@@ -262,6 +264,8 @@ def execute_case(verifier, python, packet, before, environment, admission, insta
     )
     if redrive:
         environment["DURABLE_WORKFLOW_REDRIVE_CELL"] = "1"
+    if packet["scenario"] == "pagination":
+        environment["DURABLE_WORKFLOW_PAGINATION_CELL"] = "1"
     if packet["scenario"] == "retry-cap":
         environment["DURABLE_WORKFLOW_RETRY_CAP_CELL"] = "1"
     worker_entry = entry_source("worker", test_name, packet["files"][test_name]["sha256"], installed)

scripts/ci/published-activity-acceptance.py

@@ -44,6 +44,9 @@
 SCENARIO = "activity-identity"
 ARTIFACT_SOURCE = "registry"
 SCENARIOS = {
+    "pagination": (
+        "test_published_pagination.py", "test_published_iterators_cross_real_page_boundaries",
+    ),
     "activity-identity": (
         "test_activity_identity.py", "test_retry_preserves_received_execution_id_and_changes_attempt_id",
     ),
@@ -169,7 +172,8 @@ def capture_source(report):
         "bootstrap.py": "scripts/ci/activity_acceptance_consumer.py",
         "entry.py": "scripts/ci/activity_acceptance_entry.py",
         "tuple.json": "tests/fixtures/" + (
-            {"timer-redrive": "published-timer-redrive.json", "retry-cap": "published-retry-cap.json"}.get(
+            {"timer-redrive": "published-timer-redrive.json", "retry-cap": "published-retry-cap.json",
+             "pagination": "published-pagination.json"}.get(
                 SCENARIO, "published-activity-acceptance.json"
             )
         ),

tests/fixtures/published-pagination.json

@@ -0,0 +1,11 @@
+{
+  "schema": "sdk-python.published-activity-acceptance/v1",
+  "sdk_version": "2.17.2",
+  "server_image": "ghcr.io/durable-workflow/server@sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9",
+  "server_source": "5614274d2629d03f92932a07dfee0fb10349ddd9",
+  "server_amd64_manifest": "sha256:813c9337c44be7d497efd29e1b463546519e488aa796b5ecf9dfc6f4f10d2128",
+  "server_qualification": "https://github.com/durable-workflow/server/pull/1209#issuecomment-6858753154",
+  "python_image": "docker.io/library/python@sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df",
+  "python_amd64_manifest": "sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4",
+  "python_version": "3.12.16"
+}

tests/integration/test_published_pagination.py

@@ -0,0 +1,193 @@
+"""Read real native pages through the unchanged published iterator APIs."""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.metadata
+import json
+import os
+from dataclasses import asdict
+from typing import Any
+from uuid import uuid4
+
+import pytest
+
+from durable_workflow import Client
+
+
+class ObservedClient(Client):
+    """Retain request/page evidence without replacing any Server response."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.pages: list[dict[str, Any]] = []
+
+    async def list_workflows(self, **kwargs: Any) -> Any:
+        page = await super().list_workflows(**kwargs)
+        self.pages.append({
+            "namespace": self.namespace, "request": kwargs,
+            "items": [asdict(item) for item in page.executions], "next": page.next_page_token,
+        })
+        return page
+
+    async def get_history(self, workflow_id: str, run_id: str, **kwargs: Any) -> Any:
+        page = await super().get_history(workflow_id, run_id, **kwargs)
+        assert page["workflow_id"] == workflow_id
+        assert page["run_id"] == run_id
+        self.pages.append({
+            "namespace": self.namespace,
+            "request": {"workflow_id": workflow_id, "run_id": run_id, **kwargs},
+            "items": page["events"], "next": page.get("next_page_token"),
+        })
+        return page
+
+
+def verify_pages(
+    pages: list[dict[str, Any]], expected: list[Any], request: dict[str, Any],
+    *, initial_token: str | None = None,
+) -> None:
+    """Check every request, opaque continuation, boundary and returned item."""
+    assert pages
+    token = initial_token
+    flattened = []
+    seen = set()
+    for index, page in enumerate(pages):
+        assert page["namespace"] == "default"
+        assert page["request"] == {**request, "next_page_token": token}
+        assert 0 < len(page["items"]) <= request["page_size"]
+        flattened.extend(page["items"])
+        token = page["next"]
+        if index < len(pages) - 1:
+            assert isinstance(token, str)
+            assert token
+            assert token not in seen
+            seen.add(token)
+        else:
+            assert token is None
+    assert flattened == expected
+
+
+async def check_controls(client: ObservedClient, factory: Any, expected: list[Any], page_size: int) -> dict[str, Any]:
+    client.pages.clear()
+    iterator = factory()
+    assert client.pages == []
+    await anext(iterator)
+    assert len(client.pages) == 1
+    token = client.pages[0]["next"]
+    assert isinstance(token, str)
+    assert token
+    await iterator.aclose()
+    assert len(client.pages) == 1
+
+    client.pages.clear()
+    resumed = [item async for item in factory(next_page_token=token)]
+    normalized = [asdict(item) if hasattr(item, "workflow_id") else item for item in resumed]
+    assert normalized == expected[page_size:]
+    resumed_pages = list(client.pages)
+
+    client.pages.clear()
+    iterator = factory()
+    for _ in range(page_size):
+        await anext(iterator)
+    assert len(client.pages) == 1
+    client.namespace = "must-not-be-requested"
+    try:
+        with pytest.raises(RuntimeError, match="namespace changed"):
+            await anext(iterator)
+    finally:
+        client.namespace = "default"
+        await iterator.aclose()
+    assert len(client.pages) == 1
+    return {"early_close_requests": 1, "namespace_refusal_requests": 1, "resume_pages": resumed_pages}
+
+
+async def pagination_case(server_url: str, server_token: str) -> dict[str, Any]:
+    prefix = "python-pagination-" + uuid4().hex
+    kind = "tests.pagination." + uuid4().hex
+    options = {"token": server_token, "namespace": "default"}
+    async with Client(server_url, **options) as seed:
+        targets = []
+        for index in range(5):
+            handle = await seed.start_workflow(
+                workflow_type=kind, task_queue=prefix, workflow_id=f"{prefix}-{index}", input=[],
+            )
+            assert isinstance(handle.run_id, str)
+            assert handle.run_id
+            targets.append((handle.workflow_id, handle.run_id))
+            await seed.cancel_workflow(handle.workflow_id, reason="finite pagination fixture")
+        negatives = []
+        for suffix, workflow_type, workflow_id, cancel in (
+            ("other-type", kind + ".other", prefix + "-other-type", True),
+            ("running", kind, prefix + "-running", False),
+            ("other-query", kind, "outside-" + uuid4().hex, True),
+        ):
+            handle = await seed.start_workflow(
+                workflow_type=workflow_type, task_queue=prefix, workflow_id=workflow_id, input=[],
+            )
+            if cancel:
+                await seed.cancel_workflow(handle.workflow_id, reason="pagination filter control")
+            negatives.append({"control": suffix, "workflow_id": handle.workflow_id, "run_id": handle.run_id})
+
+        filters = {"workflow_type": kind, "status": "failed", "query": prefix}
+        baseline = await seed.list_workflows(**filters, page_size=200)
+        assert baseline.next_page_token is None
+        expected_list = [asdict(item) for item in baseline.executions]
+        assert len(expected_list) == 5
+        assert {(item["workflow_id"], item["run_id"]) for item in expected_list} == set(targets)
+        assert all(item["workflow_type"] == kind and item["status"] == "cancelled" for item in expected_list)
+        workflow_id, run_id = targets[0]
+        history = await seed.get_history(workflow_id, run_id, page_size=200)
+        assert history.get("next_page_token") is None
+        expected_history = history["events"]
+        assert len(expected_history) >= 3
+        assert {"StartAccepted", "WorkflowStarted", "WorkflowCancelled"} <= {
+            event["event_type"] for event in expected_history
+        }
+        assert all(isinstance(event["timestamp"], str) for event in expected_history)
+
+        async with ObservedClient(server_url, **options) as observed:
+            actual_list = [asdict(item) async for item in observed.iter_workflows(**filters, page_size=2)]
+            list_pages = list(observed.pages)
+            list_request = {**filters, "page_size": 2}
+            verify_pages(list_pages, expected_list, list_request)
+            assert actual_list == expected_list
+            assert [len(page["items"]) for page in list_pages] == [2, 2, 1]
+            list_controls = await check_controls(
+                observed, lambda **kw: observed.iter_workflows(**filters, page_size=2, **kw), expected_list, 2,
+            )
+            verify_pages(list_controls["resume_pages"], expected_list[2:], list_request,
+                         initial_token=list_pages[0]["next"])
+
+            observed.pages.clear()
+            handle = observed.get_workflow_handle(workflow_id, run_id=run_id)
+            actual_history = [event async for event in handle.iter_history(page_size=1)]
+            history_pages = list(observed.pages)
+            history_request = {"workflow_id": workflow_id, "run_id": run_id, "page_size": 1}
+            verify_pages(history_pages, expected_history, history_request)
+            assert actual_history == expected_history
+            assert len(history_pages) == len(expected_history)
+            assert len({event["sequence"] for event in actual_history}) == len(expected_history)
+            history_controls = await check_controls(
+                observed, lambda **kw: observed.iter_history(workflow_id, run_id, page_size=1, **kw),
+                expected_history, 1,
+            )
+            verify_pages(history_controls["resume_pages"], expected_history[1:], history_request,
+                         initial_token=history_pages[0]["next"])
+
+        # The cancelled runs cannot change while the independent traversals run.
+        assert (await seed.list_workflows(**filters, page_size=200)) == baseline
+        assert (await seed.get_history(workflow_id, run_id, page_size=200)) == history
+        await seed.cancel_workflow(negatives[1]["workflow_id"], reason="pagination fixture cleanup")
+    return {
+        "namespace": "default", "targets": targets, "negative_filters": negatives,
+        "list_pages": list_pages, "history_pages": history_pages,
+        "list_controls": list_controls, "history_controls": history_controls,
+    }
+
+
+async def test_published_iterators_cross_real_page_boundaries(server_url: str, server_token: str) -> None:
+    if os.environ.get("DURABLE_WORKFLOW_PAGINATION_CELL") != "1":
+        pytest.skip("select the reviewed published pagination cell")
+    assert importlib.metadata.version("durable-workflow") == os.environ["DURABLE_WORKFLOW_EXPECT_INSTALLED_VERSION"]
+    result = await asyncio.wait_for(pagination_case(server_url, server_token), timeout=60)
+    print(json.dumps({"scope": "published pagination", **result}, sort_keys=True))

tests/test_activity_acceptance_consumer.py

@@ -144,7 +144,7 @@ def test_closed_consumer_refuses_sdk_wheel_omission_or_substitution():
         consumer.validate_setup(packet)
 
 
-@pytest.mark.parametrize("scenario", ["activity-identity", "timer-redrive", "foreign"])
+@pytest.mark.parametrize("scenario", ["activity-identity", "timer-redrive", "pagination", "foreign"])
 def test_consumer_scenario_has_closed_captured_file_membership(scenario):
     inputs = {"requirements.txt", "verify_activity_install.py", "conftest.py"}
     inputs |= consumer.CASE_FILES.get(scenario, set())