diff --git a/CHANGELOG.md b/CHANGELOG.md
index 47d06c3..9709329 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+## 4.10.1
+
+- Preserve valid outer workflow-failure messages when a nested exception message
+  is absent or blank, matching published Server fallback semantics. Keep supplied
+  small report fields unchanged, while reducing redundant outer copies when a
+  lower encoded-body budget requires it. Preserve type, retry flags and claim IDs.
+
 ## 4.10.0
 
 - Bound outgoing activity, workflow-task and terminal workflow failure messages,
diff --git a/Cargo.lock b/Cargo.lock
index 30659a8..57f7c0f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -250,7 +250,7 @@ dependencies = [
 
 [[package]]
 name = "durable-workflow"
-version = "4.10.0"
+version = "4.10.1"
 dependencies = [
  "apache-avro",
  "base64 0.22.1",
diff --git a/Cargo.toml b/Cargo.toml
index be663b6..1bdad67 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "durable-workflow"
-version = "4.10.0"
+version = "4.10.1"
 edition = "2021"
 description = "Rust client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 license = "MIT"
@@ -14,7 +14,7 @@ categories = ["api-bindings", "asynchronous"]
 include = ["/src/**", "/schema/**", "/examples/**", "/Cargo.toml", "/README.md", "/CHANGELOG.md", "/LICENSE", "/tests/fixtures/replay-regressions/published-php-complete-explicit-backoff.json", "/tests/fixtures/sdk-metrics-contract.json"]
 
 [package.metadata.durable-workflow]
-product-train = "4.10.0"
+product-train = "4.10.1"
 compatibility-authority = "protocol-manifests"
 supported-server-versions = "2.0.0"
 qualified-server-version = "2.8.0"
diff --git a/src/failure_reporting.rs b/src/failure_reporting.rs
index 4d6b445..de97a3a 100644
--- a/src/failure_reporting.rs
+++ b/src/failure_reporting.rs
@@ -62,10 +62,15 @@ fn clipped(value: &str, limit: usize, marker: &str, force: bool) -> String {
     format!("{}{marker}", prefix(value, available))
 }
 
+#[cfg(test)]
 fn metadata_size(value: &Map<String, Value>) -> Result<usize> {
+    metadata_size_with_message(value, false)
+}
+
+fn metadata_size_with_message(value: &Map<String, Value>, include_message: bool) -> Result<usize> {
     let metadata: Map<_, _> = value
         .iter()
-        .filter(|(key, _)| !core(key))
+        .filter(|(key, _)| !core(key) || (include_message && key.as_str() == "message"))
         .map(|(key, value)| (key.clone(), value.clone()))
         .collect();
     encoded_size(&metadata)
@@ -237,6 +242,7 @@ struct Failure {
     metadata_bytes: usize,
     field_sizes: Vec<usize>,
     inline_diagnostics: bool,
+    fallback: Option<(String, String)>,
 }
 impl Failure {
     fn take(
@@ -244,7 +250,24 @@ impl Failure {
         pointer: String,
         limit: usize,
         inline_diagnostics: bool,
+        fallback_pointer: Option<String>,
     ) -> Result<Self> {
+        let fallback = if let Some(pointer) = fallback_pointer {
+            let value = body
+                .pointer_mut(&pointer)
+                .ok_or(FailureReportError::InvalidReport(
+                    "missing fallback message",
+                ))?
+                .take();
+            let Value::String(message) = value else {
+                return Err(
+                    FailureReportError::InvalidReport("fallback message must be a string").into(),
+                );
+            };
+            Some((pointer, message))
+        } else {
+            None
+        };
         let value = body
             .pointer_mut(&pointer)
             .ok_or(FailureReportError::InvalidReport("missing failure object"))?
@@ -252,7 +275,7 @@ impl Failure {
         let Value::Object(original) = value else {
             return Err(FailureReportError::InvalidReport("failure must be an object").into());
         };
-        if !original.get("message").is_some_and(Value::is_string) {
+        if fallback.is_none() && !original.get("message").is_some_and(Value::is_string) {
             return Err(
                 FailureReportError::InvalidReport("failure.message must be a string").into(),
             );
@@ -288,7 +311,7 @@ impl Failure {
             if core(key) && !matches!(key.as_str(), "message" | "stack_trace") {
                 required_bytes = required_bytes.saturating_add(size).saturating_add(1);
             }
-            if !core(key) {
+            if !core(key) || (fallback.is_some() && key == "message") {
                 metadata_bytes = metadata_bytes
                     .saturating_add(size)
                     .saturating_add(usize::from(metadata_fields > 0));
@@ -308,35 +331,20 @@ impl Failure {
             metadata_bytes,
             field_sizes,
             inline_diagnostics,
+            fallback,
         })
     }
 
-    fn render(&self, budget: Budget) -> Result<Value> {
-        let message = self.original["message"]
-            .as_str()
-            .expect("validated failure message");
-        let stack = self.original.get("stack_trace").and_then(Value::as_str);
-        let mut truncated = message.len() > budget.message
-            || stack.is_some_and(|s| budget.stack == 0 || s.len() > budget.stack)
-            || self.metadata_bytes > budget.metadata;
-        // Preserve byte ordering too when serde_json's preserve_order feature is
-        // enabled by a downstream application.
-        if !truncated {
-            return Ok(Value::Object(self.original.clone()));
-        }
+    fn bounded_fields(&self, budget: Budget) -> Result<(Map<String, Value>, Omitted)> {
         let mut omitted = Omitted::new(budget.names);
         let mut output = Map::new();
         let mut used = 2_usize;
-        // Reserve space for diagnostics only when a limit already requires a marker.
-        let metadata_limit =
-            budget
-                .metadata
-                .saturating_sub(if truncated { MARKER_RESERVE } else { 0 });
+        let metadata_limit = budget.metadata.saturating_sub(MARKER_RESERVE);
         for (index, (key, value)) in self.original.iter().enumerate() {
-            if matches!(key.as_str(), "message" | "stack_trace") {
+            if key == "stack_trace" || (key == "message" && self.fallback.is_none()) {
                 continue;
             }
-            if core(key) {
+            if core(key) && !(key == "message" && self.fallback.is_some()) {
                 output.insert(key.clone(), value.clone());
                 continue;
             }
@@ -345,10 +353,35 @@ impl Failure {
                 used += size;
                 output.insert(key.clone(), value.clone());
             } else {
-                truncated = true;
                 omitted.add(key)?;
             }
         }
+        Ok((output, omitted))
+    }
+
+    fn render(&self, budget: Budget) -> Result<(Value, Option<Value>)> {
+        let message = self
+            .fallback
+            .as_ref()
+            .map(|(_, message)| message.as_str())
+            .unwrap_or_else(|| {
+                self.original["message"]
+                    .as_str()
+                    .expect("validated failure message")
+            });
+        let stack = self.original.get("stack_trace").and_then(Value::as_str);
+        let truncated = message.len() > budget.message
+            || stack.is_some_and(|s| budget.stack == 0 || s.len() > budget.stack)
+            || self.metadata_bytes > budget.metadata;
+        // Preserve byte ordering too when serde_json's preserve_order feature is
+        // enabled by a downstream application.
+        if !truncated {
+            return Ok((
+                Value::Object(self.original.clone()),
+                self.fallback.as_ref().map(|_| json!(message)),
+            ));
+        }
+        let (mut output, mut omitted) = self.bounded_fields(budget)?;
         if let Some(stack) = stack {
             let marker = format!(
                 "\n[truncated: original stack_trace {} UTF-8 bytes]",
@@ -398,19 +431,28 @@ impl Failure {
             );
             output.insert("diagnostics".into(), Value::Object(diagnostics));
         }
-        output.insert(
-            "message".into(),
-            Value::String(clipped(message, budget.message, &message_marker, truncated)),
-        );
+        let rendered_message =
+            Value::String(clipped(message, budget.message, &message_marker, truncated));
+        let fallback_message = if self.fallback.is_some() {
+            // Keep an ignored nested value PHP-blank after omission. Adding a
+            // marker there would change which message Server makes authoritative.
+            if self.original.contains_key("message") && !output.contains_key("message") {
+                output.insert("message".into(), json!(""));
+            }
+            Some(rendered_message)
+        } else {
+            output.insert("message".into(), rendered_message);
+            None
+        };
         // Count the actual serialized metadata after inserting markers, including
         // escaped field names and any wrapper around existing diagnostics.
-        if metadata_size(&output)? > budget.metadata {
+        if metadata_size_with_message(&output, self.fallback.is_some())? > budget.metadata {
             return Err(FailureReportError::InvalidReport(
                 "truncation diagnostics exceed their serialized budget",
             )
             .into());
         }
-        Ok(Value::Object(output))
+        Ok((Value::Object(output), fallback_message))
     }
 
     fn refresh_details(&mut self, body: &Value) -> Result<()> {
@@ -432,63 +474,98 @@ impl Failure {
 }
 
 /// Captures originals once so later transport-budget reductions retain true sizes.
+struct MessageMirror {
+    target: String,
+    source: String,
+    original_bytes: usize,
+}
+
+#[derive(Default)]
+struct ReportPositions {
+    pointers: Vec<(String, Option<String>)>,
+    texts: Vec<(String, LocalText)>,
+    mirrors: Vec<MessageMirror>,
+    guarded: bool,
+}
+impl ReportPositions {
+    fn complete(&mut self, body: &Value) {
+        for (index, command) in body["commands"]
+            .as_array()
+            .into_iter()
+            .flatten()
+            .enumerate()
+        {
+            if command["type"] == "fail_workflow" || command["type"] == "fail_update" {
+                self.guarded = true;
+                if command["exception"]["message"]
+                    .as_str()
+                    .is_some_and(|message| !committed_message(message).is_empty())
+                {
+                    self.pointers
+                        .push((format!("/commands/{index}/exception"), None));
+                    if command.get("message").is_some() {
+                        self.mirrors.push(MessageMirror {
+                            target: format!("/commands/{index}/message"),
+                            source: format!("/commands/{index}/exception/message"),
+                            original_bytes: command["message"].as_str().map_or(0, str::len),
+                        });
+                    }
+                } else if let Some(message) = command["message"].as_str() {
+                    if command["exception"].is_object() {
+                        self.pointers.push((
+                            format!("/commands/{index}/exception"),
+                            Some(format!("/commands/{index}/message")),
+                        ));
+                    } else {
+                        self.texts.push((
+                            format!("/commands/{index}/message"),
+                            LocalText::new(message),
+                        ));
+                    }
+                }
+            }
+            if command["type"] == "record_local_activity" {
+                self.guarded |= command.get("message").is_some()
+                    || command["attempts"]
+                        .as_array()
+                        .into_iter()
+                        .flatten()
+                        .any(|v| v.get("message").is_some());
+            }
+        }
+    }
+}
+
 pub(super) struct Prepared {
     failures: Vec<Failure>,
     texts: Vec<(String, LocalText)>,
-    mirrors: Vec<(String, String)>,
+    mirrors: Vec<MessageMirror>,
 }
 impl Prepared {
     pub(super) fn new(body: &mut Value, path: &str, limit: usize) -> Result<Option<Self>> {
         let parts: Vec<_> = path.trim_matches('/').split('/').collect();
-        let mut pointers = Vec::new();
-        let mut texts = Vec::new();
-        let mut mirrors = Vec::new();
-        let mut local_reports = false;
+        let mut positions = ReportPositions::default();
         match parts.as_slice() {
             ["worker", "activity-tasks" | "workflow-tasks" | "query-tasks", _, "fail"] => {
-                local_reports = true;
+                positions.guarded = true;
                 // Preserve Server validation and ordinary payload handling for
                 // malformed raw requests; their final body is still guarded.
                 if body["failure"]["message"].is_string() {
-                    pointers.push("/failure".into());
+                    positions.pointers.push(("/failure".into(), None));
                 }
             }
             ["worker", "workflow-tasks", _, "complete"] => {
-                for (index, command) in body["commands"]
-                    .as_array()
-                    .into_iter()
-                    .flatten()
-                    .enumerate()
-                {
-                    if command["type"] == "fail_workflow" || command["type"] == "fail_update" {
-                        if command["exception"]["message"].is_string() {
-                            pointers.push(format!("/commands/{index}/exception"));
-                            if command.get("message").is_some() {
-                                mirrors.push((
-                                    format!("/commands/{index}/message"),
-                                    format!("/commands/{index}/exception/message"),
-                                ));
-                            }
-                        } else if let Some(message) = command["message"].as_str() {
-                            texts.push((
-                                format!("/commands/{index}/message"),
-                                LocalText::new(message),
-                            ));
-                        }
-                    }
-                    if command["type"] == "record_local_activity" {
-                        local_reports |= command.get("message").is_some()
-                            || command["attempts"]
-                                .as_array()
-                                .into_iter()
-                                .flatten()
-                                .any(|v| v.get("message").is_some());
-                    }
-                }
+                positions.complete(body);
             }
             _ => return Ok(None),
         }
-        if pointers.is_empty() && texts.is_empty() && !local_reports {
+        let ReportPositions {
+            pointers,
+            texts,
+            mirrors,
+            guarded,
+        } = positions;
+        if pointers.is_empty() && texts.is_empty() && !guarded {
             return Ok(None);
         }
         if path.len() > limit {
@@ -503,8 +580,14 @@ impl Prepared {
             parts.as_slice(),
             ["worker", "workflow-tasks" | "query-tasks", _, "fail"]
         );
-        for pointer in pointers {
-            failures.push(Failure::take(body, pointer, limit, inline_diagnostics)?);
+        for (pointer, fallback) in pointers {
+            failures.push(Failure::take(
+                body,
+                pointer,
+                limit,
+                inline_diagnostics,
+                fallback,
+            )?);
         }
         let prepared = Self {
             failures,
@@ -517,21 +600,35 @@ impl Prepared {
 
     fn render(&self, body: &mut Value, budget: Budget) -> Result<()> {
         for failure in &self.failures {
+            let (rendered, fallback_message) = failure.render(budget)?;
             *body
                 .pointer_mut(&failure.pointer)
-                .expect("retained failure location") = failure.render(budget)?;
+                .expect("retained failure location") = rendered;
+            if let (Some((pointer, _)), Some(message)) = (&failure.fallback, fallback_message) {
+                *body
+                    .pointer_mut(pointer)
+                    .expect("retained fallback message") = message;
+            }
         }
         for (pointer, original) in &self.texts {
             *body
                 .pointer_mut(pointer)
                 .expect("retained message location") = json!(original.render(budget.message));
         }
-        for (target, source) in &self.mirrors {
+        for mirror in &self.mirrors {
+            // Server selects a nonblank nested message. Preserve supplied small
+            // fields, reducing only a redundant outer copy that exceeds this
+            // iteration's budget (which may reflect a lower transport limit).
+            if mirror.original_bytes <= budget.message {
+                continue;
+            }
             let message = body
-                .pointer(source)
+                .pointer(&mirror.source)
                 .expect("retained exception message")
                 .clone();
-            *body.pointer_mut(target).expect("retained command message") = message;
+            *body
+                .pointer_mut(&mirror.target)
+                .expect("retained command message") = message;
         }
         Ok(())
     }
@@ -748,14 +845,41 @@ mod tests {
             );
             assert_eq!(body["commands"][0]["non_retryable"], true);
             if structured {
-                assert_eq!(
-                    body["commands"][0]["message"],
-                    body["commands"][0]["exception"]["message"]
-                );
+                assert_eq!(body["commands"][0]["exception"]["message"], "");
+                assert!(body["commands"][0]["message"]
+                    .as_str()
+                    .unwrap()
+                    .contains("[truncated"));
+                assert!(body["commands"][0]["exception"]["diagnostics"][MARKER_KEY]
+                    ["omitted_fields"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .any(|v| v == "message"));
             }
         }
     }
 
+    #[test]
+    fn nonblank_nested_message_and_redundant_outer_copy_fit_after_escaping() {
+        let original = "\u{0000}界🦀".repeat(400_000);
+        let mut body = json!({"lease_owner":"owner","commands":[{"type":"fail_workflow","message":original,
+            "exception":{"type":"Original","message":original,"stack_trace":"\u{0001}".repeat(3*1024*1024)}}]});
+        let mut report = Prepared::new(&mut body, "/worker/workflow-tasks/task/complete", 4096)
+            .unwrap()
+            .unwrap();
+        let bytes = report.encode(&mut body, 4096).unwrap();
+        assert!(bytes.len() <= 4096);
+        assert_eq!(
+            body["commands"][0]["message"],
+            body["commands"][0]["exception"]["message"]
+        );
+        assert!(body["commands"][0]["message"]
+            .as_str()
+            .unwrap()
+            .contains("[truncated"));
+    }
+
     #[test]
     fn local_attempt_messages_fit_before_observation_without_mutating_prior_commands() {
         let original = "\u{0000}界🦀".repeat(400_000);
diff --git a/tests/failure_fallback.rs b/tests/failure_fallback.rs
new file mode 100644
index 0000000..e1ddbf9
--- /dev/null
+++ b/tests/failure_fallback.rs
@@ -0,0 +1,42 @@
+mod failure_reporting_support;
+use durable_workflow::{json, Client};
+use failure_reporting_support::{Server, MIB};
+#[tokio::test]
+async fn small_terminal_failure_preserves_outer_fallback_and_all_supplied_fields() {
+    let server = Server::start(2 * MIB, vec![]).await;
+    let client = Client::new(&server.url).unwrap();
+    let commands = vec![
+        json!({"type":"fail_workflow","message":"outer fallback","exception_type":"Original","non_retryable":true,"exception":{"type":"Original","message":""}}),
+        json!({"type":"fail_workflow","message":"outer fallback","exception_type":"Original","exception":{"type":"Original","message":" \t\u{0000}"}}),
+        json!({"type":"fail_workflow","message":"outer supplied","exception":{"type":"Original","message":"nested authoritative"}}),
+        json!({"type":"fail_workflow","message":"outer fallback","exception":{"type":"Original"}}),
+    ];
+    for command in &commands {
+        client
+            .complete_workflow_task("task", "owner", 4, vec![command.clone()])
+            .await
+            .unwrap();
+    }
+    let reports = server.reports();
+    assert_eq!(reports.len(), commands.len());
+    for ((_, _, body), command) in reports.into_iter().zip(commands) {
+        assert_eq!(body["commands"][0],command,"small valid reports keep their original wire values; published Server chooses nonblank nested message or the outer fallback");
+    }
+}
+
+#[tokio::test]
+async fn redundant_outer_message_reduces_only_when_the_lower_encoded_budget_requires_it() {
+    let server = Server::start(4096, vec![]).await;
+    let client = Client::new(&server.url).unwrap();
+    client.cluster_info().await.unwrap();
+    client.complete_workflow_task("exact-task","exact-owner",4,vec![json!({"type":"fail_workflow","message":"\u{0000}".repeat(2000),
+        "exception_type":"Original","non_retryable":true,"exception":{"type":"Original","message":"nested authoritative"}})]).await.unwrap();
+    let reports = server.reports();
+    assert_eq!(reports.len(), 1);
+    assert!(reports[0].1.len() <= 4096);
+    let body = &reports[0].2;
+    assert_eq!(body["lease_owner"], "exact-owner");
+    assert_eq!(body["workflow_task_attempt"], 4);
+    assert_eq!(body["commands"][0]["message"], "nested authoritative");
+    assert_eq!(body["commands"][0]["non_retryable"], true);
+}
diff --git a/tests/failure_reporting.rs b/tests/failure_reporting.rs
index 31d7856..a69f835 100644
--- a/tests/failure_reporting.rs
+++ b/tests/failure_reporting.rs
@@ -83,7 +83,8 @@ async fn workflow_task_and_terminal_workflow_report_large_failure() {
     }
     assert_eq!(reports[0].2["failure"]["type"], "RustWorkflowTaskFailure");
     let command = &reports[1].2["commands"][0];
-    assert_eq!(command["message"], command["exception"]["message"]);
+    assert_eq!(command["exception"]["message"], "");
+    assert!(command["message"].as_str().unwrap().contains("[truncated"));
     assert_eq!(command["exception_type"], "Original");
     assert_eq!(command["non_retryable"], true);
 }
diff --git a/tests/failure_reporting_server.rs b/tests/failure_reporting_server.rs
index f39f283..8f2591b 100644
--- a/tests/failure_reporting_server.rs
+++ b/tests/failure_reporting_server.rs
@@ -335,3 +335,56 @@ async fn workflow_task_failure_records_large_stack_and_omitted_valid_details() {
         .is_some_and(|id| id != task.task_id));
     println!("published workflow-task failure passed:3MiB message/stack/valid details, recorded receipt, exact identity and retry task; originaltype=OriginalTaskFailure,task_id={}",task.task_id);
 }
+
+#[tokio::test]
+#[ignore = "requires isolated published Server"]
+async fn terminal_failure_preserves_blank_fallback_and_nonblank_nested_authority() {
+    let client = client();
+    let queue = format!("rust-fallback-{}", durable_workflow::Uuid::new_v4());
+    let worker_id = format!("{queue}-worker");
+    let mut worker = Worker::new(client.clone(), &queue).worker_id(&worker_id);
+    worker.register_workflow("tests.failure-fallback", |_, _| async { Ok(json!(null)) });
+    worker.register().await.unwrap();
+    for (index, (nested, expected)) in [
+        ("", "outer fallback"),
+        (" \t\0", "outer fallback"),
+        ("nested authoritative", "nested authoritative"),
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let workflow_id = format!("{queue}-{index}");
+        let handle = client
+            .start_workflow("tests.failure-fallback", &queue, &workflow_id, json!([]))
+            .await
+            .unwrap();
+        let task = tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                if let Some(task) = client
+                    .poll_workflow_task(&worker_id, &queue, Duration::from_millis(10))
+                    .await
+                    .unwrap()
+                {
+                    break task;
+                }
+            }
+        })
+        .await
+        .unwrap();
+        client.complete_workflow_task(&task.task_id,task.lease_owner.as_deref().unwrap(),task.workflow_task_attempt,
+            vec![json!({"type":"fail_workflow","message":"outer fallback","exception_type":"OriginalFallbackType","non_retryable":true,
+                "exception":{"type":"OriginalFallbackType","message":nested}})]).await.unwrap();
+        let events = history(&client, &handle).await;
+        let failed = events
+            .iter()
+            .find(|event| event["event_type"] == "WorkflowFailed")
+            .expect("valid fallback failure recorded");
+        assert_eq!(failed["payload"]["message"], expected);
+        assert_eq!(failed["payload"]["exception_type"], "OriginalFallbackType");
+        assert_eq!(
+            handle.describe().await.unwrap().status.as_deref(),
+            Some("failed")
+        );
+    }
+    println!("published terminal fallback passed: empty/whitespace nested values preserve outer message; nonblank nested value is authoritative; original type and claim identity retained");
+}
diff --git a/tests/fixtures/grouped-condition-worker-consumer/Cargo.lock b/tests/fixtures/grouped-condition-worker-consumer/Cargo.lock
index 29834a4..3b0b047 100644
--- a/tests/fixtures/grouped-condition-worker-consumer/Cargo.lock
+++ b/tests/fixtures/grouped-condition-worker-consumer/Cargo.lock
@@ -250,7 +250,7 @@ dependencies = [
 
 [[package]]
 name = "durable-workflow"
-version = "4.10.0"
+version = "4.10.1"
 dependencies = [
  "apache-avro",
  "base64 0.22.1",
diff --git a/tests/nexus/Cargo.lock b/tests/nexus/Cargo.lock
index 30659a8..57f7c0f 100644
--- a/tests/nexus/Cargo.lock
+++ b/tests/nexus/Cargo.lock
@@ -250,7 +250,7 @@ dependencies = [
 
 [[package]]
 name = "durable-workflow"
-version = "4.10.0"
+version = "4.10.1"
 dependencies = [
  "apache-avro",
  "base64 0.22.1",