[Brook] Appended Workflow stream records at activation completion.

crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h

@@ -109,7 +109,8 @@ typedef struct TemporalCoreRuntime TemporalCoreRuntime;
 typedef struct TemporalCoreSlotReserveCompletionCtx TemporalCoreSlotReserveCompletionCtx;
 
 /**
- * One process's stream store, serving the stream calls lang makes.
+ * One process's stream store, serving the stream calls lang makes. Workers take it in their
+ * options, so a Workflow's output and outside producers share it.
  */
 typedef struct TemporalCoreStreamStore TemporalCoreStreamStore;
 
@@ -912,6 +913,12 @@ typedef struct TemporalCoreWorkerOptions {
    * a workflow task. Zero disables eager activity execution.
    */
   uint32_t max_eager_activity_reservations_per_workflow_task;
+  /**
+   * The store this worker appends workflow stream output to, or null for none. It must
+   * outlive the worker.
+   * NOTE: Experimental
+   */
+  const struct TemporalCoreStreamStore *stream_store;
 } TemporalCoreWorkerOptions;
 
 /**

crates/sdk-core-c-bridge/src/streams.rs

@@ -6,7 +6,8 @@ use prost::Message;
 use std::sync::Arc;
 use temporalio_sdk_core::streams::{StreamService, connect_stream_service, proto};
 
-/// One process's stream store, serving the stream calls lang makes.
+/// One process's stream store, serving the stream calls lang makes. Workers take it in their
+/// options, so a Workflow's output and outside producers share it.
 pub struct StreamStore {
     pub(crate) runtime: Runtime,
     pub(crate) service: Arc<StreamService>,

crates/sdk-core-c-bridge/src/worker.rs

@@ -61,6 +61,10 @@ pub struct WorkerOptions {
     /// Maximum number of activity slots that may be reserved for eager execution when completing
     /// a workflow task. Zero disables eager activity execution.
     pub max_eager_activity_reservations_per_workflow_task: u32,
+    /// The store this worker appends workflow stream output to, or null for none. It must
+    /// outlive the worker.
+    /// NOTE: Experimental
+    pub stream_store: *const crate::streams::StreamStore,
 }
 
 #[repr(C)]
@@ -1304,6 +1308,9 @@ impl TryFrom<&WorkerOptions> for temporalio_sdk_core::WorkerConfig {
                     .collect::<HashSet<_>>(),
             )
             .disable_payload_error_limit(opt.disable_payload_error_limit)
+            .maybe_stream_store(
+                unsafe { opt.stream_store.as_ref() }.map(|store| store.service.store()),
+            )
             .build()
             .map_err(|err| anyhow::anyhow!(err))
     }
@@ -1510,6 +1517,7 @@ mod tests {
             storage_drivers: crate::ByteArrayRefArray::empty(),
             disable_payload_error_limit: false,
             max_eager_activity_reservations_per_workflow_task: 3,
+            stream_store: std::ptr::null(),
         }
     }
 

crates/sdk-core/CHANGELOG.md

@@ -45,6 +45,19 @@ relevant information.
   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.
+* Experimental: a Workflow publishes stream records with `WorkflowOutputStreamCommit` in its
+  activation completion. The Worker appends them to its `stream_store` before it reports the
+  completion, so readers see them while the Workflow Task runs, and nothing goes to History. The
+  producer is `@<run id>`. The attempt is the last completed task's WorkflowTaskStarted event id
+  shifted left 16 bits, plus 1 and the task's failed or timed-out reruns since. Sequences start at
+  1 per attempt. Replay appends nothing. A store that refuses the append fails the Workflow Task.
+  Once the server accepts a task, Core retires the dedupe sessions of the run's earlier attempts.
+  The C bridge's worker options take the process's stream store as `stream_store`.
+* Experimental: a Workflow Task append that the store refuses as closed drops that topic's
+  records, logs a warning, counts `workflow_stream_output_dropped`, and the task goes on. A
+  retry would be refused the same way, also when the run closed the topic itself before its task
+  reran. Any other store failure still fails the task. Local Activities that a completion with
+  stream output schedules start only once that output is in the store.
 
 ### Breaking Changes
 * Replay detects changes to the service or operation of scheduled Nexus operations, reporting

crates/sdk-core/Cargo.toml

@@ -49,7 +49,8 @@ antithesis_assertions = ["dep:antithesis_sdk"]
 dynamic-tls = ["temporalio-client/dynamic-tls"]
 # Used by the integration-test runner to make Cloud exclusions native libtest ignores.
 cloud-test-mode = []
-# The stream store integration: lang reaches the store in process.
+# The stream store integration: a Worker appends a Workflow's stream output, and lang reaches the
+# store in process.
 streams = ["dep:temporalio-streams"]
 
 [dependencies]
@@ -115,7 +116,7 @@ tokio = { version = "1.47", default-features = false, features = [
   "process",
   "macros",
 ] }
-tokio-util = { version = "0.7", features = ["io", "io-util"] }
+tokio-util = { version = "0.7", features = ["io", "io-util", "rt"] }
 tokio-stream = { version = "0.1", default-features = false }
 tonic = { workspace = true, default-features = false, features = [
   "tls-native-roots",
@@ -155,6 +156,8 @@ path = "../macros"
 version = "~1.1.0"
 
 [dev-dependencies]
+# Turns on the stream store integration for this crate's own tests.
+temporalio-sdk-core = { path = ".", features = ["streams"] }
 assert_matches = "1.5"
 bimap = "0.6.3"
 bytes = "1.10"