diff --git a/Sources/tart/Commands/Clone.swift b/Sources/tart/Commands/Clone.swift
index 62cc75c..cf600c9 100644
--- a/Sources/tart/Commands/Clone.swift
+++ b/Sources/tart/Commands/Clone.swift
@@ -56,15 +56,28 @@ struct Clone: AsyncParsableCommand {
       guard remoteName != nil else {
         throw ValidationError("--stacked requires a remote image")
       }
+      try DiskImageStack.requireSupport()
     }
 
     if let remoteName, try !ociStorage.hasUsableCachedImageForClone(remoteName, requireManifest: stacked) {
       // Pull the VM in case it's OCI-based and doesn't exist locally yet
       let registry = try Registry(host: remoteName.host, namespace: remoteName.namespace, insecure: insecure)
+
+      // Fail before pulling disk content when this host cannot create a writable stacked disk.
+      if !stacked {
+        let (manifest, _) = try await registry.pullManifest(reference: remoteName.reference.value)
+        if manifest.layers.contains(where: { $0.mediaType == asifOverlayMediaType }) {
+          try DiskImageStack.requireSupport()
+        }
+      }
+
       try await ociStorage.pull(remoteName, registry: registry, concurrency: concurrency, deduplicate: deduplicate)
     }
 
     let sourceVM = try VMStorageHelper.open(sourceName)
+    if sourceVM.isStackedVM || sourceVM.isStackedCachedImage {
+      try DiskImageStack.requireSupport()
+    }
     let tmpVMDir = try VMDirectory.temporary()
 
     // Lock the temporary VM directory to prevent it's garbage collection
@@ -126,7 +139,7 @@ struct Clone: AsyncParsableCommand {
         }
       }
     }, onCancel: {
-      try? FileManager.default.removeItem(at: tmpVMDir.baseURL)
+      try? tmpVMDir.removeFromDisk()
     })
   }
 }
diff --git a/Sources/tart/Commands/Import.swift b/Sources/tart/Commands/Import.swift
index 7fc955c..a3779bd 100644
--- a/Sources/tart/Commands/Import.swift
+++ b/Sources/tart/Commands/Import.swift
@@ -21,6 +21,9 @@ struct Import: AsyncParsableCommand {
 
     // Create a temporary VM directory to which we will load the export file
     let tmpVMDir = try VMDirectory.temporary()
+    defer {
+      try? tmpVMDir.removeFromDisk()
+    }
 
     // Lock the temporary VM directory to prevent it's garbage collection
     // while we're running
@@ -30,10 +33,8 @@ struct Import: AsyncParsableCommand {
     // Populate the temporary VM directory with the export file contents
     print("importing...")
     try tmpVMDir.importFromArchive(path: path)
-
-    if tmpVMDir.isStackedVM || tmpVMDir.isStackedCachedImage {
-      try? FileManager.default.removeItem(at: tmpVMDir.baseURL)
-      throw RuntimeError.ImportFailed("importing stacked VMs is not supported yet")
+    guard tmpVMDir.initialized else {
+      throw RuntimeError.ImportFailed("archive does not contain a runnable VM")
     }
 
     try await withTaskCancellationHandler(operation: {
@@ -50,7 +51,7 @@ struct Import: AsyncParsableCommand {
 
       try lock.unlock()
     }, onCancel: {
-      try? FileManager.default.removeItem(at: tmpVMDir.baseURL)
+      try? tmpVMDir.removeFromDisk()
     })
   }
 }
diff --git a/Sources/tart/Commands/Prune.swift b/Sources/tart/Commands/Prune.swift
index 4ff12ff..baa490c 100644
--- a/Sources/tart/Commands/Prune.swift
+++ b/Sources/tart/Commands/Prune.swift
@@ -81,27 +81,34 @@ struct Prune: AsyncParsableCommand {
   }
 
   static func pruneSpaceBudget(prunableStorages: [PrunableStorage], spaceBudgetBytes: UInt64) throws {
-    let prunables: [Prunable] = try prunableStorages
-      .flatMap { try $0.prunables() }
-      .sorted { try $0.accessDate() > $1.accessDate() }
+    while true {
+      let prunables: [Prunable] = try prunableStorages
+        .flatMap { try $0.prunables() }
+        .sorted { try $0.accessDate() > $1.accessDate() }
 
-    var spaceBudgetBytes = spaceBudgetBytes
-    var prunablesToDelete: [Prunable] = []
+      var remainingBudgetBytes = spaceBudgetBytes
+      var prunableToDelete: Prunable?
 
-    for prunable in prunables {
-      let prunableSizeBytes = UInt64(try prunable.allocatedSizeBytes())
+      for prunable in prunables {
+        let prunableSizeBytes = UInt64(try prunable.allocatedSizeBytes())
 
-      if prunableSizeBytes <= spaceBudgetBytes {
-        // Don't mark for deletion as
-        // there's a budget available
-        spaceBudgetBytes -= prunableSizeBytes
-      } else {
-        // Mark for deletion
-        prunablesToDelete.append(prunable)
+        if prunableSizeBytes <= remainingBudgetBytes {
+          // Don't mark for deletion as there is budget available
+          remainingBudgetBytes -= prunableSizeBytes
+        } else {
+          prunableToDelete = prunable
+          break
+        }
       }
-    }
 
-    try prunablesToDelete.forEach { try $0.delete() }
+      guard let prunableToDelete else {
+        return
+      }
+
+      // Deleting one cached stacked image can change which remaining image
+      // owns shared immutable content. Rebuild before choosing another.
+      try prunableToDelete.delete()
+    }
   }
 
   static func reclaimIfNeeded(_ requiredBytes: UInt64, _ initiator: Prunable? = nil) throws {
@@ -145,46 +152,51 @@ struct Prune: AsyncParsableCommand {
     try Prune.reclaimIfPossible(requiredBytes - volumeAvailableCapacityCalculated, initiator)
   }
 
-  private static func reclaimIfPossible(_ reclaimBytes: UInt64, _ initiator: Prunable? = nil) throws {
+  static func reclaimIfPossible(_ reclaimBytes: UInt64, _ initiator: Prunable? = nil) throws {
     let span = OTel.shared.tracer.spanBuilder(spanName: "prune").startSpan()
     defer { span.end() }
 
     let prunableStorages: [PrunableStorage] = [try VMStorageOCI(), try IPSWCache()]
-    let prunables: [Prunable] = try prunableStorages
-      .flatMap { try $0.prunables() }
-      .sorted { try $0.accessDate() < $1.accessDate() }
+    let prunables = {
+      try prunableStorages
+        .flatMap { try $0.prunables() }
+        .sorted { try $0.accessDate() < $1.accessDate() }
+    }
 
     // Does it even make sense to start?
-    let cacheUsedBytes = try prunables.map { try $0.allocatedSizeBytes() }.reduce(0, +)
-    if cacheUsedBytes < reclaimBytes {
+    let initialPrunables = try prunables()
+    let initialCacheUsedBytes = try initialPrunables.map { try $0.allocatedSizeBytes() }.reduce(0, +)
+    guard let reclaimBytes = Int(exactly: reclaimBytes), initialCacheUsedBytes >= reclaimBytes else {
       return
     }
 
-    var cacheReclaimedBytes: Int = 0
+    let targetCacheUsedBytes = initialCacheUsedBytes - reclaimBytes
+    var currentCacheUsedBytes = initialCacheUsedBytes
+    let initiatorPath = initiator.map {
+      $0.url.resolvingSymlinksInPath().standardizedFileURL.path
+    }
 
-    var it = prunables.makeIterator()
-
-    while cacheReclaimedBytes <= reclaimBytes {
-      guard let prunable = it.next() else {
+    while currentCacheUsedBytes > targetCacheUsedBytes {
+      // Deleting one cached stacked image can transfer ownership of shared
+      // immutable content to another record without reclaiming those bytes.
+      // Rebuild the candidates after every deletion so automatic pruning
+      // measures the cache that remains rather than a stale ownership snapshot.
+      guard let prunable = try prunables().first(where: {
+        $0.url.resolvingSymlinksInPath().standardizedFileURL.path != initiatorPath
+      }) else {
         break
       }
 
-      if prunable.url == initiator?.url.resolvingSymlinksInPath() {
-        // do not prune the initiator
-        continue
-      }
-
       let allocatedSizeBytes = try prunable.allocatedSizeBytes()
 
       OpenTelemetry.instance.contextProvider.activeSpan?
         .addEvent(name: "Pruned \(allocatedSizeBytes) bytes for \(prunable.url.path)")
 
-      cacheReclaimedBytes += allocatedSizeBytes
-
       try prunable.delete()
+      currentCacheUsedBytes = try prunables().map { try $0.allocatedSizeBytes() }.reduce(0, +)
     }
 
     OpenTelemetry.instance.contextProvider.activeSpan?
-      .addEvent(name: "Reclaimed \(cacheReclaimedBytes) bytes")
+      .addEvent(name: "Reclaimed \(initialCacheUsedBytes - currentCacheUsedBytes) bytes")
   }
 }
diff --git a/Sources/tart/Commands/Run.swift b/Sources/tart/Commands/Run.swift
index e39a716..dbd946c 100644
--- a/Sources/tart/Commands/Run.swift
+++ b/Sources/tart/Commands/Run.swift
@@ -1039,20 +1039,24 @@ struct AdditionalDisk {
         // A cached stacked image has no writable top overlay. Create one in a
         // disposable directory for this additional-disk attachment.
         let temporaryVMDir = try VMDirectory.temporary()
-        try FileManager.default.copyItem(at: vmDir.configURL, to: temporaryVMDir.configURL)
-        try FileManager.default.copyItem(at: vmDir.nvramURL, to: temporaryVMDir.nvramURL)
-        try FileManager.default.copyItem(at: vmDir.manifestURL, to: temporaryVMDir.manifestURL)
-        let lock = try FileLock(lockURL: temporaryVMDir.baseURL)
-        try lock.lock()
+        let temporaryVMDirLock = try FileLock(lockURL: temporaryVMDir.baseURL)
+        try temporaryVMDirLock.lock()
+        try vmDir.cloneStacked(
+          to: temporaryVMDir,
+          copyWritableOverlay: false,
+          generateMAC: false
+        )
         let stack = try temporaryVMDir.diskImageStack()
-        try stack.createWritableOverlay()
         let attachment = try stack.makeAttachment(
           readOnly: diskReadOnly,
           cachingMode: try VZDiskImageCachingMode(cachingModeRaw) ?? .automatic,
           synchronizationMode: try VZDiskImageSynchronizationMode(syncModeRaw)
         )
 
-        return AdditionalDisk(configuration: VZVirtioBlockDeviceConfiguration(attachment: attachment), temporaryDiskLock: lock)
+        return AdditionalDisk(
+          configuration: VZVirtioBlockDeviceConfiguration(attachment: attachment),
+          temporaryDiskLock: temporaryVMDirLock
+        )
       }
 
       // Unfortunately, VZDiskImageStorageDeviceAttachment does not support
diff --git a/Sources/tart/Config.swift b/Sources/tart/Config.swift
index 64af7de..a04ba82 100644
--- a/Sources/tart/Config.swift
+++ b/Sources/tart/Config.swift
@@ -33,7 +33,7 @@ struct Config {
         continue
       }
 
-      try FileManager.default.removeItem(at: entry)
+      try VMDirectory(baseURL: entry).removeFromDisk()
 
       try lock.unlock()
     }
diff --git a/Sources/tart/ContentStore.swift b/Sources/tart/ContentStore.swift
index 7608e61..fe9344e 100644
--- a/Sources/tart/ContentStore.swift
+++ b/Sources/tart/ContentStore.swift
@@ -16,6 +16,7 @@ struct ContentStore {
 
   let baseURL: URL
   private let digestDirectoryURL: URL
+  private let pruneLockURL: URL
 
   init() throws {
     try self.init(baseURL: Config().tartCacheDir.appendingPathComponent("content", isDirectory: true))
@@ -24,13 +25,41 @@ struct ContentStore {
   init(baseURL: URL) throws {
     self.baseURL = baseURL
     self.digestDirectoryURL = baseURL.appendingPathComponent(Self.digestAlgorithm, isDirectory: true)
+    self.pruneLockURL = baseURL.appendingPathComponent(".gc.lock")
     try FileManager.default.createDirectory(at: digestDirectoryURL, withIntermediateDirectories: true)
+    if !FileManager.default.fileExists(atPath: pruneLockURL.path) {
+      _ = FileManager.default.createFile(atPath: pruneLockURL.path, contents: Data())
+    }
+  }
+
+  /// Serializes reference publication with the final reference check and
+  /// deletion of immutable cache entries across Tart processes.
+  func withPruneLock<T>(_ body: () throws -> T) throws -> T {
+    let lock = try FileLock(lockURL: pruneLockURL)
+    try lock.lock()
+    defer { try? lock.unlock() }
+
+    return try body()
+  }
+
+  /// Waits for any prune already scanning references to finish. After this
+  /// returns, later prune runs can see a reference the caller already wrote.
+  func synchronizePublishedReferences() throws {
+    try withPruneLock {}
   }
 
   func contentURL(for contentDigest: String) throws -> URL {
+    try contentURL(for: contentDigest, under: baseURL)
+  }
+
+  /// Returns the canonical path for a digest under an arbitrary content-store
+  /// root without creating directories or lock files.
+  func contentURL(for contentDigest: String, under baseURL: URL) throws -> URL {
     let digestHex = try validatedDigestHex(contentDigest)
 
-    return digestDirectoryURL.appendingPathComponent(digestHex)
+    return baseURL
+      .appendingPathComponent(Self.digestAlgorithm, isDirectory: true)
+      .appendingPathComponent(digestHex)
   }
 
   func temporaryContentURL(for contentDigest: String) throws -> URL {
@@ -61,9 +90,9 @@ struct ContentStore {
     return lockURL
   }
 
-  /// Returns a digest-addressed entry without rereading it. Pull verifies
-  /// content hashes before accepting a cache hit; clone only needs a cheap
-  /// structural check, like Tart's existing disk.img path.
+  /// Returns an immutable digest-addressed entry without rereading it. Files
+  /// are verified when installed and when deciding whether a pull is a cache
+  /// hit; normal clone/run/push paths trust the store like Tart's disk.img.
   func contentURLIfPresent(for contentDigest: String) throws -> URL? {
     let url = try contentURL(for: contentDigest)
 
@@ -71,6 +100,8 @@ struct ContentStore {
       return nil
     }
 
+    try url.updateAccessDate()
+
     return url
   }
 
@@ -88,6 +119,33 @@ struct ContentStore {
     return url
   }
 
+  /// Returns immutable content files that no retained cached image or local VM
+  /// references. Callers may prune these like other cache entries.
+  func prunables(excluding referencedContentDigests: Swift.Set<String>) throws -> [URL] {
+    guard let enumerator = FileManager.default.enumerator(
+      at: digestDirectoryURL,
+      includingPropertiesForKeys: [.isRegularFileKey],
+      options: [.skipsSubdirectoryDescendants]
+    ) else {
+      return []
+    }
+
+    return try enumerator.compactMap { element in
+      guard let url = element as? URL,
+            try url.resourceValues(forKeys: [.isRegularFileKey]).isRegularFile == true else {
+        return nil
+      }
+
+      let contentDigest = "\(Self.digestPrefix)\(url.lastPathComponent)"
+      guard (try? validatedDigestHex(contentDigest)) != nil,
+            !referencedContentDigests.contains(contentDigest) else {
+        return nil
+      }
+
+      return url
+    }
+  }
+
   /// Move a fully reconstructed temporary file into the cache after verifying
   /// its semantic identity. The caller should create the temporary file with
   /// temporaryContentURL(for:) or resumableContentURL(for:) so rename stays on
diff --git a/Sources/tart/DiskImageStack.swift b/Sources/tart/DiskImageStack.swift
index 2a60e76..3b67e8e 100644
--- a/Sources/tart/DiskImageStack.swift
+++ b/Sources/tart/DiskImageStack.swift
@@ -45,6 +45,22 @@ struct DiskImageStack {
   let blockSize: UInt64
   let blockCount: UInt64
 
+  static var isSupported: Bool {
+    #if canImport(DiskImageKit)
+      if #available(macOS 27.0, *) {
+        return true
+      }
+    #endif
+
+    return false
+  }
+
+  static func requireSupport() throws {
+    guard isSupported else {
+      throw DiskImageStackError.unavailable
+    }
+  }
+
   /// Reads a disk image's current block layout without resolving or validating a
   /// whole stack. This is used for the VM's private writable overlay, whose
   /// size may be newer than the pinned immutable parent manifest.
@@ -69,21 +85,7 @@ struct DiskImageStack {
     #if canImport(DiskImageKit)
       if #available(macOS 27.0, *) {
         let image = try DiskImage(opening: .open(url: url, mode: .readOnly))
-        let matchesFormat = switch expectedFormat {
-        case .raw:
-          image.format == .raw
-        case .asif:
-          image.format == .asif
-        }
-        guard matchesFormat else {
-          throw DiskImageStackError.invalidDiskImage(url, "base disk format does not match")
-        }
-        guard image.layerType == nil, image.parentUUID == nil else {
-          throw DiskImageStackError.invalidDiskImage(url, "base disk must not be an overlay")
-        }
-        if expectedFormat == .asif && image.layerUUID == nil {
-          throw DiskImageStackError.invalidDiskImage(url, "ASIF base disk is missing a UUID")
-        }
+        try validateBase(image, at: url, expectedFormat: expectedFormat)
 
         return DiskImageBlockLayout(
           blockSize: UInt64(image.blockSize.rawValue),
@@ -214,7 +216,7 @@ struct DiskImageStack {
       }
 
       let baseImage = try DiskImage(opening: .open(url: baseURL, mode: .readOnly))
-      try validateBase(baseImage, at: baseURL, expectedFormat: baseFormat)
+      try Self.validateBase(baseImage, at: baseURL, expectedFormat: baseFormat)
 
       var image = baseImage
 
@@ -239,7 +241,7 @@ struct DiskImageStack {
     }
 
     @available(macOS 27.0, *)
-    private func validateBase(
+    private static func validateBase(
       _ image: DiskImage,
       at url: URL,
       expectedFormat: DiskImageFormat
diff --git a/Sources/tart/OCI/Manifest.swift b/Sources/tart/OCI/Manifest.swift
index 76801a6..08c84dd 100644
--- a/Sources/tart/OCI/Manifest.swift
+++ b/Sources/tart/OCI/Manifest.swift
@@ -37,6 +37,24 @@ struct TartDiskFileGroup: Equatable {
   /// Whole reconstructed-file digest. Existing flat manifests do not have
   /// this until a macOS 27 clone normalizes its local manifest copy.
   var contentDigest: String?
+
+  /// Expected size of the complete disk file reconstructed from these chunks.
+  func uncompressedSize() -> UInt64? {
+    var result: UInt64 = 0
+    for chunk in chunks {
+      guard let size = chunk.uncompressedSize() else {
+        return nil
+      }
+
+      let addition = result.addingReportingOverflow(size)
+      guard !addition.overflow else {
+        return nil
+      }
+      result = addition.partialValue
+    }
+
+    return result
+  }
 }
 
 enum TartDiskRepresentation: Equatable {
@@ -172,6 +190,16 @@ struct OCIManifest: Codable, Equatable {
     return .stacked(base: base, overlays: overlays)
   }
 
+  /// Returns content-store digests needed to reconstruct this disk stack.
+  func diskContentDigests() throws -> [String] {
+    switch try tartDiskRepresentation() {
+    case .flat(let base):
+      return base.contentDigest.map { [$0] } ?? []
+    case .stacked(let base, let overlays):
+      return ([base] + overlays).compactMap(\.contentDigest)
+    }
+  }
+
   private func validateChunkMetadata(_ chunks: [OCIManifestLayer]) throws {
     guard chunks.allSatisfy({ $0.uncompressedSize() != nil && $0.uncompressedContentDigest() != nil }) else {
       throw OCIManifestValidationError.invalidDiskMetadata("disk chunks need uncompressed size and content digest")
diff --git a/Sources/tart/VMDirectory+Archive.swift b/Sources/tart/VMDirectory+Archive.swift
index 969a84b..7ebbe95 100644
--- a/Sources/tart/VMDirectory+Archive.swift
+++ b/Sources/tart/VMDirectory+Archive.swift
@@ -1,3 +1,4 @@
+import Foundation
 import System
 import AppleArchive
 
@@ -10,8 +11,14 @@ fileprivate let permissions = FilePermissions(rawValue: 0o644)
 // [2]: https://developer.apple.com/documentation/compression/algorithm/lzfse
 extension VMDirectory {
   func exportToArchive(path: String) throws {
-    guard !isStackedVM && !isStackedCachedImage else {
-      throw RuntimeError.ExportFailed("exporting stacked VMs is not supported yet")
+    let temporaryArchive = try stackedArchiveDirectoryIfNeeded()
+    let archiveSourceURL = temporaryArchive?.vmDirectory.baseURL ?? baseURL
+
+    defer {
+      if let temporaryArchive {
+        try? temporaryArchive.lock.unlock()
+        try? temporaryArchive.vmDirectory.removeFromDisk()
+      }
     }
 
     guard let fileStream = ArchiveByteStream.fileStream(
@@ -53,7 +60,7 @@ extension VMDirectory {
       return
     }
 
-    try encodeStream.writeDirectoryContents(archiveFrom: FilePath(baseURL.path), keySet: keySet)
+    try encodeStream.writeDirectoryContents(archiveFrom: FilePath(archiveSourceURL.path), keySet: keySet)
   }
 
   func importFromArchive(path: String) throws {
@@ -96,5 +103,131 @@ extension VMDirectory {
     }
 
     _ = try ArchiveStream.process(readingFrom: decodeStream, writingTo: extractStream)
+
+    if isStackedVM {
+      try restoreStackedArchive()
+    }
   }
+
+  /// Builds a self-contained staging directory for a stacked archive, if this
+  /// directory currently resolves to a stacked VM or cached image.
+  private func stackedArchiveDirectoryIfNeeded() throws -> (vmDirectory: VMDirectory, lock: FileLock)? {
+    guard isStackedVM || isStackedCachedImage else {
+      return nil
+    }
+    try DiskImageStack.requireSupport()
+
+    let contentStore = try ContentStore()
+    let archiveVMDir = try VMDirectory.temporary()
+    let archiveVMDirLock = try FileLock(lockURL: archiveVMDir.baseURL)
+    try archiveVMDirLock.lock()
+
+    do {
+      let stagedSource: (isStackedVM: Bool, contentDigests: [String])? = try contentStore.withPruneLock {
+        // OCI tags are mutable symlinks. Resolve one digest record while tag
+        // replacement and cached-image deletion are blocked, then copy every
+        // source-owned file before releasing the lock.
+        let sourceVMDir = VMDirectory(baseURL: baseURL.resolvingSymlinksInPath())
+        guard sourceVMDir.isStackedVM || sourceVMDir.isStackedCachedImage else {
+          return nil
+        }
+
+        let sourceIsStackedVM = sourceVMDir.isStackedVM
+        if sourceIsStackedVM {
+          guard try sourceVMDir.state() == .Stopped else {
+            throw RuntimeError.ExportFailed("VM \"\(sourceVMDir.name)\" must be stopped before export")
+          }
+        }
+
+        try FileManager.default.copyItem(at: sourceVMDir.configURL, to: archiveVMDir.configURL)
+        try FileManager.default.copyItem(at: sourceVMDir.nvramURL, to: archiveVMDir.nvramURL)
+        try FileManager.default.copyItem(at: sourceVMDir.manifestURL, to: archiveVMDir.manifestURL)
+        if sourceIsStackedVM {
+          try FileManager.default.copyItem(at: sourceVMDir.overlayURL, to: archiveVMDir.overlayURL)
+        }
+
+        return (sourceIsStackedVM, try archiveVMDir.diskContentDigests())
+      }
+
+      guard let stagedSource else {
+        try archiveVMDirLock.unlock()
+        try archiveVMDir.removeFromDisk()
+        return nil
+      }
+
+      if !stagedSource.isStackedVM {
+        try archiveVMDir.diskImageStack().createWritableOverlay()
+      }
+
+      // The staged manifest is now an in-progress reference, so immutable
+      // content remains protected while these potentially large copies run
+      // without holding the global prune lock.
+      for contentDigest in stagedSource.contentDigests {
+        guard let sourceURL = try contentStore.contentURLIfPresent(for: contentDigest) else {
+          throw RuntimeError.ExportFailed("VM is missing cached disk content \(contentDigest)")
+        }
+
+        let destinationURL = try contentStore.contentURL(
+          for: contentDigest,
+          under: archiveContentStoreURL(in: archiveVMDir)
+        )
+        try FileManager.default.createDirectory(
+          at: destinationURL.deletingLastPathComponent(),
+          withIntermediateDirectories: true
+        )
+        try FileManager.default.copyItem(at: sourceURL, to: destinationURL)
+      }
+
+      return (archiveVMDir, archiveVMDirLock)
+    } catch {
+      try? archiveVMDirLock.unlock()
+      try? archiveVMDir.removeFromDisk()
+      throw error
+    }
+  }
+
+  /// Restores immutable files from an archive into the shared content store,
+  /// removes the archive-only payload, then validates the resulting stack.
+  private func restoreStackedArchive() throws {
+    try DiskImageStack.requireSupport()
+
+    let contentStore = try ContentStore()
+    // The extracted manifest is already a reference; synchronize publication
+    // with a concurrent prune before installing its immutable content.
+    try contentStore.synchronizePublishedReferences()
+    for contentDigest in try diskContentDigests() {
+      if try contentStore.existingContentURL(for: contentDigest) != nil {
+        continue
+      }
+
+      let archivedContentURL = try contentStore.contentURL(
+        for: contentDigest,
+        under: archiveContentStoreURL(in: self)
+      )
+      guard FileManager.default.fileExists(atPath: archivedContentURL.path) else {
+        throw RuntimeError.ImportFailed("archive is missing disk content \(contentDigest)")
+      }
+
+      let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
+      do {
+        try FileManager.default.copyItem(at: archivedContentURL, to: temporaryURL)
+        _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
+      } catch {
+        try? FileManager.default.removeItem(at: temporaryURL)
+        throw error
+      }
+    }
+
+    try FileManager.default.removeItem(at: archiveContentStoreURL(in: self))
+    try? FileManager.default.removeItem(at: stateURL)
+
+    // Opening the attachment validates the reconstructed immutable stack and
+    // imported writable overlay before the VM enters local storage.
+    _ = try diskImageStack().makeAttachment()
+  }
+
+  private func archiveContentStoreURL(in vmDir: VMDirectory) -> URL {
+    vmDir.baseURL.appendingPathComponent("content", isDirectory: true)
+  }
+
 }
diff --git a/Sources/tart/VMDirectory+DiskImageStack.swift b/Sources/tart/VMDirectory+DiskImageStack.swift
index 027672f..7ce6708 100644
--- a/Sources/tart/VMDirectory+DiskImageStack.swift
+++ b/Sources/tart/VMDirectory+DiskImageStack.swift
@@ -1,6 +1,12 @@
 import Foundation
 
 extension VMDirectory {
+  /// Returns content-store digests needed to reconstruct this VM's disk stack.
+  func diskContentDigests() throws -> [String] {
+    let manifest = try OCIManifest(fromJSON: Data(contentsOf: manifestURL))
+    return try manifest.diskContentDigests()
+  }
+
   func diskImageStack(contentStore providedStore: ContentStore? = nil) throws -> DiskImageStack {
     let manifest = try OCIManifest(fromJSON: Data(contentsOf: manifestURL))
     let base: TartDiskFileGroup
@@ -42,13 +48,18 @@ extension VMDirectory {
     generateMAC: Bool,
     contentStore: ContentStore? = nil
   ) throws {
-    try FileManager.default.copyItem(at: configURL, to: destination.configURL)
-    try FileManager.default.copyItem(at: nvramURL, to: destination.nvramURL)
-    try FileManager.default.copyItem(at: manifestURL, to: destination.manifestURL)
+    let contentStore = try contentStore ?? ContentStore()
+    try contentStore.withPruneLock {
+      try FileManager.default.copyItem(at: configURL, to: destination.configURL)
+      try FileManager.default.copyItem(at: nvramURL, to: destination.nvramURL)
+      try FileManager.default.copyItem(at: manifestURL, to: destination.manifestURL)
 
-    if copyWritableOverlay {
-      try FileManager.default.copyItem(at: overlayURL, to: destination.overlayURL)
-    } else {
+      if copyWritableOverlay {
+        try FileManager.default.copyItem(at: overlayURL, to: destination.overlayURL)
+      }
+    }
+
+    if !copyWritableOverlay {
       try destination.diskImageStack(contentStore: contentStore).createWritableOverlay()
     }
 
@@ -67,17 +78,6 @@ extension VMDirectory {
     let contentDigest = try Digest.hash(diskURL)
     let contentStore = try providedStore ?? ContentStore()
 
-    if try contentStore.existingContentURL(for: contentDigest) == nil {
-      let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
-      do {
-        try FileManager.default.copyItem(at: diskURL, to: temporaryURL)
-        _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
-      } catch {
-        try? FileManager.default.removeItem(at: temporaryURL)
-        throw error
-      }
-    }
-
     var manifest = try OCIManifest(fromJSON: Data(contentsOf: manifestURL))
     guard case .flat = try manifest.tartDiskRepresentation() else {
       throw RuntimeError.VMConfigurationError("--stacked cannot use an image that already has a stacked disk")
@@ -101,7 +101,24 @@ extension VMDirectory {
 
     try FileManager.default.copyItem(at: configURL, to: destination.configURL)
     try FileManager.default.copyItem(at: nvramURL, to: destination.nvramURL)
-    try manifest.toJSON().write(to: destination.manifestURL)
+    try contentStore.withPruneLock {
+      try manifest.toJSON().write(to: destination.manifestURL)
+    }
+
+    // Publish the temporary VM's manifest before installing the shared base.
+    // Reference-aware pruning includes in-progress manifests, so the content
+    // cannot be collected in the window before this VM is moved into place.
+    if try contentStore.contentURLIfPresent(for: contentDigest) == nil {
+      let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
+      do {
+        try FileManager.default.copyItem(at: diskURL, to: temporaryURL)
+        _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
+      } catch {
+        try? FileManager.default.removeItem(at: temporaryURL)
+        throw error
+      }
+    }
+
     try destination.diskImageStack(contentStore: contentStore).createWritableOverlay()
 
     if generateMAC {
diff --git a/Sources/tart/VMDirectory+OCI.swift b/Sources/tart/VMDirectory+OCI.swift
index ed0ec64..c5d1a87 100644
--- a/Sources/tart/VMDirectory+OCI.swift
+++ b/Sources/tart/VMDirectory+OCI.swift
@@ -114,7 +114,7 @@ extension VMDirectory {
 
   /// Reconstructs one complete immutable base disk or published ASIF overlay
   /// from its Tart disk chunks, unless the shared content store already has a
-  /// verified copy.
+  /// size-matching copy.
   private func pullDiskFile(
     registry: Registry,
     group: TartDiskFileGroup,
@@ -133,7 +133,10 @@ extension VMDirectory {
     try lock.lock()
     defer { try? lock.unlock() }
 
-    if let existingURL = try contentStore.existingContentURL(for: contentDigest) {
+    if let existingURL = try contentStore.contentURLIfPresent(for: contentDigest),
+       let actualSize = UInt64(exactly: try existingURL.sizeBytes()),
+       let expectedSize = group.uncompressedSize(),
+       actualSize == expectedSize {
       progress.completedUnitCount += group.chunks.reduce(0) { $0 + Int64($1.size) }
       return existingURL
     }
@@ -189,7 +192,6 @@ extension VMDirectory {
     var annotations = diskAnnotations
     annotations[uploadTimeAnnotation] = Date().toISO()
     manifest.annotations = annotations
-
     // Manifest
     for reference in references {
       defaultLogger.appendNewLine("pushing manifest for \(reference)...")
@@ -250,26 +252,13 @@ extension VMDirectory {
       ))
     }
 
-    // Keep the snapshot out of startup GC while this potentially long push
-    // hashes, uploads, and inspects it.
-    let frozenOverlayDirectory = try VMDirectory.temporary()
-    let frozenOverlayLock = try FileLock(lockURL: frozenOverlayDirectory.baseURL)
-    try frozenOverlayLock.lock()
-    defer {
-      try? frozenOverlayLock.unlock()
-      try? FileManager.default.removeItem(at: frozenOverlayDirectory.baseURL)
-    }
-
-    let frozenOverlayURL = frozenOverlayDirectory.baseURL.appendingPathComponent("overlay.asif")
-    try FileManager.default.copyItem(at: overlayURL, to: frozenOverlayURL)
-
-    let overlaySize = try FileManager.default.attributesOfItem(atPath: frozenOverlayURL.path)[.size] as! Int64
+    let overlaySize = try FileManager.default.attributesOfItem(atPath: overlayURL.path)[.size] as! Int64
     defaultLogger.appendNewLine("pushing overlay...")
     let progress = Progress(totalUnitCount: overlaySize)
     ProgressObserver(progress).log(defaultLogger)
-    let contentDigest = try Digest.hash(frozenOverlayURL)
+    let contentDigest = try Digest.hash(overlayURL)
     let chunks = try await DiskV2.push(
-      diskURL: frozenOverlayURL,
+      diskURL: overlayURL,
       mediaType: asifOverlayMediaType,
       registry: registry,
       chunkSizeMb: chunkSizeMb,
@@ -278,7 +267,7 @@ extension VMDirectory {
     )
     layers.append(contentsOf: annotatedChunks(chunks, kind: .asifOverlay, contentDigest: contentDigest))
 
-    let blockLayout = try DiskImageStack.diskImageBlockLayout(at: frozenOverlayURL)
+    let blockLayout = try DiskImageStack.diskImageBlockLayout(at: overlayURL)
     let diskSize = blockLayout.blockSize.multipliedReportingOverflow(by: blockLayout.blockCount)
     guard !diskSize.overflow else {
       throw DiskImageStackError.invalidBlockLayout("stacked disk block layout overflows UInt64")
@@ -316,6 +305,8 @@ extension VMDirectory {
       return group.chunks
     }
 
+    // Rebuilding transport blobs republishes this file under the pinned
+    // whole-file digest, so validate the cached bytes at this boundary.
     guard let contentURL = try contentStore.existingContentURL(for: contentDigest) else {
       throw RuntimeError.VMMissingFiles("stacked VM is missing cached disk content \(contentDigest)")
     }
diff --git a/Sources/tart/VMDirectory.swift b/Sources/tart/VMDirectory.swift
index 89a4922..45caeb3 100644
--- a/Sources/tart/VMDirectory.swift
+++ b/Sources/tart/VMDirectory.swift
@@ -236,17 +236,29 @@ struct VMDirectory: Prunable {
     contentStore: ContentStore? = nil
   ) throws {
     if isStackedVM {
-      guard try state() == .Stopped else {
+      // Resolve the stack before taking the config.json PID lock. Reading
+      // config.json after acquiring an fcntl lock would release that lock
+      // when the read file descriptor is closed.
+      let stack = try diskImageStack(contentStore: contentStore)
+      let lock = try lock()
+      guard try lock.trylock() else {
+        throw RuntimeError.VMConfigurationError("VM \"\(name)\" must be stopped before resizing its disk")
+      }
+      defer { try? lock.unlock() }
+
+      // Holding the PID lock proves that the VM is not running. A saved state
+      // file is the remaining suspended state that must also reject resize.
+      guard !FileManager.default.fileExists(atPath: stateURL.path) else {
         throw RuntimeError.VMConfigurationError("VM \"\(name)\" must be stopped before resizing its disk")
       }
 
-      let stack = try diskImageStack(contentStore: contentStore)
       let desiredSizeBytes = UInt64(sizeGB) * 1000 * 1000 * 1000
       guard desiredSizeBytes.isMultiple(of: stack.blockSize) else {
         throw RuntimeError.InvalidDiskSize("new disk size must align to the stacked disk block size")
       }
 
-      try stack.growWritableOverlay(toBlockCount: desiredSizeBytes / stack.blockSize)
+      let desiredBlockCount = desiredSizeBytes / stack.blockSize
+      try stack.growWritableOverlay(toBlockCount: desiredBlockCount)
       return
     }
 
@@ -373,11 +385,20 @@ struct VMDirectory: Prunable {
       throw RuntimeError.VMIsRunning(name)
     }
 
-    try FileManager.default.removeItem(at: baseURL)
+    try removeFromDisk()
 
     try lock.unlock()
   }
 
+  /// Removes a VM directory while preserving the content-store reference
+  /// protocol for any complete or partially published manifest it contains.
+  func removeFromDisk() throws {
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      try FileManager.default.removeItem(at: baseURL)
+    }
+  }
+
   func accessDate() throws -> Date {
     try baseURL.accessDate()
   }
diff --git a/Sources/tart/VMStorageLocal.swift b/Sources/tart/VMStorageLocal.swift
index 6a39420..4297caa 100644
--- a/Sources/tart/VMStorageLocal.swift
+++ b/Sources/tart/VMStorageLocal.swift
@@ -35,11 +35,27 @@ class VMStorageLocal: PrunableStorage {
 
   func move(_ name: String, from: VMDirectory) throws {
     _ = try FileManager.default.createDirectory(at: baseURL, withIntermediateDirectories: true)
-    _ = try FileManager.default.replaceItemAt(vmURL(name), withItemAt: from.baseURL)
+    try replace(VMDirectory(baseURL: vmURL(name)), with: from)
   }
 
   func rename(_ name: String, _ newName: String) throws {
-    _ = try FileManager.default.replaceItemAt(vmURL(newName), withItemAt: vmURL(name))
+    let source = VMDirectory(baseURL: vmURL(name))
+    let destination = VMDirectory(baseURL: vmURL(newName))
+    try replace(destination, with: source)
+  }
+
+  /// References in a manifest must not disappear while content GC is deciding
+  /// whether their immutable disk files are still in use.
+  private func replace(_ destination: VMDirectory, with source: VMDirectory) throws {
+    if FileManager.default.fileExists(atPath: source.manifestURL.path) ||
+      FileManager.default.fileExists(atPath: destination.manifestURL.path) {
+      let contentStore = try ContentStore()
+      try contentStore.withPruneLock {
+        _ = try FileManager.default.replaceItemAt(destination.baseURL, withItemAt: source.baseURL)
+      }
+    } else {
+      _ = try FileManager.default.replaceItemAt(destination.baseURL, withItemAt: source.baseURL)
+    }
   }
 
   func delete(_ name: String) throws {
diff --git a/Sources/tart/VMStorageOCI.swift b/Sources/tart/VMStorageOCI.swift
index 4848f77..0ca6cf2 100644
--- a/Sources/tart/VMStorageOCI.swift
+++ b/Sources/tart/VMStorageOCI.swift
@@ -22,9 +22,8 @@ class VMStorageOCI: PrunableStorage {
   }
 
   /// Whether clone can use a cached image without pulling. Standalone images keep
-  /// Tart's existing structural check. Stacked cached images cheaply require every
-  /// immutable file with its expected length; explicit pull remains the path
-  /// that hashes content and repairs same-sized corruption.
+  /// Tart's existing structural check. Stacked cached images require every
+  /// immutable file with its expected length.
   func hasUsableCachedImageForClone(_ name: RemoteName, requireManifest: Bool = false) throws -> Bool {
     guard exists(name) else {
       return false
@@ -45,25 +44,7 @@ class VMStorageOCI: PrunableStorage {
 
     let contentStore = try ContentStore()
     for group in [base] + overlays {
-      guard let contentDigest = group.contentDigest,
-            let contentURL = try contentStore.contentURLIfPresent(for: contentDigest) else {
-        return false
-      }
-
-      var expectedSize: UInt64 = 0
-      for chunk in group.chunks {
-        guard let uncompressedSize = chunk.uncompressedSize() else {
-          return false
-        }
-        let addition = expectedSize.addingReportingOverflow(uncompressedSize)
-        guard !addition.overflow else {
-          return false
-        }
-        expectedSize = addition.partialValue
-      }
-
-      guard let actualSize = UInt64(exactly: try contentURL.sizeBytes()),
-            actualSize == expectedSize else {
+      guard try hasUsableCachedDiskFile(group, contentStore: contentStore) else {
         return false
       }
     }
@@ -164,35 +145,44 @@ class VMStorageOCI: PrunableStorage {
 
     let vmDir = try create(name, overwrite: exists(name))
 
-    if source.isStackedVM {
-      guard case .stacked(_, let overlays) = try manifest.tartDiskRepresentation(),
-            let contentDigest = overlays.last?.contentDigest else {
-        throw RuntimeError.VMConfigurationError("pushed image is missing its writable ASIF overlay")
-      }
-
-      // The pushed top overlay becomes immutable in the cached image. Keep a
-      // semantic copy so later clones do not need to fetch it back.
-      let contentStore = try ContentStore()
-      if try contentStore.existingContentURL(for: contentDigest) == nil {
-        let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
-        do {
-          try FileManager.default.copyItem(at: source.overlayURL, to: temporaryURL)
-          _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
-        } catch {
-          try? FileManager.default.removeItem(at: temporaryURL)
-          throw error
+    do {
+      if source.isStackedVM {
+        guard case .stacked(_, let overlays) = try manifest.tartDiskRepresentation(),
+              let contentDigest = overlays.last?.contentDigest else {
+          throw RuntimeError.VMConfigurationError("pushed image is missing its writable ASIF overlay")
         }
+
+        // The pushed top overlay becomes immutable in the cached image. Keep a
+        // semantic copy so later clones do not need to fetch it back.
+        let contentStore = try ContentStore()
+        try contentStore.withPruneLock {
+          try FileManager.default.copyItem(at: source.configURL, to: vmDir.configURL)
+          try FileManager.default.copyItem(at: source.nvramURL, to: vmDir.nvramURL)
+          // Publish the reference before installing the immutable top overlay,
+          // so reference-aware pruning cannot collect it in between.
+          try manifest.toJSON().write(to: vmDir.manifestURL)
+        }
+
+        if try contentStore.contentURLIfPresent(for: contentDigest) == nil {
+          let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
+          do {
+            try FileManager.default.copyItem(at: source.overlayURL, to: temporaryURL)
+            _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
+          } catch {
+            try? FileManager.default.removeItem(at: temporaryURL)
+            throw error
+          }
+        }
+      } else {
+        try source.clone(to: vmDir, generateMAC: false)
+        // Keep the exact manifest Tart submitted so tag links and later pushes
+        // refer to the same digest-addressed cached image.
+        try manifest.toJSON().write(to: vmDir.manifestURL)
       }
-
-      try FileManager.default.copyItem(at: source.configURL, to: vmDir.configURL)
-      try FileManager.default.copyItem(at: source.nvramURL, to: vmDir.nvramURL)
-    } else {
-      try source.clone(to: vmDir, generateMAC: false)
+    } catch {
+      try? vmDir.removeFromDisk()
+      throw error
     }
-
-    // Keep the exact manifest Tart submitted so tag links and later pushes
-    // refer to the same digest-addressed cached image.
-    try manifest.toJSON().write(to: vmDir.manifestURL)
   }
 
   func move(_ name: RemoteName, from: VMDirectory) throws{
@@ -203,11 +193,20 @@ class VMStorageOCI: PrunableStorage {
     try FileManager.default.createDirectory(at: targetURL.deletingLastPathComponent(),
                                             withIntermediateDirectories: true)
 
-    _ = try FileManager.default.replaceItemAt(targetURL, withItemAt: from.baseURL)
+    let target = VMDirectory(baseURL: targetURL)
+    if FileManager.default.fileExists(atPath: from.manifestURL.path) ||
+      FileManager.default.fileExists(atPath: target.manifestURL.path) {
+      let contentStore = try ContentStore()
+      try contentStore.withPruneLock {
+        _ = try FileManager.default.replaceItemAt(targetURL, withItemAt: from.baseURL)
+      }
+    } else {
+      _ = try FileManager.default.replaceItemAt(targetURL, withItemAt: from.baseURL)
+    }
   }
 
   func delete(_ name: RemoteName) throws {
-    try FileManager.default.removeItem(at: vmURL(name))
+    try removeRecord(at: vmURL(name))
     try gc()
   }
 
@@ -216,6 +215,7 @@ class VMStorageOCI: PrunableStorage {
 
     guard let enumerator = FileManager.default.enumerator(at: baseURL,
                                                           includingPropertiesForKeys: [.isSymbolicLinkKey]) else {
+      try gcContent()
       return
     }
 
@@ -243,7 +243,28 @@ class VMStorageOCI: PrunableStorage {
       let vmDir = VMDirectory(baseURL: baseURL)
 
       if !vmDir.isExplicitlyPulled() && incRefCount == 0 {
-        try FileManager.default.removeItem(at: baseURL)
+        try removeRecord(at: baseURL)
+      }
+    }
+
+    try gcContent()
+  }
+
+  /// Cached images with a manifest publish references into the shared content
+  /// store. Remove them through VMDirectory so reference removal is serialized
+  /// with clone, export, pull, and content GC, even if a record is incomplete.
+  private func removeRecord(at url: URL) throws {
+    try VMDirectory(baseURL: url).removeFromDisk()
+  }
+
+  /// Remove immutable files whose final published or in-progress reference
+  /// has disappeared, without collecting unrelated cached images.
+  fileprivate func gcContent() throws {
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      let referencedContentDigests = try referencedContentDigests(includeCachedImages: true)
+      for contentURL in try contentStore.prunables(excluding: referencedContentDigests) {
+        try FileManager.default.removeItem(at: contentURL)
       }
     }
   }
@@ -287,9 +308,57 @@ class VMStorageOCI: PrunableStorage {
   }
 
   func prunables() throws -> [Prunable] {
-    try list().filter { (_, vmDir, isSymlink) in
-      !isSymlink && vmDir.isStandalone
+    let records = try list().filter { (_, _, isSymlink) in
+      !isSymlink
     }.map { (_, vmDir, _) in vmDir }
+
+    // Attribute shared content to the newest cached image that references it.
+    // This counts each file once while charging it to the last record that
+    // normally needs to be removed before the file becomes reclaimable.
+    let nonCacheContentDigests = try referencedContentDigests(includeCachedImages: false)
+    var contentOwners = [String: VMDirectory]()
+    for record in records where record.isStackedCachedImage {
+      // Interrupted cache population can leave a truncated manifest in an
+      // otherwise recognizable cached record. It has no reliable content
+      // references, but it must not prevent pruning other cache entries.
+      for contentDigest in (try? record.diskContentDigests()) ?? []
+        where !nonCacheContentDigests.contains(contentDigest) {
+        guard let currentOwner = contentOwners[contentDigest] else {
+          contentOwners[contentDigest] = record
+          continue
+        }
+
+        let recordAccessDate = try record.accessDate()
+        let currentAccessDate = try currentOwner.accessDate()
+        if recordAccessDate > currentAccessDate ||
+          (recordAccessDate == currentAccessDate && record.url.path > currentOwner.url.path) {
+          contentOwners[contentDigest] = record
+        }
+      }
+    }
+
+    let contentStore = try ContentStore()
+    var ownedContentURLs = [URL: [URL]]()
+    for (contentDigest, owner) in contentOwners {
+      let contentURL = try contentStore.contentURL(for: contentDigest)
+      guard FileManager.default.fileExists(atPath: contentURL.path) else {
+        continue
+      }
+
+      ownedContentURLs[owner.url, default: []].append(contentURL)
+    }
+
+    var result: [Prunable] = records.map { record in
+      CachedImagePrunable(
+        vmDir: record,
+        ownedContentURLs: ownedContentURLs[record.url] ?? []
+      )
+    }
+
+    result += try contentStore.prunables(excluding: referencedContentDigests(includeCachedImages: true))
+      .map(ContentPrunable.init)
+
+    return result
   }
 
   func pull(_ name: RemoteName, registry: Registry, concurrency: UInt, deduplicate: Bool) async throws {
@@ -345,6 +414,12 @@ class VMStorageOCI: PrunableStorage {
       let tmpVMDirLock = try FileLock(lockURL: tmpVMDir.baseURL)
       try tmpVMDirLock.lock()
 
+      // Make in-progress stacked content references visible before reclaiming
+      // space or reconstructing immutable files.
+      try ContentStore().withPruneLock {
+        try manifestData.write(to: tmpVMDir.manifestURL)
+      }
+
       // A previously pulled standalone image already has the complete base
       // disk locally as disk.img. Promote that file into the content store
       // before sizing or pulling so a stacked child only fetches overlays.
@@ -402,16 +477,13 @@ class VMStorageOCI: PrunableStorage {
           return .throw
         }
 
-        // Preserve the exact manifest bytes received from the registry. Its
-        // digest identifies this cached image and stacked VMs pin it.
-        try manifestData.write(to: tmpVMDir.manifestURL)
         if preserveExplicitlyPulledMark {
           tmpVMDir.markExplicitlyPulled()
         }
 
         try move(digestName, from: tmpVMDir)
       }, onCancel: {
-        try? FileManager.default.removeItem(at: tmpVMDir.baseURL)
+        try? tmpVMDir.removeFromDisk()
       })
     } else {
       defaultLogger.appendNewLine("\(digestName) image is already cached! creating a symlink...")
@@ -430,9 +502,10 @@ class VMStorageOCI: PrunableStorage {
     _ = try VMStorageOCI().open(name)
   }
 
-  /// Returns `nil` for standalone images and the missing immutable disk-file groups
-  /// for stacked images. `ContentStore.existingContentURL()` intentionally
-  /// validates the digest so corrupt entries are repaired by a normal pull.
+  /// Returns nil for standalone images and the missing immutable disk-file
+  /// groups for stacked images. Like existing standalone cached images, cache hits trust
+  /// already-installed files; checking size still repairs truncated entries
+  /// without hashing a large prewarmed base on every pull.
   private func missingStackedDiskFileGroups(for manifest: OCIManifest) throws -> [TartDiskFileGroup]? {
     guard case .stacked(let base, let overlays) = try manifest.tartDiskRepresentation() else {
       return nil
@@ -441,10 +514,7 @@ class VMStorageOCI: PrunableStorage {
     let contentStore = try ContentStore()
     var missingGroups: [TartDiskFileGroup] = []
     for group in [base] + overlays {
-      guard let contentDigest = group.contentDigest else {
-        throw OCIManifestValidationError.invalidDiskMetadata("stacked disk files need a whole-file content digest")
-      }
-      if try contentStore.existingContentURL(for: contentDigest) == nil {
+      if try !hasUsableCachedDiskFile(group, contentStore: contentStore) {
         missingGroups.append(group)
       }
     }
@@ -462,30 +532,32 @@ class VMStorageOCI: PrunableStorage {
     }
 
     let contentStore = try ContentStore()
-    // Content-store entries are verified when installed. Avoid hashing a
-    // potentially large prewarmed base again on every stacked pull.
-    guard try contentStore.contentURLIfPresent(for: contentDigest) == nil else {
-      return
-    }
-
-    for (_, vmDir, isSymlink) in try list() where !isSymlink && vmDir.isStandalone {
-      guard let manifestData = try? Data(contentsOf: vmDir.manifestURL),
-            let candidateManifest = try? OCIManifest(fromJSON: manifestData),
-            case .flat(let candidateBase) = try? candidateManifest.tartDiskRepresentation(),
-            diskChunksMatch(candidateBase.chunks, base.chunks) else {
-        continue
+    try contentStore.withPruneLock {
+      // Content-store entries are verified when installed. Avoid hashing a
+      // potentially large prewarmed base again on every stacked pull.
+      guard try contentStore.contentURLIfPresent(for: contentDigest) == nil else {
+        return
       }
 
-      let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
-      do {
-        try FileManager.default.copyItem(at: vmDir.diskURL, to: temporaryURL)
-        _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
-        return
-      } catch ContentStoreError.contentDigestMismatch {
-        try? FileManager.default.removeItem(at: temporaryURL)
-      } catch {
-        try? FileManager.default.removeItem(at: temporaryURL)
-        throw error
+      for (_, vmDir, isSymlink) in try list() where !isSymlink && vmDir.isStandalone {
+        guard let manifestData = try? Data(contentsOf: vmDir.manifestURL),
+              let candidateManifest = try? OCIManifest(fromJSON: manifestData),
+              case .flat(let candidateBase) = try? candidateManifest.tartDiskRepresentation(),
+              diskChunksMatch(candidateBase.chunks, base.chunks) else {
+          continue
+        }
+
+        let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
+        do {
+          try FileManager.default.copyItem(at: vmDir.diskURL, to: temporaryURL)
+          _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
+          return
+        } catch ContentStoreError.contentDigestMismatch {
+          try? FileManager.default.removeItem(at: temporaryURL)
+        } catch {
+          try? FileManager.default.removeItem(at: temporaryURL)
+          throw error
+        }
       }
     }
   }
@@ -506,6 +578,17 @@ class VMStorageOCI: PrunableStorage {
     }
   }
 
+  private func hasUsableCachedDiskFile(_ group: TartDiskFileGroup, contentStore: ContentStore) throws -> Bool {
+    guard let contentDigest = group.contentDigest,
+          let contentURL = try contentStore.contentURLIfPresent(for: contentDigest),
+          let actualSize = UInt64(exactly: try contentURL.sizeBytes()),
+          let expectedSize = group.uncompressedSize() else {
+      return false
+    }
+
+    return actualSize == expectedSize
+  }
+
   func linked(from: RemoteName, to: RemoteName) -> Bool {
     do {
       let resolvedFrom = try FileManager.default.destinationOfSymbolicLink(atPath: vmURL(from).path)
@@ -516,9 +599,13 @@ class VMStorageOCI: PrunableStorage {
   }
 
   func link(from: RemoteName, to: RemoteName) throws {
-    try? FileManager.default.removeItem(at: vmURL(from))
-
-    try FileManager.default.createSymbolicLink(at: vmURL(from), withDestinationURL: vmURL(to))
+    // Export resolves mutable tags while holding this same lock, so replace
+    // the symlink atomically with respect to stacked archive staging.
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      try? FileManager.default.removeItem(at: vmURL(from))
+      try FileManager.default.createSymbolicLink(at: vmURL(from), withDestinationURL: vmURL(to))
+    }
 
     try gc()
   }
@@ -594,6 +681,108 @@ class VMStorageOCI: PrunableStorage {
       try LocalLayerCache(choosen.name, choosen.deduplicatedBytes, choosen.vmDir.diskURL, choosen.manifest)
     })
   }
+
+  /// Returns content referenced outside the OCI cache, optionally including
+  /// references published by retained cached images.
+  private func referencedContentDigests(includeCachedImages: Bool) throws -> Swift.Set<String> {
+    var result = Swift.Set<String>()
+
+    for (_, vmDir) in try VMStorageLocal().list() where vmDir.isStackedVM {
+      result.formUnion(try vmDir.diskContentDigests())
+    }
+
+    // Clone, pull, and import publish their manifest before installing
+    // immutable content. Include partially populated temporary directories so
+    // pruning cannot race those operations.
+    for url in try FileManager.default.contentsOfDirectory(
+      at: Config().tartTmpDir,
+      includingPropertiesForKeys: [],
+      options: .skipsHiddenFiles
+    ) {
+      let vmDir = VMDirectory(baseURL: url)
+      guard FileManager.default.fileExists(atPath: vmDir.manifestURL.path),
+            let contentDigests = try? vmDir.diskContentDigests() else {
+        continue
+      }
+
+      result.formUnion(contentDigests)
+    }
+
+    if includeCachedImages {
+      for (_, vmDir, isSymlink) in try list() where !isSymlink && vmDir.isStackedCachedImage {
+        // Malformed cached records are invalid references. Keep scanning so
+        // one interrupted population does not disable content GC globally.
+        if let contentDigests = try? vmDir.diskContentDigests() {
+          result.formUnion(contentDigests)
+        }
+      }
+    }
+
+    return result
+  }
+
+  fileprivate func deleteContentIfUnused(_ url: URL) throws {
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      let referencedContentDigests = try referencedContentDigests(includeCachedImages: true)
+      let stillPrunable = try contentStore.prunables(excluding: referencedContentDigests).contains {
+        $0.resolvingSymlinksInPath() == url.resolvingSymlinksInPath()
+      }
+      if stillPrunable {
+        try FileManager.default.removeItem(at: url)
+      }
+    }
+  }
+}
+
+private struct ContentPrunable: Prunable {
+  let url: URL
+
+  func delete() throws {
+    try VMStorageOCI().deleteContentIfUnused(url)
+  }
+
+  func accessDate() throws -> Date {
+    try url.accessDate()
+  }
+
+  func sizeBytes() throws -> Int {
+    try url.sizeBytes()
+  }
+
+  func allocatedSizeBytes() throws -> Int {
+    try url.allocatedSizeBytes()
+  }
+}
+
+/// A digest-addressed cached image plus immutable content attributed to the
+/// final remote reference that can release it.
+private struct CachedImagePrunable: Prunable {
+  let vmDir: VMDirectory
+  let ownedContentURLs: [URL]
+
+  var url: URL {
+    vmDir.url
+  }
+
+  func delete() throws {
+    try vmDir.delete()
+    // Deleting a record can make attributed content unreferenced. Run GC now
+    // so one prune invocation reclaims those bytes.
+    try VMStorageOCI().gcContent()
+  }
+
+  func accessDate() throws -> Date {
+    try vmDir.accessDate()
+  }
+
+  func sizeBytes() throws -> Int {
+    try vmDir.sizeBytes() + ownedContentURLs.map { try $0.sizeBytes() }.reduce(0, +)
+  }
+
+  func allocatedSizeBytes() throws -> Int {
+    try vmDir.allocatedSizeBytes() + ownedContentURLs.map { try $0.allocatedSizeBytes() }.reduce(0, +)
+  }
 }
 
 extension URL {
diff --git a/Tests/TartTests/ContentStoreTests.swift b/Tests/TartTests/ContentStoreTests.swift
index 95224cc..6a98e90 100644
--- a/Tests/TartTests/ContentStoreTests.swift
+++ b/Tests/TartTests/ContentStoreTests.swift
@@ -31,6 +31,7 @@ final class ContentStoreTests: XCTestCase {
     try Data("corrupt".utf8).write(to: contentURL)
 
     XCTAssertNil(try store.existingContentURL(for: expectedDigest))
+    XCTAssertEqual(try store.contentURLIfPresent(for: expectedDigest), contentURL)
   }
 
   func testResumableAndLockURLsAreStablePerDigest() throws {
@@ -113,6 +114,24 @@ final class ContentStoreTests: XCTestCase {
     }
   }
 
+  func testContentURLUnderArbitraryRootHasNoSideEffects() throws {
+    let rootURL = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
+    addTeardownBlock {
+      try? FileManager.default.removeItem(at: rootURL)
+    }
+    let digest = Digest.hash(Data("content".utf8))
+
+    let contentStore = try temporaryStore()
+    let contentURL = try contentStore.contentURL(for: digest, under: rootURL)
+
+    XCTAssertEqual(
+      contentURL,
+      rootURL.appendingPathComponent("sha256", isDirectory: true)
+        .appendingPathComponent(String(digest.dropFirst("sha256:".count)))
+    )
+    XCTAssertFalse(FileManager.default.fileExists(atPath: rootURL.path))
+  }
+
   private func temporaryStore() throws -> ContentStore {
     let url = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
     addTeardownBlock {
diff --git a/Tests/TartTests/OCIManifestTests.swift b/Tests/TartTests/OCIManifestTests.swift
index 63b68bd..9926827 100644
--- a/Tests/TartTests/OCIManifestTests.swift
+++ b/Tests/TartTests/OCIManifestTests.swift
@@ -21,6 +21,33 @@ final class OCIManifestTests: XCTestCase {
     ))
   }
 
+  func testDiskFileGroupUncompressedSize() {
+    let first = chunk(mediaType: diskV2MediaType, suffix: "base-0")
+    let second = chunk(mediaType: diskV2MediaType, suffix: "base-1")
+    XCTAssertEqual(TartDiskFileGroup(kind: .base, chunks: [first, second], contentDigest: nil).uncompressedSize(), 2)
+
+    var overflowing = first
+    overflowing.annotations?[uncompressedSizeAnnotation] = String(UInt64.max)
+    XCTAssertNil(TartDiskFileGroup(kind: .base, chunks: [overflowing, second], contentDigest: nil).uncompressedSize())
+  }
+
+  func testDiskContentDigests() throws {
+    let flatBase = chunk(mediaType: diskV2MediaType, suffix: "flat", diskFileDigest: "sha256:flat")
+    XCTAssertEqual(try manifest(diskDescriptors: [flatBase]).diskContentDigests(), ["sha256:flat"])
+
+    let stackedBase = chunk(mediaType: diskV2MediaType, suffix: "base", diskFileDigest: "sha256:base")
+    let overlay = chunk(
+      mediaType: asifOverlayMediaType,
+      suffix: "overlay",
+      diskFileDigest: "sha256:overlay",
+      chunkCount: 1
+    )
+    XCTAssertEqual(
+      try manifest(diskDescriptors: [stackedBase, overlay]).diskContentDigests(),
+      ["sha256:base", "sha256:overlay"]
+    )
+  }
+
   func testStackedRepresentationRequiresBaseDigest() throws {
     let base = chunk(mediaType: diskV2MediaType, suffix: "base-0")
     let overlay = chunk(mediaType: asifOverlayMediaType, suffix: "overlay-0", diskFileDigest: "sha256:overlay", chunkCount: 1)
diff --git a/Tests/TartTests/VMDirectoryDiskImageStackTests.swift b/Tests/TartTests/VMDirectoryDiskImageStackTests.swift
index 63ae535..f0049b4 100644
--- a/Tests/TartTests/VMDirectoryDiskImageStackTests.swift
+++ b/Tests/TartTests/VMDirectoryDiskImageStackTests.swift
@@ -79,7 +79,37 @@ import XCTest
       XCTAssertTrue(FileManager.default.fileExists(atPath: fresh.overlayURL.path))
     }
 
-    func testResizeDiskGrowsWritableOverlay() throws {
+    func testStackedRemoteAdditionalDiskRetainsTemporaryVM() throws {
+      try withTemporaryTartHome {
+        let source = try flatSource()
+        let stacked = try temporaryVMDirectory()
+        try source.cloneAsStackedBase(to: stacked, generateMAC: false)
+
+        let storage = try VMStorageOCI()
+        let name = try RemoteName("example.com/org/image:latest")
+        let cachedImage = try storage.create(name)
+        try FileManager.default.copyItem(at: stacked.configURL, to: cachedImage.configURL)
+        try FileManager.default.copyItem(at: stacked.nvramURL, to: cachedImage.nvramURL)
+        try FileManager.default.copyItem(at: stacked.manifestURL, to: cachedImage.manifestURL)
+
+        do {
+          let additionalDisk = try AdditionalDisk(parseFrom: name.description)
+          let entries = try temporaryEntries()
+          XCTAssertEqual(entries.count, 1)
+          XCTAssertTrue(VMDirectory(baseURL: entries[0]).isStackedVM)
+
+          try Config().gc()
+          XCTAssertEqual(try temporaryEntries(), entries)
+
+          withExtendedLifetime(additionalDisk) {}
+        }
+
+        try Config().gc()
+        XCTAssertTrue(try temporaryEntries().isEmpty)
+      }
+    }
+
+    func testResizeDiskGrowsWritableOverlayAndPreservesParentGeometry() throws {
       let contentStore = try temporaryContentStore()
       let source = try flatSource()
       let stacked = try temporaryVMDirectory()
@@ -90,6 +120,107 @@ import XCTest
       let image = try DiskImage(opening: .open(url: stacked.overlayURL, mode: .readOnly))
       XCTAssertEqual(image.blockCount, 1_000_000_000 / 512)
       XCTAssertEqual(try stacked.diskSizeBytes(), 1_000_000_000)
+
+      let manifest = try OCIManifest(fromJSON: Data(contentsOf: stacked.manifestURL))
+      XCTAssertEqual(manifest.diskBlockSize(), 512)
+      XCTAssertEqual(manifest.diskBlockCount(), 8)
+
+      _ = try stacked.diskImageStack(contentStore: contentStore).makeAttachment()
+    }
+
+    func testStackedArchiveRoundTripsImmutableContentAndOverlay() throws {
+      try withTemporaryTartHome {
+        let source = try flatSource()
+        let stacked = try temporaryVMDirectory()
+        try source.cloneAsStackedBase(to: stacked, generateMAC: false)
+
+        let contentDigest = try Digest.hash(source.diskURL)
+        let contentStore = try ContentStore()
+        let archivedOverlayDigest = try Digest.hash(stacked.overlayURL)
+        let archiveURL = try temporaryDirectory().appendingPathComponent("stacked.tvm")
+        try stacked.exportToArchive(path: archiveURL.path)
+
+        let cachedBaseURL = try XCTUnwrap(try contentStore.existingContentURL(for: contentDigest))
+        // Import must repair a corrupt cache entry from the valid archive
+        // instead of discarding the archive copy as an apparent cache hit.
+        try Data("corrupt".utf8).write(to: cachedBaseURL)
+        XCTAssertNil(try contentStore.existingContentURL(for: contentDigest))
+
+        let imported = try temporaryVMDirectory()
+        try imported.importFromArchive(path: archiveURL.path)
+
+        XCTAssertTrue(imported.isStackedVM)
+        XCTAssertEqual(try Digest.hash(imported.overlayURL), archivedOverlayDigest)
+        XCTAssertNotNil(try contentStore.existingContentURL(for: contentDigest))
+        _ = try imported.diskImageStack().makeAttachment()
+      }
+    }
+
+    func testStackedOCIArchiveSurvivesConcurrentRecordDeletion() throws {
+      try withTemporaryTartHome {
+        let source = try flatSource()
+        let stacked = try temporaryVMDirectory()
+        try source.cloneAsStackedBase(to: stacked, generateMAC: false)
+
+        let manifest = try OCIManifest(fromJSON: Data(contentsOf: stacked.manifestURL))
+        let storage = try VMStorageOCI()
+        let record = try storage.create(RemoteName(
+          host: "example.com",
+          namespace: "org/image",
+          reference: Reference(digest: try manifest.digest())
+        ))
+        try FileManager.default.copyItem(at: stacked.configURL, to: record.configURL)
+        try FileManager.default.copyItem(at: stacked.nvramURL, to: record.nvramURL)
+        try FileManager.default.copyItem(at: stacked.manifestURL, to: record.manifestURL)
+        XCTAssertTrue(record.isStackedCachedImage)
+
+        let archiveURL = try temporaryDirectory().appendingPathComponent("stacked-race.tvm")
+        let contentStore = try ContentStore()
+        let lockHeld = DispatchSemaphore(value: 0)
+        let releaseLock = DispatchSemaphore(value: 0)
+        let exportStarted = DispatchSemaphore(value: 0)
+        let exportFinished = DispatchSemaphore(value: 0)
+        let deletionStarted = DispatchSemaphore(value: 0)
+        let deletionFinished = DispatchSemaphore(value: 0)
+
+        DispatchQueue.global().async {
+          try? contentStore.withPruneLock {
+            lockHeld.signal()
+            releaseLock.wait()
+          }
+        }
+        XCTAssertEqual(lockHeld.wait(timeout: .now() + 1), .success)
+
+        // Queue export first so it is the next prune-lock waiter, then queue
+        // deletion behind it. Export must finish staging everything it needs
+        // before deletion can remove the source cached image.
+        DispatchQueue.global().async {
+          exportStarted.signal()
+          try? record.exportToArchive(path: archiveURL.path)
+          exportFinished.signal()
+        }
+        XCTAssertEqual(exportStarted.wait(timeout: .now() + 1), .success)
+        Thread.sleep(forTimeInterval: 0.1)
+
+        DispatchQueue.global().async {
+          deletionStarted.signal()
+          try? record.delete()
+          deletionFinished.signal()
+        }
+        XCTAssertEqual(deletionStarted.wait(timeout: .now() + 1), .success)
+        XCTAssertEqual(exportFinished.wait(timeout: .now() + 0.1), .timedOut)
+        XCTAssertEqual(deletionFinished.wait(timeout: .now() + 0.1), .timedOut)
+
+        releaseLock.signal()
+        XCTAssertEqual(exportFinished.wait(timeout: .now() + 5), .success)
+        XCTAssertEqual(deletionFinished.wait(timeout: .now() + 5), .success)
+        XCTAssertFalse(FileManager.default.fileExists(atPath: record.baseURL.path))
+
+        let imported = try temporaryVMDirectory()
+        try imported.importFromArchive(path: archiveURL.path)
+        XCTAssertTrue(imported.isStackedVM)
+        _ = try imported.diskImageStack().makeAttachment()
+      }
     }
 
     func testResolvesPublishedOverlayFromManifestAndCache() throws {
@@ -168,6 +299,28 @@ import XCTest
       return try ContentStore(baseURL: url)
     }
 
+    private func temporaryEntries() throws -> [URL] {
+      try FileManager.default.contentsOfDirectory(
+        at: Config().tartTmpDir,
+        includingPropertiesForKeys: nil
+      )
+    }
+
+    private func withTemporaryTartHome(_ body: () throws -> Void) throws {
+      let home = try temporaryDirectory()
+      let previousHome = ProcessInfo.processInfo.environment["TART_HOME"]
+      setenv("TART_HOME", home.path, 1)
+      defer {
+        if let previousHome {
+          setenv("TART_HOME", previousHome, 1)
+        } else {
+          unsetenv("TART_HOME")
+        }
+      }
+
+      try body()
+    }
+
     private func temporaryVMDirectory() throws -> VMDirectory {
       VMDirectory(baseURL: try temporaryDirectory())
     }
diff --git a/Tests/TartTests/VMDirectoryLayoutTests.swift b/Tests/TartTests/VMDirectoryLayoutTests.swift
index 364a5fd..5f8fcce 100644
--- a/Tests/TartTests/VMDirectoryLayoutTests.swift
+++ b/Tests/TartTests/VMDirectoryLayoutTests.swift
@@ -75,29 +75,39 @@ final class VMDirectoryLayoutTests: XCTestCase {
     )
   }
 
-  func testStackedExportIsRejected() throws {
+  func testStackedCachedImageAccountingUsesManifestBlockLayout() throws {
     let vmDir = try temporaryVMDirectory()
-    try touch(vmDir.configURL)
-    try touch(vmDir.nvramURL)
-    try touch(vmDir.manifestURL)
-    try touch(vmDir.overlayURL)
-    let archiveURL = vmDir.baseURL.appendingPathComponent("export.tvm")
 
-    XCTAssertThrowsError(try vmDir.exportToArchive(path: archiveURL.path)) { error in
-      guard case RuntimeError.ExportFailed(let message) = error else {
-        return XCTFail("unexpected error: \(error)")
-      }
-      XCTAssertEqual(message, "exporting stacked VMs is not supported yet")
+    try Data("config".utf8).write(to: vmDir.configURL)
+    try Data("nvram".utf8).write(to: vmDir.nvramURL)
+    try stackedManifest(blockSize: 512, blockCount: 8).toJSON().write(to: vmDir.manifestURL)
+
+    XCTAssertEqual(
+      try vmDir.sizeBytes(),
+      try vmDir.configURL.sizeBytes() + vmDir.nvramURL.sizeBytes()
+    )
+    XCTAssertEqual(
+      try vmDir.allocatedSizeBytes(),
+      try vmDir.configURL.allocatedSizeBytes() + vmDir.nvramURL.allocatedSizeBytes()
+    )
+    XCTAssertEqual(try vmDir.diskSizeBytes(), 4096)
+  }
+
+  func testStackedArchiveRequiresMacOS27() throws {
+    if #available(macOS 27.0, *) {
+      throw XCTSkip("macOS 26 compatibility test")
     }
-    XCTAssertFalse(FileManager.default.fileExists(atPath: archiveURL.path))
 
-    try FileManager.default.removeItem(at: vmDir.overlayURL)
-    XCTAssertTrue(vmDir.isStackedCachedImage)
+    let vmDir = try temporaryVMDirectory()
+    try Data("config".utf8).write(to: vmDir.configURL)
+    try Data("nvram".utf8).write(to: vmDir.nvramURL)
+    try Data("overlay".utf8).write(to: vmDir.overlayURL)
+    try stackedManifest(blockSize: 512, blockCount: 8).toJSON().write(to: vmDir.manifestURL)
+    let archiveURL = try temporaryVMDirectory().baseURL.appendingPathComponent("stacked.tvm")
     XCTAssertThrowsError(try vmDir.exportToArchive(path: archiveURL.path)) { error in
-      guard case RuntimeError.ExportFailed(let message) = error else {
+      guard case DiskImageStackError.unavailable = error else {
         return XCTFail("unexpected error: \(error)")
       }
-      XCTAssertEqual(message, "exporting stacked VMs is not supported yet")
     }
     XCTAssertFalse(FileManager.default.fileExists(atPath: archiveURL.path))
   }
diff --git a/Tests/TartTests/VMStorageOCITests.swift b/Tests/TartTests/VMStorageOCITests.swift
index 5a4a8fc..af3fbba 100644
--- a/Tests/TartTests/VMStorageOCITests.swift
+++ b/Tests/TartTests/VMStorageOCITests.swift
@@ -85,10 +85,10 @@ final class VMStorageOCITests: XCTestCase {
     }
   }
 
-  func testStackedCacheHitRequiresVerifiedContentAndSizesMissingFiles() throws {
+  func testStackedCacheHitRequiresExpectedContentSizes() throws {
     try withTemporaryTartHome {
-      let baseData = Data("base".utf8)
-      let overlayData = Data("overlay".utf8)
+      let baseData = Data(repeating: 0x41, count: 10)
+      let overlayData = Data(repeating: 0x42, count: 20)
       let baseDigest = Digest.hash(baseData)
       let overlayDigest = Digest.hash(overlayData)
       let manifest = try stackedManifest(
@@ -216,6 +216,518 @@ final class VMStorageOCITests: XCTestCase {
     }
   }
 
+  func testGCPrunesOnlyUnreferencedContent() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let referenced = try installContent(Data("referenced".utf8), into: contentStore)
+      let unreferenced = try installContent(Data("unreferenced".utf8), into: contentStore)
+
+      let stacked = try VMStorageLocal().create("stacked")
+      try config().save(toURL: stacked.configURL)
+      XCTAssertTrue(FileManager.default.createFile(atPath: stacked.nvramURL.path, contents: Data()))
+      XCTAssertTrue(FileManager.default.createFile(atPath: stacked.overlayURL.path, contents: Data()))
+      try pinnedBaseManifest(contentDigest: referenced.digest).toJSON().write(to: stacked.manifestURL)
+
+      try VMStorageOCI().gc()
+
+      XCTAssertTrue(FileManager.default.fileExists(atPath: referenced.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: unreferenced.url.path))
+    }
+  }
+
+  func testGCPrunesContentWithoutOCIStorageDirectory() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let unreferenced = try installContent(Data("unreferenced".utf8), into: contentStore)
+      let storage = try VMStorageOCI()
+
+      XCTAssertFalse(FileManager.default.fileExists(atPath: storage.baseURL.path))
+
+      try storage.gc()
+
+      XCTAssertFalse(FileManager.default.fileExists(atPath: unreferenced.url.path))
+    }
+  }
+
+  func testGCDoesNotPruneContentReferencedByInProgressManifest() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let referenced = try installContent(Data("in-progress".utf8), into: contentStore)
+      let temporaryVMDir = try VMDirectory.temporary()
+
+      // Pull and clone publish the manifest before config, NVRAM, or a
+      // writable overlay necessarily exist.
+      try pinnedBaseManifest(contentDigest: referenced.digest).toJSON().write(to: temporaryVMDir.manifestURL)
+
+      try VMStorageOCI().gc()
+
+      XCTAssertTrue(FileManager.default.fileExists(atPath: referenced.url.path))
+    }
+  }
+
+  func testStalePrunableDoesNotDeleteNewlyReferencedContent() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let content = try installContent(Data("new-reference".utf8), into: contentStore)
+      let storage = try VMStorageOCI()
+      let candidate = try XCTUnwrap(storage.prunables().first {
+        $0.url.resolvingSymlinksInPath() == content.url.resolvingSymlinksInPath()
+      })
+
+      // Simulate a clone or pull publishing its manifest after prune built
+      // the candidate list but before deletion starts.
+      let temporaryVMDir = try VMDirectory.temporary()
+      try contentStore.withPruneLock {
+        try pinnedBaseManifest(contentDigest: content.digest).toJSON().write(to: temporaryVMDir.manifestURL)
+      }
+
+      try candidate.delete()
+
+      XCTAssertTrue(FileManager.default.fileExists(atPath: content.url.path))
+    }
+  }
+
+  func testVMDirectoryDeletionOfStackedOCIRecordWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let storage = try VMStorageOCI()
+      let record = try createRecord(for: stackedManifest(), in: storage)
+
+      try assertDeletionWaitsForPruneLock(record: record) {
+        try record.delete()
+      }
+    }
+  }
+
+  func testStorageDeletionOfStackedOCIRecordWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let storage = try VMStorageOCI()
+      let manifest = try stackedManifest()
+      let name = try digestName(for: manifest)
+      let record = try createRecord(for: manifest, in: storage)
+
+      try assertDeletionWaitsForPruneLock(record: record) {
+        try storage.delete(name)
+      }
+    }
+  }
+
+  func testStorageDeletionOfIncompleteManifestRecordWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let storage = try VMStorageOCI()
+      let name = RemoteName(
+        host: "example.com",
+        namespace: "org/image",
+        reference: Reference(digest: "sha256:incomplete")
+      )
+      let record = try storage.create(name)
+      try Data("{}".utf8).write(to: record.manifestURL)
+
+      try assertDeletionWaitsForPruneLock(record: record) {
+        try storage.delete(name)
+      }
+    }
+  }
+
+  func testVMDirectoryDeletionOfUnpublishedRecordWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let storage = try VMStorageOCI()
+      let record = try storage.create(RemoteName(
+        host: "example.com",
+        namespace: "org/image",
+        reference: Reference(digest: "sha256:unpublished")
+      ))
+
+      try assertDeletionWaitsForPruneLock(record: record) {
+        try record.removeFromDisk()
+      }
+    }
+  }
+
+  func testTagReplacementWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let storage = try VMStorageOCI()
+      let firstManifest = try stackedManifest(baseContentDigest: "sha256:first")
+      let secondManifest = try stackedManifest(baseContentDigest: "sha256:second")
+      let firstName = try digestName(for: firstManifest)
+      let secondName = try digestName(for: secondManifest)
+      _ = try createRecord(for: firstManifest, in: storage)
+      _ = try createRecord(for: secondManifest, in: storage)
+      let tagName = RemoteName(
+        host: secondName.host,
+        namespace: secondName.namespace,
+        reference: Reference(tag: "latest")
+      )
+
+      let contentStore = try ContentStore()
+      let lockHeld = DispatchSemaphore(value: 0)
+      let releaseLock = DispatchSemaphore(value: 0)
+      let replacementStarted = DispatchSemaphore(value: 0)
+      let replacementFinished = DispatchSemaphore(value: 0)
+
+      DispatchQueue.global().async {
+        try? contentStore.withPruneLock {
+          lockHeld.signal()
+          releaseLock.wait()
+        }
+      }
+      XCTAssertEqual(lockHeld.wait(timeout: .now() + 1), .success)
+
+      DispatchQueue.global().async {
+        replacementStarted.signal()
+        try? storage.link(from: tagName, to: secondName)
+        replacementFinished.signal()
+      }
+      XCTAssertEqual(replacementStarted.wait(timeout: .now() + 1), .success)
+      XCTAssertEqual(replacementFinished.wait(timeout: .now() + 0.1), .timedOut)
+
+      releaseLock.signal()
+      XCTAssertEqual(replacementFinished.wait(timeout: .now() + 1), .success)
+      XCTAssertTrue(storage.linked(from: tagName, to: secondName))
+      XCTAssertFalse(storage.linked(from: tagName, to: firstName))
+    }
+  }
+
+  func testGCDeletionOfStackedOCIRecordWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let storage = try VMStorageOCI()
+      let record = try createRecord(for: stackedManifest(), in: storage)
+
+      try assertDeletionWaitsForPruneLock(record: record) {
+        try storage.gc()
+      }
+    }
+  }
+
+  func testTemporaryManifestGCWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let temporaryVMDir = try VMDirectory.temporary()
+      try stackedManifest().toJSON().write(to: temporaryVMDir.manifestURL)
+
+      try assertDeletionWaitsForPruneLock(record: temporaryVMDir) {
+        try Config().gc()
+      }
+    }
+  }
+
+  func testLockedTemporaryDirectorySurvivesGarbageCollection() throws {
+    try withTemporaryTartHome {
+      let temporaryVMDir = try VMDirectory.temporary()
+      let lock = try FileLock(lockURL: temporaryVMDir.baseURL)
+      try lock.lock()
+
+      try Config().gc()
+      XCTAssertTrue(FileManager.default.fileExists(atPath: temporaryVMDir.baseURL.path))
+
+      try lock.unlock()
+      try Config().gc()
+      XCTAssertFalse(FileManager.default.fileExists(atPath: temporaryVMDir.baseURL.path))
+    }
+  }
+
+  func testMovingStackedOCIRecordWaitsForPruneLock() throws {
+    try withTemporaryTartHome {
+      let storage = try VMStorageOCI()
+      let source = try temporaryVMDirectory()
+      let manifest = try stackedManifest()
+      let name = try digestName(for: manifest)
+      try config().save(toURL: source.configURL)
+      XCTAssertTrue(FileManager.default.createFile(atPath: source.nvramURL.path, contents: Data()))
+      try manifest.toJSON().write(to: source.manifestURL)
+
+      let contentStore = try ContentStore()
+      let lockHeld = DispatchSemaphore(value: 0)
+      let releaseLock = DispatchSemaphore(value: 0)
+      let moveStarted = DispatchSemaphore(value: 0)
+      let moveFinished = DispatchSemaphore(value: 0)
+
+      DispatchQueue.global().async {
+        try? contentStore.withPruneLock {
+          lockHeld.signal()
+          releaseLock.wait()
+        }
+      }
+      XCTAssertEqual(lockHeld.wait(timeout: .now() + 1), .success)
+
+      DispatchQueue.global().async {
+        moveStarted.signal()
+        try? storage.move(name, from: source)
+        moveFinished.signal()
+      }
+      XCTAssertEqual(moveStarted.wait(timeout: .now() + 1), .success)
+      XCTAssertEqual(moveFinished.wait(timeout: .now() + 0.1), .timedOut)
+      XCTAssertTrue(FileManager.default.fileExists(atPath: source.baseURL.path))
+
+      releaseLock.signal()
+      XCTAssertEqual(moveFinished.wait(timeout: .now() + 1), .success)
+      XCTAssertFalse(FileManager.default.fileExists(atPath: source.baseURL.path))
+      XCTAssertTrue(try storage.open(name).isStackedCachedImage)
+    }
+  }
+
+  func testPruningLastStackedOCIRecordReclaimsItsContent() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let baseContent = try installContent(Data("record-only-base".utf8), into: contentStore)
+      let overlayContent = try installContent(Data("record-only-overlay".utf8), into: contentStore)
+      let manifest = try stackedManifest(
+        baseContentDigest: baseContent.digest,
+        overlayContentDigest: overlayContent.digest,
+        baseUncompressedSize: UInt64(try baseContent.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try overlayContent.url.sizeBytes())
+      )
+      let name = try digestName(for: manifest)
+      let storage = try VMStorageOCI()
+      let record = try storage.create(name)
+      try config().save(toURL: record.configURL)
+      XCTAssertTrue(FileManager.default.createFile(atPath: record.nvramURL.path, contents: Data()))
+      try manifest.toJSON().write(to: record.manifestURL)
+
+      let candidate = try XCTUnwrap(storage.prunables().first {
+        $0.url.lastPathComponent == record.url.lastPathComponent
+      })
+      XCTAssertGreaterThanOrEqual(
+        try candidate.allocatedSizeBytes(),
+        try baseContent.url.allocatedSizeBytes() + overlayContent.url.allocatedSizeBytes()
+      )
+
+      // The record itself fits in this budget, so pruning only succeeds if it
+      // accounts for the immutable content released with the final reference.
+      try Prune.pruneSpaceBudget(
+        prunableStorages: [storage],
+        spaceBudgetBytes: UInt64(try record.allocatedSizeBytes())
+      )
+
+      XCTAssertFalse(FileManager.default.fileExists(atPath: record.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: baseContent.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: overlayContent.url.path))
+    }
+  }
+
+  func testMalformedStackedOCIRecordDoesNotBlockPruning() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let baseContent = try installContent(Data("valid-base".utf8), into: contentStore)
+      let overlayContent = try installContent(Data("valid-overlay".utf8), into: contentStore)
+      let storage = try VMStorageOCI()
+      let validRecord = try createRecord(for: stackedManifest(
+        baseContentDigest: baseContent.digest,
+        overlayContentDigest: overlayContent.digest,
+        baseUncompressedSize: UInt64(try baseContent.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try overlayContent.url.sizeBytes())
+      ), in: storage)
+
+      let malformedRecord = try storage.create(RemoteName(
+        host: "example.com",
+        namespace: "org/image",
+        reference: Reference(digest: "sha256:malformed")
+      ))
+      try config().save(toURL: malformedRecord.configURL)
+      XCTAssertTrue(FileManager.default.createFile(atPath: malformedRecord.nvramURL.path, contents: Data()))
+      try Data("{".utf8).write(to: malformedRecord.manifestURL)
+
+      XCTAssertNoThrow(try storage.prunables())
+      try Prune.pruneSpaceBudget(prunableStorages: [storage], spaceBudgetBytes: 0)
+
+      XCTAssertFalse(FileManager.default.fileExists(atPath: validRecord.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: malformedRecord.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: baseContent.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: overlayContent.url.path))
+    }
+  }
+
+  func testPruningOneStackedOCIRecordPreservesSharedContent() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let baseContent = try installContent(Data("shared-base".utf8), into: contentStore)
+      let firstOverlay = try installContent(Data("first-overlay".utf8), into: contentStore)
+      let secondOverlay = try installContent(Data("second-overlay".utf8), into: contentStore)
+      let storage = try VMStorageOCI()
+
+      let firstManifest = try stackedManifest(
+        baseContentDigest: baseContent.digest,
+        overlayContentDigest: firstOverlay.digest,
+        baseUncompressedSize: UInt64(try baseContent.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try firstOverlay.url.sizeBytes())
+      )
+      let secondManifest = try stackedManifest(
+        baseContentDigest: baseContent.digest,
+        overlayContentDigest: secondOverlay.digest,
+        baseUncompressedSize: UInt64(try baseContent.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try secondOverlay.url.sizeBytes())
+      )
+      let firstRecord = try createRecord(for: firstManifest, in: storage)
+      let secondRecord = try createRecord(for: secondManifest, in: storage)
+
+      let firstCandidate = try XCTUnwrap(storage.prunables().first {
+        $0.url.lastPathComponent == firstRecord.url.lastPathComponent
+      })
+      try firstCandidate.delete()
+
+      XCTAssertTrue(FileManager.default.fileExists(atPath: baseContent.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: firstOverlay.url.path))
+      XCTAssertTrue(FileManager.default.fileExists(atPath: secondOverlay.url.path))
+
+      let secondCandidate = try XCTUnwrap(storage.prunables().first {
+        $0.url.lastPathComponent == secondRecord.url.lastPathComponent
+      })
+      try secondCandidate.delete()
+
+      XCTAssertFalse(FileManager.default.fileExists(atPath: baseContent.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: secondOverlay.url.path))
+    }
+  }
+
+  func testSpaceBudgetRecomputesSharedContentAfterOwnerDeletion() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let baseContent = try installContent(Data("shared-base".utf8), into: contentStore)
+      let firstOverlay = try installContent(Data("first-overlay".utf8), into: contentStore)
+      let secondOverlay = try installContent(Data("second-overlay".utf8), into: contentStore)
+      let storage = try VMStorageOCI()
+
+      let firstManifest = try stackedManifest(
+        baseContentDigest: baseContent.digest,
+        overlayContentDigest: firstOverlay.digest,
+        baseUncompressedSize: UInt64(try baseContent.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try firstOverlay.url.sizeBytes())
+      )
+      let secondManifest = try stackedManifest(
+        baseContentDigest: baseContent.digest,
+        overlayContentDigest: secondOverlay.digest,
+        baseUncompressedSize: UInt64(try baseContent.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try secondOverlay.url.sizeBytes())
+      )
+      let olderRecord = try createRecord(for: firstManifest, in: storage)
+      let newerRecord = try createRecord(for: secondManifest, in: storage)
+      try olderRecord.url.updateAccessDate(Date(timeIntervalSince1970: 1))
+      try newerRecord.url.updateAccessDate(Date(timeIntervalSince1970: 2))
+
+      let olderCandidate = try XCTUnwrap(storage.prunables().first {
+        $0.url.lastPathComponent == olderRecord.url.lastPathComponent
+      })
+      let budget = UInt64(try olderCandidate.allocatedSizeBytes())
+
+      // On the first pass the newer record owns the shared base and is
+      // selected for deletion, while the older record fits this budget.
+      // Recomputing must then charge the surviving record for the base and
+      // prune it too.
+      try Prune.pruneSpaceBudget(
+        prunableStorages: [storage],
+        spaceBudgetBytes: budget
+      )
+
+      XCTAssertFalse(FileManager.default.fileExists(atPath: olderRecord.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: newerRecord.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: baseContent.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: firstOverlay.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: secondOverlay.url.path))
+    }
+  }
+
+  func testSpaceBudgetRecomputesBeforeDeletingAnotherCandidate() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let sharedBase = try installContent(Data(repeating: 0x41, count: 128 * 1024), into: contentStore)
+      let newestOverlay = try installContent(Data("newest-overlay".utf8), into: contentStore)
+      let middleOverlay = try installContent(Data("middle-overlay".utf8), into: contentStore)
+      let retainedBase = try installContent(Data(repeating: 0x42, count: 32 * 1024), into: contentStore)
+      let retainedOverlay = try installContent(Data("retained-overlay".utf8), into: contentStore)
+      let storage = try VMStorageOCI()
+
+      let newest = try createRecord(for: stackedManifest(
+        baseContentDigest: sharedBase.digest,
+        overlayContentDigest: newestOverlay.digest,
+        baseUncompressedSize: UInt64(try sharedBase.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try newestOverlay.url.sizeBytes())
+      ), in: storage)
+      let middle = try createRecord(for: stackedManifest(
+        baseContentDigest: sharedBase.digest,
+        overlayContentDigest: middleOverlay.digest,
+        baseUncompressedSize: UInt64(try sharedBase.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try middleOverlay.url.sizeBytes())
+      ), in: storage)
+      let retained = try createRecord(for: stackedManifest(
+        baseContentDigest: retainedBase.digest,
+        overlayContentDigest: retainedOverlay.digest,
+        baseUncompressedSize: UInt64(try retainedBase.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try retainedOverlay.url.sizeBytes())
+      ), in: storage)
+      try newest.url.updateAccessDate(Date(timeIntervalSince1970: 3))
+      try middle.url.updateAccessDate(Date(timeIntervalSince1970: 2))
+      try retained.url.updateAccessDate(Date(timeIntervalSince1970: 1))
+
+      let retainedCandidate = try XCTUnwrap(storage.prunables().first {
+        $0.url.lastPathComponent == retained.url.lastPathComponent
+      })
+
+      // Initially the newest record owns the shared base and is too large.
+      // The middle record appears small enough to retain, making the oldest
+      // unrelated record look like a second deletion candidate. After the
+      // first deletion, ownership moves to the middle record; recomputing
+      // before selecting again must delete it and preserve the unrelated one.
+      try Prune.pruneSpaceBudget(
+        prunableStorages: [storage],
+        spaceBudgetBytes: UInt64(try retainedCandidate.allocatedSizeBytes())
+      )
+
+      XCTAssertFalse(FileManager.default.fileExists(atPath: newest.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: middle.url.path))
+      XCTAssertTrue(FileManager.default.fileExists(atPath: retained.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: sharedBase.url.path))
+      XCTAssertTrue(FileManager.default.fileExists(atPath: retainedBase.url.path))
+      XCTAssertTrue(FileManager.default.fileExists(atPath: retainedOverlay.url.path))
+    }
+  }
+
+  func testAutomaticReclaimRecomputesSharedContentAfterOwnerDeletion() throws {
+    try withTemporaryTartHome {
+      let contentStore = try ContentStore()
+      let sharedBase = try installContent(Data(repeating: 0x41, count: 128 * 1024), into: contentStore)
+      let initiatorOverlay = try installContent(Data("initiator-overlay".utf8), into: contentStore)
+      let ownerOverlay = try installContent(Data("owner-overlay".utf8), into: contentStore)
+      let unrelatedBase = try installContent(Data("unrelated-base".utf8), into: contentStore)
+      let unrelatedOverlay = try installContent(Data("unrelated-overlay".utf8), into: contentStore)
+      let storage = try VMStorageOCI()
+
+      let initiatorManifest = try stackedManifest(
+        baseContentDigest: sharedBase.digest,
+        overlayContentDigest: initiatorOverlay.digest,
+        baseUncompressedSize: UInt64(try sharedBase.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try initiatorOverlay.url.sizeBytes())
+      )
+      let ownerManifest = try stackedManifest(
+        baseContentDigest: sharedBase.digest,
+        overlayContentDigest: ownerOverlay.digest,
+        baseUncompressedSize: UInt64(try sharedBase.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try ownerOverlay.url.sizeBytes())
+      )
+      let unrelatedManifest = try stackedManifest(
+        baseContentDigest: unrelatedBase.digest,
+        overlayContentDigest: unrelatedOverlay.digest,
+        baseUncompressedSize: UInt64(try unrelatedBase.url.sizeBytes()),
+        overlayUncompressedSize: UInt64(try unrelatedOverlay.url.sizeBytes())
+      )
+      let initiator = try createRecord(for: initiatorManifest, in: storage)
+      let owner = try createRecord(for: ownerManifest, in: storage)
+      let unrelated = try createRecord(for: unrelatedManifest, in: storage)
+      try initiator.url.updateAccessDate(Date(timeIntervalSince1970: 1))
+      try owner.url.updateAccessDate(Date(timeIntervalSince1970: 2))
+      try unrelated.url.updateAccessDate(Date(timeIntervalSince1970: 3))
+
+      let sharedBaseSize = UInt64(try sharedBase.url.allocatedSizeBytes())
+
+      // The owner is the first deletable record and is initially charged for
+      // the shared base. Deleting it cannot reclaim that base because the
+      // protected initiator still references it, so reclaim must continue.
+      try Prune.reclaimIfPossible(sharedBaseSize, initiator)
+
+      XCTAssertTrue(FileManager.default.fileExists(atPath: initiator.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: owner.url.path))
+      XCTAssertFalse(FileManager.default.fileExists(atPath: unrelated.url.path))
+      XCTAssertTrue(FileManager.default.fileExists(atPath: sharedBase.url.path))
+    }
+  }
+
   #if canImport(DiskImageKit)
     @available(macOS 27.0, *)
     func testPopulateStackedPushedImageCachesImmutableTopOverlay() throws {
@@ -265,6 +777,47 @@ final class VMStorageOCITests: XCTestCase {
     return vmDir
   }
 
+  private func createRecord(for manifest: OCIManifest, in storage: VMStorageOCI) throws -> VMDirectory {
+    let record = try storage.create(try digestName(for: manifest))
+    try config().save(toURL: record.configURL)
+    XCTAssertTrue(FileManager.default.createFile(atPath: record.nvramURL.path, contents: Data()))
+    try manifest.toJSON().write(to: record.manifestURL)
+
+    return record
+  }
+
+  private func assertDeletionWaitsForPruneLock(
+    record: VMDirectory,
+    deletion: @escaping () throws -> Void
+  ) throws {
+    let contentStore = try ContentStore()
+    let lockHeld = DispatchSemaphore(value: 0)
+    let releaseLock = DispatchSemaphore(value: 0)
+    let deletionStarted = DispatchSemaphore(value: 0)
+    let deletionFinished = DispatchSemaphore(value: 0)
+
+    DispatchQueue.global().async {
+      try? contentStore.withPruneLock {
+        lockHeld.signal()
+        releaseLock.wait()
+      }
+    }
+    XCTAssertEqual(lockHeld.wait(timeout: .now() + 1), .success)
+
+    DispatchQueue.global().async {
+      deletionStarted.signal()
+      try? deletion()
+      deletionFinished.signal()
+    }
+    XCTAssertEqual(deletionStarted.wait(timeout: .now() + 1), .success)
+    XCTAssertEqual(deletionFinished.wait(timeout: .now() + 0.1), .timedOut)
+    XCTAssertTrue(FileManager.default.fileExists(atPath: record.baseURL.path))
+
+    releaseLock.signal()
+    XCTAssertEqual(deletionFinished.wait(timeout: .now() + 1), .success)
+    XCTAssertFalse(FileManager.default.fileExists(atPath: record.baseURL.path))
+  }
+
   #if canImport(DiskImageKit)
     @available(macOS 27.0, *)
     private func diskImageSource() throws -> VMDirectory {
@@ -337,6 +890,34 @@ final class VMStorageOCITests: XCTestCase {
     _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
   }
 
+  private func pinnedBaseManifest(contentDigest: String) -> OCIManifest {
+    var disk = OCIManifestLayer(
+      mediaType: diskV2MediaType,
+      size: 1,
+      digest: "sha256:disk-transport",
+      uncompressedSize: 1,
+      uncompressedContentDigest: "sha256:disk-chunk"
+    )
+    disk.annotations?[diskFileContentDigestAnnotation] = contentDigest
+
+    return OCIManifest(
+      config: OCIManifestConfig(size: 1, digest: "sha256:oci-config"),
+      layers: [
+        OCIManifestLayer(mediaType: configMediaType, size: 1, digest: "sha256:config"),
+        disk,
+        OCIManifestLayer(mediaType: nvramMediaType, size: 1, digest: "sha256:nvram"),
+      ]
+    )
+  }
+
+  private func installContent(_ data: Data, into contentStore: ContentStore) throws -> (digest: String, url: URL) {
+    let digest = Digest.hash(data)
+    let temporaryURL = try contentStore.temporaryContentURL(for: digest)
+    try data.write(to: temporaryURL)
+
+    return (digest, try contentStore.install(temporaryURL, contentDigest: digest))
+  }
+
   private func digestName(for manifest: OCIManifest) throws -> RemoteName {
     RemoteName(
       host: "example.com",
diff --git a/benchmark/README.md b/benchmark/README.md
index a307058..83c8934 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -22,6 +22,20 @@ You can also enable the debugging output to diagnose issues:
 go run cmd/main.go fio --debug
 ```
 
+To compare a cold stacked OCI pull with the same pull after its immutable
+base has been prewarmed, provide a flat remote base image and a stacked image
+built from it:
+
+```shell
+go run cmd/main.go stacked-oci \
+  --base-image ghcr.io/example/macos-base:latest \
+  --image ghcr.io/example/macos-child:latest
+```
+
+The command uses disposable `TART_HOME` directories. Its prewarmed scenario
+keeps the VM created by `tart clone --stacked` alive while pulling the child, so
+the shared immutable base layer remains referenced and available for reuse.
+
 ## Results
 
 ### Mar 27, 2024
diff --git a/benchmark/internal/command/root.go b/benchmark/internal/command/root.go
index 77a33bc..e93d4bf 100644
--- a/benchmark/internal/command/root.go
+++ b/benchmark/internal/command/root.go
@@ -2,6 +2,7 @@ package command
 
 import (
 	"github.com/cirruslabs/tart/benchmark/internal/command/fio"
+	"github.com/cirruslabs/tart/benchmark/internal/command/stackedoci"
 	"github.com/cirruslabs/tart/benchmark/internal/command/xcode"
 	"github.com/spf13/cobra"
 )
@@ -15,6 +16,7 @@ func NewCommand() *cobra.Command {
 
 	cmd.AddCommand(
 		fio.NewCommand(),
+		stackedoci.NewCommand(),
 		xcode.NewCommand(),
 	)
 
diff --git a/benchmark/internal/command/stackedoci/stackedoci.go b/benchmark/internal/command/stackedoci/stackedoci.go
new file mode 100644
index 0000000..f087d08
--- /dev/null
+++ b/benchmark/internal/command/stackedoci/stackedoci.go
@@ -0,0 +1,143 @@
+package stackedoci
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+
+	"github.com/gosuri/uitable"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapio"
+)
+
+var (
+	debug        bool
+	baseImage    string
+	stackedImage string
+	insecure     bool
+	concurrency  uint
+)
+
+func NewCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "stacked-oci",
+		Short: "benchmark cold and prewarmed stacked OCI pulls",
+		Long: "Compare a cold stacked-image pull with a pull whose immutable base " +
+			"has already been materialized by tart clone --stacked. Both scenarios use " +
+			"disposable TART_HOME directories and leave the user's Tart home untouched.",
+		RunE: run,
+	}
+
+	cmd.Flags().BoolVar(&debug, "debug", false, "enable debug logging")
+	cmd.Flags().StringVar(&baseImage, "base-image", "", "remote flat OCI image used as the stacked image's base")
+	cmd.Flags().StringVar(&stackedImage, "image", "", "remote stacked OCI image to pull and clone")
+	cmd.Flags().BoolVar(&insecure, "insecure", false, "connect to the OCI registry via insecure HTTP")
+	cmd.Flags().UintVar(&concurrency, "concurrency", 4, "network concurrency passed to tart pull and clone")
+	_ = cmd.MarkFlagRequired("base-image")
+	_ = cmd.MarkFlagRequired("image")
+
+	return cmd
+}
+
+func run(cmd *cobra.Command, _ []string) error {
+	if concurrency < 1 {
+		return fmt.Errorf("concurrency cannot be less than 1")
+	}
+
+	config := zap.NewProductionConfig()
+	if debug {
+		config.Level = zap.NewAtomicLevelAt(zap.DebugLevel)
+	}
+	logger, err := config.Build()
+	if err != nil {
+		return err
+	}
+	defer func() { _ = logger.Sync() }()
+
+	coldHome, err := os.MkdirTemp("", "tart-stacked-oci-cold-*")
+	if err != nil {
+		return err
+	}
+	defer os.RemoveAll(coldHome)
+
+	warmHome, err := os.MkdirTemp("", "tart-stacked-oci-warm-*")
+	if err != nil {
+		return err
+	}
+	defer os.RemoveAll(warmHome)
+
+	table := uitable.New()
+	table.AddRow("Scenario", "Operation", "Time")
+
+	duration, err := timedTart(cmd.Context(), logger, coldHome, pullArguments(stackedImage)...)
+	if err != nil {
+		return fmt.Errorf("cold stacked pull failed: %w", err)
+	}
+	table.AddRow("cold", "pull stacked image", duration)
+
+	duration, err = timedTart(cmd.Context(), logger, coldHome, "clone", stackedImage, "cold-clone")
+	if err != nil {
+		return fmt.Errorf("cold stacked clone failed: %w", err)
+	}
+	table.AddRow("cold", "clone stacked image", duration)
+
+	duration, err = timedTart(cmd.Context(), logger, warmHome, cloneBaseArguments(baseImage)...)
+	if err != nil {
+		return fmt.Errorf("base prewarm failed: %w", err)
+	}
+	table.AddRow("prewarmed", "clone --stacked base image", duration)
+
+	duration, err = timedTart(cmd.Context(), logger, warmHome, pullArguments(stackedImage)...)
+	if err != nil {
+		return fmt.Errorf("prewarmed stacked pull failed: %w", err)
+	}
+	table.AddRow("prewarmed", "pull stacked image", duration)
+
+	duration, err = timedTart(cmd.Context(), logger, warmHome, "clone", stackedImage, "warm-clone")
+	if err != nil {
+		return fmt.Errorf("prewarmed stacked clone failed: %w", err)
+	}
+	table.AddRow("prewarmed", "clone stacked image", duration)
+
+	fmt.Println(table.String())
+	return nil
+}
+
+func pullArguments(image string) []string {
+	args := []string{"pull", "--concurrency", fmt.Sprint(concurrency)}
+	if insecure {
+		args = append(args, "--insecure")
+	}
+	return append(args, image)
+}
+
+func cloneBaseArguments(image string) []string {
+	args := []string{"clone", "--stacked", "--concurrency", fmt.Sprint(concurrency)}
+	if insecure {
+		args = append(args, "--insecure")
+	}
+	return append(args, image, "prewarmed-base")
+}
+
+func timedTart(
+	ctx context.Context,
+	logger *zap.Logger,
+	tartHome string,
+	args ...string,
+) (time.Duration, error) {
+	logger.Sugar().Debugf("TART_HOME=%s tart %s", tartHome, strings.Join(args, " "))
+	start := time.Now()
+
+	command := exec.CommandContext(ctx, "tart", args...)
+	command.Env = append(os.Environ(), "TART_HOME="+tartHome)
+	loggerWriter := &zapio.Writer{Log: logger, Level: zap.DebugLevel}
+	command.Stdout = loggerWriter
+	command.Stderr = loggerWriter
+
+	err := command.Run()
+	return time.Since(start).Round(time.Millisecond), err
+}
diff --git a/docs/faq.md b/docs/faq.md
index 77c254a..0d676fe 100644
--- a/docs/faq.md
+++ b/docs/faq.md
@@ -231,6 +231,28 @@ export TART_NO_AUTO_PRUNE=
 TART_NO_AUTO_PRUNE= tart pull ...
 ```
 
+## Stacked disk images
+
+On macOS 27 or newer, `tart clone --stacked` can create a VM from a remote,
+standalone macOS OCI image whose writes are stored in a private ASIF overlay while
+its source disk remains a shared read-only base:
+
+```shell
+tart clone --stacked ghcr.io/example/macos-base:latest macos-build
+```
+
+Running and pushing `macos-build` preserves that disk relationship. Pulling
+another image from the same lineage only downloads immutable disk files that
+are not already present in Tart's cache. For a stopped stacked VM,
+`tart set --disk-size` grows its private writable overlay without changing the
+base; a subsequent push records the new guest-visible disk size.
+
+`tart pull` can cache a stacked image without assembling its disk. Clone, run,
+import, and export require a Tart build with DiskImageKit support and macOS 27
+or newer. Existing standalone raw and ASIF images continue to work on older
+hosts. Keep published lineages shallow when possible: every additional parent
+overlay adds another ASIF file to validate and assemble at run time.
+
 ## Disk resizing
 
 Disk resizing works on most cloud-ready Linux distributions out-of-the box (e.g. Ubuntu Cloud Images have the `cloud-initramfs-growroot` package installed that runs on boot) and on the rest of the distributions by running the `growpart` or `resize2fs` commands.
diff --git a/docs/quick-start.md b/docs/quick-start.md
index eeed28a..c13beda 100644
--- a/docs/quick-start.md
+++ b/docs/quick-start.md
@@ -265,3 +265,15 @@ tart clone acme.io/remoteorg/name:latest my-local-vm-name
 ```
 
 If the specified image is not already present, this invocation calls the `tart pull` implicitly before cloning.
+
+### Creating a Stacked Disk
+
+On macOS 27 or newer, use `--stacked` to create a VM that keeps a remote standalone
+macOS OCI image as an immutable base and stores only its own writes separately:
+
+```bash
+tart clone --stacked acme.io/remoteorg/macos-base:latest my-local-vm-name
+```
+
+Pushing this VM preserves the disk relationship. Pulling another image from the
+same lineage reuses immutable disk files that are already in Tart's cache.
diff --git a/integration-tests/test_stacked_oci.py b/integration-tests/test_stacked_oci.py
new file mode 100644
index 0000000..7042a7c
--- /dev/null
+++ b/integration-tests/test_stacked_oci.py
@@ -0,0 +1,87 @@
+import os
+import platform
+import uuid
+from typing import Optional
+
+import pytest
+from paramiko.client import AutoAddPolicy, SSHClient
+
+
+def _macos_major_version() -> int:
+    version = platform.mac_ver()[0]
+    return int(version.split(".", maxsplit=1)[0]) if version else 0
+
+
+def _shutdown_vm(tart, vm_name: str, guest_command: Optional[str] = None) -> None:
+    tart_run_process = tart.run_async(["run", "--no-graphics", vm_name])
+
+    try:
+        stdout, _ = tart.run(["ip", vm_name, "--wait", "180"])
+        client = SSHClient()
+        client.set_missing_host_key_policy(AutoAddPolicy)
+        client.connect(stdout.strip(), username="admin", password="admin")
+        if guest_command:
+            _, command_stdout, command_stderr = client.exec_command(guest_command)
+            assert command_stdout.channel.recv_exit_status() == 0, command_stderr.read().decode()
+        client.exec_command("sudo shutdown -h now")
+        client.close()
+
+        tart_run_process.wait(timeout=180)
+        assert tart_run_process.returncode == 0
+    finally:
+        if tart_run_process.poll() is None:
+            tart_run_process.terminate()
+            tart_run_process.wait(timeout=30)
+
+
+@pytest.mark.skipif(
+    _macos_major_version() < 27,
+    reason="stacked disk images require DiskImageKit on macOS 27 or newer",
+)
+def test_stacked_oci_round_trip_and_boot(tart, docker_registry):
+    suffix = str(uuid.uuid4())
+    source_image = os.environ.get(
+        "TART_STACKED_INTEGRATION_BASE_IMAGE",
+        "ghcr.io/cirruslabs/macos-tahoe-base:latest",
+    )
+    source_clone_args = ["clone"]
+    if os.environ.get("TART_STACKED_INTEGRATION_BASE_INSECURE") == "1":
+        source_clone_args.append("--insecure")
+    standalone_vm = f"stacked-base-{suffix}"
+    stacked_vm = f"stacked-child-{suffix}"
+    restored_vm = f"stacked-restored-{suffix}"
+    base_remote = docker_registry.remote_name(f"stacked-base-{suffix}")
+    child_remote = docker_registry.remote_name(f"stacked-child-{suffix}")
+
+    try:
+        # Publish a normal Tart image, then start a stacked lineage from that
+        # remote image. The base remains a normal flat OCI image.
+        tart.run(source_clone_args + [source_image, standalone_vm])
+        tart.run(["push", "--insecure", standalone_vm, base_remote])
+        tart.run(["clone", "--insecure", "--stacked", base_remote, stacked_vm])
+
+        stacked_path = os.path.join(tart.home(), "vms", stacked_vm)
+        assert os.path.isfile(os.path.join(stacked_path, "overlay.asif"))
+        assert os.path.isfile(os.path.join(stacked_path, "manifest.json"))
+        assert not os.path.exists(os.path.join(stacked_path, "disk.img"))
+
+        # Boot once so the top ASIF overlay contains real guest writes, then
+        # exercise stacked push, pull, clone, assembly, and a second boot.
+        _shutdown_vm(tart, stacked_vm, "touch /Users/admin/stacked-round-trip-marker")
+        tart.run(["push", "--insecure", stacked_vm, child_remote])
+        tart.run(["delete", stacked_vm])
+        tart.run(["pull", "--insecure", child_remote])
+        tart.run(["clone", child_remote, restored_vm])
+
+        restored_path = os.path.join(tart.home(), "vms", restored_vm)
+        assert os.path.isfile(os.path.join(restored_path, "overlay.asif"))
+        assert os.path.isfile(os.path.join(restored_path, "manifest.json"))
+        assert not os.path.exists(os.path.join(restored_path, "disk.img"))
+
+        _shutdown_vm(tart, restored_vm, "test -f /Users/admin/stacked-round-trip-marker")
+    finally:
+        for vm_name in (restored_vm, stacked_vm, standalone_vm):
+            try:
+                tart.run(["delete", vm_name])
+            except Exception:
+                pass