workflow_streams: parametrize WorkflowStreamItem on decoded data type

Make WorkflowStreamItem generic in T so subscribers get a typed
data field that matches the result_type passed to subscribe:
  - subscribe(result_type=T)        -> WorkflowStreamItem[T]
  - subscribe()                     -> WorkflowStreamItem[Any]
  - subscribe(result_type=RawValue) -> WorkflowStreamItem[RawValue]

Adds def-style overloads to WorkflowStreamClient.subscribe (matching
the existing TopicHandle/WorkflowTopicHandle generic style) and
tightens TopicHandle.subscribe to AsyncIterator[WorkflowStreamItem[T]].
The internal workflow-side _log is annotated as
list[WorkflowStreamItem[Payload]] since the workflow does not decode.

No runtime behavior change; existing tests (which use unparameterized
WorkflowStreamItem) continue to type-check as WorkflowStreamItem[Any].

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>

Author: jssmith

temporalio/contrib/workflow_streams/_client.py

@@ -456,14 +456,33 @@ async def _run_flusher(self) -> None:
             self._flush_event.clear()
             await self._flush()
 
+    @overload
+    def subscribe(
+        self,
+        topics: str | list[str] | None = ...,
+        from_offset: int = ...,
+        *,
+        result_type: type[T],
+        poll_cooldown: timedelta = ...,
+    ) -> AsyncIterator[WorkflowStreamItem[T]]: ...
+    @overload
+    def subscribe(
+        self,
+        topics: str | list[str] | None = ...,
+        from_offset: int = ...,
+        *,
+        result_type: None = None,
+        poll_cooldown: timedelta = ...,
+    ) -> AsyncIterator[WorkflowStreamItem[Any]]: ...
+
     async def subscribe(
         self,
         topics: str | list[str] | None = None,
         from_offset: int = 0,
         *,
         result_type: type | None = None,
         poll_cooldown: timedelta = timedelta(milliseconds=100),
-    ) -> AsyncIterator[WorkflowStreamItem]:
+    ) -> AsyncIterator[WorkflowStreamItem[Any]]:
         """Async iterator that polls for new items.
 
         Automatically follows continue-as-new chains when the client

temporalio/contrib/workflow_streams/_stream.py

@@ -136,7 +136,7 @@ def __init__(self, prior_state: WorkflowStreamState | None = None) -> None:
             )
 
         if prior_state is not None:
-            self._log: list[WorkflowStreamItem] = [
+            self._log: list[WorkflowStreamItem[Payload]] = [
                 WorkflowStreamItem(topic=item.topic, data=_decode_payload(item.data))
                 for item in prior_state.log
             ]

temporalio/contrib/workflow_streams/_topic_handle.py

@@ -91,7 +91,7 @@ async def subscribe(
         from_offset: int = 0,
         *,
         poll_cooldown: timedelta = timedelta(milliseconds=100),
-    ) -> AsyncIterator[WorkflowStreamItem]:
+    ) -> AsyncIterator[WorkflowStreamItem[T]]:
         """Async iterator over items on this topic, decoded as ``T``.
 
         For raw ``Payload`` access, or any other decode type that

temporalio/contrib/workflow_streams/_types.py

@@ -20,10 +20,12 @@
 import base64
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Any
+from typing import Generic, TypeVar
 
 from temporalio.api.common.v1 import Payload
 
+T = TypeVar("T")
+
 
 # basedpyright flags _-prefixed module-level functions as unused even when
 # sibling modules import them (_stream.py, _client.py). Vanilla pyright does
@@ -41,26 +43,26 @@ def _decode_payload(wire: str) -> Payload:  # pyright: ignore[reportUnusedFuncti
 
 
 @dataclass
-class WorkflowStreamItem:
+class WorkflowStreamItem(Generic[T]):
     """A single item in the workflow stream's log.
 
     .. warning::
         This class is experimental and may change in future versions.
 
-    The ``data`` field always carries the decoded value produced by
-    :meth:`WorkflowStreamClient.subscribe`: the converter's default
-    ``Any`` decoding when ``result_type`` is omitted, an instance of
-    ``T`` when ``result_type=T`` is passed, or a
+    The ``data`` field carries the decoded value produced by
+    :meth:`WorkflowStreamClient.subscribe`. The generic parameter ``T``
+    matches the ``result_type`` passed to ``subscribe``: an instance of
+    ``T`` when ``result_type=T``, the converter's default ``Any``
+    decoding when ``result_type`` is omitted, or a
     :class:`temporalio.common.RawValue` wrapping the original
-    ``Payload`` when ``result_type=RawValue`` is passed. The dataclass
-    is typed as ``Any`` to accommodate all three.
+    ``Payload`` when ``result_type=RawValue``.
 
     The ``offset`` field is populated at poll time from the item's
     position in the global log.
     """
 
     topic: str
-    data: Any
+    data: T
     offset: int = 0