[Brook] Refused stream owners this release does not support.

temporalio/contrib/streams/__init__.py

@@ -47,6 +47,10 @@
 Every read follows the Workflow's run chain across Continue-as-New and ends
 when the chain closes. A reader can't pin one run.
 
+In this release only a Workflow owns a stream, and only Activities and
+clients read one. Reading inside a Workflow (:func:`workflow_reader`) and
+the other owner kinds raise :class:`StreamUnsupportedError`.
+
 The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in
 :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body``
 as an ordinary payload.
@@ -82,6 +86,7 @@
 )
 from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef
 from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic
+from temporalio.contrib.streams._workflow import workflow_reader
 
 __all__ = [
     "ActivityStreamHandle",
@@ -112,4 +117,5 @@
     "activity_handle",
     "get_stream_handle",
     "topic",
+    "workflow_reader",
 ]

temporalio/contrib/streams/_handles.py

@@ -542,9 +542,10 @@ def activity_handle() -> ActivityStreamHandle:
     instead.
 
     Raises:
-        ValueError: The Activity was not scheduled by a Workflow, or no stream
-            store is registered on the Worker's client.
-        StreamUnsupportedError: The Activity is a Local Activity.
+        StreamUnsupportedError: The Activity is a Local Activity, or it was
+            not scheduled by a Workflow, since streams owned by an Activity
+            are not supported in this release.
+        ValueError: No stream store is registered on the Worker's client.
         RuntimeError: Not called from inside an Activity.
     """
     info = temporalio.activity.info()
@@ -554,9 +555,9 @@ def activity_handle() -> ActivityStreamHandle:
             "attempt isn't durable; use a regular Activity instead"
         )
     if info.workflow_id is None or info.workflow_run_id is None:
-        raise ValueError(
-            "this Activity was not scheduled by a Workflow, so it has no Workflow "
-            "stream"
+        raise StreamUnsupportedError(
+            "this Activity was not scheduled by a Workflow, and streams owned by "
+            "an Activity are not supported in this release"
         )
     client = temporalio.activity.client()
     ref = StreamRef.for_workflow(info.workflow_id, chain_run_id=info.workflow_run_id)
@@ -621,4 +622,5 @@ def get_stream_handle(
         ref = StreamRef.for_workflow(
             workflow_id, topic=topic, chain_run_id=chain_run_id
         )
+    ref._require_supported()
     return StreamHandle(store_for_client(client), client, ref)

temporalio/contrib/streams/_ref.py

@@ -13,6 +13,7 @@
 from dataclasses import dataclass
 from typing import Any, Literal
 
+from temporalio.contrib.streams._errors import StreamUnsupportedError
 from temporalio.contrib.streams._topic import (
     DEFAULT_TOPIC,
     StreamTopic,
@@ -25,8 +26,7 @@
 """The owner kinds this release opens. Only a Workflow owns a stream.
 
 :attr:`StreamRef.kind` is a plain string, so a ref that names an owner kind
-a later release adds still reaches the ref's own check when it is decoded,
-instead of failing the converter's type check.
+a later release adds still decodes, and is refused where a handle opens.
 """
 
 
@@ -55,19 +55,35 @@ class StreamRef:
     chain_run_id: str | None = None
 
     def __post_init__(self) -> None:
-        """Refuse a ref this release cannot open.
+        """Refuse a ref that names no stream.
+
+        Any ``kind`` decodes, since a ref of a later release can arrive as
+        Workflow input, where raising would fail the Workflow Task on every
+        retry. :meth:`_require_supported` refuses it where a handle opens.
 
         Raises:
-            ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or
-                ``topic`` is empty.
+            ValueError: ``kind`` or ``topic`` is empty, or a Workflow ref has
+                no ``workflow_id``.
         """
-        if self.kind != "workflow":
-            raise ValueError(f"unknown StreamRef kind {self.kind!r}")
-        if not self.workflow_id:
+        if not self.kind:
+            raise ValueError("a StreamRef needs a kind")
+        if self.kind == "workflow" and not self.workflow_id:
             raise ValueError("a Workflow StreamRef needs a workflow_id")
         if not self.topic:
             raise ValueError("a StreamRef needs a topic name")
 
+    def _require_supported(self) -> None:
+        """Refuse to open a stream whose owner kind this release lacks.
+
+        Raises:
+            StreamUnsupportedError: ``kind`` is not ``"workflow"``.
+        """
+        if self.kind != "workflow":
+            raise StreamUnsupportedError(
+                f"streams owned by kind {self.kind!r} are not supported in this "
+                "release; only Workflow-owned streams are"
+            )
+
     @classmethod
     def for_workflow(
         cls,

temporalio/contrib/streams/_workflow.py

@@ -0,0 +1,30 @@
+"""Streams inside Workflow code.
+
+Reading a stream inside a Workflow is not part of this release.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from temporalio.contrib.streams._errors import StreamUnsupportedError
+
+__all__ = ["workflow_reader"]
+
+
+def workflow_reader(*_args: Any, **_kwargs: Any) -> Any:
+    """Reading a stream inside a Workflow is not supported in this release.
+
+    Read from an Activity or a client with
+    :func:`temporalio.contrib.streams.activity_handle` or
+    :func:`temporalio.contrib.streams.get_stream_handle`, and send the
+    Workflow what it needs, for example as a Signal.
+
+    Raises:
+        StreamUnsupportedError: Always.
+    """
+    raise StreamUnsupportedError(
+        "reading a stream inside a Workflow is not supported in this release; "
+        "read it from an Activity or a client and send the Workflow what it "
+        "needs, for example as a Signal"
+    )

tests/contrib/streams/test_outside_handles.py

@@ -396,19 +396,6 @@ async def test_a_named_chain_stays_apart_from_a_newer_chain(client: Client):
     assert [r.value for r in new_records] == ["new"]
 
 
-async def test_an_activity_with_no_workflow_has_no_workflow_stream():
-    env = ActivityEnvironment()
-    env.info = dataclasses.replace(
-        ActivityEnvironment.default_info(), workflow_id=None, workflow_run_id=None
-    )
-
-    async def standalone() -> None:
-        activity_handle()
-
-    with pytest.raises(ValueError, match="not scheduled by a Workflow"):
-        await env.run(standalone)
-
-
 @activity.defn
 async def count_twice() -> str:
     producer = activity_handle().producer(topic=TOKENS)

tests/contrib/streams/test_topics_refs.py

@@ -9,6 +9,7 @@
     DEFAULT_TOPIC,
     StreamRef,
     StreamTopic,
+    StreamUnsupportedError,
     topic,
 )
 from temporalio.contrib.streams._topic import (
@@ -63,24 +64,26 @@ def test_a_ref_with_the_old_run_id_decodes_as_the_newest_chain():
 
 
 def test_a_ref_refuses_what_it_cannot_name():
-    with pytest.raises(ValueError, match="unknown"):
-        StreamRef("nexus", "x")
     with pytest.raises(ValueError):
         StreamRef("workflow", "")
     with pytest.raises(ValueError):
         StreamRef("workflow", "wf", topic="")
 
 
-def test_a_ref_from_a_later_release_decodes_to_the_kind_check():
-    # The kind is a plain string, so the converter builds the ref and the ref
-    # itself refuses the kind, rather than the converter failing a type check.
+@pytest.mark.parametrize("kind", ["activity", "nexus"])
+def test_a_ref_from_a_later_release_decodes_and_fails_where_it_opens(kind: str):
+    # Decoding happens while a Workflow Task applies its input, where a raise
+    # fails the task on every retry. Opening the handle is where the caller
+    # can catch it.
     converter = DataConverter.default.payload_converter
     payload = Payload(
         metadata={"encoding": b"json/plain"},
-        data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}',
+        data=f'{{"kind": "{kind}", "workflow_id": "wf", "topic": "out"}}'.encode(),
     )
-    with pytest.raises(ValueError, match="'activity'"):
-        converter.from_payloads([payload], [StreamRef])
+    ref = converter.from_payloads([payload], [StreamRef])[0]
+    assert ref.kind == kind
+    with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"):
+        ref._require_supported()
 
 
 @pytest.mark.parametrize(

tests/contrib/streams/test_unsupported_owners.py

@@ -0,0 +1,73 @@
+"""Owners and readers this release does not support say so.
+
+Only a Workflow owns a stream in this release, and only Activities and
+clients read one. Everything else raises ``StreamUnsupportedError``.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+
+import pytest
+
+from temporalio import workflow
+from temporalio.client import Client
+from temporalio.contrib.streams import (
+    StreamRef,
+    StreamUnsupportedError,
+    activity_handle,
+    get_stream_handle,
+    topic,
+    workflow_reader,
+)
+from temporalio.contrib.streams.memory import MemoryStreams
+from temporalio.testing import ActivityEnvironment
+from tests.contrib.streams._support import connect_with, new_workflow_id
+from tests.helpers import new_worker
+
+EVENTS = topic("events", dict)
+
+
+async def test_a_ref_refuses_owner_kinds_this_release_lacks_where_it_opens(
+    client: Client,
+):
+    streams_client = await connect_with(client, MemoryStreams())
+    for kind in ("activity", "standalone"):
+        ref = StreamRef(kind, "x")  # type: ignore[arg-type]
+        with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"):
+            get_stream_handle(streams_client, ref)
+
+
+@workflow.defn
+class TriesToRead:
+    @workflow.run
+    async def run(self) -> str:
+        try:
+            workflow_reader(EVENTS)
+        except StreamUnsupportedError as error:
+            return str(error)
+        return "read"
+
+
+async def test_a_workflow_cannot_read_a_stream(client: Client):
+    streams_client = await connect_with(client, MemoryStreams())
+    async with new_worker(streams_client, TriesToRead) as worker:
+        said = await streams_client.execute_workflow(
+            TriesToRead.run,
+            id=new_workflow_id(),
+            task_queue=worker.task_queue,
+        )
+    assert "not supported in this release" in said
+
+
+async def test_an_activity_with_no_workflow_has_no_stream():
+    env = ActivityEnvironment()
+    env.info = dataclasses.replace(
+        ActivityEnvironment.default_info(), workflow_id=None, workflow_run_id=None
+    )
+
+    async def standalone() -> None:
+        activity_handle()
+
+    with pytest.raises(StreamUnsupportedError, match="owned by an Activity"):
+        await env.run(standalone)