[FROZEN][Bridge] Generated the stream service from the crate's Nexus contract.

.github/workflows/ci.yml

@@ -244,6 +244,24 @@ jobs:
         env:
           TEMPORAL_TEST_PROTO3: 1
 
+  # The stream service's bindings come from the Nexus contract in the
+  # temporalio-streams crate that the bridge pins. Cargo fetches the crate, so
+  # this job needs no bridge build.
+  check-nexus-streams-api:
+    timeout-minutes: 20
+    runs-on: ubuntu-latest
+    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.10"
+      - uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8
+      - run: uv sync --only-dev
+      - run: uv run --no-sync python scripts/gen_nexus_streams_api.py --check
+
   test-latest-deps:
     timeout-minutes: 30
     runs-on: ubuntu-latest

.gitignore

@@ -15,3 +15,4 @@ temporalio/bridge/temporal_sdk_bridge*
 tags
 /.claude
 tmpclaude-*
+/.nexgen

pyproject.toml

@@ -109,6 +109,7 @@ format = [
 ]
 gen-docs = "uv run scripts/gen_docs.py"
 gen-nexus-system-api = "uv run scripts/gen_nexus_system_api.py"
+gen-nexus-streams-api = "uv run scripts/gen_nexus_streams_api.py"
 gen-protos = [
   { cmd = "uv run scripts/gen_protos.py" },
   { ref = "gen-nexus-system-api" },
@@ -194,7 +195,7 @@ 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|_generated|\\.)).*"
 add_ignore = [
   # We like to wrap at a certain number of chars, even long summary sentences.
   # https://github.com/PyCQA/pydocstyle/issues/184

scripts/gen_bridge_client.py

@@ -13,6 +13,7 @@
 import temporalio.api.testservice.v1.service_pb2 as test_service
 import temporalio.api.workflowservice.v1.service_pb2 as workflow_service
 import temporalio.bridge.proto.health.v1.health_pb2 as health_service
+import temporalio.bridge.proto.streams.nexus_pb2 as stream_nexus_service
 import temporalio.bridge.proto.streams.streams_pb2 as stream_service
 
 
@@ -245,44 +246,71 @@ def python_module_for(file_name: str) -> str:
 
 
 def generate_python_stream_service(
-    service_descriptor: ServiceDescriptor,
+    service_descriptors: list[ServiceDescriptor],
     output_file: str = "temporalio/bridge/streams_generated.py",
 ):
-    print("generating python stream service")
+    print("generating python stream services")
 
     def type_name(message) -> str:
         return f"{python_module_for(message.file.name)}.{message.name}"
 
-    methods = sorted(
-        (
-            method
-            for method in service_descriptor.methods
-            if not method.client_streaming and not method.server_streaming
-        ),
-        key=lambda m: m.name,
-    )
+    def unary(service_descriptor: ServiceDescriptor) -> list:
+        return sorted(
+            (
+                method
+                for method in service_descriptor.methods
+                if not method.client_streaming and not method.server_streaming
+            ),
+            key=lambda m: m.name,
+        )
+
     modules = sorted(
-        {python_module_for(service_descriptor.file.name)}
-        | {python_module_for(m.input_type.file.name) for m in methods}
-        | {python_module_for(m.output_type.file.name) for m in methods}
+        {python_module_for(d.file.name) for d in service_descriptors}
+        | {
+            python_module_for(m.input_type.file.name)
+            for d in service_descriptors
+            for m in unary(d)
+        }
+        | {
+            python_module_for(m.output_type.file.name)
+            for d in service_descriptors
+            for m in unary(d)
+        }
     )
     method_template = Template('''
     async def $method_name(self, req: $request_type) -> $response_type:
         """Invokes the $service_name.$method_name call."""
         return await self._store.call("$rpc_name", req, $response_type)
 ''')
-    method_calls = [
-        method_template.substitute(
+    # Every service's calls go through the store's one entry, which the stream
+    # runtime dispatches by name, so each service is a class over the same store.
+    class_template = Template('''
+
+class $service_name:
+    """Calls for the $service_name, which the stream runtime serves in process."""
+
+    def __init__(self, store: StreamStore) -> None:
+        """Initialize service with the provided store."""
+        self._store = store
+$method_calls''')
+    classes = [
+        class_template.substitute(
             service_name=service_descriptor.name,
-            method_name=pascal_to_snake(method.name),
-            rpc_name=method.name,
-            request_type=type_name(method.input_type),
-            response_type=type_name(method.output_type),
+            method_calls="".join(
+                method_template.substitute(
+                    service_name=service_descriptor.name,
+                    method_name=pascal_to_snake(method.name),
+                    rpc_name=method.name,
+                    request_type=type_name(method.input_type),
+                    response_type=type_name(method.output_type),
+                )
+                for method in unary(service_descriptor)
+            ),
         )
-        for method in methods
+        for service_descriptor in service_descriptors
     ]
     file_template = Template('''# Generated file. DO NOT EDIT
-"""Generated calls for the stream runtime's in-process $service_name."""
+"""Generated calls for the stream runtime's in-process stream services."""
 
 from __future__ import annotations
 
@@ -292,21 +320,12 @@ async def $method_name(self, req: $request_type) -> $response_type:
 
 if TYPE_CHECKING:
     from temporalio.bridge.streams import StreamStore
-
-
-class $service_name:
-    """Calls for the $service_name, which the stream runtime serves in process."""
-
-    def __init__(self, store: StreamStore) -> None:
-        """Initialize service with the provided store."""
-        self._store = store
-$method_calls''')
+$classes''')
     with open(output_file, "w") as f:
         f.write(
             file_template.substitute(
-                service_name=service_descriptor.name,
                 imports="\n".join(f"import {module}" for module in modules),
-                method_calls="".join(method_calls),
+                classes="".join(classes),
             )
         )
 
@@ -355,5 +374,8 @@ def sanitize_proto_name(input: str) -> str:
     )
 
     generate_python_stream_service(
-        stream_service.DESCRIPTOR.services_by_name["StreamService"]
+        [
+            stream_service.DESCRIPTOR.services_by_name["StreamService"],
+            stream_nexus_service.DESCRIPTOR.services_by_name["StreamNexusService"],
+        ]
     )

scripts/gen_nexus_streams_api.py

@@ -0,0 +1,104 @@
+"""Generate the stream service's Python bindings from its Nexus contract.
+
+The stream runtime serves the contract, so its one copy is the
+``crates/streams/nexus/temporal_streams.nexusrpc.yaml`` of the temporalio-streams
+crate that the bridge pins, and the bindings land in
+``temporalio/contrib/streams/nexus/_generated``. With
+``--check`` the script generates into a scratch directory and fails when the
+result differs from what is checked in, which is how CI keeps the two in step.
+
+The contract uses the ``x-nexus-long-poll``, ``x-nexus-cursor``,
+``x-nexus-stream-ref``, ``x-nexus-handle`` and ``x-nexus-operation-token``
+annotations, and the bindings include the forwarding handler, which no nexgen
+release carries yet. So it uses the nexgen fork revision that
+``gen_nexus_system_api.py`` pins, and the same install.
+"""
+
+import argparse
+import filecmp
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+import gen_protos
+from gen_nexus_system_api import nex_gen_command
+
+base_dir = Path(__file__).parent.parent
+package_dir = base_dir / "temporalio" / "contrib" / "streams" / "nexus"
+contract_path = (
+    gen_protos.streams_proto_dir.parent / "nexus" / "temporal_streams.nexusrpc.yaml"
+)
+output_dir = package_dir / "_generated"
+
+
+def generate(target: Path) -> None:
+    """Generate the bindings into ``target``, formatted as the repository formats."""
+    subprocess.check_call(
+        [
+            *nex_gen_command(),
+            "python",
+            "--client",
+            "--forwarder",
+            "--output",
+            str(target),
+            str(contract_path),
+        ]
+    )
+    for fix in (["check", "--select", "I", "--fix"], ["format"]):
+        subprocess.check_call(
+            [sys.executable, "-m", "ruff", *fix, "--quiet", str(target)]
+        )
+
+
+def differences(left: Path, right: Path) -> list[str]:
+    """The relative paths whose content differs between two trees."""
+    comparison = filecmp.dircmp(left, right, ignore=["__pycache__"])
+    found = [
+        *comparison.left_only,
+        *comparison.right_only,
+        *comparison.diff_files,
+        *comparison.funny_files,
+    ]
+    for name, sub in comparison.subdirs.items():
+        found.extend(
+            f"{name}/{path}" for path in differences(Path(sub.left), Path(sub.right))
+        )
+    _, mismatch, errors = filecmp.cmpfiles(
+        left, right, comparison.same_files, shallow=False
+    )
+    return sorted({*found, *mismatch, *errors})
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="fail when regenerating would change the checked-in bindings",
+    )
+    args = parser.parse_args()
+    if not contract_path.exists():
+        raise RuntimeError(f"missing contract: {contract_path}")
+
+    with tempfile.TemporaryDirectory(dir=base_dir) as temp_dir:
+        scratch = Path(temp_dir) / "_generated"
+        generate(scratch)
+        if args.check:
+            changed = differences(scratch, output_dir) if output_dir.exists() else ["."]
+            if changed:
+                print(
+                    "The stream service bindings are stale; run "
+                    "`uv run scripts/gen_nexus_streams_api.py`. Changed: "
+                    + ", ".join(changed),
+                    file=sys.stderr,
+                )
+                sys.exit(1)
+            return
+        shutil.rmtree(output_dir, ignore_errors=True)
+        shutil.copytree(scratch, output_dir)
+
+
+if __name__ == "__main__":
+    main()

scripts/gen_protos.py

@@ -50,6 +50,8 @@ def find_streams_proto_dir() -> Path:
 # The stream API that the crate carries until it lands upstream. Core's api_upstream does not have
 # it, so its modules go into temporalio.api from here.
 streams_api_proto_dir = streams_proto_dir / "api_stream"
+# The Nexus progress and sync call API, which the crate carries for the same reason.
+nexus_api_proto_dir = streams_proto_dir / "api_nexus"
 
 # Exclude testsrv dependencies protos
 proto_paths = [
@@ -62,6 +64,7 @@ def find_streams_proto_dir() -> Path:
 # The crate's api_include copy is left out, since api_upstream generates those modules.
 proto_paths.extend((streams_proto_dir / "temporal").glob("**/*.proto"))
 proto_paths.extend(streams_api_proto_dir.glob("**/*.proto"))
+proto_paths.extend(nexus_api_proto_dir.glob("**/*.proto"))
 proto_paths.extend(test_proto_dir.glob("**/*.proto"))
 proto_paths.extend(additional_proto_dir.glob("**/*.proto"))
 
@@ -221,6 +224,7 @@ def generate_protos(output_dir: Path):
             "-mgrpc_tools.protoc",
             f"--proto_path={api_proto_dir}",
             f"--proto_path={streams_api_proto_dir}",
+            f"--proto_path={nexus_api_proto_dir}",
             f"--proto_path={api_cloud_proto_dir}",
             f"--proto_path={core_proto_dir}",
             f"--proto_path={testsrv_proto_dir}",