Added one stream interface with its four providers and the Nexus front.

.github/workflows/ci.yml

@@ -100,6 +100,56 @@ jobs:
           npx doctoc README.md
           [[ -z $(git status --porcelain README.md) ]] || (git diff README.md; echo "README changed"; exit 1)
 
+  # The client-side (Redis) stream provider's own evidence. Its tests need a store
+  # this repo does not otherwise stand up, so without this job nothing that proves
+  # retention, cursor ownership, the staged commit or the paired producer write ever
+  # runs anywhere but a developer's machine.
+  streams-redis:
+    timeout-minutes: 30
+    runs-on: ubuntu-latest
+    services:
+      redis:
+        image: redis:8-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 5s
+          --health-timeout 3s
+          --health-retries 10
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          submodules: recursive
+      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
+        with:
+          python-version: "3.13"
+      - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
+        with:
+          workspaces: temporalio/bridge -> target
+          key: streams-redis-${{ env.pythonLocation }}
+      - uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3
+        with:
+          version: "23.x"
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8
+      - run: uv tool install poethepoet
+      - run: uv sync --all-extras
+      - run: poe build-develop
+      # The dev server comes from the test environment, the store from the service
+      # above. Run serially: the cases measure real timing and share one Redis.
+      - run: uv run pytest tests/streams -p no:randomly -s
+        timeout-minutes: 20
+        env:
+          STREAMS_LIVE: redis
+          TEMPORAL_TEST_REDIS_URL: redis://127.0.0.1:6379
+          AI198_REDIS_URL: redis://127.0.0.1:6379
+      # Also without the store, so the gate itself keeps working and the memory
+      # provider's conformance run stays honest.
+      - run: uv run pytest tests/streams -p no:randomly -s
+        timeout-minutes: 10
+
   # Verify the optional FIPS build: the Rust core must link aws-lc-fips-sys
   # (aws-lc-rs FIPS mode) and must NOT link `ring` (the cargo-tree guard, ported
   # from sdk-ruby PR #466's `fips_tree` guard); then run the test suite against the

.gitignore

@@ -15,3 +15,6 @@ temporalio/bridge/temporal_sdk_bridge*
 tags
 /.claude
 tmpclaude-*
+
+# Demo run output, written per provider; the numbers live in the doc.
+streams_demo/results-*/

.gitmodules

@@ -1,3 +1,4 @@
 [submodule "sdk-core"]
 	path = temporalio/bridge/sdk-core
-	url = https://github.com/temporalio/sdk-rust.git
+	url = https://github.com/moedash/sdk-rust.git
+	branch = moe/AI-198-core-all

CHANGELOG.md

@@ -37,6 +37,97 @@ to include examples, links to docs, or any other relevant information.
   worker-side factories registered with `StrandsPlugin(sandboxes=...)`.
 
 - Added the `temporalio.contrib.gcp.cloud_run.id` module with the `CloudRunIdPlugin` client plugin to set the worker identity on Cloud Run.
+- **Experimental**: `temporalio.streams` defines one stream interface a workflow
+  can read, decide on, and write. A provider is registered once as a plugin,
+  `Client.connect(plugins=[provider])`, and workers built from that client
+  inherit it; each context then asks for its stream the same way:
+  `workflow.stream_reader()` and `workflow.stream_writer()` in workflow code,
+  `activity.stream_handle()` in an activity, and `client.get_stream_handle()`
+  anywhere a client is held. A topic is a typed definition,
+  `streams.topic("inputs", Token)`, shared by workflow, activity and client
+  code; a plain string names a topic decided at runtime. The record on the wire
+  is `temporal.api.stream.v1.StreamRecord` on every provider, and
+  code; a plain string names a topic decided at runtime, and a call that names
+  no topic addresses the default topic, `streams.DEFAULT_TOPIC` (`"output"`,
+  the server's default stream name). The record on the wire
+  is `temporal.api.stream.v1.StreamRecord` on every provider. A stream is
+  handed to another process as a `streams.StreamRef`, plain data naming the
+  owner and the topic, which `client.get_stream_handle(ref)` and
+  `activity.stream_handle(ref)` open; `client.create_stream(stream_id, ...)`
+  creates a standalone stream with a retention policy, and its handle's
+  `close()` seals it. A provider runs record bodies through the client's data
+  converter, so a payload codec and external storage apply to them.
+  `temporalio.streams.providers.memory.MemoryStreams` is the in-memory
+  reference provider the conformance tests run against.
+- **Experimental**: `temporalio.streams.providers.redis.RedisStreams` serves the
+  stream interface over External Workflow Streams, holding one topic as an
+  input and an output stream.
+- **Experimental**: `temporalio.streams.providers.workflow_streams.WorkflowStreamsProvider`
+  serves the stream interface over the shipped Workflow Streams transport as a
+  worker plugin, so a workflow reads and publishes through
+  `temporalio.contrib.workflow_streams` without naming it. Records are the
+  `StreamRecord` proto inside the shipped item payload, and a handle without a
+  run id follows continue-as-new run by run.
+- **Experimental**: `temporalio.streams.providers.nexus.NexusStreams` puts one
+  Nexus endpoint in front of a storage provider, so a caller reaches a stream
+  through the endpoint and never names the store, and
+  `TemporalStreamsHandler` serves that endpoint by fronting the provider's own
+  handles. Its contract is defined in `temporal_streams.nexusrpc.yaml` and the
+  bindings are generated from it; a record crosses as the serialized
+  `StreamRecord` proto. Configure the front with `data_converter=` to run a
+  payload codec on the caller side, so records are encoded before they leave
+  the process.
+- **Experimental**: server-side streams. A workflow publishes to a stream it
+  owns with a command the server applies in its Workflow Task's commit, and
+  reads the ranges the server delivers on its Workflow Tasks, through
+  `temporalio.workflow.append_stream_records`, `subscribe_stream` and
+  `read_stream_records`. `temporalio.client_stream` and
+  `temporalio.contrib.server_streams` reach the same stream from outside a
+  workflow, and `temporalio.streams.providers.native.NativeStreams` puts it
+  behind the shared stream interface with one owned stream per topic. Requires
+  a server that serves the stream service. `Replayer(stream_client=)` replays a
+  workflow that read such a stream while the server still holds it: History
+  records only the offsets each task consumed, so the replayer fetches the
+  records from the stream service and hands them to the replay with the
+  history. A range the stream no longer holds fails the replay with
+  `StreamNotFoundError`. A handle without a run id follows a workflow reset as
+  it follows a continue-as-new, reading the reset run from the floor its stream
+  reports, and the replayer fetches the ranges recorded before a reset point
+  from the run the workflow was reset from. For offline replay,
+  `Replayer.fetch_stream_slices(client, history)` attaches the records to a
+  `WorkflowHistory` while the stream is retained, `to_json()` and `from_json()`
+  carry them as `streamSlices` beside the events, and a history that carries
+  them replays with no server.
+- Added experimental External Workflow Streams in
+  `temporalio.contrib.external_workflow_streams`. Workflow stream payloads are
+  stored in a configured external backend instead of Temporal History, with a
+  Redis Streams provider included. Workflows subscribe with `external_stream`,
+  external processes publish with `ExternalStreamProducer`, and Workers are
+  configured with `external_stream_backend`.
+- Added the output direction for External Workflow Streams.
+  Workflows publish with `external_output_stream`, Activities and external
+  processes use `ExternalOutputStreamProducer`, and external consumers resume
+  through `ExternalOutputStreamClient`. Workflow output is staged outside
+  History and becomes readable only after its compact Workflow Task marker is
+  committed.
+- `ExternalStreamSubscription.records()` yields each value with the provider
+  offset it was read from, for a reader that has to name where it got to.
+- Added the `temporalio.contrib.gcp.cloud_run.id` module with the `CloudRunIdPlugin` client plugin to set the worker identity on Cloud Run.
+  `StreamRecord` proto, and both operations address a stream by a `StreamRef`
+  naming its owner (a workflow, an activity or a standalone stream) and topic,
+  which the handler maps onto the store's accessor for that owner. Configure
+  the front with `data_converter=` to run a payload codec on the caller side,
+  so records are encoded before they leave the process.
+- **Experimental**: `temporalio.streams.providers.workflow_streams.WorkflowStreamsProvider`
+  serves the stream interface over the shipped Workflow Streams transport as a
+  worker plugin, so a workflow reads and publishes through
+  `temporalio.contrib.workflow_streams` without naming it. Records are the
+  `StreamRecord` proto inside the shipped item payload, and a handle without a
+  run id follows continue-as-new run by run and a reset into the run reset to.
+  An outside publish is an Update that answers with the batch's position and
+  refuses a conflicting repeat, falling back to the shipped Signal on a
+  workflow whose worker predates it. A workflow's activity keeps its own
+  streams in the workflow's log under `activity/<id>/<name>`.
 
 ### Changed
 
@@ -52,6 +143,22 @@ to include examples, links to docs, or any other relevant information.
 
 ### Fixed
 
+- Preserve empty activations in the shared External Workflow Streams input and
+  output replay schedule. Workflows that read input, publish decisions, and
+  schedule Activities now reproduce that schedule during replay. Inconsistent
+  prerelease markers are rejected explicitly rather than guessing where omitted
+  activations belonged.
+- Resume external input waits when cold replay encounters a wake in an already
+  loaded History page, including Workers with workflow caching disabled.
+- Avoid an unnecessary output replacement Workflow Task after stream input has
+  resumed the Workflow and it is waiting on an Activity or timer.
+- Keep an incomplete retained external stream task alive when workflow caching
+  is disabled; evict it after its normal task boundary instead of repeatedly
+  interrupting input readiness with shutdown markers.
+- `WorkflowEnvironment.start_time_skipping()`: concurrent `WorkflowHandle.result()` waiters now
+  share one time-skipping unlock. The test server holds one lock per in-flight task, so a second
+  unlock let the clock jump while another workflow still had a task in flight, and that task then
+  timed out.
 - `GoogleAdkPlugin` now passes the optional `anthropic`, `litellm`, and `openai` SDKs through
   the workflow sandbox.
 - `contrib.deepagents`: prevent duplicate input messages after continue-as-new.
@@ -72,6 +179,28 @@ to include examples, links to docs, or any other relevant information.
 ### Added
 
 #### Standalone Activity operator commands
+- **Experimental**: `temporalio.streams` defines one stream interface a workflow
+  can read, decide on, and write. A provider is registered once as a plugin,
+  `Client.connect(plugins=[provider])`, and workers built from that client
+  inherit it; each context then asks for its stream the same way:
+  `workflow.stream_reader()` and `workflow.stream_writer()` in workflow code,
+  `activity.stream_handle()` in an activity, and `client.get_stream_handle()`
+  anywhere a client is held. A topic is a typed definition,
+  `streams.topic("inputs", Token)`, shared by workflow, activity and client
+  code; a plain string names a topic decided at runtime. The record on the wire
+  is `temporal.api.stream.v1.StreamRecord` on every provider. A stream is
+  handed to another process as a `streams.StreamRef`, plain data naming the
+  owner and, when it has one, the topic, which `client.get_stream_handle(ref)`
+  and `activity.stream_handle(ref)` open; `client.create_stream(stream_id, ...)`
+  creates a standalone stream with a retention policy, and its handle's
+  `close()` seals it. A provider runs record bodies through the client's data
+  converter, so a payload codec and external storage apply to them.
+  `temporalio.streams.providers.memory.MemoryStreams` is the in-memory
+  reference provider the conformance tests run against, and
+  `temporalio.streams.providers.redis.RedisStreams` serves the same interface
+  over External Workflow Streams, one topic as an input and an output stream.
+- `ExternalStreamSubscription.records()` yields each value with the provider
+  offset it was read from, for a reader that has to name where it got to.
 
 - `ActivityHandle` now supports operator commands for standalone activities: `pause`,
   `unpause`, `update_options` and `restore_original_options`.

pyproject.toml

@@ -95,6 +95,7 @@ dev = [
   "pytest-rerunfailures>=16.1",
   "pytest-xdist>=3.6,<4",
   "moto[s3,server]>=5",
+  "redis>=5,<9",
   "langgraph>=1.1.0",
   "langsmith>=0.7.34,<0.9",
   "deepagents>=0.6.12,<0.7; python_version >= '3.11'",
@@ -125,16 +126,19 @@ format = [
 ]
 gen-docs = "uv run scripts/gen_docs.py"
 gen-nexus-system-api = "uv run scripts/gen_nexus_system_api.py"
+gen-streams-nexus-api = "uv run scripts/gen_streams_nexus_api.py"
 gen-protos = [
   { cmd = "uv run scripts/gen_protos.py" },
   { ref = "gen-nexus-system-api" },
+  { ref = "gen-streams-nexus-api" },
   { cmd = "uv run scripts/gen_payload_visitor.py" },
   { cmd = "uv run scripts/gen_bridge_client.py" },
   { ref = "format" },
 ]
 gen-protos-docker = [
   { cmd = "uv run scripts/gen_protos_docker.py" },
   { ref = "gen-nexus-system-api" },
+  { ref = "gen-streams-nexus-api" },
   { cmd = "uv run scripts/gen_payload_visitor.py" },
   { cmd = "uv run scripts/gen_bridge_client.py" },
   { ref = "format" },
@@ -198,16 +202,21 @@ exclude = [
   'temporalio/api',
   'temporalio/bridge/proto',
   'temporalio/nexus/system/workflow_service',
+  'temporalio/streams/providers/_nexus_generated',
 ]
 
 [[tool.mypy.overrides]]
 module = "temporalio.nexus.system.workflow_service.*"
 ignore_errors = true
 
+[[tool.mypy.overrides]]
+module = "temporalio.streams.providers._nexus_generated.*"
+ignore_errors = true
+
 [tool.pydocstyle]
 convention = "google"
 # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088
-match_dir = "^(?!(docs|scripts|tests|api|proto|system|\\.)).*"
+match_dir = "^(?!(docs|scripts|tests|api|proto|system|_nexus_generated|\\.)).*"
 add_ignore = [
   # We like to wrap at a certain number of chars, even long summary sentences.
   # https://github.com/PyCQA/pydocstyle/issues/184
@@ -241,6 +250,7 @@ privacy = [
   "HIDDEN:temporalio.worker.workflow_sandbox.importer",
   "HIDDEN:temporalio.worker.workflow_sandbox.in_sandbox",
   "HIDDEN:**.*_pb2*",
+  "HIDDEN:temporalio.streams.providers._nexus_generated._definitions",
 ]
 project-name = "Temporal Python"
 sidebar-expand-depth = 2
@@ -267,7 +277,7 @@ reportUnnecessaryIsInstance = "none"
 reportUnnecessaryTypeIgnoreComment = "none"
 reportUnusedCallResult = "none"
 reportUnknownLambdaType = "none"
-include = ["temporalio", "tests"]
+include = ["temporalio", "tests", "streams_demo"]
 exclude = [
   # Exclude auto generated files
   "temporalio/api",

scripts/gen_payload_visitor.py

@@ -63,13 +63,10 @@ def name_for(desc: Descriptor) -> str:
 
 
 def field_is_repeated(field: FieldDescriptor) -> bool:
-    return bool(
-        getattr(
-            field,
-            "is_repeated",
-            getattr(field, "label") == FieldDescriptor.LABEL_REPEATED,
-        )
-    )
+    is_repeated = getattr(field, "is_repeated", None)
+    if is_repeated is not None:
+        return bool(is_repeated)
+    return getattr(field, "label") == FieldDescriptor.LABEL_REPEATED
 
 
 def emit_loop(