[Brook] Reached Core's stream store from Python.

changelog/added/irretraceable-branda.md

@@ -0,0 +1,6 @@
+Core: Experimental: a workflow can commit the stream records it published with the new `WorkflowOutputStreamCommit` command, each with its plaintext content hash and size. Core builds the output manifest from them and records it in a `core_external_stream` marker ahead of the completion's other commands, so the output becomes visible only when the workflow task is accepted. While replaying, lang commits the same records again without bodies, and Core checks the manifest it builds from them against the recorded one (the stage token, run id and store name aside), or the task fails as nondeterministic. A workflow that wrote this marker can only move forward. A Worker on an older Core, or one whose lang doesn't commit the records again on replay, fails that workflow's next task, so don't roll such a fleet back.
+Core: Experimental: behind the new `streams` feature, `WorkerConfig::stream_store` takes the store a workflow's stream output goes to. Core stages each completion's records in it before the completion is sent, and fails the workflow task when the store refuses them or the Worker has none, so a task the server accepts always has its output staged.
+Core: Experimental: Core promotes a workflow task's staged stream output once the server accepts the completion, and aborts it when the server says the task or its commands were not applied. When the answer is lost, the next replay promotes what History shows committed and the next accepted completion aborts the rest. Output another Worker staged and History proves is promoted once.
+Core: Experimental: after a publishing workflow's final task (complete, fail or cancel, not continue-as-new) is accepted, Core closes its run chain's streams in the store, so readers end and producers are refused. It checks the chain's latest run first and leaves the streams open for a retry or cron successor. A failed close is retried with backoff until it lands or the Worker stops.
+Core: Experimental: behind `streams`, `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. Its store goes to each Worker's `stream_store`, so a Workflow's output and outside producers share it.
+Core: Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the stream service, built with the Redis store, and a `stream_store` field on its worker options.

changelog/breaking-changes/irretraceable-branda.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/irretraceable-branda.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