diff --git a/src/iceberg/test/cherry_pick_operation_test.cc b/src/iceberg/test/cherry_pick_operation_test.cc
index cb55de9d0..bb0069552 100644
--- a/src/iceberg/test/cherry_pick_operation_test.cc
+++ b/src/iceberg/test/cherry_pick_operation_test.cc
@@ -31,6 +31,7 @@
 #include "iceberg/avro/avro_register.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/metrics/metrics_reporter.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/row/partition_values.h"
 #include "iceberg/schema.h"
@@ -653,6 +654,18 @@ TEST_F(CherryPickOperationTest, CherryPickReplayedAsFastForwardAfterRollback) {
   ICEBERG_UNWRAP_OR_FAIL(auto parent_id, CommitAppend(file_a_));
   ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_));
   ASSERT_THAT(CommitAppend(conflict_a_), IsOk());
+
+  class MockReporter : public MetricsReporter {
+   public:
+    MOCK_METHOD(Status, Report, (const MetricsReport&), (override));
+  };
+  auto reporter = std::make_shared<MockReporter>();
+  EXPECT_CALL(*reporter, Report(::testing::_)).Times(0);
+  ICEBERG_UNWRAP_OR_FAIL(
+      table_, Table::Make(table_->name(), table_->metadata(),
+                          std::string(table_->metadata_file_location()), file_io_,
+                          catalog_, table_->full_name(), reporter));
+
   ICEBERG_UNWRAP_OR_FAIL(auto transaction, StageCherrypick(staged_id));
   ICEBERG_UNWRAP_OR_FAIL(auto draft, transaction->current().Snapshot());
   ASSERT_NE(draft->snapshot_id, staged_id);
diff --git a/src/iceberg/test/fast_append_test.cc b/src/iceberg/test/fast_append_test.cc
index d75e1ef23..3efceaa78 100644
--- a/src/iceberg/test/fast_append_test.cc
+++ b/src/iceberg/test/fast_append_test.cc
@@ -429,7 +429,7 @@ TEST_F(FastAppendTest, TransactionApplyFailureCleansUpStagedFiles) {
   EXPECT_THAT(deleted_paths, ::testing::SizeIs(2U));
 }
 
-TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) {
+TEST_F(FastAppendTest, RebaseReusesCopiedAppendManifest) {
   table_->metadata()->format_version = 1;
   ASSERT_THAT(
       TableMetadataUtil::Write(*file_io_, std::string(table_->metadata_file_location()),
@@ -449,12 +449,20 @@ TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) {
       ASSERT_THAT(properties->Commit(), IsOk());
     }
     ICEBERG_UNWRAP_OR_FAIL(auto snapshot, txn->current().Snapshot());
+    EXPECT_THAT(snapshot->summary,
+                ::testing::Contains(::testing::Pair(
+                    std::string(SnapshotSummaryFields::kAddedRecords), "100")));
     SnapshotReader snapshot_reader(snapshot.get());
     ICEBERG_UNWRAP_OR_FAIL(auto manifests, snapshot_reader.DataManifests(file_io_));
     ASSERT_THAT(manifests, ::testing::SizeIs(1));
     const auto& manifest_path = manifests[0].manifest_path;
     EXPECT_NE(manifest_path, path);
-    EXPECT_THAT(attempt_manifests, ::testing::Not(::testing::Contains(manifest_path)));
+    if (!attempt_manifests.empty()) {
+      EXPECT_EQ(manifest_path, attempt_manifests.front());
+    }
+    // The copy is immutable, but each attempt needs a new manifest list.
+    EXPECT_THAT(attempt_lists,
+                ::testing::Not(::testing::Contains(snapshot->manifest_list)));
     attempt_manifests.push_back(manifest_path);
     attempt_lists.push_back(snapshot->manifest_list);
     ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadEntries(manifests[0]));
@@ -487,12 +495,13 @@ TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) {
   std::unordered_set<std::string> unique_deleted_paths(deleted_paths.begin(),
                                                        deleted_paths.end());
   EXPECT_THAT(unique_deleted_paths,
-              ::testing::UnorderedElementsAre(attempt_manifests[0], attempt_lists[0],
-                                              attempt_manifests[1], attempt_lists[1]));
+              ::testing::UnorderedElementsAre(attempt_lists[0], attempt_lists[1]));
+  EXPECT_THAT(deleted_paths, ::testing::SizeIs(2));
   EXPECT_THAT(deleted_paths, ::testing::Not(::testing::Contains(path)));
   EXPECT_THAT(deleted_paths,
               ::testing::Not(::testing::Contains(manifests[0].manifest_path)));
   EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk());
+  EXPECT_THAT(file_io_->ReadFile(manifests[0].manifest_path, std::nullopt), IsOk());
   EXPECT_THAT(file_io_->ReadFile(attempt_lists[2], std::nullopt), IsOk());
 }
 
@@ -1241,10 +1250,10 @@ TEST_F(FastAppendTest, ReplayBecomingNoopCleansStagingWithoutReporting) {
       EXPECT_EQ(committed->properties().configs().at("effective"), "value");
       EXPECT_EQ(txn->state(), TransactionState::kCommitted);
     }
-    EXPECT_EQ(deletes, 4);
+    EXPECT_EQ(deletes, 3);
     EXPECT_EQ(reports, 0);
     EXPECT_THAT(append->Commit(), IsError(ErrorKind::kValidationFailed));
-    EXPECT_EQ(deletes, 4);
+    EXPECT_EQ(deletes, 3);
   }
 }
 
diff --git a/src/iceberg/test/merging_snapshot_update_test.cc b/src/iceberg/test/merging_snapshot_update_test.cc
index d31a1eb29..64b1b8866 100644
--- a/src/iceberg/test/merging_snapshot_update_test.cc
+++ b/src/iceberg/test/merging_snapshot_update_test.cc
@@ -56,10 +56,12 @@
 #include "iceberg/test/update_test_base.h"
 #include "iceberg/transaction.h"
 #include "iceberg/update/delete_files.h"
+#include "iceberg/update/expire_snapshots.h"
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/merge_append.h"
 #include "iceberg/update/row_delta.h"
 #include "iceberg/update/snapshot_manager.h"
+#include "iceberg/update/update_partition_spec.h"
 #include "iceberg/update/update_properties.h"
 #include "iceberg/util/macros.h"
 
@@ -1380,7 +1382,137 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestCopiesManifestWithAssignedSnapshotI
   EXPECT_NE(data_manifests[0].manifest_path, path);
 }
 
-TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
+TEST_F(MergingSnapshotUpdateTest, ReplayRejectsRemovedPartitionSpec) {
+  for (bool fast_append : {true, false}) {
+    SCOPED_TRACE(fast_append);
+    RegisterTableFromResource(MetadataResource());
+    const auto original_metadata = table_->metadata();
+    ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+    if (fast_append) {
+      ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
+      append->AppendFile(file_a_);
+      ASSERT_THAT(append->Commit(), IsOk());
+    } else {
+      ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewMergeAppend());
+      append->AppendFile(file_a_);
+      ASSERT_THAT(append->Commit(), IsOk());
+    }
+
+    ICEBERG_UNWRAP_OR_FAIL(auto update_spec, table_->NewUpdatePartitionSpec());
+    update_spec->AddField("y");
+    ASSERT_THAT(update_spec->Commit(), IsOk());
+    ASSERT_THAT(table_->Refresh(), IsOk());
+    ICEBERG_UNWRAP_OR_FAIL(auto expire, table_->NewExpireSnapshots());
+    expire->CleanExpiredMetadata(true);
+    ASSERT_THAT(expire->Commit(), IsOk());
+    ASSERT_THAT(table_->Refresh(), IsOk());
+    ASSERT_THAT(table_->metadata()->PartitionSpecById(spec_->spec_id()),
+                IsError(ErrorKind::kNotFound));
+    ASSERT_NE(original_metadata, table_->metadata());
+
+    EXPECT_THAT(txn->Commit(), ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+                                                HasErrorMessage("Partition spec")));
+    EXPECT_EQ(txn->state(), TransactionState::kFailed);
+    EXPECT_TRUE(ReloadMetadata()->snapshots.empty());
+  }
+}
+
+TEST_F(MergingSnapshotUpdateTest, ReplayKeepsMergedDVReferencedOnlyByStagedSnapshot) {
+  UpgradeTableToV3();
+  CommitFileA();
+  const auto main_snapshot_id = table_->metadata()->current_snapshot_id;
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto first,
+      WriteDeletionVector(table_location_ + "/data/first.puffin", file_a_, {1}));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto second,
+      WriteDeletionVector(table_location_ + "/data/second.puffin", file_a_, {2, 3}));
+  FailCommits(1, [&](int attempt) {
+    if (attempt != 0) return;
+    ICEBERG_UNWRAP_OR_FAIL(auto concurrent, catalog_->LoadTable(table_ident_));
+    ICEBERG_UNWRAP_OR_FAIL(auto properties, concurrent->NewUpdateProperties());
+    properties->Set("replay-conflict", "true");
+    ASSERT_THAT(properties->Commit(), IsOk());
+  });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto delta, txn->NewRowDelta());
+  delta->AddDeletes(first).AddDeletes(second).StageOnly();
+  ASSERT_THAT(delta->Commit(), IsOk());
+  auto staged = txn->current().snapshots.back();
+  ICEBERG_UNWRAP_OR_FAIL(auto committed, txn->Commit());
+  ICEBERG_UNWRAP_OR_FAIL(auto retained,
+                         committed->metadata()->SnapshotById(staged->snapshot_id));
+  SnapshotReader reader(retained.get());
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, reader.DeleteManifests(file_io_));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto entries,
+      ReadAllEntries({manifests.begin(), manifests.end()}, *committed->metadata()));
+  ASSERT_THAT(entries, ::testing::SizeIs(1));
+  EXPECT_THAT(entries[0].data_file->file_path, ::testing::HasSubstr("merged-dvs-"));
+  ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(entries[0].data_file, file_io_));
+  EXPECT_EQ(positions.Cardinality(), 3);
+  EXPECT_EQ(committed->metadata()->current_snapshot_id, main_snapshot_id);
+  EXPECT_NE(retained->manifest_list, staged->manifest_list);
+  EXPECT_FALSE(file_io_->ReadFile(staged->manifest_list, std::nullopt).has_value());
+}
+
+TEST_F(MergingSnapshotUpdateTest, ReplayRefreshesInheritedManifestMetadata) {
+  UpgradeTableToV3();
+  ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties());
+  properties->Set(TableProperties::kManifestMinMergeCount.key(), "1");
+  ASSERT_THAT(properties->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto initial, table_->NewFastAppend());
+  initial->AppendFile(file_c_);
+  ASSERT_THAT(initial->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto base_snapshot, table_->current_snapshot());
+  ICEBERG_UNWRAP_OR_FAIL(auto manager, table_->NewSnapshotManager());
+  manager->CreateBranch("audit", base_snapshot->snapshot_id);
+  ASSERT_THAT(manager->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  FailCommits(2, [&](int attempt) {
+    if (attempt == 2) return;
+    ICEBERG_UNWRAP_OR_FAIL(auto concurrent, catalog_->LoadTable(table_ident_));
+    ICEBERG_UNWRAP_OR_FAIL(auto append, concurrent->NewFastAppend());
+    // First change the merge bin. Then change only its inherited metadata.
+    append->ToBranch(attempt == 0 ? "main" : "audit");
+    append->AppendFile(
+        MakeDataFile(std::format("/data/concurrent-{}.parquet", attempt), 4));
+    ASSERT_THAT(append->Commit(), IsOk());
+  });
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto first, txn->NewFastAppend());
+  first->AppendFile(file_a_);
+  ASSERT_THAT(first->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto staged, txn->current().Snapshot());
+  ICEBERG_UNWRAP_OR_FAIL(auto merge, txn->NewMergeAppend());
+  merge->AppendFile(file_b_);
+  ASSERT_THAT(merge->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto committed, txn->Commit());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto replayed_append,
+                         committed->metadata()->SnapshotById(staged->snapshot_id));
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, committed->current_snapshot());
+  SnapshotReader reader(snapshot.get());
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, reader.DataManifests(file_io_));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto entries,
+      ReadAllEntries({manifests.begin(), manifests.end()}, *committed->metadata()));
+  auto entry = std::ranges::find_if(entries, [&](const ManifestEntry& entry) {
+    return entry.data_file->file_path == file_a_->file_path;
+  });
+  ASSERT_NE(entry, entries.end());
+  EXPECT_EQ(entry->snapshot_id, replayed_append->snapshot_id);
+  EXPECT_EQ(entry->sequence_number, replayed_append->sequence_number);
+  EXPECT_EQ(entry->file_sequence_number, replayed_append->sequence_number);
+  EXPECT_EQ(entry->data_file->first_row_id, replayed_append->first_row_id);
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsReplaced), "3");
+}
+
+TEST_F(MergingSnapshotUpdateTest, AddManifestRetryReusesCopiedManifest) {
   auto path = table_location_ + "/metadata/retry-input.avro";
   ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_}));
   manifest.added_snapshot_id = 12345;
@@ -1390,12 +1522,20 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
   FailCommits(2, [&](int attempt) {
     SCOPED_TRACE(attempt);
     ICEBERG_UNWRAP_OR_FAIL(auto snapshot, op->StagedSnapshot());
+    EXPECT_THAT(snapshot->summary,
+                ::testing::Contains(::testing::Pair(
+                    std::string(SnapshotSummaryFields::kAddedRecords), "100")));
     SnapshotReader snapshot_reader(snapshot.get());
     ICEBERG_UNWRAP_OR_FAIL(auto manifests, snapshot_reader.DataManifests(file_io_));
     ASSERT_THAT(manifests, ::testing::SizeIs(1));
     const auto& manifest_path = manifests[0].manifest_path;
     EXPECT_NE(manifest_path, path);
-    EXPECT_THAT(attempt_manifests, ::testing::Not(::testing::Contains(manifest_path)));
+    if (!attempt_manifests.empty()) {
+      EXPECT_EQ(manifest_path, attempt_manifests.front());
+    }
+    // The copy is immutable, but each attempt needs a new manifest list.
+    EXPECT_THAT(attempt_lists,
+                ::testing::Not(::testing::Contains(snapshot->manifest_list)));
     attempt_manifests.push_back(manifest_path);
     attempt_lists.push_back(snapshot->manifest_list);
   });
@@ -1412,8 +1552,7 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
   ASSERT_THAT(attempt_lists, ::testing::SizeIs(3));
   EXPECT_EQ(manifests[0].manifest_path, attempt_manifests[2]);
   EXPECT_THAT(deleted,
-              ::testing::UnorderedElementsAre(attempt_manifests[0], attempt_lists[0],
-                                              attempt_manifests[1], attempt_lists[1]));
+              ::testing::UnorderedElementsAre(attempt_lists[0], attempt_lists[1]));
   EXPECT_THAT(deleted, ::testing::Not(::testing::Contains(path)));
   EXPECT_THAT(deleted, ::testing::Not(::testing::Contains(manifests[0].manifest_path)));
   EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk());
@@ -1424,6 +1563,33 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
   EXPECT_EQ(entries[0].data_file->file_path, file_a_->file_path);
 }
 
+TEST_F(MergingSnapshotUpdateTest, ResumesManifestCopiesAfterApplyFailure) {
+  ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
+  auto first_path = table_location_ + "/metadata/copy-first.avro";
+  auto second_path = table_location_ + "/metadata/copy-second.avro";
+  ICEBERG_UNWRAP_OR_FAIL(auto first, WriteManifest(first_path, {file_a_}));
+  ICEBERG_UNWRAP_OR_FAIL(auto second, WriteManifest(second_path, {file_b_}));
+  first.added_snapshot_id = second.added_snapshot_id = 12345;
+  ASSERT_THAT(op->AppendManifest(first), IsOk());
+  ASSERT_THAT(op->AppendManifest(second), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto contents, file_io_->ReadFile(second_path, std::nullopt));
+  ASSERT_THAT(file_io_->DeleteFile(second_path), IsOk());
+  EXPECT_FALSE(op->ApplyForTest(*table_->metadata(), nullptr).has_value());
+  ASSERT_THAT(file_io_->WriteFile(second_path, contents), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, op->CommitManifests());
+  ASSERT_THAT(manifests, ::testing::SizeIs(2));
+  ASSERT_THAT(table_->Refresh(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedRecords), "200");
+  ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadAllEntries(manifests, *table_->metadata()));
+  ASSERT_THAT(entries, ::testing::SizeIs(2));
+  EXPECT_THAT(
+      (std::vector{entries[0].data_file->file_path, entries[1].data_file->file_path}),
+      ::testing::UnorderedElementsAre(file_a_->file_path, file_b_->file_path));
+}
+
 TEST_F(MergingSnapshotUpdateTest, AddManifestRejectsManifestWithFirstRowId) {
   auto path = table_location_ + "/metadata/rowid.avro";
   ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_}));
diff --git a/src/iceberg/test/rewrite_manifests_test.cc b/src/iceberg/test/rewrite_manifests_test.cc
index 1bf4dd0d3..9c746ba75 100644
--- a/src/iceberg/test/rewrite_manifests_test.cc
+++ b/src/iceberg/test/rewrite_manifests_test.cc
@@ -2488,6 +2488,87 @@ TEST_P(RewriteManifestsTest, RewriteManifestsOnBranch) {
   EXPECT_EQ(std::ranges::distance(audit_manifests), 1);
 }
 
+TEST_P(RewriteManifestsTest, ReplaysReplacementAfterEarlierStagedAppend) {
+  SetCommitRetryProperties(2);
+  FailCommits(2, [&](int attempt) {
+    if (attempt < 2) {
+      ICEBERG_UNWRAP_OR_FAIL(auto concurrent, catalog_->LoadTable(table_ident_));
+      ICEBERG_UNWRAP_OR_FAIL(auto properties, concurrent->NewUpdateProperties());
+      properties->Set(std::format("replay-conflict-{}", attempt), "true");
+      ASSERT_THAT(properties->Commit(), IsOk());
+    }
+  });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
+  append->AppendFile(file_a_);
+  ASSERT_THAT(append->Commit(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto staged_snapshot, txn->current().Snapshot());
+  SnapshotReader staged_reader(staged_snapshot.get());
+  ICEBERG_UNWRAP_OR_FAIL(auto staged_manifests, staged_reader.DataManifests(file_io_));
+  ASSERT_THAT(staged_manifests, ::testing::SizeIs(1));
+  ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadManifestEntries(staged_manifests[0]));
+  ASSERT_THAT(entries, ::testing::SizeIs(1));
+  ICEBERG_UNWRAP_OR_FAIL(auto replacement,
+                         WriteExistingManifest("txn-replay-replacement", entries));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto rewrite, txn->NewRewriteManifests());
+  rewrite->DeleteManifest(staged_manifests[0]).AddManifest(replacement);
+  ASSERT_THAT(rewrite->Commit(), IsOk());
+
+  ASSERT_THAT(txn->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentManifests());
+  ASSERT_THAT(manifests, ::testing::SizeIs(1));
+  ICEBERG_UNWRAP_OR_FAIL(auto final_entries, ReadManifestEntries(manifests[0]));
+  ASSERT_THAT(final_entries, ::testing::SizeIs(1));
+  EXPECT_EQ(final_entries[0].data_file->file_path, file_a_->file_path);
+
+  // The append snapshot still needs the manifest replaced by the rewrite.
+  ICEBERG_UNWRAP_OR_FAIL(auto committed_append,
+                         table_->metadata()->SnapshotById(staged_snapshot->snapshot_id));
+  SnapshotReader committed_reader(committed_append.get());
+  EXPECT_THAT(committed_reader.DataManifests(file_io_), IsOk());
+  EXPECT_TRUE(FileExists(staged_manifests[0].manifest_path));
+}
+
+TEST_P(RewriteManifestsTest, RejectsReplacementAfterStagedManifestMetadataChanges) {
+  if (GetParam() == 1) {
+    GTEST_SKIP() << "Sequence and row-ID inheritance require format v2 or higher";
+  }
+  ASSERT_THAT(AppendFiles({file_c_}), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto base_snapshot, table_->current_snapshot());
+  ICEBERG_UNWRAP_OR_FAIL(auto manager, table_->NewSnapshotManager());
+  manager->CreateBranch("audit", base_snapshot->snapshot_id);
+  ASSERT_THAT(manager->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
+  append->AppendFile(file_a_);
+  ASSERT_THAT(append->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto staged, txn->current().Snapshot());
+  SnapshotReader reader(staged.get());
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, reader.DataManifests(file_io_));
+  ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadManifestEntries(manifests.front()));
+  ICEBERG_UNWRAP_OR_FAIL(auto replacement,
+                         WriteExistingManifest("stale-inheritance", entries));
+  ICEBERG_UNWRAP_OR_FAIL(auto rewrite, txn->NewRewriteManifests());
+  rewrite->DeleteManifest(manifests.front()).AddManifest(replacement);
+  ASSERT_THAT(rewrite->Commit(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto concurrent, table_->NewFastAppend());
+  concurrent->ToBranch("audit");
+  concurrent->AppendFile(file_d_);
+  ASSERT_THAT(concurrent->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+  auto result = txn->Commit();
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(result, HasErrorMessage("Deleted manifest metadata changed"));
+  EXPECT_EQ(txn->state(), TransactionState::kFailed);
+  EXPECT_TRUE(FileExists(replacement.manifest_path));
+}
+
 TEST_P(RewriteManifestsTest, RewriteDataManifestsPreservesDeletes) {
   if (GetParam() == 1) {
     GTEST_SKIP() << "Delete manifests require format version 2 or higher";
diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc
index eada4eb90..53b5f1bd3 100644
--- a/src/iceberg/transaction.cc
+++ b/src/iceberg/transaction.cc
@@ -21,11 +21,16 @@
 #include <format>
 #include <iterator>
 #include <memory>
+#include <mutex>
+#include <optional>
 #include <string>
+#include <unordered_set>
 
 #include "iceberg/catalog.h"
 #include "iceberg/location_provider.h"
 #include "iceberg/logging/log_macros.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/manifest/manifest_reader.h"
 #include "iceberg/schema.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/statistics_file.h"
@@ -99,6 +104,60 @@ std::string FormatCommittedSnapshots(
   return detail;
 }
 
+/// Collect files that must survive transaction finalization.
+///
+/// The set includes manifests and manifest lists from all snapshots created by
+/// this transaction. It also includes owned content files still referenced by
+/// retained delete manifests.
+///
+/// \param metadata The metadata after the transaction commit.
+/// \param io FileIO used to read manifest lists and delete manifests.
+/// \param snapshot_ids Snapshot IDs in the committed AddSnapshot changes.
+/// \param content_paths Owned files whose live references have not been found.
+Result<std::unordered_set<std::string>> CommittedTransactionFiles(
+    const TableMetadata& metadata, const std::shared_ptr<FileIO>& io,
+    std::unordered_set<int64_t> snapshot_ids,
+    std::unordered_set<std::string> content_paths) {
+  std::unordered_set<std::string> committed;
+  for (const auto& snapshot : metadata.snapshots) {
+    if (!snapshot || snapshot_ids.erase(snapshot->snapshot_id) == 0) {
+      continue;
+    }
+    committed.insert(snapshot->manifest_list);
+    SnapshotReader snapshot_reader(snapshot.get());
+    ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.Manifests(io));
+    for (const auto& manifest : manifests) {
+      if (!committed.insert(manifest.manifest_path).second || content_paths.empty() ||
+          manifest.content != ManifestContent::kDeletes) {
+        continue;
+      }
+      // A merged DV can be shared by several retained snapshots.
+      ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
+      ICEBERG_ASSIGN_OR_RAISE(auto spec,
+                              metadata.PartitionSpecById(manifest.partition_spec_id));
+      ICEBERG_ASSIGN_OR_RAISE(auto reader,
+                              ManifestReader::Make(manifest, io, schema, spec));
+      ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntriesStream());
+      while (!content_paths.empty()) {
+        ICEBERG_ASSIGN_OR_RAISE(auto entry, entries->Next());
+        if (!entry) {
+          break;
+        }
+        ICEBERG_CHECK(entry->data_file != nullptr,
+                      "Missing content file in committed manifest {}",
+                      manifest.manifest_path);
+        if (content_paths.erase(entry->data_file->file_path) != 0) {
+          committed.insert(entry->data_file->file_path);
+        }
+      }
+    }
+  }
+  // A catalog may omit snapshots. Never clean files with a partial keep-set.
+  ICEBERG_CHECK(snapshot_ids.empty(), "Committed metadata is missing {} added snapshots",
+                snapshot_ids.size());
+  return committed;
+}
+
 }  // namespace
 
 // ---------------------------------------------------------------------------
@@ -222,6 +281,7 @@ Status Transaction::ReplayUpdates() {
   ICEBERG_CHECK(state_ == TransactionState::kReady,
                 "Replay requires a ready transaction");
   for (const auto& update : pending_updates_) {
+    ICEBERG_RETURN_UNEXPECTED(update->PrepareForReplay());
     ICEBERG_RETURN_UNEXPECTED(ApplyUpdate(*update));
   }
   return {};
@@ -357,6 +417,8 @@ Status Transaction::ApplyUpdateSchema(UpdateSchema& update) {
 Status Transaction::ApplyUpdateSnapshot(SnapshotUpdate& update) {
   const auto& base = ctx_->metadata_builder->current();
 
+  // Report only snapshots generated by this attempt.
+  update.staged_snapshot_.reset();
   ICEBERG_ASSIGN_OR_RAISE(auto result, update.Apply());
   if (result.snapshot == nullptr) {
     return {};
@@ -375,10 +437,8 @@ Status Transaction::ApplyUpdateSnapshot(SnapshotUpdate& update) {
   ICEBERG_RETURN_UNEXPECTED(temp_update->CheckErrors());
 
   if (temp_update->changes().empty()) {
-    // A no-op may still write temporary files. Clean them now; the update remains
-    // registered so it can be replayed after a refresh.
-    internal::LogAndIgnoreFailure("Update staging cleanup",
-                                  [&update] { return update.CleanStaged(); });
+    // A later replay may reuse these files. Terminal cleanup will remove leftovers.
+    update.staged_snapshot_.reset();
     return {};
   }
 
@@ -534,7 +594,40 @@ void Transaction::CleanupUpdates() noexcept {
 }
 
 void Transaction::FinalizeUpdates(const TableMetadata& committed) noexcept {
+  std::optional<std::unordered_set<std::string>> committed_files;
+  internal::LogAndIgnoreFailure("Transaction file discovery", [&]() -> Status {
+    std::unordered_set<int64_t> snapshot_ids;
+    for (const auto& change : ctx_->metadata_builder->changes()) {
+      if (change->kind() == TableUpdate::Kind::kAddSnapshot) {
+        const auto& snapshot =
+            internal::checked_cast<const table::AddSnapshot&>(*change).snapshot();
+        snapshot_ids.insert(snapshot->snapshot_id);
+      }
+    }
+    std::unordered_set<std::string> content_paths;
+    for (const auto& update : pending_updates_) {
+      if (update->kind() != PendingUpdate::Kind::kUpdateSnapshot) {
+        continue;
+      }
+      auto& snapshot_update = internal::checked_cast<SnapshotUpdate&>(*update);
+      std::lock_guard lock(snapshot_update.staging_mutex_);
+      content_paths.insert(snapshot_update.staged_content_files_.begin(),
+                           snapshot_update.staged_content_files_.end());
+    }
+    ICEBERG_ASSIGN_OR_RAISE(
+        committed_files,
+        CommittedTransactionFiles(committed, ctx_->table->io(), std::move(snapshot_ids),
+                                  std::move(content_paths)));
+    return {};
+  });
   for (const auto& update : pending_updates_) {
+    if (update->kind() == PendingUpdate::Kind::kUpdateSnapshot && committed_files) {
+      // Failed discovery must never turn into an empty keep-set.
+      internal::LogAndIgnoreFailure("Snapshot cleanup", [&] {
+        return internal::checked_cast<SnapshotUpdate&>(*update).CleanUncommittedFiles(
+            *committed_files);
+      });
+    }
     internal::LogAndIgnoreFailure("Update finalization", [&update, &committed] {
       return update->Finalize(committed);
     });
@@ -556,7 +649,6 @@ Result<std::shared_ptr<Table>> Transaction::CommitOnce(bool is_first_attempt) {
     if (metadata_changed || standalone_retry) {
       ICEBERG_CHECK(CanRetry(),
                     "Cannot rebase a transaction containing a non-retryable update");
-      CleanupUpdates();
       ctx_->metadata_builder =
           TableMetadataBuilder::BuildFrom(ctx_->table->metadata().get());
       auto applied = ReplayUpdates();
diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc
index fd14308e8..036531dd9 100644
--- a/src/iceberg/update/expire_snapshots.cc
+++ b/src/iceberg/update/expire_snapshots.cc
@@ -918,6 +918,8 @@ Result<ExpireSnapshots::SnapshotToRef> ExpireSnapshots::ComputeRetainedRefs(
 }
 
 Result<ExpireSnapshots::ApplyResult> ExpireSnapshots::Apply() {
+  // A no-op apply must not leave an earlier cleanup plan active.
+  apply_result_.reset();
   ICEBERG_RETURN_UNEXPECTED(CheckErrors());
 
   const TableMetadata& base = this->base();
diff --git a/src/iceberg/update/fast_append.cc b/src/iceberg/update/fast_append.cc
index 1cb1e41cf..60c63e57b 100644
--- a/src/iceberg/update/fast_append.cc
+++ b/src/iceberg/update/fast_append.cc
@@ -97,13 +97,12 @@ Result<std::vector<ManifestFile>> FastAppend::Apply(
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests());
-  // A retry cleanup deletes copied append manifests and clears the rewritten
-  // list; rebuild them from the original appended manifests before re-applying.
-  if (rewritten_append_manifests_.empty() && !append_manifests_to_copy_.empty()) {
-    for (const auto& manifest : append_manifests_to_copy_) {
-      ICEBERG_ASSIGN_OR_RAISE(auto copied_manifest, CopyManifest(manifest));
-      rewritten_append_manifests_.push_back(std::move(copied_manifest));
-    }
+  // Immutable copies and their summaries are reused across replay attempts.
+  for (size_t i = rewritten_append_manifests_.size();
+       i < append_manifests_to_copy_.size(); ++i) {
+    ICEBERG_ASSIGN_OR_RAISE(auto copied_manifest,
+                            CopyManifest(append_manifests_to_copy_[i]));
+    rewritten_append_manifests_.push_back(std::move(copied_manifest));
   }
 
   manifests.reserve(new_written_manifests.size() + append_manifests_.size() +
@@ -147,6 +146,7 @@ std::unordered_map<std::string, std::string> FastAppend::Summary() {
       base().properties.Get(TableProperties::kWritePartitionSummaryLimit));
   summary_.Merge(added_data_files_summary_);
   summary_.Merge(appended_manifests_summary_);
+  summary_.Merge(copied_manifests_summary_);
   for (const auto& [property, value] : custom_summary_properties_) {
     summary_.Set(property, value);
   }
@@ -160,28 +160,11 @@ void FastAppend::SetSummaryProperty(const std::string& property,
   SnapshotUpdate::SetSummaryProperty(property, value);
 }
 
-Status FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) {
-  // Clean up new manifests that were written but not committed
-  if (!new_manifests_.empty()) {
-    for (const auto& manifest : new_manifests_) {
-      if (!committed.contains(manifest.manifest_path)) {
-        std::ignore = DeleteFile(manifest.manifest_path);
-      }
-    }
-    new_manifests_.clear();
-  }
-
-  // Clean up only rewritten append manifests as they are always owned by the table
-  // Don't clean up append manifests as they are added to the manifest list and are
-  // not compacted
-  if (!rewritten_append_manifests_.empty()) {
-    for (const auto& manifest : rewritten_append_manifests_) {
-      if (!committed.contains(manifest.manifest_path)) {
-        std::ignore = DeleteFile(manifest.manifest_path);
-      }
-    }
-    rewritten_append_manifests_.clear();
-  }
+Status FastAppend::CleanUncommitted(
+    const std::unordered_set<std::string>& /*committed*/) {
+  new_manifests_.clear();
+  rewritten_append_manifests_.clear();
+  copied_manifests_summary_.Clear();
   return {};
 }
 
@@ -199,18 +182,19 @@ Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) {
   std::string new_manifest_path = ManifestPath();
   int64_t snapshot_id = SnapshotId();
 
-  // Copy the manifest with the new snapshot ID.
-  return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, snapshot_id,
-                            new_manifest_path, current.format_version,
-                            &appended_manifests_summary_);
+  // A failed copy must not leave partial counts in the cached summary.
+  SnapshotSummaryBuilder summary;
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto copied,
+      CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, snapshot_id,
+                         new_manifest_path, current.format_version, &summary));
+  copied_manifests_summary_.Merge(summary);
+  return copied;
 }
 
 Result<std::vector<ManifestFile>> FastAppend::WriteNewManifests() {
-  // If there are new files and manifests were already written, clean them up
+  // New additions invalidate the cache. Old paths stay staged until cleanup.
   if (has_new_files_ && !new_manifests_.empty()) {
-    for (const auto& manifest : new_manifests_) {
-      std::ignore = DeleteFile(manifest.manifest_path);
-    }
     new_manifests_.clear();
   }
 
diff --git a/src/iceberg/update/fast_append.h b/src/iceberg/update/fast_append.h
index e20576654..81682578a 100644
--- a/src/iceberg/update/fast_append.h
+++ b/src/iceberg/update/fast_append.h
@@ -106,10 +106,12 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate {
   // Stable input summaries for retry-safe summary_ rebuilds.
   SnapshotSummaryBuilder added_data_files_summary_;
   SnapshotSummaryBuilder appended_manifests_summary_;
+  // Copies may be reused across attempts, so their summary must survive too.
+  SnapshotSummaryBuilder copied_manifests_summary_;
   // User-provided summary properties restored after summary_ rebuilds.
   std::unordered_map<std::string, std::string> custom_summary_properties_;
   std::vector<ManifestFile> append_manifests_;
-  // Original manifests kept to recreate copied manifests after retry cleanup.
+  // Original inputs for manifests that must be copied.
   std::vector<ManifestFile> append_manifests_to_copy_;
   std::vector<ManifestFile> rewritten_append_manifests_;
   std::vector<ManifestFile> new_manifests_;
diff --git a/src/iceberg/update/merging_snapshot_update.cc b/src/iceberg/update/merging_snapshot_update.cc
index 154a27b91..0033fd4fa 100644
--- a/src/iceberg/update/merging_snapshot_update.cc
+++ b/src/iceberg/update/merging_snapshot_update.cc
@@ -411,8 +411,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name,
       delete_expression_(Expressions::AlwaysFalse()) {
   auto file_io = ctx_->table->io();
   auto data_filter_manager = ManifestFilterManager::Make(
-      ManifestContent::kData, file_io,
-      [this](const std::string& location) { return DeleteFile(location); });
+      ManifestContent::kData, file_io, [this](const std::string& location) {
+        RegisterStagedFile(location);
+        return Status{};
+      });
   if (!data_filter_manager.has_value()) {
     AddError(data_filter_manager.error());
   } else {
@@ -420,8 +422,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name,
   }
 
   auto delete_filter_manager = ManifestFilterManager::Make(
-      ManifestContent::kDeletes, file_io,
-      [this](const std::string& location) { return DeleteFile(location); });
+      ManifestContent::kDeletes, file_io, [this](const std::string& location) {
+        RegisterStagedFile(location);
+        return Status{};
+      });
   if (!delete_filter_manager.has_value()) {
     AddError(delete_filter_manager.error());
   } else {
@@ -437,7 +441,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name,
   auto data_merge_manager = ManifestMergeManager::Make(
       ManifestContent::kData, target_size_bytes, min_count_to_merge, merge_enabled,
       file_io, [this] { return SnapshotId(); },
-      [this](const std::string& location) { return DeleteFile(location); });
+      [this](const std::string& location) {
+        RegisterStagedFile(location);
+        return Status{};
+      });
   if (!data_merge_manager.has_value()) {
     AddError(data_merge_manager.error());
   } else {
@@ -447,7 +454,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name,
   auto delete_merge_manager = ManifestMergeManager::Make(
       ManifestContent::kDeletes, target_size_bytes, min_count_to_merge, merge_enabled,
       file_io, [this] { return SnapshotId(); },
-      [this](const std::string& location) { return DeleteFile(location); });
+      [this](const std::string& location) {
+        RegisterStagedFile(location);
+        return Status{};
+      });
   if (!delete_merge_manager.has_value()) {
     AddError(delete_merge_manager.error());
   } else {
@@ -675,8 +685,14 @@ Result<ManifestFile> MergingSnapshotUpdate::CopyManifest(const ManifestFile& man
   ICEBERG_ASSIGN_OR_RAISE(auto spec,
                           current.PartitionSpecById(manifest.partition_spec_id));
   std::string path = ManifestPath();
-  return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, SnapshotId(), path,
-                            current.format_version, &appended_manifests_summary_);
+  // A failed copy must not leave partial counts in the cached summary.
+  SnapshotSummaryBuilder summary;
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto copied,
+      CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, SnapshotId(), path,
+                         current.format_version, &summary));
+  copied_manifests_summary_.Merge(summary);
+  return copied;
 }
 
 // -------------------------------------------------------------------------
@@ -729,11 +745,8 @@ ManifestWriterFactory MergingSnapshotUpdate::MakeWriterFactory(
 }
 
 Result<std::vector<ManifestFile>> MergingSnapshotUpdate::WriteNewDataManifests() {
-  // If new files were staged after the cache was populated (commit retry), invalidate.
+  // New additions invalidate the cache. Old paths stay staged until cleanup.
   if (has_new_data_files_ && !cached_new_data_manifests_.empty()) {
-    for (const auto& m : cached_new_data_manifests_) {
-      std::ignore = DeleteFile(m.manifest_path);
-    }
     cached_new_data_manifests_.clear();
   }
 
@@ -819,10 +832,9 @@ MergingSnapshotUpdate::MergeDVs() {
   auto output_path = location_provider->NewDataLocation(
       std::format("merged-dvs-{}-{}.puffin", SnapshotId(), ++dv_merge_attempt_));
 
-  RegisterStagedFile(output_path);
+  RegisterStagedContentFile(output_path);
   auto merged_files = DVUtil::MergeAndWriteDVs(groups, output_path, ctx_->table->io());
   if (!merged_files) {
-    std::ignore = DeleteFile(output_path);
     return ::iceberg::unexpected<Error>(std::move(merged_files.error()));
   }
 
@@ -846,30 +858,16 @@ MergingSnapshotUpdate::MergeDVs() {
                   groups[pos].referenced_data_file);
     PendingDeleteFile merged{.file = std::move(iter->second),
                              .data_sequence_number = data_sequence_numbers[pos]};
-    merged_dvs_.push_back(merged);
     result.push_back(std::move(merged));
   }
 
-  // The operation-specific merged DV cache now owns cleanup for this file.
-  UnregisterStagedFile(output_path);
-
   return result;
 }
 
 Result<std::vector<ManifestFile>> MergingSnapshotUpdate::WriteNewDeleteManifests() {
-  // If new files were staged after the cache was populated (commit retry), invalidate.
+  // New additions invalidate the cache. Old paths stay staged until cleanup.
   if (has_new_delete_files_ && !cached_new_delete_manifests_.empty()) {
-    for (const auto& m : cached_new_delete_manifests_) {
-      std::ignore = DeleteFile(m.manifest_path);
-    }
     cached_new_delete_manifests_.clear();
-    std::unordered_set<std::string> deleted_merged_dv_paths;
-    for (const auto& dv : merged_dvs_) {
-      if (deleted_merged_dv_paths.insert(dv.file->file_path).second) {
-        std::ignore = DeleteFile(dv.file->file_path);
-      }
-    }
-    merged_dvs_.clear();
     added_delete_files_summary_.Clear();
   }
 
@@ -921,6 +919,12 @@ Result<std::vector<ManifestFile>> MergingSnapshotUpdate::WriteNewDeleteManifests
   return result;
 }
 
+Status MergingSnapshotUpdate::PrepareForReplay() {
+  // The same path can inherit new sequence numbers and row IDs on replay.
+  // Drop derived caches, but keep their files staged for terminal cleanup.
+  return CleanUncommittedManagers({});
+}
+
 Result<std::vector<ManifestFile>> MergingSnapshotUpdate::Apply(
     const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) {
   appended_manifests_summary_.Clear();
@@ -990,11 +994,10 @@ Result<std::vector<ManifestFile>> MergingSnapshotUpdate::Apply(
 
   // Step 4: Write (or retrieve cached) new data manifests.
   ICEBERG_ASSIGN_OR_RAISE(auto written_data_manifests, WriteNewDataManifests());
-  if (rewritten_append_manifests_.empty() && !append_manifests_to_copy_.empty()) {
-    for (const auto& manifest : append_manifests_to_copy_) {
-      ICEBERG_ASSIGN_OR_RAISE(auto copied, CopyManifest(manifest));
-      rewritten_append_manifests_.push_back(std::move(copied));
-    }
+  for (size_t i = rewritten_append_manifests_.size();
+       i < append_manifests_to_copy_.size(); ++i) {
+    ICEBERG_ASSIGN_OR_RAISE(auto copied, CopyManifest(append_manifests_to_copy_[i]));
+    rewritten_append_manifests_.push_back(std::move(copied));
   }
 
   // Incorporate append manifests (from AddManifest), stamping each with the
@@ -1027,6 +1030,7 @@ Result<std::vector<ManifestFile>> MergingSnapshotUpdate::Apply(
   summary_builder().Merge(added_data_files_summary_);
   summary_builder().Merge(added_delete_files_summary_);
   summary_builder().Merge(appended_manifests_summary_);
+  summary_builder().Merge(copied_manifests_summary_);
   for (const auto& [property, value] : custom_summary_properties_) {
     summary_builder().Set(property, value);
   }
@@ -1061,57 +1065,39 @@ Result<std::vector<ManifestFile>> MergingSnapshotUpdate::Apply(
 
 Status MergingSnapshotUpdate::CleanUncommitted(
     const std::unordered_set<std::string>& committed) {
+  ICEBERG_RETURN_UNEXPECTED(CleanUncommittedManagers(committed));
+  CleanUncommittedAppends(committed);
+  return {};
+}
+
+Status MergingSnapshotUpdate::CleanUncommittedManagers(
+    const std::unordered_set<std::string>& committed) {
   ICEBERG_RETURN_UNEXPECTED(ManagersReady());
   ICEBERG_RETURN_UNEXPECTED(data_merge_manager_->CleanUncommitted(committed));
   ICEBERG_RETURN_UNEXPECTED(data_filter_manager_->CleanUncommitted(committed));
   ICEBERG_RETURN_UNEXPECTED(delete_merge_manager_->CleanUncommitted(committed));
-  ICEBERG_RETURN_UNEXPECTED(delete_filter_manager_->CleanUncommitted(committed));
-  ICEBERG_RETURN_UNEXPECTED(CleanUncommittedAppends(committed));
-  return {};
+  return delete_filter_manager_->CleanUncommitted(committed);
 }
 
-Status MergingSnapshotUpdate::CleanUncommittedAppends(
+void MergingSnapshotUpdate::CleanUncommittedAppends(
     const std::unordered_set<std::string>& committed) {
-  ICEBERG_RETURN_UNEXPECTED(
-      DeleteUncommitted(cached_new_data_manifests_, committed, /*clear=*/true));
-  ICEBERG_RETURN_UNEXPECTED(
-      DeleteUncommitted(cached_new_delete_manifests_, committed, /*clear=*/true));
-  // rewritten_append_manifests_ are always owned by the table.
-  ICEBERG_RETURN_UNEXPECTED(
-      DeleteUncommitted(rewritten_append_manifests_, committed, /*clear=*/true));
-  if (committed.empty()) {
-    std::unordered_set<std::string> deleted_merged_dv_paths;
-    for (const auto& dv : merged_dvs_) {
-      if (deleted_merged_dv_paths.insert(dv.file->file_path).second) {
-        std::ignore = DeleteFile(dv.file->file_path);
-      }
-    }
-  }
-  merged_dvs_.clear();
+  // Generated files are already tracked by SnapshotUpdate.
+  cached_new_data_manifests_.clear();
+  cached_new_delete_manifests_.clear();
+  rewritten_append_manifests_.clear();
+  copied_manifests_summary_.Clear();
 
-  // append_manifests_ are only owned by the table if the commit succeeded.
+  // External append manifests become table-owned only after a successful commit.
   if (!committed.empty()) {
-    ICEBERG_RETURN_UNEXPECTED(
-        DeleteUncommitted(append_manifests_, committed, /*clear=*/false));
+    for (const auto& manifest : append_manifests_) {
+      if (!committed.contains(manifest.manifest_path)) {
+        RegisterStagedFile(manifest.manifest_path);
+      }
+    }
   }
 
   has_new_data_files_ = false;
   has_new_delete_files_ = false;
-  return {};
-}
-
-Status MergingSnapshotUpdate::DeleteUncommitted(
-    std::vector<ManifestFile>& manifests,
-    const std::unordered_set<std::string>& committed, bool clear) {
-  for (const auto& manifest : manifests) {
-    if (!committed.contains(manifest.manifest_path)) {
-      std::ignore = DeleteFile(manifest.manifest_path);
-    }
-  }
-  if (clear) {
-    manifests.clear();
-  }
-  return {};
 }
 
 std::unordered_map<std::string, std::string> MergingSnapshotUpdate::Summary() {
diff --git a/src/iceberg/update/merging_snapshot_update.h b/src/iceberg/update/merging_snapshot_update.h
index e1cd7303b..3938a9d1e 100644
--- a/src/iceberg/update/merging_snapshot_update.h
+++ b/src/iceberg/update/merging_snapshot_update.h
@@ -71,6 +71,7 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
       const std::shared_ptr<Snapshot>& snapshot) override;
 
   Status CleanUncommitted(const std::unordered_set<std::string>& committed) override;
+  Status PrepareForReplay() override;
 
   std::unordered_map<std::string, std::string> Summary() override;
 
@@ -348,10 +349,8 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
   /// \brief Write new delete manifests for staged delete files; caches the result.
   Result<std::vector<ManifestFile>> WriteNewDeleteManifests();
 
-  Status CleanUncommittedAppends(const std::unordered_set<std::string>& committed);
-
-  Status DeleteUncommitted(std::vector<ManifestFile>& manifests,
-                           const std::unordered_set<std::string>& committed, bool clear);
+  Status CleanUncommittedManagers(const std::unordered_set<std::string>& committed);
+  void CleanUncommittedAppends(const std::unordered_set<std::string>& committed);
 
   // Used for commit event notifications and diagnostic log messages.
   std::string table_name_;
@@ -363,6 +362,8 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
   SnapshotSummaryBuilder added_data_files_summary_;
   SnapshotSummaryBuilder added_delete_files_summary_;
   SnapshotSummaryBuilder appended_manifests_summary_;
+  // Copies may be reused across attempts, so their summary must survive too.
+  SnapshotSummaryBuilder copied_manifests_summary_;
   std::unordered_map<std::string, std::string> custom_summary_properties_;
 
   std::unique_ptr<ManifestFilterManager> data_filter_manager_;
@@ -378,7 +379,7 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
   // Manifests passed via AddManifest(): inherit path (no copy needed) and
   // rewrite path (must be copied with the current snapshot ID).
   std::vector<ManifestFile> append_manifests_;
-  // Original manifests kept to recreate copied manifests after retry cleanup.
+  // Original inputs for manifests that must be copied.
   std::vector<ManifestFile> append_manifests_to_copy_;
   std::vector<ManifestFile> rewritten_append_manifests_;
 
@@ -389,7 +390,6 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
 
   std::vector<ManifestFile> cached_new_data_manifests_;
   std::vector<ManifestFile> cached_new_delete_manifests_;
-  std::vector<PendingDeleteFile> merged_dvs_;
   int32_t dv_merge_attempt_ = 0;
 };
 
diff --git a/src/iceberg/update/pending_update.h b/src/iceberg/update/pending_update.h
index 81bf4563e..972f77c9e 100644
--- a/src/iceberg/update/pending_update.h
+++ b/src/iceberg/update/pending_update.h
@@ -93,6 +93,10 @@ class ICEBERG_EXPORT PendingUpdate : public ErrorCollector,
 
   /// \brief Discard this generation's staging, retaining applied operation intent.
   virtual Status CleanStaged() { return {}; }
+
+  /// \brief Prepare retry-safe state after the transaction metadata is refreshed.
+  virtual Status PrepareForReplay() { return {}; }
+
   /// \brief Complete a known successful commit. Called only by Transaction with the
   /// final committed table metadata.
   virtual Status Finalize(const TableMetadata& committed);
diff --git a/src/iceberg/update/rewrite_manifests.cc b/src/iceberg/update/rewrite_manifests.cc
index 653663b61..38abe8d93 100644
--- a/src/iceberg/update/rewrite_manifests.cc
+++ b/src/iceberg/update/rewrite_manifests.cc
@@ -84,6 +84,19 @@ Status CatchStatus(std::string_view operation, Fn&& fn) {
   }
 }
 
+Status FinishWithCloseStatus(Status rewrite_status, Status close_status) {
+  // Keep the rewrite error when both operations fail.
+  if (!rewrite_status.has_value()) {
+    auto error = std::move(rewrite_status).error();
+    if (!close_status.has_value()) {
+      error.message += "; additionally failed to close manifest writer: ";
+      error.message += close_status.error().message;
+    }
+    return ::iceberg::unexpected<Error>(std::move(error));
+  }
+  return close_status;
+}
+
 }  // namespace
 
 class RewriteManifests::WriterWrapper {
@@ -177,7 +190,7 @@ RewriteManifests& RewriteManifests::AddManifest(const ManifestFile& manifest) {
   if (can_inherit_snapshot_id() && !manifest.added_snapshot_id.has_value()) {
     added_manifests_.push_back(manifest);
   } else {
-    // Retain the input so Apply can recreate the copy after retry cleanup.
+    // Retain the input so replay can rebuild the copy.
     added_manifests_to_copy_.push_back(manifest);
   }
   return *this;
@@ -185,6 +198,11 @@ RewriteManifests& RewriteManifests::AddManifest(const ManifestFile& manifest) {
 
 std::string RewriteManifests::operation() { return DataOperation::kReplace; }
 
+Status RewriteManifests::PrepareForReplay() {
+  // Rebuild against the latest manifest list on every replay.
+  return CleanUncommitted({});
+}
+
 Result<std::vector<ManifestFile>> RewriteManifests::Apply(
     const TableMetadata& /*metadata_to_update*/,
     const std::shared_ptr<Snapshot>& snapshot) {
@@ -262,12 +280,9 @@ std::unordered_map<std::string, std::string> RewriteManifests::Summary() {
 }
 
 Status RewriteManifests::CleanUncommitted(
-    const std::unordered_set<std::string>& committed) {
-  DeleteUncommitted(new_manifests_, committed);
-  DeleteUncommitted(rewritten_added_manifests_, committed);
-  entry_count_ = 0;
-  kept_manifests_.clear();
-  rewritten_manifests_.clear();
+    const std::unordered_set<std::string>& /*committed*/) {
+  Reset();
+  rewritten_added_manifests_.clear();
   return {};
 }
 
@@ -311,11 +326,20 @@ Status RewriteManifests::ValidateDeletedManifests(
     int64_t current_snapshot_id) const {
   // Directly deleted manifests must still be present in the current snapshot.
   for (const auto& manifest : deleted_manifests_) {
-    if (!current_manifests.contains(manifest)) {
+    auto current = current_manifests.find(manifest);
+    if (current == current_manifests.end()) {
       return ValidationFailed(
           "Deleted manifest {} could not be found in the latest snapshot {}",
           manifest.manifest_path, current_snapshot_id);
     }
+    // A prior update may reuse this path with new inherited values on replay.
+    // The caller's replacement still contains the original values.
+    if (manifest.sequence_number != current->sequence_number ||
+        manifest.min_sequence_number != current->min_sequence_number ||
+        manifest.first_row_id != current->first_row_id) {
+      return ValidationFailed("Deleted manifest metadata changed for {} in snapshot {}",
+                              manifest.manifest_path, current_snapshot_id);
+    }
   }
   return {};
 }
@@ -467,36 +491,19 @@ Status RewriteManifests::PerformRewrite(std::span<const ManifestFile> current_ma
     return process_batch();
   });
 
-  auto close_status = CloseWriters();
-
-  if (!write_status.has_value()) {
-    auto error = std::move(write_status).error();
-    if (!close_status.has_value()) {
-      error.message += "; additionally failed to close manifest writer: ";
-      error.message += close_status.error().message;
-    }
-    return ::iceberg::unexpected<Error>(std::move(error));
-  }
-  if (!close_status.has_value()) {
-    return close_status;
+  // Close every writer before returning the first failure.
+  auto status = FinishWithCloseStatus(std::move(write_status), CloseWriters());
+  if (!status.has_value()) {
+    return status;
   }
   rewrite_dirty_ = false;
   return {};
 }
 
-void RewriteManifests::DeleteUncommitted(
-    std::vector<ManifestFile>& manifests,
-    const std::unordered_set<std::string>& committed) {
-  for (const auto& manifest : manifests) {
-    if (!committed.contains(manifest.manifest_path)) {
-      std::ignore = DeleteFile(manifest.manifest_path);
-    }
-  }
-  manifests.clear();
-}
-
 void RewriteManifests::Reset() {
-  DeleteUncommitted(new_manifests_, {});
+  // Replay keeps staged outputs alive until the transaction reaches a terminal
+  // state. Clear only the in-memory generation.
+  new_manifests_.clear();
   entry_count_ = 0;
   kept_manifests_.clear();
   rewritten_manifests_.clear();
diff --git a/src/iceberg/update/rewrite_manifests.h b/src/iceberg/update/rewrite_manifests.h
index 669e0fb1f..3aa968cb9 100644
--- a/src/iceberg/update/rewrite_manifests.h
+++ b/src/iceberg/update/rewrite_manifests.h
@@ -102,6 +102,7 @@ class ICEBERG_EXPORT RewriteManifests : public SnapshotUpdate {
       const std::shared_ptr<Snapshot>& snapshot) override;
   std::unordered_map<std::string, std::string> Summary() override;
   Status CleanUncommitted(const std::unordered_set<std::string>& committed) override;
+  Status PrepareForReplay() override;
 
   explicit RewriteManifests(std::string table_name,
                             std::shared_ptr<TransactionContext> ctx);
@@ -139,8 +140,6 @@ class ICEBERG_EXPORT RewriteManifests : public SnapshotUpdate {
       const std::shared_ptr<PartitionSpec>& spec);
   Status CloseWriters();
 
-  void DeleteUncommitted(std::vector<ManifestFile>& manifests,
-                         const std::unordered_set<std::string>& committed);
   void Reset();
 
  private:
diff --git a/src/iceberg/update/snapshot_update.cc b/src/iceberg/update/snapshot_update.cc
index d6a1066ac..c2604a9c0 100644
--- a/src/iceberg/update/snapshot_update.cc
+++ b/src/iceberg/update/snapshot_update.cc
@@ -23,6 +23,7 @@
 #include <format>
 #include <ranges>
 #include <span>
+#include <unordered_set>
 
 #include "iceberg/constants.h"
 #include "iceberg/file_io.h"
@@ -193,6 +194,18 @@ Result<ManifestFile> AddMetadata(const ManifestFile& manifest, std::shared_ptr<F
   return enriched;
 }
 
+// Cached files can outlive a spec removed by concurrent metadata cleanup.
+Status ValidateManifestPartitionSpecs(std::span<const ManifestFile> manifests,
+                                      const TableMetadata& metadata) {
+  std::unordered_set<int32_t> checked_spec_ids;
+  for (const auto& manifest : manifests) {
+    if (checked_spec_ids.insert(manifest.partition_spec_id).second) {
+      ICEBERG_RETURN_UNEXPECTED(metadata.PartitionSpecById(manifest.partition_spec_id));
+    }
+  }
+  return {};
+}
+
 }  // anonymous namespace
 
 SnapshotUpdate::~SnapshotUpdate() = default;
@@ -322,6 +335,7 @@ Result<SnapshotUpdate::ApplyResult> SnapshotUpdate::Apply() {
   ICEBERG_RETURN_UNEXPECTED(Validate(base(), parent_snapshot));
 
   ICEBERG_ASSIGN_OR_RAISE(auto manifests, Apply(base(), parent_snapshot));
+  ICEBERG_RETURN_UNEXPECTED(ValidateManifestPartitionSpecs(manifests, base()));
   auto metadata_tasks = TaskGroup().SetExecutor(plan_executor_);
   for (auto& manifest : manifests) {
     if (manifest.added_snapshot_id.has_value()) {
@@ -384,43 +398,33 @@ Result<SnapshotUpdate::ApplyResult> SnapshotUpdate::Apply() {
                      .stage_only = stage_only_};
 }
 
-Status SnapshotUpdate::Finalize([[maybe_unused]] const TableMetadata& metadata) {
-  if (staged_snapshot_ == nullptr) {
-    return {};
-  }
-
-  // Only files created by this update are tracked, and committed paths are kept below.
-  internal::LogAndIgnoreFailure("Snapshot cleanup", [this]() -> Status {
-    auto snapshot_reader = SnapshotReader(staged_snapshot_.get());
-    ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.Manifests(ctx_->table->io()));
-    auto committed = manifests | std::views::transform([](const auto& manifest) {
-                       return manifest.manifest_path;
-                     }) |
-                     std::ranges::to<std::unordered_set<std::string>>();
-    internal::LogAndIgnoreFailure("Snapshot operation cleanup", [this, &committed] {
-      return CleanUncommitted(committed);
-    });
-    committed.insert(staged_snapshot_->manifest_list);
-    std::vector<std::string> unused;
-    {
-      std::lock_guard lock(staging_mutex_);
-      for (const auto& path : staged_files_) {
-        if (!committed.contains(path)) {
-          unused.push_back(path);
-        }
-      }
-    }
-    for (const auto& path : unused) {
-      std::ignore = DeleteFile(path);
-    }
-    return {};
+Status SnapshotUpdate::CleanUncommittedFiles(
+    const std::unordered_set<std::string>& committed) {
+  // All retained snapshots share this set. A later update may reuse our output.
+  internal::LogAndIgnoreFailure("Snapshot operation cleanup", [this, &committed] {
+    return CleanUncommitted(committed);
   });
-
+  std::vector<std::string> unused;
   {
     std::lock_guard lock(staging_mutex_);
-    staged_files_.clear();
+    for (const auto& path : staged_files_) {
+      if (!committed.contains(path)) {
+        unused.push_back(path);
+      }
+    }
+    for (const auto& path : committed) {
+      staged_files_.erase(path);
+      staged_content_files_.erase(path);
+    }
+  }
+  for (const auto& path : unused) {
+    std::ignore = DeleteStagedFile(path);
   }
-  auto report_status = ReportCommit();
+  return {};
+}
+
+Status SnapshotUpdate::Finalize([[maybe_unused]] const TableMetadata& committed) {
+  auto report_status = staged_snapshot_ ? ReportCommit() : Status{};
   staged_snapshot_.reset();
   summary_.Clear();
   return report_status;
@@ -479,14 +483,14 @@ Result<std::unordered_map<std::string, std::string>> SnapshotUpdate::ComputeSumm
 Status SnapshotUpdate::CleanStaged() {
   internal::LogAndIgnoreFailure("Snapshot staging cleanup",
                                 [this] { return CleanUncommitted({}); });
-  // Retry paths that operation-specific cleanup did not delete.
+  // Delete all remaining staged paths, including partial writes.
   std::vector<std::string> paths;
   {
     std::lock_guard lock(staging_mutex_);
     paths.assign(staged_files_.begin(), staged_files_.end());
   }
   for (const auto& path : paths) {
-    std::ignore = DeleteFile(path);
+    std::ignore = DeleteStagedFile(path);
   }
   staged_snapshot_.reset();
   summary_.Clear();
@@ -498,26 +502,25 @@ void SnapshotUpdate::RegisterStagedFile(const std::string& path) {
   staged_files_.insert(path);
 }
 
-void SnapshotUpdate::UnregisterStagedFile(const std::string& path) {
+void SnapshotUpdate::RegisterStagedContentFile(const std::string& path) {
   std::lock_guard lock(staging_mutex_);
-  staged_files_.erase(path);
+  staged_files_.insert(path);
+  staged_content_files_.insert(path);
 }
 
-Status SnapshotUpdate::DeleteFile(const std::string& path) noexcept {
+Status SnapshotUpdate::DeleteStagedFile(const std::string& path) noexcept {
   try {
     auto result = delete_func_ ? delete_func_(path) : ctx_->table->io()->DeleteFile(path);
     if (!result && result.error().kind != ErrorKind::kNotFound) {
-      RegisterStagedFile(path);
       ICEBERG_LOG_WARN("Cannot clean staged file {}: {}", path, result.error().message);
       return {};
     }
     std::lock_guard lock(staging_mutex_);
     staged_files_.erase(path);
+    staged_content_files_.erase(path);
   } catch (const std::exception& e) {
-    RegisterStagedFile(path);
     ICEBERG_LOG_WARN("Cannot clean staged file {}: {}", path, e.what());
   } catch (...) {
-    RegisterStagedFile(path);
     ICEBERG_LOG_WARN("Cannot clean staged file {}: unknown exception", path);
   }
   return {};
diff --git a/src/iceberg/update/snapshot_update.h b/src/iceberg/update/snapshot_update.h
index 18e81e4f9..42b1ae984 100644
--- a/src/iceberg/update/snapshot_update.h
+++ b/src/iceberg/update/snapshot_update.h
@@ -236,16 +236,14 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
   int32_t attempt() const { return attempt_.load(std::memory_order_relaxed); }
   int64_t target_manifest_size_bytes() const { return target_manifest_size_bytes_; }
 
-  /// \brief Clean up any uncommitted manifests that were created.
+  /// \brief Clear operation caches and register any remaining owned files.
   ///
-  /// Manifests may not be committed if Apply is called multiple times
-  /// because a commit conflict has occurred. Implementations may keep
-  /// around manifests because the same changes will be made by both
-  /// Apply calls. This method instructs the implementation to clean up
-  /// those manifests and passes the paths of the manifests that were
-  /// actually committed.
+  /// Do not delete files here. SnapshotUpdate deletes uncommitted paths through
+  /// its staging registry. Generated paths must be registered before writing.
+  /// External inputs become table-owned only after a successful commit.
   ///
-  /// \param committed A set of manifest paths that were actually committed
+  /// \param committed File paths protected by retained transaction snapshots.
+  /// Empty when discarding all staged files.
   virtual Status CleanUncommitted(const std::unordered_set<std::string>& committed) = 0;
 
   /// \brief A string that describes the action that produced the new snapshot.
@@ -287,11 +285,9 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
   /// \brief Get or generate the snapshot ID for the new snapshot.
   int64_t SnapshotId();
 
-  /// Best-effort delete. Failed paths remain registered for later cleanup.
-  Status DeleteFile(const std::string& path) noexcept;
   void RegisterStagedFile(const std::string& path);
-  /// Transfer cleanup ownership to the derived update.
-  void UnregisterStagedFile(const std::string& path);
+  /// Track an owned content file as well as its cleanup path.
+  void RegisterStagedContentFile(const std::string& path);
 
   std::string ManifestPath();
   std::string ManifestListPath();
@@ -306,6 +302,8 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
       const TableMetadata& previous);
 
   Status ReportCommit() const;
+  Status CleanUncommittedFiles(const std::unordered_set<std::string>& committed);
+  Status DeleteStagedFile(const std::string& path) noexcept;
 
  protected:
   SnapshotSummaryBuilder summary_;
@@ -364,9 +362,16 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
   int32_t write_manifest_parallelism_{1};
   std::atomic<int32_t> manifest_count_{0};
   std::atomic<int32_t> attempt_{0};
-  // All paths are registered before writes, including partial/failed writes.
+  // Protect both staging sets. All paths are registered before writes, including
+  // partial and failed writes.
   std::mutex staging_mutex_;
+  // Every file path created or owned by this update: manifests, manifest lists,
+  // and content files. Terminal cleanup may delete paths from this set.
   std::unordered_set<std::string> staged_files_;
+  // A subset of staged_files_ containing owned content files, such as merged
+  // deletion vectors. Successful cleanup keeps these only when a retained
+  // manifest still references them.
+  std::unordered_set<std::string> staged_content_files_;
   const int64_t target_manifest_size_bytes_;
   std::optional<int64_t> snapshot_id_;
   OptionalExecutor plan_executor_;