diff --git a/Cargo.toml b/Cargo.toml
index 76bdd46f6..e44afa24c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -8,6 +8,7 @@ members = [
     "crates/macros",
     "crates/sdk",
     "crates/workflow",
+    "crates/streams",
     "crates/sdk-core-c-bridge",
     "crates/changelog-release-notes",
 ]
diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md
index 29d2f5bd3..c4887f965 100644
--- a/crates/sdk-core/CHANGELOG.md
+++ b/crates/sdk-core/CHANGELOG.md
@@ -40,6 +40,8 @@ relevant information.
   name the run's last completed workflow task, and `failed_tasks`, the started event ids of the
   run's last 100 failed or timed out workflow tasks. Retries that Core runs locally keep the
   value. Core leaves it unset for other activities.
+* A new experimental crate, `temporalio-streams`, has the Stream runtime that the SDKs link
+  into their bridges. Core itself has no Stream code.
 
 ### Security
 * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency
diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml
new file mode 100644
index 000000000..3153fccf2
--- /dev/null
+++ b/crates/streams/Cargo.toml
@@ -0,0 +1,36 @@
+[package]
+name = "temporalio-streams"
+version = "0.1.0"
+edition = "2024"
+rust-version = "1.88.0"
+authors = ["Temporal Technologies Inc. <sdk@temporal.io>"]
+license-file = { workspace = true }
+description = "The experimental Stream runtime that Temporal SDK bridges link"
+homepage = "https://temporal.io/"
+repository = "https://github.com/temporalio/sdk-rust"
+keywords = ["temporal", "workflow", "streams"]
+categories = ["development-tools"]
+
+[dependencies]
+prost = { workspace = true }
+prost-types = { workspace = true }
+thiserror = { workspace = true }
+tonic = { workspace = true }
+
+[dependencies.temporalio-common]
+path = "../common"
+version = "~1.1.0"
+default-features = false
+
+# Only for its descriptor set, which the build script reads for the API protos that the Stream
+# protos import.
+[dependencies.temporalio-protos]
+path = "../protos"
+version = "~0.10.0"
+
+[build-dependencies]
+protox = "0.9.1"
+tonic-prost-build = { workspace = true }
+
+[lints]
+workspace = true
diff --git a/crates/streams/README.md b/crates/streams/README.md
new file mode 100644
index 000000000..d1102dd09
--- /dev/null
+++ b/crates/streams/README.md
@@ -0,0 +1,17 @@
+# temporalio-streams
+
+Experimental. The Stream runtime that the Temporal SDKs link into their bridges. A bridge builds
+the runtime per client, as a plain object, and calls it with a method name and proto bytes. A
+Workflow makes each Stream call from a Local Activity. Core has no Stream code.
+
+## Protos
+
+- `protos/temporal/api/stream/v1/message.proto`,
+  `protos/temporal/api/streamservice/v1/request_response.proto` and
+  `protos/temporal/api/enums/v1/stream.proto` are the Stream protos of the API proposal, byte for
+  byte. The API protos of this repo don't have them yet, so the crate generates them. Lang
+  generates the same files. They import API protos that the build reads from the descriptor set
+  of `temporalio-protos`, and that map to the types of `temporalio-common`. So the packaged crate
+  needs no file outside itself.
+- `protos/temporal/sdk/streams/v1/runtime.proto` has the crate's own calls, such as the writer
+  calls.
diff --git a/crates/streams/build.rs b/crates/streams/build.rs
new file mode 100644
index 000000000..289758e82
--- /dev/null
+++ b/crates/streams/build.rs
@@ -0,0 +1,50 @@
+use protox::{
+    Compiler,
+    file::{ChainFileResolver, DescriptorSetFileResolver, GoogleFileResolver, IncludeFileResolver},
+};
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    println!("cargo:rerun-if-changed=protos");
+    let protos = [
+        "protos/temporal/api/enums/v1/stream.proto",
+        "protos/temporal/api/stream/v1/message.proto",
+        "protos/temporal/api/streamservice/v1/request_response.proto",
+        "protos/temporal/sdk/streams/v1/runtime.proto",
+    ];
+    // The build reads the API protos that these import from the descriptor set of
+    // `temporalio-protos`, so the packaged crate needs no file outside itself.
+    let api = std::env::var("DEP_TEMPORALIO_PROTOS_DESCRIPTOR_PATH")?;
+    println!("cargo:rerun-if-changed={api}");
+    let mut resolver = ChainFileResolver::new();
+    resolver.add(IncludeFileResolver::new("protos".into()));
+    resolver.add(DescriptorSetFileResolver::decode(
+        std::fs::read(&api)?.as_slice(),
+    )?);
+    resolver.add(GoogleFileResolver::new());
+    let mut compiler = Compiler::with_file_resolver(resolver);
+    compiler
+        .include_source_info(true)
+        .include_imports(true)
+        .open_files(protos)?;
+    // protox compiles the protos in Rust, so the crate builds with no protoc, as Core's
+    // `vendored-protox` build does.
+    let descriptor_file = std::path::PathBuf::from(std::env::var("OUT_DIR")?).join("streams.bin");
+    std::fs::write(&descriptor_file, compiler.encode_file_descriptor_set())?;
+    tonic_prost_build::configure()
+        .build_server(false)
+        .build_client(false)
+        .skip_protoc_run()
+        .file_descriptor_set_path(&descriptor_file)
+        // Use the Core types for the Temporal API messages, so that values need no conversion
+        // between the crates. Not `.temporal.api.enums`, since Core lacks the Stream enums.
+        .extern_path(
+            ".temporal.api.common",
+            "::temporalio_common::protos::temporal::api::common",
+        )
+        .extern_path(
+            ".temporal.api.failure",
+            "::temporalio_common::protos::temporal::api::failure",
+        )
+        .compile_protos(&protos, &["protos"])?;
+    Ok(())
+}
diff --git a/crates/streams/protos/temporal/api/enums/v1/stream.proto b/crates/streams/protos/temporal/api/enums/v1/stream.proto
new file mode 100644
index 000000000..b64f7d5d4
--- /dev/null
+++ b/crates/streams/protos/temporal/api/enums/v1/stream.proto
@@ -0,0 +1,90 @@
+syntax = "proto3";
+
+package temporal.api.enums.v1;
+
+option go_package = "go.temporal.io/api/enums/v1;enums";
+option java_package = "io.temporal.api.enums.v1";
+option java_multiple_files = true;
+option java_outer_classname = "StreamProto";
+option ruby_package = "Temporalio::Api::Enums::V1";
+option csharp_namespace = "Temporalio.Api.Enums.V1";
+
+// Experimental. Might change in the future.
+enum StreamProducerKind {
+    STREAM_PRODUCER_KIND_UNSPECIFIED = 0;
+    STREAM_PRODUCER_KIND_WORKFLOW = 1;
+    STREAM_PRODUCER_KIND_LOCAL_ACTIVITY = 2;
+    STREAM_PRODUCER_KIND_ACTIVITY = 3;
+    STREAM_PRODUCER_KIND_CLIENT = 4;
+    STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY = 5;
+}
+
+// Experimental. Might change in the future.
+// How a Stream closed, so a reader tells a terminated or timed out owner from a failed one
+// without parsing the failure.
+// (-- api-linter: core::0216::synonyms=disabled
+//     aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --)
+enum StreamCloseStatus {
+    STREAM_CLOSE_STATUS_UNSPECIFIED = 0;
+    STREAM_CLOSE_STATUS_COMPLETED = 1;
+    STREAM_CLOSE_STATUS_FAILED = 2;
+    STREAM_CLOSE_STATUS_CANCELED = 3;
+    // TERMINATED and TIMED_OUT come only from the owner's end, never from a caller.
+    STREAM_CLOSE_STATUS_TERMINATED = 4;
+    STREAM_CLOSE_STATUS_TIMED_OUT = 5;
+}
+
+// Experimental. Might change in the future.
+// (-- api-linter: core::0216::synonyms=disabled
+//     aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --)
+enum StreamStatus {
+    STREAM_STATUS_UNSPECIFIED = 0;
+    STREAM_STATUS_OPEN = 1;
+    STREAM_STATUS_CLOSED = 2;
+    STREAM_STATUS_EXPIRED = 3;
+}
+
+// Experimental. Might change in the future.
+enum StreamReadStart {
+    STREAM_READ_START_UNSPECIFIED = 0;
+    STREAM_READ_START_BEGINNING = 1;
+    STREAM_READ_START_END = 2;
+}
+
+// Experimental. Might change in the future.
+// Why a Stream call failed. Each reason has one gRPC status code.
+enum StreamErrorReason {
+    STREAM_ERROR_REASON_UNSPECIFIED = 0;
+    // Nobody created the Stream, even after `create_wait`. NOT_FOUND.
+    STREAM_ERROR_REASON_NOT_FOUND = 1;
+    // The Stream's Retention or safety TTL passed. NOT_FOUND.
+    STREAM_ERROR_REASON_EXPIRED = 2;
+    // The Stream is closed. FAILED_PRECONDITION.
+    STREAM_ERROR_REASON_CLOSED = 3;
+    // The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a
+    // Workflow retry withdrew its run. FAILED_PRECONDITION.
+    STREAM_ERROR_REASON_FENCED = 4;
+    // The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION.
+    STREAM_ERROR_REASON_SEQUENCE = 5;
+    // The batch has the `sequence` of the last one of its epoch, and other content.
+    // FAILED_PRECONDITION.
+    STREAM_ERROR_REASON_SEQ_DIVERGENT = 6;
+    // A create names another owner than the Stream has, or the owner doesn't allow this close.
+    // FAILED_PRECONDITION.
+    STREAM_ERROR_REASON_NOT_OWNER = 7;
+    // A newer format version wrote the Stream. FAILED_PRECONDITION.
+    STREAM_ERROR_REASON_FORMAT_TOO_NEW = 8;
+    // The offset is past the newest offset + 1, for example after a failover lost the newest
+    // Records. A read from it works again once the Stream has Records up to it. OUT_OF_RANGE.
+    STREAM_ERROR_REASON_OFFSET = 9;
+    // A Record or a batch is over the Store's limit. INVALID_ARGUMENT.
+    STREAM_ERROR_REASON_TOO_LARGE = 10;
+    // The request is malformed. INVALID_ARGUMENT.
+    STREAM_ERROR_REASON_INVALID = 11;
+    // The Store failed or did not answer. UNAVAILABLE.
+    STREAM_ERROR_REASON_STORAGE = 12;
+    // No writer has the id. The SDK's in-process writer only. NOT_FOUND.
+    STREAM_ERROR_REASON_UNKNOWN_WRITER = 13;
+    // The Store lacks a capability that the call needs. UNIMPLEMENTED.
+    STREAM_ERROR_REASON_UNSUPPORTED = 14;
+}
diff --git a/crates/streams/protos/temporal/api/stream/v1/message.proto b/crates/streams/protos/temporal/api/stream/v1/message.proto
new file mode 100644
index 000000000..e4984081e
--- /dev/null
+++ b/crates/streams/protos/temporal/api/stream/v1/message.proto
@@ -0,0 +1,127 @@
+syntax = "proto3";
+
+package temporal.api.stream.v1;
+
+option go_package = "go.temporal.io/api/stream/v1;stream";
+option java_package = "io.temporal.api.stream.v1";
+option java_multiple_files = true;
+option java_outer_classname = "MessageProto";
+option ruby_package = "Temporalio::Api::Stream::V1";
+option csharp_namespace = "Temporalio.Api.Stream.V1";
+
+import "google/protobuf/timestamp.proto";
+
+import "temporal/api/common/v1/message.proto";
+import "temporal/api/enums/v1/stream.proto";
+import "temporal/api/failure/v1/message.proto";
+
+// Experimental. Might change in the future.
+// Identifies one Stream. The id is opaque to callers.
+message StreamRef {
+    string id = 1;
+    // Unset for a client's Stream. For a Workflow, business_id is the Workflow ID and run_id is
+    // the first execution run id of its run chain. For a Standalone Activity, business_id is the
+    // activity ID and run_id is its run id. A create and a close of a Workflow-owned Stream must
+    // name its owner.
+    temporal.api.common.v1.Execution owner = 2;
+    // The name of the backend that holds the Stream, as the Worker and client config register
+    // it, so every process routes a ref it receives. Empty is the default backend. The Store
+    // never reads it.
+    string backend = 3;
+}
+
+// Experimental. Might change in the future.
+// The ordered attempt of a Producer. Compared in this order: task_started_event_id,
+// task_attempt, task_started_time, attempt_scheduled_time, attempt. A field that a kind does
+// not use is empty and compares as zero. A timestamp compares seconds, then nanos.
+message ProducerEpoch {
+    // The Workflow Task of a Workflow, or the one that dispatched a Local Activity.
+    int64 task_started_event_id = 1;
+    int32 task_attempt = 2;
+    // Tells a reused started event id apart, for example after a dropped speculative task.
+    google.protobuf.Timestamp task_started_time = 5;
+    // When the server scheduled an Activity's attempt. A reset makes it later.
+    google.protobuf.Timestamp attempt_scheduled_time = 3;
+    // Activity, Local Activity and client attempt.
+    int32 attempt = 4;
+}
+
+// Experimental. Might change in the future.
+message Producer {
+    temporal.api.enums.v1.StreamProducerKind kind = 1;
+    string id = 2;
+    ProducerEpoch epoch = 3;
+    // The run of the Producer, so nobody parses the id. Empty for a client.
+    string run_id = 4;
+}
+
+// Experimental. Might change in the future.
+message WorkflowTaskEpoch {
+    int64 started_event_id = 1;
+    int32 attempt = 2;
+    google.protobuf.Timestamp started_time = 3;
+}
+
+// Experimental. Might change in the future.
+message AppendRecord {
+    reserved 1;
+    reserved "topic";
+    temporal.api.common.v1.Payload body = 2;
+}
+
+// Experimental. Might change in the future.
+message Supersession {
+    temporal.api.enums.v1.StreamProducerKind producer_kind = 1;
+    string producer_id = 2;
+    ProducerEpoch epoch = 3;
+    // Workflow only.
+    WorkflowTaskEpoch previous_completed_task = 4;
+    repeated int64 failed_tasks = 5;
+    // Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn.
+    string withdraw_run = 6;
+    // The run of the Producer, so nobody parses the id. Empty for a client.
+    string run_id = 7;
+}
+
+// Experimental. Might change in the future.
+// One Record that a Producer appended.
+message StreamData {
+    reserved 1;
+    reserved "topic";
+    Producer producer = 2;
+    temporal.api.common.v1.Payload body = 3;
+}
+
+// Experimental. Might change in the future.
+// The close of a Stream. A Stream has at most one, and a read returns it last.
+message StreamClose {
+    // The Producer that closed the Stream.
+    Producer producer = 1;
+    temporal.api.common.v1.Payload result = 2;
+    temporal.api.failure.v1.Failure failure = 3;
+    // The result or the failure stays next to it.
+    temporal.api.enums.v1.StreamCloseStatus status = 4;
+}
+
+// Experimental. Might change in the future.
+// An offset is ordered and unique in one Stream, and it is sparse. Offsets have gaps, so they
+// don't count Records. Zero is never an offset. A read names its Stream, so the read checks the
+// offset against that Stream.
+message StreamRecord {
+    int64 offset = 1;
+    // A read item is a Record, a Superseded item or the close.
+    oneof item {
+        StreamData data = 2;
+        // The first batch of a higher epoch carries it. It withdraws the earlier Records that its
+        // facts name, and a reader drops them.
+        Supersession superseded = 3;
+        StreamClose close = 4;
+    }
+}
+
+// Experimental. Might change in the future.
+// The details of a failed Stream call, in the gRPC status details and in the details of the
+// application failure of a Nexus operation.
+message StreamErrorDetails {
+    temporal.api.enums.v1.StreamErrorReason reason = 1;
+}
diff --git a/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto b/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto
new file mode 100644
index 000000000..a2bfeba35
--- /dev/null
+++ b/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto
@@ -0,0 +1,190 @@
+syntax = "proto3";
+
+package temporal.api.streamservice.v1;
+
+option go_package = "go.temporal.io/api/streamservice/v1;streamservice";
+option java_package = "io.temporal.api.streamservice.v1";
+option java_multiple_files = true;
+option java_outer_classname = "RequestResponseProto";
+option ruby_package = "Temporalio::Api::StreamService::V1";
+option csharp_namespace = "Temporalio.Api.StreamService.V1";
+
+// The messages of the Stream service. An SDK Store (Redis or memory) and the server answer the
+// same five operations with them. A field that only the server reads says so.
+
+import "google/protobuf/duration.proto";
+
+import "temporal/api/common/v1/message.proto";
+import "temporal/api/enums/v1/stream.proto";
+import "temporal/api/failure/v1/message.proto";
+import "temporal/api/stream/v1/message.proto";
+
+// Experimental. Might change in the future.
+// Creates the Stream, or gets it when it exists. Then created is false, and the Stream keeps its
+// own Retention. It fails with CLOSED on a closed Stream, EXPIRED on an expired one, and NOT_OWNER
+// on one with another owner.
+// (-- api-linter: core::0133=disabled
+//     aip.dev/not-precedent: The request names a Stream by namespace and stream_id. --)
+message CreateStreamRequest {
+    reserved 5;
+    reserved "auto_close";
+    string namespace = 1;
+    // Required, and never empty.
+    string stream_id = 2;
+    // The age after which a Record is dropped. Empty is the Store default.
+    google.protobuf.Duration retention = 3;
+    // Workflow only. The previous run on a retry run's first create, to withdraw its Records.
+    string withdraw_run = 4;
+    // The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch. It needs a
+    // ref whose owner names the run's chain.
+    temporal.api.stream.v1.Producer owner = 6;
+    // Only the server reads it. The server closes the Stream with the outcome of the owner's run
+    // chain, terminate and timeout included. It needs owner, and a ref with a Workflow owner.
+    // (-- api-linter: core::0140::prepositions=disabled
+    //     aip.dev/not-precedent: "with" ties the close to the owner's run chain. --)
+    bool close_with_run = 7;
+    // The ref of the Stream, with stream_id as its id. A Workflow-owned Stream sets the ref's
+    // owner, which names the owner's run chain.
+    temporal.api.stream.v1.StreamRef ref = 8;
+    // Only the server reads it. The server checks that the token is the Activity's current
+    // attempt, so a caller can't put a Stream into a Standalone Activity it isn't.
+    bytes task_token = 9;
+}
+
+// Experimental. Might change in the future.
+message CreateStreamResponse {
+    reserved 3;
+    reserved "status";
+    // The ref, with the owner of a Workflow-owned Stream.
+    temporal.api.stream.v1.StreamRef stream = 1;
+    bool created = 2;
+}
+
+// Experimental. Might change in the future.
+// FAILED_PRECONDITION if closed, fenced, or if the Stream's meta has a higher format version v
+// (detail stream_format_too_new). NOT_FOUND if absent.
+message AppendRecordsRequest {
+    string namespace = 1;
+    temporal.api.stream.v1.StreamRef stream = 2;
+    // Required outside a Workflow. In a Workflow the transport sets it.
+    temporal.api.stream.v1.Producer producer = 3;
+    // The batch number in this epoch. With the Producer and epoch it is the dedupe key. Zero takes
+    // the next number of the epoch. Such a batch has no dedupe, so a retry writes it again.
+    int64 sequence = 4;
+    repeated temporal.api.stream.v1.AppendRecord records = 5;
+    // The run's last completed Workflow Task (started_event_id 0 before the first) and its failed
+    // tasks, from a Local Activity's start. The batch withdraws the run's Local Activity Records
+    // whose task never completed.
+    temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6;
+    repeated int64 failed_tasks = 7;
+    // Local Activity only. The previous run on a retry run's first batch, to withdraw its Records.
+    string withdraw_run = 8;
+    // Only the server reads it, on a Workflow Producer's batch. The batch withdraws no Workflow
+    // Task and keeps no task facts, but withdraw_run still applies.
+    bool defer_withdrawal = 9;
+    // Only the server reads it. The server checks the Producer against the Activity's current
+    // attempt, so nobody else writes as that Activity.
+    bytes task_token = 10;
+}
+
+// Experimental. Might change in the future.
+message AppendRecordsResponse {
+    int64 first_offset = 1;
+    int64 last_offset = 2;
+}
+
+// Experimental. Might change in the future.
+message CloseStreamRequest {
+    string namespace = 1;
+    temporal.api.stream.v1.StreamRef stream = 2;
+    temporal.api.stream.v1.Producer producer = 3;
+    temporal.api.common.v1.Payload result = 4;
+    // The close status comes from it. No failure is COMPLETED. A failure is CANCELED when it has
+    // canceled_failure_info, or when it is an Activity, child Workflow or Nexus Operation failure
+    // whose cause has canceled_failure_info. Any other failure is FAILED.
+    temporal.api.failure.v1.Failure failure = 5;
+    // Only the server reads these. They are the run's task facts, as on the append.
+    temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6;
+    repeated int64 failed_tasks = 7;
+    // Only the server reads it. The previous run on a retry run's close, to withdraw its Records.
+    string withdraw_run = 8;
+    reserved 9;
+    reserved "status";
+    // Only the server reads it. It checks the token as on the append.
+    bytes task_token = 10;
+}
+
+// Experimental. Might change in the future.
+message CloseStreamResponse {
+}
+
+// Experimental. Might change in the future.
+message ReadRecordsRequest {
+    string namespace = 1;
+    temporal.api.stream.v1.StreamRef stream = 2;
+    // Start at this offset. Zero uses start. OUT_OF_RANGE if past the newest offset + 1.
+    // (-- api-linter: core::0140::prepositions=disabled
+    //     aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --)
+    int64 from_offset = 3;
+    temporal.api.enums.v1.StreamReadStart start = 4;
+    reserved 5, 10;
+    reserved "topics", "topic_prefixes";
+    // The most items in one page, Superseded items included. Zero is 1000. A Superseded item and
+    // the Record after it stay in one page, so a page of one Record can hold two items.
+    int32 max_records = 6;
+    google.protobuf.Duration wait = 7;
+    // How long to wait for a Stream that nobody created yet. Empty fails at once with
+    // NOT_FOUND. At most 60 s.
+    google.protobuf.Duration create_wait = 8;
+    // The exclusive end of the read. The read scans no entry at or after it, the close Record
+    // included, so a replay reads the same range. Zero means no end. Only the server serves it,
+    // and a Store refuses it.
+    // (-- api-linter: core::0140::prepositions=disabled
+    //     aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --)
+    int64 to_offset = 11;
+    // Park as a normal read, then answer with no Records and next_offset = from_offset, when a
+    // Record at or after from_offset exists, when the Stream closes (done), or when wait ends. Its
+    // wait may be days, since a parked read adds nothing to History while it waits. Only the
+    // server serves it, and a Store refuses it.
+    bool wake_only = 12;
+    // Read the newest `last` Records, oldest first. It needs no from_offset and no start. Not with
+    // wake_only, and not in Workflow code, where replay needs a fixed range.
+    int32 last = 13;
+    // The most Record bytes in one page, so a page fits the caller's limit, such as a History
+    // marker. A larger Record still comes alone, so the read always moves on. Zero is no cap.
+    int64 max_bytes = 14;
+    // The lowest offset this run read on replay. A reset run copies the earlier run's reads into
+    // its History, so its floor must cover them too, not only its first live page. Only the
+    // server reads it.
+    // (-- api-linter: core::0140::prepositions=disabled
+    //     aip.dev/not-precedent: The floor is the lowest from_offset of the run's reads. --)
+    int64 floor_from = 15;
+}
+
+// Experimental. Might change in the future.
+message ReadRecordsResponse {
+    repeated temporal.api.stream.v1.StreamRecord records = 1;
+    // The from_offset of the next read, which is the newest offset read + 1.
+    int64 next_offset = 2;
+    bool done = 3;
+    // This reader's floor keeps more than half of the Stream's byte cap. The writer fails when the
+    // cap fills, and only the reader's continue-as-new frees it. Only the server sets it.
+    bool continue_as_new_suggested = 4;
+}
+
+// Experimental. Might change in the future.
+message DescribeStreamRequest {
+    string namespace = 1;
+    temporal.api.stream.v1.StreamRef stream = 2;
+}
+
+// Experimental. Might change in the future.
+message DescribeStreamResponse {
+    reserved 4;
+    reserved "auto_close";
+    temporal.api.enums.v1.StreamStatus status = 1;
+    // The newest offset, of a Record, a Superseded item or the close. Zero when the Stream has
+    // none.
+    int64 last_offset = 2;
+    google.protobuf.Duration retention = 3;
+}
diff --git a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto
new file mode 100644
index 000000000..4ca123e94
--- /dev/null
+++ b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto
@@ -0,0 +1,84 @@
+syntax = "proto3";
+
+// The messages of the crate's own calls, in addition to the five Stream service operations. Lang
+// sends them through the bridge's call on the runtime object, by the method name in each comment.
+// Field numbers are part of the contract with every SDK.
+package temporal.sdk.streams.v1;
+
+import "google/protobuf/duration.proto";
+import "temporal/api/stream/v1/message.proto";
+
+// Method `OpenWriter`, asynchronous. A writer for an Activity, a Local Activity or a client
+// Producer, which keeps its epoch and `sequence` order. A writer of a Local Activity, or of an
+// Activity attempt above 1, first sends an empty batch, so that the new epoch withdraws the earlier
+// one's Records. The handles of one Producer epoch on one Stream share one writer. Each open adds a
+// handle, and each `CloseWriter` or `FreeWriter` releases one. The batches of every writer in a
+// process take their `sequence` from one counter, so a writer that opens again after its close
+// continues above its earlier batches.
+message OpenWriterRequest {
+    reserved 4, 7;
+    string namespace = 1;
+    temporal.api.stream.v1.StreamRef stream = 2;
+    temporal.api.stream.v1.Producer producer = 3;
+    // A write sends at most this many Records in one batch. Zero is 1000.
+    int32 max_batch_size = 5;
+    // How long a write retries a Store failure. Empty is 10 s.
+    google.protobuf.Duration max_retry_duration = 6;
+    // Local Activity only. The run's task facts, which each batch sends as AppendRecords does, so
+    // the writes withdraw the run's orphaned Local Activity Records.
+    temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 8;
+    repeated int64 failed_tasks = 9;
+}
+
+message OpenWriterResponse {
+    uint64 writer_id = 1;
+}
+
+// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer and responds when
+// the Store has them. It uses the writer's Producer epoch and `sequence` order. After a Store
+// failure it resends the same batch, so that the Store writes it once. Records past the Store's
+// batch limit go in more batches, in order. An awaited publish of an Activity or a client uses this
+// call. A Store failure after the retries fails only that write, and the next write takes a higher
+// `sequence`. Any other failure fails the writer.
+message WriteRecordsRequest {
+    uint64 writer_id = 1;
+    repeated temporal.api.stream.v1.AppendRecord records = 2;
+}
+
+message WriteRecordsResponse {
+    // The offsets of the first and the last Record. Zero when the request has no Record.
+    int64 first_offset = 1;
+    int64 last_offset = 2;
+}
+
+// Method `CloseWriter`, asynchronous. Releases one handle of the writer. The last handle waits for
+// a write in flight, then drops the writer. It doesn't close the Stream. It fails with the
+// writer's error if the writer failed.
+message CloseWriterRequest {
+    uint64 writer_id = 1;
+}
+
+message CloseWriterResponse {
+    // The offset of the newest Record the writer wrote. Zero if it wrote none.
+    int64 last_offset = 1;
+}
+
+// Method `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang
+// drops, such as from a finalizer. The bridge may also call it with no async runtime. A write in
+// flight still completes. The writer's error, if any, is lost.
+message FreeWriterRequest {
+    uint64 writer_id = 1;
+}
+
+message FreeWriterResponse {
+}
+
+// Method `CloseWriters`, asynchronous. Drops every open writer, for example when the backend's
+// client or Worker shuts down.
+message CloseWritersRequest {
+}
+
+message CloseWritersResponse {
+    // The writers that had failed.
+    int32 failed = 1;
+}
diff --git a/crates/streams/src/error.rs b/crates/streams/src/error.rs
new file mode 100644
index 000000000..0c0aa4a52
--- /dev/null
+++ b/crates/streams/src/error.rs
@@ -0,0 +1,163 @@
+//! The error of every failed call, and its form for lang.
+
+use crate::proto::{StreamErrorDetails, StreamErrorReason};
+use prost::Message;
+use tonic::Code;
+
+/// The result of every fallible call in the crate.
+pub type StreamResult<T> = Result<T, StreamError>;
+
+/// Why a call failed. Each reason has one status code and one case of `StreamErrorReason`, and lang
+/// maps the reason to one error class.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum Reason {
+    /// Nobody created the Stream, even after `create_wait`.
+    NotFound,
+    /// The Stream's Retention or safety TTL passed.
+    Expired,
+    /// The Stream is closed.
+    Closed,
+    /// The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a
+    /// Workflow retry withdrew its run.
+    Fenced,
+    /// The batch has a lower `sequence` than the last one of its epoch.
+    Sequence,
+    /// The batch has the `sequence` of the last one of its epoch, and other content.
+    SeqDivergent,
+    /// The Stream's owner doesn't allow this Producer to close it.
+    NotOwner,
+    /// A newer format version wrote the Stream.
+    FormatTooNew,
+    /// The offset is past the newest offset + 1, for example after a failover lost the newest
+    /// Records.
+    Offset,
+    /// A Record or a batch is over the Store's limit.
+    TooLarge,
+    /// The request is malformed.
+    Invalid,
+    /// The Store failed or did not answer.
+    Storage,
+    /// No writer has the id.
+    UnknownWriter,
+    /// The Store lacks a capability that the call needs.
+    Unsupported,
+}
+
+impl Reason {
+    /// The token that starts the error's text. Lang reads the reason from
+    /// `StreamErrorDetails.reason`, not from this text.
+    pub fn token(self) -> &'static str {
+        match self {
+            Reason::NotFound => "stream_not_found",
+            Reason::Expired => "stream_expired",
+            Reason::Closed => "stream_closed",
+            Reason::Fenced => "stream_fenced",
+            Reason::Sequence => "stream_sequence",
+            Reason::SeqDivergent => "seq_divergent",
+            Reason::NotOwner => "stream_not_owner",
+            Reason::FormatTooNew => "stream_format_too_new",
+            Reason::Offset => "stream_offset",
+            Reason::TooLarge => "stream_too_large",
+            Reason::Invalid => "stream_invalid",
+            Reason::Storage => "stream_storage",
+            Reason::UnknownWriter => "stream_unknown_writer",
+            Reason::Unsupported => "stream_unsupported",
+        }
+    }
+
+    /// The reason in the error details.
+    pub fn proto(self) -> StreamErrorReason {
+        match self {
+            Reason::NotFound => StreamErrorReason::NotFound,
+            Reason::Expired => StreamErrorReason::Expired,
+            Reason::Closed => StreamErrorReason::Closed,
+            Reason::Fenced => StreamErrorReason::Fenced,
+            Reason::Sequence => StreamErrorReason::Sequence,
+            Reason::SeqDivergent => StreamErrorReason::SeqDivergent,
+            Reason::NotOwner => StreamErrorReason::NotOwner,
+            Reason::FormatTooNew => StreamErrorReason::FormatTooNew,
+            Reason::Offset => StreamErrorReason::Offset,
+            Reason::TooLarge => StreamErrorReason::TooLarge,
+            Reason::Invalid => StreamErrorReason::Invalid,
+            Reason::Storage => StreamErrorReason::Storage,
+            Reason::UnknownWriter => StreamErrorReason::UnknownWriter,
+            Reason::Unsupported => StreamErrorReason::Unsupported,
+        }
+    }
+
+    /// The status code of the reason, as in the proposal's Appendix A.
+    pub fn code(self) -> Code {
+        match self {
+            Reason::NotFound | Reason::Expired | Reason::UnknownWriter => Code::NotFound,
+            Reason::Closed
+            | Reason::Fenced
+            | Reason::Sequence
+            | Reason::SeqDivergent
+            | Reason::NotOwner
+            | Reason::FormatTooNew => Code::FailedPrecondition,
+            Reason::TooLarge | Reason::Invalid => Code::InvalidArgument,
+            // A read from the offset works again once the Stream has Records up to it.
+            Reason::Offset => Code::OutOfRange,
+            Reason::Storage => Code::Unavailable,
+            Reason::Unsupported => Code::Unimplemented,
+        }
+    }
+}
+
+/// A failed Stream call.
+#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
+#[error("{}: {message}", reason.token())]
+pub struct StreamError {
+    /// Why the call failed.
+    pub reason: Reason,
+    /// What happened, for a log or the facade error.
+    pub message: String,
+}
+
+impl StreamError {
+    /// Creates an error.
+    pub fn new(reason: Reason, message: impl Into<String>) -> Self {
+        Self {
+            reason,
+            message: message.into(),
+        }
+    }
+
+    /// A malformed request.
+    pub fn invalid(message: impl Into<String>) -> Self {
+        Self::new(Reason::Invalid, message)
+    }
+
+    /// A Store failure.
+    pub fn storage(message: impl Into<String>) -> Self {
+        Self::new(Reason::Storage, message)
+    }
+
+    /// The status code that lang raises the error with.
+    pub fn code(&self) -> Code {
+        self.reason.code()
+    }
+
+    /// The encoded [StreamErrorDetails], which lang reads the reason from, as from the details of
+    /// a gRPC status.
+    pub fn details(&self) -> Vec<u8> {
+        StreamErrorDetails {
+            reason: self.reason.proto() as i32,
+        }
+        .encode_to_vec()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn an_error_carries_its_code_and_reason_to_lang() {
+        let error = StreamError::new(Reason::FormatTooNew, "format 2 is newer than 1");
+        assert_eq!(error.code(), Code::FailedPrecondition);
+        let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap();
+        assert_eq!(details.reason(), StreamErrorReason::FormatTooNew);
+    }
+}
diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs
new file mode 100644
index 000000000..95e6f23b4
--- /dev/null
+++ b/crates/streams/src/lib.rs
@@ -0,0 +1,20 @@
+#![warn(missing_docs)]
+
+//! The Stream runtime of Temporal SDKs. It is experimental.
+//!
+//! The crate serves the Stream service on a Stream Store. Lang's bridge builds the runtime per
+//! client and calls it with an operation name and proto bytes. Core has no hook for it. A Workflow
+//! makes each call from a Local Activity.
+
+mod error;
+pub mod protos;
+
+pub use error::{Reason, StreamError, StreamResult};
+
+/// The protos that lang and the crate exchange.
+pub mod proto {
+    pub use crate::protos::temporal::{
+        api::{enums::v1::*, stream::v1::*, streamservice::v1::*},
+        sdk::streams::v1::*,
+    };
+}
diff --git a/crates/streams/src/protos.rs b/crates/streams/src/protos.rs
new file mode 100644
index 000000000..7b939ed84
--- /dev/null
+++ b/crates/streams/src/protos.rs
@@ -0,0 +1,35 @@
+//! The generated protos of the Stream service of the API proposal and of the crate's own calls.
+
+#![allow(missing_docs, unreachable_pub, clippy::all)]
+
+/// The Temporal API packages that the crate generates. Others map to `temporalio-common`, and
+/// `enums` has only the Stream enums.
+pub mod temporal {
+    pub mod api {
+        pub mod enums {
+            pub mod v1 {
+                tonic::include_proto!("temporal.api.enums.v1");
+            }
+        }
+
+        pub mod stream {
+            pub mod v1 {
+                tonic::include_proto!("temporal.api.stream.v1");
+            }
+        }
+
+        pub mod streamservice {
+            pub mod v1 {
+                tonic::include_proto!("temporal.api.streamservice.v1");
+            }
+        }
+    }
+
+    pub mod sdk {
+        pub mod streams {
+            pub mod v1 {
+                tonic::include_proto!("temporal.sdk.streams.v1");
+            }
+        }
+    }
+}