[Brook] Reached Core's stream store with records and errors.

.gitattributes

@@ -1 +1,11 @@
 CHANGELOG.md merge=union
+# GitHub folds generated files in diffs.
+temporalio/api/** linguist-generated=true
+temporalio/bridge/proto/** linguist-generated=true
+*_pb2*.py linguist-generated=true
+*_pb2*.pyi linguist-generated=true
+*_generated.py linguist-generated=true
+*_generated.rs linguist-generated=true
+**/_generated/** linguist-generated=true
+**/testdata/** linguist-generated=true
+*_vectors.json linguist-generated=true

changelog/added/cornfed-tanesha.md

@@ -0,0 +1 @@
+Core: Experimental: the stream service's `RetireProducer` call drops one producer attempt's dedupe state on a stream, which the Redis store keeps as one field of the topic's meta hash until retention. A producer that writes once per short-lived attempt calls it after the attempt's last append returned, so the meta hash tracks the attempts in flight. Calling it earlier lets a retried batch land twice.

changelog/added/hypermagical-celena.md

@@ -0,0 +1,2 @@
+Core: Experimental: behind the new `streams` feature, `streams::connect_stream_service` builds one stream store from a `StreamStoreConfig` (Redis or memory) and serves lang's stream calls in process: append, read, latest, close and owner delete, as serialized `coresdk.streams` requests.
+Core: Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the stream service, built with the Redis store.

changelog/breaking-changes/hypermagical-celena.md

@@ -0,0 +1 @@
+Core: Replay detects changes to the service or operation of scheduled Nexus operations, reporting nondeterminism while retaining compatibility with histories recorded before these checks.

changelog/fixed/hypermagical-celena.md

@@ -0,0 +1,2 @@
+Core: Local activities still running when their workflow is evicted or the worker shuts down now always receive a cancellation. Local activities no longer retry ordinary failures or start-to-close timeouts after cancellation has been requested, matching non-local activities. Workflow code no longer observes a spurious local activity cancellation when its run is evicted or completes while the local activity is backing off between retries.
+Core: Workers no longer panic when evicting a workflow with buffered local activity results. Completed activations' output is recorded before eviction, without recording eviction-induced cancellations.

pyproject.toml

@@ -183,6 +183,8 @@ exclude = [
   'temporalio/api',
   'temporalio/bridge/proto',
   'temporalio/nexus/system/workflow_service',
+  # Core's own sources, which hold a Python script for Core's cross-SDK tests
+  'temporalio/bridge/sdk-core',
 ]
 
 [[tool.mypy.overrides]]
@@ -258,11 +260,15 @@ exclude = [
   "temporalio/api",
   "temporalio/bridge/proto",
   "temporalio/bridge/_visitor.py",
+  # Core's own sources, which hold a Python script for Core's cross-SDK tests
+  "temporalio/bridge/sdk-core",
   "tests/worker/workflow_sandbox/testmodules/proto",
 ]
 
 [tool.ruff]
 target-version = "py310"
+# Core's own sources, which hold a Python script for Core's cross-SDK tests
+extend-exclude = ["temporalio/bridge/sdk-core"]
 
 [build-system]
 requires = ["maturin>=1.0,<2.0"]

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.streams_pb2 as stream_service
 
 
 def generate_python_services(
@@ -231,6 +232,85 @@ def generate_rust_match_arm(
     )
 
 
+def python_module_for(file_name: str) -> str:
+    """The Python package that re-exports the messages of a proto file."""
+    directory = file_name.rsplit("/", 1)[0]
+    for proto_prefix, python_prefix in (
+        ("temporal/sdk/core/", "temporalio/bridge/proto/"),
+        ("temporal/api/", "temporalio/api/"),
+    ):
+        if directory.startswith(proto_prefix):
+            return (python_prefix + directory[len(proto_prefix) :]).replace("/", ".")
+    raise ValueError(f"no Python package for {file_name}")
+
+
+def generate_python_stream_service(
+    service_descriptor: ServiceDescriptor,
+    output_file: str = "temporalio/bridge/streams_generated.py",
+):
+    print("generating python stream service")
+
+    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,
+    )
+    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}
+    )
+    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(
+            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 methods
+    ]
+    file_template = Template('''# Generated file. DO NOT EDIT
+"""Generated calls for Core's in-process $service_name."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+$imports
+
+if TYPE_CHECKING:
+    from temporalio.bridge.streams import StreamStore
+
+
+class $service_name:
+    """Calls for the $service_name, which Core dispatches in process."""
+
+    def __init__(self, store: StreamStore) -> None:
+        """Initialize service with the provided store."""
+        self._store = store
+$method_calls''')
+    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),
+            )
+        )
+
+
 def pascal_to_snake(input: str) -> str:
     return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", input).lower()
 
@@ -273,3 +353,7 @@ def sanitize_proto_name(input: str) -> str:
             health_service.DESCRIPTOR,
         ]
     )
+
+    generate_python_stream_service(
+        stream_service.DESCRIPTOR.services_by_name["StreamService"]
+    )

scripts/gen_protos.py

@@ -50,6 +50,10 @@
     partial(
         re.compile(r"from temporal\.sdk\.core\.").sub, r"from temporalio.bridge.proto."
     ),
+    partial(
+        re.compile(r"from temporal\.sdk\.streams\.").sub,
+        r"from temporalio.bridge.proto.streams.",
+    ),
     partial(
         re.compile(r"'__module__' : 'temporal\.api\.").sub,
         r"'__module__' : 'temporalio.api.",
@@ -71,6 +75,9 @@
         r"temporalio.api.dependencies.nexusannotations.",
     ),
     partial(re.compile(r"temporal\.sdk\.core\.").sub, r"temporalio.bridge.proto."),
+    partial(
+        re.compile(r"temporal\.sdk\.streams\.").sub, r"temporalio.bridge.proto.streams."
+    ),
 ]
 
 find_class_re = re.compile(r"\nclass ([^_\(\:]+)")
@@ -218,6 +225,11 @@ def generate_protos(output_dir: Path):
     for p in (output_dir / "temporal" / "sdk" / "core").iterdir():
         shutil.rmtree(sdk_out_dir / p.name, ignore_errors=True)
         p.replace(sdk_out_dir / p.name)
+    # The stream record envelope sits beside Core's own stream messages, which the loop above
+    # has just moved in.
+    (output_dir / "temporal" / "sdk" / "streams" / "v1").replace(
+        sdk_out_dir / "streams" / "v1"
+    )
     shutil.rmtree(sdk_out_dir / "health", ignore_errors=True)
     (output_dir / "health").replace(sdk_out_dir / "health")
     # Move test protos