fix(ecstore): enable native Windows creation witnesses

Exercise explicit epoch preservation, same-instance reload revocation, and publication between intent persistence and proof grant. Reorder the cohort fixture only after real retry metadata reads.

Validation: rustfmt on both modified files and git diff --check passed. Behavioral validation is pending independent acceptance.

crates/ecstore/src/bucket/metadata_sys.rs

@@ -1553,9 +1553,27 @@ struct MetadataPublishGuard {
     _guard: tokio::sync::OwnedMutexGuard<MetadataPublishLockState>,
 }
 
+#[derive(Debug, Clone)]
+pub(crate) struct PendingBucketCreationProof {
+    pub(crate) generation: Uuid,
+    pub(crate) intent: Vec<u8>,
+    pub(crate) versioning_requested: bool,
+    pub(crate) witnesses: Vec<crate::disk::os::NativeBucketCreationWitness>,
+}
+
+impl PendingBucketCreationProof {
+    pub(crate) fn matches_metadata(&self, metadata: &BucketMetadata) -> Result<bool> {
+        Ok(!metadata.bucket_creation_committed
+            && metadata.bucket_incarnation_sidecar
+            && metadata.bucket_incarnation_id == self.generation
+            && metadata.marshal_msg()? == self.intent)
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct BucketMetadataSys {
     metadata_map: Arc<RwLock<HashMap<String, Arc<BucketMetadata>>>>,
+    pending_bucket_creations: Arc<RwLock<HashMap<String, PendingBucketCreationProof>>>,
     /// Serializes metadata-map commits and their derived cache updates for one
     /// bucket. Namespace locks, when present, are acquired before this lock.
     metadata_publish_locks: Arc<MetadataPublishLockRegistry>,
@@ -1599,6 +1617,7 @@ impl BucketMetadataSys {
     pub fn new(api: Arc<ECStore>) -> Self {
         Self {
             metadata_map: Arc::new(RwLock::new(HashMap::new())),
+            pending_bucket_creations: Arc::new(RwLock::new(HashMap::new())),
             metadata_publish_locks: Arc::new(MetadataPublishLockRegistry {
                 locks: StdMutex::new(HashMap::new()),
             }),
@@ -1629,6 +1648,43 @@ impl BucketMetadataSys {
             .expect("bucket metadata object store should still be live")
     }
 
+    pub(crate) async fn pending_bucket_creation(&self, bucket: &str) -> Option<PendingBucketCreationProof> {
+        self.pending_bucket_creations.read().await.get(bucket).cloned()
+    }
+
+    pub(crate) async fn grant_pending_bucket_creation(&self, bucket: &str, proof: PendingBucketCreationProof) {
+        self.pending_bucket_creations.write().await.insert(bucket.to_owned(), proof);
+    }
+
+    pub(crate) async fn revoke_pending_bucket_creation(&self, bucket: &str) {
+        self.pending_bucket_creations.write().await.remove(bucket);
+    }
+
+    pub(crate) async fn consume_pending_bucket_creation(
+        &self,
+        bucket: &str,
+        metadata: &BucketMetadata,
+        witnesses: &[crate::disk::os::NativeBucketCreationWitness],
+    ) -> Result<bool> {
+        let mut pending = self.pending_bucket_creations.write().await;
+        let Some(proof) = pending.get(bucket) else {
+            return Ok(false);
+        };
+        if !proof.matches_metadata(metadata)?
+            || proof.witnesses.len() != witnesses.len()
+            || !proof
+                .witnesses
+                .iter()
+                .zip(witnesses)
+                .all(|(old, current)| old.matches(current))
+            || !crate::disk::os::consume_native_creation_witnesses(witnesses)
+        {
+            return Ok(false);
+        }
+        pending.remove(bucket);
+        Ok(true)
+    }
+
     fn object_store_if_live(&self) -> Option<Arc<ECStore>> {
         self.api.upgrade()
     }
@@ -1819,6 +1875,7 @@ impl BucketMetadataSys {
         expected: Option<&Arc<BucketMetadata>>,
         namespace_guard: &rustfs_lock::NamespaceLockGuard,
     ) -> Result<()> {
+        self.revoke_pending_bucket_creation(bucket).await;
         if !await_bucket_namespace_operation(
             Some(namespace_guard),
             bucket,
@@ -1902,6 +1959,7 @@ impl BucketMetadataSys {
         persisted: bool,
         namespace_guard: &rustfs_lock::NamespaceLockGuard,
     ) -> Result<()> {
+        self.revoke_pending_bucket_creation(bucket).await;
         if !persisted {
             let _publish_guard = self
                 .lock_metadata_publish(bucket, namespace_guard, "refreshed bucket metadata absence publish")
@@ -1963,6 +2021,7 @@ impl BucketMetadataSys {
     }
 
     pub async fn set(&self, bucket: String, bm: Arc<BucketMetadata>) {
+        self.revoke_pending_bucket_creation(&bucket).await;
         if !is_meta_bucketname(&bucket) {
             let publish_lock = self.metadata_publish_lock(&bucket);
             let _publish_guard = publish_lock.lock().await;
@@ -1984,6 +2043,7 @@ impl BucketMetadataSys {
         if is_meta_bucketname(bucket) {
             return false;
         }
+        self.revoke_pending_bucket_creation(bucket).await;
         let publish_lock = self.metadata_publish_lock(bucket);
         let _publish_guard = publish_lock.lock().await;
         let mut map = self.metadata_map.write().await;
@@ -2001,6 +2061,7 @@ impl BucketMetadataSys {
     }
 
     async fn _reset(&mut self) {
+        self.pending_bucket_creations.write().await.clear();
         let mut map = self.metadata_map.write().await;
         map.clear();
         drop(map);
@@ -2139,6 +2200,7 @@ impl BucketMetadataSys {
     /// server's metadata never leaks into the ambient (first) instance.
     pub(crate) async fn persist_and_set(&self, bm: BucketMetadata) -> Result<()> {
         let mut bm = bm;
+        self.revoke_pending_bucket_creation(&bm.name).await;
         bm.save_with_store(self.object_store()).await?;
 
         self.set(bm.name.clone(), Arc::new(bm)).await;
@@ -2147,6 +2209,7 @@ impl BucketMetadataSys {
     }
 
     async fn persist_new_and_set(&self, mut bm: BucketMetadata) -> Result<()> {
+        self.revoke_pending_bucket_creation(&bm.name).await;
         bm.bucket_creation_committed = true;
         bm.save_with_store_committed(self.object_store()).await?;
         save_bucket_incarnation(self.object_store(), &bm.name, bm.bucket_incarnation_id).await?;
@@ -2156,6 +2219,7 @@ impl BucketMetadataSys {
     }
 
     async fn persist_new_bucket_metadata_intent(&self, mut bm: BucketMetadata) -> Result<()> {
+        self.revoke_pending_bucket_creation(&bm.name).await;
         bm.bucket_creation_committed = false;
         bm.save_with_store(self.object_store()).await?;
         save_bucket_incarnation(self.object_store(), &bm.name, bm.bucket_incarnation_id).await?;
@@ -2166,6 +2230,7 @@ impl BucketMetadataSys {
     }
 
     async fn commit_bucket_metadata(&self, mut bm: BucketMetadata) -> Result<()> {
+        self.revoke_pending_bucket_creation(&bm.name).await;
         bm.parse_all_configs()?;
         bm.bucket_incarnation_sidecar = true;
         // The caller holds the full bucket creation fence (lifecycle, metadata
@@ -2214,6 +2279,7 @@ impl BucketMetadataSys {
         bucket: &str,
         namespace_guard: &rustfs_lock::NamespaceLockGuard,
     ) -> Result<()> {
+        self.revoke_pending_bucket_creation(bucket).await;
         let expected = self.metadata_map.read().await.get(bucket).cloned();
         if !self
             .bucket_exists(bucket, namespace_guard, "peer bucket metadata existence check")
@@ -2270,6 +2336,7 @@ impl BucketMetadataSys {
             #[cfg(test)]
             self.lazy_disk_loads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
+            self.revoke_pending_bucket_creation(bucket).await;
             let lock = self.object_store().new_ns_lock(bucket, bucket).await?;
             let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
             #[cfg(test)]
@@ -2578,6 +2645,7 @@ impl BucketMetadataSys {
     /// revalidated at write quorum, so a stale snapshot can neither revert an
     /// acknowledged configuration update nor outlive a delete/recreate.
     async fn migrate_bucket_creation_commit(&self, bucket: &str) -> Result<BucketMetadataAuthority> {
+        self.revoke_pending_bucket_creation(bucket).await;
         let transaction_lock = self
             .object_store()
             .new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket))
@@ -2655,6 +2723,7 @@ impl BucketMetadataSys {
         &self,
         bucket: &str,
     ) -> Result<BucketMetadataAuthority> {
+        self.revoke_pending_bucket_creation(bucket).await;
         #[cfg(test)]
         if self.object_lock_disk_read_errors.write().await.remove(bucket) {
             return Err(Error::other(format!("injected Object Lock metadata disk read failure: {bucket}")));

crates/ecstore/src/disk/disk_store.rs

@@ -2320,6 +2320,15 @@ impl DiskAPI for LocalDiskWrapper {
         })
     }
 
+    async fn bucket_creation_witness(&self, volume: &str) -> Result<Option<super::os::NativeBucketCreationWitness>> {
+        self.track_disk_health_with_op(
+            "bucket_creation_witness",
+            || async { self.disk.bucket_creation_witness(volume).await },
+            Duration::ZERO,
+        )
+        .await
+    }
+
     async fn make_volume(&self, volume: &str) -> Result<()> {
         // Scoped heal must drain directory creation before releasing its lifecycle owner.
         let timeout = if crate::store::bucket_heal_scope(volume).is_some() {

crates/ecstore/src/disk/fs.rs

@@ -117,6 +117,9 @@ pub const O_APPEND: FileMode = 0x00400;
 //     create_new: bool,
 
 pub async fn open_file(path: impl AsRef<Path>, mode: FileMode) -> io::Result<File> {
+    if mode & O_CREATE != 0 {
+        super::os::record_native_publication(path.as_ref());
+    }
     let base_opts = match mode & (O_RDONLY | O_WRONLY | O_RDWR) {
         O_RDONLY => get_readonly_options(),
         O_WRONLY => get_writeonly_options(),
@@ -160,6 +163,7 @@ pub fn lstat_std(path: impl AsRef<Path>) -> io::Result<Metadata> {
 }
 
 pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
+    super::os::record_native_publication(path.as_ref());
     fs::create_dir_all(path.as_ref()).await
 }
 
@@ -211,15 +215,18 @@ pub fn remove_all_std(path: impl AsRef<Path>) -> io::Result<()> {
 }
 
 pub async fn mkdir(path: impl AsRef<Path>) -> io::Result<()> {
+    super::os::record_native_publication(path.as_ref());
     fs::create_dir(path.as_ref()).await
 }
 
 pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
+    super::os::record_native_publication(to.as_ref());
     fs::rename(from, to).await
 }
 
 #[cfg(any(not(windows), test))]
 pub fn rename_std(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
+    super::os::record_native_publication(to.as_ref());
     std::fs::rename(from, to)
 }
 

crates/ecstore/src/disk/local.rs

@@ -262,6 +262,7 @@ fn snapshot_part_transaction_file(src: &Path, backup: &Path, absent: &Path) -> s
 }
 
 fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, restore: &Path) -> std::io::Result<()> {
+    os::record_native_publication(current);
     match std::fs::symlink_metadata(backup) {
         Ok(metadata) if metadata.is_file() => {
             remove_file_if_exists(restore)?;
@@ -287,6 +288,7 @@ async fn write_delete_rollback_file(
     namespace_owner: Option<Arc<dyn Send + Sync>>,
 ) -> Result<()> {
     let backup_dir = object_dir.join(rollback_dir.to_string());
+    os::record_native_publication(&backup_dir);
     let path = backup_dir.join(name);
     let lease = os::acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
     let data = data.to_vec();
@@ -2148,6 +2150,7 @@ impl AsyncWrite for DirectWriter {
 fn open_direct_writer(file_path: &Path, state: &DirectIoWriteState) -> Result<Option<DirectWriter>> {
     use std::os::unix::fs::OpenOptionsExt;
 
+    os::record_native_publication(file_path);
     let open_result = std::fs::OpenOptions::new()
         .create(true)
         .write(true)
@@ -3920,6 +3923,7 @@ impl LocalIoBackend for StdBackend {
                 if is_direct_io_write_enabled() && self.direct_io_write.supported.load(Ordering::Relaxed) {
                     let write_state = self.direct_io_write.clone();
                     let direct_path = file_path.clone();
+                    os::record_native_publication(&direct_path);
                     let direct = tokio::task::spawn_blocking(move || open_direct_writer(&direct_path, &write_state))
                         .await
                         .map_err(|err| DiskError::other(format!("O_DIRECT open task failed: {err}")))??;
@@ -5167,6 +5171,7 @@ async fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
 pub struct LocalDisk {
     pub root: PathBuf,
     publication_root: os::PublicationRoot,
+    publication_history: Option<os::NativePublicationRegistration>,
     /// I/O root pinned to the mount instance that was opened while the disk
     /// was initialized. On Linux this is `/proc/self/fd/<dirfd>/.`; resolving
     /// paths beneath it keeps repair I/O on that mount even if the configured
@@ -5535,6 +5540,8 @@ impl LocalDisk {
         #[cfg(not(target_os = "linux"))]
         let io_root = Self::open_mount_lease(&root)?;
 
+        let publication_history = os::register_native_publication_root(&root, &io_root);
+
         ensure_data_usage_layout(&io_root)
             .await
             .map_err(|e| e.narrow_to_disk().unwrap_or_else(DiskError::other))
@@ -5670,6 +5677,7 @@ impl LocalDisk {
         let mut disk = Self {
             root: root.clone(),
             publication_root,
+            publication_history,
             io_root: io_root.clone(),
             #[cfg(target_os = "linux")]
             mount_lease,
@@ -7439,6 +7447,7 @@ impl LocalDisk {
                 };
                 if let Some(rollback_dir) = rollback_dir {
                     let rollback_path = object_dir.join(rollback_dir.to_string());
+                    os::record_native_publication(&rollback_path);
                     if let Err(err) = fs::create_dir_all(&rollback_path).await {
                         let err: DiskError = to_file_error(err).into();
                         if reserved_version_delete {
@@ -7961,6 +7970,7 @@ impl LocalDisk {
             }
             InternalBuf::Owned(buf) => {
                 let path = file_path.to_path_buf();
+                os::record_native_publication(&path);
                 if let Some(parent) = path.parent()
                     && parent != skip_parent
                 {
@@ -7968,6 +7978,7 @@ impl LocalDisk {
                 }
 
                 tokio::task::spawn_blocking(move || {
+                    os::record_native_publication(&path);
                     #[cfg(test)]
                     run_owned_file_write_before_open(&path);
 
@@ -9581,6 +9592,9 @@ impl DiskAPI for LocalDisk {
     }
 
     async fn close(&self) -> Result<()> {
+        if let Some(history) = self.publication_history.as_ref() {
+            history.invalidate();
+        }
         // This disk instance is being retired (e.g. replaced by renew_disk on
         // reconnect). Drop its cached descriptors so a replacement instance's
         // invalidations are never defeated by this one continuing to serve stale
@@ -9738,6 +9752,7 @@ impl DiskAPI for LocalDisk {
             let file_path = self.io_get_object_path(volume, path)?;
             let path = path.to_string();
             let sync_metadata = effective_durability(volume).syncs_commit_metadata();
+            os::record_native_publication(&file_path);
             return Ok(tokio::task::spawn_blocking(move || {
                 // A persistent directory lock bounds metadata growth. Removing
                 // per-target lock files can split flock ownership across inodes.
@@ -10126,6 +10141,7 @@ impl DiskAPI for LocalDisk {
                     .map_err(to_file_error)?;
             }
         }
+        os::record_native_publication(&transaction_path);
         tokio::task::spawn_blocking(move || {
             let source = std::fs::symlink_metadata(&src_file_path).map_err(to_file_error)?;
             if !source.is_file() {
@@ -10197,6 +10213,7 @@ impl DiskAPI for LocalDisk {
         }
         let durability = effective_durability(volume);
 
+        os::record_native_publication(&current_data_path);
         tokio::task::spawn_blocking(move || {
             match std::fs::symlink_metadata(&transaction_path) {
                 Ok(metadata) if metadata.is_dir() => {}
@@ -10807,12 +10824,27 @@ impl DiskAPI for LocalDisk {
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
+    async fn bucket_creation_witness(&self, volume: &str) -> Result<Option<os::NativeBucketCreationWitness>> {
+        let Some(history) = self.publication_history.as_ref() else {
+            return Ok(None);
+        };
+        let Some(disk_id) = self.get_disk_id().await? else {
+            return Ok(None);
+        };
+        Ok(history.witness(volume, disk_id, Arc::clone(&self.io_backend)))
+    }
+
     async fn make_volume(&self, volume: &str) -> Result<()> {
         if !Self::is_valid_volname(volume) {
             return Err(Error::other("Invalid arguments specified"));
         }
 
         let volume_dir = self.io_get_bucket_path(volume)?;
+        #[cfg(all(test, not(windows)))]
+        os::prepared_publication_test_hooks::run(
+            os::prepared_publication_test_hooks::Stage::CreateDirectory,
+            &self.root.join(volume),
+        );
 
         // Volume creation is a mutation boundary, so it must observe the live
         // filesystem rather than a potentially stale existence-cache entry.
@@ -11552,6 +11584,54 @@ mod test {
     use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
     use tracing_subscriber::fmt::MakeWriter;
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn owned_direct_child_open_attempt_consumes_native_history() {
+        let dir = tempfile::tempdir().expect("native disk directory");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("native local disk");
+        let bucket = format!("owned-open-attempt-{}", Uuid::new_v4());
+        let witness = disk
+            .publication_history
+            .as_ref()
+            .expect("native registration")
+            .witness(&bucket, Uuid::new_v4(), Arc::clone(&disk.io_backend))
+            .expect("fresh history");
+
+        disk.write_all_public(&bucket, "direct-child", Bytes::from_static(b"payload"))
+            .await
+            .expect_err("the direct child has no bucket parent");
+        assert!(
+            !os::validate_native_creation_witnesses(&[witness]),
+            "even a failed O_CREATE in the real owned executor must permanently consume native history"
+        );
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn synchronous_direct_open_attempt_consumes_native_history() {
+        let dir = tempfile::tempdir().expect("native disk directory");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("native local disk");
+        let bucket = format!("direct-io-open-attempt-{}", Uuid::new_v4());
+        let witness = disk
+            .publication_history
+            .as_ref()
+            .expect("native registration")
+            .witness(&bucket, Uuid::new_v4(), Arc::clone(&disk.io_backend))
+            .expect("fresh history");
+        let path = disk.io_get_object_path(&bucket, "direct-child").expect("direct I/O path");
+
+        assert!(
+            open_direct_writer(&path, &DirectIoWriteState::new()).is_err(),
+            "the actual synchronous O_DIRECT entry must observe the missing bucket parent"
+        );
+        assert!(
+            !os::validate_native_creation_witnesses(&[witness]),
+            "a directly callable O_CREATE helper cannot retain retry permission after a failed publication attempt"
+        );
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn read_dir_entries_skip_symlinks_without_truncating_listing() {

crates/ecstore/src/disk/mod.rs

@@ -327,6 +327,13 @@ impl DiskAPI for Disk {
         }
     }
 
+    async fn bucket_creation_witness(&self, volume: &str) -> Result<Option<os::NativeBucketCreationWitness>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.bucket_creation_witness(volume).await,
+            Disk::Remote(_) => Ok(None),
+        }
+    }
+
     #[tracing::instrument(level = "trace", skip_all)]
     async fn make_volume(&self, volume: &str) -> Result<()> {
         match self {
@@ -1121,6 +1128,11 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     // NSScanner
 
     // Volume operations.
+    /// Unknown or remote backends cannot attest process-local publication history.
+    async fn bucket_creation_witness(&self, _volume: &str) -> Result<Option<os::NativeBucketCreationWitness>> {
+        Ok(None)
+    }
+
     async fn make_volume(&self, volume: &str) -> Result<()>;
     async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>;
     async fn list_volumes(&self) -> Result<Vec<VolumeInfo>>;

crates/ecstore/src/disk/os.rs

@@ -21,7 +21,7 @@ use futures::TryStreamExt;
 use parking_lot::Mutex;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::{HashMap, HashSet, VecDeque},
     io,
     path::{Component, Path, PathBuf},
     sync::{Arc, LazyLock, Weak},
@@ -276,6 +276,10 @@ pub(crate) mod prepared_publication_test_hooks {
         Rollback,
         #[cfg(test)]
         DirFsync,
+        #[cfg(test)]
+        CreateDirectory,
+        #[cfg(test)]
+        NativeRootIdentity,
     }
 
     type Hook = Box<dyn FnOnce() + Send>;
@@ -1381,6 +1385,228 @@ fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'stati
 }
 static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
+
+#[cfg(windows)]
+type NativeRootIdentity = (u64, [u8; 16]);
+#[cfg(not(windows))]
+type NativeRootIdentity = (u64, u64);
+
+// Publication history outlives disk handles, cleanup, and metadata snapshots.
+// Reusing a pathname or descriptor for another physical root is not new proof.
+#[derive(Debug, Default)]
+struct NativePublicationHistory {
+    roots: HashMap<NativeRootIdentity, NativeRootHistory>,
+    aliases: HashMap<PathBuf, HashSet<NativeRootIdentity>>,
+}
+
+#[derive(Debug, Default)]
+struct NativeRootHistory {
+    unknown: bool,
+    started: HashSet<String>,
+}
+
+static NATIVE_PUBLICATION_HISTORY: LazyLock<Mutex<NativePublicationHistory>> =
+    LazyLock::new(|| Mutex::new(NativePublicationHistory::default()));
+
+#[derive(Clone, Debug)]
+pub(crate) struct NativePublicationRegistration {
+    identity: NativeRootIdentity,
+    root: PathBuf,
+    io_root: PathBuf,
+}
+
+/// Opaque, process-local observation. Unsupported disks cannot construct it.
+#[derive(Clone, Debug)]
+pub struct NativeBucketCreationWitness {
+    registration: NativePublicationRegistration,
+    backend: Arc<dyn super::local::LocalIoBackend>,
+    bucket: String,
+    disk_id: uuid::Uuid,
+}
+
+fn native_root_identity(path: &Path) -> Option<NativeRootIdentity> {
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::MetadataExt;
+        #[cfg(test)]
+        prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::NativeRootIdentity, path);
+        let metadata = std::fs::metadata(path).ok()?;
+        metadata.is_dir().then(|| (metadata.dev(), metadata.ino()))
+    }
+    #[cfg(windows)]
+    {
+        // Resolve and validate the same directory anchor used for publication;
+        // keep its volume and complete 128-bit file ID in the history key.
+        let (_, directory) = open_windows_publication_root(path).ok()?;
+        windows_file_identity(directory.handle.as_ref()).ok()
+    }
+    #[cfg(all(not(unix), not(windows)))]
+    {
+        let _ = path;
+        None
+    }
+}
+
+pub(crate) fn register_native_publication_root(root: &Path, io_root: &Path) -> Option<NativePublicationRegistration> {
+    let identity = native_root_identity(root)?;
+    if native_root_identity(io_root) != Some(identity) {
+        return None;
+    }
+    let mut history = NATIVE_PUBLICATION_HISTORY.lock();
+    history.roots.entry(identity).or_default();
+    for alias in [root, io_root] {
+        let alias: PathBuf = alias.components().collect();
+        let previous = history.aliases.entry(alias).or_default();
+        let collisions: Vec<_> = previous.iter().copied().filter(|old| *old != identity).collect();
+        previous.insert(identity);
+        if !collisions.is_empty() {
+            history.roots.get_mut(&identity)?.unknown = true;
+            for old in collisions {
+                history.roots.get_mut(&old)?.unknown = true;
+            }
+        }
+    }
+    Some(NativePublicationRegistration {
+        identity,
+        root: root.to_path_buf(),
+        io_root: io_root.to_path_buf(),
+    })
+}
+
+/// Record before queueing filesystem work. Cancellation never removes Started.
+pub(crate) fn record_native_publication(path: &Path) {
+    let aliases: Vec<_> = {
+        let history = NATIVE_PUBLICATION_HISTORY.lock();
+        path.ancestors()
+            .filter(|alias| history.aliases.contains_key(*alias))
+            .map(Path::to_path_buf)
+            .collect()
+    };
+    // Physical lookups must never hold the process-wide publication mutex.
+    let observations: Vec<_> = aliases
+        .into_iter()
+        .filter_map(|alias| {
+            let relative = path.strip_prefix(&alias).ok()?;
+            let mut components = relative.components().filter(|part| !matches!(part, Component::CurDir));
+            let Some(first) = components.next() else {
+                return None;
+            };
+            let bucket = match first {
+                Component::Normal(bucket) if components.all(|part| matches!(part, Component::Normal(_))) => {
+                    bucket.to_str().map(str::to_owned)
+                }
+                _ => None,
+            };
+            let physical = native_root_identity(&alias);
+            Some((alias, physical, bucket))
+        })
+        .collect();
+    let mut history = NATIVE_PUBLICATION_HISTORY.lock();
+    for (alias, physical, bucket) in observations {
+        // Aliases only accumulate identities. Include registration concurrent
+        // with the stat, rather than using a stale captured identity list.
+        let identities = history.aliases.get(&alias).cloned().unwrap_or_default();
+        if physical.is_none() || identities.iter().any(|identity| Some(*identity) != physical) {
+            for identity in identities {
+                history.roots.entry(identity).or_default().unknown = true;
+            }
+            if let Some(identity) = physical {
+                history.roots.entry(identity).or_default().unknown = true;
+            }
+        } else if let Some(identity) = physical {
+            let root = history.roots.entry(identity).or_default();
+            match bucket {
+                Some(bucket) => {
+                    root.started.insert(bucket);
+                }
+                None => root.unknown = true,
+            }
+        }
+    }
+}
+
+impl NativePublicationRegistration {
+    pub(crate) fn invalidate(&self) {
+        if let Some(root) = NATIVE_PUBLICATION_HISTORY.lock().roots.get_mut(&self.identity) {
+            root.unknown = true;
+        }
+    }
+
+    pub(crate) fn witness(
+        &self,
+        bucket: &str,
+        disk_id: uuid::Uuid,
+        backend: Arc<dyn super::local::LocalIoBackend>,
+    ) -> Option<NativeBucketCreationWitness> {
+        if disk_id.is_nil()
+            || !matches!(Path::new(bucket).components().next(), Some(Component::Normal(_)))
+            || Path::new(bucket).components().count() != 1
+        {
+            return None;
+        }
+        let witness = NativeBucketCreationWitness {
+            registration: self.clone(),
+            backend,
+            bucket: bucket.to_owned(),
+            disk_id,
+        };
+        validate_native_creation_witnesses(std::slice::from_ref(&witness)).then_some(witness)
+    }
+}
+
+impl NativeBucketCreationWitness {
+    pub(crate) fn matches(&self, other: &Self) -> bool {
+        self.registration.identity == other.registration.identity
+            && Arc::ptr_eq(&self.backend, &other.backend)
+            && self.disk_id == other.disk_id
+            && self.bucket == other.bucket
+    }
+
+    fn physical_root_is_current(&self) -> bool {
+        let root = native_root_identity(&self.registration.root);
+        let io_root = native_root_identity(&self.registration.io_root);
+        if root == Some(self.registration.identity) && io_root == Some(self.registration.identity) {
+            return true;
+        }
+        let mut history = NATIVE_PUBLICATION_HISTORY.lock();
+        history.roots.entry(self.registration.identity).or_default().unknown = true;
+        for identity in [root, io_root].into_iter().flatten() {
+            history.roots.entry(identity).or_default().unknown = true;
+        }
+        false
+    }
+
+    fn valid_in(&self, history: &NativePublicationHistory) -> bool {
+        history
+            .roots
+            .get(&self.registration.identity)
+            .is_some_and(|root| !root.unknown && !root.started.contains(&self.bucket))
+    }
+}
+
+pub(crate) fn validate_native_creation_witnesses(witnesses: &[NativeBucketCreationWitness]) -> bool {
+    if witnesses.is_empty() || !witnesses.iter().all(NativeBucketCreationWitness::physical_root_is_current) {
+        return false;
+    }
+    let history = NATIVE_PUBLICATION_HISTORY.lock();
+    witnesses.iter().all(|witness| witness.valid_in(&history))
+}
+
+pub(crate) fn consume_native_creation_witnesses(witnesses: &[NativeBucketCreationWitness]) -> bool {
+    if witnesses.is_empty() || !witnesses.iter().all(NativeBucketCreationWitness::physical_root_is_current) {
+        return false;
+    }
+    let mut history = NATIVE_PUBLICATION_HISTORY.lock();
+    if !witnesses.iter().all(|witness| witness.valid_in(&history)) {
+        return false;
+    }
+    for witness in witnesses {
+        if let Some(root) = history.roots.get_mut(&witness.registration.identity) {
+            root.started.insert(witness.bucket.clone());
+        }
+    }
+    true
+}
 type NamespaceMutationLock = AsyncMutex<()>;
 type NamespaceMutationLockRegistry = HashMap<PathBuf, Weak<NamespaceMutationLock>>;
 static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
@@ -2083,6 +2309,7 @@ pub(in crate::disk) async fn rename_with_namespace_owner(
     dst: &Path,
     namespace_owner: Option<Arc<dyn Send + Sync>>,
 ) -> io::Result<()> {
+    record_native_publication(dst);
     if namespace_owner.is_none() {
         return tokio::fs::rename(src, dst).await;
     }
@@ -2104,12 +2331,19 @@ pub(in crate::disk) async fn create_dir_all_with_namespace_owner(
     path: &Path,
     namespace_owner: Option<Arc<dyn Send + Sync>>,
 ) -> io::Result<()> {
+    record_native_publication(path);
     if namespace_owner.is_none() {
         return tokio::fs::create_dir_all(path).await;
     }
     let path = path.to_path_buf();
     let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
-    run_blocking_namespace_operation(lease, move || std::fs::create_dir_all(path)).await
+    run_blocking_namespace_operation(lease, move || {
+        #[cfg(all(test, not(windows)))]
+        prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::CreateDirectory, &path);
+        record_native_publication(&path);
+        std::fs::create_dir_all(path)
+    })
+    .await
 }
 
 #[tracing::instrument(name = "rename_all", level = "debug", skip_all)]
@@ -2120,6 +2354,7 @@ pub(crate) async fn rename_all_with_owner(
     publication_root: &PublicationRoot,
     namespace_owner: Option<Arc<dyn Send + Sync>>,
 ) -> Result<()> {
+    record_native_publication(dst_file_path.as_ref());
     let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await;
     rename_all_with_lease(src_file_path, dst_file_path, base_dir, publication_root, lease).await
 }
@@ -2131,6 +2366,7 @@ pub(crate) async fn rename_all_with_lease(
     publication_root: &PublicationRoot,
     lease: Arc<NamespaceMutationLease>,
 ) -> Result<()> {
+    record_native_publication(dst_file_path.as_ref());
     reliable_rename_inner_with_lease(
         src_file_path.as_ref().to_path_buf(),
         dst_file_path.as_ref().to_path_buf(),
@@ -2155,6 +2391,7 @@ pub(crate) async fn rename_reconciled_metadata(
     owner: Option<Arc<dyn Send + Sync>>,
     authority: Arc<crate::bucket::lifecycle::legacy_transition_state_reconcile::TransitionStateReconcileAuthority>,
 ) -> Result<()> {
+    record_native_publication(&destination);
     let lease = acquire_namespace_mutation_lease_with_owner(&destination, owner).await;
     run_blocking_namespace_operation(lease, move || {
         let preparation = prepare_rename_with_retry(&source, &destination, &base_dir, &publication_root)?;
@@ -2399,6 +2636,7 @@ pub(crate) async fn rename_all_ignore_missing_source_with_owner(
     publication_root: &PublicationRoot,
     namespace_owner: Option<Arc<dyn Send + Sync>>,
 ) -> Result<()> {
+    record_native_publication(dst_file_path.as_ref());
     let src_file_path = src_file_path.as_ref();
     let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await;
     match reliable_rename_inner_with_lease(
@@ -2453,6 +2691,7 @@ async fn reliable_rename_inner(
     publication_root: &PublicationRoot,
     warn_on_missing_source: bool,
 ) -> io::Result<()> {
+    record_native_publication(dst_file_path.as_ref());
     let src_file_path = src_file_path.as_ref().to_path_buf();
     let dst_file_path = dst_file_path.as_ref().to_path_buf();
     let base_dir = base_dir.as_ref().to_path_buf();
@@ -2476,6 +2715,7 @@ async fn reliable_rename_inner_with_lease(
     warn_on_missing_source: bool,
     lease: Arc<NamespaceMutationLease>,
 ) -> io::Result<()> {
+    record_native_publication(&dst_file_path);
     let operation = {
         let src_file_path = src_file_path.clone();
         let dst_file_path = dst_file_path.clone();
@@ -2878,6 +3118,7 @@ fn rename_into_existing_parent(
     dst_file_path: &Path,
     parent_guard: Option<&ExistingBaseDirectoryGuard>,
 ) -> io::Result<()> {
+    record_native_publication(dst_file_path);
     use rustix::fs::{Mode, OFlags, open, renameat};
 
     let Some(parent_guard) = parent_guard else {
@@ -2926,6 +3167,7 @@ fn rename_into_existing_parent(
     parent_guard: Option<&ExistingBaseDirectoryGuard>,
     source: &winapi_util::Handle,
 ) -> io::Result<()> {
+    record_native_publication(dst_file_path);
     use std::{
         mem::size_of,
         os::windows::{ffi::OsStrExt, io::AsRawHandle},
@@ -4214,6 +4456,7 @@ pub(crate) fn mkdir_all_below_existing_base_std(
     base_dir: &Path,
     publication_root: &PublicationRoot,
 ) -> io::Result<ExistingBaseDirectoryGuard> {
+    record_native_publication(dir_path);
     let relative = dir_path
         .strip_prefix(base_dir)
         .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory"))?;
@@ -5718,6 +5961,77 @@ mod tests {
         assert_eq!(err.raw_os_error(), Some(disk_full));
     }
 
+    #[cfg(windows)]
+    #[test]
+    fn windows_native_root_identity_matches_directory_handles() {
+        let temp = tempdir().expect("create native identity fixture");
+        let first = temp.path().join("first");
+        let second = temp.path().join("second");
+        std::fs::create_dir(&first).expect("create first root");
+        std::fs::create_dir(&second).expect("create second root");
+        let (_, first_anchor) = open_windows_publication_root(&first).expect("open first directory anchor");
+        let (_, second_anchor) = open_windows_publication_root(&second).expect("open second directory anchor");
+        let first_identity = windows_file_identity(first_anchor.handle.as_ref()).expect("query full first identity");
+        let second_identity = windows_file_identity(second_anchor.handle.as_ref()).expect("query full second identity");
+
+        assert_eq!(native_root_identity(&first), Some(first_identity));
+        assert_eq!(native_root_identity(&first.join(".")), Some(first_identity));
+        assert_eq!(native_root_identity(&second), Some(second_identity));
+        assert_ne!(first_identity, second_identity, "distinct physical roots must have distinct identities");
+    }
+
+    #[cfg(windows)]
+    #[tokio::test]
+    async fn windows_native_creation_witness_retains_started_history() {
+        let temp = tempdir().expect("create native witness fixture");
+        let root = std::fs::canonicalize(temp.path()).expect("canonical native root");
+        let _publication_root = PublicationRoot::new(&root).expect("retain native directory anchor");
+        let registration = register_native_publication_root(&root, &root.join(".")).expect("register same physical root");
+        let backend: Arc<dyn super::super::local::LocalIoBackend> = Arc::new(super::super::local::StdBackend::new(root.clone()));
+        let disk_id = uuid::Uuid::new_v4();
+        let witness = registration
+            .witness("published", disk_id, Arc::clone(&backend))
+            .expect("fresh native witness");
+        let bucket = root.join("published");
+
+        super::super::fs::mkdir(&bucket)
+            .await
+            .expect("publish through the actual native entry");
+        assert!(bucket.is_dir(), "the production entry must create the directory");
+        assert!(!validate_native_creation_witnesses(std::slice::from_ref(&witness)));
+        std::fs::remove_dir(&bucket).expect("remove published directory");
+        assert!(registration.witness("published", disk_id, Arc::clone(&backend)).is_none());
+        let reconnected = register_native_publication_root(&root, &root).expect("reconnect same physical root");
+        assert!(reconnected.witness("published", disk_id, Arc::clone(&backend)).is_none());
+
+        let unstarted = reconnected
+            .witness("unstarted", disk_id, Arc::clone(&backend))
+            .expect("untouched bucket keeps its witness");
+        assert!(consume_native_creation_witnesses(std::slice::from_ref(&unstarted)));
+        assert!(!validate_native_creation_witnesses(std::slice::from_ref(&unstarted)));
+        assert!(reconnected.witness("unstarted", disk_id, backend).is_none());
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn windows_native_creation_witness_rejects_unknown_roots() {
+        let temp = tempdir().expect("create unknown identity fixture");
+        let root = temp.path().join("root");
+        let other = temp.path().join("other");
+        let file = temp.path().join("file");
+        std::fs::create_dir(&root).expect("create native root");
+        std::fs::create_dir(&other).expect("create distinct native root");
+        std::fs::write(&file, b"not a directory").expect("create regular file");
+
+        assert!(native_root_identity(&temp.path().join("missing")).is_none());
+        assert!(native_root_identity(&file).is_none());
+        assert!(register_native_publication_root(&root, &other).is_none());
+        let registration = register_native_publication_root(&root, &root).expect("register actual native root");
+        let backend: Arc<dyn super::super::local::LocalIoBackend> = Arc::new(super::super::local::StdBackend::new(root));
+        registration.invalidate();
+        assert!(registration.witness("unstarted", uuid::Uuid::new_v4(), backend).is_none());
+    }
+
     #[cfg(windows)]
     #[test]
     fn windows_rename_all_legacy_file_identity_distinguishes_sources() {
@@ -7737,3 +8051,272 @@ mod tests {
         assert_eq!(err.kind(), io::ErrorKind::NotFound);
     }
 }
+
+#[cfg(all(test, unix))]
+mod native_creation_history_tests {
+    use super::*;
+
+    struct NativeFixture {
+        _temp: tempfile::TempDir,
+        root: PathBuf,
+        registration: NativePublicationRegistration,
+        backend: Arc<dyn super::super::local::LocalIoBackend>,
+        disk_id: uuid::Uuid,
+    }
+
+    impl NativeFixture {
+        fn new() -> Self {
+            let temp = tempfile::tempdir().expect("native root");
+            let root = std::fs::canonicalize(temp.path()).expect("canonical root");
+            let registration = register_native_publication_root(&root, &root).expect("native registration");
+            let backend = Arc::new(super::super::local::StdBackend::new(root.clone()));
+            Self {
+                _temp: temp,
+                root,
+                registration,
+                backend,
+                disk_id: uuid::Uuid::new_v4(),
+            }
+        }
+
+        fn witness(&self, bucket: &str) -> Option<NativeBucketCreationWitness> {
+            self.registration.witness(bucket, self.disk_id, Arc::clone(&self.backend))
+        }
+    }
+
+    #[tokio::test]
+    async fn every_native_directory_publication_entry_invalidates_retry() {
+        let fixture = NativeFixture::new();
+        for operation in [
+            "recursive",
+            "single",
+            "open",
+            "rename",
+            "rename_std",
+            "fallback",
+            "owned_mkdir",
+            "owned_rename",
+            "handle_relative",
+        ] {
+            let bucket = format!("native-{operation}-{}", uuid::Uuid::new_v4());
+            let path = fixture.root.join(&bucket);
+            let witness = fixture.witness(&bucket).expect("fresh native witness");
+            match operation {
+                "recursive" => super::super::fs::make_dir_all(path.join("object/child"))
+                    .await
+                    .expect("recursive mkdir"),
+                "single" => super::super::fs::mkdir(&path).await.expect("single mkdir"),
+                "open" => {
+                    let file = super::super::fs::open_file(&path, super::super::fs::O_CREATE | super::super::fs::O_WRONLY)
+                        .await
+                        .expect("create file");
+                    drop(file);
+                }
+                "rename" | "rename_std" | "owned_rename" => {
+                    let source = fixture.root.join(format!("source-{}", uuid::Uuid::new_v4()));
+                    std::fs::create_dir(&source).expect("rename source");
+                    match operation {
+                        "rename" => super::super::fs::rename(&source, &path).await.expect("async rename"),
+                        "rename_std" => super::super::fs::rename_std(&source, &path).expect("sync rename"),
+                        _ => rename_with_namespace_owner(&source, &path, Some(Arc::new(())))
+                            .await
+                            .expect("owned rename"),
+                    }
+                }
+                "fallback" => make_dir_all(path.join("object/child"), &path)
+                    .await
+                    .expect("implicit volume fallback"),
+                "owned_mkdir" => create_dir_all_with_namespace_owner(&path.join("child"), Some(Arc::new(())))
+                    .await
+                    .expect("owned recursive mkdir"),
+                "handle_relative" => {
+                    let publication_root = PublicationRoot::new(&fixture.root).expect("publication root");
+                    let guard = mkdir_all_below_existing_base_std(&path.join("child"), &fixture.root, &publication_root)
+                        .expect("first-level handle-relative mkdir");
+                    drop(guard);
+                }
+                _ => unreachable!(),
+            }
+            assert!(
+                !validate_native_creation_witnesses(std::slice::from_ref(&witness)),
+                "{operation}: old proof remains live"
+            );
+            if path.is_dir() {
+                std::fs::remove_dir_all(&path).expect("remove published directory");
+            } else {
+                std::fs::remove_file(&path).expect("remove published file");
+            }
+            assert!(fixture.witness(&bucket).is_none(), "{operation}: removal must not regrant proof");
+        }
+    }
+
+    #[tokio::test]
+    async fn cancelled_namespace_waiter_cannot_restore_native_proof() {
+        let fixture = NativeFixture::new();
+        let bucket = format!("native-cancel-{}", uuid::Uuid::new_v4());
+        let path = fixture.root.join(&bucket);
+        let witness = fixture.witness(&bucket).expect("fresh witness");
+        let lease = acquire_namespace_mutation_lease(&path).await;
+        let mut attempt = Box::pin(create_dir_all_with_namespace_owner(&path, Some(Arc::new(()))));
+        assert!(matches!(futures::poll!(&mut attempt), std::task::Poll::Pending));
+        assert!(
+            !validate_native_creation_witnesses(std::slice::from_ref(&witness)),
+            "Started must precede admission wait"
+        );
+        assert!(!path.exists(), "the namespace wait must still precede the syscall");
+        drop(attempt);
+        drop(lease);
+        assert!(fixture.witness(&bucket).is_none(), "cancellation cannot undo an admitted attempt");
+    }
+
+    #[tokio::test]
+    async fn queued_mkdir_cancellation_retains_lease_and_consumed_history() {
+        use futures::FutureExt;
+        use std::sync::mpsc;
+
+        let fixture = NativeFixture::new();
+        let bucket = format!("native-queued-cancel-{}", uuid::Uuid::new_v4());
+        let path = fixture.root.join(&bucket);
+        let witness = fixture.witness(&bucket).expect("fresh witness");
+        let (entered_tx, entered_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        let _hook = prepared_publication_test_hooks::install_at(
+            prepared_publication_test_hooks::Stage::CreateDirectory,
+            &path,
+            move || {
+                entered_tx.send(()).expect("signal real executor entry");
+                let _ = release_rx.recv();
+            },
+        );
+        let operation_path = path.clone();
+        let operation =
+            tokio::spawn(async move { create_dir_all_with_namespace_owner(&operation_path, Some(Arc::new(()))).await });
+        tokio::task::spawn_blocking(move || entered_rx.recv_timeout(std::time::Duration::from_secs(30)))
+            .await
+            .expect("entry waiter")
+            .expect("the real blocking executor must enter");
+        operation.abort();
+        assert!(operation.await.expect_err("cancel the async waiter").is_cancelled());
+        assert!(
+            !validate_native_creation_witnesses(std::slice::from_ref(&witness)),
+            "queued publication is permanently Started"
+        );
+        assert!(
+            acquire_namespace_mutation_lease(&path).now_or_never().is_none(),
+            "the actual blocked executor must still own the namespace lease"
+        );
+        release_tx.send(()).expect("release the syscall");
+        let drained = tokio::time::timeout(std::time::Duration::from_secs(30), acquire_namespace_mutation_lease(&path))
+            .await
+            .expect("the lease must drain with the syscall");
+        drop(drained);
+        assert!(path.is_dir(), "the cancelled waiter cannot discard the already queued syscall");
+        std::fs::remove_dir_all(&path).expect("remove the published directory");
+        assert!(fixture.witness(&bucket).is_none(), "drain and removal cannot restore proof");
+    }
+
+    #[tokio::test]
+    async fn publication_detects_retargeted_physical_alias_without_registration() {
+        let source = NativeFixture::new();
+        let destination = NativeFixture::new();
+        let bucket = format!("native-retarget-{}", uuid::Uuid::new_v4());
+        let witness = destination.witness(&bucket).expect("destination witness");
+        let retired = source.root.with_extension("retired");
+        std::fs::rename(&source.root, &retired).expect("move the original physical root");
+        std::os::unix::fs::symlink(&destination.root, &source.root).expect("retarget the registered alias");
+        super::super::fs::mkdir(source.root.join(&bucket))
+            .await
+            .expect("publish through the changed alias");
+        assert!(
+            !validate_native_creation_witnesses(std::slice::from_ref(&witness)),
+            "publication-time alias drift must invalidate the actual destination root"
+        );
+        std::fs::remove_dir(destination.root.join(&bucket)).expect("remove the published bucket");
+        assert!(destination.witness(&bucket).is_none(), "removal must not undo alias uncertainty");
+        std::fs::remove_file(&source.root).expect("remove the symlink");
+        std::fs::rename(retired, &source.root).expect("restore the test root");
+    }
+
+    #[test]
+    fn blocked_physical_identity_lookup_does_not_block_other_publication() {
+        use std::sync::mpsc;
+        let stalled = NativeFixture::new();
+        let healthy = NativeFixture::new();
+        let witness = stalled.witness("unpublished").expect("stalled root witness");
+        let (entered_tx, entered_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        let _hook = prepared_publication_test_hooks::install_at(
+            prepared_publication_test_hooks::Stage::NativeRootIdentity,
+            &stalled.root,
+            move || {
+                entered_tx.send(()).expect("signal physical lookup");
+                let _ = release_rx.recv();
+            },
+        );
+        let validator = std::thread::spawn(move || validate_native_creation_witnesses(&[witness]));
+        entered_rx
+            .recv_timeout(std::time::Duration::from_secs(30))
+            .expect("enter the actual identity lookup");
+        let (published_tx, published_rx) = mpsc::channel();
+        let path = healthy.root.join("published");
+        let publisher = std::thread::spawn(move || {
+            record_native_publication(&path);
+            published_tx.send(()).expect("signal independent publication");
+        });
+        let published = published_rx.recv_timeout(std::time::Duration::from_secs(5));
+        release_tx.send(()).expect("release the identity lookup");
+        validator.join().expect("identity lookup thread");
+        publisher.join().expect("publication thread");
+        assert!(
+            published.is_ok(),
+            "a slow physical lookup must not retain the process-wide publication mutex"
+        );
+    }
+
+    #[test]
+    fn native_consume_is_atomic_and_history_survives_reconnect() {
+        let fixture = NativeFixture::new();
+        let first = format!("native-first-{}", uuid::Uuid::new_v4());
+        let second = format!("native-second-{}", uuid::Uuid::new_v4());
+        let first_witness = fixture.witness(&first).expect("first witness");
+        let second_witness = fixture.witness(&second).expect("second witness");
+        record_native_publication(&fixture.root.join(&second));
+        assert!(!consume_native_creation_witnesses(&[first_witness.clone(), second_witness]));
+        assert!(
+            validate_native_creation_witnesses(std::slice::from_ref(&first_witness)),
+            "failed cohort consume must be atomic"
+        );
+        assert!(consume_native_creation_witnesses(std::slice::from_ref(&first_witness)));
+        assert!(!validate_native_creation_witnesses(std::slice::from_ref(&first_witness)));
+        let reconnected = register_native_publication_root(&fixture.root, &fixture.root).expect("reconnected registration");
+        let backend: Arc<dyn super::super::local::LocalIoBackend> =
+            Arc::new(super::super::local::StdBackend::new(fixture.root.clone()));
+        assert!(
+            reconnected.witness(&first, fixture.disk_id, backend).is_none(),
+            "reconnect must retain Started"
+        );
+    }
+
+    #[test]
+    fn physical_alias_reuse_is_unknown_and_backend_identity_is_real() {
+        let temp = tempfile::tempdir().expect("alias parent");
+        let root = temp.path().join("root");
+        std::fs::create_dir(&root).expect("first physical root");
+        let old = register_native_publication_root(&root, &root).expect("first registration");
+        let backend: Arc<dyn super::super::local::LocalIoBackend> = Arc::new(super::super::local::StdBackend::new(root.clone()));
+        let id = uuid::Uuid::new_v4();
+        let witness = old.witness("same-name", id, Arc::clone(&backend)).expect("old witness");
+        let other_backend: Arc<dyn super::super::local::LocalIoBackend> =
+            Arc::new(super::super::local::StdBackend::new(root.clone()));
+        let other = old.witness("same-name", id, other_backend).expect("new backend witness");
+        assert!(!witness.matches(&other), "a different actual backend cannot reuse a pending observation");
+        std::fs::rename(&root, temp.path().join("retired-root")).expect("retire physical root");
+        std::fs::create_dir(&root).expect("replacement root at the same alias");
+        let replacement = register_native_publication_root(&root, &root).expect("replacement registration");
+        assert!(replacement.witness("same-name", id, backend).is_none(), "alias reuse must remain Unknown");
+        assert!(
+            !validate_native_creation_witnesses(&[witness]),
+            "old physical identity must not validate at a new root"
+        );
+    }
+}

crates/ecstore/src/store/bucket.rs

@@ -17,6 +17,11 @@ use crate::bucket::{
     metadata::{BUCKET_TABLE_RESERVED_PREFIX, BUCKET_TAGGING_CONFIG, ConfigState, table_bucket_catalog_metadata_prefix},
     utils::is_meta_bucketname,
 };
+use crate::disk::{
+    DiskAPI,
+    error::DiskError,
+    error_reduce::{BUCKET_OP_IGNORED_ERRS, reduce_write_quorum_errs},
+};
 use crate::error::is_err_bucket_not_found;
 use crate::runtime::sources as runtime_sources;
 use crate::set_disk::{BucketInfoQuorum, get_lock_acquire_timeout};
@@ -32,6 +37,104 @@ const SCANNER_BUCKET_LIST_SET_CONCURRENCY: usize = 4;
 const EVENT_BUCKET_DELETE_BLOCKED: &str = "bucket_delete_blocked";
 const EVENT_BUCKET_DELETE_ROLLBACK_FAILED: &str = "bucket_delete_rollback_failed";
 
+#[cfg(test)]
+static FAIL_BEFORE_BUCKET_PUBLICATION: std::sync::LazyLock<std::sync::Mutex<std::collections::HashSet<String>>> =
+    std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new()));
+
+#[cfg(test)]
+pub(crate) fn fail_next_bucket_creation_after_intent(bucket: &str) {
+    FAIL_BEFORE_BUCKET_PUBLICATION
+        .lock()
+        .expect("creation fault mutex")
+        .insert(bucket.to_owned());
+}
+
+#[cfg(test)]
+#[derive(Clone, Copy, Eq, Hash, PartialEq)]
+pub(crate) enum BucketCreationPausePoint {
+    AfterIntentPersistence,
+    BeforeRetryPublication,
+}
+
+#[cfg(test)]
+struct BucketCreationPause {
+    entered: tokio::sync::oneshot::Sender<()>,
+    release: tokio::sync::oneshot::Receiver<()>,
+}
+
+#[cfg(test)]
+static BUCKET_CREATION_PAUSES: std::sync::LazyLock<
+    std::sync::Mutex<std::collections::HashMap<(String, BucketCreationPausePoint), BucketCreationPause>>,
+> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashMap::new()));
+
+#[cfg(test)]
+pub(crate) struct BucketCreationPauseGuard {
+    bucket: String,
+    point: BucketCreationPausePoint,
+}
+
+#[cfg(test)]
+impl Drop for BucketCreationPauseGuard {
+    fn drop(&mut self) {
+        BUCKET_CREATION_PAUSES
+            .lock()
+            .expect("creation pause mutex")
+            .remove(&(self.bucket.clone(), self.point));
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn pause_next_bucket_creation(
+    bucket: &str,
+    point: BucketCreationPausePoint,
+) -> (
+    BucketCreationPauseGuard,
+    tokio::sync::oneshot::Receiver<()>,
+    tokio::sync::oneshot::Sender<()>,
+) {
+    let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+    let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+    BUCKET_CREATION_PAUSES.lock().expect("creation pause mutex").insert(
+        (bucket.to_owned(), point),
+        BucketCreationPause {
+            entered: entered_tx,
+            release: release_rx,
+        },
+    );
+    (
+        BucketCreationPauseGuard {
+            bucket: bucket.to_owned(),
+            point,
+        },
+        entered_rx,
+        release_tx,
+    )
+}
+
+#[cfg(test)]
+async fn pause_bucket_creation(bucket: &str, point: BucketCreationPausePoint) {
+    let pause = BUCKET_CREATION_PAUSES
+        .lock()
+        .expect("creation pause mutex")
+        .remove(&(bucket.to_owned(), point));
+    if let Some(pause) = pause {
+        let _ = pause.entered.send(());
+        let _ = pause.release.await;
+    }
+}
+
+fn orphaned_creation_requires_recovery(generation: Uuid) -> StorageError {
+    StorageError::InvalidArgument(
+        "CreateBucket".to_string(),
+        "bucket".to_string(),
+        format!("orphaned bucket generation {generation} requires administrator recovery before recreation"),
+    )
+}
+
+fn creation_intent_lock_lost(bucket: &str) -> StorageError {
+    Error::other(format!("bucket metadata creation intent lock was lost: {bucket}"))
+}
+
 /// Record why `DeleteBucket` refused, and at a level that matches who can act
 /// on it.
 ///
@@ -352,6 +455,169 @@ impl ECStore {
         Ok(())
     }
 
+    async fn native_creation_witnesses(&self, bucket: &str) -> Option<Vec<crate::disk::os::NativeBucketCreationWitness>> {
+        let mut witnesses = Vec::new();
+        for (_, _, set) in self.bucket_sets() {
+            let disks = set.disks.read().await.clone();
+            if disks.is_empty() {
+                return None;
+            }
+            for disk in disks {
+                let disk = disk?;
+                if !matches!(disk.stat_volume(bucket).await, Err(DiskError::VolumeNotFound)) {
+                    return None;
+                }
+                witnesses.push(disk.bucket_creation_witness(bucket).await.ok()??);
+            }
+        }
+        crate::disk::os::validate_native_creation_witnesses(&witnesses).then_some(witnesses)
+    }
+
+    async fn grant_native_pending_creation<F>(
+        &self,
+        bucket: &str,
+        opts: &MakeBucketOptions,
+        metadata: &BucketMetadata,
+        before: &[crate::disk::os::NativeBucketCreationWitness],
+        fences_valid: F,
+    ) -> Option<()>
+    where
+        F: Fn() -> bool + Send,
+    {
+        let after = self.native_creation_witnesses(bucket).await?;
+        if before.len() != after.len() || !before.iter().zip(&after).all(|(old, current)| old.matches(current)) {
+            return None;
+        }
+        let (persisted, present) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket)
+            .await
+            .ok()?;
+        let mut expected = metadata.clone();
+        expected.bucket_creation_committed = false;
+        expected.bucket_incarnation_sidecar = true;
+        expected.default_timestamps();
+        let intent = persisted.marshal_msg().ok()?;
+        if !present
+            || persisted.bucket_creation_committed
+            || !persisted.bucket_incarnation_sidecar
+            || intent != expected.marshal_msg().ok()?
+        {
+            return None;
+        }
+        let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx).ok()?;
+        let sys = sys.read().await.clone();
+        if !fences_valid() || !crate::disk::os::validate_native_creation_witnesses(&after) {
+            return None;
+        }
+        sys.grant_pending_bucket_creation(
+            bucket,
+            metadata_sys::PendingBucketCreationProof {
+                generation: persisted.bucket_incarnation_id,
+                intent,
+                versioning_requested: opts.versioning_enabled,
+                witnesses: after,
+            },
+        )
+        .await;
+        Some(())
+    }
+
+    async fn continue_native_bucket_creation<F>(
+        &self,
+        bucket: &str,
+        opts: &MakeBucketOptions,
+        metadata: &BucketMetadata,
+        fences_valid: F,
+    ) -> Result<()>
+    where
+        F: Fn() -> bool + Send,
+    {
+        let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
+        let sys = sys.read().await.clone();
+        let sets: Vec<_> = self.bucket_sets().map(|(_, _, set)| set).collect();
+        let mut inventories = Vec::with_capacity(sets.len());
+        for set in &sets {
+            inventories.push(set.disks.read().await);
+        }
+        if inventories.is_empty() || inventories.iter().any(|disks| disks.is_empty()) {
+            return Err(orphaned_creation_requires_recovery(metadata.bucket_incarnation_id));
+        }
+        let mut witnesses = Vec::new();
+        for disks in &inventories {
+            for disk in disks.iter() {
+                let Some(disk) = disk else {
+                    return Err(orphaned_creation_requires_recovery(metadata.bucket_incarnation_id));
+                };
+                match disk.stat_volume(bucket).await {
+                    Err(DiskError::VolumeNotFound) => {}
+                    Ok(_) => {
+                        return Err(StorageError::InvalidArgument(
+                            "CreateBucket".to_owned(),
+                            "bucket".to_owned(),
+                            "an old bucket volume remains on disk; administrator recovery refused".to_owned(),
+                        ));
+                    }
+                    Err(error) => return Err(error.into()),
+                }
+                let Some(witness) = disk.bucket_creation_witness(bucket).await? else {
+                    return Err(orphaned_creation_requires_recovery(metadata.bucket_incarnation_id));
+                };
+                witnesses.push(witness);
+            }
+        }
+        if !fences_valid() {
+            return Err(creation_intent_lock_lost(bucket));
+        }
+        if !sys.consume_pending_bucket_creation(bucket, metadata, &witnesses).await? {
+            return Err(orphaned_creation_requires_recovery(metadata.bucket_incarnation_id));
+        }
+        if !fences_valid() {
+            return Err(creation_intent_lock_lost(bucket));
+        }
+
+        // Use the verified handles, retaining inventory locks through fanout.
+        // Calling set.make_bucket here would recursively read the same locks.
+        let results = futures::future::join_all(inventories.iter().map(|disks| async move {
+            let results = futures::future::join_all(disks.iter().map(|disk| async move {
+                match disk {
+                    Some(disk) => match disk.make_volume(bucket).await {
+                        Ok(()) => Ok(()),
+                        Err(err) if opts.force_create && matches!(err, DiskError::VolumeExists) => Ok(()),
+                        Err(err) => Err(err),
+                    },
+                    None => Err(DiskError::DiskNotFound),
+                }
+            }))
+            .await;
+            let errs = results
+                .into_iter()
+                .map(|result| result.err())
+                .collect::<Vec<Option<DiskError>>>();
+            match reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, (disks.len() / 2) + 1) {
+                Some(err) => Err(err.into()),
+                None => Ok(()),
+            }
+        }))
+        .await;
+        let mut bucket_exists_error = None;
+        let mut first_hard_error = None;
+        for result in results {
+            let Err(err) = result else {
+                continue;
+            };
+            if matches!(err, StorageError::VolumeExists) || is_err_bucket_exists(&err) {
+                if bucket_exists_error.is_none() {
+                    bucket_exists_error = Some(err);
+                }
+            } else if first_hard_error.is_none() {
+                first_hard_error = Some(err);
+            }
+        }
+        match first_hard_error.or(bucket_exists_error) {
+            Some(err) => Err(err),
+            None => Ok(()),
+        }
+    }
+
     /// Return whether an old bucket generation has a durable retirement record.
     /// A live matching incarnation is never considered retired, even if a stale
     /// record exists from a previous generation.
@@ -901,6 +1167,7 @@ impl ECStore {
             }
         };
         let confirmed_missing = existing_bucket_info.is_none();
+        let mut retry_metadata = None;
         if confirmed_missing
             && !opts.no_lock
             && !is_meta_bucketname(bucket)
@@ -921,13 +1188,30 @@ impl ECStore {
                     ),
                     other => other,
                 })?;
-            return Err(StorageError::InvalidArgument(
-                "CreateBucket".to_string(),
-                "bucket".to_string(),
-                format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
-            ));
+            let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
+            let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
+            let sys = sys.read().await.clone();
+            let proof = sys.pending_bucket_creation(bucket).await;
+            let metadata_store = metadata_sys::object_store_in(&self.ctx).await?;
+            if !persisted
+                || metadata.bucket_incarnation_id != old_incarnation
+                || !metadata.lock_enabled
+                || !opts.lock_enabled
+                || opts.created_at.is_some_and(|created| created != metadata.created)
+                || crate::bucket::retirement::is_retired(metadata_store, bucket, old_incarnation).await?
+                || !match proof {
+                    Some(proof) => proof.versioning_requested == opts.versioning_enabled && proof.matches_metadata(&metadata)?,
+                    None => false,
+                }
+            {
+                return Err(orphaned_creation_requires_recovery(old_incarnation));
+            }
+            retry_metadata = Some(metadata);
         }
-        let existing_metadata = if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
+        let continuing_creation = retry_metadata.is_some();
+        let existing_metadata = if let Some(metadata) = retry_metadata {
+            Some(metadata)
+        } else if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
             let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
             if !persisted {
                 metadata = BucketMetadata::new(bucket);
@@ -966,7 +1250,7 @@ impl ECStore {
             }
         });
         let existing_incarnation_is_authoritative = meta.bucket_incarnation_sidecar;
-        if confirmed_missing || is_meta_bucketname(bucket) {
+        if (confirmed_missing || is_meta_bucketname(bucket)) && !continuing_creation {
             meta.set_created(opts.created_at);
 
             if opts.versioning_enabled {
@@ -982,7 +1266,7 @@ impl ECStore {
         // replica unlocked, so replicated versions could be deleted without the
         // retention the source enforces.
         let lock_newly_enabled = opts.lock_enabled && !meta.lock_enabled;
-        if opts.lock_enabled {
+        if opts.lock_enabled && !continuing_creation {
             meta.lock_enabled = true;
             meta.object_lock_config_xml = crate::bucket::utils::serialize(&*crate::store::ENABLED_OBJECT_LOCK_CONFIG)?;
             meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&ENABLED_VERSIONING_CONFIG)?;
@@ -990,12 +1274,61 @@ impl ECStore {
 
         let metadata_persisted_before_physical = confirmed_missing && !is_meta_bucketname(bucket) && opts.lock_enabled;
         if metadata_persisted_before_physical {
-            metadata_sys::set_new_bucket_metadata_intent_in(&self.ctx, meta.clone()).await?;
+            if !continuing_creation {
+                // Only the fenced producer that saw neither authoritative
+                // metadata nor a sidecar may grant a continuation witness.
+                // Unknown observations disable proof, without changing the
+                // ordinary fresh creation's persistence and error behavior.
+                let prior_intent_absent = if !opts.no_lock {
+                    match metadata_sys::object_store_if_initialized_in(&self.ctx).await {
+                        Some(store) => {
+                            matches!(crate::bucket::metadata::load_bucket_incarnation(store, bucket).await, Ok(None))
+                                && matches!(
+                                    metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await,
+                                    Ok((_, false))
+                                )
+                        }
+                        None => false,
+                    }
+                } else {
+                    false
+                };
+                let before = if prior_intent_absent {
+                    self.native_creation_witnesses(bucket).await
+                } else {
+                    None
+                };
+                metadata_sys::set_new_bucket_metadata_intent_in(&self.ctx, meta.clone()).await?;
+                #[cfg(test)]
+                pause_bucket_creation(bucket, BucketCreationPausePoint::AfterIntentPersistence).await;
+                if let Some(before) = before {
+                    // Optional proof failure does not change an ordinary fresh create.
+                    let _ = self
+                        .grant_native_pending_creation(bucket, opts, &meta, &before, || {
+                            !bucket_lifecycle_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                                && !metadata_transaction_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                                && !ns_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                        })
+                        .await;
+                }
+            }
             if bucket_lifecycle_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
                 || metadata_transaction_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
                 || ns_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
             {
-                return Err(Error::other(format!("bucket metadata creation intent lock was lost: {bucket}")));
+                if let Ok(sys) = metadata_sys::require_bucket_metadata_sys_in(&self.ctx) {
+                    let sys = sys.read().await.clone();
+                    sys.revoke_pending_bucket_creation(bucket).await;
+                }
+                return Err(creation_intent_lock_lost(bucket));
+            }
+            #[cfg(test)]
+            if FAIL_BEFORE_BUCKET_PUBLICATION
+                .lock()
+                .expect("creation fault mutex")
+                .remove(bucket)
+            {
+                return Err(Error::other("injected failure after bucket creation intent persistence"));
             }
         }
 
@@ -1017,9 +1350,23 @@ impl ECStore {
                 bucket,
                 "bucket creation metadata transaction",
                 async {
-                    self.make_bucket_on_sets(bucket, opts)
+                    let result = if continuing_creation {
+                        #[cfg(test)]
+                        pause_bucket_creation(bucket, BucketCreationPausePoint::BeforeRetryPublication).await;
+                        self.continue_native_bucket_creation(bucket, opts, &meta, || {
+                            !bucket_lifecycle_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                                && !metadata_transaction_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                                && !ns_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                        })
                         .await
-                        .map_err(|err| to_object_err(err, vec![bucket]))
+                    } else {
+                        if let Ok(sys) = metadata_sys::require_bucket_metadata_sys_in(&self.ctx) {
+                            let sys = sys.read().await.clone();
+                            sys.revoke_pending_bucket_creation(bucket).await;
+                        }
+                        self.make_bucket_on_sets(bucket, opts).await
+                    };
+                    result.map_err(|err| to_object_err(err, vec![bucket]))
                 },
             ),
         )

crates/ecstore/src/store/init.rs

@@ -866,7 +866,6 @@ mod tests {
         should_defer_rebalance_auto_start, should_retry_format_load, wait_for_local_decommission_resume_delay,
     };
     use crate::core::pools::PoolMetaBootstrapAuthority;
-    #[cfg(feature = "test-util")]
     use crate::disk::DiskAPI;
     #[cfg(feature = "test-util")]
     use crate::{
@@ -18316,11 +18315,25 @@ mod tests {
         crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
 
         let bucket = format!("lock-create-intent-retry-{}", uuid::Uuid::new_v4());
-        let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket);
-        intent.lock_enabled = true;
-        crate::bucket::metadata_sys::set_new_bucket_metadata_in(&ctx, intent)
+        crate::store::bucket::fail_next_bucket_creation_after_intent(&bucket);
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
             .await
-            .expect("persist pre-visibility lock intent");
+            .expect_err("fail the real fenced producer before physical publication");
+        let (intent, persisted) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+            .await
+            .expect("read the real pre-visibility lock intent");
+        assert!(persisted, "the original complete intent must be persisted");
+        assert!(!intent.bucket_creation_committed, "the producer must not have committed");
+        assert!(intent.bucket_incarnation_sidecar, "the pending UUID must have a sidecar");
+        assert!(!intent.bucket_incarnation_id.is_nil(), "the fresh producer must bind a real UUID");
+        let original_intent = intent.marshal_msg().expect("encode the original intent");
         assert!(
             store
                 .peer_sys
@@ -18346,6 +18359,531 @@ mod tests {
                 .expect("read retried lock state"),
             crate::bucket::metadata_sys::ObjectLockConfigState::Configured { .. }
         ));
+        let (mut committed, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+            .await
+            .expect("read committed retry metadata");
+        assert!(committed.bucket_creation_committed, "the retry must commit the original generation");
+        assert_eq!(
+            committed.bucket_incarnation_id, intent.bucket_incarnation_id,
+            "retry must not rotate the UUID"
+        );
+        committed.bucket_creation_committed = false;
+        assert_eq!(
+            committed.marshal_msg().expect("encode retried intent"),
+            original_intent,
+            "retry must preserve all original creation and configuration values"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn lock_enabled_creation_retry_preserves_explicit_unix_epoch() {
+        let temp = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "lock-retry-epoch", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("lock-retry-epoch-{}", uuid::Uuid::new_v4());
+        let options = MakeBucketOptions {
+            lock_enabled: true,
+            created_at: Some(OffsetDateTime::UNIX_EPOCH),
+            ..Default::default()
+        };
+        crate::store::bucket::fail_next_bucket_creation_after_intent(&bucket);
+        store
+            .make_bucket(&bucket, &options)
+            .await
+            .expect_err("stop the real epoch producer before physical publication");
+        let (intent, persisted) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+            .await
+            .expect("read the persisted epoch intent");
+        assert!(persisted, "the epoch intent must be persisted");
+        assert_eq!(intent.created, OffsetDateTime::UNIX_EPOCH, "persist the exact supplied creation time");
+        let original_intent = intent.marshal_msg().expect("encode the complete epoch intent");
+
+        store
+            .make_bucket(&bucket, &options)
+            .await
+            .expect("retry the original epoch intent");
+        assert!(matches!(
+            crate::bucket::metadata_sys::get_object_lock_config_state_in(&ctx, &bucket)
+                .await
+                .expect("read the retried epoch lock state"),
+            crate::bucket::metadata_sys::ObjectLockConfigState::Configured { .. }
+        ));
+        let (mut committed, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+            .await
+            .expect("read the committed epoch metadata");
+        assert!(committed.bucket_creation_committed, "the epoch retry must commit");
+        assert_eq!(committed.created, OffsetDateTime::UNIX_EPOCH, "retry must preserve the supplied epoch");
+        committed.bucket_creation_committed = false;
+        assert_eq!(
+            committed.marshal_msg().expect("encode the retried epoch intent"),
+            original_intent,
+            "retry must preserve the original generation and complete epoch intent"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn creation_retry_rejects_publication_between_intent_persistence_and_proof() {
+        use crate::store::bucket::{BucketCreationPausePoint, pause_next_bucket_creation};
+
+        let temp = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "lock-post-persist", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("lock-post-persist-{}", uuid::Uuid::new_v4());
+        let disks = store.pools[0].disk_set[0].disks.read().await.clone();
+        let disk = disks[0].clone().expect("native disk");
+        let (pause, entered, release) = pause_next_bucket_creation(&bucket, BucketCreationPausePoint::AfterIntentPersistence);
+        crate::store::bucket::fail_next_bucket_creation_after_intent(&bucket);
+        let producer_store = store.clone();
+        let producer_bucket = bucket.clone();
+        let producer = tokio::spawn(async move {
+            producer_store
+                .make_bucket(
+                    &producer_bucket,
+                    &MakeBucketOptions {
+                        lock_enabled: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(30), entered)
+            .await
+            .expect("producer persistence pause timeout")
+            .expect("producer must finish real intent persistence");
+        let (intent, persisted) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+            .await
+            .expect("read the intent while its original fences remain held");
+        assert!(persisted, "the original intent must precede the competing publication");
+        let original_intent = intent
+            .marshal_msg()
+            .expect("encode the original competing-publication intent");
+        disk.make_volume(&bucket)
+            .await
+            .expect("finish the competing native publication");
+        disk.delete_volume(&bucket, true)
+            .await
+            .expect("remove the competing physical volume");
+        release.send(()).expect("resume the original producer");
+        let result = producer.await;
+        drop(pause);
+        let error = result
+            .expect("original producer task")
+            .expect_err("the existing fault must stop the original producer before its own fanout");
+        assert!(
+            error
+                .to_string()
+                .contains("injected failure after bucket creation intent persistence")
+        );
+        let sys = crate::bucket::metadata_sys::require_bucket_metadata_sys_in(&ctx).expect("instance metadata sys");
+        let sys = sys.read().await.clone();
+        assert!(
+            sys.pending_bucket_creation(&bucket).await.is_none(),
+            "a completed publication attempt must prevent the original proof grant"
+        );
+        let (after_producer, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+            .await
+            .expect("read the original intent after the interrupted producer");
+        assert_eq!(
+            after_producer.marshal_msg().expect("encode the interrupted producer intent"),
+            original_intent,
+            "competing publication must not replace the original generation or intent"
+        );
+        for disk in &disks {
+            assert!(matches!(
+                disk.as_ref().expect("native disk").stat_volume(&bucket).await,
+                Err(crate::disk::error::DiskError::VolumeNotFound)
+            ));
+        }
+        let error = store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("the old same-generation create must remain manual recovery");
+        assert!(error.to_string().contains("administrator recovery"));
+        let (after_retry, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+            .await
+            .expect("read the unchanged intent after retry rejection");
+        assert_eq!(
+            after_retry.marshal_msg().expect("encode the rejected retry intent"),
+            original_intent,
+            "the denied retry must preserve the complete original intent"
+        );
+        for disk in &disks {
+            assert!(matches!(
+                disk.as_ref().expect("native disk").stat_volume(&bucket).await,
+                Err(crate::disk::error::DiskError::VolumeNotFound)
+            ));
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn committed_lock_intent_without_volumes_requires_recovery() {
+        let temp = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "lock-committed-intent-reject", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("lock-committed-{}", uuid::Uuid::new_v4());
+        let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket);
+        intent.lock_enabled = true;
+        crate::bucket::metadata_sys::set_new_bucket_metadata_in(&ctx, intent)
+            .await
+            .expect("persist the original committed fixture");
+        let error = store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("committed metadata is not a never-published proof");
+        assert!(error.to_string().contains("administrator recovery"));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn pending_lock_intent_setter_cannot_grant_creation_retry() {
+        let temp = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "lock-pending-setter-reject", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("lock-pending-setter-reject-{}", uuid::Uuid::new_v4());
+        let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket);
+        intent.lock_enabled = true;
+        crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, intent)
+            .await
+            .expect("persist an uncommitted fixture without a fresh producer");
+        let error = store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("false alone is not trusted publication history");
+        assert!(error.to_string().contains("administrator recovery"));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn metadata_without_sidecar_cannot_become_fresh_retry_proof() {
+        let temp = tempfile::tempdir().expect("create temp store dir");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "lock-metadata-only-residue", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("lock-metadata-only-residue-{}", uuid::Uuid::new_v4());
+        let mut residue = crate::bucket::metadata::BucketMetadata::new(&bucket);
+        residue.lock_enabled = true;
+        residue
+            .save_with_store(store.clone())
+            .await
+            .expect("persist only the old metadata, without a sidecar");
+
+        crate::store::bucket::fail_next_bucket_creation_after_intent(&bucket);
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("stop the real producer after replacing the metadata residue");
+        let error = store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("replacing old metadata must not turn it into never-published proof");
+        assert!(error.to_string().contains("administrator recovery"));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn creation_retry_rejects_publication_removal_and_changed_intent() {
+        use crate::storage_api_contracts::namespace::NamespaceLocking as _;
+        use crate::store::bucket::{BucketCreationPausePoint, pause_next_bucket_creation};
+
+        let temp = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "lock-retry-history", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        for case in [
+            "volume",
+            "implicit_parent",
+            "lock_intent",
+            "version_intent",
+            "reload",
+            "same_instance_reload",
+            "setter_restore",
+            "cohort_order",
+            "created_at",
+            "generation",
+            "retired",
+        ] {
+            let bucket = format!("retry-{}-{}", case.replace('_', "-"), uuid::Uuid::new_v4());
+            let original = MakeBucketOptions {
+                lock_enabled: true,
+                ..Default::default()
+            };
+            crate::store::bucket::fail_next_bucket_creation_after_intent(&bucket);
+            store
+                .make_bucket(&bucket, &original)
+                .await
+                .expect_err("stop the real producer before publication");
+            let sys = crate::bucket::metadata_sys::require_bucket_metadata_sys_in(&ctx).expect("instance metadata sys");
+            let sys = sys.read().await.clone();
+            sys.pending_bucket_creation(&bucket)
+                .await
+                .expect("the fresh producer must issue positive proof");
+            let mut retry = MakeBucketOptions {
+                lock_enabled: true,
+                ..Default::default()
+            };
+            match case {
+                "volume" | "implicit_parent" => {
+                    let set = &store.pools[0].disk_set[0];
+                    let disk = set.disks.read().await[0].clone().expect("native disk");
+                    if case == "volume" {
+                        disk.make_volume(&bucket).await.expect("raw disk publication");
+                    } else {
+                        crate::disk::fs::make_dir_all(disk.path().join(&bucket).join("implicit/parent"))
+                            .await
+                            .expect("implicit parent publication");
+                    }
+                    disk.delete_volume(&bucket, true).await.expect("remove the published volume");
+                }
+                "lock_intent" => retry.lock_enabled = false,
+                "version_intent" => retry.versioning_enabled = true,
+                "reload" => crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await,
+                "same_instance_reload" => {
+                    let (intent, persisted) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+                        .await
+                        .expect("read the original intent before same-instance reload");
+                    assert!(persisted, "same-instance reload starts from a real persisted intent");
+                    let original_intent = intent.marshal_msg().expect("encode the intent before reload");
+                    sys.reload_from_store(&bucket)
+                        .await
+                        .expect_err("an absent physical bucket cannot be peer-adopted");
+                    assert!(
+                        sys.pending_bucket_creation(&bucket).await.is_none(),
+                        "same-instance reload must revoke the old retry proof even when its bucket is absent"
+                    );
+                    let (after_reload, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+                        .await
+                        .expect("read the persisted intent after same-instance reload rejection");
+                    assert_eq!(
+                        after_reload.marshal_msg().expect("encode the intent after reload"),
+                        original_intent,
+                        "reload rejection must preserve the original generation and complete intent"
+                    );
+                }
+                "setter_restore" => {
+                    let (intent, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+                        .await
+                        .expect("load the producer's original intent");
+                    let mut changed = intent.clone();
+                    changed.created += time::Duration::seconds(1);
+                    crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, changed)
+                        .await
+                        .expect("ordinary setter replaces the intent");
+                    crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, intent)
+                        .await
+                        .expect("ordinary setter restores the original bytes and generation");
+                }
+                "cohort_order" => {}
+                "created_at" => {
+                    let (intent, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+                        .await
+                        .expect("read the original creation time");
+                    retry.created_at = Some(intent.created + time::Duration::seconds(1));
+                }
+                "generation" => {
+                    let (mut intent, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+                        .await
+                        .expect("read the original generation");
+                    intent.bucket_incarnation_id = uuid::Uuid::new_v4();
+                    intent
+                        .save_with_store(store.clone())
+                        .await
+                        .expect("replace the actual persisted metadata generation");
+                    crate::bucket::metadata::save_bucket_incarnation(store.clone(), &bucket, intent.bucket_incarnation_id)
+                        .await
+                        .expect("replace the actual sidecar generation");
+                }
+                "retired" => {
+                    let (intent, _) = crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&ctx, &bucket)
+                        .await
+                        .expect("read the pending generation");
+                    let lifecycle_guard = store
+                        .acquire_bucket_lifecycle_write_lock(&bucket)
+                        .await
+                        .expect("retirement lifecycle fence");
+                    let namespace = store.new_ns_lock(&bucket, &bucket).await.expect("retirement namespace");
+                    let namespace_guard = namespace
+                        .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
+                        .await
+                        .expect("retirement namespace fence");
+                    let mut opts = ObjectOptions {
+                        max_parity: true,
+                        ..Default::default()
+                    };
+                    opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
+                    opts.add_namespace_lock_guard(&namespace_guard);
+                    crate::bucket::retirement::commit_retirement(store.clone(), &bucket, intent.bucket_incarnation_id, &opts)
+                        .await
+                        .expect("persist retirement of the pending generation under the real fences");
+                }
+                _ => unreachable!(),
+            }
+            let result = if case == "cohort_order" {
+                let (pause, entered, release) =
+                    pause_next_bucket_creation(&bucket, BucketCreationPausePoint::BeforeRetryPublication);
+                let retry_store = store.clone();
+                let retry_bucket = bucket.clone();
+                let retry_task = tokio::spawn(async move { retry_store.make_bucket(&retry_bucket, &retry).await });
+                tokio::time::timeout(std::time::Duration::from_secs(30), entered)
+                    .await
+                    .expect("retry cohort pause timeout")
+                    .expect("retry must finish metadata reads under its real fences");
+                store.pools[0].disk_set[0].disks.write().await.rotate_left(1);
+                release.send(()).expect("resume retry with a reordered physical cohort");
+                let result = retry_task.await;
+                store.pools[0].disk_set[0].disks.write().await.rotate_right(1);
+                drop(pause);
+                result.expect("cohort retry task")
+            } else {
+                store.make_bucket(&bucket, &retry).await
+            };
+            let error = result.expect_err("unproven continuation must remain manual");
+            assert!(error.to_string().contains("administrator recovery"), "{case}: {error}");
+        }
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(storage_class_env)]
+    async fn creation_retry_retains_the_verified_cohort_through_physical_fanout() {
+        use crate::disk::os::prepared_publication_test_hooks;
+        use std::sync::mpsc;
+
+        let temp = tempfile::tempdir().expect("create temp store dir");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "lock-retry-cohort-lease", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("lock-retry-cohort-lease-{}", uuid::Uuid::new_v4());
+        crate::store::bucket::fail_next_bucket_creation_after_intent(&bucket);
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("stop the real producer before physical publication");
+        let set = store.pools[0].disk_set[0].clone();
+        let path = set.disks.read().await[0].as_ref().expect("native disk").path().join(&bucket);
+        let (entered_tx, entered_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        let _hook = prepared_publication_test_hooks::install_at(
+            prepared_publication_test_hooks::Stage::CreateDirectory,
+            &path,
+            move || {
+                entered_tx.send(()).expect("signal real make-volume entry");
+                let _ = release_rx.recv();
+            },
+        );
+        let retry_store = store.clone();
+        let retry_bucket = bucket.clone();
+        let retry = tokio::spawn(async move {
+            retry_store
+                .make_bucket(
+                    &retry_bucket,
+                    &MakeBucketOptions {
+                        lock_enabled: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::task::spawn_blocking(move || entered_rx.recv_timeout(std::time::Duration::from_secs(30)))
+            .await
+            .expect("publication waiter")
+            .expect("retry must reach the actual disk publication call");
+        let inventory_is_locked = set.disks.try_write().is_err();
+        release_tx.send(()).expect("release physical publication");
+        retry
+            .await
+            .expect("retry task")
+            .expect("the verified cohort should finish physical creation");
+        assert!(
+            inventory_is_locked,
+            "the whole verified cohort must remain read-locked through actual disk fanout"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn creation_retry_rejects_a_present_missing_or_unknown_native_disk() {
+        for case in ["present", "missing", "unknown"] {
+            let temp = tempfile::tempdir().expect("create temp store dir");
+            let (_ctx, store, _shutdown) =
+                without_storage_class_env(build_isolated_test_store(temp.path(), "lock-retry-disk-denial", &[4])).await;
+            crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+            let bucket = format!("lock-retry-disk-{case}-{}", uuid::Uuid::new_v4());
+            crate::store::bucket::fail_next_bucket_creation_after_intent(&bucket);
+            store
+                .make_bucket(
+                    &bucket,
+                    &MakeBucketOptions {
+                        lock_enabled: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("stop the real producer before publication");
+            let set = &store.pools[0].disk_set[0];
+            let disk = set.disks.read().await[0].clone().expect("native disk");
+            match case {
+                "present" => disk.make_volume(&bucket).await.expect("leave one physical volume present"),
+                "missing" => set.disks.write().await[0] = None,
+                "unknown" => disk.close().await.expect("invalidate the actual native backend"),
+                _ => unreachable!(),
+            }
+            store
+                .make_bucket(
+                    &bucket,
+                    &MakeBucketOptions {
+                        lock_enabled: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("one disk's absent-or-never-published evidence cannot be assumed");
+            if case == "missing" {
+                set.disks.write().await[0] = Some(disk);
+            }
+        }
     }
 
     #[tokio::test]