feat(python): yield the items of a poll, with a resume point

GUIDE.md

@@ -2303,7 +2303,7 @@ type or transform expression refers to an otherwise-unavailable module.
 ### @nexus.poll
 
 **Placement:** Operation (function) **Syntax:**
-`@nexus.poll python="<support-helper>" [target="workflow|client"]`
+`@nexus.poll python="<support-helper>" [target="workflow|client"] [items="<field>"] [python-resume="<support-helper>"]`
 
 Generates an `iter_<name>` async iterator for the operation, and for each
 resource method that binds it. The iterator sends the request, yields the
@@ -2338,6 +2338,28 @@ def next_read_request(request, response):
 The referenced helper must be available from the package specified by
 `--support-package`.
 
+`items` names a list field of the response. The iterator then yields the items,
+not the pages, and it is a `PollIterator` of the package. Its `next_request` is
+the request that resumes after the last item that it yielded, and it is `None`
+after the last page. `resume` names a helper `(page request, item) -> request`
+that resumes after one item. After a page's last item, or after a page with no
+item, `next_request` is the page helper's request. A caller saves it and resumes
+with it.
+
+```wit
+/// @nexus.poll python="next_read_request" items="records"
+///   python-resume="resume_read_request" target="client"
+read-records: func(request: read-records-request) -> read-records-response;
+```
+
+```python
+def resume_read_request(request, item):
+    return dataclasses.replace(request, from_offset=(item.offset or 0) + 1, last=None)
+```
+
+nexgen refuses `items` that names no list field of the response, and `resume`
+with no `items`.
+
 ---
 
 ### @nexus.flatten-in-api

advanced/samples/README.md

@@ -75,7 +75,7 @@ sample-project instructions for [Go](go/README.md), [Python](python/README.md),
 Operations with `@nexus.poll`, in a service with an endpoint and in a service
 with no endpoint, and a resource method that binds one. Only Python generates
 the iterators, for both targets. The service with no endpoint also takes a
-parameter of an enum type.
+parameter of an enum type, and its poll yields the executions of each page.
 
 - WIT: [`inputs/poll-service.wit`](inputs/poll-service.wit)
 - Python: [`python/wit/poll_service/`](python/wit/poll_service/),
@@ -213,7 +213,8 @@ each target, which the sample support code gives in place of Brook's runtime.
 The create takes the backend of the Stream's ref, and the ref copies the id. In
 Python, the factory `stream` builds a handle from an id or a ref with no call,
 and a handle passes as its ref. `publish` takes the values, and returns the
-offset of the last one.
+offset of the last one. The read iterator yields each Record, and its
+`next_request` resumes after the last one.
 
 - WIT: [`inputs/stream-service.wit`](inputs/stream-service.wit),
   [`inputs/stream-writer.wit`](inputs/stream-writer.wit)

advanced/samples/inputs/poll-service.wit

@@ -17,10 +17,69 @@ interface page-service {
     query: option<string>,
   }
 
+  /// @nexus.proto "temporal.api.common.v1.WorkflowExecution"
+  record workflow-execution {
+    workflow-id: string,
+    run-id: string,
+  }
+
+  /// @nexus.proto "temporal.api.workflow.v1.WorkflowExecutionInfo"
+  record workflow-execution-info {
+    execution: workflow-execution,
+    parent-execution: option<workflow-execution>,
+    /// @nexus.omit
+    %type: placeholder,
+    /// @nexus.omit
+    start-time: placeholder,
+    /// @nexus.omit
+    close-time: placeholder,
+    /// @nexus.omit
+    status: placeholder,
+    /// @nexus.omit
+    history-length: placeholder,
+    /// @nexus.omit
+    parent-namespace-id: placeholder,
+    /// @nexus.omit
+    execution-time: placeholder,
+    /// @nexus.omit
+    memo: placeholder,
+    /// @nexus.omit
+    search-attributes: placeholder,
+    /// @nexus.omit
+    auto-reset-points: placeholder,
+    /// @nexus.omit
+    task-queue: placeholder,
+    /// @nexus.omit
+    state-transition-count: placeholder,
+    /// @nexus.omit
+    history-size-bytes: placeholder,
+    /// @nexus.omit
+    most-recent-worker-version-stamp: placeholder,
+    /// @nexus.omit
+    execution-duration: placeholder,
+    /// @nexus.omit
+    root-execution: placeholder,
+    /// @nexus.omit
+    assigned-build-id: placeholder,
+    /// @nexus.omit
+    inherited-build-id: placeholder,
+    /// @nexus.omit
+    first-run-id: placeholder,
+    /// @nexus.omit
+    versioning-info: placeholder,
+    /// @nexus.omit
+    worker-deployment-name: placeholder,
+    /// @nexus.omit
+    priority: placeholder,
+    /// @nexus.omit
+    external-payload-size-bytes: placeholder,
+    /// @nexus.omit
+    external-payload-count: placeholder,
+  }
+
   /// @nexus.proto "temporal.api.workflowservice.v1.ListWorkflowExecutionsResponse"
   record list-workflows-response {
-    /// @nexus.omit
-    executions: placeholder,
+    executions: list<workflow-execution-info>,
     next-page-token: option<list<u8>>,
   }
 
@@ -58,6 +117,6 @@ interface feed-service {
 
   open-feed: func(request: open-feed-request) -> feed-result;
 
-  /// @nexus.poll python="next_page_request"
+  /// @nexus.poll python="next_page_request" items="executions"
   list-feed: func(request: list-workflows-request) -> list-workflows-response;
 }

advanced/samples/inputs/stream-service.wit

@@ -428,7 +428,8 @@ interface stream-service {
   /// @nexus.doc
   ///   "A long poll. NOT_FOUND for an uncreated Stream. Workflow code reads from a Local Activity."
   /// @nexus.operation name="ReadRecords"
-  /// @nexus.poll python="next_read_request" target="client"
+  /// @nexus.poll python="next_read_request" items="records" python-resume="resume_read_request"
+  ///   target="client"
   read-records: func(request: read-records-request) -> read-records-response;
 
   /// @nexus.operation name="DescribeStream"

advanced/samples/invalid/poll/invalid-resume-helper/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit-directive"
+contains = ["@nexus.poll", "invalid `resume` support helper `resume_read(request, item)`"]

advanced/samples/invalid/poll/invalid-resume-helper/input.wit

@@ -0,0 +1,14 @@
+package temporal:poll-invalid-resume-helper@1.0.0;
+
+world service { export reader; }
+
+/// @nexus.endpoint "reader"
+interface reader {
+  record read-request { offset: option<s64> }
+  record read-item { offset: option<s64> }
+  record read-response { items: list<read-item>, done: option<bool> }
+
+  // The resume helper names a support helper, not a call.
+  /// @nexus.poll python="next_read" items="items" python-resume="resume_read(request, item)"
+  read: func(request: read-request) -> read-response;
+}

advanced/samples/invalid/poll/items-not-a-list/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit-directive"
+contains = ["@nexus.poll", "`done` isn't a list"]

advanced/samples/invalid/poll/items-not-a-list/input.wit

@@ -0,0 +1,14 @@
+package temporal:poll-items-not-a-list@1.0.0;
+
+world service { export reader; }
+
+/// @nexus.endpoint "reader"
+interface reader {
+  record read-request { offset: option<s64> }
+  record read-item { offset: option<s64> }
+  record read-response { items: list<read-item>, done: option<bool> }
+
+  // A poll yields the items of a list.
+  /// @nexus.poll python="next_read" items="done"
+  read: func(request: read-request) -> read-response;
+}

advanced/samples/invalid/poll/items-unknown-field/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit-directive"
+contains = ["@nexus.poll", "`records` isn't a field of the response"]

advanced/samples/invalid/poll/items-unknown-field/input.wit

@@ -0,0 +1,14 @@
+package temporal:poll-items-unknown-field@1.0.0;
+
+world service { export reader; }
+
+/// @nexus.endpoint "reader"
+interface reader {
+  record read-request { offset: option<s64> }
+  record read-item { offset: option<s64> }
+  record read-response { items: list<read-item>, done: option<bool> }
+
+  // `records` isn't a field of the response.
+  /// @nexus.poll python="next_read" items="records"
+  read: func(request: read-request) -> read-response;
+}

advanced/samples/invalid/poll/resume-without-items/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit-directive"
+contains = ["@nexus.poll", "`resume` needs `items`"]

advanced/samples/invalid/poll/resume-without-items/input.wit

@@ -0,0 +1,14 @@
+package temporal:poll-resume-without-items@1.0.0;
+
+world service { export reader; }
+
+/// @nexus.endpoint "reader"
+interface reader {
+  record read-request { offset: option<s64> }
+  record read-item { offset: option<s64> }
+  record read-response { items: list<read-item>, done: option<bool> }
+
+  // A resume point is after an item, so it needs the items.
+  /// @nexus.poll python="next_read" python-resume="resume_read"
+  read: func(request: read-request) -> read-response;
+}

advanced/samples/python/temporal_support/__init__.py

@@ -5,5 +5,6 @@
 from .operations import *  # noqa: F401,F403
 from .poll import next_page_request as next_page_request
 from .poll import next_read_request as next_read_request
+from .poll import resume_read_request as resume_read_request
 from .temporal_model_converters import *  # noqa: F401,F403
 from .streams import *  # noqa: F401,F403

advanced/samples/python/temporal_support/poll.py

@@ -47,3 +47,13 @@ def next_page_request(request: _RequestT, response: _PageResponse) -> _RequestT
     if not response.next_page_token:
         return None
     return dataclasses.replace(request, next_page_token=response.next_page_token)
+
+
+class _ReadItem(typing.Protocol):
+    @property
+    def offset(self) -> int | None: ...
+
+
+def resume_read_request(request: _ReadRequestT, item: _ReadItem) -> _ReadRequestT:
+    # A read that resumes after an item starts past its offset, and reads no `last` again.
+    return dataclasses.replace(request, from_offset=(item.offset or 0) + 1, last=None)

advanced/samples/python/tests/test_client_target.py

@@ -280,9 +280,9 @@ async def test_a_read_poll_keeps_its_offset_on_an_empty_page() -> None:
     ]
     client = FakeClient({"ReadRecords": list(pages)})
 
-    read = [page async for page in client.iter_read_records(str, stream=ref, last=2)]
+    read = [item async for item in client.iter_read_records(str, stream=ref, last=2)]
 
-    assert read == pages
+    assert read == []
     assert [call.input for call in client.calls] == [
         ReadRecordsRequest(stream=ref, last=2),
         ReadRecordsRequest(stream=ref, from_offset=3),
@@ -291,23 +291,40 @@ async def test_a_read_poll_keeps_its_offset_on_an_empty_page() -> None:
     ]
 
 
-async def test_a_handle_reads_every_page() -> None:
+def stream_record(body: str, offset: int) -> StreamRecord[str]:
+    return StreamRecord(item=StreamItemData(StreamData(body=body)), offset=offset)
+
+
+async def test_a_handle_yields_items_and_resumes_after_each() -> None:
     ref = StreamRef(id="s1")
     pages = [
-        ReadRecordsResponse[str](next_offset=2),
-        ReadRecordsResponse[str](done=True),
+        ReadRecordsResponse(
+            records=[stream_record("a", 10), stream_record("b", 12)], next_offset=15
+        ),
+        ReadRecordsResponse[str](next_offset=15),
+        ReadRecordsResponse(
+            records=[stream_record("c", 20)], next_offset=21, done=True
+        ),
     ]
     client = FakeClient(
         {"CreateStream": [CreateStreamResult(stream=ref)], "ReadRecords": list(pages)}
     )
     stream = await client.create_stream(Texts, id="s1")
 
-    read = [page async for page in stream.iter_read(from_offset=1)]
+    read = stream.iter_read(from_offset=10)
+    seen: list[tuple[int | None, int | None]] = []
+    async for record in read:
+        seen.append(
+            (record.offset, read.next_request and read.next_request.from_offset)
+        )
 
-    assert read == pages
+    # After an item, the read resumes past it, and after a page's last item, where the page ends.
+    assert seen == [(10, 11), (12, 15), (20, None)]
+    assert read.next_request is None
     assert [call.input for call in client.calls[1:]] == [
-        ReadRecordsRequest(stream=ref, from_offset=1),
-        ReadRecordsRequest(stream=ref, from_offset=2),
+        ReadRecordsRequest(stream=ref, from_offset=10),
+        ReadRecordsRequest(stream=ref, from_offset=15),
+        ReadRecordsRequest(stream=ref, from_offset=15),
     ]
 
 

advanced/samples/python/tests/test_poll_service.py

@@ -23,14 +23,26 @@
     ListWorkflowsRequest,
     ListWorkflowsResponse,
     OpenFeedRequest,
+    WorkflowExecution,
+    WorkflowExecutionInfo,
 )
 
-# Each page names the next one, and the last page names none.
+
+def info(workflow_id: str) -> WorkflowExecutionInfo:
+    return WorkflowExecutionInfo(
+        execution=WorkflowExecution(workflow_id=workflow_id, run_id="r1")
+    )
+
+
+# Each page names the next one, and the last page names none. The middle page is empty.
 PAGES = {
-    None: ListWorkflowsResponse(next_page_token=b"2"),
+    None: ListWorkflowsResponse(
+        executions=[info("w1"), info("w2")], next_page_token=b"2"
+    ),
     b"2": ListWorkflowsResponse(next_page_token=b"3"),
-    b"3": ListWorkflowsResponse(),
+    b"3": ListWorkflowsResponse(executions=[info("w3")]),
 }
+ITEMS = [info("w1"), info("w2"), info("w3")]
 
 
 def page(request: ListWorkflowsRequest) -> ListWorkflowsResponse:
@@ -102,9 +114,12 @@ async def test_a_client_polls_an_operation_until_the_helper_stops() -> None:
 async def test_a_client_polls_a_service_with_no_endpoint() -> None:
     client = FakeClient()
 
-    pages = [response async for response in client.iter_list_feed(namespace="default")]
+    read = client.iter_list_feed(namespace="default")
+    items = [item async for item in read]
 
-    assert pages == list(PAGES.values())
+    # The feed's poll yields the executions of each page.
+    assert items == ITEMS
+    assert read.next_request is None
     assert client.tokens("ListFeed") == [None, b"2", b"3"]
     assert {endpoint for endpoint, _, _ in client.calls} == {None}
 
@@ -115,9 +130,9 @@ async def test_a_client_resource_polls_its_method() -> None:
         namespace="default", query="q", order=FeedOrder.Oldest
     )
 
-    pages = [response async for response in feed.iter_read(page_size=10)]
+    items = [item async for item in feed.iter_read(page_size=10)]
 
-    assert pages == list(PAGES.values())
+    assert items == ITEMS
     assert client.calls[0][2] == OpenFeedRequest(
         namespace="default", query="q", order=FeedOrder.Oldest
     )
@@ -167,12 +182,12 @@ class PollWorkflow:
     async def run(self) -> list[int]:
         pages = [p async for p in poll_service.iter_list_workflows(namespace="default")]
         service = poll_service.FeedServiceClient("feeds")
-        feed_pages = [p async for p in service.iter_list_feed(namespace="default")]
+        feed_items = [i async for i in service.iter_list_feed(namespace="default")]
         feed = await service.open_feed(
             namespace="default", query="q", order=FeedOrder.Oldest
         )
-        read_pages = [p async for p in feed.iter_read()]
-        return [len(pages), len(feed_pages), len(read_pages)]
+        read_items = [i async for i in feed.iter_read()]
+        return [len(pages), len(feed_items), len(read_items)]
 
 
 async def test_workflow_code_polls_each_kind_of_operation(