From 29889d7955733bb179c5639b0fafe9692a8786c1 Mon Sep 17 00:00:00 2001
From: Mohammad Dashti <mohammad.dashti@temporal.io>
Date: Wed, 14 Oct 2026 14:47:26 -0700
Subject: [PATCH] Added the Workflow output stream commit to the activation
 completion.

Lang sends a Workflow's stream records with its completion, so Core can append them before it reports the Workflow Task. The commit carries no marker and puts nothing in History.
---
 .../workflow_commands/workflow_commands.proto | 25 +++++++++++++++++++
 crates/protos/src/protos/mod.rs               | 10 ++++++++
 .../workflow/machines/workflow_machines.rs    |  6 +++++
 crates/sdk-core/src/worker/workflow/mod.rs    |  5 ++++
 4 files changed, 46 insertions(+)

diff --git a/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto b/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto
index 2db4ccfb6f..9cae959d4b 100644
--- a/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto
+++ b/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto
@@ -20,6 +20,7 @@ import "temporal/api/workflow/v1/message.proto";
 import "temporal/sdk/core/child_workflow/child_workflow.proto";
 import "temporal/sdk/core/nexus/nexus.proto";
 import "temporal/sdk/core/common/common.proto";
+import "temporal/sdk/streams/v1/message.proto";
 
 message WorkflowCommand {
     // User metadata that may or may not be persisted into history depending on the command type.
@@ -55,9 +56,33 @@ message WorkflowCommand {
         UpdateResponse update_response = 20;
         ScheduleNexusOperation schedule_nexus_operation = 21;
         RequestCancelNexusOperation request_cancel_nexus_operation = 22;
+        WorkflowOutputStreamCommit workflow_output_stream_commit = 23;
     }
 }
 
+// The stream records one activation published, sent with its completion.
+//
+// Core appends them to the Worker's stream store before it reports the completion to the
+// server, so readers see them before the Workflow Task completes. Nothing about them goes to
+// History. Lang sends no commit while it replays, and Core drops one it gets then, since the
+// records are in the store already. Core takes at most one commit per completion.
+message WorkflowOutputStreamCommit {
+    repeated OutputRecord records = 1;
+}
+
+// One record a Workflow published, in publish order across topics.
+message OutputRecord {
+    string topic = 1;
+    // DATA or FINISH.
+    temporal.sdk.streams.v1.StreamRecordKind kind = 2;
+    // The value on DATA, after lang's payload codec. Absent on FINISH.
+    temporal.api.common.v1.Payload body = 3;
+    // The SHA-256 of the body as the payload converter made it, before the codec. Set on DATA.
+    // Core builds each topic's batch digest from these, so a repeat of a batch matches whatever
+    // the codec made of the bodies.
+    bytes content_hash = 4;
+}
+
 message StartTimer {
     // Lang's incremental sequence number, used as the operation identifier
     uint32 seq = 1;
diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs
index ffd7640263..26d77ce504 100644
--- a/crates/protos/src/protos/mod.rs
+++ b/crates/protos/src/protos/mod.rs
@@ -1620,6 +1620,16 @@ pub mod coresdk {
                 }
             }
 
+            impl Display for WorkflowOutputStreamCommit {
+                fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+                    write!(
+                        f,
+                        "WorkflowOutputStreamCommit({} record(s))",
+                        self.records.len()
+                    )
+                }
+            }
+
             impl QueryResult {
                 /// Helper to construct the Temporal API query result types.
                 pub fn into_components(
diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs
index 6c1b1a59f7..dabbfa0521 100644
--- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs
+++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs
@@ -1586,6 +1586,12 @@ impl WorkflowMachines {
                         annotations
                     );
                 }
+                WFCommandVariant::OutputStreamCommit(_) => {
+                    return Err(fatal!(
+                        "WorkflowOutputStreamCommit reached the state machines. The run takes it \
+                         out when it processes the completion"
+                    ));
+                }
                 WFCommandVariant::NoCommandsFromLang => (),
             }
         }
diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs
index 4dc05d1556..7bbb8164b4 100644
--- a/crates/sdk-core/src/worker/workflow/mod.rs
+++ b/crates/sdk-core/src/worker/workflow/mod.rs
@@ -1593,6 +1593,8 @@ enum WFCommandVariant {
     UpdateResponse(UpdateResponse),
     ScheduleNexusOperation(ScheduleNexusOperation),
     RequestCancelNexusOperation(RequestCancelNexusOperation),
+    /// The stream records lang published in this activation.
+    OutputStreamCommit(WorkflowOutputStreamCommit),
 }
 
 impl TryFrom<WorkflowCommand> for WFCommand {
@@ -1654,6 +1656,9 @@ impl TryFrom<WorkflowCommand> for WFCommand {
             workflow_command::Variant::RequestCancelNexusOperation(s) => {
                 WFCommandVariant::RequestCancelNexusOperation(s)
             }
+            workflow_command::Variant::WorkflowOutputStreamCommit(commit) => {
+                WFCommandVariant::OutputStreamCommit(commit)
+            }
         };
         Ok(Self {
             variant,