diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs
index d094f8cdb..c8ddcecf3 100644
--- a/crates/e2e_test/src/fake_s3_target/mod.rs
+++ b/crates/e2e_test/src/fake_s3_target/mod.rs
@@ -76,6 +76,18 @@ const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-sourc
 const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] =
     ["x-rustfs-source-replication-request", "x-minio-source-replication-request"];
 const SOURCE_ETAG_HEADERS: [&str; 2] = ["x-rustfs-source-etag", "x-minio-source-etag"];
+const SOURCE_TAGGING_TIMESTAMP_HEADERS: [&str; 2] = [
+    "x-rustfs-source-replication-tagging-timestamp",
+    "x-minio-source-replication-tagging-timestamp",
+];
+const SOURCE_RETENTION_TIMESTAMP_HEADERS: [&str; 2] = [
+    "x-rustfs-source-replication-retention-timestamp",
+    "x-minio-source-replication-retention-timestamp",
+];
+const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [
+    "x-rustfs-source-replication-legalhold-timestamp",
+    "x-minio-source-replication-legalhold-timestamp",
+];
 const RESERVED_BUCKET_PREFIXES: [&str; 3] = ["xn--", "sthree-", "amzn-s3-demo-"];
 const RESERVED_BUCKET_SUFFIXES: [&str; 6] = ["-s3alias", "--ol-s3", ".mrap", "--x-s3", "--table-s3", "-an"];
 
@@ -118,6 +130,25 @@ pub enum FaultAction {
     WrongEtag,
 }
 
+/// Replication LWW timestamp headers observed on a request, journaled so
+/// sender-side tests can assert what a real target would receive.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ReplicationTimestampHeaders {
+    pub tagging: Option<String>,
+    pub retention: Option<String>,
+    pub legalhold: Option<String>,
+}
+
+impl ReplicationTimestampHeaders {
+    fn from_headers(headers: &HeaderMap) -> Self {
+        Self {
+            tagging: header_value(headers, &SOURCE_TAGGING_TIMESTAMP_HEADERS).map(bounded_journal_value),
+            retention: header_value(headers, &SOURCE_RETENTION_TIMESTAMP_HEADERS).map(bounded_journal_value),
+            legalhold: header_value(headers, &SOURCE_LEGALHOLD_TIMESTAMP_HEADERS).map(bounded_journal_value),
+        }
+    }
+}
+
 /// Credential-free request metadata retained for deterministic assertions.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RequestRecord {
@@ -131,6 +162,7 @@ pub struct RequestRecord {
     pub part_number: Option<i32>,
     pub content_length: Option<u64>,
     pub consumed_bytes: Option<usize>,
+    pub replication_timestamps: ReplicationTimestampHeaders,
     pub fault: Option<FaultAction>,
 }
 
@@ -536,7 +568,15 @@ impl S3Access for FaultAccess {
             .get(CONTENT_LENGTH)
             .and_then(|value| value.to_str().ok())
             .and_then(|value| value.parse().ok());
-        let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length);
+        let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
+        let fault = record_request(
+            &self.control,
+            operation,
+            context.method().clone(),
+            parsed,
+            content_length,
+            replication_timestamps,
+        );
         if let Some(RequestFault {
             action: FaultAction::Status(status),
             ..
@@ -589,6 +629,7 @@ fn record_request(
     method: Method,
     parsed: ParsedRequest,
     content_length: Option<u64>,
+    replication_timestamps: ReplicationTimestampHeaders,
 ) -> Option<RequestFault> {
     let mut state = lock(control);
     let action = parsed
@@ -613,6 +654,7 @@ fn record_request(
         part_number: parsed.part_number,
         content_length,
         consumed_bytes: None,
+        replication_timestamps,
         fault: action.clone(),
     });
     action.map(|action| RequestFault { sequence, action })
@@ -1699,6 +1741,52 @@ mod tests {
             .await?)
     }
 
+    #[tokio::test]
+    async fn journals_replication_timestamp_headers() -> Result<(), BoxError> {
+        let target = FakeS3Target::start().await?;
+        target.create_bucket("target-bucket");
+        let client = client(&target);
+
+        client
+            .put_object()
+            .bucket("target-bucket")
+            .key("plain")
+            .body(ByteStream::from_static(b"plain"))
+            .send()
+            .await?;
+        client
+            .put_object()
+            .bucket("target-bucket")
+            .key("stamped")
+            .body(ByteStream::from_static(b"stamped"))
+            .customize()
+            .map_request(move |mut request| {
+                let headers = request.headers_mut();
+                headers.insert("x-rustfs-source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
+                headers.insert("x-minio-source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
+                headers.insert("x-rustfs-source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
+                Ok::<_, std::convert::Infallible>(request)
+            })
+            .send()
+            .await?;
+
+        let requests = target.requests();
+        let plain = requests
+            .iter()
+            .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("plain"))
+            .expect("plain PUT must be journaled");
+        assert_eq!(plain.replication_timestamps, ReplicationTimestampHeaders::default());
+
+        let stamped = requests
+            .iter()
+            .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("stamped"))
+            .expect("stamped PUT must be journaled");
+        assert_eq!(stamped.replication_timestamps.tagging.as_deref(), Some("2026-01-02T03:04:05Z"));
+        assert_eq!(stamped.replication_timestamps.retention.as_deref(), Some("2026-01-02T03:04:06Z"));
+        assert_eq!(stamped.replication_timestamps.legalhold.as_deref(), Some("2026-01-02T03:04:07Z"));
+        Ok(())
+    }
+
     macro_rules! assert_sdk_error {
         ($error:expr, $status:expr, $code:expr) => {{
             let error = &$error;
@@ -2985,6 +3073,7 @@ mod tests {
                         part_number: None,
                     },
                     Some(0),
+                    ReplicationTimestampHeaders::default(),
                 );
             }
             let records = lock(&control).requests.clone();
@@ -3006,6 +3095,7 @@ mod tests {
                     part_number: None,
                 },
                 None,
+                ReplicationTimestampHeaders::default(),
             );
             {
                 let bounded_records = lock(&bounded_control);
diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs
index 1e1e18dc0..c6bcc5742 100644
--- a/crates/ecstore/src/bucket/bucket_target_sys.rs
+++ b/crates/ecstore/src/bucket/bucket_target_sys.rs
@@ -58,7 +58,9 @@ use rustfs_utils::http::{
 };
 use rustfs_utils::http::{
     SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
-    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
+    SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
+    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
+    insert_header,
 };
 use rustls_pki_types::pem::PemObject;
 use serde::{Deserialize, Serialize};
@@ -1476,9 +1478,12 @@ impl Default for AdvancedPutOptions {
             replication_status: ReplicationStatusType::Pending,
             source_mtime: OffsetDateTime::now_utc(),
             replication_request: false,
-            retention_timestamp: OffsetDateTime::now_utc(),
-            tagging_timestamp: OffsetDateTime::now_utc(),
-            legalhold_timestamp: OffsetDateTime::now_utc(),
+            // UNIX_EPOCH means "never modified": header() must not emit a
+            // timestamp header for it, otherwise a receiver would treat an
+            // unset category as a modification made right now.
+            retention_timestamp: OffsetDateTime::UNIX_EPOCH,
+            tagging_timestamp: OffsetDateTime::UNIX_EPOCH,
+            legalhold_timestamp: OffsetDateTime::UNIX_EPOCH,
             replication_validity_check: false,
         }
     }
@@ -1675,6 +1680,16 @@ impl PutObjectOptions {
             );
         }
 
+        for (suffix, timestamp) in [
+            (SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, self.internal.tagging_timestamp),
+            (SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, self.internal.retention_timestamp),
+            (SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, self.internal.legalhold_timestamp),
+        ] {
+            if timestamp.unix_timestamp() != 0 {
+                insert_header(&mut header, suffix, timestamp.format(&Rfc3339).unwrap_or_default());
+            }
+        }
+
         if self.internal.replication_request {
             insert_header(&mut header, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
         }
@@ -2842,6 +2857,57 @@ mod tests {
         );
     }
 
+    #[test]
+    fn put_object_headers_carry_replication_timestamp_headers() {
+        // MinIO receivers resolve concurrent tag/retention/legal-hold edits by
+        // last-writer-wins on these headers (object-api-options.go parses them
+        // as RFC3339); a replica without them loses every conflict resolution.
+        let mut opts = PutObjectOptions::default();
+        opts.internal.replication_request = true;
+        let tagging = OffsetDateTime::from_unix_timestamp(1_700_000_001).expect("valid timestamp");
+        let retention = OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("valid timestamp");
+        let legalhold = OffsetDateTime::from_unix_timestamp(1_700_000_003).expect("valid timestamp");
+        opts.internal.tagging_timestamp = tagging;
+        opts.internal.retention_timestamp = retention;
+        opts.internal.legalhold_timestamp = legalhold;
+
+        let header = opts.header();
+        for (suffix, expected) in [
+            ("source-replication-tagging-timestamp", tagging),
+            ("source-replication-retention-timestamp", retention),
+            ("source-replication-legalhold-timestamp", legalhold),
+        ] {
+            assert_eq!(
+                rustfs_utils::http::get_header(&header, suffix).as_deref(),
+                Some(expected.format(&Rfc3339).expect("RFC3339 timestamp").as_str()),
+                "replication put requests must carry the {suffix} header"
+            );
+        }
+    }
+
+    #[test]
+    fn put_object_headers_omit_unset_replication_timestamps() {
+        // UNIX_EPOCH means "never modified on the source"; sending it would
+        // make the receiver treat an unset category as a fresh modification.
+        let mut opts = PutObjectOptions::default();
+        opts.internal.replication_request = true;
+        opts.internal.tagging_timestamp = OffsetDateTime::UNIX_EPOCH;
+        opts.internal.retention_timestamp = OffsetDateTime::UNIX_EPOCH;
+        opts.internal.legalhold_timestamp = OffsetDateTime::UNIX_EPOCH;
+
+        let header = opts.header();
+        for suffix in [
+            "source-replication-tagging-timestamp",
+            "source-replication-retention-timestamp",
+            "source-replication-legalhold-timestamp",
+        ] {
+            assert!(
+                rustfs_utils::http::get_header(&header, suffix).is_none(),
+                "unset {suffix} must not be sent to replication targets"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn get_remote_target_client_internal_rejects_loopback_endpoint() {
         let sys = BucketTargetSys::default();
diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs
index 156c94cd7..e82ab7e52 100644
--- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs
+++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs
@@ -27,8 +27,10 @@ use rustfs_utils::http::{
     AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID,
     AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE,
     HeaderExt as _, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
-    SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map,
-    is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key, ssec_replication_transport_header,
+    SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
+    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP,
+    get_str, insert_header_map, is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key,
+    ssec_replication_transport_header,
 };
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
@@ -119,6 +121,27 @@ fn classify_replication_source_encryption(metadata: &HashMap<String, String>) ->
     }
 }
 
+fn is_legacy_source_replication_timestamp_key(key: &str) -> bool {
+    fn has_prefix_and_suffix(key: &str, prefix: &str, suffix: &str) -> bool {
+        let key = key.as_bytes();
+        key.len() == prefix.len() + suffix.len()
+            && key[..prefix.len()].eq_ignore_ascii_case(prefix.as_bytes())
+            && key[prefix.len()..].eq_ignore_ascii_case(suffix.as_bytes())
+    }
+
+    [
+        SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP,
+        SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
+        SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
+    ]
+    .iter()
+    .any(|suffix| {
+        ["x-rustfs-", "x-minio-"]
+            .iter()
+            .any(|prefix| has_prefix_and_suffix(key, prefix, suffix))
+    })
+}
+
 pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String, String>) -> bool {
     rustfs_replication::is_ssec_encrypted(user_defined)
 }
@@ -176,6 +199,11 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
             continue;
         }
 
+        if is_legacy_source_replication_timestamp_key(key) {
+            meta.insert(format!("x-amz-meta-{key}"), value.to_string());
+            continue;
+        }
+
         if is_internal_key(key) || is_standard_header(key) {
             continue;
         }
@@ -259,15 +287,23 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
 
         if !tags.is_empty() {
             put_options.user_tags = tags;
-            put_options.internal.tagging_timestamp =
-                if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) {
-                    OffsetDateTime::parse(&timestamp, &Rfc3339)
-                        .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
-                } else {
-                    object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-                };
         }
     }
+    // Load the stored tagging timestamp independently of whether any tags
+    // remain: DeleteObjectTagging leaves the object tagless but stamps this
+    // key, and the deletion's LWW timestamp must still reach the replica.
+    // With no stored key, fall back to mod_time only while tags exist
+    // (MinIO parity); a tagless object without the key was never tagged and
+    // keeps the epoch default (no header).
+    put_options.internal.tagging_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP)
+    {
+        OffsetDateTime::parse(&timestamp, &Rfc3339)
+            .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
+    } else if !put_options.user_tags.is_empty() {
+        object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
+    } else {
+        OffsetDateTime::UNIX_EPOCH
+    };
 
     let metadata = &*object_info.user_defined;
 
@@ -283,13 +319,15 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
         put_options.cache_control = cache_control.to_string();
     }
 
-    if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE) {
+    if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE).filter(|mode| !mode.is_empty()) {
         put_options.mode = Some(ObjectLockRetentionMode::from(mode.to_uppercase().as_str()));
     }
 
     if let Some(retain_until_date) = metadata.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
-        put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
-            .map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
+        if !retain_until_date.is_empty() {
+            put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
+                .map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
+        }
         put_options.internal.retention_timestamp =
             if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP) {
                 OffsetDateTime::parse(&timestamp, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH)
@@ -694,6 +732,110 @@ mod tests {
         assert!(options.internal.replication_request);
     }
 
+    /// DeleteObjectTagging leaves the object tagless but stamps the
+    /// tagging-timestamp internal key; the deletion's LWW timestamp must
+    /// still be loaded (and therefore sent) so the replica can order the
+    /// deletion against concurrent tag edits.
+    #[test]
+    fn replication_put_options_carry_tagging_timestamp_after_tag_deletion() {
+        let mut metadata = std::collections::HashMap::new();
+        rustfs_utils::http::insert_str(&mut metadata, SUFFIX_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
+
+        let object_info = ObjectInfo {
+            user_defined: Arc::new(metadata),
+            user_tags: Arc::new(String::new()),
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            version_id: Some(Uuid::nil()),
+            ..Default::default()
+        };
+
+        let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
+
+        assert!(options.user_tags.is_empty());
+        assert_eq!(
+            options.internal.tagging_timestamp,
+            OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp"),
+            "the stored tagging timestamp must load independently of remaining tags"
+        );
+
+        // A tagless object without the stored key was never tagged: the epoch
+        // default keeps the header unsent.
+        let untagged = ObjectInfo {
+            user_tags: Arc::new(String::new()),
+            mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
+            version_id: Some(Uuid::nil()),
+            ..Default::default()
+        };
+        let (options, _) = replication_put_object_options("", &untagged).expect("build put options");
+        assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
+    }
+
+    #[test]
+    fn replication_put_options_do_not_promote_legacy_user_timestamp_metadata() {
+        let legacy_keys = [
+            "x-rustfs-source-replication-tagging-timestamp",
+            "x-rustfs-source-replication-retention-timestamp",
+            "x-rustfs-source-replication-legalhold-timestamp",
+            "x-minio-source-replication-tagging-timestamp",
+            "x-minio-source-replication-retention-timestamp",
+            "x-minio-source-replication-legalhold-timestamp",
+        ];
+        let object_info = ObjectInfo {
+            user_defined: Arc::new(
+                legacy_keys
+                    .iter()
+                    .map(|key| (key.to_string(), "2099-01-02T03:04:05Z".to_string()))
+                    .collect(),
+            ),
+            ..Default::default()
+        };
+
+        let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
+
+        for legacy_key in legacy_keys {
+            assert!(!options.user_metadata.contains_key(legacy_key));
+            assert_eq!(
+                options
+                    .user_metadata
+                    .get(&format!("x-amz-meta-{legacy_key}"))
+                    .map(String::as_str),
+                Some("2099-01-02T03:04:05Z")
+            );
+        }
+        assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(options.internal.retention_timestamp, OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(options.internal.legalhold_timestamp, OffsetDateTime::UNIX_EPOCH);
+    }
+
+    #[test]
+    fn replication_put_options_carry_retention_timestamp_after_clear() {
+        let mut metadata = HashMap::from([
+            (AMZ_OBJECT_LOCK_MODE.to_string(), String::new()),
+            (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.to_string(), String::new()),
+        ]);
+        rustfs_utils::http::insert_str(&mut metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
+        let object_info = ObjectInfo {
+            user_defined: Arc::new(metadata),
+            ..Default::default()
+        };
+
+        let (options, _) = replication_put_object_options("", &object_info).expect("retention clear must replicate");
+
+        assert!(options.mode.is_none());
+        assert_eq!(options.retain_until_date, OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(
+            options.internal.retention_timestamp,
+            OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp")
+        );
+        let headers = options.header();
+        assert!(!headers.contains_key(AMZ_OBJECT_LOCK_MODE));
+        assert!(!headers.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE));
+        assert_eq!(
+            rustfs_utils::http::get_header(&headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP).as_deref(),
+            Some("2026-01-02T03:04:05Z")
+        );
+    }
+
     #[test]
     fn replication_put_options_strip_encryption_metadata_from_plaintext_objects() {
         use rustfs_utils::http::object_encryption_keys::{INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, SSEC_ORIGINAL_SIZE_HEADER};
diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs
index 291b9039a..624cb7593 100644
--- a/crates/ecstore/src/object_api/types.rs
+++ b/crates/ecstore/src/object_api/types.rs
@@ -277,6 +277,12 @@ pub struct ObjectOptions {
     /// fence avoids recursively acquiring the read lock behind a queued writer.
     pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
     pub replication_request: bool,
+    /// Source-cluster LWW timestamps carried by an authorized replication
+    /// request; None when the source never modified the category. Only the
+    /// replication-authorized options builders may set these.
+    pub replication_tagging_timestamp: Option<OffsetDateTime>,
+    pub replication_retention_timestamp: Option<OffsetDateTime>,
+    pub replication_legalhold_timestamp: Option<OffsetDateTime>,
     /// Authorized SSE-C replication passthrough: the body is already
     /// ciphertext, so the write path must not encrypt or compress it and
     /// stores the restored encryption metadata verbatim. Only the
diff --git a/crates/utils/src/http/header_compat.rs b/crates/utils/src/http/header_compat.rs
index 38486ca07..c36632d8e 100644
--- a/crates/utils/src/http/header_compat.rs
+++ b/crates/utils/src/http/header_compat.rs
@@ -50,6 +50,14 @@ pub const SUFFIX_SOURCE_DELETEMARKER: &str = "source-deletemarker";
 pub const SUFFIX_SOURCE_PROXY_REQUEST: &str = "source-proxy-request";
 pub const SUFFIX_SOURCE_REPLICATION_REQUEST: &str = "source-replication-request";
 pub const SUFFIX_SOURCE_REPLICATION_CHECK: &str = "source-replication-check";
+// LWW timestamps for replicated tag/retention/legal-hold modifications. MinIO
+// declares these with mixed case (internal/http/headers.go:
+// X-Minio-Source-Replication-Tagging-Timestamp / -Retention-Timestamp /
+// -LegalHold-Timestamp); HTTP header names compare case-insensitively, so the
+// lowercase suffix forms interoperate. Values are RFC3339 on the wire.
+pub const SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP: &str = "source-replication-tagging-timestamp";
+pub const SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP: &str = "source-replication-retention-timestamp";
+pub const SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP: &str = "source-replication-legalhold-timestamp";
 pub const SUFFIX_REPLICATION_SSEC_CRC: &str = "replication-ssec-crc";
 
 /// Returns true if the key is object-encryption metadata understood by RustFS or MinIO.
@@ -196,6 +204,27 @@ mod tests {
         assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42));
     }
 
+    #[test]
+    fn replication_timestamp_headers_match_minio_wire_names() {
+        let mut headers = HeaderMap::new();
+        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z");
+        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, "2026-01-02T03:04:06Z");
+        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, "2026-01-02T03:04:07Z");
+
+        // The exact names MinIO's object-api-options.go reads (its Get()
+        // canonicalizes case, so a case-insensitive match is wire-equivalent).
+        for name in [
+            "X-Minio-Source-Replication-Tagging-Timestamp",
+            "X-Minio-Source-Replication-Retention-Timestamp",
+            "X-Minio-Source-Replication-LegalHold-Timestamp",
+            "x-rustfs-source-replication-tagging-timestamp",
+            "x-rustfs-source-replication-retention-timestamp",
+            "x-rustfs-source-replication-legalhold-timestamp",
+        ] {
+            assert!(headers.contains_key(name), "replication timestamp header {name} must be written");
+        }
+    }
+
     #[test]
     fn test_get_header() {
         let mut headers = HeaderMap::new();
diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs
index 1b871a0df..a20b13188 100644
--- a/rustfs/src/app/multipart_usecase.rs
+++ b/rustfs/src/app/multipart_usecase.rs
@@ -44,8 +44,8 @@ use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, Wri
 use super::storage_api::multipart_usecase::object_utils::to_s3s_etag;
 use super::storage_api::multipart_usecase::options::{
     copy_src_opts, extract_metadata_from_mime, get_complete_multipart_upload_opts_with_replication_authorization,
-    get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, parse_copy_source_range,
-    put_opts_with_replication_authorization, validate_archive_content_encoding,
+    get_content_sha256_with_query, get_opts, has_replication_retention_update, namespace_reserved_user_metadata,
+    parse_copy_source_range, put_opts_with_replication_authorization, validate_archive_content_encoding,
 };
 use super::storage_api::multipart_usecase::request_context::spawn_traced_join;
 use super::storage_api::multipart_usecase::s3_api::multipart::{
@@ -777,7 +777,9 @@ impl DefaultMultipartUsecase {
 
         let mut metadata = create_multipart_upload_metadata(input_metadata, &req.headers, tagging, storage_class.as_ref());
 
-        let has_explicit_object_lock_retention = object_lock_mode.is_some() || object_lock_retain_until_date.is_some();
+        let has_explicit_object_lock_retention = object_lock_mode.is_some()
+            || object_lock_retain_until_date.is_some()
+            || has_replication_retention_update(&req.headers, replication_authorized);
         let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?;
         if let Some(object_lock_metadata) = build_put_like_object_lock_metadata(
             &bucket,
diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs
index 3bf35bc1b..913d6eb52 100644
--- a/rustfs/src/app/object_usecase.rs
+++ b/rustfs/src/app/object_usecase.rs
@@ -86,8 +86,8 @@ use super::storage_api::object_usecase::object_utils::to_s3s_etag;
 use super::storage_api::object_usecase::options::{
     copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata,
     extract_metadata_from_mime_with_object_name, filter_object_metadata, get_content_sha256_with_query, get_opts,
-    namespace_reserved_user_metadata, normalize_content_encoding_for_storage, preserve_unclassified_user_metadata,
-    put_opts_with_replication_authorization, validate_archive_content_encoding,
+    has_replication_retention_update, namespace_reserved_user_metadata, normalize_content_encoding_for_storage,
+    preserve_unclassified_user_metadata, put_opts_with_replication_authorization, validate_archive_content_encoding,
 };
 use super::storage_api::object_usecase::request_context::{self, spawn_traced, spawn_traced_join};
 use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params;
@@ -5799,7 +5799,9 @@ impl DefaultObjectUsecase {
         )?;
 
         let mut metadata = metadata.unwrap_or_default();
-        let has_explicit_object_lock_retention = object_lock_mode.is_some() || object_lock_retain_until_date.is_some();
+        let has_explicit_object_lock_retention = object_lock_mode.is_some()
+            || object_lock_retain_until_date.is_some()
+            || has_replication_retention_update(&req.headers, inbound_replication_put);
         let object_lock_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?;
         rustfs_io_metrics::record_put_object_stage_duration_from("app_object_lock_config_lookup", object_lock_config_stage_start);
@@ -10081,6 +10083,19 @@ mod tests {
         assert_eq!(metadata.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str), Some("COMPLIANCE"));
         assert!(metadata.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER));
         assert_eq!(metadata.get("x-amz-meta-x-amz-object-lock-mode").map(String::as_str), Some("GOVERNANCE"));
+
+        let mut replication_headers = HeaderMap::new();
+        insert_header(&mut replication_headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
+        insert_header(
+            &mut replication_headers,
+            rustfs_utils::http::SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
+            "2026-01-01T00:00:00Z",
+        );
+        let mut replica_metadata = HashMap::new();
+        let explicit_clear = has_replication_retention_update(&replication_headers, true);
+        apply_bucket_default_lock_retention("bucket", &state, &mut replica_metadata, explicit_clear).unwrap();
+        assert!(!replica_metadata.contains_key(AMZ_OBJECT_LOCK_MODE_LOWER));
+        assert!(!replica_metadata.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER));
     }
 
     fn pax_record(key: &str, value: &[u8]) -> Vec<u8> {
diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs
index b97b16bcb..839b8a38f 100644
--- a/rustfs/src/app/storage_api.rs
+++ b/rustfs/src/app/storage_api.rs
@@ -996,8 +996,9 @@ pub(crate) mod options {
         copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata,
         extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata,
         get_complete_multipart_upload_opts_with_replication_authorization, get_content_sha256_with_query, get_opts,
-        namespace_reserved_user_metadata, normalize_content_encoding_for_storage, parse_copy_source_range,
-        preserve_unclassified_user_metadata, put_opts_with_replication_authorization, validate_archive_content_encoding,
+        has_replication_retention_update, namespace_reserved_user_metadata, normalize_content_encoding_for_storage,
+        parse_copy_source_range, preserve_unclassified_user_metadata, put_opts_with_replication_authorization,
+        validate_archive_content_encoding,
     };
 }
 
diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs
index ce4be49d5..9369d7224 100644
--- a/rustfs/src/storage/ecfs.rs
+++ b/rustfs/src/storage/ecfs.rs
@@ -45,7 +45,7 @@ use rustfs_targets::EventName;
 use rustfs_utils::http::headers::{
     AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
 };
-use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, insert_str};
+use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, insert_str};
 use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error};
 use std::collections::HashMap;
 use std::fmt::Debug;
@@ -461,6 +461,11 @@ impl S3 for FS {
             let mut eval_metadata = HashMap::new();
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default());
+            insert_str(
+                &mut eval_metadata,
+                SUFFIX_TAGGING_TIMESTAMP,
+                OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(),
+            );
             opts.eval_metadata = Some(eval_metadata);
         }
 
@@ -1645,6 +1650,11 @@ impl S3 for FS {
             let mut eval_metadata = HashMap::new();
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default());
+            insert_str(
+                &mut eval_metadata,
+                SUFFIX_TAGGING_TIMESTAMP,
+                OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(),
+            );
             opts.eval_metadata = Some(eval_metadata);
         }
 
diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs
index 609b2ea62..ac920b5c3 100644
--- a/rustfs/src/storage/options.rs
+++ b/rustfs/src/storage/options.rs
@@ -17,11 +17,13 @@ use crate::storage::storage_api::options_consumer::contract::{object::HTTPPrecon
 use http::header::{IF_MATCH, IF_NONE_MATCH};
 use http::{HeaderMap, HeaderValue};
 use rustfs_utils::http::{
-    AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
-    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST,
-    SUFFIX_SOURCE_VERSION_ID, get_header,
+    AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
+    SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
+    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
+    SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header,
     header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
-    insert_header_map,
+    insert_header_map, insert_str,
     metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX},
 };
 use rustfs_utils::http::{
@@ -422,6 +424,9 @@ pub fn get_complete_multipart_upload_opts_with_replication_authorization(
         preserve_etag,
         ..Default::default()
     };
+    if replication_request {
+        apply_replication_timestamps_from_headers(headers, &mut opts);
+    }
     apply_replica_status_from_headers(headers, &mut opts, replication_request_authorized);
 
     fill_conditional_writes_opts_from_header(headers, &mut opts)?;
@@ -458,6 +463,12 @@ pub fn put_opts_from_headers(headers: &HeaderMap<HeaderValue>, metadata: HashMap
     put_opts_from_headers_with_replication_authorization(headers, metadata, false)
 }
 
+pub(crate) fn has_replication_retention_update(headers: &HeaderMap<HeaderValue>, replication_request_authorized: bool) -> bool {
+    replication_request_authorized
+        && get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true")
+        && replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP).is_some()
+}
+
 pub fn put_opts_from_headers_with_replication_authorization(
     headers: &HeaderMap<HeaderValue>,
     metadata: HashMap<String, String>,
@@ -479,6 +490,7 @@ pub fn put_opts_from_headers_with_replication_authorization(
         if let Some(crc) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) {
             insert_header_map(&mut opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC, crc.into_owned());
         }
+        apply_replication_timestamps_from_headers(headers, &mut opts);
     }
     Ok(opts)
 }
@@ -504,6 +516,47 @@ fn replication_source_mtime(headers: &HeaderMap<HeaderValue>) -> Option<time::Of
     }
 }
 
+/// Parses one replication LWW timestamp header. Invalid values are dropped
+/// with a warning (same tolerance as [`replication_source_mtime`]) so a
+/// malformed source header cannot wedge the replication queue.
+fn replication_timestamp_header(headers: &HeaderMap<HeaderValue>, suffix: &str) -> Option<time::OffsetDateTime> {
+    let value = get_header(headers, suffix)?;
+    let value = value.trim();
+    match time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339) {
+        Ok(timestamp) => Some(timestamp),
+        Err(err) => {
+            tracing::warn!("Invalid {} value '{}' (replication request=true): {}", suffix, value, err);
+            None
+        }
+    }
+}
+
+/// Callers must gate on an authorized replication request: these headers are
+/// trusted source-cluster state, not client input.
+fn apply_replication_timestamps_from_headers(headers: &HeaderMap<HeaderValue>, opts: &mut ObjectOptions) {
+    opts.replication_tagging_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP);
+    opts.replication_retention_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP);
+    opts.replication_legalhold_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP);
+
+    // Persist into the internal metadata keys so a later outbound replication
+    // pass (replication_target_boundary) reads the source's modification
+    // times instead of falling back to mod_time.
+    // TODO(P1-6): receiver-side LWW is still missing — when the stored
+    // per-category timestamp is newer than the inbound one, the existing
+    // tags/retention/legal-hold should win instead of being overwritten.
+    for (timestamp, suffix) in [
+        (opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP),
+        (opts.replication_retention_timestamp, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP),
+        (opts.replication_legalhold_timestamp, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP),
+    ] {
+        if let Some(timestamp) = timestamp
+            && let Ok(value) = timestamp.format(&time::format_description::well_known::Rfc3339)
+        {
+            insert_str(&mut opts.user_defined, suffix, value);
+        }
+    }
+}
+
 fn apply_replica_status_from_headers(headers: &HeaderMap<HeaderValue>, opts: &mut ObjectOptions, authorized: bool) {
     if !authorized {
         return;
@@ -663,6 +716,13 @@ fn is_reserved_user_metadata_key(key: &str) -> bool {
         || starts_with_ignore_ascii_case(key, MINIO_INTERNAL_PREFIX)
         || starts_with_ignore_ascii_case(key, RUSTFS_ENCRYPTION_PREFIX)
         || starts_with_ignore_ascii_case(key, MINIO_ENCRYPTION_PREFIX)
+        // Replication transport names (source-replication timestamps,
+        // source-mtime/-etag/-version-id, ...). A bare stored key with one of
+        // these names is forwarded verbatim by the outbound replication
+        // header builder on a server-authorized request, so the receiver
+        // would persist attacker-chosen values as trusted internal LWW state.
+        || starts_with_ignore_ascii_case(key, "x-rustfs-source-")
+        || starts_with_ignore_ascii_case(key, "x-minio-source-")
 }
 
 fn stored_user_metadata_key(key: &str) -> String {
@@ -1082,15 +1142,16 @@ mod tests {
         del_opts_with_versioning, detect_content_type_from_object_name, extract_metadata, extract_metadata_from_mime,
         extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts,
         get_complete_multipart_upload_opts_with_replication_authorization, get_default_opts, get_opts,
-        namespace_reserved_user_metadata, parse_copy_source_range, put_opts, put_opts_from_headers,
-        put_opts_from_headers_with_replication_authorization, put_opts_with_replication_authorization,
+        has_replication_retention_update, namespace_reserved_user_metadata, parse_copy_source_range, put_opts,
+        put_opts_from_headers, put_opts_from_headers_with_replication_authorization, put_opts_with_replication_authorization,
         validate_archive_content_encoding,
     };
     use http::{HeaderMap, HeaderValue};
     use rustfs_utils::http::{
         AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER,
         AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG,
-        SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
+        SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
+        SUFFIX_SOURCE_VERSION_ID, insert_header,
     };
     use s3s::S3ErrorCode;
     use s3s::dto::{BucketVersioningStatus, ExcludedPrefix, VersioningConfiguration};
@@ -1520,6 +1581,24 @@ mod tests {
         assert!(opts.preserve_etag.is_none());
     }
 
+    #[test]
+    fn test_replication_retention_update_requires_authorization() {
+        let mut headers = HeaderMap::new();
+        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
+        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, "2026-01-01T00:00:00Z");
+
+        assert!(!has_replication_retention_update(&headers, false));
+        assert!(has_replication_retention_update(&headers, true));
+
+        let mut missing_request = HeaderMap::new();
+        insert_header(
+            &mut missing_request,
+            SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
+            "2026-01-01T00:00:00Z",
+        );
+        assert!(!has_replication_retention_update(&missing_request, true));
+    }
+
     #[test]
     fn test_put_opts_from_headers_gates_ssec_passthrough_on_authorization() {
         use rustfs_utils::http::object_encryption_keys::{
@@ -1599,6 +1678,136 @@ mod tests {
         assert!(opts_invalid.mod_time.is_none());
     }
 
+    /// A client PUT must not materialize the replication transport names as
+    /// bare stored user-metadata keys: the outbound replication header
+    /// builder forwards user metadata verbatim on a server-authorized
+    /// request, so a bare `x-rustfs-source-replication-*-timestamp` key would
+    /// deliver an attacker-chosen value into the replica's trusted internal
+    /// LWW state (for a tagless object nothing later overwrites it).
+    #[test]
+    fn test_replication_transport_names_cannot_be_forged_via_user_metadata() {
+        let mut headers = HeaderMap::new();
+        for name in [
+            "x-amz-meta-x-rustfs-source-replication-tagging-timestamp",
+            "x-amz-meta-x-minio-source-replication-legalhold-timestamp",
+            "x-rustfs-meta-x-rustfs-source-replication-retention-timestamp",
+            "x-amz-meta-x-rustfs-source-mtime",
+        ] {
+            headers.insert(
+                http::header::HeaderName::from_static(name),
+                HeaderValue::from_static("2026-01-02T03:04:05Z"),
+            );
+        }
+
+        let metadata = extract_metadata(&headers);
+
+        for forged in [
+            "x-rustfs-source-replication-tagging-timestamp",
+            "x-minio-source-replication-legalhold-timestamp",
+            "x-rustfs-source-replication-retention-timestamp",
+            "x-rustfs-source-mtime",
+        ] {
+            assert!(
+                !metadata.contains_key(forged),
+                "{forged} must not be storable as a bare user-metadata key"
+            );
+        }
+        // The values survive, namespaced back under the user-metadata prefix.
+        assert_eq!(
+            metadata
+                .get("x-amz-meta-x-rustfs-source-replication-tagging-timestamp")
+                .map(String::as_str),
+            Some("2026-01-02T03:04:05Z")
+        );
+    }
+
+    #[test]
+    fn test_put_opts_from_headers_gates_replication_timestamp_persistence_on_authorization() {
+        // Sender-side LWW state (replication_target_boundary.rs) is read back
+        // from these internal metadata keys, so an authorized replication PUT
+        // must persist the inbound timestamp headers; an unauthorized client
+        // must not be able to forge them.
+        let mut headers = HeaderMap::new();
+        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
+        insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
+        insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
+        insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
+
+        let untrusted = put_opts_from_headers(&headers, HashMap::new()).expect("ordinary PUT options should be created");
+        for suffix in [
+            "tagging-timestamp",
+            "objectlock-retention-timestamp",
+            "objectlock-legalhold-timestamp",
+        ] {
+            assert!(
+                rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(),
+                "unauthorized clients must not persist the {suffix} internal key"
+            );
+        }
+        assert!(untrusted.replication_tagging_timestamp.is_none());
+        assert!(untrusted.replication_retention_timestamp.is_none());
+        assert!(untrusted.replication_legalhold_timestamp.is_none());
+
+        let trusted = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true)
+            .expect("authorized replication request should parse");
+        let parse = |value: &str| {
+            time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).expect("valid RFC3339")
+        };
+        assert_eq!(trusted.replication_tagging_timestamp, Some(parse("2026-01-02T03:04:05Z")));
+        assert_eq!(trusted.replication_retention_timestamp, Some(parse("2026-01-02T03:04:06Z")));
+        assert_eq!(trusted.replication_legalhold_timestamp, Some(parse("2026-01-02T03:04:07Z")));
+        for (suffix, expected) in [
+            ("tagging-timestamp", "2026-01-02T03:04:05Z"),
+            ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"),
+            ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"),
+        ] {
+            assert_eq!(
+                rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(),
+                Some(expected),
+                "authorized replication must persist the {suffix} internal key"
+            );
+        }
+    }
+
+    #[test]
+    fn test_complete_multipart_opts_persist_replication_timestamps_when_authorized() {
+        let mut headers = HeaderMap::new();
+        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
+        insert_header(&mut headers, "replication-actual-object-size", "1");
+        insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
+        insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
+        insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
+
+        let untrusted = get_complete_multipart_upload_opts(&headers).expect("ordinary multipart options should be created");
+        for suffix in [
+            "tagging-timestamp",
+            "objectlock-retention-timestamp",
+            "objectlock-legalhold-timestamp",
+        ] {
+            assert!(
+                rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(),
+                "unauthorized multipart completes must not persist the {suffix} internal key"
+            );
+        }
+
+        let trusted = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true)
+            .expect("authorized multipart complete should parse");
+        for (suffix, expected) in [
+            ("tagging-timestamp", "2026-01-02T03:04:05Z"),
+            ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"),
+            ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"),
+        ] {
+            assert_eq!(
+                rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(),
+                Some(expected),
+                "authorized multipart completes must persist the {suffix} internal key"
+            );
+        }
+        assert!(trusted.replication_tagging_timestamp.is_some());
+        assert!(trusted.replication_retention_timestamp.is_some());
+        assert!(trusted.replication_legalhold_timestamp.is_some());
+    }
+
     #[test]
     fn test_put_opts_from_headers_with_replica_status() {
         let mut headers = HeaderMap::new();
diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs
index f4f084a31..5d217f18e 100644
--- a/rustfs/src/storage/storage_api.rs
+++ b/rustfs/src/storage/storage_api.rs
@@ -185,8 +185,9 @@ pub(crate) mod options_consumer {
         copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata,
         extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata,
         get_complete_multipart_upload_opts_with_replication_authorization, get_content_sha256_with_query, get_opts,
-        namespace_reserved_user_metadata, normalize_content_encoding_for_storage, parse_copy_source_range,
-        preserve_unclassified_user_metadata, put_opts_with_replication_authorization, validate_archive_content_encoding,
+        has_replication_retention_update, namespace_reserved_user_metadata, normalize_content_encoding_for_storage,
+        parse_copy_source_range, preserve_unclassified_user_metadata, put_opts_with_replication_authorization,
+        validate_archive_content_encoding,
     };
 
     pub(crate) mod contract {