workflow_streams: rename stop_polling to detach_pollers

Better captures the operation. The method releases the in-flight
__temporal_workflow_stream_poll update handlers (subscribers were
"attached" to the stream's drain signal; this detaches them so
they return to the caller, who can then follow continue-as-new
or stop) and rejects new poll attachments. "stop_polling"
ambiguously suggested the stream itself was the one polling;
"detach_pollers" names the actor (the pollers / subscribers) and
captures the relationship.

Updates the continue_as_new helper and the explicit-recipe
docstring/test accordingly.

Author: jssmith

temporalio/contrib/workflow_streams/_stream.py

@@ -264,17 +264,17 @@ def get_state(
             publishers=active_publishers,
         )
 
-    def stop_polling(self) -> None:
-        """Release waiting subscribers and reject new poll updates.
+    def detach_pollers(self) -> None:
+        """Release waiting pollers and reject new poll updates.
 
         After this call the stream's ``__temporal_workflow_stream_poll``
-        update handler stops serving subscribers on this run: any
-        in-flight polls return their current item batch (often empty)
-        and new polls are rejected at the validator. Publishes still
-        land in the in-memory log and ``get_state`` /
-        ``continue_as_new`` remain valid — the stream is being held
-        open just long enough to snapshot state and hand off to the
-        next run.
+        update handler releases its in-flight subscribers on this run:
+        each waiting poll returns its current item batch (often empty)
+        so the consumer can either follow continue-as-new or stop, and
+        new polls are rejected at the validator. Publishes still land
+        in the in-memory log and ``get_state`` / ``continue_as_new``
+        remain valid — the stream is being held open just long enough
+        to snapshot state and hand off to the next run.
 
         Call this before
         ``await workflow.wait_condition(workflow.all_handlers_finished)``
@@ -288,15 +288,15 @@ async def continue_as_new(
         *,
         publisher_ttl: timedelta = timedelta(seconds=900),
     ) -> NoReturn:
-        """Stop polling, wait for handlers, continue-as-new with built args.
+        """Detach pollers, wait for handlers, continue-as-new with built args.
 
-        Replaces the three-line recipe ``stop_polling()`` →
+        Replaces the three-line recipe ``detach_pollers()`` →
         ``wait_condition(all_handlers_finished)`` →
         ``workflow.continue_as_new(args=...)`` for the common case where
         the only CAN parameter that varies is ``args``.
 
-        ``build_args`` is invoked *after* poll-serving has stopped, with
-        the post-stop :class:`WorkflowStreamState` as its single
+        ``build_args`` is invoked *after* pollers have been detached,
+        with the post-detach :class:`WorkflowStreamState` as its single
         argument. The caller threads that state into whatever input
         dataclass the workflow expects:
 
@@ -309,19 +309,19 @@ async def continue_as_new(
 
         Workflows that need to override other CAN parameters
         (``task_queue``, ``retry_policy``, ``run_timeout``, etc.) should
-        keep using the explicit ``stop_polling`` / ``wait_condition`` /
+        keep using the explicit ``detach_pollers`` / ``wait_condition`` /
         ``workflow.continue_as_new(...)`` recipe.
 
         Args:
-            build_args: Callable that receives the post-stop stream
+            build_args: Callable that receives the post-detach stream
                 state and returns the positional ``args`` for the new
                 run.
             publisher_ttl: Forwarded to :meth:`get_state`.
 
         Does not return; ``workflow.continue_as_new`` raises an internal
         exception that the SDK uses to close the run.
         """
-        self.stop_polling()
+        self.detach_pollers()
         await workflow.wait_condition(workflow.all_handlers_finished)
         workflow.continue_as_new(
             args=build_args(self.get_state(publisher_ttl=publisher_ttl)),

tests/contrib/workflow_streams/test_workflow_streams.py

@@ -1685,7 +1685,7 @@ async def run(self, _input: CANWorkflowInputTyped) -> None:
                 return
             if self._should_continue:
                 self._should_continue = False
-                self.stream.stop_polling()
+                self.stream.detach_pollers()
                 await workflow.wait_condition(workflow.all_handlers_finished)
                 workflow.continue_as_new(
                     args=[
@@ -1829,7 +1829,7 @@ async def run(self, _input: CANWorkflowInputTyped) -> None:
 @pytest.mark.asyncio
 async def test_continue_as_new_helper(client: Client) -> None:
     """The ``WorkflowStream.continue_as_new`` helper preserves log and dedup state
-    just like the explicit stop_polling/wait/CAN recipe."""
+    just like the explicit detach_pollers/wait/CAN recipe."""
     async with new_worker(
         client,
         ContinueAsNewHelperWorkflow,