lint and format

Author: mjameswh

arch_docs/workflow_task_chunking.md

@@ -4,20 +4,20 @@ One source of complexity in Core is the chunking of history into "logical" Workf
 
 Workflow tasks (WFTs) always take the following form in event history:
 
-* \[Preceding Events\] (optional)
-* WFT Scheduled
-* WFT Started
-* WFT Completed
-* \[Commands\] (optional)
+- \[Preceding Events\] (optional)
+- WFT Scheduled
+- WFT Started
+- WFT Completed
+- \[Commands\] (optional)
 
 In the typical case, the "logical" WFT consists of all the commands from the last workflow task,
 any events generated in the interrim, and the scheduled/started preamble. So:
 
-* WFT Completed
-* \[Commands\] (optional)
-* \[Events\] (optional)
-* WFT Scheduled
-* WFT Started
+- WFT Completed
+- \[Commands\] (optional)
+- \[Events\] (optional)
+- WFT Scheduled
+- WFT Started
 
 Commands and events are both "optional" in the sense that:
 
@@ -32,8 +32,22 @@ There may be no events for more nuanced reasons:
    rather as a "protocol message" attached to the task.
 3. Server can forcibly generate a new WFT with some obscure APIs
 
-Core does not consider such empty WFT sequences as worthy of waking lang (on replay - as a new
-task, they always will), since nothing meaningful has happened. Thus, they are grouped together
+## Handling of empty WFTs (March 2026)
+
+Until now, Core would try to avoid waking lang when replaying empty WFT sequences (i.e. a WFT
+Completed immediately followed by a WFT Scheduled and WFT Started, with no commands and no events
+in-between), since they would presumably be no-ops and therefore waste execution resources. Instead,
+the second WFT (and potentially subsequent WFTs) would be collapsed into the first, resulting in a
+single WFT that has the inbound events and start time of the first WFT, but the resulting commands
+of the last WFT in the collapsed sequence.
+
+It was found that there are some particular edge cases where WFT collapsing might result in
+incorrect replay behavior. One particular example of this is the case where an update request would
+be sent after an empty WFT. This led to the development of some heuristics to avoid incorrect
+chunking in presence of known problematic patterns. Unfortunately, it has been found that these
+heuristics may
+
+Thus, they are grouped together task, they always will), since nothing meaningful has happened. Thus, they are grouped together
 as part of a "logical" WFT with the last WFT that had any real work in it.
 
 ## Possible issues as of this writing (5/25)
@@ -44,8 +58,8 @@ NDE.
 
 ### Possible solutions
 
-* Core can attach a flag on WFT completes in order to be explicit that that WFT may be skipped on
+- Core can attach a flag on WFT completes in order to be explicit that that WFT may be skipped on
   replay. IE: During WFT heartbeating for LAs.
-* We could legislate that server should never send empty WFTs. Seemingly the only case of this
+- We could legislate that server should never send empty WFTs. Seemingly the only case of this
   is
   the [obscure api](https://github.com/temporalio/temporal/blob/d189737aa2ed1b07c221abb9fbdd28ecf68f0492/proto/internal/temporal/server/api/adminservice/v1/service.proto#L151)

crates/sdk-core/src/core_tests/workflow_tasks.rs

@@ -2463,7 +2463,7 @@ async fn core_internal_flags() {
                 .iter()
                 .copied()
                 .collect::<HashSet<_>>(),
-            CoreInternalFlags::all_except_too_high()
+            CoreInternalFlags::all_except_unknown()
                 .map(|f| f as u32)
                 .collect()
         );

crates/sdk-core/src/internal_flags.rs

@@ -25,9 +25,11 @@ pub enum CoreInternalFlags {
     /// In this flag additional checks were added to a number of state machines to ensure that
     /// the ID and type of activities, local activities, and child workflows match during replay.
     IdAndTypeDeterminismChecks = 1,
+
     /// Introduced automatically upserting search attributes for each patched call, and
     /// nondeterminism checks for upserts.
     UpsertSearchAttributeOnPatch = 2,
+
     /// Prior to this flag, we truncated commands received from lang at the
     /// first terminal (i.e. workflow-terminating) command. With this flag, we
     /// reorder commands such that all non-terminal commands come first,
@@ -37,8 +39,57 @@ pub enum CoreInternalFlags {
     /// if in the sequence delivered by lang they came after a terminal command.
     /// See <https://github.com/temporalio/features/issues/481>.
     MoveTerminalCommands = 3,
+
+    /// Indicates that this workflow uses the improved WFT heartbeat heuristic that
+    /// is more conservative when deciding whether to collapse consecutive empty WFTs.
+    /// The improved heuristic avoids incorrectly collapsing WFTs that carry updates,
+    /// and requires more look-ahead before committing to a chunking decision when
+    /// history is paginated.
+    ///
+    /// Only set on workflows started after this flag was introduced (first-WFT-only).
+    /// Existing workflows continue to use the legacy heuristic.
+    ImprovedHeartbeatHeuristic = 4,
+
     /// We received a value higher than this code can understand.
-    TooHigh = u32::MAX,
+    UnknownFlag = u32::MAX,
+}
+
+impl CoreInternalFlags {
+    fn from_u32(v: u32) -> Self {
+        match v {
+            1 => Self::IdAndTypeDeterminismChecks,
+            2 => Self::UpsertSearchAttributeOnPatch,
+            3 => Self::MoveTerminalCommands,
+            4 => Self::ImprovedHeartbeatHeuristic,
+            _ => Self::UnknownFlag,
+        }
+    }
+
+    /// Returns all cumulative flags that should be enabled by default on every WFT completion.
+    pub(crate) fn all_cumulative_default_enabled() -> impl Iterator<Item = CoreInternalFlags> {
+        [
+            CoreInternalFlags::IdAndTypeDeterminismChecks,
+            CoreInternalFlags::UpsertSearchAttributeOnPatch,
+            CoreInternalFlags::MoveTerminalCommands,
+        ]
+        .iter()
+        .copied()
+    }
+
+    /// Returns cumulative flags that should only be enabled on the first WFT of new workflows.
+    /// These are not written on subsequent WFTs to avoid changing behavior of existing workflows.
+    pub(crate) fn all_first_wft_only_default_enabled() -> impl Iterator<Item = CoreInternalFlags> {
+        [CoreInternalFlags::ImprovedHeartbeatHeuristic]
+            .iter()
+            .copied()
+    }
+
+    /// Returns all known flag variants (excluding the sentinel).
+    #[cfg(test)]
+    pub(crate) fn all_except_unknown() -> impl Iterator<Item = CoreInternalFlags> {
+        enum_iterator::all::<CoreInternalFlags>()
+            .filter(|f| !matches!(f, CoreInternalFlags::UnknownFlag))
+    }
 }
 
 #[allow(clippy::large_enum_variant)]
@@ -137,15 +188,19 @@ impl InternalFlags {
         }
     }
 
-    /// Writes all known core flags to the set which should be recorded in the current WFT if not
-    /// already known. Must only be called if not replaying.
-    pub(crate) fn write_all_known(&mut self) {
+    /// Writes all core flags that should be enabled by default to the set which should be recorded
+    /// in the current WFT if not already known. Must only be called if not replaying.
+    pub(crate) fn write_all_cumulative_default_enabled(&mut self, is_first_wft: bool) {
         if let Self::Enabled {
             core_since_last_complete,
             ..
         } = self
         {
-            core_since_last_complete.extend(CoreInternalFlags::all_except_too_high());
+            if is_first_wft {
+                core_since_last_complete
+                    .extend(CoreInternalFlags::all_first_wft_only_default_enabled());
+            }
+            core_since_last_complete.extend(CoreInternalFlags::all_cumulative_default_enabled());
         }
     }
 
@@ -214,22 +269,6 @@ impl InternalFlags {
     }
 }
 
-impl CoreInternalFlags {
-    fn from_u32(v: u32) -> Self {
-        match v {
-            1 => Self::IdAndTypeDeterminismChecks,
-            2 => Self::UpsertSearchAttributeOnPatch,
-            3 => Self::MoveTerminalCommands,
-            _ => Self::TooHigh,
-        }
-    }
-
-    pub(crate) fn all_except_too_high() -> impl Iterator<Item = CoreInternalFlags> {
-        enum_iterator::all::<CoreInternalFlags>()
-            .filter(|f| !matches!(f, CoreInternalFlags::TooHigh))
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -266,7 +305,7 @@ mod tests {
 
     #[test]
     fn all_have_u32_from_impl() {
-        let all_known = CoreInternalFlags::all_except_too_high();
+        let all_known = CoreInternalFlags::all_except_unknown();
         for flag in all_known {
             let as_u32 = flag as u32;
             assert_eq!(CoreInternalFlags::from_u32(as_u32), flag);

crates/sdk-core/src/replay/history_builder.rs

@@ -1,4 +1,4 @@
-use crate::replay::HistoryInfo;
+use crate::{internal_flags::CoreInternalFlags, replay::HistoryInfo};
 use anyhow::bail;
 use prost_types::Timestamp;
 use std::{
@@ -48,6 +48,11 @@ pub struct TestHistoryBuilder {
     final_workflow_task_started_event_id: i64,
     previous_task_completed_id: i64,
     original_run_id: String,
+    /// When true, the builder auto-sets `ImprovedHeartbeatHeuristic` on the first WFTCompleted.
+    use_improved_heartbeat_heuristic: bool,
+    /// Tracks whether we've already emitted the first WFTCompleted
+    /// (to know when to set the cumulative `ImprovedHeartbeatHeuristic` flag).
+    has_seen_wft_completed: bool,
 }
 
 impl TestHistoryBuilder {
@@ -69,6 +74,8 @@ impl TestHistoryBuilder {
             original_run_id: extract_original_run_id_from_events(&events)
                 .expect("Run id must be discoverable")
                 .to_string(),
+            use_improved_heartbeat_heuristic: false,
+            has_seen_wft_completed: false,
             events,
         }
     }
@@ -89,6 +96,17 @@ impl TestHistoryBuilder {
         self.add(attribs)
     }
 
+    /// Enable the improved heartbeat heuristic flag for this builder. When enabled,
+    /// the first WFTCompleted event gets the cumulative `ImprovedHeartbeatHeuristic` flag.
+    pub fn set_use_improved_heartbeat_heuristic(&mut self) {
+        if self.has_seen_wft_completed {
+            panic!(
+                "Improved heartbeat heuristic can only be enabled before the first WFTCompleted"
+            );
+        }
+        self.use_improved_heartbeat_heuristic = true;
+    }
+
     /// Adds the following events:
     /// ```text
     /// EVENT_TYPE_WORKFLOW_TASK_SCHEDULED
@@ -127,6 +145,13 @@ impl TestHistoryBuilder {
             ..Default::default()
         });
         self.previous_task_completed_id = id;
+        if self.use_improved_heartbeat_heuristic && !self.has_seen_wft_completed {
+            self.set_flags_on_wft_completed(
+                id,
+                &[CoreInternalFlags::ImprovedHeartbeatHeuristic as u32],
+            );
+        }
+        self.has_seen_wft_completed = true;
     }
 
     /// Add a workflow task timed out event.
@@ -608,6 +633,17 @@ impl TestHistoryBuilder {
         Self::set_flags(self.events.iter_mut().rev(), core, lang)
     }
 
+    /// Sets core flags on the WFTCompleted event with the given event ID.
+    pub fn set_flags_on_wft_completed(&mut self, event_id: i64, core: &[u32]) {
+        if let Some(event) = self.events.iter_mut().find(|e| e.event_id == event_id)
+            && let Some(Attributes::WorkflowTaskCompletedEventAttributes(ref mut a)) =
+                event.attributes
+        {
+            let sdk_dat = a.sdk_metadata.get_or_insert_with(Default::default);
+            sdk_dat.core_used_flags.extend_from_slice(core);
+        }
+    }
+
     /// Get the event ID of the most recently added event
     pub fn current_event_id(&self) -> i64 {
         self.current_event_id