diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f44b1db..3b7ab1e 100644
--- a/.github/workflows/ci.yml
+++ b/.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]
diff --git a/scripts/ci/activity_acceptance_consumer.py b/scripts/ci/activity_acceptance_consumer.py
index 1b51e06..c92c714 100644
--- a/scripts/ci/activity_acceptance_consumer.py
+++ b/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)
diff --git a/scripts/ci/published-activity-acceptance.py b/scripts/ci/published-activity-acceptance.py
index 63f19df..851730b 100644
--- a/scripts/ci/published-activity-acceptance.py
+++ b/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"
             )
         ),
diff --git a/tests/fixtures/published-pagination.json b/tests/fixtures/published-pagination.json
new file mode 100644
index 0000000..47c7440
--- /dev/null
+++ b/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"
+}
diff --git a/tests/integration/test_published_pagination.py b/tests/integration/test_published_pagination.py
new file mode 100644
index 0000000..960dc8e
--- /dev/null
+++ b/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))
diff --git a/tests/test_activity_acceptance_consumer.py b/tests/test_activity_acceptance_consumer.py
index 0bcb27d..293c201 100644
--- a/tests/test_activity_acceptance_consumer.py
+++ b/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())
diff --git a/tests/test_published_pagination_evidence.py b/tests/test_published_pagination_evidence.py
new file mode 100644
index 0000000..cd5a3b5
--- /dev/null
+++ b/tests/test_published_pagination_evidence.py
@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+from copy import deepcopy
+from pathlib import Path
+
+import httpx
+import pytest
+
+PATH = Path(__file__).parent / "integration/test_published_pagination.py"
+SPEC = importlib.util.spec_from_file_location("pagination_case", PATH)
+assert SPEC is not None
+assert SPEC.loader is not None
+case = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(case)
+
+
+def evidence():
+    request = {"workflow_id": "fixed-workflow", "run_id": "fixed-run", "page_size": 1}
+    events = [{"sequence": n, "timestamp": "2026-11-14T12:34:56.123456Z"} for n in range(3)]
+    pages = [
+        {"namespace": "default", "request": {**request, "next_page_token": previous},
+         "items": [event], "next": following}
+        for event, previous, following in zip(
+            events, [None, "opaque+/=", "tail"], ["opaque+/=", "tail", None], strict=True,
+        )
+    ]
+    return request, events, pages
+
+
+def test_real_page_evidence_accepts_complete_ordered_traversal_and_resume():
+    request, events, pages = evidence()
+    case.verify_pages(pages, events, request)
+    case.verify_pages(pages[1:], events[1:], request, initial_token="opaque+/=")
+
+
+@pytest.mark.parametrize("mutation", ["namespace", "run", "cursor", "omission", "duplicate", "open-tail", "order"])
+def test_incomplete_or_substituted_history_cannot_pass_evidence(mutation):
+    request, events, pages = evidence()
+    pages = deepcopy(pages)
+    if mutation == "namespace":
+        pages[1]["namespace"] = "other"
+    elif mutation == "run":
+        pages[1]["request"]["run_id"] = "other"
+    elif mutation == "cursor":
+        pages[1]["request"]["next_page_token"] = "decoded-or-replaced"
+    elif mutation == "omission":
+        pages[1]["items"] = []
+    elif mutation == "duplicate":
+        pages[1]["items"] = pages[0]["items"]
+    elif mutation == "open-tail":
+        pages[-1]["next"] = "unread"
+    else:
+        pages[1]["items"], pages[2]["items"] = pages[2]["items"], pages[1]["items"]
+    with pytest.raises(AssertionError):
+        case.verify_pages(pages, events, request)
+
+
+@pytest.mark.parametrize("field", ["workflow_type", "status", "query", "page_size"])
+def test_filtered_list_evidence_refuses_changed_filter(field):
+    request = {"workflow_type": "fixed-type", "status": "failed", "query": "fixed-prefix", "page_size": 2}
+    pages = [{"namespace": "default", "request": {**request, "next_page_token": None}, "items": [1], "next": None}]
+    pages[0]["request"][field] = "changed"
+    with pytest.raises(AssertionError):
+        case.verify_pages(pages, [1], request)
+
+
+@pytest.mark.parametrize("kind", ["workflows", "history"])
+async def test_case_observer_preserves_public_responses_and_namespace_before_next_request(kind):
+    requests = []
+
+    def respond(request):
+        requests.append(request)
+        index = 0 if request.url.params.get("next_page_token") is None else 1
+        token = "opaque+/=" if index == 0 else None
+        if kind == "workflows":
+            value = {"workflows": [{"workflow_id": str(index), "run_id": "run", "workflow_type": "type"}],
+                     "next_page_token": token}
+        else:
+            value = {"workflow_id": "wf", "run_id": "run", "events": [{"sequence": index}], "next_page_token": token}
+        return httpx.Response(200, json=value)
+
+    client = case.ObservedClient("https://runtime.example", control_token="synthetic-control", namespace="default")
+    client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(respond))
+    async with client:
+        factory = (lambda **kw: client.iter_workflows(page_size=1, **kw)) if kind == "workflows" else (
+            lambda **kw: client.iter_history("wf", "run", page_size=1, **kw)
+        )
+        expected = [item async for item in factory()]
+        if kind == "workflows":
+            expected = [case.asdict(item) for item in expected]
+        controls = await case.check_controls(client, factory, expected, 1)
+        assert controls["resume_pages"][0]["request"]["next_page_token"] == "opaque+/="
+    assert all(request.headers["X-Namespace"] == "default" for request in requests)
+
+
+@pytest.mark.parametrize("field", ["workflow_id", "run_id"])
+async def test_history_observer_refuses_foreign_response_identity(field):
+    response = {"workflow_id": "wf", "run_id": "run", "events": []}
+    response[field] = "foreign"
+    client = case.ObservedClient("https://runtime.example", control_token="synthetic-control", namespace="default")
+    client._http = httpx.AsyncClient(
+        base_url=client.base_url, transport=httpx.MockTransport(lambda _: httpx.Response(200, json=response)),
+    )
+    async with client:
+        with pytest.raises(AssertionError):
+            await client.get_history("wf", "run", page_size=1)
+    assert client.pages == []
+
+
+async def test_complete_fixture_uses_timestamp_sequence_public_shape_without_worker(monkeypatch):
+    runs = {}
+    requests = []
+
+    def respond(request):
+        requests.append(request)
+        path = request.url.path
+        if path == "/api/cluster/info":
+            return httpx.Response(200, json={"namespace": {"name": "default", "external_payload_storage": None}})
+        if request.method == "POST" and path == "/api/workflows":
+            body = json.loads(request.content)
+            item = {key: body[key] for key in ("workflow_id", "workflow_type", "task_queue")}
+            item.update(run_id="run-" + str(len(runs)), status="pending")
+            runs[item["workflow_id"]] = item
+            return httpx.Response(202, json=item)
+        if request.method == "POST" and path.endswith("/cancel"):
+            runs[path.split("/")[3]]["status"] = "cancelled"
+            return httpx.Response(200, json={})
+        size = int(request.url.params["page_size"])
+        token = request.url.params.get("next_page_token")
+        start = 0 if token is None else {"opaque+/=1": 1, "opaque+/=2": 2, "opaque+/=4": 4}[token]
+        if path == "/api/workflows":
+            items = [item for item in reversed(list(runs.values())) if (
+                item["workflow_type"] == request.url.params["workflow_type"]
+                and item["status"] == "cancelled"
+                and request.url.params["query"] in item["workflow_id"]
+            )]
+            result = {"workflows": items[start:start + size]}
+        else:
+            parts = path.split("/")
+            assert parts[4] == "runs"
+            assert parts[6] == "history"
+            assert runs[parts[3]]["run_id"] == parts[5]
+            items = [{"sequence": index, "event_type": kind, "timestamp": "2026-11-14T12:34:56.123456Z",
+                      "principal": None, "payload": {}} for index, kind in enumerate(
+                          ["StartAccepted", "WorkflowStarted", "WorkflowCancelled"], start=1,
+                      )]
+            result = {"workflow_id": parts[3], "run_id": parts[5], "events": items[start:start + size]}
+        following = start + size
+        result["next_page_token"] = "opaque+/=" + str(following) if following < len(items) else None
+        return httpx.Response(200, json=result)
+
+    original = case.Client.__init__
+
+    def initialize(self, *args, **kwargs):
+        original(self, *args, **kwargs)
+        self._http = httpx.AsyncClient(base_url=self.base_url, transport=httpx.MockTransport(respond))
+
+    monkeypatch.setattr(case.Client, "__init__", initialize)
+    result = await case.pagination_case("https://runtime.example", "synthetic-control")
+    assert [len(page["items"]) for page in result["list_pages"]] == [2, 2, 1]
+    assert len(result["history_pages"]) == 3
+    assert len(runs) == 8
+    assert all(item["status"] == "cancelled" for item in runs.values())
+    assert all(request.headers["X-Namespace"] == "default" for request in requests)
+    assert not any("worker" in request.url.path for request in requests)