From a4f747a17eced1300bb1ae1598a0cfef8ce46f5e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 14:56:54 -0700 Subject: [PATCH 1/5] Added stream topics and StreamRef. A topic is defined once with its record type, and a StreamRef names a Workflow's stream as data another process can open. --- temporalio/contrib/streams/__init__.py | 16 ++++ temporalio/contrib/streams/_ref.py | 83 ++++++++++++++++++++ temporalio/contrib/streams/_topic.py | 92 +++++++++++++++++++++++ tests/contrib/streams/test_topics_refs.py | 56 ++++++++++++++ 4 files changed, 247 insertions(+) create mode 100644 temporalio/contrib/streams/_ref.py create mode 100644 temporalio/contrib/streams/_topic.py create mode 100644 tests/contrib/streams/test_topics_refs.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index d71ef1947d..0221a926ca 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -19,6 +19,9 @@ or another stream with :class:`StreamCursorError`, and one whose record retention dropped with :class:`StreamExpiredError`. A read with no cursor starts at :data:`BEGINNING` or :data:`END`. +3. **Topics are defined once.** :func:`topic` defines a topic with the type + its records decode to, and every party shares that definition. A call + that names no topic addresses :data:`DEFAULT_TOPIC`. The record on the wire is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.contrib.streams.proto.v1`, with the user's value in ``body`` @@ -49,11 +52,19 @@ StreamRecord, Supersession, ) +from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef +from temporalio.contrib.streams._topic import ( + DEFAULT_TOPIC, + StreamTopic, + resolve_topic, + topic, +) __all__ = [ "BEGINNING", "CONTENT_HASH_KEY", "Cursor", + "DEFAULT_TOPIC", "END", "RecordKind", "StreamClosedError", @@ -62,12 +73,17 @@ "StreamExpiredError", "StreamNotFoundError", "StreamOutcomeUnknownError", + "StreamOwnerKind", "StreamProducerError", "StreamRecord", + "StreamRef", + "StreamTopic", "StreamUnsupportedError", "Supersession", "content_fingerprint", "content_hash", "decode_body", "encode_body", + "resolve_topic", + "topic", ] diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py new file mode 100644 index 0000000000..0cd5d02588 --- /dev/null +++ b/temporalio/contrib/streams/_ref.py @@ -0,0 +1,83 @@ +"""A reference to one stream that crosses a process boundary as data. + +A handle is bound to a client and a provider, so it cannot be a Workflow +argument or an Activity result. A :class:`StreamRef` can: it names the owner +and the topic, nothing more, and whoever receives it opens the stream on the +provider its own client carries. It carries no cursor, because a position +belongs to a reader, and no provider name. +""" + +from __future__ import annotations + +import dataclasses +from dataclasses import dataclass +from typing import Any, Literal + +from temporalio.contrib.streams._topic import ( + DEFAULT_TOPIC, + StreamTopic, + resolve_topic, +) + +__all__ = ["StreamOwnerKind", "StreamRef"] + +StreamOwnerKind = Literal["workflow"] +"""What owns a stream. Only a Workflow does in this release.""" + + +@dataclass(frozen=True) +class StreamRef: + """One stream, named by its owner and its topic. + + ``kind`` says what owns the stream; only ``"workflow"`` exists in this + release. A Workflow ref carries ``workflow_id`` and, when pinned to one + run, ``run_id``. Without ``run_id`` it follows the run chain, so a reader + keeps reading across Continue-as-New. ``topic`` is the topic a handle + opened from the ref addresses when a call names none. + + The default data converter carries it as JSON, so it can be a Workflow + argument or an Activity result. + """ + + kind: StreamOwnerKind + workflow_id: str + run_id: str | None = None + topic: str = DEFAULT_TOPIC + + def __post_init__(self) -> None: + """Refuse a ref this release cannot open. + + Raises: + ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or + ``topic`` is empty. + """ + if self.kind != "workflow": + raise ValueError(f"unknown StreamRef kind {self.kind!r}") + if 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") + + @classmethod + def for_workflow( + cls, + workflow_id: str, + *, + run_id: str | None = None, + topic: str | StreamTopic[Any] | None = None, + ) -> StreamRef: + """A ref to ``topic`` of ``workflow_id``'s stream. + + Without ``topic`` it names + :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + """ + name, _ = resolve_topic(topic) + return cls("workflow", workflow_id, run_id=run_id, topic=name) + + def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef: + """The same owner, naming ``topic`` instead. + + ``None`` names :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + """ + name, _ = resolve_topic(topic) + return dataclasses.replace(self, topic=name) diff --git a/temporalio/contrib/streams/_topic.py b/temporalio/contrib/streams/_topic.py new file mode 100644 index 0000000000..3e4b3de9ac --- /dev/null +++ b/temporalio/contrib/streams/_topic.py @@ -0,0 +1,92 @@ +"""Typed topic definitions. + +A topic is defined once, at module level, with the type its records decode +to, and the Workflow, its Activities and the client code all refer to that +one definition, the way they share a Signal or an Activity. A plain string +names a topic too, for a name decided at runtime; then the decode hint +travels as ``result_type=`` on each read. + +On the wire a topic is a string, and +:attr:`temporalio.contrib.streams.StreamRecord.topic` is that string. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Generic, TypeVar, overload + +__all__ = ["DEFAULT_TOPIC", "StreamTopic", "resolve_topic", "topic"] + +T = TypeVar("T") + +DEFAULT_TOPIC = "output" +"""The topic a call addresses when it names none. + +It is an ordinary name, so naming it explicitly addresses the same topic. +""" + + +@dataclass(frozen=True) +class StreamTopic(Generic[T]): + """A topic of a stream, with the type its records decode to. + + Made with :func:`topic`. Pass it wherever a call takes a topic, and the + record and value types follow from it. + """ + + name: str + result_type: type[T] | None = None + """The type records decode to, or ``None`` for the converter's default.""" + + +@overload +def topic(name: str, result_type: type[T]) -> StreamTopic[T]: ... + + +@overload +def topic(name: str, result_type: None = None) -> StreamTopic[Any]: ... + + +def topic(name: str, result_type: type | None = None) -> StreamTopic[Any]: + """Define a topic. + + Args: + name: The topic's name, as it appears on every record. + result_type: The type records on this topic decode to. Without one, + the payload converter's default applies. + + Raises: + ValueError: ``name`` is empty. + """ + if not name: + raise ValueError("topic name must not be empty") + return StreamTopic(name, result_type) + + +def resolve_topic( + topic: str | StreamTopic[Any] | None = None, result_type: type | None = None +) -> tuple[str, type | None]: + """The name and decode hint a call means, from either form of topic. + + ``None`` means :data:`DEFAULT_TOPIC`. A provider calls this once at the + top of each call that takes a topic, so a definition and a string reach + the store the same way. + + Raises: + ValueError: A definition was given together with ``result_type``, + which would name two types for one topic, or the name is empty. + """ + if isinstance(topic, StreamTopic): + if result_type is not None: + raise ValueError( + f"topic {topic.name!r} already carries its type; do not pass " + "result_type= with a definition" + ) + name, result_type = topic.name, topic.result_type + elif topic is None: + name = DEFAULT_TOPIC + else: + name = topic + if not name: + raise ValueError("topic must not be empty") + return name, result_type diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py new file mode 100644 index 0000000000..9c8ad48cdc --- /dev/null +++ b/tests/contrib/streams/test_topics_refs.py @@ -0,0 +1,56 @@ +"""Typed topics and the stream reference that crosses process boundaries.""" + +from __future__ import annotations + +import pytest + +from temporalio.contrib.streams import ( + DEFAULT_TOPIC, + StreamRef, + StreamTopic, + resolve_topic, + topic, +) +from temporalio.converter import DataConverter + +OUT = topic("out", dict) + + +def test_a_definition_carries_its_name_and_type(): + assert OUT == StreamTopic("out", dict) + assert resolve_topic(OUT) == ("out", dict) + assert resolve_topic("free", int) == ("free", int) + assert resolve_topic(None) == (DEFAULT_TOPIC, None) + + +def test_topic_mistakes_are_value_errors(): + with pytest.raises(ValueError): + topic("") + with pytest.raises(ValueError): + resolve_topic("") + with pytest.raises(ValueError, match="already carries its type"): + resolve_topic(OUT, dict) + + +def test_a_ref_names_a_workflow_stream(): + ref = StreamRef.for_workflow("wf", topic=OUT) + assert ref == StreamRef("workflow", "wf", topic="out") + assert StreamRef.for_workflow("wf").topic == DEFAULT_TOPIC + assert ref.with_topic(None).topic == DEFAULT_TOPIC + assert ref.with_topic("other").workflow_id == "wf" + + +def test_a_ref_round_trips_through_the_default_converter(): + ref = StreamRef.for_workflow("wf", run_id="run", topic=OUT) + converter = DataConverter.default.payload_converter + payload = converter.to_payloads([ref])[0] + assert converter.from_payloads([payload], [StreamRef])[0] == ref + + +def test_a_ref_refuses_what_it_cannot_name(): + with pytest.raises(ValueError, match="unknown"): + StreamRef("nexus", "x") # type: ignore[arg-type] + with pytest.raises(ValueError): + StreamRef("workflow", "") + with pytest.raises(ValueError): + StreamRef("workflow", "wf", topic="") From 9f61ca84711008265fe2b9df3840da40a71bb971 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 15:10:28 -0700 Subject: [PATCH 2/5] Widened the stream reference kind to any string. A ref that names an owner kind from a later release now reaches the ref's own kind check when decoded, instead of failing the converter's type check. --- temporalio/contrib/streams/_ref.py | 9 +++++++-- tests/contrib/streams/test_topics_refs.py | 15 ++++++++++++++- 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index 0cd5d02588..3bc6fd82e9 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -22,7 +22,12 @@ __all__ = ["StreamOwnerKind", "StreamRef"] StreamOwnerKind = Literal["workflow"] -"""What owns a stream. Only a Workflow does in this release.""" +"""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. +""" @dataclass(frozen=True) @@ -39,7 +44,7 @@ class StreamRef: argument or an Activity result. """ - kind: StreamOwnerKind + kind: str workflow_id: str run_id: str | None = None topic: str = DEFAULT_TOPIC diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 9c8ad48cdc..b2708a9106 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -4,6 +4,7 @@ import pytest +from temporalio.api.common.v1 import Payload from temporalio.contrib.streams import ( DEFAULT_TOPIC, StreamRef, @@ -49,8 +50,20 @@ def test_a_ref_round_trips_through_the_default_converter(): def test_a_ref_refuses_what_it_cannot_name(): with pytest.raises(ValueError, match="unknown"): - StreamRef("nexus", "x") # type: ignore[arg-type] + 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. + converter = DataConverter.default.payload_converter + payload = Payload( + metadata={"encoding": b"json/plain"}, + data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}', + ) + with pytest.raises(ValueError, match="'activity'"): + converter.from_payloads([payload], [StreamRef]) From 3ee8c3a742b60cee194e033bbbb2aedc0e0519a3 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 17:20:49 -0700 Subject: [PATCH 3/5] Kept resolve_topic out of the public stream surface. Only providers resolve a topic argument, so applications do not need it and the package should not commit to it. --- temporalio/contrib/streams/__init__.py | 2 -- tests/contrib/streams/test_public_surface.py | 1 + tests/contrib/streams/test_topics_refs.py | 4 +++- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index d524dbfbea..2b89ff5a6c 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -51,7 +51,6 @@ from temporalio.contrib.streams._topic import ( DEFAULT_TOPIC, StreamTopic, - resolve_topic, topic, ) @@ -76,6 +75,5 @@ "StreamTopic", "StreamUnsupportedError", "Supersession", - "resolve_topic", "topic", ] diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py index 76a1e09099..3abae317f5 100644 --- a/tests/contrib/streams/test_public_surface.py +++ b/tests/contrib/streams/test_public_surface.py @@ -12,6 +12,7 @@ "content_hash", "decode_body", "encode_body", + "resolve_topic", } diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index b2708a9106..1d00dceb62 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -9,9 +9,11 @@ DEFAULT_TOPIC, StreamRef, StreamTopic, - resolve_topic, topic, ) +from temporalio.contrib.streams._topic import ( + resolve_topic, +) from temporalio.converter import DataConverter OUT = topic("out", dict) From 8d0a673b3184e3d5128993421b2ae3e9c6b1e0ec Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 13:03:29 -0700 Subject: [PATCH 4/5] Bounded topic names and said what a ref's run id changes. A name goes into store keys, cursor hashes and Core's manifest budget, and a control character in it broke the pending stage encoding. A pinned ref was read as returning only that run's records, which no provider does. --- temporalio/contrib/streams/_ref.py | 9 +++++--- temporalio/contrib/streams/_topic.py | 28 ++++++++++++++++++----- tests/contrib/streams/test_topics_refs.py | 17 ++++++++++++++ 3 files changed, 45 insertions(+), 9 deletions(-) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index 3bc6fd82e9..6faa99f970 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -36,9 +36,12 @@ class StreamRef: ``kind`` says what owns the stream; only ``"workflow"`` exists in this release. A Workflow ref carries ``workflow_id`` and, when pinned to one - run, ``run_id``. Without ``run_id`` it follows the run chain, so a reader - keeps reading across Continue-as-New. ``topic`` is the topic a handle - opened from the ref addresses when a call names none. + run, ``run_id``. The stream belongs to the run chain either way, and a + read returns the chain's records. ``run_id`` changes two things: which + chain the handle looks up, and that a read ends when that run closes, + Continue-as-New included. Without it a reader keeps reading across + Continue-as-New. ``topic`` is the topic a handle opened from the ref + addresses when a call names none. The default data converter carries it as JSON, so it can be a Workflow argument or an Activity result. diff --git a/temporalio/contrib/streams/_topic.py b/temporalio/contrib/streams/_topic.py index 3e4b3de9ac..6eb854d238 100644 --- a/temporalio/contrib/streams/_topic.py +++ b/temporalio/contrib/streams/_topic.py @@ -12,6 +12,7 @@ from __future__ import annotations +import unicodedata from dataclasses import dataclass from typing import Any, Generic, TypeVar, overload @@ -56,10 +57,10 @@ def topic(name: str, result_type: type | None = None) -> StreamTopic[Any]: the payload converter's default applies. Raises: - ValueError: ``name`` is empty. + ValueError: ``name`` is empty, longer than 256 UTF-8 bytes, or holds + a control character. """ - if not name: - raise ValueError("topic name must not be empty") + _check_name(name) return StreamTopic(name, result_type) @@ -74,7 +75,8 @@ def resolve_topic( Raises: ValueError: A definition was given together with ``result_type``, - which would name two types for one topic, or the name is empty. + which would name two types for one topic, or the name is empty, + longer than 256 UTF-8 bytes, or holds a control character. """ if isinstance(topic, StreamTopic): if result_type is not None: @@ -87,6 +89,20 @@ def resolve_topic( name = DEFAULT_TOPIC else: name = topic - if not name: - raise ValueError("topic must not be empty") + _check_name(name) return name, result_type + + +_MAX_NAME_BYTES = 256 + + +def _check_name(name: str) -> None: + # The same limit for every provider: the name goes into store keys, cursor + # hashes and Core's manifest budget, and a provider may join names with a + # control character. + if not name: + raise ValueError("topic name must not be empty") + if len(name.encode()) > _MAX_NAME_BYTES: + raise ValueError(f"topic name must be at most {_MAX_NAME_BYTES} UTF-8 bytes") + if any(unicodedata.category(c) == "Cc" for c in name): + raise ValueError(f"topic name {name!r} must not hold a control character") diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 1d00dceb62..3237a011f7 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -69,3 +69,20 @@ def test_a_ref_from_a_later_release_decodes_to_the_kind_check(): ) with pytest.raises(ValueError, match="'activity'"): converter.from_payloads([payload], [StreamRef]) + + +@pytest.mark.parametrize( + "name", + ["x" * 257, "é" * 129, "a\x1fb", "line\nbreak", "\x00"], +) +def test_a_topic_name_that_is_too_long_or_has_a_control_character_is_refused(name: str): + # Every provider puts the name in keys, cursors and Core's manifest budget. + with pytest.raises(ValueError): + topic(name) + with pytest.raises(ValueError): + resolve_topic(name) + + +def test_a_topic_name_of_256_bytes_is_accepted(): + assert topic("x" * 256).name == "x" * 256 + assert resolve_topic("é" * 128)[0] == "é" * 128 From f6dce6f714e5659beb2df215d6dddef8b8897840 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:08:45 -0700 Subject: [PATCH 5/5] Said a stream ref opens on its client's store. There's no provider any more, only the store a client's plugin configures. --- temporalio/contrib/streams/__init__.py | 7 ++----- temporalio/contrib/streams/_ref.py | 6 +++--- temporalio/contrib/streams/_topic.py | 6 +++--- tests/contrib/streams/test_public_surface.py | 1 + tests/contrib/streams/test_topics_refs.py | 2 +- 5 files changed, 10 insertions(+), 12 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 33e3e0eaf5..fe83ad676e 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -49,11 +49,7 @@ Supersession, ) from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef -from temporalio.contrib.streams._topic import ( - DEFAULT_TOPIC, - StreamTopic, - topic, -) +from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic __all__ = [ "BEGINNING", @@ -71,6 +67,7 @@ "StreamProducerError", "StreamRecord", "StreamRecordError", + "StreamRef", "StreamRefusedError", "StreamStorageError", "StreamTopic", diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index 6faa99f970..bc8308398a 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -1,10 +1,10 @@ """A reference to one stream that crosses a process boundary as data. -A handle is bound to a client and a provider, so it cannot be a Workflow +A handle is bound to a client and its store, so it cannot be a Workflow argument or an Activity result. A :class:`StreamRef` can: it names the owner and the topic, nothing more, and whoever receives it opens the stream on the -provider its own client carries. It carries no cursor, because a position -belongs to a reader, and no provider name. +store its own client carries. It carries no cursor, because a position +belongs to a reader, and no store name. """ from __future__ import annotations diff --git a/temporalio/contrib/streams/_topic.py b/temporalio/contrib/streams/_topic.py index 6eb854d238..0f2ec562a5 100644 --- a/temporalio/contrib/streams/_topic.py +++ b/temporalio/contrib/streams/_topic.py @@ -69,7 +69,7 @@ def resolve_topic( ) -> tuple[str, type | None]: """The name and decode hint a call means, from either form of topic. - ``None`` means :data:`DEFAULT_TOPIC`. A provider calls this once at the + ``None`` means :data:`DEFAULT_TOPIC`. A handle calls this once at the top of each call that takes a topic, so a definition and a string reach the store the same way. @@ -97,8 +97,8 @@ def resolve_topic( def _check_name(name: str) -> None: - # The same limit for every provider: the name goes into store keys, cursor - # hashes and Core's manifest budget, and a provider may join names with a + # The same limit in every SDK: the name goes into store keys, cursor + # hashes and Core's manifest budget, and a store may join names with a # control character. if not name: raise ValueError("topic name must not be empty") diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py index a3647d4eba..1161287e40 100644 --- a/tests/contrib/streams/test_public_surface.py +++ b/tests/contrib/streams/test_public_surface.py @@ -12,6 +12,7 @@ "decode_body", "encode_bodies", "error_from_failure", + "resolve_topic", } diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 3237a011f7..3cbbe1e6b0 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -76,7 +76,7 @@ def test_a_ref_from_a_later_release_decodes_to_the_kind_check(): ["x" * 257, "é" * 129, "a\x1fb", "line\nbreak", "\x00"], ) def test_a_topic_name_that_is_too_long_or_has_a_control_character_is_refused(name: str): - # Every provider puts the name in keys, cursors and Core's manifest budget. + # Every store puts the name in keys, cursors and Core's manifest budget. with pytest.raises(ValueError): topic(name) with pytest.raises(ValueError):