[Brook] Added RedisStreamPlugin over the linked temporalio-streams crate.

pyproject.toml

@@ -97,8 +97,8 @@ dev = [
 ]
 
 [tool.poe.tasks]
-build-develop = "uv run maturin develop --uv"
-build-develop-with-release = { cmd = "uv run maturin develop --release --uv" }
+build-develop = "uv run maturin develop --uv --features streams"
+build-develop-with-release = { cmd = "uv run maturin develop --release --uv --features streams" }
 # FIPS builds: swap the rustls stack onto aws-lc-rs FIPS mode (aws-lc-fips-sys), eliminating `ring`.
 build-develop-fips = { cmd = "uv run maturin develop --uv --no-default-features --features fips", env = { TEMPORALIO_FIPS = "1" } }
 build-wheel-fips = { cmd = "uv run maturin build --release --no-default-features --features fips", env = { TEMPORALIO_FIPS = "1" } }
@@ -164,18 +164,20 @@ filterwarnings = [
 before-all = "pip install protoc-wheel-0"
 build = "cp310-win_amd64 cp310-manylinux_x86_64 cp310-manylinux_aarch64 cp310-musllinux_x86_64 cp310-musllinux_aarch64 cp310-macosx_x86_64 cp310-macosx_arm64"
 build-verbosity = 1
+# The official wheels link the Stream runtime.
+environment = { MATURIN_PEP517_ARGS = "--features streams" }
 
 [tool.cibuildwheel.macos]
-environment = { MACOSX_DEPLOYMENT_TARGET = "10.12" }
+environment = { MACOSX_DEPLOYMENT_TARGET = "10.12", MATURIN_PEP517_ARGS = "--features streams" }
 
 [tool.cibuildwheel.linux]
 before-all = "sh scripts/cibuildwheel_before_all_linux.sh"
 before-build = "sh scripts/cibuildwheel_before_build_linux.sh"
-environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true" }
+environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", MATURIN_PEP517_ARGS = "--features streams" }
 
 [[tool.cibuildwheel.overrides]]
 select = "*musllinux*"
-environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", RUSTFLAGS = "-C target-feature=-crt-static" }
+environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", RUSTFLAGS = "-C target-feature=-crt-static", MATURIN_PEP517_ARGS = "--features streams" }
 
 [tool.mypy]
 ignore_missing_imports = true

temporalio/bridge/Cargo.toml

@@ -29,14 +29,20 @@ default = ["tls-ring"]
 tls-ring = [
     "temporalio-sdk-core/tls-ring",
     "temporalio-client/tls-ring",
+    "temporalio-streams?/tls-ring",
     "tokio-rustls/ring",
 ]
 tls-aws-lc = [
     "temporalio-sdk-core/tls-aws-lc",
     "temporalio-client/tls-aws-lc",
+    "temporalio-streams?/tls-aws-lc",
     "tokio-rustls/aws_lc_rs",
 ]
-fips = ["tls-aws-lc", "dep:aws-lc-rs"]
+fips = ["tls-aws-lc", "dep:aws-lc-rs", "streams", "temporalio-streams?/fips"]
+# Links the Stream runtime as the `temporalio-streams-redis` worker extension. No stream code
+# runs unless a plugin builds the extension. The official builds turn it on. It is off by
+# default, since its git dependencies would break offline builds of the sdist.
+streams = ["dep:temporalio-streams"]
 
 [dependencies]
 anyhow = "1.0"
@@ -84,6 +90,20 @@ tonic = "0.14"
 tracing = "0.1"
 url = "2.2"
 
+# The rev moves with the sdk-core pin, and the patch below makes the crate use the submodule's
+# crates, so the bridge sees one type of each.
+[dependencies.temporalio-streams]
+git = "https://github.com/moetemp/sdk-streams"
+rev = "0981f62b1454e6501e3173d2115d94cb25eb9036"
+default-features = false
+optional = true
+
+# The key must match the sdk-rust URL in the sdk-streams manifest byte for byte, or Cargo
+# ignores the patch with only a warning.
+[patch."https://github.com/moetemp/sdk-rust"]
+temporalio-common = { path = "./sdk-core/crates/common" }
+temporalio-sdk-core = { path = "./sdk-core/crates/sdk-core" }
+
 [profile.release]
 opt-level = 3
 debug = false

temporalio/bridge/src/extension.rs

@@ -61,8 +61,11 @@ fn call_error(err: ExtensionError) -> PyErr {
 
 /// Builds the extension with `name` from its encoded config. Only the extensions that this build
 /// links are known.
-// A release build links no extension yet, so it never reads the config.
-#[cfg_attr(not(debug_assertions), allow(unused_variables))]
+// A release build with no extension feature never reads the config.
+#[cfg_attr(
+    not(any(debug_assertions, feature = "streams")),
+    allow(unused_variables)
+)]
 pub fn new_worker_extension(
     runtime_ref: &runtime::RuntimeRef,
     name: &str,
@@ -73,6 +76,12 @@ pub fn new_worker_extension(
     // An extension may start tasks on the runtime that it lives on.
     let _guard = runtime.core.tokio_handle().enter();
     let extension: Arc<dyn WorkerExtension> = match name {
+        #[cfg(feature = "streams")]
+        temporalio_streams::EXTENSION_NAME => {
+            let meter = runtime.core.telemetry().get_temporal_metric_meter();
+            temporalio_streams::RedisStreamExtension::from_config_bytes(&config, meter)
+                .map_err(|err| PyValueError::new_err(err.to_string()))?
+        }
         #[cfg(debug_assertions)]
         test_extension::NAME => Arc::new(test_extension::TestExtension::new(&config)?),
         _ => {