[Jetty] Merged Brook #365 and made native the default Stream backend.

changelog/added/babbling-brook.md

@@ -0,0 +1,3 @@
+Add experimental Streams to `temporalio.common`. A Stream is an ordered log of Records that Workflows, Activities and clients write, and that clients and Activities read. Register a backend with `stream_backends=` on `Client.connect`, `Replayer` or `WorkflowEnvironment`, then use `workflow.create_stream`, `Client.stream` or `WorkflowHandle.stream`.
+Add a Redis Stream backend in `temporalio.contrib.stream`, and an in-memory one for tests in `temporalio.contrib.stream.testing`. The Redis backend needs a build with the bridge's `streams` feature, which the official builds have.
+Add `activity.Info.workflow_task`, the Workflow Task attempt in which a Worker dispatched a Local Activity.

changelog/added/jaunty-jetty.md

@@ -0,0 +1,2 @@
+Add `temporalio.common.NativeStreamBackend`, a Stream backend that stores Streams in the Temporal server. It needs a server with the Stream service. It's the default and needs no registration. A client can register more than one backend, and its Workers serve all of them.
+Add Workflow reads of a Stream with `WorkflowStream.subscribe`, which returns a `temporalio.common.WorkflowItems`. Add Streams owned by Standalone Activities with `activity.create_stream` and `client.ActivityHandle.stream`, and client-owned Streams with `Client.create_stream`.

changelog/changed/babbling-brook.md

@@ -0,0 +1 @@
+Convert values of a `transfer_type_convertible` class inside JSON values too, such as list items and dataclass fields, not only at the top level of a payload. A dataclass with the decorator is still written as its fields inside a JSON value.

pyproject.toml

@@ -186,9 +186,9 @@ exclude = [
   'temporalio/api',
   'temporalio/bridge/proto',
   'temporalio/nexus/system/workflow_service',
-  'temporalio/contrib/streams/proto',
-  'temporalio/contrib/streams/system',
-  'temporalio/contrib/streams/writer_system',
+  'temporalio/common/_streams/proto',
+  'temporalio/common/_streams/system',
+  'temporalio/common/_streams/writer_system',
 ]
 
 [[tool.mypy.overrides]]
@@ -197,9 +197,9 @@ ignore_errors = true
 
 [[tool.mypy.overrides]]
 module = [
-  "temporalio.contrib.streams.proto.*",
-  "temporalio.contrib.streams.system.*",
-  "temporalio.contrib.streams.writer_system.*",
+  "temporalio.common._streams.proto.*",
+  "temporalio.common._streams.system.*",
+  "temporalio.common._streams.writer_system.*",
 ]
 ignore_errors = true
 
@@ -239,6 +239,8 @@ privacy = [
   "HIDDEN:temporalio.worker.workflow_instance",
   "HIDDEN:temporalio.worker.workflow_sandbox.importer",
   "HIDDEN:temporalio.worker.workflow_sandbox.in_sandbox",
+  "HIDDEN:temporalio.common._streams.system.client",
+  "HIDDEN:temporalio.common._streams.writer_system.client",
   "HIDDEN:**.*_pb2*",
 ]
 project-name = "Temporal Python"
@@ -272,7 +274,7 @@ exclude = [
   "temporalio/api",
   "temporalio/bridge/proto",
   "temporalio/bridge/_visitor.py",
-  "temporalio/contrib/streams/proto",
+  "temporalio/common/_streams/proto",
   "tests/worker/workflow_sandbox/testmodules/proto",
 ]
 

scripts/_proto/temporal/api/streamservice/v1/stream_service.proto

@@ -131,8 +131,7 @@ message StreamRecord {
     CloseStatus close_status = 9;
 }
 
-// Idempotent for one id, one Retention and one auto_close. ALREADY_EXISTS for another
-// Retention or another auto_close.
+// Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention.
 message CreateStreamRequest {
     string namespace = 1;
     // Required from Workflow code.
@@ -141,11 +140,8 @@ message CreateStreamRequest {
     google.protobuf.Duration retention = 3;
     // Workflow only. A retry run's first create: the previous run, whose Records it withdraws.
     string withdraw_run = 4;
-    // Close the Stream with no result when no Record was appended for this long, counted from
-    // the create. Empty never closes on its own. It is separate from Retention and the safety
-    // TTL. The Store applies it on the next access, so no call sees the Stream open after the
-    // period. UNIMPLEMENTED (detail stream_unsupported) on a Store that can't.
-    google.protobuf.Duration auto_close = 5;
+    reserved 5;
+    reserved "auto_close";
     // The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch.
     Producer owner = 6;
     // Close the Stream with the result or the failure of the owner's run chain when the chain
@@ -282,8 +278,8 @@ message DescribeStreamResponse {
     // Zero when the Stream has no Record.
     int64 last_offset = 2;
     google.protobuf.Duration retention = 3;
-    // The auto_close of the create. Empty when the Stream has none.
-    google.protobuf.Duration auto_close = 4;
+    reserved 4;
+    reserved "auto_close";
 }
 
 // The details of a failed Stream call, in the gRPC status details and in the details of the

scripts/_proto/temporal/sdk/streams/v1/extension.proto

@@ -18,7 +18,7 @@ message NativeStreamConfig {
     google.protobuf.Duration retention = 2;
 }
 
-// The settings of the Redis Stream extension. Lang builds it from the plugin options.
+// The settings of the Redis Stream extension. Lang builds it from the backend options.
 message RedisStreamConfig {
     // One URL, or the seed URLs of a cluster. Credentials and TLS come from the URL.
     repeated string urls = 1;
@@ -120,7 +120,7 @@ message FreeWriterRequest {
 message FreeWriterResponse {
 }
 
-// Method `CloseWriters`, asynchronous. Drops every open writer, for example when the plugin's
+// Method `CloseWriters`, asynchronous. Drops every open writer, for example when the backend's
 // client or Worker shuts down.
 message CloseWritersRequest {
 }

scripts/_wit/stream-service.wit

@@ -168,17 +168,15 @@ interface stream-service {
     close-status: option<close-status>,
   }
 
-  /// @nexus.doc "Create a Stream idempotently. ALREADY_EXISTS for another Retention or auto-close."
+  /// @nexus.doc "Create a Stream idempotently. ALREADY_EXISTS for another Retention."
   /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamRequest"
   record create-stream-request {
-    /// @nexus.doc "Required from Workflow code. Empty elsewhere asks the service for an id."
+    /// @nexus.doc "Required on Redis and memory. On native, empty asks the server for an id."
     stream-id: string,
     /// @nexus.doc "The age after which a Record is dropped. Empty is the Store default."
     retention: option<duration>,
     /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws."
     withdraw-run: option<string>,
-    /// @nexus.doc "Close with no result after this long with no append. Empty never auto-closes."
-    auto-close: option<duration>,
     /// @nexus.doc "The Workflow run that owns the Stream. The Workflow transport sets it."
     owner: option<producer>,
     /// @nexus.doc "Close with the owner run chain's result when it closes. Native only."
@@ -329,8 +327,6 @@ interface stream-service {
     /// @nexus.offset "stream-offset"
     last-offset: option<s64>,
     retention: option<duration>,
-    /// @nexus.doc "The auto-close of the create. Empty when the Stream has none."
-    auto-close: option<duration>,
   }
 
   // Its stream-ref is its serialized form, so a handle passes through Workflow input, Activity
@@ -374,7 +370,7 @@ interface stream-service {
       create-wait: option<duration>,
     ) -> read-records-response;
 
-    /// @nexus.doc "Describe the Stream: its status, newest offset, Retention and auto-close."
+    /// @nexus.doc "Describe the Stream: its status, newest offset and Retention."
     /// @nexus.operation "describe-stream"
     describe: func() -> describe-stream-response;
   }

scripts/gen_protos.py

@@ -33,7 +33,7 @@
 
 api_out_dir = base_dir / "temporalio" / "api"
 sdk_out_dir = base_dir / "temporalio" / "bridge" / "proto"
-streams_out_dir = base_dir / "temporalio" / "contrib" / "streams" / "proto"
+streams_out_dir = base_dir / "temporalio" / "common" / "_streams" / "proto"
 
 py_fixes = [
     partial(re.compile(r"from temporal\.api\.").sub, r"from temporalio.api."),
@@ -53,11 +53,11 @@
     ),
     partial(
         re.compile(r"from temporal\.sdk\.streams\.").sub,
-        r"from temporalio.contrib.streams.proto.",
+        r"from temporalio.common._streams.proto.",
     ),
     partial(
         re.compile(r"'__module__' : 'temporal\.sdk\.streams\.").sub,
-        r"'__module__' : 'temporalio.contrib.streams.proto.",
+        r"'__module__' : 'temporalio.common._streams.proto.",
     ),
     partial(
         re.compile(r"'__module__' : 'temporal\.api\.").sub,
@@ -82,7 +82,7 @@
     partial(re.compile(r"temporal\.sdk\.core\.").sub, r"temporalio.bridge.proto."),
     partial(
         re.compile(r"temporal\.sdk\.streams\.").sub,
-        r"temporalio.contrib.streams.proto.",
+        r"temporalio.common._streams.proto.",
     ),
 ]
 
@@ -231,7 +231,7 @@ def generate_protos(output_dir: Path):
     for p in (output_dir / "temporal" / "sdk" / "core").iterdir():
         shutil.rmtree(sdk_out_dir / p.name, ignore_errors=True)
         p.replace(sdk_out_dir / p.name)
-    # The stream extension's own protos go with the contrib package that uses them
+    # Only the Streams package imports the stream extension protos
     streams_out_dir.mkdir(parents=True, exist_ok=True)
     for p in (output_dir / "temporal" / "sdk" / "streams").iterdir():
         shutil.rmtree(streams_out_dir / p.name, ignore_errors=True)

scripts/gen_stream_service.py

@@ -1,10 +1,9 @@
-"""Generates the Stream service bindings of temporalio.contrib.streams.
+"""Generates the Stream service bindings of the Streams API in temporalio.common.
 
 The WIT and the protos are copies of the temporalio-streams crate's, in scripts/_wit and
 scripts/_proto. Run scripts/gen_protos.py first for the proto modules. No nexgen release has the
-offset, long-poll and client features yet, so the script installs nexgen at
-NEX_GEN_REVISION from moetemp/nexgen. NEX_GEN_STREAMS_BIN may name a build of that revision
-instead.
+offset, long-poll and client features, so the script installs nexgen at NEX_GEN_REVISION from
+moetemp/nexgen. Set NEX_GEN_STREAMS_BIN to use a local build of that revision instead.
 """
 
 import os
@@ -30,12 +29,12 @@
     / "v1"
     / "stream_service.proto"
 )
-output_dir = base_dir / "temporalio" / "contrib" / "streams" / "system"
-# The writer is client-only, so it has a caller and its resource class, and no Workflow code.
-writer_output_dir = base_dir / "temporalio" / "contrib" / "streams" / "writer_system"
+output_dir = base_dir / "temporalio" / "common" / "_streams" / "system"
+# Only clients call the writer service, so its bindings have no Workflow code.
+writer_output_dir = base_dir / "temporalio" / "common" / "_streams" / "writer_system"
 
 NEX_GEN_REPOSITORY = "https://github.com/moetemp/nexgen"
-# The head of moe/brook/v3/n8-stream-resource. A branch moves, so the pin is the commit.
+# A commit on the moe/brook/v3/n8-stream-resource branch.
 NEX_GEN_REVISION = "204b566ee996939b165ccd1d18c8c432bfdcc0dd"
 nex_gen_root = base_dir / ".nexgen-streams" / NEX_GEN_REVISION
 
@@ -81,14 +80,14 @@ def generate_stream_service() -> None:
             raise RuntimeError(f"missing generator input: {path}")
     with tempfile.TemporaryDirectory(dir=base_dir) as temp_dir:
         descriptor_path = Path(temp_dir) / "stream_service.bin"
-        # The extension messages live in temporalio.contrib.streams.proto, so the descriptor
-        # names their file by that module path.
+        # The generated extension messages are in temporalio.common._streams.proto, so the
+        # descriptor must name their file by that module path.
         extension_root = Path(temp_dir) / "extension"
         extension_path = (
             extension_root
             / "temporalio"
-            / "contrib"
-            / "streams"
+            / "common"
+            / "_streams"
             / "proto"
             / "v1"
             / "extension.proto"

temporalio/activity.py

@@ -21,6 +21,7 @@
     TYPE_CHECKING,
     Any,
     NoReturn,
+    TypeVar,
     overload,
 )
 
@@ -37,6 +38,9 @@
 
 if TYPE_CHECKING:
     from temporalio.client import Client
+    from temporalio.common import ActivityStream, Creating
+
+_StreamDefinitionT = TypeVar("_StreamDefinitionT")
 
 
 @overload
@@ -112,7 +116,7 @@ class WorkflowTaskAttempt:
     """ID of the ``WorkflowTaskStarted`` event of the run's last completed task. 0 if none."""
 
     failed_tasks: Sequence[int] = ()
-    """``WorkflowTaskStarted`` event IDs of the run's failed and timed out tasks, the last 100."""
+    """``WorkflowTaskStarted`` event IDs of the run's last 100 failed and timed out tasks."""
 
 
 @dataclass(frozen=True)
@@ -167,7 +171,7 @@ class Info:
 
     workflow_task: WorkflowTaskAttempt | None = None
     """For a local activity, the Workflow Task in which the Worker dispatched this execution.
-    Retries that the Worker runs without a new task keep it. None for other activities.
+    Retries that the Worker runs in the same task have the same value. None for other activities.
 
     .. warning::
         This API is experimental.
@@ -515,6 +519,47 @@ def metric_meter() -> temporalio.common.MetricMeter:
     return _Context.current().metric_meter
 
 
+def create_stream(
+    definition: type[_StreamDefinitionT],
+    *,
+    retention: timedelta | None = None,
+    backend: str | None = None,
+) -> Creating[ActivityStream[_StreamDefinitionT]]:
+    """Create a Stream that this Standalone Activity owns, on the native backend.
+
+    The Stream id is ``<activity id>:<activity run id>:<name>``, so a client can get it from the
+    Activity handle with :py:meth:`temporalio.client.ActivityHandle.stream` without a call. The
+    Stream stays open across the Activity's retries, and the server closes it when the Activity
+    ends, with its status and result. So the end of an ``async with`` block doesn't close it.
+    Await the call to get the handle::
+
+        stream = await activity.create_stream(Events)
+        await stream.topic(Events.tokens).publish("hi")
+
+    Only the native backend can store the Stream of a Standalone Activity, since no other
+    backend would close it when the Activity ends.
+
+    .. warning::
+        This API is experimental.
+
+    Args:
+        definition: The Stream definition class. Its name is the Stream name.
+        retention: The age after which a Record is deleted. A closed Stream is kept for one
+            Retention after its close. Defaults to the backend's Retention.
+        backend: The name of a native backend. Defaults to ``native``.
+
+    Raises:
+        StreamConfigError: The Activity runs for a Workflow, which owns its Streams, or the
+            backend isn't native. No call is made.
+        StreamClosedError: The Stream was created and closed before.
+    """
+    from temporalio.common._streams import _activity_stream
+
+    return _activity_stream.create_stream(
+        definition, retention=retention, backend=backend
+    )
+
+
 class LoggerAdapter(logging.LoggerAdapter):
     """Adapter that adds details to the log about the running activity.