feat: add the Stream runtime API with stubbed calls

crates/protos/protos/local/temporal/sdk/core/streams/streams.proto

@@ -6,8 +6,23 @@ option ruby_package = "Temporalio::Internal::Bridge::Api::Streams";
 import "google/protobuf/duration.proto";
 import "temporal/api/stream/v1/message.proto";
 
-// The writer messages that `temporalio-streams` adds to the five Stream service operations. A
-// bridge passes them as bytes to the runtime's byte calls, so it decodes no proto.
+// The writer messages that `temporalio-streams` adds to the five Stream service operations, and
+// the config of its memory runtime. A bridge passes them as bytes to the runtime's byte calls, so
+// it decodes no proto.
+
+// The settings of the memory Stream runtime, for tests. Each field has the number of the same
+// field in the config of the other Stores, such as Redis, so a backend can send that config here
+// too.
+message MemoryStreamConfig {
+    // The default Retention of a new Stream. Empty is 7 days.
+    google.protobuf.Duration retention = 4;
+    // How long an open Stream stays after its last write. Empty keeps an open Stream until its
+    // close.
+    google.protobuf.Duration last_write_ttl = 5;
+    // The backend name that lang registered this runtime under. The runtime sets it on each
+    // StreamRef that it builds, such as the one in a create response.
+    string backend = 10;
+}
 
 // The open of a writer, through the runtime's `open_writer`. A writer is for an Activity, a Local
 // Activity or a client Producer, and it keeps its epoch and `sequence` order. A writer of a Local

crates/sdk-core/CHANGELOG.md

@@ -33,6 +33,10 @@ relevant information.
 
 ## Unreleased
 
+### Added
+* A new experimental crate, `temporalio-streams`, has the Stream runtime that the SDKs link
+  into their bridges. Core itself has no Stream code.
+
 ## [0.11.0] - 2026-11-09
 
 ### Added

crates/streams/Cargo.toml

@@ -22,5 +22,8 @@ path = "../common"
 version = "~1.2.0"
 default-features = false
 
+[dev-dependencies]
+tokio = { version = "1.47", features = ["macros", "rt-multi-thread", "time"] }
+
 [lints]
 workspace = true

crates/streams/src/dispatch.rs

@@ -0,0 +1,92 @@
+//! The byte form of the runtime's and the writer's calls, for bridges. A bridge passes the encoded
+//! request and gets the encoded response, so it decodes no proto.
+
+use crate::{
+    Reason, StreamError, StreamResult, StreamRuntime, StreamWriter, proto::WriteRecordsRequest,
+};
+use prost::Message;
+
+pub(crate) fn decode<M: Message + Default>(input: &[u8]) -> StreamResult<M> {
+    M::decode(input).map_err(|error| StreamError::invalid(format!("The request is bad: {error}")))
+}
+
+impl StreamRuntime {
+    /// Runs one of the five Stream service operations on encoded input, by its name, and returns
+    /// the encoded output. The bridge raises an error with [StreamError::code] and
+    /// [StreamError::details].
+    pub async fn call(&self, method: &str, input: &[u8]) -> StreamResult<Vec<u8>> {
+        let output = match method {
+            "CreateStream" => self.create_stream(decode(input)?).await?.encode_to_vec(),
+            "AppendRecords" => self.append_records(decode(input)?).await?.encode_to_vec(),
+            "CloseStream" => self.close_stream(decode(input)?).await?.encode_to_vec(),
+            "ReadRecords" => self.read_records(decode(input)?).await?.encode_to_vec(),
+            "DescribeStream" => self.describe_stream(decode(input)?).await?.encode_to_vec(),
+            _ => {
+                return Err(StreamError::new(
+                    Reason::Unsupported,
+                    format!("Unknown method {method}"),
+                ));
+            }
+        };
+        Ok(output)
+    }
+
+    /// Opens a writer from an encoded `OpenWriterRequest`.
+    pub async fn open_writer_bytes(&self, input: &[u8]) -> StreamResult<StreamWriter> {
+        self.open_writer(decode(input)?).await
+    }
+}
+
+impl StreamWriter {
+    /// Writes an encoded `WriteRecordsRequest`, and returns the encoded `WriteRecordsResponse`.
+    pub async fn write_bytes(&self, input: &[u8]) -> StreamResult<Vec<u8>> {
+        let request = decode::<WriteRecordsRequest>(input)?;
+        Ok(self.write(request).await?.encode_to_vec())
+    }
+
+    /// Closes the writer, and returns the encoded `CloseWriterResponse`.
+    pub async fn close_bytes(&self) -> StreamResult<Vec<u8>> {
+        Ok(self.close().await?.encode_to_vec())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::store::Store;
+    use std::sync::Arc;
+
+    struct NoStore;
+
+    impl Store for NoStore {}
+
+    const METHODS: [&str; 5] = [
+        "CreateStream",
+        "AppendRecords",
+        "CloseStream",
+        "ReadRecords",
+        "DescribeStream",
+    ];
+
+    fn runtime() -> Arc<StreamRuntime> {
+        StreamRuntime::new("test", "", Arc::new(NoStore), None)
+    }
+
+    // Each method decodes its request before anything else. A typo in a method name would make it
+    // unknown, which is unsupported, so the loop also checks the names that a bridge calls.
+    #[tokio::test]
+    async fn bad_bytes_are_invalid_and_an_unknown_method_is_unsupported() {
+        let runtime = runtime();
+        for method in METHODS {
+            let error = runtime.call(method, &[0xff]).await.unwrap_err();
+            assert_eq!(error.reason, Reason::Invalid, "{method}");
+        }
+        let error = runtime.open_writer_bytes(&[0xff]).await.unwrap_err();
+        assert_eq!(error.reason, Reason::Invalid);
+        let error = runtime.call("Nope", &[]).await.unwrap_err();
+        assert_eq!(
+            (error.reason, error.message.as_str()),
+            (Reason::Unsupported, "Unknown method Nope")
+        );
+    }
+}

crates/streams/src/lib.rs

@@ -6,9 +6,15 @@
 //! client and calls its typed methods, or its byte form with an operation name and proto bytes.
 //! Core has no hook for it. A Workflow makes each call from a Local Activity.
 
+mod dispatch;
 mod error;
+mod runtime;
+pub mod store;
+mod writer;
 
 pub use error::{Reason, StreamError, StreamResult};
+pub use runtime::{MemoryStreamRuntime, StreamRuntime};
+pub use writer::StreamWriter;
 
 /// The protos that lang and the crate exchange. `temporalio-protos` generates them, so a value
 /// needs no conversion between the crates.

crates/streams/src/runtime.rs

@@ -0,0 +1,142 @@
+//! The Stream runtime, a plain object that lang's bridge builds per client.
+
+use crate::{
+    Reason, StreamError, StreamResult, StreamWriter,
+    proto::{
+        AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, CloseStreamResponse,
+        CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse,
+        MemoryStreamConfig, OpenWriterRequest, ReadRecordsRequest, ReadRecordsResponse,
+    },
+    store::Store,
+};
+use prost::Message;
+use std::sync::Arc;
+use temporalio_common::telemetry::metrics::TemporalMeter;
+
+/// The Stream runtime on one Store. It serves the five Stream service operations and the writers.
+///
+/// A bridge builds one for each client with a Stream backend, and that client's Workers share it.
+/// A Store crate, such as `temporalio-streams-redis`, builds it with its own Store.
+pub struct StreamRuntime {
+    name: &'static str,
+    backend: String,
+}
+
+impl std::fmt::Debug for StreamRuntime {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("StreamRuntime")
+            .field("name", &self.name)
+            .field("backend", &self.backend)
+            .finish_non_exhaustive()
+    }
+}
+
+pub(crate) fn not_implemented() -> StreamError {
+    StreamError::new(Reason::Unsupported, "Not implemented")
+}
+
+impl StreamRuntime {
+    /// Builds a runtime on `store`. `name` identifies the runtime in logs. `backend` is the name
+    /// that lang registered the runtime under, and `meter` is lang's runtime meter, if set. No
+    /// call of this runtime uses the Store or records a metric, so the runtime keeps neither.
+    pub fn new(
+        name: &'static str,
+        backend: impl Into<String>,
+        store: Arc<dyn Store>,
+        meter: Option<TemporalMeter>,
+    ) -> Arc<Self> {
+        let _ = (store, meter);
+        Arc::new(Self {
+            name,
+            backend: backend.into(),
+        })
+    }
+
+    /// A name that identifies the runtime in logs.
+    pub fn name(&self) -> &str {
+        self.name
+    }
+
+    /// Creates a Stream, or gets it with USE_EXISTING.
+    pub async fn create_stream(
+        &self,
+        request: CreateStreamRequest,
+    ) -> StreamResult<CreateStreamResponse> {
+        let _ = request;
+        Err(not_implemented())
+    }
+
+    /// Appends one batch as one Producer.
+    pub async fn append_records(
+        &self,
+        request: AppendRecordsRequest,
+    ) -> StreamResult<AppendRecordsResponse> {
+        let _ = request;
+        Err(not_implemented())
+    }
+
+    /// Closes a Stream.
+    pub async fn close_stream(
+        &self,
+        request: CloseStreamRequest,
+    ) -> StreamResult<CloseStreamResponse> {
+        let _ = request;
+        Err(not_implemented())
+    }
+
+    /// Reads one page of Records after an offset.
+    pub async fn read_records(
+        &self,
+        request: ReadRecordsRequest,
+    ) -> StreamResult<ReadRecordsResponse> {
+        let _ = request;
+        Err(not_implemented())
+    }
+
+    /// Gets the status and the newest offset of a Stream.
+    pub async fn describe_stream(
+        &self,
+        request: DescribeStreamRequest,
+    ) -> StreamResult<DescribeStreamResponse> {
+        let _ = request;
+        Err(not_implemented())
+    }
+
+    /// Opens a writer for an Activity, a Local Activity or a client Producer. The bridge holds the
+    /// writer, and dropping it releases it.
+    pub async fn open_writer(&self, request: OpenWriterRequest) -> StreamResult<StreamWriter> {
+        let _ = request;
+        Err(not_implemented())
+    }
+}
+
+/// The Stream runtime for tests.
+#[derive(Debug)]
+pub struct MemoryStreamRuntime;
+
+impl MemoryStreamRuntime {
+    /// Takes an encoded [MemoryStreamConfig]. Its field numbers are those of the same fields in
+    /// the config of another Store, such as Redis, so a backend can pass that config too. A bad
+    /// config fails with [Reason::Invalid]. The crate has no memory Store, so a good config fails
+    /// with [Reason::Unsupported].
+    pub fn from_config_bytes(
+        config: &[u8],
+        meter: Option<TemporalMeter>,
+    ) -> StreamResult<Arc<StreamRuntime>> {
+        MemoryStreamConfig::decode(config)
+            .map_err(|error| StreamError::invalid(format!("The config is bad: {error}")))?;
+        let _ = meter;
+        Err(not_implemented())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_bad_memory_config_is_invalid() {
+        let error = MemoryStreamRuntime::from_config_bytes(&[0xff], None).unwrap_err();
+        assert_eq!(error.reason, Reason::Invalid);
+    }
+}

crates/streams/src/store.rs

@@ -0,0 +1,5 @@
+//! The Store under a Stream runtime.
+
+/// Keeps the Records and the state of Streams. A [crate::StreamRuntime] serves the Stream service
+/// on one Store, which a bridge or a Store crate, such as `temporalio-streams-redis`, gives it.
+pub trait Store: Send + Sync + 'static {}

crates/streams/src/writer.rs

@@ -0,0 +1,30 @@
+//! Writers for Activities and clients. The bridge holds each writer as an object, and dropping it
+//! releases it.
+
+use crate::{
+    StreamResult,
+    proto::{CloseWriterResponse, WriteRecordsRequest, WriteRecordsResponse},
+};
+
+/// Has no value, so no writer exists until the runtime opens one.
+#[derive(Debug)]
+enum Never {}
+
+/// A writer of one Producer epoch on one Stream, for an Activity, a Local Activity or a client.
+/// The bridge holds it, and dropping it releases it.
+#[derive(Debug)]
+pub struct StreamWriter(Never);
+
+impl StreamWriter {
+    /// Appends Records at once and answers with the offsets of the first and the last one.
+    pub async fn write(&self, request: WriteRecordsRequest) -> StreamResult<WriteRecordsResponse> {
+        let _ = request;
+        match self.0 {}
+    }
+
+    /// Closes the writer. It waits for a write in flight, then answers with the offset of the
+    /// newest Record that the writer wrote. It doesn't close the Stream.
+    pub async fn close(&self) -> StreamResult<CloseWriterResponse> {
+        match self.0 {}
+    }
+}