diff --git a/runtime/src/deterministic.rs b/runtime/src/deterministic.rs
index dcf1252b23..f07a935f3b 100644
--- a/runtime/src/deterministic.rs
+++ b/runtime/src/deterministic.rs
@@ -1117,6 +1117,14 @@ impl Context {
         self.executor().auditor.clone()
     }
 
+    fn fill_random(&self, dest: &mut [u8]) {
+        let executor = self.executor();
+        executor.auditor.event(b"rand", |hasher| {
+            hasher.update(b"fill_bytes");
+        });
+        executor.rng.lock().fill_bytes(dest);
+    }
+
     /// Compute a [Sha256] digest of all storage contents.
     pub fn storage_audit(&self) -> Digest {
         self.storage.inner().inner().inner().audit()
@@ -1593,11 +1601,7 @@ impl TryRng for Context {
     }
 
     fn try_fill_bytes(&mut self, dest: &mut [u8]) -> Result<(), Self::Error> {
-        let executor = self.executor();
-        executor.auditor.event(b"rand", |hasher| {
-            hasher.update(b"fill_bytes");
-        });
-        executor.rng.lock().fill_bytes(dest);
+        self.fill_random(dest);
         Ok(())
     }
 }
@@ -1625,6 +1629,32 @@ impl crate::Storage for Context {
     }
 }
 
+impl crate::atomic::Backend for Context {
+    type Worker = Self;
+
+    fn atomic_worker(&self) -> Self {
+        <Self as crate::Supervisor>::child(self, "atomic_storage")
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.storage.atomic_resources()
+    }
+
+    fn new_atomic_identifier(&self) -> [u8; 16] {
+        let mut identifier = [0; 16];
+        self.fill_random(&mut identifier);
+        identifier
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        self.storage.open_atomic_existing(partition, name).await
+    }
+}
+
 impl crate::BufferPooler for Context {
     fn network_buffer_pool(&self) -> &crate::BufferPool {
         &self.network_buffer_pool
@@ -1641,8 +1671,9 @@ mod tests {
     #[cfg(feature = "external")]
     use crate::FutureExt;
     use crate::{
-        Blob, Metrics as _, Resolver, Runner as _, Spawner as _, Storage, Strategizer,
-        Supervisor as _, WriteOptions, deterministic, reschedule,
+        AtomicBlob as _, AtomicStorage as _, BatchOperation, Blob, Metrics as _, Resolver,
+        Runner as _, Spawner as _, Storage, Strategizer, Supervisor as _, WriteOptions,
+        deterministic, reschedule,
     };
     use commonware_macros::test_traced;
     use commonware_parallel::Strategy;
@@ -1700,6 +1731,23 @@ mod tests {
         assert_ne!(state_a, state_b);
     }
 
+    #[test]
+    fn test_try_fill_bytes_is_deterministic_and_audited() {
+        fn run(seed: u64) -> ([u8; 32], String) {
+            deterministic::Runner::seeded(seed).start(|mut context| async move {
+                let before = context.auditor().state();
+                let mut bytes = [0; 32];
+                context.try_fill_bytes(&mut bytes).unwrap();
+                let after = context.auditor().state();
+                assert_ne!(after, before);
+                (bytes, after)
+            })
+        }
+
+        assert_eq!(run(7), run(7));
+        assert_ne!(run(7), run(8));
+    }
+
     #[test]
     fn test_same_seed_same_order() {
         // Generate initial outputs
@@ -1914,6 +1962,46 @@ mod tests {
         });
     }
 
+    #[test]
+    fn test_recover_escaped_atomic_blob_cannot_publish_crash_lost_bytes() {
+        let (stale, checkpoint) =
+            deterministic::Runner::default().start_and_recover(|context| async move {
+                let (blob, _) = context.open_atomic("stale", b"blob").await.unwrap();
+                blob.append(b"lost").await.unwrap();
+                blob
+            });
+
+        deterministic::Runner::from(checkpoint).start(|context| async move {
+            assert!(stale.sync().await.is_err());
+            drop(stale);
+
+            let (_, len) = context.open_atomic("stale", b"blob").await.unwrap();
+            assert_eq!(len, 0);
+        });
+    }
+
+    #[test]
+    fn test_recover_rejects_escaped_atomic_batch_participant() {
+        let (stale, checkpoint) =
+            deterministic::Runner::default().start_and_recover(|context| async move {
+                let (blob, _) = context.open_atomic("stale_batch", b"blob").await.unwrap();
+                blob.append(b"lost").await.unwrap();
+                blob
+            });
+
+        deterministic::Runner::from(checkpoint).start(|context| async move {
+            assert!(
+                context
+                    .apply(vec![BatchOperation::Publish(stale)])
+                    .await
+                    .is_err()
+            );
+
+            let (_, len) = context.open_atomic("stale_batch", b"blob").await.unwrap();
+            assert_eq!(len, 0);
+        });
+    }
+
     #[test]
     fn test_recover_snapshots_fault_configuration() {
         let (stale_config, checkpoint) =
diff --git a/runtime/src/storage/audited.rs b/runtime/src/storage/audited.rs
index 43a273a6f8..8d23063d4b 100644
--- a/runtime/src/storage/audited.rs
+++ b/runtime/src/storage/audited.rs
@@ -72,6 +72,52 @@ impl<S: crate::Storage> crate::Storage for Storage<S> {
     }
 }
 
+impl<S: crate::atomic::Backend> crate::atomic::Backend for Storage<S> {
+    type Worker = Storage<S::Worker>;
+
+    fn atomic_worker(&self) -> Self::Worker {
+        Storage {
+            inner: self.inner.atomic_worker(),
+            auditor: self.auditor.clone(),
+        }
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.inner.atomic_resources()
+    }
+
+    fn new_atomic_identifier(&self) -> [u8; 16] {
+        self.inner.new_atomic_identifier()
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        self.auditor.event(b"open_atomic_existing", |hasher| {
+            hasher.update(partition.as_bytes());
+            hasher.update(name);
+        });
+        self.inner
+            .open_atomic_existing(partition, name)
+            .await
+            .map(|opened| {
+                opened.map(|(blob, len)| {
+                    (
+                        Blob {
+                            auditor: self.auditor.clone(),
+                            inner: blob,
+                            partition: partition.into(),
+                            name: name.to_vec(),
+                        },
+                        len,
+                    )
+                })
+            })
+    }
+}
+
 #[derive(Clone)]
 pub struct Blob<B: crate::Blob> {
     auditor: Arc<Auditor>,
@@ -156,8 +202,9 @@ mod tests {
         Storage as _, WriteOptions,
         deterministic::Auditor,
         storage::{
-            audited::Storage as AuditedStorage, memory::Storage as MemStorage,
-            tests::run_storage_tests,
+            audited::Storage as AuditedStorage,
+            memory::Storage as MemStorage,
+            tests::{RecordingAtomicBackend, run_storage_tests},
         },
         telemetry::metrics::Registry,
     };
@@ -178,6 +225,35 @@ mod tests {
         run_storage_tests(storage).await;
     }
 
+    #[tokio::test]
+    async fn test_audited_atomic_backend_forwards_worker_and_identifier() {
+        let auditor = Arc::new(Auditor::default());
+        let inner = RecordingAtomicBackend::new(MemStorage::new(test_pool()));
+        let storage = AuditedStorage::new(inner, auditor);
+
+        let worker = crate::atomic::Backend::atomic_worker(&storage);
+        assert!(Arc::ptr_eq(&storage.auditor, &worker.auditor));
+        let resources = crate::atomic::Backend::atomic_resources(&storage);
+        let worker_resources = crate::atomic::Backend::atomic_resources(&worker);
+        assert!(Arc::ptr_eq(
+            &resources.exclusion,
+            &worker_resources.exclusion
+        ));
+        assert!(Arc::ptr_eq(
+            &resources.payload_budget,
+            &worker_resources.payload_budget
+        ));
+        let mut expected_identifier = [2; 16];
+        expected_identifier[8..].fill(0);
+        assert_eq!(
+            crate::atomic::Backend::new_atomic_identifier(&worker),
+            expected_identifier
+        );
+
+        assert_eq!(storage.inner().worker_calls(), 1);
+        assert_eq!(storage.inner().identifier_tags(), vec![2]);
+    }
+
     #[tokio::test]
     async fn test_audited_storage_separates_partition_and_blob_names() {
         let auditor1 = Arc::new(Auditor::default());
diff --git a/runtime/src/storage/faulty.rs b/runtime/src/storage/faulty.rs
index d3becd0383..e99d6dbdaf 100644
--- a/runtime/src/storage/faulty.rs
+++ b/runtime/src/storage/faulty.rs
@@ -571,6 +571,39 @@ impl<S: crate::Storage> crate::Storage for Storage<S> {
     }
 }
 
+impl<S: crate::atomic::Backend> crate::atomic::Backend for Storage<S> {
+    type Worker = Storage<S::Worker>;
+
+    fn atomic_worker(&self) -> Self::Worker {
+        Storage {
+            inner: self.inner.atomic_worker(),
+            ctx: self.ctx.clone(),
+            pending: self.pending.clone(),
+            generations: self.generations.clone(),
+        }
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.inner.atomic_resources()
+    }
+
+    fn new_atomic_identifier(&self) -> [u8; 16] {
+        self.inner.new_atomic_identifier()
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        if self.ctx.should_fail(Op::Open) {
+            return Err(injected_io_error().into());
+        }
+        let opened = self.inner.open_atomic_existing(partition, name).await?;
+        Ok(opened.map(|(blob, len)| (self.wrap_blob(partition, name, blob, len), len)))
+    }
+}
+
 /// A blob wrapper that injects deterministic faults based on configuration.
 #[derive(Clone)]
 pub struct Blob<B: crate::Blob> {
diff --git a/runtime/src/storage/iouring.rs b/runtime/src/storage/iouring.rs
index 64a05b1369..453f651c4a 100644
--- a/runtime/src/storage/iouring.rs
+++ b/runtime/src/storage/iouring.rs
@@ -28,10 +28,10 @@ use crate::{
     utils,
 };
 use commonware_formatting::{from_hex, hex};
-use commonware_utils::sync::Mutex;
+use commonware_utils::sync::{AsyncMutex, AsyncRwLock};
 use std::{
     fs::{self, File},
-    io::{Error as IoError, Read, Seek, SeekFrom, Write},
+    io::{Error as IoError, ErrorKind, Read, Seek, SeekFrom, Write},
     ops::RangeInclusive,
     path::{Path, PathBuf},
     sync::{Arc, atomic::AtomicBool},
@@ -71,10 +71,16 @@ fn sync_dir(path: &Path) -> Result<(), Error> {
     })
 }
 
-/// Configuration for a [Storage].
+/// Configuration for a [`Storage`].
+///
+/// `storage_directory` must be owned by exactly one independently constructed [`Storage`] lineage
+/// for its full lifetime. Clones belong to that lineage. Another process, storage instance, path
+/// alias, or direct filesystem mutation must not access the directory concurrently.
+/// Partition names and hex-encoded blob names each occupy one filesystem path component and must
+/// fit the filesystem's component-length limit.
 #[derive(Clone, Debug)]
 pub struct Config {
-    /// Where to store blobs.
+    /// Directory exclusively owned by the resulting storage lineage.
     pub storage_directory: PathBuf,
     /// Configuration for the iouring instance.
     pub iouring_config: iouring::Config,
@@ -84,7 +90,7 @@ pub struct Config {
 
 #[derive(Clone)]
 pub struct Storage {
-    lock: Arc<Mutex<()>>,
+    atomic_resources: crate::atomic::AtomicResources,
     storage_directory: PathBuf,
     io_handle: iouring::Handle,
     pool: BufferPool,
@@ -108,7 +114,12 @@ impl Storage {
         let (io_handle, iouring_loop) = iouring::IoUringLoop::new(iouring_config, registry);
 
         let storage = Self {
-            lock: Arc::new(Mutex::new(())),
+            atomic_resources: crate::atomic::AtomicResources {
+                driver: crate::atomic::Driver::background(),
+                exclusion: Arc::new(AsyncRwLock::new(())),
+                namespace: Arc::new(AsyncMutex::new(())),
+                payload_budget: Arc::new(crate::atomic::PayloadBudget::default()),
+            },
             storage_directory,
             io_handle,
             pool,
@@ -131,7 +142,7 @@ impl crate::Storage for Storage {
         super::validate_partition_name(partition)?;
 
         // Acquire the filesystem lock
-        let _guard = self.lock.lock();
+        let _guard = self.atomic_resources.namespace.lock().await;
 
         // Construct the full path
         let path = self.storage_directory.join(partition).join(hex(name));
@@ -197,13 +208,18 @@ impl crate::Storage for Storage {
         super::validate_partition_name(partition)?;
 
         // Acquire the filesystem lock
-        let _guard = self.lock.lock();
+        let _guard = self.atomic_resources.namespace.lock().await;
 
         let path = self.storage_directory.join(partition);
         if let Some(name) = name {
             let blob_path = path.join(hex(name));
-            fs::remove_file(blob_path)
-                .map_err(|_| Error::BlobMissing(partition.into(), hex(name)))?;
+            match fs::remove_file(blob_path) {
+                Ok(()) => {}
+                Err(error) if error.kind() == ErrorKind::NotFound => {
+                    return Err(Error::BlobMissing(partition.into(), hex(name)));
+                }
+                Err(error) => return Err(Error::Io(error.into())),
+            }
 
             // Sync the partition directory to ensure the removal is durable.
             sync_dir(&path)?;
@@ -220,7 +236,7 @@ impl crate::Storage for Storage {
         super::validate_partition_name(partition)?;
 
         // Acquire the filesystem lock
-        let _guard = self.lock.lock();
+        let _guard = self.atomic_resources.namespace.lock().await;
 
         let path = self.storage_directory.join(partition);
 
@@ -236,7 +252,8 @@ impl crate::Storage for Storage {
                 return Err(Error::PartitionCorrupt(partition.into()));
             }
 
-            if let Some(name) = entry.file_name().to_str() {
+            let file_name = entry.file_name();
+            if let Some(name) = file_name.to_str() {
                 // Reject anything that isn't canonical lowercase hex (no `0x`
                 // prefix, no whitespace) since `from_hex` is lenient and
                 // storage only ever writes the canonical form via `hex()`.
@@ -248,11 +265,63 @@ impl crate::Storage for Storage {
                 blobs.push(decoded);
             }
         }
-
         Ok(blobs)
     }
 }
 
+impl crate::atomic::Backend for Storage {
+    type Worker = Self;
+
+    fn atomic_worker(&self) -> Self {
+        self.clone()
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.atomic_resources.clone()
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        super::validate_partition_name(partition)?;
+        let _guard = self.atomic_resources.namespace.lock().await;
+        let path = self.storage_directory.join(partition).join(hex(name));
+        let mut file = match fs::OpenOptions::new().read(true).write(true).open(&path) {
+            Ok(file) => file,
+            Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
+            Err(error) => {
+                return Err(Error::BlobOpenFailed(
+                    partition.into(),
+                    hex(name),
+                    error.into(),
+                ));
+            }
+        };
+
+        let raw_len = file.metadata().map_err(|_| Error::ReadFailed)?.len();
+        let versions = crate::DEFAULT_BLOB_VERSION..=crate::DEFAULT_BLOB_VERSION;
+        let Some((logical_len, _, data_offset)) =
+            resolve_header(&mut file, raw_len, &versions, partition, name)?
+        else {
+            return Ok(None);
+        };
+        super::header::require_atomic_layout(data_offset, partition, name)?;
+        Ok(Some((
+            Blob::new(
+                partition.into(),
+                name,
+                file,
+                self.io_handle.clone(),
+                self.pool.clone(),
+                data_offset,
+            ),
+            logical_len,
+        )))
+    }
+}
+
 pub struct Blob {
     /// The partition this blob lives in
     partition: String,
@@ -424,7 +493,8 @@ impl crate::Blob for Blob {
 mod tests {
     use super::{Header, *};
     use crate::{
-        Blob as _, BufferPool, BufferPoolConfig, IoBuf, IoBufMut, Storage as _,
+        AtomicBlob as _, AtomicStorage as _, BatchOperation, Blob as _, BufferPool,
+        BufferPoolConfig, IoBuf, IoBufMut, Storage as _,
         storage::{Layout, tests::run_storage_tests},
         telemetry::metrics::Registry,
         utils::thread,
@@ -480,10 +550,6 @@ mod tests {
         storage_directory
     }
 
-    /// Verify the end-to-end storage-page alignment invariant on the io_uring backend: paged
-    /// data written to a V1 blob with a 4096-byte physical page size occupies exactly one
-    /// aligned 4096-byte disk page per physical page (header page included), so page reads
-    /// never straddle a page boundary.
     #[tokio::test]
     async fn test_v1_paged_alignment() {
         let (storage, storage_directory) = create_test_storage();
@@ -543,6 +609,129 @@ mod tests {
         let _ = std::fs::remove_dir_all(storage_directory);
     }
 
+    #[tokio::test]
+    async fn test_atomic_exact_open_leaves_container_recovery_to_ordinary_open() {
+        let (storage, storage_directory) = create_test_storage();
+        let partition_path = storage_directory.join("partition");
+        std::fs::create_dir_all(&partition_path).unwrap();
+        let path = partition_path.join(hex(b"blob"));
+        let interrupted =
+            Header::create(&(crate::DEFAULT_BLOB_VERSION..=crate::DEFAULT_BLOB_VERSION)).0
+                [..Header::PRELUDE_SIZE]
+                .to_vec();
+        std::fs::write(&path, &interrupted).unwrap();
+
+        assert!(
+            crate::atomic::Backend::open_atomic_existing(&storage, "partition", b"blob")
+                .await
+                .unwrap()
+                .is_none()
+        );
+        assert_eq!(std::fs::read(&path).unwrap(), interrupted);
+
+        let (_, len) = storage.open_atomic("partition", b"blob").await.unwrap();
+        assert_eq!(len, 0);
+
+        let _ = std::fs::remove_dir_all(storage_directory);
+    }
+
+    #[tokio::test]
+    async fn test_atomic_rejects_v0_ordinary_lookalikes_without_mutation() {
+        let storage_directory = create_test_directory();
+        let original = crate::storage::header::tests::v0_atomic_lookalike_bytes();
+        for (partition, name) in [("v0_atomic_open", b"open"), ("v0_atomic_scan", b"scan")] {
+            let partition_path = storage_directory.join(partition);
+            std::fs::create_dir_all(&partition_path).unwrap();
+            std::fs::write(partition_path.join(hex(name)), &original).unwrap();
+        }
+        let mut registry = Registry::default();
+        let storage = Storage::start(
+            Config {
+                storage_directory: storage_directory.clone(),
+                iouring_config: Default::default(),
+                thread_stack_size: thread::system_thread_stack_size(),
+            },
+            &mut registry.sub_registry("storage"),
+            test_pool(&mut registry.sub_registry("pool")),
+        );
+
+        let rejected_open = storage.open_atomic("v0_atomic_open", b"open").await;
+        let rejected_scan = storage.scan_atomic("v0_atomic_scan").await;
+        assert!(
+            std::fs::read(storage_directory.join("v0_atomic_open").join(hex(b"open"))).unwrap()
+                == original,
+            "atomic open mutated a legacy ordinary blob"
+        );
+        assert!(
+            std::fs::read(storage_directory.join("v0_atomic_scan").join(hex(b"scan"))).unwrap()
+                == original,
+            "atomic scan mutated a legacy ordinary blob"
+        );
+        assert!(matches!(
+            rejected_open,
+            Err(crate::Error::BlobCorrupt(_, _, reason)) if reason == "expected V1 header layout"
+        ));
+        assert!(matches!(
+            rejected_scan,
+            Err(crate::Error::BlobCorrupt(_, _, reason)) if reason == "expected V1 header layout"
+        ));
+
+        drop(storage);
+        let _ = std::fs::remove_dir_all(storage_directory);
+    }
+
+    #[tokio::test]
+    async fn test_atomic_mixed_batch_lifecycle() {
+        let (storage, storage_directory) = create_test_storage();
+        let partition = "atomic_batch";
+        let (a, _) = storage.open_atomic(partition, b"a").await.unwrap();
+        let (b, _) = storage.open_atomic(partition, b"b").await.unwrap();
+        let (c, _) = storage.open_atomic(partition, b"c").await.unwrap();
+        a.append(b"a").await.unwrap();
+        b.append(b"bb").await.unwrap();
+        c.append(b"c").await.unwrap();
+        storage
+            .apply(vec![
+                BatchOperation::Publish(a),
+                BatchOperation::Publish(b),
+                BatchOperation::Publish(c),
+            ])
+            .await
+            .unwrap();
+
+        let (a, _) = storage.open_atomic(partition, b"a").await.unwrap();
+        let (b, _) = storage.open_atomic(partition, b"b").await.unwrap();
+        let (c, _) = storage.open_atomic(partition, b"c").await.unwrap();
+        a.append(b"-next").await.unwrap();
+        storage
+            .apply(vec![
+                BatchOperation::Remove(c),
+                BatchOperation::Publish(a),
+                BatchOperation::Rewind { blob: b, len: 1 },
+            ])
+            .await
+            .unwrap();
+
+        assert_eq!(
+            storage.scan_atomic(partition).await.unwrap(),
+            vec![b"a".to_vec(), b"b".to_vec()]
+        );
+        let (a, len) = storage.open_atomic(partition, b"a").await.unwrap();
+        assert_eq!(len, 6);
+        assert_eq!(a.read_at(0, 6).await.unwrap().coalesce(), b"a-next");
+        drop(a);
+        let (b, len) = storage.open_atomic(partition, b"b").await.unwrap();
+        assert_eq!(len, 1);
+        assert_eq!(b.read_at(0, 1).await.unwrap().coalesce(), b"b");
+        drop(b);
+        assert_eq!(
+            storage.scan(partition).await.unwrap(),
+            vec![b"a".to_vec(), b"b".to_vec()]
+        );
+
+        let _ = std::fs::remove_dir_all(storage_directory);
+    }
+
     #[tokio::test]
     async fn test_blob_header_handling() {
         // Verify header creation, logical offsets, resize, reopen, and corruption recovery.
@@ -843,6 +1032,15 @@ mod tests {
             .unwrap_err();
         assert_eq!(err.to_string(), "blob missing: partition/6d697373696e67");
 
+        let blob_path = storage_directory.join("partition").join(hex(b"directory"));
+        std::fs::create_dir(&blob_path).unwrap();
+        let err = storage
+            .remove("partition", Some(b"directory"))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::Io(_)));
+        assert!(blob_path.is_dir());
+
         let _ = std::fs::remove_dir_all(&storage_directory);
     }
 
diff --git a/runtime/src/storage/metered.rs b/runtime/src/storage/metered.rs
index 9e67573158..b5c4ddd1d7 100644
--- a/runtime/src/storage/metered.rs
+++ b/runtime/src/storage/metered.rs
@@ -116,6 +116,47 @@ impl<S: crate::Storage> crate::Storage for Storage<S> {
     }
 }
 
+impl<S: crate::atomic::Backend> crate::atomic::Backend for Storage<S> {
+    type Worker = Storage<S::Worker>;
+
+    fn atomic_worker(&self) -> Self::Worker {
+        Storage {
+            inner: self.inner.atomic_worker(),
+            metrics: self.metrics.clone(),
+        }
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.inner.atomic_resources()
+    }
+
+    fn new_atomic_identifier(&self) -> [u8; 16] {
+        self.inner.new_atomic_identifier()
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        self.inner
+            .open_atomic_existing(partition, name)
+            .await
+            .map(|opened| {
+                opened.map(|(inner, len)| {
+                    (
+                        Blob {
+                            inner,
+                            partition: partition.into(),
+                            metrics: Arc::new(MetricsHandle::new(self.metrics.clone())),
+                        },
+                        len,
+                    )
+                })
+            })
+    }
+}
+
 /// A wrapper around a `Blob` implementation that tracks metrics
 #[derive(Clone)]
 pub struct Blob<B> {
@@ -241,7 +282,10 @@ mod tests {
     use super::*;
     use crate::{
         Blob, BufferPool, BufferPoolConfig, Storage as _,
-        storage::{memory::Storage as MemoryStorage, tests::run_storage_tests},
+        storage::{
+            memory::Storage as MemoryStorage,
+            tests::{RecordingAtomicBackend, run_storage_tests},
+        },
         telemetry::metrics::Registry,
     };
 
@@ -258,6 +302,37 @@ mod tests {
         run_storage_tests(storage).await;
     }
 
+    #[tokio::test]
+    async fn test_metered_atomic_backend_forwards_worker_and_identifier() {
+        let mut registry = Registry::default();
+        let inner = RecordingAtomicBackend::new(MemoryStorage::new(test_pool(
+            &mut registry.sub_registry("pool"),
+        )));
+        let storage = Storage::new(inner, &mut registry.sub_registry("storage"));
+
+        let worker = crate::atomic::Backend::atomic_worker(&storage);
+        assert!(Arc::ptr_eq(&storage.metrics, &worker.metrics));
+        let resources = crate::atomic::Backend::atomic_resources(&storage);
+        let worker_resources = crate::atomic::Backend::atomic_resources(&worker);
+        assert!(Arc::ptr_eq(
+            &resources.exclusion,
+            &worker_resources.exclusion
+        ));
+        assert!(Arc::ptr_eq(
+            &resources.payload_budget,
+            &worker_resources.payload_budget
+        ));
+        let mut expected_identifier = [2; 16];
+        expected_identifier[8..].fill(0);
+        assert_eq!(
+            crate::atomic::Backend::new_atomic_identifier(&worker),
+            expected_identifier
+        );
+
+        assert_eq!(storage.inner().worker_calls(), 1);
+        assert_eq!(storage.inner().identifier_tags(), vec![2]);
+    }
+
     /// Test that a failed open does not count an open blob.
     #[tokio::test]
     async fn test_failed_open_does_not_count_open_blob() {
diff --git a/runtime/src/storage/mod.rs b/runtime/src/storage/mod.rs
index 0c1b4e2e26..f89535344b 100644
--- a/runtime/src/storage/mod.rs
+++ b/runtime/src/storage/mod.rs
@@ -80,8 +80,103 @@ stability_scope!(BETA {
 
 #[cfg(test)]
 pub(crate) mod tests {
-    use crate::{Blob, Buf, IoBuf, IoBufMut, IoBufs, IoBufsMut, Storage, WriteOptions};
+    use crate::{Blob, Buf, Error, IoBuf, IoBufMut, IoBufs, IoBufsMut, Storage, WriteOptions};
+    use commonware_utils::sync::Mutex;
     use futures::FutureExt;
+    use std::{
+        ops::RangeInclusive,
+        sync::{
+            Arc,
+            atomic::{AtomicUsize, Ordering},
+        },
+    };
+
+    #[derive(Default)]
+    struct AtomicBackendCalls {
+        workers: AtomicUsize,
+        identifiers: AtomicUsize,
+        identifier_tags: Mutex<Vec<u8>>,
+    }
+
+    /// Backend with an observable worker identity for wrapper contract tests.
+    #[derive(Clone)]
+    pub(crate) struct RecordingAtomicBackend {
+        inner: super::memory::Storage,
+        calls: Arc<AtomicBackendCalls>,
+        tag: u8,
+    }
+
+    impl RecordingAtomicBackend {
+        pub(crate) fn new(inner: super::memory::Storage) -> Self {
+            Self {
+                inner,
+                calls: Arc::new(AtomicBackendCalls::default()),
+                tag: 1,
+            }
+        }
+
+        pub(crate) fn worker_calls(&self) -> usize {
+            self.calls.workers.load(Ordering::Relaxed)
+        }
+
+        pub(crate) fn identifier_tags(&self) -> Vec<u8> {
+            self.calls.identifier_tags.lock().clone()
+        }
+    }
+
+    impl Storage for RecordingAtomicBackend {
+        type Blob = <super::memory::Storage as Storage>::Blob;
+
+        async fn open_versioned(
+            &self,
+            partition: &str,
+            name: &[u8],
+            versions: RangeInclusive<u16>,
+        ) -> Result<(Self::Blob, u64, u16), Error> {
+            self.inner.open_versioned(partition, name, versions).await
+        }
+
+        async fn remove(&self, partition: &str, name: Option<&[u8]>) -> Result<(), Error> {
+            self.inner.remove(partition, name).await
+        }
+
+        async fn scan(&self, partition: &str) -> Result<Vec<Vec<u8>>, Error> {
+            self.inner.scan(partition).await
+        }
+    }
+
+    impl crate::atomic::Backend for RecordingAtomicBackend {
+        type Worker = Self;
+
+        fn atomic_worker(&self) -> Self {
+            self.calls.workers.fetch_add(1, Ordering::Relaxed);
+            Self {
+                inner: crate::atomic::Backend::atomic_worker(&self.inner),
+                calls: self.calls.clone(),
+                tag: self.tag + 1,
+            }
+        }
+
+        fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+            crate::atomic::Backend::atomic_resources(&self.inner)
+        }
+
+        fn new_atomic_identifier(&self) -> [u8; 16] {
+            self.calls.identifier_tags.lock().push(self.tag);
+            let ordinal = self.calls.identifiers.fetch_add(1, Ordering::Relaxed) as u64;
+            let mut identifier = [self.tag; 16];
+            identifier[8..].copy_from_slice(&ordinal.to_be_bytes());
+            identifier
+        }
+
+        async fn open_atomic_existing(
+            &self,
+            partition: &str,
+            name: &[u8],
+        ) -> Result<Option<(Self::Blob, u64)>, Error> {
+            crate::atomic::Backend::open_atomic_existing(&self.inner, partition, name).await
+        }
+    }
 
     /// Runs the full suite of tests on the provided storage implementation.
     pub(crate) async fn run_storage_tests<S>(storage: S)
diff --git a/runtime/src/tokio/runtime.rs b/runtime/src/tokio/runtime.rs
index 6033d51383..d424b29746 100644
--- a/runtime/src/tokio/runtime.rs
+++ b/runtime/src/tokio/runtime.rs
@@ -912,6 +912,26 @@ impl crate::Storage for Context {
     }
 }
 
+impl crate::atomic::Backend for Context {
+    type Worker = Storage;
+
+    fn atomic_worker(&self) -> Self::Worker {
+        self.storage.clone()
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        crate::atomic::Backend::atomic_resources(&self.storage)
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        crate::atomic::Backend::open_atomic_existing(&self.storage, partition, name).await
+    }
+}
+
 impl crate::BufferPooler for Context {
     fn network_buffer_pool(&self) -> &BufferPool {
         &self.network_buffer_pool
@@ -926,8 +946,8 @@ impl crate::BufferPooler for Context {
 mod tests {
     use super::*;
     use crate::{
-        Metrics, Network, Resolver, Runner as _, Sink, Stream, telemetry::metrics::raw::Counter,
-        tokio::telemetry,
+        AtomicBlob as _, AtomicStorage as _, BatchOperation, Metrics, Network, Resolver,
+        Runner as _, Sink, Stream, telemetry::metrics::raw::Counter, tokio::telemetry,
     };
     use bytes::Bytes;
     use std::{
@@ -1138,6 +1158,40 @@ mod tests {
         );
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn test_atomic_open_and_remove_run_through_tokio_context() {
+        let cfg = Config::new();
+        let storage_directory = cfg.storage_directory().clone();
+        Runner::new(cfg).start(|context| async move {
+            let (blob, len) = context
+                .open_atomic("atomic_context", b"blob")
+                .await
+                .unwrap();
+            assert_eq!(len, 0);
+            blob.append(b"value").await.unwrap();
+            blob.sync().await.unwrap();
+            context
+                .apply(vec![BatchOperation::Remove(blob)])
+                .await
+                .unwrap();
+            assert!(
+                context
+                    .scan_atomic("atomic_context")
+                    .await
+                    .unwrap()
+                    .is_empty()
+            );
+
+            let (_, len) = context
+                .open_atomic("atomic_context", b"blob")
+                .await
+                .unwrap();
+            assert_eq!(len, 0);
+        });
+        let _ = std::fs::remove_dir_all(storage_directory);
+    }
+
     #[test]
     fn test_thread_stack_size_override() {
         let cfg = Config::new().with_thread_stack_size(4 * 1024 * 1024);