Replayed Max's Option 7 (external streams) prototype onto the base.

CHANGELOG.md

@@ -92,6 +92,18 @@ to include examples, links to docs, or any other relevant information.
   stream's notification channel, reads on each delivery and completes when the stream closes.
   `temporalio.streams.providers.nexus_consumer_service` hosts it on its own and needs the
   `streams-nexus` extra.
+- Added experimental External Workflow Streams in
+  `temporalio.contrib.external_workflow_streams`. Workflow stream payloads are
+  stored in a configured external backend instead of Temporal History, with a
+  Redis Streams provider included. Workflows subscribe with `external_stream`,
+  external processes publish with `ExternalStreamProducer`, and Workers are
+  configured with `external_stream_backend`.
+- Added the output direction for External Workflow Streams.
+  Workflows publish with `external_output_stream`, Activities and external
+  processes use `ExternalOutputStreamProducer`, and external consumers resume
+  through `ExternalOutputStreamClient`. Workflow output is staged outside
+  History and becomes readable only after its compact Workflow Task marker is
+  committed.
 
 ### Changed
 

pyproject.toml

@@ -96,6 +96,7 @@ dev = [
   "pytest-rerunfailures>=16.1",
   "pytest-xdist>=3.6,<4",
   "moto[s3,server]>=5",
+  "redis>=5,<9",
   "langgraph>=1.1.0",
   "langsmith>=0.7.34,<0.9",
   "deepagents>=0.6.12,<0.7; python_version >= '3.11'",

temporalio/bridge/src/worker.rs

@@ -17,9 +17,9 @@ use temporalio_common::protos::temporal::api::history::v1::History;
 use temporalio_common::protos::temporal::api::worker::v1::{PluginInfo, StorageDriverInfo};
 use temporalio_sdk_core::replay::{HistoryForReplay, ReplayWorkerInput};
 use temporalio_sdk_core::{
-    PollError, SlotInfo, SlotInfoTrait, SlotKind, SlotKindType, SlotMarkUsedContext,
-    SlotReleaseContext, SlotReservationContext, SlotSupplier as SlotSupplierTrait,
-    SlotSupplierPermit, WorkflowErrorType,
+    ExternalStreamReadyResult, ExternalStreamRunStatus, PollError, SlotInfo, SlotInfoTrait,
+    SlotKind, SlotKindType, SlotMarkUsedContext, SlotReleaseContext, SlotReservationContext,
+    SlotSupplier as SlotSupplierTrait, SlotSupplierPermit, WorkflowErrorType,
 };
 use tokio::sync::mpsc::{channel, Sender};
 use tokio_stream::wrappers::ReceiverStream;
@@ -671,6 +671,63 @@ impl WorkerRef {
         Ok(())
     }
 
+    /// Tell Core a record is buffered for one external stream wait.
+    ///
+    /// **Synchronous and blocking**, like `record_activity_heartbeat`, but unlike it
+    /// **acknowledged**: the caller is a watcher whose next action -- keep waiting, re-probe, send
+    /// a wake Signal, or tear itself down -- depends entirely on which of the five results comes
+    /// back, so a fire-and-forget send would leave it unable to tell "Core will activate" from
+    /// "nothing here will".
+    ///
+    /// Synchronous rather than a Python awaitable on purpose. There is one watcher per
+    /// subscription and they do not share an event loop, and a future produced by
+    /// `future_into_py` is bound to the loop that created it -- so an awaitable here would work
+    /// from the worker's own loop and hang from anywhere else. Blocking with the GIL released is
+    /// safe: the answer comes off Core's serialized local-input lane, which does no I/O.
+    ///
+    /// Returns one of `Accepted`, `Stale`, `Parked`, `NoOpenWorkflowTask`, `RunNotFound`.
+    fn notify_external_stream_ready(
+        &self,
+        py: Python<'_>,
+        run_id: String,
+        wait_id: u32,
+        wait_generation: u64,
+    ) -> PyResult<&'static str> {
+        let worker = self.worker.as_ref().unwrap().clone();
+        // Deliberately *not* `enter_sync!`: entering the tokio handle would put this thread in an
+        // async context, and `Handle::block_on` panics there.
+        let handle = self.runtime.core.tokio_handle();
+        let result = py.detach(move || {
+            handle.block_on(worker.notify_external_stream_ready(&run_id, wait_id, wait_generation))
+        });
+        Ok(match result {
+            ExternalStreamReadyResult::Accepted => "Accepted",
+            ExternalStreamReadyResult::Stale => "Stale",
+            ExternalStreamReadyResult::Parked => "Parked",
+            ExternalStreamReadyResult::NoOpenWorkflowTask => "NoOpenWorkflowTask",
+            ExternalStreamReadyResult::RunNotFound => "RunNotFound",
+        })
+    }
+
+    /// Ask what state a run's external stream wait set is in, changing nothing.
+    ///
+    /// Deliberately not the readiness call: readiness means "a record is buffered", so probing
+    /// with it would assert something false and manufacture a spurious activation on the way out
+    /// of a shutting-down worker.
+    ///
+    /// Returns one of `WftOpen`, `Parked`, `NoOpenWorkflowTask`, `RunNotFound`.
+    fn external_stream_run_status(&self, py: Python<'_>, run_id: String) -> PyResult<&'static str> {
+        let worker = self.worker.as_ref().unwrap().clone();
+        let handle = self.runtime.core.tokio_handle();
+        let status = py.detach(move || handle.block_on(worker.external_stream_run_status(&run_id)));
+        Ok(match status {
+            ExternalStreamRunStatus::WftOpen => "WftOpen",
+            ExternalStreamRunStatus::Parked => "Parked",
+            ExternalStreamRunStatus::NoOpenWorkflowTask => "NoOpenWorkflowTask",
+            ExternalStreamRunStatus::RunNotFound => "RunNotFound",
+        })
+    }
+
     fn request_workflow_eviction(&self, run_id: &str) -> PyResult<()> {
         enter_sync!(self.runtime);
         self.worker

temporalio/bridge/worker.py

@@ -5,6 +5,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import enum
 from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import dataclass
 from typing import (
@@ -32,6 +34,62 @@
 from temporalio.worker._command_aware_visitor import CommandAwarePayloadVisitor
 
 
+class ExternalStreamReadyResult(enum.Enum):
+    """What Core did with an external stream readiness notification.
+
+    Five values, not two, because a watcher does something different with each
+    and an operator should conclude something different from each.
+
+    ``NO_OPEN_WORKFLOW_TASK`` and ``RUN_NOT_FOUND`` are separate on purpose. A
+    Run cached between Workflow Tasks is the healthy post-completion state;
+    reporting it as a missing Run would both corrupt the metric and tell the
+    watcher to tear itself down while it is still needed.
+    """
+
+    ACCEPTED = "Accepted"
+    """Serialized into an open Workflow Task. Core will activate; do nothing."""
+
+    STALE = "Stale"
+    """The wait moved on. Re-probe; do **not** send a Signal."""
+
+    PARKED = "Parked"
+    """A confirmed park generation exists. Send the reserved wake Signal."""
+
+    NO_OPEN_WORKFLOW_TASK = "NoOpenWorkflowTask"
+    """Cached with waits registered, no task open. Signal, and **keep** the watcher."""
+
+    RUN_NOT_FOUND = "RunNotFound"
+    """Absent from this Worker's cache. Signal, then tear the watcher down."""
+
+    @property
+    def needs_wake_signal(self) -> bool:
+        """Whether local readiness could not be delivered and a Signal is owed."""
+        return self in (
+            ExternalStreamReadyResult.PARKED,
+            ExternalStreamReadyResult.NO_OPEN_WORKFLOW_TASK,
+            ExternalStreamReadyResult.RUN_NOT_FOUND,
+        )
+
+
+class ExternalStreamRunStatus(enum.Enum):
+    """What state a Run's external stream wait set is in.
+
+    The answer to a **read-only** probe. Asking changes nothing.
+    """
+
+    WFT_OPEN = "WftOpen"
+    """A Workflow Task is open, retained by the wait set."""
+
+    PARKED = "Parked"
+    """The complete wait set is parked."""
+
+    NO_OPEN_WORKFLOW_TASK = "NoOpenWorkflowTask"
+    """Cached with active waits, but no Workflow Task open."""
+
+    RUN_NOT_FOUND = "RunNotFound"
+    """Absent from this Worker's cache."""
+
+
 @dataclass
 class WorkerConfig:
     """Python representation of the Rust struct for configuring a worker."""
@@ -259,6 +317,64 @@ def record_activity_heartbeat(
         """Record an activity heartbeat."""
         self._ref.record_activity_heartbeat(comp.SerializeToString())  # type: ignore[reportOptionalMemberAccess]
 
+    def notify_external_stream_ready_sync(
+        self, run_id: str, wait_id: int, wait_generation: int
+    ) -> ExternalStreamReadyResult:
+        """Tell Core a record is **buffered** for one external stream wait.
+
+        Buffered, not merely available: a notification for a record the caller
+        has not yet read into its buffer would produce an activation whose
+        drain must block, which is the deadlock hazard the out-of-thread buffer
+        exists to remove.
+
+        Acknowledged rather than fire-and-forget, because what the caller does
+        next depends entirely on the answer -- see
+        :py:class:`ExternalStreamReadyResult`.
+
+        Blocks while Core answers on its serialized local-input lane. Safe from
+        any thread, and safe to block in -- the lane does no I/O.
+
+        .. warning::
+            Do **not** call this from the event loop running the Worker. That
+            loop is what drives the poll this answer comes from, so blocking it
+            here waits for something only it can produce. Use
+            :py:meth:`notify_external_stream_ready` from a loop.
+        """
+        return ExternalStreamReadyResult(
+            self._ref.notify_external_stream_ready(  # type: ignore[reportOptionalMemberAccess]
+                run_id, wait_id, wait_generation
+            )
+        )
+
+    async def notify_external_stream_ready(
+        self, run_id: str, wait_id: int, wait_generation: int
+    ) -> ExternalStreamReadyResult:
+        """:py:meth:`notify_external_stream_ready_sync`, off the calling loop.
+
+        There is one watcher per subscription and they do not share a loop, so
+        this runs the blocking call in a thread rather than binding a future to
+        whichever loop happened to create it.
+        """
+        return await asyncio.to_thread(
+            self.notify_external_stream_ready_sync, run_id, wait_id, wait_generation
+        )
+
+    def external_stream_run_status_sync(self, run_id: str) -> ExternalStreamRunStatus:
+        """Ask what state a Run's external stream wait set is in, changing nothing.
+
+        A separate call from :py:meth:`notify_external_stream_ready` on purpose:
+        readiness means "a record is buffered", so probing with it would assert
+        something false and manufacture a spurious Workflow Task on the way out
+        of a shutting-down Worker.
+        """
+        return ExternalStreamRunStatus(
+            self._ref.external_stream_run_status(run_id)  # type: ignore[reportOptionalMemberAccess]
+        )
+
+    async def external_stream_run_status(self, run_id: str) -> ExternalStreamRunStatus:
+        """:py:meth:`external_stream_run_status_sync`, off the calling loop."""
+        return await asyncio.to_thread(self.external_stream_run_status_sync, run_id)
+
     def request_workflow_eviction(self, run_id: str) -> None:
         """Request a workflow be evicted."""
         self._ref.request_workflow_eviction(run_id)  # type: ignore[reportOptionalMemberAccess]