[SDK Sentinel] Prevent replay shutdown from dropping nondeterminism results

CHANGELOG.md

@@ -56,6 +56,8 @@ relevant information.
 
 
 ### Fixed
+* Batch workflow replay no longer reports a final nondeterministic history as successful when
+  shutdown races with delivery of its replay result.
 * Ephemeral server startup now enforces its elapsed-time deadline even when a connection stalls,
   and startup timeout messages include the last connection error when available.
 * Avoid scanning the workflow cache when no workflow tasks are waiting for a cache slot, reducing

crates/sdk-core/CHANGELOG.md

@@ -45,6 +45,8 @@ relevant information.
   nondeterminism while retaining compatibility with histories recorded before these checks.
 
 ### Fixed
+* Replay workers now deliver a failed history's eviction before polling the next history, preventing
+  shutdown from dropping the final nondeterminism result.
 * Dev and test server startup now stops waiting at the configured deadline even if a readiness
   request stalls. Timeout messages include the last connection failure when available.
 * Workers with large workflow caches avoid unnecessary CPU usage when no workflow tasks are waiting

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

@@ -32,6 +32,7 @@ use temporalio_common::{
         coresdk::workflow_activation::remove_from_cache::EvictionReason,
         temporal::api::{
             common::v1::WorkflowExecution,
+            enums::v1::WorkflowTaskFailedCause,
             history::v1::History,
             workflowservice::v1::{
                 DescribeNamespaceResponse, RespondWorkflowTaskCompletedResponse,
@@ -136,8 +137,10 @@ where
             });
         client
             .expect_fail_workflow_task()
-            .returning(move |_, _, _| {
-                hist_allow_tx.send("Failed".to_string()).unwrap();
+            .returning(move |_, cause, _| {
+                if cause != WorkflowTaskFailedCause::NonDeterministicError {
+                    hist_allow_tx.send("Failed".to_string()).unwrap();
+                }
                 async move { Ok(RespondWorkflowTaskFailedResponse::default()) }.boxed()
             });
         let mut worker = Worker::new(self.config, None, Arc::new(client), None, None, None)?;