Refused a stream publish from a query handler at the call.

temporalio/workflow/_streams.py

@@ -29,6 +29,7 @@
 from temporalio.streams._topic import StreamTopic, resolve_topic
 from temporalio.streams._wire import RecordDecoder, to_wire
 from temporalio.workflow._context import _Runtime, payload_converter
+from temporalio.workflow._exceptions import ReadOnlyContextError
 from temporalio.workflow._sandbox import logger
 
 __all__ = ["StreamReader", "StreamWriter", "stream_reader", "stream_writer"]
@@ -169,9 +170,18 @@ def publish(self, value: T) -> None:
         Raises:
             ValueError: The topic was already finished in this run, by this
                 writer or by another one on the same topic.
+            temporalio.workflow.ReadOnlyContextError: Called from a query or an
+                update validator, which commit nothing.
         """
         if self._topic in self._finished:
             raise ValueError(f"topic {self._topic!r} was already finished")
+        # A query is answered on a task that carries the answer and nothing else, so a
+        # record published here could only ever be dropped. Said at the call rather than
+        # thrown away later without a word.
+        if _Runtime.current().workflow_is_read_only():
+            raise ReadOnlyContextError(
+                "While in read-only function, action attempted: publish to a stream"
+            )
         self._sink.publish(
             to_wire(
                 payload_converter(),

tests/streams/test_streams_workflow.py

@@ -25,6 +25,7 @@
 from temporalio import workflow
 from temporalio.client import Client
 from temporalio.streams import (
+    BEGINNING,
     DEFAULT_TOPIC,
     END,
     Cursor,
@@ -507,6 +508,41 @@ async def read_everything() -> list[Any]:
     ]
 
 
+@workflow.defn
+class PublishFromQuery:
+    """Publishes from a query handler, which commits nothing."""
+
+    @workflow.run
+    async def run(self) -> None:
+        await workflow.wait_condition(lambda: False)
+
+    @workflow.query
+    def peek(self) -> str:
+        workflow.stream_writer(DECISIONS).publish({"from": "query"})
+        return "unreachable"
+
+
+async def test_a_publish_from_a_query_handler_is_refused_at_the_call(
+    client: Client, provider: MemoryStreams
+):
+    # A query is answered on a task that carries the answer alone, so the record
+    # could only ever be dropped on the way out. The call says so instead.
+    workflow_id = f"streams-wf-{uuid.uuid4().hex}"
+    async with new_worker(client, PublishFromQuery, plugins=[provider]) as worker:
+        handle = await client.start_workflow(
+            PublishFromQuery.run, id=workflow_id, task_queue=worker.task_queue
+        )
+        try:
+            with pytest.raises(Exception) as caught:
+                await handle.query(PublishFromQuery.peek)
+            assert "publish to a stream" in str(caught.value)
+
+            stream = provider.get_stream_handle(client, workflow_id)
+            assert await stream.latest(topic=DECISIONS) == BEGINNING
+        finally:
+            await handle.terminate()
+
+
 @workflow.defn
 class DefaultTopicEcho:
     """Reads one value on its default topic and answers on the same topic."""