[FROZEN][Bridge] Served the stream service from a Worker's store.

temporalio/contrib/streams/_body.py

@@ -1,12 +1,18 @@
 """What the SDK owes a record's body between the converter and the store.
 
-The payload converter turns a value into the body and stops there. Every other
-payload the SDK sends then passes through the payload codec and external
-storage, and a stream body owes the same, or a codec-protected deployment would
-leak plaintext through its streams. The stream runtime never sees a plaintext
-body, so both hashes that need one are taken here, before the codec. A codec
-that encrypts with a fresh nonce makes every retry's bytes differ, so a store
-that compared encoded bytes would refuse a retry as a divergent write.
+The payload converter turns a value into the body and stops there. Every
+other payload the SDK sends then passes through the payload codec and
+external storage, and a stream body owes the same, or a codec-protected
+deployment would leak plaintext through its streams. On the direct paths, a
+Workflow Producer and the outside handles, the stream runtime never sees a
+plaintext body, so both hashes that need one are taken here, before the codec.
+A codec that encrypts with a fresh nonce makes every retry's bytes differ, so a
+store that compared encoded bytes would refuse a retry as a divergent write.
+
+The stream service's Nexus path is different. The stream runtime gets
+the caller's plain bodies, takes both hashes over the bytes the caller sent,
+and asks the handler to run the codec. Only the codec step uses this module
+there.
 """
 
 from __future__ import annotations
@@ -63,6 +69,19 @@ async def encode_bodies(
     return await converter._external_store_payload_sequence(encoded)
 
 
+async def decode_bodies(
+    converter: DataConverter, payloads: Sequence[Payload]
+) -> list[Payload]:
+    """Undo :func:`encode_bodies` on bodies read back from the store, together.
+
+    Raises:
+        RuntimeError: A body is a claim and ``converter`` has no external
+            storage to redeem it with.
+    """
+    retrieved = await converter._external_retrieve_payload_sequence(payloads)
+    return await converter._decode_payload_sequence(retrieved)
+
+
 async def decode_body(converter: DataConverter, payload: Payload) -> Payload:
     """Undo :func:`encode_bodies` on a body read back from the store.
 

temporalio/contrib/streams/nexus/__init__.py

@@ -24,6 +24,11 @@
   nothing beyond the standard library.
 * ``TemporalStreamsForwarder``, which hands each operation's input to the
   stream runtime as the contract's JSON transfer form.
+
+:class:`TemporalStreamsHandler` is the one hand-written piece. It runs its
+authorizer and the namespace's payload codec, the parts only the SDK has, and
+the stream runtime runs every rule of the contract on the store of its Worker's
+client.
 """
 
 from temporalio.contrib.streams import StreamRef
@@ -40,6 +45,10 @@
     HTTPStatusError,
     TemporalStreamsHttpClient,
 )
+from temporalio.contrib.streams.nexus._handler import (
+    StreamAccess,
+    TemporalStreamsHandler,
+)
 
 __all__ = [
     "AppendInput",
@@ -48,8 +57,10 @@
     "ReadInput",
     "ReadOutput",
     "RecordWire",
+    "StreamAccess",
     "StreamCursor",
     "StreamRef",
     "TemporalStreams",
+    "TemporalStreamsHandler",
     "TemporalStreamsHttpClient",
 ]