diff --git a/xet_client/src/cas_client/simulation/deletion_controls.rs b/xet_client/src/cas_client/simulation/deletion_controls.rs
index 35f4585cb..bc8a3ca10 100644
--- a/xet_client/src/cas_client/simulation/deletion_controls.rs
+++ b/xet_client/src/cas_client/simulation/deletion_controls.rs
@@ -4,11 +4,35 @@ use xet_core_structures::merklehash::MerkleHash;
 
 use crate::error::Result;
 
-/// An opaque 32-byte tag used for conditional deletion (compare-and-delete).
+/// An opaque 32-byte etag used for conditional deletion (compare-and-delete), standing in
+/// for S3's ETag.
 ///
-/// Implementations should derive this from object metadata/content with enough entropy
-/// to reduce false matches when objects are rapidly rewritten.
-pub type ObjectTag = [u8; 32];
+/// Derived from what is stored, not from when it was written: a byte-identical re-upload
+/// leaves it unchanged, as S3's content-derived ETag does. A caller that needs to tell a
+/// re-upload apart cannot do it from this value — see [`LAST_UPLOAD_TAG_KEY`].
+pub type ObjectETag = [u8; 32];
+
+/// S3-style `(key, value)` tag set attached to an object.
+///
+/// Distinct from [`ObjectETag`]: writing this leaves the object's bytes, and so its
+/// [`ObjectETag`], untouched. That is what lets a caller record something about an object
+/// without invalidating anything keyed on its content.
+pub type ObjectTagSet = Vec<(String, String)>;
+
+/// Tag key CAS stamps with the unix seconds of a xorb's most recent write.
+///
+/// Every simulated xorb upload stamps this, as CAS does. What the value
+/// *means* is the reader's business — xet-core only reproduces the stamp.
+pub const LAST_UPLOAD_TAG_KEY: &str = "last-upload";
+
+/// The tag set CAS puts on a xorb it has just written.
+pub fn last_upload_tag_set_now() -> ObjectTagSet {
+    let unix = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or_default();
+    vec![(LAST_UPLOAD_TAG_KEY.to_string(), unix.to_string())]
+}
 
 /// Trait for clients that support deletion and integrity operations on shards and file entries.
 ///
@@ -41,19 +65,25 @@ pub trait DeletionControlableClient: Send + Sync {
     /// Deletes a XORB by hash.
     async fn delete_xorb(&self, hash: &MerkleHash);
 
-    /// Returns all XORB hashes with their associated object tags.
-    async fn list_xorbs_and_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>>;
+    /// Returns all XORB hashes with their associated object etags.
+    async fn list_xorbs_and_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>>;
+
+    /// Deletes a XORB only if its current etag matches the provided etag.
+    /// Returns `Ok(true)` if deleted, `Ok(false)` if the etag did not match.
+    async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool>;
+
+    /// Returns a XORB's tag set, empty if it has none.
+    async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result<ObjectTagSet>;
 
-    /// Deletes a XORB only if its current tag matches the provided tag.
-    /// Returns `Ok(true)` if deleted, `Ok(false)` if the tag did not match.
-    async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool>;
+    /// Replaces a XORB's tag set wholesale, as S3 `PutObjectTagging` does.
+    async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()>;
 
-    /// Returns all shard hashes with their associated object tags.
-    async fn list_shards_with_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>>;
+    /// Returns all shard hashes with their associated object etags.
+    async fn list_shards_with_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>>;
 
-    /// Deletes a shard only if its current tag matches the provided tag.
-    /// Returns `Ok(true)` if deleted, `Ok(false)` if the tag did not match.
-    async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool>;
+    /// Deletes a shard only if its current etag matches the provided etag.
+    /// Returns `Ok(true)` if deleted, `Ok(false)` if the etag did not match.
+    async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool>;
 
     /// Verifies referential integrity of all shards on disk.
     async fn verify_integrity(&self) -> Result<()>;
diff --git a/xet_client/src/cas_client/simulation/deletion_unit_testing.rs b/xet_client/src/cas_client/simulation/deletion_unit_testing.rs
index 798f57a93..15f1e26cb 100644
--- a/xet_client/src/cas_client/simulation/deletion_unit_testing.rs
+++ b/xet_client/src/cas_client/simulation/deletion_unit_testing.rs
@@ -13,7 +13,7 @@ use std::sync::Arc;
 use xet_core_structures::merklehash::MerkleHash;
 
 use super::client_testing_utils::RandomFileContents;
-use super::deletion_controls::ObjectTag;
+use super::deletion_controls::ObjectETag;
 use super::{ClientTestingUtils, DeletionControlableClient, DirectAccessClient};
 
 /// Runs all common DeletionControlableClient tests using a factory that creates fresh clients.
@@ -340,19 +340,19 @@ async fn test_verify_integrity_after_file_deletion<C: DirectAccessClient + Delet
     client.verify_integrity().await.unwrap();
 }
 
-/// Tests that list_xorbs_and_tags returns entries matching list_xorbs with non-zero tags.
+/// Tests that list_xorbs_and_etags returns entries matching list_xorbs with non-zero tags.
 async fn test_list_xorbs_and_tags<C: DirectAccessClient + DeletionControlableClient + 'static>(client: Arc<C>) {
-    assert!(client.list_xorbs_and_tags().await.unwrap().is_empty());
+    assert!(client.list_xorbs_and_etags().await.unwrap().is_empty());
 
     let file = client.upload_random_file(&[(1, (0, 3))], 2048).await.unwrap();
     let expected_xorbs = expected_xorb_hashes(&[&file]);
 
-    let xorbs_and_tags = client.list_xorbs_and_tags().await.unwrap();
-    let listed_hashes: HashSet<MerkleHash> = xorbs_and_tags.iter().map(|(h, _)| *h).collect();
+    let xorbs_and_etags = client.list_xorbs_and_etags().await.unwrap();
+    let listed_hashes: HashSet<MerkleHash> = xorbs_and_etags.iter().map(|(h, _)| *h).collect();
     assert_eq!(listed_hashes, expected_xorbs);
 
-    let zero_tag: ObjectTag = [0u8; 32];
-    for (_, tag) in &xorbs_and_tags {
+    let zero_tag: ObjectETag = [0u8; 32];
+    for (_, tag) in &xorbs_and_etags {
         assert_ne!(tag, &zero_tag, "Tag should be non-zero");
     }
 }
@@ -362,32 +362,32 @@ async fn test_delete_xorb_if_tag_matches<C: DirectAccessClient + DeletionControl
     let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
     let xorb_hash = file.terms[0].xorb_hash;
 
-    let xorbs_and_tags = client.list_xorbs_and_tags().await.unwrap();
-    let (_, correct_tag) = xorbs_and_tags.iter().find(|(h, _)| *h == xorb_hash).unwrap();
+    let xorbs_and_etags = client.list_xorbs_and_etags().await.unwrap();
+    let (_, correct_tag) = xorbs_and_etags.iter().find(|(h, _)| *h == xorb_hash).unwrap();
 
-    let wrong_tag: ObjectTag = [0xFFu8; 32];
-    let deleted = client.delete_xorb_if_tag_matches(&xorb_hash, &wrong_tag).await.unwrap();
+    let wrong_tag: ObjectETag = [0xFFu8; 32];
+    let deleted = client.delete_xorb_if_etag_matches(&xorb_hash, &wrong_tag).await.unwrap();
     assert!(!deleted, "Wrong tag should not delete the xorb");
     assert!(client.xorb_exists(&xorb_hash).await.unwrap(), "Xorb should still exist after wrong tag");
 
-    let deleted = client.delete_xorb_if_tag_matches(&xorb_hash, correct_tag).await.unwrap();
+    let deleted = client.delete_xorb_if_etag_matches(&xorb_hash, correct_tag).await.unwrap();
     assert!(deleted, "Correct tag should delete the xorb");
     assert!(!client.xorb_exists(&xorb_hash).await.unwrap(), "Xorb should be gone after correct tag");
 }
 
-/// Tests that list_shards_with_tags returns entries matching list_shard_entries with non-zero tags.
+/// Tests that list_shards_with_etags returns entries matching list_shard_entries with non-zero tags.
 async fn test_list_shards_with_tags<C: DirectAccessClient + DeletionControlableClient + 'static>(client: Arc<C>) {
-    assert!(client.list_shards_with_tags().await.unwrap().is_empty());
+    assert!(client.list_shards_with_etags().await.unwrap().is_empty());
 
     client.upload_random_file(&[(1, (0, 3))], 2048).await.unwrap();
 
-    let shards_and_tags = client.list_shards_with_tags().await.unwrap();
+    let shards_and_tags = client.list_shards_with_etags().await.unwrap();
     let shard_entries = client.list_shard_entries().await.unwrap();
     let listed_hashes: HashSet<MerkleHash> = shards_and_tags.iter().map(|(h, _)| *h).collect();
     let expected_hashes: HashSet<MerkleHash> = shard_entries.into_iter().collect();
     assert_eq!(listed_hashes, expected_hashes);
 
-    let zero_tag: ObjectTag = [0u8; 32];
+    let zero_tag: ObjectETag = [0u8; 32];
     for (_, tag) in &shards_and_tags {
         assert_ne!(tag, &zero_tag, "Tag should be non-zero");
     }
@@ -397,16 +397,16 @@ async fn test_list_shards_with_tags<C: DirectAccessClient + DeletionControlableC
 async fn test_delete_shard_if_tag_matches<C: DirectAccessClient + DeletionControlableClient + 'static>(client: Arc<C>) {
     let _file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
 
-    let shards_and_tags = client.list_shards_with_tags().await.unwrap();
+    let shards_and_tags = client.list_shards_with_etags().await.unwrap();
     assert!(!shards_and_tags.is_empty());
     let (shard_hash, correct_tag) = &shards_and_tags[0];
 
-    let wrong_tag: ObjectTag = [0xFFu8; 32];
-    let deleted = client.delete_shard_if_tag_matches(shard_hash, &wrong_tag).await.unwrap();
-    assert!(!deleted, "Wrong tag should not delete the shard");
-    assert!(!client.list_shard_entries().await.unwrap().is_empty(), "Shard should still exist after wrong tag");
+    let wrong_tag: ObjectETag = [0xFFu8; 32];
+    let deleted = client.delete_shard_if_etag_matches(shard_hash, &wrong_tag).await.unwrap();
+    assert!(!deleted, "Wrong etag should not delete the shard");
+    assert!(!client.list_shard_entries().await.unwrap().is_empty(), "Shard should still exist after wrong etag");
 
-    let deleted = client.delete_shard_if_tag_matches(shard_hash, correct_tag).await.unwrap();
-    assert!(deleted, "Correct tag should delete the shard");
-    assert!(client.list_shard_entries().await.unwrap().is_empty(), "Shard should be gone after correct tag");
+    let deleted = client.delete_shard_if_etag_matches(shard_hash, correct_tag).await.unwrap();
+    assert!(deleted, "Correct etag should delete the shard");
+    assert!(client.list_shard_entries().await.unwrap().is_empty(), "Shard should be gone after correct etag");
 }
diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs
index 7b78e8236..67cf34f99 100644
--- a/xet_client/src/cas_client/simulation/local_client.rs
+++ b/xet_client/src/cas_client/simulation/local_client.rs
@@ -31,7 +31,7 @@ use xet_runtime::core::XetContext;
 use xet_runtime::fd_diagnostics::{report_fd_count, track_fd_scope};
 use xet_runtime::file_utils::SafeFileCreator;
 
-use super::deletion_controls::ObjectTag;
+use super::deletion_controls::{ObjectETag, ObjectTagSet, last_upload_tag_set_now};
 use super::direct_access_client::DirectAccessClient;
 use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil};
 use crate::cas_client::Client;
@@ -375,6 +375,31 @@ impl LocalClient {
         PathBuf::from(name)
     }
 
+    /// Path of a xorb's S3-style tag set: `<canonical>.tagset`, holding JSON.
+    /// A sidecar so writing tags cannot disturb the bytes the [`ObjectETag`] is
+    /// derived from.
+    fn tag_set_xorb_path(&self, hash: &MerkleHash) -> PathBuf {
+        let canonical = self.get_path_for_entry(hash);
+        let mut name = canonical.into_os_string();
+        name.push(".tagset");
+        PathBuf::from(name)
+    }
+
+    /// Drops a xorb's tag-set sidecar. Called from every path that removes the
+    /// object or condemns it: a hard delete takes the tags with it, and GC's
+    /// `gc-delete` write replaces the whole set, dropping `last-upload` either
+    /// way. Leaving the sidecar behind would orphan it on disk, and let a later
+    /// upload of the same hash inherit tags from the object that used to live
+    /// there.
+    fn clear_tag_set_xorb(&self, hash: &MerkleHash) {
+        let path = self.tag_set_xorb_path(hash);
+        #[cfg(windows)]
+        if path.exists() {
+            Self::clear_readonly(&path);
+        }
+        let _ = std::fs::remove_file(path);
+    }
+
     /// Path used to park a tagged-for-deletion shard: `<hex>.mdb.gctag`.
     fn gctag_shard_path(&self, hash: &MerkleHash) -> PathBuf {
         let canonical = self.shard_dir.join(shard_file_name(hash));
@@ -422,26 +447,26 @@ impl LocalClient {
         }
     }
 
-    /// Builds an `ObjectTag` from file metadata at the given path.
+    /// Builds an `ObjectETag` from file metadata at the given path.
     ///
     /// We hash multiple metadata fields to increase entropy and reduce false
     /// matches during rapid rewrite/delete races.
-    fn object_tag_from_path(path: &Path) -> Result<ObjectTag> {
-        let meta = std::fs::metadata(path).map_err(ClientError::internal)?;
-        let modified = meta.modified().map_err(ClientError::internal)?;
-        let modified_nanos = modified.duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos();
-        let created_nanos = meta
-            .created()
-            .ok()
-            .and_then(|ts| ts.duration_since(std::time::UNIX_EPOCH).ok())
-            .map_or(0u128, |d| d.as_nanos());
-
-        let mut entropy = Vec::with_capacity(16 + 16 + 8 + 1);
-        entropy.extend_from_slice(&modified_nanos.to_le_bytes());
-        entropy.extend_from_slice(&created_nanos.to_le_bytes());
-        entropy.extend_from_slice(&meta.len().to_le_bytes());
-        entropy.push(u8::from(meta.permissions().readonly()));
-
+    /// The object's etag, derived from its key and stored length rather than
+    /// from filesystem timestamps.
+    ///
+    /// S3's ETag is a function of content, so a byte-identical re-upload leaves
+    /// it unchanged. Xorbs and shards are content-addressed, so the key already
+    /// fixes the content and the length is what distinguishes a differently
+    /// serialized rewrite of the same key. Hashing mtime/ctime instead made
+    /// every write look like different content — including a re-upload, and
+    /// including the rename onto a temp path that the conditional deletes do.
+    fn object_etag_for(prefix: &[u8], key: &MerkleHash, path: &Path) -> Result<ObjectETag> {
+        let len = std::fs::metadata(path).map_err(ClientError::internal)?.len();
+        let key_bytes: [u8; 32] = (*key).into();
+        let mut entropy = Vec::with_capacity(prefix.len() + key_bytes.len() + 8);
+        entropy.extend_from_slice(prefix);
+        entropy.extend_from_slice(&key_bytes);
+        entropy.extend_from_slice(&len.to_le_bytes());
         Ok(compute_data_hash(&entropy).into())
     }
 
@@ -994,9 +1019,10 @@ impl super::DeletionControlableClient for LocalClient {
         } else {
             let _ = std::fs::remove_file(file_path);
         }
+        self.clear_tag_set_xorb(hash);
     }
 
-    async fn list_xorbs_and_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>> {
+    async fn list_xorbs_and_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>> {
         let mut ret = Vec::new();
         for entry in self.xorb_dir.read_dir().map_err(ClientError::internal)? {
             let entry = entry.map_err(ClientError::internal)?;
@@ -1007,15 +1033,15 @@ impl super::DeletionControlableClient for LocalClient {
             if let Some(pos) = name.rfind('.') {
                 let hex = &name[(pos + 1)..];
                 if let Ok(hash) = MerkleHash::from_hex(hex) {
-                    let tag = Self::object_tag_from_path(&path)?;
-                    ret.push((hash, tag));
+                    let etag = Self::object_etag_for(b"xorb", &hash, &path)?;
+                    ret.push((hash, etag));
                 }
             }
         }
         Ok(ret)
     }
 
-    async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool> {
+    async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool> {
         let file_path = self.get_path_for_entry(hash);
 
         // Atomically move the file out of the namespace before checking the
@@ -1026,7 +1052,7 @@ impl super::DeletionControlableClient for LocalClient {
             return Err(ClientError::XORBNotFound(*hash));
         }
 
-        let current_tag = match Self::object_tag_from_path(&tmp_path) {
+        let current_etag = match Self::object_etag_for(b"xorb", hash, &tmp_path) {
             Ok(t) => t,
             Err(e) => {
                 Self::restore_from_tmp(&tmp_path, &file_path);
@@ -1034,7 +1060,7 @@ impl super::DeletionControlableClient for LocalClient {
             },
         };
 
-        if &current_tag != tag {
+        if &current_etag != etag {
             Self::restore_from_tmp(&tmp_path, &file_path);
             return Ok(false);
         }
@@ -1050,19 +1076,51 @@ impl super::DeletionControlableClient for LocalClient {
         } else {
             std::fs::remove_file(&tmp_path)?;
         }
+        self.clear_tag_set_xorb(hash);
         Ok(true)
     }
 
-    async fn list_shards_with_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>> {
+    async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result<ObjectTagSet> {
+        if !self.get_path_for_entry(hash).exists() {
+            return Err(ClientError::Other(format!("XORB not found: {}", hash.hex())));
+        }
+        let path = self.tag_set_xorb_path(hash);
+        if !path.exists() {
+            return Ok(ObjectTagSet::new());
+        }
+        let raw = std::fs::read(&path)?;
+        serde_json::from_slice(&raw)
+            .map_err(|e| ClientError::Other(format!("invalid tag set at {}: {e}", path.display())))
+    }
+
+    async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()> {
+        if !self.get_path_for_entry(hash).exists() {
+            return Err(ClientError::Other(format!("XORB not found: {}", hash.hex())));
+        }
+        let path = self.tag_set_xorb_path(hash);
+        let raw = serde_json::to_vec(&tags).map_err(|e| ClientError::Other(format!("serialize tag set: {e}")))?;
+        #[cfg(windows)]
+        if path.exists() {
+            Self::clear_readonly(&path);
+        }
+        // Temp file plus rename: `get_xorb_tag_set` reads this path without coordination,
+        // so a truncate-in-place write would let it observe a half-written tag set.
+        let mut file = SafeFileCreator::replace_existing(&path)?;
+        file.write_all(&raw)?;
+        file.close()?;
+        Ok(())
+    }
+
+    async fn list_shards_with_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>> {
         let mut ret = Vec::new();
         for (hash, path) in self.shard_file_paths()? {
-            let tag = Self::object_tag_from_path(&path)?;
-            ret.push((hash, tag));
+            let etag = Self::object_etag_for(b"shard", &hash, &path)?;
+            ret.push((hash, etag));
         }
         Ok(ret)
     }
 
-    async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool> {
+    async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool> {
         let path = self.shard_path_for_hash(hash)?;
 
         let tmp_path = path.with_extension(format!("gc_del_{:x}", rand::random::<u64>()));
@@ -1070,7 +1128,7 @@ impl super::DeletionControlableClient for LocalClient {
             return Err(ClientError::Other(format!("Shard not found: {}", hash.hex())));
         }
 
-        let current_tag = match Self::object_tag_from_path(&tmp_path) {
+        let current_etag = match Self::object_etag_for(b"shard", hash, &tmp_path) {
             Ok(t) => t,
             Err(e) => {
                 Self::restore_from_tmp(&tmp_path, &path);
@@ -1078,7 +1136,7 @@ impl super::DeletionControlableClient for LocalClient {
             },
         };
 
-        if &current_tag != tag {
+        if &current_etag != etag {
             Self::restore_from_tmp(&tmp_path, &path);
             return Ok(false);
         }
@@ -1606,7 +1664,7 @@ impl Client for LocalClient {
 
         // Always rewrite: even if the xorb already exists, the file must be
         // re-created so its filesystem metadata (mtime/ctime) changes, producing
-        // a new tag for delete_xorb_if_tag_matches.  SafeFileCreator uses
+        // a new etag for delete_xorb_if_etag_matches.  SafeFileCreator uses
         // temp-file + atomic rename, so concurrent readers are safe.
 
         // Reconstruct footer if not present
@@ -1663,6 +1721,23 @@ impl Client for LocalClient {
         // readable again. Mirrors S3 PutObject overwriting a tagged object.
         let _ = std::fs::remove_file(self.gctag_xorb_path(&hash));
 
+        // CAS stamps `last-upload` on every xorb write, and PutObject replaces
+        // the whole tag set, so the stamp both records this write and clears
+        // whatever was there.
+        let tag_set_path = self.tag_set_xorb_path(&hash);
+        match serde_json::to_vec(&last_upload_tag_set_now()) {
+            Ok(raw) => {
+                #[cfg(windows)]
+                if tag_set_path.exists() {
+                    Self::clear_readonly(&tag_set_path);
+                }
+                if let Err(e) = std::fs::write(&tag_set_path, raw) {
+                    warn!("failed to stamp last-upload tag at {}: {e}", tag_set_path.display());
+                }
+            },
+            Err(e) => warn!("failed to serialize last-upload tag set: {e}"),
+        }
+
         info!("{file_path:?} successfully written with {bytes_written} bytes.");
 
         Ok(bytes_written as u64)
@@ -2278,15 +2353,114 @@ mod tests {
             .expect("Integrity should pass: the old shard's stale file entry with dangling xorb refs is not consulted");
     }
 
-    /// Tests that list_xorbs_and_tags tags change after file re-creation with a timestamp delay.
+    /// A tag set is stored, replaced wholesale on the next write (as S3
+    /// `PutObjectTagging` does), and never disturbs the xorb's `ObjectETag` or
+    /// its appearance in the xorb listing — the sidecar must not read as a xorb.
     #[tokio::test]
-    async fn test_list_xorbs_and_tags_timestamp_changes() {
+    async fn test_xorb_tag_set_round_trip_leaves_object_tag_intact() {
+        let client = LocalClient::temporary(test_context()).await.unwrap();
+        let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        let xorb_hash = file.terms[0].xorb_hash;
+
+        let before = client.list_xorbs_and_etags().await.unwrap();
+        let (_, tag_before) = before.iter().find(|(h, _)| *h == xorb_hash).unwrap();
+
+        client
+            .set_xorb_tag_set(&xorb_hash, vec![("last-upload".to_string(), "1234".to_string())])
+            .await
+            .unwrap();
+        assert_eq!(
+            client.get_xorb_tag_set(&xorb_hash).await.unwrap(),
+            vec![("last-upload".to_string(), "1234".to_string())]
+        );
+
+        // A second write replaces the set rather than merging into it.
+        client
+            .set_xorb_tag_set(&xorb_hash, vec![("other".to_string(), "x".to_string())])
+            .await
+            .unwrap();
+        assert_eq!(
+            client.get_xorb_tag_set(&xorb_hash).await.unwrap(),
+            vec![("other".to_string(), "x".to_string())],
+            "PutObjectTagging semantics replace the whole set"
+        );
+
+        // Both listing paths must ignore the sidecar, not just the tagged one.
+        let listed = client.list_xorbs().await.unwrap();
+        assert_eq!(listed, vec![xorb_hash], "the .tagset sidecar must not be listed as a xorb");
+
+        let after = client.list_xorbs_and_etags().await.unwrap();
+        assert_eq!(after.len(), before.len(), "the sidecar must not appear as an extra tagged xorb");
+        let (_, tag_after) = after.iter().find(|(h, _)| *h == xorb_hash).unwrap();
+        assert_eq!(tag_before, tag_after, "tagging must not move the ObjectETag");
+    }
+
+    /// With upload tagging on, an uploaded xorb carries `last-upload` and the
+    /// sidecar still does not show up as a xorb in either listing.
+    #[tokio::test]
+    async fn test_upload_tagging_stamps_last_upload() {
+        let client = LocalClient::temporary(test_context()).await.unwrap();
+
+        let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        let xorb_hash = file.terms[0].xorb_hash;
+
+        let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap();
+        let (key, value) = tags.first().expect("upload must stamp a tag set");
+        assert_eq!(key, super::super::deletion_controls::LAST_UPLOAD_TAG_KEY);
+        assert!(value.parse::<u64>().is_ok(), "value must be unix seconds, got {value:?}");
+
+        assert_eq!(client.list_xorbs().await.unwrap(), vec![xorb_hash]);
+    }
+
+    /// A hard delete removes the sidecar, so it is not orphaned on disk and a
+    /// later upload of the same hash cannot inherit the old tags.
+    #[tokio::test]
+    async fn test_delete_xorb_clears_its_tag_set_sidecar() {
+        let client = LocalClient::temporary(test_context()).await.unwrap();
+        let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        let xorb_hash = file.terms[0].xorb_hash;
+
+        client
+            .set_xorb_tag_set(&xorb_hash, vec![("stale".to_string(), "value".to_string())])
+            .await
+            .unwrap();
+        let sidecar = client.tag_set_xorb_path(&xorb_hash);
+        assert!(sidecar.exists());
+
+        client.delete_xorb(&xorb_hash).await;
+        assert!(!sidecar.exists(), "the sidecar outlived the xorb");
+
+        let refile = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        assert_eq!(refile.terms[0].xorb_hash, xorb_hash);
+        let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap();
+        assert!(!tags.iter().any(|(k, _)| k == "stale"), "stale tags survived a delete: {tags:?}");
+    }
+
+    /// Tagging an absent xorb is an error rather than creating an orphan sidecar.
+    #[tokio::test]
+    async fn test_xorb_tag_set_requires_the_xorb_to_exist() {
+        let client = LocalClient::temporary(test_context()).await.unwrap();
+        let missing = MerkleHash::default();
+        assert!(client.get_xorb_tag_set(&missing).await.is_err());
+        assert!(client.set_xorb_tag_set(&missing, vec![]).await.is_err());
+    }
+
+    /// Tests that list_xorbs_and_etags etags change after file re-creation with a timestamp delay.
+    /// A xorb deleted and re-uploaded with the same content keeps its etag,
+    /// even across a filesystem timestamp change.
+    ///
+    /// This is S3's contract — the ETag is a function of content — and the
+    /// resurrection hazard it creates is real: a stale etag snapshot still
+    /// matches the new object. That is what the `last-upload` tag is for; the
+    /// etag deliberately no longer hides it.
+    #[tokio::test]
+    async fn test_etag_is_stable_across_an_identical_reupload() {
         let client = LocalClient::temporary(test_context()).await.unwrap();
 
         let file1 = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
         let xorb_hash = file1.terms[0].xorb_hash;
 
-        let tags1 = client.list_xorbs_and_tags().await.unwrap();
+        let tags1 = client.list_xorbs_and_etags().await.unwrap();
         let (_, tag1) = tags1.iter().find(|(h, _)| *h == xorb_hash).unwrap();
 
         // Delete and wait 1 second so the filesystem timestamp advances.
@@ -2296,11 +2470,16 @@ mod tests {
         // Re-upload a file that creates a new xorb with the same hash seed.
         let file2 = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
         let xorb_hash2 = file2.terms[0].xorb_hash;
+        assert_eq!(xorb_hash, xorb_hash2, "the seed must reproduce the same xorb");
 
-        let tags2 = client.list_xorbs_and_tags().await.unwrap();
+        let tags2 = client.list_xorbs_and_etags().await.unwrap();
         let (_, tag2) = tags2.iter().find(|(h, _)| *h == xorb_hash2).unwrap();
 
-        assert_ne!(tag1, tag2, "Tags should differ after re-creation with timestamp delay");
+        assert_eq!(tag1, tag2, "identical content must keep its etag despite a newer mtime");
+
+        // The `last-upload` tag is what moved instead.
+        let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap();
+        assert_eq!(tags.first().map(|(k, _)| k.as_str()), Some(super::super::deletion_controls::LAST_UPLOAD_TAG_KEY));
     }
 
     // ── Lifecycle-tag deletion mode tests ───────────────────────────────
diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs b/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs
index 750cab89a..02a6fceb4 100644
--- a/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs
+++ b/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs
@@ -10,13 +10,13 @@ use xet_core_structures::xorb_object::XorbObject;
 use xet_runtime::core::XetContext;
 
 use super::simulation_types::{
-    FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry, FileSizeResponse, HashWithTag, TagDeleteRequest,
-    TagDeleteResponse, XorbExistsResponse, XorbLengthResponse, XorbRangesRequest, XorbRangesResponse,
-    XorbRawLengthResponse,
+    ETagDeleteRequest, ETagDeleteResponse, FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry,
+    FileSizeResponse, HashWithETag, TagSetBody, XorbExistsResponse, XorbLengthResponse, XorbRangesRequest,
+    XorbRangesResponse, XorbRawLengthResponse,
 };
 use crate::cas_client::RemoteClient;
 use crate::cas_client::interface::Client;
-use crate::cas_client::simulation::deletion_controls::ObjectTag;
+use crate::cas_client::simulation::deletion_controls::{ObjectETag, ObjectTagSet};
 use crate::cas_client::simulation::xorb_utils::duration_to_expiration_secs_ceil;
 use crate::cas_client::simulation::{DeletionControlableClient, DirectAccessClient};
 use crate::cas_types::{
@@ -554,22 +554,22 @@ impl DeletionControlableClient for SimulationControlClient {
         let _ = self.http_client.delete(&url).send().await;
     }
 
-    async fn list_xorbs_and_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>> {
+    async fn list_xorbs_and_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>> {
         let resp = self
             .http_client
-            .get(self.sim_url("/xorbs_with_tags"))
+            .get(self.sim_url("/xorbs_with_etags"))
             .send()
             .await
             .map_err(|e| ClientError::Other(e.to_string()))?;
         let resp = Self::check_status(resp).await?;
-        let entries: Vec<HashWithTag> = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
-        Ok(entries.into_iter().map(|e| (e.hash, e.tag)).collect())
+        let entries: Vec<HashWithETag> = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
+        Ok(entries.into_iter().map(|e| (e.hash, e.etag)).collect())
     }
 
-    async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool> {
+    async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool> {
         let hex = HexMerkleHash::from(*hash);
-        let url = self.sim_url(&format!("/xorbs/{hex}/tag_delete"));
-        let body = TagDeleteRequest { tag: *tag };
+        let url = self.sim_url(&format!("/xorbs/{hex}/etag_delete"));
+        let body = ETagDeleteRequest { etag: *etag };
         let resp = self
             .http_client
             .post(&url)
@@ -578,26 +578,52 @@ impl DeletionControlableClient for SimulationControlClient {
             .await
             .map_err(|e| ClientError::Other(e.to_string()))?;
         let resp = Self::check_status(resp).await?;
-        let result: TagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
+        let result: ETagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
         Ok(result.deleted)
     }
 
-    async fn list_shards_with_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>> {
+    async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result<ObjectTagSet> {
+        let hex = HexMerkleHash::from(*hash);
+        let resp = self
+            .http_client
+            .get(self.sim_url(&format!("/xorbs/{hex}/tag_set")))
+            .send()
+            .await
+            .map_err(|e| ClientError::Other(e.to_string()))?;
+        let resp = Self::check_status(resp).await?;
+        let body: TagSetBody = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
+        Ok(body.tags)
+    }
+
+    async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()> {
+        let hex = HexMerkleHash::from(*hash);
+        let resp = self
+            .http_client
+            .put(self.sim_url(&format!("/xorbs/{hex}/tag_set")))
+            .json(&TagSetBody { tags })
+            .send()
+            .await
+            .map_err(|e| ClientError::Other(e.to_string()))?;
+        Self::check_status(resp).await?;
+        Ok(())
+    }
+
+    async fn list_shards_with_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>> {
         let resp = self
             .http_client
-            .get(self.sim_url("/shards_with_tags"))
+            .get(self.sim_url("/shards_with_etags"))
             .send()
             .await
             .map_err(|e| ClientError::Other(e.to_string()))?;
         let resp = Self::check_status(resp).await?;
-        let entries: Vec<HashWithTag> = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
-        Ok(entries.into_iter().map(|e| (e.hash, e.tag)).collect())
+        let entries: Vec<HashWithETag> = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
+        Ok(entries.into_iter().map(|e| (e.hash, e.etag)).collect())
     }
 
-    async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool> {
+    async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool> {
         let hex = HexMerkleHash::from(*hash);
-        let url = self.sim_url(&format!("/shards/{hex}/tag_delete"));
-        let body = TagDeleteRequest { tag: *tag };
+        let url = self.sim_url(&format!("/shards/{hex}/etag_delete"));
+        let body = ETagDeleteRequest { etag: *etag };
         let resp = self
             .http_client
             .post(&url)
@@ -606,7 +632,7 @@ impl DeletionControlableClient for SimulationControlClient {
             .await
             .map_err(|e| ClientError::Other(e.to_string()))?;
         let resp = Self::check_status(resp).await?;
-        let result: TagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
+        let result: ETagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?;
         Ok(result.deleted)
     }
 
diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs b/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs
index 6179805b8..abc3c4916 100644
--- a/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs
+++ b/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs
@@ -9,9 +9,9 @@ use http::header::RANGE;
 
 use super::handlers::{FileRangeVariant, ServerState, error_to_response, parse_range_header};
 use super::simulation_types::{
-    ConfigDelayRangeRequest, ConfigDurationRequest, FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry,
-    FileSizeResponse, HashWithTag, TagDeleteRequest, TagDeleteResponse, XorbExistsResponse, XorbLengthResponse,
-    XorbRangesRequest, XorbRangesResponse, XorbRawLengthResponse,
+    ConfigDelayRangeRequest, ConfigDurationRequest, ETagDeleteRequest, ETagDeleteResponse, FetchTermDataRequest,
+    FetchTermDataResponse, FileShardsEntry, FileSizeResponse, HashWithETag, TagSetBody, XorbExistsResponse,
+    XorbLengthResponse, XorbRangesRequest, XorbRangesResponse, XorbRawLengthResponse,
 };
 use crate::cas_types::{FileRange, HexMerkleHash};
 
@@ -36,12 +36,13 @@ pub fn simulation_routes() -> Router<ServerState> {
         .route("/config/api_delay", post(set_api_delay))
         .route("/fetch_term_data", post(fetch_term_data))
         // DeletionControlableClient routes
-        .route("/xorbs_with_tags", get(list_xorbs_and_tags))
-        .route("/xorbs/{hash}/tag_delete", post(delete_xorb_if_tag_matches))
+        .route("/xorbs_with_etags", get(list_xorbs_and_etags))
+        .route("/xorbs/{hash}/etag_delete", post(delete_xorb_if_etag_matches))
+        .route("/xorbs/{hash}/tag_set", get(get_xorb_tag_set).put(set_xorb_tag_set))
         .route("/shards", get(list_shard_entries))
-        .route("/shards_with_tags", get(list_shards_with_tags))
+        .route("/shards_with_etags", get(list_shards_with_etags))
         .route("/shards/{hash}", get(get_shard_bytes).delete(delete_shard_entry))
-        .route("/shards/{hash}/tag_delete", post(delete_shard_if_tag_matches))
+        .route("/shards/{hash}/etag_delete", post(delete_shard_if_etag_matches))
         .route("/shards/{hash}/dedup_entries", delete(remove_shard_dedup_entries))
         .route("/file_entries", get(list_file_shard_entries))
         .route("/file_entries/{hash}", delete(delete_file_entry))
@@ -203,56 +204,85 @@ async fn delete_xorb(State(state): State<ServerState>, Path(HexMerkleHash(hash))
     StatusCode::NO_CONTENT.into_response()
 }
 
-async fn list_xorbs_and_tags(State(state): State<ServerState>) -> Response {
+async fn list_xorbs_and_etags(State(state): State<ServerState>) -> Response {
     let Some(dc) = &state.deletion_client else {
         return not_implemented();
     };
-    match dc.list_xorbs_and_tags().await {
+    match dc.list_xorbs_and_etags().await {
         Ok(entries) => {
-            let response: Vec<HashWithTag> = entries.into_iter().map(|(hash, tag)| HashWithTag { hash, tag }).collect();
+            let response: Vec<HashWithETag> =
+                entries.into_iter().map(|(hash, etag)| HashWithETag { hash, etag }).collect();
             Json(response).into_response()
         },
         Err(e) => error_to_response(e),
     }
 }
 
-async fn delete_xorb_if_tag_matches(
+async fn delete_xorb_if_etag_matches(
     State(state): State<ServerState>,
     Path(HexMerkleHash(hash)): Path<HexMerkleHash>,
-    Json(body): Json<TagDeleteRequest>,
+    Json(body): Json<ETagDeleteRequest>,
 ) -> Response {
     let Some(dc) = &state.deletion_client else {
         return not_implemented();
     };
-    match dc.delete_xorb_if_tag_matches(&hash, &body.tag).await {
-        Ok(deleted) => Json(TagDeleteResponse { deleted }).into_response(),
+    match dc.delete_xorb_if_etag_matches(&hash, &body.etag).await {
+        Ok(deleted) => Json(ETagDeleteResponse { deleted }).into_response(),
         Err(e) => error_to_response(e),
     }
 }
 
-async fn list_shards_with_tags(State(state): State<ServerState>) -> Response {
+async fn get_xorb_tag_set(
+    State(state): State<ServerState>,
+    Path(HexMerkleHash(hash)): Path<HexMerkleHash>,
+) -> Response {
+    let Some(dc) = &state.deletion_client else {
+        return not_implemented();
+    };
+    match dc.get_xorb_tag_set(&hash).await {
+        Ok(tags) => Json(TagSetBody { tags }).into_response(),
+        Err(e) => error_to_response(e),
+    }
+}
+
+async fn set_xorb_tag_set(
+    State(state): State<ServerState>,
+    Path(HexMerkleHash(hash)): Path<HexMerkleHash>,
+    Json(body): Json<TagSetBody>,
+) -> Response {
+    let Some(dc) = &state.deletion_client else {
+        return not_implemented();
+    };
+    match dc.set_xorb_tag_set(&hash, body.tags).await {
+        Ok(()) => StatusCode::NO_CONTENT.into_response(),
+        Err(e) => error_to_response(e),
+    }
+}
+
+async fn list_shards_with_etags(State(state): State<ServerState>) -> Response {
     let Some(dc) = &state.deletion_client else {
         return not_implemented();
     };
-    match dc.list_shards_with_tags().await {
+    match dc.list_shards_with_etags().await {
         Ok(entries) => {
-            let response: Vec<HashWithTag> = entries.into_iter().map(|(hash, tag)| HashWithTag { hash, tag }).collect();
+            let response: Vec<HashWithETag> =
+                entries.into_iter().map(|(hash, etag)| HashWithETag { hash, etag }).collect();
             Json(response).into_response()
         },
         Err(e) => error_to_response(e),
     }
 }
 
-async fn delete_shard_if_tag_matches(
+async fn delete_shard_if_etag_matches(
     State(state): State<ServerState>,
     Path(HexMerkleHash(hash)): Path<HexMerkleHash>,
-    Json(body): Json<TagDeleteRequest>,
+    Json(body): Json<ETagDeleteRequest>,
 ) -> Response {
     let Some(dc) = &state.deletion_client else {
         return not_implemented();
     };
-    match dc.delete_shard_if_tag_matches(&hash, &body.tag).await {
-        Ok(deleted) => Json(TagDeleteResponse { deleted }).into_response(),
+    match dc.delete_shard_if_etag_matches(&hash, &body.etag).await {
+        Ok(deleted) => Json(ETagDeleteResponse { deleted }).into_response(),
         Err(e) => error_to_response(e),
     }
 }
diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_types.rs b/xet_client/src/cas_client/simulation/local_server/simulation_types.rs
index 273bdff6e..4a43bacf9 100644
--- a/xet_client/src/cas_client/simulation/local_server/simulation_types.rs
+++ b/xet_client/src/cas_client/simulation/local_server/simulation_types.rs
@@ -1,7 +1,7 @@
 use serde::{Deserialize, Serialize};
 use xet_core_structures::merklehash::MerkleHash;
 
-use crate::cas_client::simulation::deletion_controls::ObjectTag;
+use crate::cas_client::simulation::deletion_controls::{ObjectETag, ObjectTagSet};
 use crate::cas_types::XorbReconstructionFetchInfo;
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -64,17 +64,24 @@ pub struct FetchTermDataResponse {
 }
 
 #[derive(Debug, Serialize, Deserialize)]
-pub struct HashWithTag {
+pub struct HashWithETag {
     pub hash: MerkleHash,
-    pub tag: ObjectTag,
+    pub etag: ObjectETag,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
-pub struct TagDeleteRequest {
-    pub tag: ObjectTag,
+pub struct ETagDeleteRequest {
+    pub etag: ObjectETag,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
-pub struct TagDeleteResponse {
+pub struct ETagDeleteResponse {
     pub deleted: bool,
 }
+
+/// Body of a tag-set read or write. Ordered pairs rather than a map so the
+/// wire form matches S3's tag set, where key order is preserved.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct TagSetBody {
+    pub tags: ObjectTagSet,
+}
diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs
index 39d264fb4..0c07a0985 100644
--- a/xet_client/src/cas_client/simulation/memory_client.rs
+++ b/xet_client/src/cas_client/simulation/memory_client.rs
@@ -27,7 +27,7 @@ use super::super::interface::{ShardUploadProgressCallback, ShardUploadProgressTy
 use super::super::progress_tracked_streams::ProgressCallback;
 use super::client_testing_utils::{FileTermReference, RandomFileContents};
 #[cfg(not(target_family = "wasm"))]
-use super::deletion_controls::ObjectTag;
+use super::deletion_controls::{ObjectETag, ObjectTagSet, last_upload_tag_set_now};
 use super::direct_access_client::DirectAccessClient;
 use super::random_xorb::RandomXorb;
 use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil};
@@ -47,11 +47,11 @@ struct MaterializedXorb {
 
 /// Storage for a XORB - either fully materialized or generated on-the-fly.
 /// Each variant carries a monotonic `generation` counter that is bumped on
-/// every insert, so the tag changes even when content is identical (matching
+/// every insert, so the etag changes even when content is identical (matching
 /// production ETag semantics).
 enum XorbStorage {
-    Materialized { entry: MaterializedXorb, generation: u64 },
-    Random { xorb: RandomXorb, generation: u64 },
+    Materialized { entry: MaterializedXorb },
+    Random { xorb: RandomXorb },
 }
 
 /// In-memory client for testing purposes. Stores all data in memory using hash tables.
@@ -64,8 +64,7 @@ pub struct MemoryClient {
     global_dedup: RwLock<MerkleHashMap<Bytes>>,
     /// Upload concurrency controller
     upload_concurrency_controller: Arc<AdaptiveConcurrencyController>,
-    /// Monotonic counter for xorb upload generations (tag freshness).
-    xorb_generation: AtomicU64,
+    /// Monotonic counter for xorb upload generations (etag freshness).
     /// URL expiration in milliseconds
     url_expiration_ms: AtomicU64,
     /// Global dedup shard expiration in seconds (0 = disabled).
@@ -89,6 +88,10 @@ pub struct MemoryClient {
     /// Shard hash currently tagged for lifecycle deletion. Shard data is
     /// retained in `shard` so a re-upload can clear the tag.
     gc_tagged_shard: RwLock<Option<MerkleHash>>,
+    /// S3-style tag sets per XORB. Held separately from `xorbs` so writing one
+    /// cannot perturb the bytes the [`ObjectETag`] is derived from.
+    #[cfg(not(target_family = "wasm"))]
+    xorb_tag_sets: RwLock<MerkleHashMap<ObjectTagSet>>,
 }
 
 impl MemoryClient {
@@ -99,7 +102,6 @@ impl MemoryClient {
             shard: RwLock::new(MDBInMemoryShard::default()),
             global_dedup: RwLock::new(MerkleHashMap::new()),
             upload_concurrency_controller: AdaptiveConcurrencyController::new_upload(ctx, "memory_uploads"),
-            xorb_generation: AtomicU64::new(0),
             url_expiration_ms: AtomicU64::new(u64::MAX),
             global_dedup_expiration_secs: AtomicU64::new(0),
             random_ms_delay_window: (AtomicU64::new(0), AtomicU64::new(0)),
@@ -108,6 +110,8 @@ impl MemoryClient {
             lifecycle_tag_deletion: AtomicBool::new(false),
             gc_tagged_xorbs: RwLock::new(HashSet::new()),
             gc_tagged_shard: RwLock::new(None),
+            #[cfg(not(target_family = "wasm"))]
+            xorb_tag_sets: RwLock::new(MerkleHashMap::new()),
         })
     }
 
@@ -124,6 +128,17 @@ impl MemoryClient {
         self.gc_tagged_xorbs.read().await.contains(hash)
     }
 
+    /// Errors unless the xorb is present and not condemned. A lifecycle-tagged
+    /// xorb reads as gone everywhere else here, and `LocalClient` errors on it
+    /// because the canonical file has been renamed away.
+    #[cfg(not(target_family = "wasm"))]
+    async fn require_readable_xorb(&self, hash: &MerkleHash) -> Result<()> {
+        if !self.xorbs.read().await.contains_key(hash) || self.xorb_is_tagged(hash).await {
+            return Err(ClientError::Other(format!("XORB not found: {}", hash.hex())));
+        }
+        Ok(())
+    }
+
     async fn shard_is_tagged(&self, hash: &MerkleHash) -> bool {
         self.gc_tagged_shard.read().await.is_some_and(|h| &h == hash)
     }
@@ -171,8 +186,7 @@ impl MemoryClient {
             shard.add_xorb_block(cas_info)?;
         }
 
-        let generation = self.xorb_generation.fetch_add(1, Ordering::Relaxed);
-        self.xorbs.write().await.insert(hash, XorbStorage::Random { xorb, generation });
+        self.xorbs.write().await.insert(hash, XorbStorage::Random { xorb });
         Ok(hash)
     }
 
@@ -271,7 +285,7 @@ impl MemoryClient {
     }
 
     #[cfg(not(target_family = "wasm"))]
-    fn object_tag_from_key_and_payload(prefix: &[u8], key: &MerkleHash, payload: &[u8]) -> ObjectTag {
+    fn object_etag_from_key_and_payload(prefix: &[u8], key: &MerkleHash, payload: &[u8]) -> ObjectETag {
         let key_bytes: [u8; 32] = (*key).into();
         let payload_hash: [u8; 32] = compute_data_hash(payload).into();
         let mut entropy = Vec::with_capacity(prefix.len() + key_bytes.len() + payload_hash.len());
@@ -282,18 +296,13 @@ impl MemoryClient {
     }
 
     #[cfg(not(target_family = "wasm"))]
-    fn xorb_tag(hash: &MerkleHash, storage: &XorbStorage) -> ObjectTag {
+    fn xorb_etag(hash: &MerkleHash, storage: &XorbStorage) -> ObjectETag {
         match storage {
-            XorbStorage::Materialized { entry, generation } => {
-                let mut payload = Vec::from(entry.serialized_data.as_ref());
-                payload.extend_from_slice(&generation.to_le_bytes());
-                Self::object_tag_from_key_and_payload(b"xorb", hash, &payload)
+            XorbStorage::Materialized { entry } => {
+                Self::object_etag_from_key_and_payload(b"xorb", hash, entry.serialized_data.as_ref())
             },
-            XorbStorage::Random { xorb, generation } => {
-                let mut entropy = Vec::with_capacity(16);
-                entropy.extend_from_slice(&xorb.num_chunks().to_le_bytes());
-                entropy.extend_from_slice(&generation.to_le_bytes());
-                Self::object_tag_from_key_and_payload(b"xorb", hash, &entropy)
+            XorbStorage::Random { xorb } => {
+                Self::object_etag_from_key_and_payload(b"xorb", hash, &xorb.num_chunks().to_le_bytes())
             },
         }
     }
@@ -908,10 +917,10 @@ impl Client for MemoryClient {
         let footer_start = serialized_xorb_object.footer_start;
         let serialized_data = serialized_xorb_object.serialized_data;
 
-        // Always overwrite: even if the xorb already exists, we must store it
-        // with a fresh generation so its tag changes, matching production ETag
-        // semantics and ensuring delete_xorb_if_tag_matches is safe under
-        // concurrent uploads.
+        // Always overwrite, but the etag is derived from the stored bytes alone,
+        // so a byte-identical re-upload leaves it unchanged — as S3 does, whose
+        // ETag is a function of content. What tells a re-upload apart is the
+        // `last-upload` tag stamped below, not the etag.
 
         info!("Storing XORB {hash:?} in memory");
 
@@ -937,10 +946,17 @@ impl Client for MemoryClient {
         };
 
         let bytes_written = serialized_data.len();
-        let generation = self.xorb_generation.fetch_add(1, Ordering::Relaxed);
 
         {
+            // The write, the un-tag and the `last-upload` stamp are one critical section:
+            // a delete interleaving between them would erase the tag of the xorb this call
+            // just wrote. Locks are taken tagged-before-xorbs everywhere, so readers that
+            // hold both cannot deadlock against this.
+            let mut tagged = self.gc_tagged_xorbs.write().await;
             let mut xorbs = self.xorbs.write().await;
+            #[cfg(not(target_family = "wasm"))]
+            let mut tag_sets = self.xorb_tag_sets.write().await;
+
             xorbs.insert(
                 hash,
                 XorbStorage::Materialized {
@@ -948,15 +964,14 @@ impl Client for MemoryClient {
                         serialized_data: Bytes::from(serialized_data),
                         xorb_object: xorb_obj,
                     },
-                    generation,
                 },
             );
-        }
 
-        // A re-upload of the same xorb hash supersedes any prior
-        // lifecycle-tagged copy: clear the tag so the xorb is readable again.
-        // Mirrors S3 PutObject overwriting a tagged object.
-        self.gc_tagged_xorbs.write().await.remove(&hash);
+            tagged.remove(&hash);
+
+            #[cfg(not(target_family = "wasm"))]
+            tag_sets.insert(hash, last_upload_tag_set_now());
+        }
 
         if let Some(ref cb) = progress_callback {
             let n = bytes_written as u64;
@@ -1199,43 +1214,73 @@ impl super::DeletionControlableClient for MemoryClient {
     }
 
     async fn delete_xorb(&self, hash: &MerkleHash) {
+        // Removal and tag clear under one critical section, so a concurrent upload cannot
+        // land between them and have its fresh `last-upload` stamp erased by this delete.
+        let mut tagged = self.gc_tagged_xorbs.write().await;
+        let mut xorbs = self.xorbs.write().await;
+        #[cfg(not(target_family = "wasm"))]
+        let mut tag_sets = self.xorb_tag_sets.write().await;
+
         if self.lifecycle_tag_deletion_enabled() {
-            self.gc_tagged_xorbs.write().await.insert(*hash);
+            tagged.insert(*hash);
         } else {
-            self.xorbs.write().await.remove(hash);
+            xorbs.remove(hash);
         }
+        #[cfg(not(target_family = "wasm"))]
+        tag_sets.remove(hash);
     }
 
-    async fn list_xorbs_and_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>> {
+    async fn list_xorbs_and_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>> {
         let tagged = self.gc_tagged_xorbs.read().await;
         let xorbs = self.xorbs.read().await;
         Ok(xorbs
             .iter()
             .filter(|(hash, _)| !tagged.contains(hash))
-            .map(|(hash, storage)| (*hash, Self::xorb_tag(hash, storage)))
+            .map(|(hash, storage)| (*hash, Self::xorb_etag(hash, storage)))
             .collect())
     }
 
-    async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool> {
-        let current_tag = {
-            let xorbs = self.xorbs.read().await;
+    async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool> {
+        // The etag comparison and the delete share one critical section. Dropping the lock
+        // between them would let a re-upload slip in and be deleted on the strength of the
+        // etag it no longer has, which is the very thing this guard exists to prevent.
+        let mut tagged = self.gc_tagged_xorbs.write().await;
+        let mut xorbs = self.xorbs.write().await;
+        #[cfg(not(target_family = "wasm"))]
+        let mut tag_sets = self.xorb_tag_sets.write().await;
+
+        let current_etag = {
             let Some(storage) = xorbs.get(hash) else {
                 return Err(ClientError::XORBNotFound(*hash));
             };
-            Self::xorb_tag(hash, storage)
+            Self::xorb_etag(hash, storage)
         };
-        if &current_tag != tag {
+        if &current_etag != etag {
             return Ok(false);
         }
+
         if self.lifecycle_tag_deletion_enabled() {
-            self.gc_tagged_xorbs.write().await.insert(*hash);
+            tagged.insert(*hash);
         } else {
-            self.xorbs.write().await.remove(hash);
+            xorbs.remove(hash);
         }
+        #[cfg(not(target_family = "wasm"))]
+        tag_sets.remove(hash);
         Ok(true)
     }
 
-    async fn list_shards_with_tags(&self) -> Result<Vec<(MerkleHash, ObjectTag)>> {
+    async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result<ObjectTagSet> {
+        self.require_readable_xorb(hash).await?;
+        Ok(self.xorb_tag_sets.read().await.get(hash).cloned().unwrap_or_default())
+    }
+
+    async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()> {
+        self.require_readable_xorb(hash).await?;
+        self.xorb_tag_sets.write().await.insert(*hash, tags);
+        Ok(())
+    }
+
+    async fn list_shards_with_etags(&self) -> Result<Vec<(MerkleHash, ObjectETag)>> {
         let shard = self.shard.read().await;
         let Some((shard_hash, shard_bytes)) = Self::current_shard_hash_and_bytes(&shard)? else {
             return Ok(Vec::new());
@@ -1243,12 +1288,12 @@ impl super::DeletionControlableClient for MemoryClient {
         if self.shard_is_tagged(&shard_hash).await {
             return Ok(Vec::new());
         }
-        let tag = Self::object_tag_from_key_and_payload(b"shard", &shard_hash, shard_bytes.as_ref());
-        Ok(vec![(shard_hash, tag)])
+        let etag = Self::object_etag_from_key_and_payload(b"shard", &shard_hash, shard_bytes.as_ref());
+        Ok(vec![(shard_hash, etag)])
     }
 
-    async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result<bool> {
-        let (current_hash, current_tag) = {
+    async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result<bool> {
+        let (current_hash, current_etag) = {
             let shard = self.shard.read().await;
             let Some((current_hash, shard_bytes)) = Self::current_shard_hash_and_bytes(&shard)? else {
                 return Err(ClientError::Other(format!("Shard not found: {}", hash.hex())));
@@ -1256,10 +1301,10 @@ impl super::DeletionControlableClient for MemoryClient {
             if &current_hash != hash {
                 return Err(ClientError::Other(format!("Shard not found: {}", hash.hex())));
             }
-            let current_tag = Self::object_tag_from_key_and_payload(b"shard", &current_hash, shard_bytes.as_ref());
-            (current_hash, current_tag)
+            let current_etag = Self::object_etag_from_key_and_payload(b"shard", &current_hash, shard_bytes.as_ref());
+            (current_hash, current_etag)
         };
-        if &current_tag != tag {
+        if &current_etag != etag {
             return Ok(false);
         }
         if self.lifecycle_tag_deletion_enabled() {
@@ -1369,26 +1414,26 @@ mod tests {
         let client = new_deletion_client();
         let file = client.upload_random_file(&[(1, (0, 3))], 2048).await.unwrap();
 
-        let xorbs_and_tags = client.list_xorbs_and_tags().await.unwrap();
-        assert!(!xorbs_and_tags.is_empty());
-        let (xorb_hash, tag) = xorbs_and_tags[0];
+        let xorbs_and_etags = client.list_xorbs_and_etags().await.unwrap();
+        assert!(!xorbs_and_etags.is_empty());
+        let (xorb_hash, tag) = xorbs_and_etags[0];
 
         let wrong_tag = [0xABu8; 32];
-        assert!(!client.delete_xorb_if_tag_matches(&xorb_hash, &wrong_tag).await.unwrap());
+        assert!(!client.delete_xorb_if_etag_matches(&xorb_hash, &wrong_tag).await.unwrap());
         assert!(client.xorb_exists(&xorb_hash).await.unwrap());
 
-        assert!(client.delete_xorb_if_tag_matches(&xorb_hash, &tag).await.unwrap());
+        assert!(client.delete_xorb_if_etag_matches(&xorb_hash, &tag).await.unwrap());
         assert!(!client.xorb_exists(&xorb_hash).await.unwrap());
 
         // file deletion is idempotent for parity with the disk-backed behavior.
         client.delete_file_entry(&file.file_hash).await.unwrap();
         client.delete_file_entry(&file.file_hash).await.unwrap();
 
-        let shards_and_tags = client.list_shards_with_tags().await.unwrap();
+        let shards_and_tags = client.list_shards_with_etags().await.unwrap();
         if !shards_and_tags.is_empty() {
-            let (shard_hash, shard_tag) = shards_and_tags[0];
-            assert!(!client.delete_shard_if_tag_matches(&shard_hash, &wrong_tag).await.unwrap());
-            assert!(client.delete_shard_if_tag_matches(&shard_hash, &shard_tag).await.unwrap());
+            let (shard_hash, shard_etag) = shards_and_tags[0];
+            assert!(!client.delete_shard_if_etag_matches(&shard_hash, &wrong_tag).await.unwrap());
+            assert!(client.delete_shard_if_etag_matches(&shard_hash, &shard_etag).await.unwrap());
             assert!(client.list_shard_entries().await.unwrap().is_empty());
         }
     }
@@ -1618,6 +1663,107 @@ mod tests {
         assert!(client.xorbs.read().await.get(&xorb_hash).is_none());
     }
 
+    /// Tag sets round-trip, replace wholesale on the next write, and leave the
+    /// xorb's `ObjectETag` alone — the same contract `LocalClient` provides.
+    #[tokio::test]
+    async fn test_xorb_tag_set_round_trip_leaves_object_tag_intact() {
+        let client = new_deletion_client();
+        let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        let xorb_hash = file.terms[0].xorb_hash;
+
+        let before = client.list_xorbs_and_etags().await.unwrap();
+
+        client
+            .set_xorb_tag_set(&xorb_hash, vec![("last-upload".to_string(), "1234".to_string())])
+            .await
+            .unwrap();
+        assert_eq!(
+            client.get_xorb_tag_set(&xorb_hash).await.unwrap(),
+            vec![("last-upload".to_string(), "1234".to_string())]
+        );
+
+        client
+            .set_xorb_tag_set(&xorb_hash, vec![("other".to_string(), "x".to_string())])
+            .await
+            .unwrap();
+        assert_eq!(
+            client.get_xorb_tag_set(&xorb_hash).await.unwrap(),
+            vec![("other".to_string(), "x".to_string())],
+            "PutObjectTagging semantics replace the whole set"
+        );
+
+        assert_eq!(client.list_xorbs_and_etags().await.unwrap(), before, "tagging must not move the ObjectETag");
+    }
+
+    /// With upload tagging on, an uploaded xorb carries `last-upload` without
+    /// its etag moving; with it off, no tag set appears at all.
+    #[tokio::test]
+    async fn test_upload_tagging_stamps_last_upload() {
+        let client = new_deletion_client();
+
+        let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        let xorb_hash = file.terms[0].xorb_hash;
+
+        let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap();
+        let (key, value) = tags.first().expect("upload must stamp a tag set");
+        assert_eq!(key, super::super::deletion_controls::LAST_UPLOAD_TAG_KEY);
+        assert!(value.parse::<u64>().is_ok(), "value must be unix seconds, got {value:?}");
+    }
+
+    /// A hard delete takes the tag set with it, so a later upload of the same
+    /// hash cannot inherit tags from the object that used to live there.
+    #[tokio::test]
+    async fn test_delete_xorb_clears_its_tag_set() {
+        let client = new_deletion_client();
+        let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        let xorb_hash = file.terms[0].xorb_hash;
+
+        client
+            .set_xorb_tag_set(&xorb_hash, vec![("stale".to_string(), "value".to_string())])
+            .await
+            .unwrap();
+        client.delete_xorb(&xorb_hash).await;
+
+        // Checked against the map directly: `get_xorb_tag_set` refuses a deleted
+        // xorb, so the leak is invisible through the public surface, and a
+        // re-upload would mask it by replacing the entry wholesale.
+        assert!(
+            !client.xorb_tag_sets.read().await.contains_key(&xorb_hash),
+            "the tag set outlived the xorb it belonged to"
+        );
+
+        // And a re-upload of the same content carries only its own fresh stamp.
+        let refile = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        assert_eq!(refile.terms[0].xorb_hash, xorb_hash);
+        let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap();
+        assert!(!tags.iter().any(|(k, _)| k == "stale"), "stale tags survived a delete: {tags:?}");
+        assert_eq!(tags.len(), 1, "only the fresh last-upload stamp should be present: {tags:?}");
+    }
+
+    /// A lifecycle-tagged xorb reads as gone everywhere else here, and
+    /// `LocalClient` errors on it because the canonical file is renamed away.
+    /// The tag-set accessors must agree rather than quietly serving it.
+    #[tokio::test]
+    async fn test_tag_set_accessors_treat_a_condemned_xorb_as_gone() {
+        let client = new_deletion_client();
+        client.set_lifecycle_tag_deletion(true);
+        let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap();
+        let xorb_hash = file.terms[0].xorb_hash;
+
+        client.delete_xorb(&xorb_hash).await;
+
+        assert!(client.get_xorb_tag_set(&xorb_hash).await.is_err());
+        assert!(client.set_xorb_tag_set(&xorb_hash, vec![]).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_xorb_tag_set_requires_the_xorb_to_exist() {
+        let client = new_deletion_client();
+        let missing = MerkleHash::default();
+        assert!(client.get_xorb_tag_set(&missing).await.is_err());
+        assert!(client.set_xorb_tag_set(&missing, vec![]).await.is_err());
+    }
+
     #[tokio::test]
     async fn test_lifecycle_tag_verify_integrity_flags_tagged_xorb() {
         let client = new_deletion_client();
diff --git a/xet_client/src/cas_client/simulation/mod.rs b/xet_client/src/cas_client/simulation/mod.rs
index d39799555..de586c1aa 100644
--- a/xet_client/src/cas_client/simulation/mod.rs
+++ b/xet_client/src/cas_client/simulation/mod.rs
@@ -30,7 +30,9 @@ mod deletion_controls;
 mod local_client;
 
 #[cfg(not(target_family = "wasm"))]
-pub use deletion_controls::{DeletionControlableClient, ObjectTag};
+pub use deletion_controls::{
+    DeletionControlableClient, LAST_UPLOAD_TAG_KEY, ObjectETag, ObjectTagSet, last_upload_tag_set_now,
+};
 #[cfg(not(target_family = "wasm"))]
 pub use local_client::LocalClient;