diff --git a/docs/testing/BINARY_S3_STORAGE.md b/docs/testing/BINARY_S3_STORAGE.md
index 4718697bb8d8..d31e1a4e8327 100644
--- a/docs/testing/BINARY_S3_STORAGE.md
+++ b/docs/testing/BINARY_S3_STORAGE.md
@@ -35,8 +35,8 @@ table may not be seen at first read, and a runtime change is ignored until resta
 overrides through `Config.setProperty` do re-read it, which is how tests switch modes. Tests
 that mock `Config` statically must call `AssetStorageFeature.reset()` themselves.
 
-With the flag off, the S3 cleanup job queues (`binaryAssetCleanup`, `binaryFieldCleanup`) are not
-registered.
+With the flag off, the S3 job queues (`binaryAssetCleanup`, `binaryFieldCleanup`,
+`binaryAssetBackfill`) are not registered.
 
 ### Enabling the flag is not rollback-safe
 
@@ -202,6 +202,93 @@ database restore, and archives are kept until removed or expired by the bucket's
 With the flag on, the `deleteAllVersionsandBackup` interceptor, previously a no-op, calls its
 implementation and the all-version deletion hooks. With the flag off it stays a no-op.
 
+## Migrating existing binaries (backfill)
+
+With the flag on, an active administrator can copy existing binaries to S3 through the job API.
+The processor checks administrator status when the job is queued and again when it runs.
+
+```http
+POST /api/v1/jobs/binaryAssetBackfill
+Content-Type: application/json
+
+{"batchSize":250}
+```
+
+Follow the returned `statusUrl` (`GET /api/v1/jobs/{jobId}/status`). `parameters.afterInode` is
+the last committed batch cursor and `parameters.verifiedBinaries` counts the binaries verified in
+completed batches. The batch size is a number of content inodes, from 1 to 1000. A successful
+result includes `complete: true`; progress reaches 100% only when the scan finishes.
+
+Each batch uses conditional, verified backfill of originals and metadata, and converts raw S3
+objects to SHA-256 references without changing their database paths. It reads persisted binary
+fields, including retired field definitions, and migrates recognized completed renditions for each
+inode. Local sources are left in place. Only a fully verified batch advances the cursor, retries
+reload the cursor from the job database, and a stale worker cannot overwrite newer progress. The
+page size is applied in the SQL query, so each batch reads only its own rows. The job sends a
+heartbeat after every inode, so a slow batch is not mistaken for an abandoned job.
+
+Problems in the data itself do not stop the scan, because a retry cannot fix them: a referenced
+binary or metadata record that exists neither locally nor in S3, a row whose JSON cannot be parsed,
+and a legacy row whose content cannot be found. Each one is logged with its inode and field, the
+rest of the row is still copied, and the job reports `skippedCount` and `skippedInodes` (the first
+1000 inodes) in its parameters and result. Storage and database errors, including an S3 object
+that conflicts with the local bytes, still fail the batch so the retry policy applies; the error
+message names the inode.
+
+`POST /api/v1/jobs/{jobId}/cancel` stops between batches. After a cancellation or exhausted
+retries, submit a new job with the last persisted parameters, for example
+`{"batchSize":250,"afterInode":"<parameters.afterInode>","verifiedBinaries":42}`, or omit
+`afterInode` to verify everything again. With the flag off, enqueueing and running this job are
+rejected. The scan covers content-referenced originals and metadata, not operational server files.
+
+### Legacy Image and File values
+
+Backfill, starter export and recovery archives use the same inventory of persisted references.
+Typed `Binary` entries are included even when their field definitions were retired. An `Image` or
+`File` entry holding a plain filename is included only when the exact
+`<inode-shards>/<inode>/<field>/<filename>` object exists in the combined filesystem and S3
+inventory, so linked asset identifiers, external URLs and unrelated text are never treated as
+binaries. Listing failures propagate instead of being read as absence.
+
+## Starter export and import
+
+With the flag on, starter asset export lists persisted binary references, restores originals from
+S3 and includes raw metadata alongside legacy local files. It holds a cache lease for one binary at
+a time, so eviction can keep trimming the files it restores, and it skips a binary whose stored
+metadata says it exceeds `maxSize` before restoring it. Restored originals still pass through the
+local cache; streaming them from S3 straight into the archive is not implemented. A missing
+referenced original or any other export error fails the export without finishing the archive: the
+client receives a truncated ZIP with no central directory, which standard ZIP readers reject,
+instead of a valid ZIP that silently lacks the remaining entries. Whether the HTTP transfer also
+ends with an error depends on the servlet container. With the flag off, export behaves as before,
+including logging and skipping a file that cannot be read.
+
+Importing a starter publishes its binaries to S3 before the database commit, and cleanup of the
+imported files runs only after both succeed. A referenced binary or metadata record that is in
+neither the starter nor S3 is logged and skipped, as before, so starters exported without assets,
+with `maxSize` or with `oldAssets=false` still import; a summary count is logged at the end. S3 and
+database errors still fail the import, and so does an error importing rules, so a starter is
+never left partly imported.
+
+Importing over a populated database with the flag on performs a full replacement: existing
+variants, workflows, templates, categories, rules and experiments are cleared before import,
+foreign keys stay enforced, and caches are flushed afterwards. With the flag off, import behaves as
+before.
+
+A starter exported with the flag on cannot be fully restored into an instance with the flag off.
+Content checked in with the flag on has its binary only at its `.revisions/<uuid>/` key, and the
+archive stores it there; a flag-off instance resolves binaries by the legacy field path and ignores
+`storageKey`, so those binaries are missing after import. Import such a starter only into an
+instance with the flag on.
+
+## Integrity checks
+
+With the flag on, the file-asset integrity checker's repair copies stored binaries to the repaired
+content before its corrected JSON is published, and removes the old sources in the same
+repair transaction. The copies are uploaded before the commit, so a rollback listener deletes
+them, and their metadata, if the repair rolls back; a failed deletion is logged and leaves an
+unreferenced object behind.
+
 ## Local cache eviction
 
 With the flag on, the local asset directory is a cache that `BinaryCacheEvictionJob` can trim.
@@ -300,7 +387,7 @@ docker run -d --rm --name binary-cleanup-postgres-test \
   -e POSTGRES_DB=binary_storage_test postgres:16-alpine
 
 ./mvnw test -pl :dotcms-core -Dmaven.build.cache.enabled=false \
-  -Dtest=AssetStorageFeatureTest,AssetStorageFeatureLatchTest,S3StorageConfigurationTest,NoWebIdentityCredentialsProviderChainTest,BinaryS3StorageTest,BinaryAssetReferenceTest,BinaryCacheEvictionJobTest,BinaryFileSystemStorageTest,BinaryAssetStorageAPIImplTest,MetadataLocalCacheTest,BinaryAssetCleanupTransactionTest,BinaryAssetCleanupProcessorTest,ContentletBackupStorageGateTest,BinaryFieldCleanupProcessorTest,AssetJobEventSerializationTest \
+  -Dtest=AssetStorageFeatureTest,AssetStorageFeatureLatchTest,S3StorageConfigurationTest,NoWebIdentityCredentialsProviderChainTest,BinaryS3StorageTest,BinaryAssetReferenceTest,BinaryCacheEvictionJobTest,BinaryFileSystemStorageTest,BinaryAssetStorageAPIImplTest,MetadataLocalCacheTest,BinaryAssetCleanupTransactionTest,BinaryAssetCleanupProcessorTest,ContentletBackupStorageGateTest,BinaryFieldCleanupProcessorTest,AssetJobEventSerializationTest,BinaryAssetBackfillCheckpointTest,ImportStarterWorkflowCleanupTest,BinaryAssetBackfillProcessorTest,BinaryAssetBackfillTest,ExportStarterFailureTest \
   -Ds3.test.endpoint=http://127.0.0.1:19002 \
   -Ds3.test.jdbc=jdbc:postgresql://127.0.0.1:19003/binary_storage_test
 
@@ -313,7 +400,22 @@ stack:
 ```sh
 ./mvnw install -pl :dotcms-core --am -DskipTests -Ddocker.skip
 ./mvnw verify -pl :dotcms-integration -Dmaven.build.cache.enabled=false -Dcoreit.test.skip=false \
-  -Dit.test=BinaryAssetStorageIntegrationTest,ContentletBackupStorageTest,SharedAssetStorageIntegrationTest
+  -Dit.test=BinaryAssetStorageIntegrationTest,ContentletBackupStorageTest,SharedAssetStorageIntegrationTest,BinaryAssetStarterRestoreTest
 ```
 
+These default to flag-off mode, where the S3 cases are skipped. To run the S3 cases, create a
+bucket in the disposable MinIO (for example `mc mb local/s3-cms-it` inside the container) and add
+`-Dit.test.forkcount=1 -Ds3.cms.enabled=true -DDOT_FEATURE_FLAG_S3_ASSET_STORAGE=true
+-DDOT_BINARY_ASSET_STORAGE_TYPE=BINARY_CHAIN -DDOT_STORAGE_FILE_METADATA_DEFAULT_CHAIN=FILE_SYSTEM,S3`
+plus the `DOT_STORAGE_FILE_METADATA_S3_BUCKET_NAME`, `_BUCKET_REGION`, `_ACCESS_KEY`,
+`_SECRET_ACCESS_KEY` and `_ENDPOINT` properties.
+
+`BinaryAssetStarterRestoreTest` is an opt-in starter acceptance test that runs in phases against
+the same bucket. Add `-Ds3.starter.restore.enabled=true` and a persistent
+`-Ds3.starter.archive=<absolute path under dotCMS/target>/starter.zip`, then run
+`-Ds3.starter.phase=export`, followed by `-Ds3.starter.phase=restore` with
+`-Dstarter.run.path` set to the same archive. `-Ds3.starter.phase=populated` (after a fresh
+export, without `-Dstarter.run.path`) tests replacing a populated database; run it only in the
+disposable harness, because it replaces that database's content.
+
 CI does not yet provide the MinIO service, so `BinaryS3StorageTest` does not run there.
diff --git a/dotCMS/src/main/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityChecker.java b/dotCMS/src/main/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityChecker.java
index 3da217c2fbca..68ac76975b05 100644
--- a/dotCMS/src/main/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityChecker.java
+++ b/dotCMS/src/main/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityChecker.java
@@ -2,11 +2,19 @@
 
 import com.dotcms.content.business.json.ContentletJsonAPI;
 import com.dotcms.content.elasticsearch.business.ContentletIndexAPI;
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotcms.storage.FetchMetadataParams;
+import com.dotcms.storage.FileMetadataAPI;
+import com.dotcms.storage.StorageKey;
+import com.dotcms.storage.StoragePersistenceProvider;
+import com.dotcms.storage.binary.BinaryAssetCleanupProcessor;
+import com.dotcms.storage.binary.BinaryAssetReference;
 import com.dotmarketing.beans.Identifier;
 import com.dotmarketing.business.APILocator;
 import com.dotmarketing.business.DotStateException;
 import com.dotmarketing.common.db.DotConnect;
 import com.dotmarketing.db.DbConnectionFactory;
+import com.dotmarketing.db.HibernateUtil;
 import com.dotmarketing.exception.DotDataException;
 import com.dotmarketing.exception.DotHibernateException;
 import com.dotmarketing.exception.DotRuntimeException;
@@ -17,6 +25,7 @@
 import com.dotmarketing.portlets.fileassets.business.FileAssetAPI;
 import com.dotmarketing.portlets.structure.model.Structure;
 import com.dotmarketing.util.Constants;
+import com.dotmarketing.util.Config;
 import com.dotmarketing.util.Logger;
 import com.dotmarketing.util.UtilMethods;
 import com.liferay.portal.model.User;
@@ -78,6 +87,25 @@ public boolean generateIntegrityResults(final String endpointId) throws Exceptio
 
     @Override
     public void executeFix(final String key) throws DotDataException, DotSecurityException {
+        if (!AssetStorageFeature.isEnabled()) {
+            executeFixInternal(key);
+            return;
+        }
+        final boolean localTransaction = HibernateUtil.startLocalTransactionIfNeeded();
+        try {
+            executeFixInternal(key);
+            if (localTransaction) {
+                HibernateUtil.commitTransaction();
+            }
+        } catch (DotDataException | DotSecurityException | RuntimeException failure) {
+            if (localTransaction) {
+                HibernateUtil.rollbackTransaction();
+            }
+            throw failure;
+        }
+    }
+
+    private void executeFixInternal(final String key) throws DotDataException, DotSecurityException {
         DotConnect dc = new DotConnect();
         // Get information from IR.
         final String getResultsQuery = new StringBuilder("SELECT ")
@@ -200,6 +228,11 @@ private void fixContentletConflicts(final Map<String, Object> contentletData,
         final ContentletAPI contentletAPI = APILocator.getContentletAPI();
         final ContentletJsonAPI contentletJsonAPI = APILocator.getContentletJsonAPI();
 
+        if (AssetStorageFeature.isEnabled()) {
+            new DotConnect().setSQL("select inode from contentlet where inode in (?, ?) order by inode for update")
+                    .addParam(localWorkingInode).addParam(localLiveInode).loadObjectResults();
+        }
+
         User systemUser = APILocator.getUserAPI().getSystemUser();
         Contentlet existingWorkingContentlet = contentletAPI.find(localWorkingInode, systemUser,
                 false);
@@ -249,6 +282,10 @@ private void fixContentletConflicts(final Map<String, Object> contentletData,
         workingCopy.setLanguageId(languageId);
         workingCopy.setModDate(new Date());
 
+        if (AssetStorageFeature.isEnabled() && structureTypeId == Structure.STRUCTURE_TYPE_FILEASSET) {
+            copyStoredBinaries(existingWorkingContentlet, workingCopy);
+        }
+
         final String workingCopyJson = Try.of(() -> contentletJsonAPI.toJson(workingCopy))
                 .getOrElseThrow(() -> new DotRuntimeException(
                         String.format("Error converting contentlet to json from local working copy with inode [%s].", localWorkingInode)));
@@ -286,6 +323,10 @@ private void fixContentletConflicts(final Map<String, Object> contentletData,
             liveCopy.setLanguageId(languageId);
             liveCopy.setModDate(new Date());
 
+            if (AssetStorageFeature.isEnabled() && structureTypeId == Structure.STRUCTURE_TYPE_FILEASSET) {
+                copyStoredBinaries(existingLiveContentlet, liveCopy);
+            }
+
             final String liveCopyJson = Try.of(() -> contentletJsonAPI.toJson(liveCopy))
                     .getOrElseThrow(() -> new DotDataException(
                             String.format("Error converting contentlet to json from local live copy with inode [%s].", localLiveInode)));
@@ -467,6 +508,79 @@ private void fixContentletConflicts(final Map<String, Object> contentletData,
 
         // Remove the Lucene index for the old page
         cleanIndex(existingWorkingContentlet, existingLiveContentlet);
+        if (AssetStorageFeature.isEnabled() && structureTypeId == Structure.STRUCTURE_TYPE_FILEASSET) {
+            BinaryAssetCleanupProcessor.enqueue(localWorkingInode);
+            if (UtilMethods.isSet(localLiveInode) && UtilMethods.isSet(remoteLiveInode)
+                    && !localLiveInode.equals(localWorkingInode)) {
+                BinaryAssetCleanupProcessor.enqueue(localLiveInode);
+            }
+        }
+    }
+
+    /**
+     * Copies each stored binary and its metadata to new revision keys owned by the repaired content,
+     * before its corrected JSON is published; source cleanup is committed with the repair. The copies
+     * are uploaded immediately, so each one registers a rollback listener that removes it again if
+     * the repair transaction rolls back.
+     *
+     * @param source      the local content whose binaries are copied
+     * @param destination the repaired content that receives the copies
+     * @throws DotDataException if a source, its referenced metadata or a copy cannot be stored
+     */
+    private void copyStoredBinaries(final Contentlet source, final Contentlet destination)
+            throws DotDataException {
+        final var metadata = APILocator.getFileMetadataAPI();
+        final var files = APILocator.getFileStorageAPI();
+        final String group = Config.getStringProperty(StoragePersistenceProvider.METADATA_GROUP_NAME,
+                FileMetadataAPI.DOT_METADATA);
+        for (final var field : source.getContentType().fields(com.dotcms.contenttype.model.field.BinaryField.class)) {
+            if (source.get(field.variable()) == null) {
+                continue;
+            }
+            try {
+                final File original = source.getBinary(field.variable());
+                if (original == null) {
+                    throw new DotDataException("Missing repair source " + source.getInode() + "/" + field.variable());
+                }
+                File copied = APILocator.getBinaryAssetStorageAPI().storeRevision(destination.getInode(),
+                        field.variable(), original.getName(), original);
+                final String revision = BinaryAssetReference.keyOf(copied, destination.getInode(), field.variable());
+                removeOnRollback(revision, () -> APILocator.getBinaryAssetStorageAPI()
+                        .deleteBinaryPaths(destination.getInode(), field.variable(), List.of(revision)));
+                final var attributes = files.retrieveRawMetaData(new StorageKey.Builder().group(group)
+                        .path(metadata.getFileName(source, field.variable()))
+                        .storage(StoragePersistenceProvider.getStorageType()).build());
+                if (attributes != null) {
+                    final String key = BinaryAssetReference.newMetadataKey(copied, destination.getInode(), field.variable());
+                    final var metadataKey = new FetchMetadataParams.Builder().cache(false)
+                            .storageKey(new StorageKey.Builder().group(group).path(key)
+                                    .storage(StoragePersistenceProvider.getStorageType()).build()).build();
+                    removeOnRollback(key, () -> files.removeMetaData(metadataKey));
+                    if (!files.setMetadata(metadataKey, attributes)) {
+                        throw new DotDataException("Unable to copy repair metadata");
+                    }
+                    copied = BinaryAssetReference.withMetadata(copied, destination.getInode(), field.variable(), key);
+                } else if (BinaryAssetReference.metadataKeyOf(original) != null) {
+                    throw new DotDataException("Missing referenced repair metadata");
+                }
+                destination.setBinary(field.variable(), copied);
+            } catch (IOException e) {
+                throw new DotDataException("Unable to copy repair source " + source.getInode(), e);
+            }
+        }
+    }
+
+    /**
+     * Registers a rollback listener that deletes an object a repair uploaded before its transaction
+     * rolled back. A failed deletion is logged rather than thrown, so it cannot mask the rollback.
+     *
+     * @param key    the uploaded object's key, for the log message
+     * @param delete deletes the object
+     */
+    private static void removeOnRollback(final String key, final io.vavr.CheckedRunnable delete) {
+        HibernateUtil.addRollbackListener(() -> Try.run(delete).onFailure(failure -> Logger.warn(
+                ContentFileAssetIntegrityChecker.class, "Unable to remove " + key
+                        + " after the integrity repair rolled back: " + failure.getMessage())));
     }
 
     private void fixFileAssetContainer(final String oldContentletIdentifier, final String newContentletIdentifier,
@@ -522,6 +636,13 @@ private Contentlet generateNewContentlet(Contentlet existingContentlet,
             final String remoteInode) throws DotContentletStateException, DotRuntimeException,
             DotSecurityException, DotDataException {
 
+        if (AssetStorageFeature.isEnabled() && structureTypeId == Structure.STRUCTURE_TYPE_FILEASSET) {
+            com.dotmarketing.business.CacheLocator.getContentletCache().remove(remoteInode);
+            final Contentlet repaired = APILocator.getContentletAPI().find(remoteInode, APILocator.systemUser(), false);
+            APILocator.getContentletIndexAPI().addContentToIndex(repaired);
+            return repaired;
+        }
+
         // If its an asset file, move the asset to a new location
         if (structureTypeId == Structure.STRUCTURE_TYPE_FILEASSET) {
             moveInodeFolder(existingContentlet, remoteInode);
diff --git a/dotCMS/src/main/java/com/dotcms/storage/AssetStorageFeature.java b/dotCMS/src/main/java/com/dotcms/storage/AssetStorageFeature.java
index 7417a755a469..af53e866af69 100644
--- a/dotCMS/src/main/java/com/dotcms/storage/AssetStorageFeature.java
+++ b/dotCMS/src/main/java/com/dotcms/storage/AssetStorageFeature.java
@@ -1,5 +1,6 @@
 package com.dotcms.storage;
 
+import com.dotcms.storage.binary.BinaryAssetBackfillProcessor;
 import com.dotcms.storage.binary.BinaryAssetCleanupProcessor;
 import com.dotcms.storage.binary.BinaryFieldCleanupProcessor;
 import com.dotmarketing.util.Config;
@@ -18,8 +19,8 @@ public final class AssetStorageFeature {
     public static final String FLAG = "FEATURE_FLAG_S3_ASSET_STORAGE";
 
     /** Job processors that belong to the S3 lifecycle and must not register while it is disabled. */
-    private static final Set<Class<?>> JOB_PROCESSORS = Set.of(BinaryAssetCleanupProcessor.class,
-            BinaryFieldCleanupProcessor.class);
+    private static final Set<Class<?>> JOB_PROCESSORS = Set.of(BinaryAssetBackfillProcessor.class,
+            BinaryAssetCleanupProcessor.class, BinaryFieldCleanupProcessor.class);
 
     private static volatile Boolean enabled;
 
diff --git a/dotCMS/src/main/java/com/dotcms/storage/binary/BinaryAssetBackfill.java b/dotCMS/src/main/java/com/dotcms/storage/binary/BinaryAssetBackfill.java
new file mode 100644
index 000000000000..eb152d73b1b6
--- /dev/null
+++ b/dotCMS/src/main/java/com/dotcms/storage/binary/BinaryAssetBackfill.java
@@ -0,0 +1,209 @@
+package com.dotcms.storage.binary;
+
+import com.dotcms.business.CloseDBIfOpened;
+import com.dotcms.content.business.json.ContentletJsonHelper;
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotcms.storage.FileMetadataAPI;
+import com.dotcms.storage.StorageKey;
+import com.dotcms.storage.StoragePersistenceProvider;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.common.db.DotConnect;
+import com.dotmarketing.db.HibernateUtil;
+import com.dotmarketing.exception.DotDataException;
+import com.dotmarketing.exception.DotSecurityException;
+import com.dotmarketing.portlets.contentlet.model.Contentlet;
+import com.dotmarketing.util.Config;
+import com.dotmarketing.util.Logger;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import java.io.File;
+import java.nio.file.NoSuchFileException;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+/** Resumable copy of referenced originals and metadata; never removes the local source. */
+public final class BinaryAssetBackfill {
+    private BinaryAssetBackfill() { }
+
+    /**
+     * The outcome of one batch.
+     *
+     * @param afterInode    the last inode visited, which is the cursor for the next batch
+     * @param binaries      the number of binaries verified in durable storage
+     * @param complete      whether the scan reached the end of the contentlet table
+     * @param skippedInodes inodes left incomplete because a referenced binary or metadata record
+     *                      exists neither locally nor in S3, or because the content row could not
+     *                      be read; each one was logged
+     */
+    public record Result(String afterInode, int binaries, boolean complete, List<String> skippedInodes) { }
+
+    /**
+     * Verifies one page of content rows in inode order. Problems in the data itself, which a retry
+     * cannot fix, are logged and reported in {@link Result#skippedInodes()} instead of failing the
+     * batch: a binary or metadata record that exists neither locally nor in S3, and a row whose JSON
+     * cannot be parsed or whose legacy content cannot be found. This matches the legacy tolerance of
+     * starter import for starters that do not carry every referenced binary. Storage and database
+     * errors still fail the batch, so a broken bucket is never mistaken for a missing file.
+     *
+     * @param afterInode     exclusive inode cursor; an empty string starts from the beginning
+     * @param limit          the number of content rows to visit, from 1 to 1000
+     * @param afterEachInode called after each visited inode, for example to send a job heartbeat
+     * @return the batch outcome
+     * @throws DotDataException if storage or the database fails; the message names the inode
+     */
+    @CloseDBIfOpened
+    public static Result runBatch(final String afterInode, final int limit, final Runnable afterEachInode)
+            throws DotDataException {
+        if (!AssetStorageFeature.isEnabled()) {
+            throw new IllegalStateException("S3 asset storage is disabled");
+        }
+        if (afterInode == null || limit < 1 || limit > 1000) {
+            throw new IllegalArgumentException("Provide an inode cursor and a batch size between 1 and 1000");
+        }
+        // The limit must be in the SQL: DotConnect.setMaxRows only trims rows that were already fetched.
+        final var rows = new DotConnect().setSQL("select inode from contentlet where inode > ? order by inode", limit)
+                .addParam(afterInode).loadObjectResults();
+        String cursor = afterInode;
+        int binaries = 0;
+        final Set<String> skipped = new LinkedHashSet<>();
+        for (final var row : rows) {
+            final String inode = row.get("inode").toString();
+            binaries += copyInode(inode, skipped);
+            cursor = inode;
+            afterEachInode.run();
+        }
+        return new Result(cursor, binaries, rows.size() < limit, List.copyOf(skipped));
+    }
+
+    /**
+     * Verifies every content row; starter import calls this before its commit. Skipped inodes are
+     * logged individually by {@link #runBatch} and counted in a summary at the end.
+     *
+     * @throws DotDataException if storage or the database fails
+     */
+    public static void runAll() throws DotDataException {
+        String cursor = "";
+        int skipped = 0;
+        while (true) {
+            final Result result = runBatch(cursor, 250, () -> { });
+            skipped += result.skippedInodes().size();
+            Logger.info(BinaryAssetBackfill.class, "S3 backfill verified " + result.binaries()
+                    + " binaries through inode " + result.afterInode());
+            if (result.complete()) {
+                if (skipped > 0) {
+                    Logger.warn(BinaryAssetBackfill.class, "S3 backfill skipped " + skipped + " content versions"
+                            + " whose binaries or metadata exist neither locally nor in S3, or whose rows could"
+                            + " not be read; the preceding warnings name each inode");
+                }
+                return;
+            }
+            cursor = result.afterInode();
+        }
+    }
+
+    /**
+     * Verifies the originals, metadata and renditions of one content row while holding its row lock.
+     * Data problems that a retry cannot fix are logged and add the inode to {@code skipped}; the other
+     * fields of the row are still copied.
+     *
+     * @param inode   the content version to copy
+     * @param skipped receives the inode when part of it had to be skipped
+     * @return the number of binaries verified
+     * @throws DotDataException if storage or the database fails
+     */
+    private static int copyInode(final String inode, final Set<String> skipped) throws DotDataException {
+        final boolean localTransaction = HibernateUtil.startLocalTransactionIfNeeded();
+        try {
+            // Coordinate with content updates/deletion, including the durable cleanup job.
+            final var rows = new DotConnect().setSQL("select contentlet_as_json from contentlet where inode = ? for update")
+                    .addParam(inode).loadObjectResults();
+            int copied = 0;
+            if (!rows.isEmpty()) {
+                final Object json = rows.get(0).get("contentlet_as_json");
+                final Contentlet content = readContent(inode, json);
+                if (content == null) {
+                    skip(inode, null, "its content row could not be read", skipped);
+                } else {
+                    final var metadataAPI = APILocator.getFileMetadataAPI();
+                    final var binaryAPI = APILocator.getBinaryAssetStorageAPI();
+                    final String snapshot = json == null || json.toString().isBlank()
+                            ? APILocator.getContentletJsonAPI().toJson(content) : json.toString();
+                    try (var lease = binaryAPI.acquireCacheLease()) {
+                        for (final var field : BinaryAssetReference.fromContentJson(snapshot, inode).entrySet()) {
+                            final var reference = field.getValue();
+                            final File file = reference.localFile(inode, field.getKey());
+                            if (!file.isFile()) {
+                                // Restores from S3. A storage error propagates; absence everywhere is skipped.
+                                try (var ignored = binaryAPI.openLocalFile(file)) {
+                                } catch (NoSuchFileException absent) {
+                                    skip(inode, field.getKey(), "the binary exists neither locally nor in S3", skipped);
+                                    continue;
+                                }
+                            }
+                            if (!binaryAPI.backfillBinary(inode, field.getKey(), file)) {
+                                throw new DotDataException("No durable copy for " + inode + "/" + field.getKey());
+                            }
+                            content.getMap().put(field.getKey(), file);
+                            final boolean metadataCopied = APILocator.getFileStorageAPI().backfillMetadata(new StorageKey.Builder()
+                                    .group(Config.getStringProperty(StoragePersistenceProvider.METADATA_GROUP_NAME, FileMetadataAPI.DOT_METADATA))
+                                    .path(metadataAPI.getFileName(content, field.getKey()))
+                                    .storage(StoragePersistenceProvider.getStorageType()).build());
+                            if (!metadataCopied && BinaryAssetReference.metadataKeyOf(file) != null) {
+                                skip(inode, field.getKey(), "its referenced metadata exists neither locally nor in S3", skipped);
+                            }
+                            copied++;
+                        }
+                        binaryAPI.backfillGeneratedFiles(inode);
+                    }
+                }
+            }
+            if (localTransaction) {
+                HibernateUtil.commitTransaction();
+            }
+            return copied;
+        } catch (Exception failure) {
+            if (localTransaction) {
+                HibernateUtil.rollbackTransaction();
+            }
+            throw new DotDataException("Unable to backfill content " + inode, failure);
+        }
+    }
+
+    /**
+     * Reads a locked content row.
+     *
+     * @param inode the content version
+     * @param json  the persisted {@code contentlet_as_json}, or null for a legacy row
+     * @return the content, or null when the JSON cannot be parsed or a legacy row has no content
+     * @throws DotDataException     if the database fails
+     * @throws DotSecurityException if the system user cannot read the content
+     */
+    private static Contentlet readContent(final String inode, final Object json)
+            throws DotDataException, DotSecurityException {
+        if (json == null || json.toString().isBlank()) {
+            final Contentlet found = APILocator.getContentletAPI().find(inode, APILocator.systemUser(), false);
+            return found == null ? null : new Contentlet(found);
+        }
+        try {
+            return APILocator.getContentletJsonAPI().toMutableContentlet(
+                    ContentletJsonHelper.INSTANCE.get().immutableFromJson(json.toString()));
+        } catch (JsonProcessingException unreadable) {
+            Logger.warn(BinaryAssetBackfill.class, "Invalid content JSON for " + inode + ": " + unreadable.getMessage());
+            return null;
+        }
+    }
+
+    /**
+     * Logs a data problem that a retry cannot fix and records its inode for the caller's report.
+     *
+     * @param inode   the affected content version
+     * @param field   the affected field, or null for the whole row
+     * @param reason  why it was skipped, phrased to follow "because"
+     * @param skipped receives the inode
+     */
+    private static void skip(final String inode, final String field, final String reason, final Set<String> skipped) {
+        Logger.warn(BinaryAssetBackfill.class, "S3 backfill skipped " + inode
+                + (field == null ? "" : "/" + field) + " because " + reason);
+        skipped.add(inode);
+    }
+}
diff --git a/dotCMS/src/main/java/com/dotcms/storage/binary/BinaryAssetBackfillProcessor.java b/dotCMS/src/main/java/com/dotcms/storage/binary/BinaryAssetBackfillProcessor.java
new file mode 100644
index 000000000000..8e91a9b1ea53
--- /dev/null
+++ b/dotCMS/src/main/java/com/dotcms/storage/binary/BinaryAssetBackfillProcessor.java
@@ -0,0 +1,179 @@
+package com.dotcms.storage.binary;
+
+import com.dotcms.business.CloseDBIfOpened;
+import com.dotcms.jobs.business.error.JobProcessingException;
+import com.dotcms.jobs.business.error.JobValidationException;
+import com.dotcms.jobs.business.job.Job;
+import com.dotcms.jobs.business.processor.Cancellable;
+import com.dotcms.jobs.business.processor.ExponentialBackoffRetryPolicy;
+import com.dotcms.jobs.business.processor.JobProcessor;
+import com.dotcms.jobs.business.processor.ProgressTracker;
+import com.dotcms.jobs.business.processor.Queue;
+import com.dotcms.jobs.business.processor.Validator;
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.business.CacheLocator;
+import com.dotmarketing.common.db.DotConnect;
+import com.dotmarketing.db.DbConnectionFactory;
+import com.dotmarketing.exception.DotDataException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.enterprise.context.Dependent;
+
+/** Admin-triggered migration using the existing job API, retries and persisted job parameters. */
+@Dependent
+@Queue(BinaryAssetBackfillProcessor.QUEUE)
+@ExponentialBackoffRetryPolicy
+public class BinaryAssetBackfillProcessor implements JobProcessor, Validator, Cancellable {
+    public static final String QUEUE = "binaryAssetBackfill";
+    private static final ObjectMapper JSON = new ObjectMapper();
+    // prune: the job row keeps the first 1000 skipped inodes; the log names every one. Page them if operators need more.
+    private static final int MAX_REPORTED_SKIPS = 1000;
+    private final AtomicBoolean cancelled = new AtomicBoolean();
+    private String afterInode = "";
+    private long verifiedBinaries;
+    private long skippedCount;
+    private List<String> skippedInodes = List.of();
+    private boolean complete;
+
+    /**
+     * Checks that the flag is on, the submitting user is an active administrator and the resume
+     * parameters (cursor, counters and the skipped inode report) are well formed.
+     *
+     * @param parameters the submitted or persisted job parameters
+     * @throws JobValidationException if any check fails
+     */
+    @Override
+    @CloseDBIfOpened
+    public void validate(final Map<String, Object> parameters) throws JobValidationException {
+        if (!AssetStorageFeature.isEnabled()) {
+            throw new JobValidationException("S3 asset storage is disabled");
+        }
+        try {
+            final Object userId = parameters.get("userId");
+            if (!(userId instanceof String) || ((String) userId).isBlank()) {
+                throw new JobValidationException("An administrator userId is required");
+            }
+            final var user = APILocator.getUserAPI().loadUserById((String) userId);
+            if (user == null || !user.isActive() || !user.isAdmin()) {
+                throw new JobValidationException("Only an active administrator can backfill all assets");
+            }
+            final int size = batchSize(parameters);
+            final String cursor = (String) parameters.getOrDefault("afterInode", "");
+            if (size < 1 || size > 1000 || cursor == null || !cursor.matches("[A-Za-z0-9_-]*")) {
+                throw new JobValidationException("Provide a valid inode cursor and a batch size between 1 and 1000");
+            }
+            if (Long.parseLong(parameters.getOrDefault("verifiedBinaries", 0).toString()) < 0) {
+                throw new JobValidationException("The verified binary count cannot be negative");
+            }
+            if (Long.parseLong(parameters.getOrDefault("skippedCount", 0).toString()) < 0
+                    || !(parameters.getOrDefault("skippedInodes", List.of()) instanceof List)) {
+                throw new JobValidationException("The skipped inode report is invalid");
+            }
+        } catch (JobValidationException failure) {
+            throw failure;
+        } catch (Exception failure) {
+            throw new JobValidationException("Invalid backfill parameters or submitting user", failure);
+        }
+    }
+
+    private static int batchSize(final Map<String, Object> parameters) {
+        return Integer.parseInt(parameters.getOrDefault("batchSize", 250).toString());
+    }
+
+    /**
+     * Runs batches from the committed checkpoint until the scan completes or the job is cancelled.
+     * Each batch's cursor, counters and skipped inodes are saved with a compare-and-set on the
+     * previous cursor, so an overlapping worker cannot overwrite newer progress. Inodes skipped
+     * because their data is missing or unreadable are reported, not retried; storage errors fail
+     * the batch so the retry policy applies. A heartbeat is sent after every inode.
+     *
+     * @param job the job to run
+     * @throws JobProcessingException if the flag is off, the job is invalid or a batch fails
+     */
+    @Override
+    @CloseDBIfOpened
+    public void process(final Job job) throws JobProcessingException {
+        if (!AssetStorageFeature.isEnabled()) {
+            throw new JobProcessingException(job.id(), "S3 asset storage is disabled");
+        }
+        try {
+            if (DbConnectionFactory.inTransaction()) {
+                throw new DotDataException("Backfill jobs require independent committed checkpoints");
+            }
+            // The manager may retry an older Job instance. Always reload the committed checkpoint.
+            CacheLocator.getJobCache().remove(job);
+            final Job current = APILocator.getJobQueueManagerAPI().getJob(job.id());
+            if (current == null || !QUEUE.equals(current.queueName())) {
+                throw new DotDataException("Backfill job does not exist");
+            }
+            validate(current.parameters());
+            afterInode = (String) current.parameters().getOrDefault("afterInode", "");
+            verifiedBinaries = Long.parseLong(current.parameters().getOrDefault("verifiedBinaries", 0).toString());
+            skippedCount = Long.parseLong(current.parameters().getOrDefault("skippedCount", 0).toString());
+            skippedInodes = ((List<?>) current.parameters().getOrDefault("skippedInodes", List.of()))
+                    .stream().map(String::valueOf).toList();
+            complete = false;
+            final int size = batchSize(current.parameters());
+            while (!cancelled.get()) {
+                if (Thread.currentThread().isInterrupted()) {
+                    throw new DotDataException("Backfill interrupted before completion");
+                }
+                final var batch = BinaryAssetBackfill.runBatch(afterInode, size,
+                        () -> job.progressTracker().ifPresent(ProgressTracker::heartbeat));
+                final long verified = Math.addExact(verifiedBinaries, batch.binaries());
+                final long skipped = Math.addExact(skippedCount, batch.skippedInodes().size());
+                final List<String> reported = new ArrayList<>(skippedInodes);
+                batch.skippedInodes().stream().limit(Math.max(0, MAX_REPORTED_SKIPS - reported.size()))
+                        .forEach(reported::add);
+                final String checkpoint = JSON.writeValueAsString(Map.of(
+                        "afterInode", batch.afterInode(), "verifiedBinaries", verified,
+                        "skippedCount", skipped, "skippedInodes", reported));
+                // Only verified batches advance. A failed/uncertain save retries idempotent copies.
+                // Compare the cursor so an overlapping worker cannot overwrite newer progress.
+                final var saved = new DotConnect().setSQL("update job set parameters = parameters || ?::jsonb, "
+                                + "updated_at = current_timestamp where id = ? and queue_name = ? "
+                                + "and coalesce(parameters->>'afterInode', '') = ? returning id")
+                        .addParam(checkpoint).addParam(job.id()).addParam(QUEUE).addParam(afterInode)
+                        .loadObjectResults();
+                if (saved.size() != 1) {
+                    throw new DotDataException("Backfill checkpoint changed or job was removed");
+                }
+                CacheLocator.getJobCache().remove(job);
+                afterInode = batch.afterInode();
+                verifiedBinaries = verified;
+                skippedCount = skipped;
+                skippedInodes = List.copyOf(reported);
+                complete = batch.complete();
+                if (complete) {
+                    job.progressTracker().ifPresent(tracker -> tracker.updateProgress(1.0f));
+                    return;
+                }
+            }
+        } catch (Exception failure) {
+            throw new JobProcessingException(job.id(), "Unable to backfill binary assets: " + failure.getMessage(), failure);
+        }
+    }
+
+    @Override
+    public void cancel(final Job job) {
+        cancelled.set(true);
+    }
+
+    /**
+     * Reports the last committed cursor and counters, whether the scan completed, and the inodes
+     * skipped because their binaries or metadata were missing everywhere or their rows were unreadable
+     * (the first 1000 of them; {@code skippedCount} has the total).
+     *
+     * @param job the job being reported
+     * @return the result metadata
+     */
+    @Override
+    public Map<String, Object> getResultMetadata(final Job job) {
+        return Map.of("afterInode", afterInode, "verifiedBinaries", verifiedBinaries, "complete", complete,
+                "skippedCount", skippedCount, "skippedInodes", skippedInodes);
+    }
+}
diff --git a/dotCMS/src/main/java/com/dotmarketing/db/DotCMSInitDb.java b/dotCMS/src/main/java/com/dotmarketing/db/DotCMSInitDb.java
index ffeb9440ac15..e1fef2b1c480 100644
--- a/dotCMS/src/main/java/com/dotmarketing/db/DotCMSInitDb.java
+++ b/dotCMS/src/main/java/com/dotmarketing/db/DotCMSInitDb.java
@@ -111,7 +111,12 @@ private static void loadStarterSite() throws Exception{
 		
 	    loadStarterSiteData() ;
 
-        DbConnectionFactory.closeAndCommit();
+        if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+            // Import cleanup is a commit listener: keep its input until durable backfill and DB commit succeed.
+            HibernateUtil.closeAndCommitTransaction();
+        } else {
+            DbConnectionFactory.closeAndCommit();
+        }
 
         removeAnyOldMetadata();
         MaintenanceUtil.flushCache();
diff --git a/dotCMS/src/main/java/com/dotmarketing/portlets/rules/util/RulesImportExportUtil.java b/dotCMS/src/main/java/com/dotmarketing/portlets/rules/util/RulesImportExportUtil.java
index a67b49f51a6c..e2c70b814e53 100644
--- a/dotCMS/src/main/java/com/dotmarketing/portlets/rules/util/RulesImportExportUtil.java
+++ b/dotCMS/src/main/java/com/dotmarketing/portlets/rules/util/RulesImportExportUtil.java
@@ -87,6 +87,9 @@ public void importRules(final File file) throws IOException {
 
 			this.importRules(importer, APILocator.systemUser());
 		} catch (Exception e) {
+			if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+				throw new IOException("Unable to import rules from " + file, e);
+			}
 			Logger.error(this.getClass(), "Error: " + e.getMessage(), e);
 		}
 	}
diff --git a/dotCMS/src/main/java/com/dotmarketing/util/starter/ExportStarterUtil.java b/dotCMS/src/main/java/com/dotmarketing/util/starter/ExportStarterUtil.java
index 01fe083184ee..3e5af8bb17f2 100644
--- a/dotCMS/src/main/java/com/dotmarketing/util/starter/ExportStarterUtil.java
+++ b/dotCMS/src/main/java/com/dotmarketing/util/starter/ExportStarterUtil.java
@@ -481,21 +481,127 @@ private List<FileEntry> getAdditionalDataAsJSON()  {
      * any backup or temporary location.
      *
      * @param zip        The {@link ZipOutputStream} which will receive all the streamed data.
-     * @param fileFilter The {@link FileFilter} being used to get the appropriate asset data.
+     * @param fileFilter The {@link AssetFileNameFilter} being used to get the appropriate asset data.
      */
-    private void getAssets(final ZipOutputStream zip, final FileFilter fileFilter) {
+    private void getAssets(final ZipOutputStream zip, final AssetFileNameFilter fileFilter) throws Exception {
 
         Logger.info(this, "Adding all Assets into compressed file:");
-        // add all our b-tree directories
-        for (int i = 0; i < assetDirs.length; i++) {
-            for (int j = 0; j < assetDirs.length; j++) {
-                final File source = Paths.get(ConfigUtils.getAssetPath() , assetDirs[i] , assetDirs[j]).toFile();
-                addFolderToZip(zip, fileFilter, source);
+        final Set<String> exported = com.dotcms.storage.AssetStorageFeature.isEnabled()
+                ? addStoredAssetsToZip(zip, fileFilter) : Set.of();
+        final FileFilter remainingFiles = exported.isEmpty() ? fileFilter : file -> fileFilter.accept(file)
+                && !exported.contains(Paths.get(ConfigUtils.getAssetPath()).toAbsolutePath().normalize()
+                .relativize(file.toPath().toAbsolutePath().normalize()).toString().replace('\\', '/'));
+        // With S3 storage on, eviction must not delete a file between this walk listing it and reading it.
+        // The walk only reads files that are already local, so the lease does not grow the cache.
+        try (var lease = com.dotcms.storage.AssetStorageFeature.isEnabled()
+                ? APILocator.getBinaryAssetStorageAPI().acquireCacheLease() : null) {
+            // add all our b-tree directories
+            for (int i = 0; i < assetDirs.length; i++) {
+                for (int j = 0; j < assetDirs.length; j++) {
+                    final File source = Paths.get(ConfigUtils.getAssetPath() , assetDirs[i] , assetDirs[j]).toFile();
+                    addFolderToZip(zip, remainingFiles, source);
+                }
+            }
+            // finally add the messages folder
+            final File source = Paths.get(ConfigUtils.getAssetPath() , "messages").toFile();
+            addFolderToZip(zip, remainingFiles, source);
+        }
+    }
+
+    /**
+     * Adds every binary and metadata record that content rows reference, restoring originals from S3
+     * as needed, so content references rather than cache residency determine the durable assets to
+     * export. Rows are paged by inode with the page size in the SQL, and each accepted row's JSON is
+     * loaded on its own, so a page never holds more than one row's JSON. A binary whose stored
+     * metadata says it exceeds the size limit is skipped before it is restored, and the cache lease
+     * is held per binary so eviction can keep trimming the restored files during a long export.
+     *
+     * @param zip    the archive being written
+     * @param filter the export's inode, name and size filter
+     * @return the asset-relative paths written, which the local folder walk then skips
+     * @throws Exception if a referenced binary or its referenced metadata cannot be read
+     */
+    private Set<String> addStoredAssetsToZip(final ZipOutputStream zip, final AssetFileNameFilter filter) throws Exception {
+        final Set<String> exported = new HashSet<>();
+        final var metadataAPI = APILocator.getFileMetadataAPI();
+        final var binaryAPI = APILocator.getBinaryAssetStorageAPI();
+        final ObjectMapper mapper = new ObjectMapper();
+        String afterInode = "";
+        while (true) {
+            // The limit must be in the SQL: DotConnect.setMaxRows only trims rows that were already fetched.
+            final var rows = new DotConnect().setSQL("select inode from contentlet where inode > ? order by inode", 500)
+                    .addParam(afterInode).loadObjectResults();
+            if (rows.isEmpty()) {
+                return exported;
+            }
+            for (final var row : rows) {
+                final String inode = row.get("inode").toString();
+                afterInode = inode;
+                final String inodePath = inode.charAt(0) + "/" + inode.charAt(1) + "/" + inode;
+                if (!filter.accept(new File(ConfigUtils.getAssetPath(), inodePath))) {
+                    continue;
+                }
+                final Contentlet content = new Contentlet(APILocator.getContentletAPI().find(inode, APILocator.systemUser(), false));
+                final var persistedRows = new DotConnect().setSQL("select contentlet_as_json from contentlet where inode = ?")
+                        .addParam(inode).loadObjectResults();
+                final Object persisted = persistedRows.isEmpty() ? null : persistedRows.get(0).get("contentlet_as_json");
+                final String snapshot = persisted == null || persisted.toString().isBlank()
+                        ? APILocator.getContentletJsonAPI().toJson(content) : persisted.toString();
+                for (final var field : com.dotcms.storage.binary.BinaryAssetReference
+                        .fromContentJson(snapshot, inode).entrySet()) {
+                    final File binary = field.getValue().localFile(inode, field.getKey());
+                    content.getMap().put(field.getKey(), binary);
+                    final String metadataPath = metadataAPI.getFileName(content, field.getKey());
+                    final var metadata = APILocator.getFileStorageAPI().retrieveRawMetaData(
+                            new com.dotcms.storage.StorageKey.Builder()
+                                    .group(Config.getStringProperty(com.dotcms.storage.StoragePersistenceProvider.METADATA_GROUP_NAME,
+                                            com.dotcms.storage.FileMetadataAPI.DOT_METADATA))
+                                    .path(metadataPath).storage(com.dotcms.storage.StoragePersistenceProvider.getStorageType()).build());
+                    final Object length = metadata == null ? null
+                            : metadata.get(com.dotcms.storage.model.BasicMetadataFields.LENGTH_META_KEY.key());
+                    if (filter.maxFileSize > 0 && !binary.isFile() && length instanceof Number
+                            && ((Number) length).longValue() > filter.maxFileSize) {
+                        Logger.warn(this, "Skipping " + binary.getAbsolutePath() + ", size: " + length
+                                + ", Max allowed: " + filter.maxFileSize);
+                        continue;
+                    }
+                    try (var lease = binaryAPI.acquireCacheLease()) {
+                        if (!binary.isFile()) {
+                            try (var ignored = binaryAPI.openLocalFile(binary)) { }
+                        }
+                        if (!filter.accept(binary)) {
+                            continue;
+                        }
+                        boolean excludedParent = false;
+                        for (File parent = binary.getParentFile(); parent != null
+                                && parent.toPath().startsWith(Paths.get(ConfigUtils.getAssetPath())); parent = parent.getParentFile()) {
+                            if (!filter.accept(parent)) {
+                                excludedParent = true;
+                                break;
+                            }
+                        }
+                        if (excludedParent) {
+                            continue;
+                        }
+                        final String revision = com.dotcms.storage.binary.BinaryAssetReference.keyOf(binary, inode, field.getKey());
+                        final String binaryPath = revision == null
+                                ? inodePath + "/" + field.getKey() + "/" + binary.getName() : revision;
+                        addFileToZip(new FileEntry(ZIP_FILE_ASSETS_FOLDER.get() + "/" + binaryPath, binary), zip);
+                        exported.add(binaryPath);
+                    }
+
+                    if (metadata != null) {
+                        // Metadata's filesystem provider uses lowercase paths, independently of binary names.
+                        final String path = metadataPath.substring(1).toLowerCase(java.util.Locale.ROOT);
+                        addFileToZip(new FileEntry(ZIP_FILE_ASSETS_FOLDER.get() + "/" + path,
+                                mapper.writeValueAsString(metadata)), zip);
+                        exported.add(path);
+                    } else if (com.dotcms.storage.binary.BinaryAssetReference.metadataKeyOf(binary) != null) {
+                        throw new IOException("Missing referenced metadata during starter export: " + metadataPath);
+                    }
+                }
             }
         }
-        // finally add the messages folder
-        final File source = Paths.get(ConfigUtils.getAssetPath() , "messages").toFile();
-        addFolderToZip(zip, fileFilter, source);
     }
 
     private void addFolderToZip(final ZipOutputStream zip, final FileFilter fileFilter, final File source){
@@ -523,6 +629,9 @@ private void addFolderToZip(final ZipOutputStream zip, final FileFilter fileFilt
                 this.addFileToZip(entry, zip);
 
             } catch (Exception e) {
+                if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+                    throw new DotRuntimeException("Unable to export asset: " + file.getPath(), e);
+                }
                 Logger.error(this, String.format("Error processing file path for %s: %s",
                         file.getPath(), e.getMessage()));
             }
@@ -578,38 +687,80 @@ public void streamCompressedAssets(final OutputStream output, boolean includeOld
      *                                    {@code true}.
      * @param includeOldAssets            If absolutely all versions of the assets must be included in the compressed
      *                                    file, set this to {@code true}.
+     * @param maxFileSize                 Asset files larger than this many bytes are left out; zero or less means
+     *                                    no limit.
      */
     private void streamCompressedData(final OutputStream output, boolean includeStarterData,
                                      final boolean includeStarterDataAndAssets, boolean includeAssetsOnly, boolean includeOldAssets, long maxFileSize) {
 
-
-
-
-        try (final ZipOutputStream zip = new ZipOutputStream(new BufferedOutputStream(output))) {
-            if (includeStarterData) {
-                Logger.debug(this, "Including Starter Data");
-                this.getStarterDataAsJSON(zip);
-            }
-            final AssetFileNameFilter fileFilter = includeOldAssets
-                    ? new AssetFileNameFilter(maxFileSize)
-                    : new AssetFileNameFilter(getLiveWorkingBloomFilter(),maxFileSize);
-            if (includeStarterDataAndAssets) {
-                Logger.debug(this, "Including Starter Data and Assets");
-                this.getAssets(zip, fileFilter);
-            }
-            if (includeAssetsOnly) {
-                Logger.debug(this, "Including Assets only");
-                this.getAssets(zip, fileFilter);
+        if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+            // Closing a ZipOutputStream writes the central directory that makes the archive readable, so a
+            // failed export must not close it: the client then receives a truncated, unreadable archive
+            // instead of a valid one that silently lacks the remaining entries.
+            final ZipOutputStream zip = new ZipOutputStream(new BufferedOutputStream(output));
+            try {
+                writeCompressedData(zip, includeStarterData, includeStarterDataAndAssets, includeAssetsOnly,
+                        includeOldAssets, maxFileSize);
+                zip.close();
+            } catch (final Exception e) {
+                throw compressionFailure(e, includeStarterData, includeStarterDataAndAssets, includeAssetsOnly,
+                        includeOldAssets);
             }
+            return;
+        }
+        try (final ZipOutputStream zip = new ZipOutputStream(new BufferedOutputStream(output))) {
+            writeCompressedData(zip, includeStarterData, includeStarterDataAndAssets, includeAssetsOnly,
+                    includeOldAssets, maxFileSize);
         } catch (final Exception e) {
-            Logger.error(this, String.format("An error occurred when generating compressed data file with [ " +
-                    "includeStarterData = %s, includeStarterDataAndAssets = %s, includeAssetsOnly = %s, " +
-                    "includeOldAssets = %s ] : %s", includeStarterData, includeStarterDataAndAssets,
-                    includeAssetsOnly, includeOldAssets, e.getMessage()), e);
-            throw new DotRuntimeException(e);
+            throw compressionFailure(e, includeStarterData, includeStarterDataAndAssets, includeAssetsOnly,
+                    includeOldAssets);
         }
     }
 
+    /**
+     * Writes the requested starter data and assets into the archive without finishing it.
+     *
+     * @param zip The archive being written. The other parameters are those of
+     *            {@link #streamCompressedData(OutputStream, boolean, boolean, boolean, boolean, long)}.
+     * @throws Exception If any data or asset cannot be written.
+     */
+    private void writeCompressedData(final ZipOutputStream zip, boolean includeStarterData,
+                                     final boolean includeStarterDataAndAssets, boolean includeAssetsOnly,
+                                     boolean includeOldAssets, long maxFileSize) throws Exception {
+        if (includeStarterData) {
+            Logger.debug(this, "Including Starter Data");
+            this.getStarterDataAsJSON(zip);
+        }
+        final AssetFileNameFilter fileFilter = includeOldAssets
+                ? new AssetFileNameFilter(maxFileSize)
+                : new AssetFileNameFilter(getLiveWorkingBloomFilter(),maxFileSize);
+        if (includeStarterDataAndAssets) {
+            Logger.debug(this, "Including Starter Data and Assets");
+            this.getAssets(zip, fileFilter);
+        }
+        if (includeAssetsOnly) {
+            Logger.debug(this, "Including Assets only");
+            this.getAssets(zip, fileFilter);
+        }
+    }
+
+    /**
+     * Logs a failed export together with its options and wraps the cause for the caller.
+     *
+     * @param e The failure. The other parameters are the export options, as passed to
+     *          {@link #streamCompressedData(OutputStream, boolean, boolean, boolean, boolean, long)}.
+     * @return The exception to throw.
+     */
+    private DotRuntimeException compressionFailure(final Exception e, boolean includeStarterData,
+                                                   final boolean includeStarterDataAndAssets,
+                                                   boolean includeAssetsOnly, boolean includeOldAssets) {
+        Logger.error(this, String.format("An error occurred when generating compressed data file with [ " +
+                "includeStarterData = %s, includeStarterDataAndAssets = %s, includeAssetsOnly = %s, " +
+                "includeOldAssets = %s ] : %s", includeStarterData, includeStarterDataAndAssets,
+                includeAssetsOnly, includeOldAssets, e.getMessage()), e);
+        return new DotRuntimeException(e);
+    }
+
     /**
      * Generates the name of the compressed dotCMS Starter file. By default, such a file name has the following format:
      * <pre>
@@ -738,15 +889,14 @@ public String toString() {
             "and " +
             "  deleted = false";
 
-    private static final String SELECT_ALL_LIVE_WORKING_CONTENTLETS_COUNT=
-            "select " +
-            "count(*) as my_count " +
-            "from (" + SELECT_ALL_LIVE_WORKING_CONTENTLETS + ") testing ";
-
     @CloseDBIfOpened
     BloomFilter<String> getLiveWorkingBloomFilter()  {
+        final String selection = com.dotcms.storage.AssetStorageFeature.isEnabled()
+                ? SELECT_ALL_LIVE_WORKING_CONTENTLETS.replace("working_inode <> live_inode",
+                        "(live_inode is null or working_inode <> live_inode)")
+                : SELECT_ALL_LIVE_WORKING_CONTENTLETS;
         long contentCount = new DotConnect()
-                .setSQL(SELECT_ALL_LIVE_WORKING_CONTENTLETS_COUNT)
+                .setSQL("select count(*) as my_count from (" + selection + ") testing ")
                 .getInt("my_count");
 
         Logger.info(this.getClass(), "Creating BloomFilter with " + contentCount + " expected size");
@@ -757,7 +907,7 @@ BloomFilter<String> getLiveWorkingBloomFilter()  {
                 0.01);
 
         try (final Connection conn = DbConnectionFactory.getDataSource().getConnection();
-            final PreparedStatement statement = conn.prepareStatement(SELECT_ALL_LIVE_WORKING_CONTENTLETS)){
+            final PreparedStatement statement = conn.prepareStatement(selection)){
             statement.setFetchSize(5000);
             try (final ResultSet rs = statement.executeQuery()) {
                 int i = 0;
diff --git a/dotCMS/src/main/java/com/dotmarketing/util/starter/ImportStarterUtil.java b/dotCMS/src/main/java/com/dotmarketing/util/starter/ImportStarterUtil.java
index 2f264edb43ca..363223744d4c 100644
--- a/dotCMS/src/main/java/com/dotmarketing/util/starter/ImportStarterUtil.java
+++ b/dotCMS/src/main/java/com/dotmarketing/util/starter/ImportStarterUtil.java
@@ -244,8 +244,15 @@ public boolean includeIt(Object obj) {
 
         cleanUpDBFromImport();
 
+        if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+            com.dotcms.storage.binary.BinaryAssetBackfill.runAll();
+        }
         Logger.info(ImportStarterUtil.class, "Done Importing");
-        deleteTempFiles();
+        if (com.dotcms.storage.AssetStorageFeature.isEnabled() && DbConnectionFactory.inTransaction()) {
+            HibernateUtil.addSyncCommitListener(this::deleteTempFiles);
+        } else {
+            deleteTempFiles();
+        }
 
     }
 
@@ -602,8 +609,8 @@ private void doJSONFileImport(final File file, final Object type, final ObjectFi
             final VariantFactory variantFactory = FactoryLocator.getVariantFactory();
             for (int j = 0; j < l.size(); j++) {
                 final Variant v = (Variant) l.get(j);
-                // Skip variants already present to avoid PK violations (the variant table is not
-                // cleared by deleteDotCMS()). We DO import the DEFAULT variant from the starter:
+                // Skip variants already present to avoid PK violations (legacy mode preserves them).
+                // Enabled full replacement clears variants, including DEFAULT, before import:
                 // on a fresh install postgres.sql does not seed it and DefaultVariantInitializer
                 // runs only AFTER the starter import (MainServlet line 129 vs 218), so an experiment
                 // referencing the DEFAULT variant would otherwise fail TrafficProportion validation
@@ -634,6 +641,9 @@ private void doJSONFileImport(final File file, final Object type, final ObjectFi
                     }
 
                     APILocator.getUserAPI().save(u, APILocator.getUserAPI().getSystemUser(), false);
+                } else if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+                    // Cleanup removed these users too; restore their archived values before role links.
+                    FactoryLocator.getUserFactory().save(u);
                 } else {
                     Logger.info(this, "");
                 }
@@ -1078,6 +1088,12 @@ List<File> endsWith(String pattern) {
     }
     private void deleteDotCMS() {
         try {
+            final boolean bulkRestore = com.dotcms.storage.AssetStorageFeature.isEnabled()
+                    && DbConnectionFactory.isPostgres();
+            final Map<String, String> deletionTriggers = Map.of(
+                    "contentlet", "content_versions_check_trigger",
+                    "identifier", "check_child_assets_trigger");
+            final Map<String, String> triggerModes = new HashMap<>();
             /* get a list of all our tables */
             final List<String> _tablesToDelete = new ArrayList<>();
 
@@ -1100,12 +1116,47 @@ private void deleteDotCMS() {
             //these tables should be deleted in this order to avoid conflicts with foreign keys
             _tablesToDelete.addAll(tablesToIgnore);
 
+            if (bulkRestore) {
+                if (!DbConnectionFactory.inTransaction()) {
+                    throw new DotStateException("Starter cleanup requires a transaction");
+                }
+                // Full replacement deletes identifiers after their dependents. These two row hooks
+                // otherwise remove a site's identifier prematurely. Foreign keys remain enforced;
+                // PostgreSQL rolls back the trigger changes together with any failed cleanup.
+                for (final var trigger : deletionTriggers.entrySet()) {
+                    final String mode = new DotConnect().setSQL(
+                            "select tgenabled from pg_trigger where tgrelid = ?::regclass and tgname = ?")
+                            .addParam(trigger.getKey()).addParam(trigger.getValue()).getString("tgenabled");
+                    triggerModes.put(trigger.getKey(), switch (mode) {
+                        case "O" -> "ENABLE";
+                        case "A" -> "ENABLE ALWAYS";
+                        case "R" -> "ENABLE REPLICA";
+                        case "D" -> "DISABLE";
+                        default -> throw new DotStateException("Unknown starter cleanup trigger mode: " + mode);
+                    });
+                    new DotConnect().setSQL("ALTER TABLE " + trigger.getKey()
+                            + " DISABLE TRIGGER " + trigger.getValue()).getResult();
+                }
+            }
             for (final String table : _tablesToDelete) {
                 Logger.info(this, "About to delete all records from " + table);
                 this.deleteTable(table);
                 Logger.info(this, "Deleted all records from " + table);
             }
+            if (bulkRestore) {
+                for (final var trigger : deletionTriggers.entrySet()) {
+                    new DotConnect().setSQL("ALTER TABLE " + trigger.getKey() + " "
+                            + triggerModes.get(trigger.getKey()) + " TRIGGER " + trigger.getValue()).getResult();
+                }
+            }
+            if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+                // Raw replacement invalidates users, roles, templates and all other cached entities.
+                CacheLocator.getCacheAdministrator().flushAll();
+            }
         } catch (HibernateException e) {
+            if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+                throw new DotRuntimeException("Starter cleanup failed", e);
+            }
             Logger.error(this,e.getMessage(),e);
         }
 
@@ -1115,22 +1166,48 @@ private void deleteDotCMS() {
 
     }
 
-    private List<String> getTablesToIgnore() {
+    static List<String> getTablesToIgnore() {
         final List<String> tablesToIgnore = new ArrayList<>();
+        if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+            // Modern workflow tables are not all Hibernate entities. Remove their children before
+            // steps, actions, schemes, roles and content types; steps may reference escalation actions.
+            tablesToIgnore.addAll(List.of("workflow_history", "workflow_comment", "workflowtask_files",
+                    "workflow_task", "workflow_action_mappings", "workflow_action_class_pars",
+                    "workflow_action_class", "workflow_action_step", "workflow_step", "workflow_action",
+                    "workflow_scheme_x_structure", "workflow_scheme"));
+            // The importer handles these outside Hibernate; version pointers precede their assets.
+            tablesToIgnore.addAll(List.of("template_version_info", "container_version_info", "link_version_info",
+                    "template_containers", "template", "dot_containers", "links", "category", "relationship"));
+            tablesToIgnore.addAll(List.of("rule_action_pars", "rule_action", "rule_condition_value",
+                    "rule_condition", "rule_condition_group", "dot_rule", "experiment"));
+        }
         tablesToIgnore.add("cms_layouts_portlets");
         tablesToIgnore.add("layouts_cms_roles");
         tablesToIgnore.add("users_cms_roles");
         tablesToIgnore.add("cms_role");
         tablesToIgnore.add("cms_layout");
 
+        if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+            // These are no longer Hibernate entities. Version pointers precede binaries/content types.
+            tablesToIgnore.add("contentlet_version_info");
+            tablesToIgnore.add("contentlet");
+            tablesToIgnore.add("variant");
+            tablesToIgnore.add("folder");
+        }
         tablesToIgnore.add("structure");
-        tablesToIgnore.add("folder");
+        if (!com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+            tablesToIgnore.add("folder");
+        }
         tablesToIgnore.add("identifier");
         tablesToIgnore.add("inode");
 
         tablesToIgnore.add("user_");
 
         tablesToIgnore.add("company");
+        if (com.dotcms.storage.AssetStorageFeature.isEnabled()) {
+            // Language is imported directly; content, workflow tasks and company reference it.
+            tablesToIgnore.add("language");
+        }
         tablesToIgnore.add("counter");
         tablesToIgnore.add("image");
         tablesToIgnore.add("portlet");
diff --git a/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureLatchTest.java b/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureLatchTest.java
index 7f1e5867ee8d..50e1fbad4be1 100644
--- a/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureLatchTest.java
+++ b/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureLatchTest.java
@@ -3,6 +3,7 @@
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.mockStatic;
 
+import com.dotcms.storage.binary.BinaryAssetBackfillProcessor;
 import com.dotcms.storage.binary.BinaryAssetCleanupProcessor;
 import com.dotcms.storage.binary.BinaryFieldCleanupProcessor;
 import com.dotmarketing.util.Config;
@@ -13,8 +14,8 @@
 import org.mockito.MockedStatic;
 
 class AssetStorageFeatureLatchTest {
-    private static final List<Class<?>> S3_PROCESSORS = List.of(BinaryAssetCleanupProcessor.class,
-            BinaryFieldCleanupProcessor.class);
+    private static final List<Class<?>> S3_PROCESSORS = List.of(BinaryAssetBackfillProcessor.class,
+            BinaryAssetCleanupProcessor.class, BinaryFieldCleanupProcessor.class);
 
     private String previous;
 
diff --git a/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureTest.java b/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureTest.java
index cab7f9a311b1..c16f2540ce3f 100644
--- a/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureTest.java
+++ b/dotCMS/src/test/java/com/dotcms/storage/AssetStorageFeatureTest.java
@@ -72,6 +72,20 @@ void disabledFlagPreventsEvictionAndAllRenditionProviderCalls() throws Exception
         verifyNoInteractions(provider);
     }
 
+    @Test
+    void disabledBackfillDoesNotCallAnyProvider() throws Exception {
+        Config.setProperty(AssetStorageFeature.FLAG, false);
+        final var provider = mock(StoragePersistenceAPI.class);
+        assertFalse(chain(provider).backfillFile(GROUP, "a/b/abc123/fileAsset/File.Txt",
+                root.resolve("missing-file").toFile()));
+        assertFalse(chain(provider).backfillObject(GROUP, "unused", new JsonWriterDelegate(),
+                new JsonReaderDelegate<>(Map.class), new java.util.HashMap<>()));
+        assertFalse(new BinaryAssetStorageAPIImpl(provider).backfillBinary("abc123", "field", null));
+        assertThrows(IllegalStateException.class, () -> com.dotcms.storage.binary.BinaryAssetBackfill.runBatch("", 1, () -> { }));
+        verifyNoInteractions(provider);
+        assertFalse(filesystem().backfillFile(GROUP, "unused", root.resolve("missing-file").toFile()));
+    }
+
     @Test
     void databaseQueryFailuresAreNotReportedAsMissingObjectsWhenEnabled() throws Exception {
         final var connection = mock(java.sql.Connection.class);
diff --git a/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillCheckpointTest.java b/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillCheckpointTest.java
new file mode 100644
index 000000000000..0198f31a69cd
--- /dev/null
+++ b/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillCheckpointTest.java
@@ -0,0 +1,122 @@
+package com.dotcms.storage.binary;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+import static org.mockito.Mockito.*;
+
+import com.dotcms.cluster.business.ServerAPI;
+import com.dotcms.jobs.business.api.JobQueueManagerAPI;
+import com.dotcms.jobs.business.error.JobProcessingException;
+import com.dotcms.jobs.business.queue.PostgresJobQueue;
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.business.UserAPI;
+import com.dotmarketing.db.DbConnectionFactory;
+import com.dotmarketing.exception.DotDataException;
+import com.dotmarketing.startup.runonce.Task250113CreatePostgresJobQueueTables;
+import com.dotmarketing.util.Config;
+import com.liferay.portal.model.User;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+/** Real PostgreSQL checkpoints with controlled copy failures; S3 byte verification has separate tests. */
+class BinaryAssetBackfillCheckpointTest {
+    @Test
+    void checkpointSurvivesReconnectionAndStaleStatusUpdates() throws Exception {
+        final String jdbc = System.getProperty("s3.test.jdbc");
+        assumeTrue(jdbc != null, "Supply s3.test.jdbc for an isolated PostgreSQL database");
+        final String previousFlag = Config.getStringProperty(AssetStorageFeature.FLAG, null);
+        final String schema = "binary_backfill_" + UUID.randomUUID().toString().replace("-", "");
+        try (Connection writer = DriverManager.getConnection(jdbc, "binary-storage-test", "binary-storage-test");
+             Connection observer = DriverManager.getConnection(jdbc, "binary-storage-test", "binary-storage-test");
+             var locator = mockStatic(APILocator.class);
+             var copies = mockStatic(BinaryAssetBackfill.class)) {
+            Config.setProperty(AssetStorageFeature.FLAG, true);
+            writer.createStatement().execute("create schema " + schema);
+            try {
+                writer.createStatement().execute("set search_path to " + schema);
+                observer.createStatement().execute("set search_path to " + schema);
+                DbConnectionFactory.setConnection(writer);
+                new Task250113CreatePostgresJobQueueTables().executeUpgrade();
+                final var server = mock(ServerAPI.class);
+                when(server.readServerId()).thenReturn("backfill-checkpoint-test");
+                locator.when(APILocator::getServerAPI).thenReturn(server);
+                final var admin = mock(User.class);
+                when(admin.isAdmin()).thenReturn(true);
+                when(admin.isActive()).thenReturn(true);
+                final var users = mock(UserAPI.class);
+                when(users.loadUserById("administrator")).thenReturn(admin);
+                locator.when(APILocator::getUserAPI).thenReturn(users);
+                final var queue = new PostgresJobQueue();
+                final var manager = mock(JobQueueManagerAPI.class);
+                when(manager.getJob(anyString())).thenAnswer(call -> queue.getJob(call.getArgument(0)));
+                locator.when(APILocator::getJobQueueManagerAPI).thenReturn(manager);
+                final String id = queue.createJob(BinaryAssetBackfillProcessor.QUEUE,
+                        Map.of("userId", "administrator", "batchSize", 1));
+                final var staleJob = queue.getJob(id).markAsRunning();
+                queue.updateJobStatus(staleJob);
+                copies.when(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()))
+                        .thenReturn(new BinaryAssetBackfill.Result("aa", 2, false, List.of()));
+                copies.when(() -> BinaryAssetBackfill.runBatch(eq("aa"), eq(1), any()))
+                        .thenThrow(new DotDataException("S3 unavailable"))
+                        .thenReturn(new BinaryAssetBackfill.Result("bb", 3, true, List.of()));
+
+                final var interrupted = assertThrows(JobProcessingException.class,
+                        () -> new BinaryAssetBackfillProcessor().process(staleJob));
+                assertEquals("S3 unavailable", interrupted.getCause().getMessage(), interrupted.toString());
+                assertEquals("aa", checkpoint(observer, id, "afterInode"));
+                assertEquals("2", checkpoint(observer, id, "verifiedBinaries"));
+                queue.putJobBackInQueue(staleJob);
+                assertEquals("aa", queue.nextJob().parameters().get("afterInode"));
+
+                // Reconnect and construct another worker, retaining only the database checkpoint.
+                DbConnectionFactory.setConnection(observer);
+                com.dotmarketing.business.CacheLocator.getJobCache().put(staleJob);
+                final var resumed = new BinaryAssetBackfillProcessor();
+                resumed.process(staleJob);
+                assertEquals("bb", checkpoint(writer, id, "afterInode"));
+                assertEquals("5", checkpoint(writer, id, "verifiedBinaries"));
+                copies.verify(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()), times(1));
+                queue.updateJobStatus(staleJob.markAsSuccessful(null));
+                assertEquals("bb", checkpoint(writer, id, "afterInode"),
+                        "The manager's original Job must not overwrite the committed checkpoint");
+                assertEquals("administrator", queue.getJob(id).parameters().get("userId"));
+
+                final String competing = queue.createJob(BinaryAssetBackfillProcessor.QUEUE,
+                        Map.of("userId", "administrator", "batchSize", 1));
+                copies.when(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any())).thenAnswer(call -> {
+                    try (var update = writer.prepareStatement(
+                            "update job set parameters = parameters || '{\"afterInode\":\"cc\",\"verifiedBinaries\":4}'::jsonb where id = ?")) {
+                        update.setString(1, competing);
+                        assertEquals(1, update.executeUpdate());
+                    }
+                    return new BinaryAssetBackfill.Result("aa", 2, false, List.of());
+                });
+                assertThrows(JobProcessingException.class,
+                        () -> new BinaryAssetBackfillProcessor().process(queue.getJob(competing)));
+                assertEquals("cc", checkpoint(writer, competing, "afterInode"));
+                assertEquals("4", checkpoint(writer, competing, "verifiedBinaries"));
+            } finally {
+                writer.createStatement().execute("drop schema " + schema + " cascade");
+            }
+        } finally {
+            DbConnectionFactory.closeConnection();
+            Config.setProperty(AssetStorageFeature.FLAG, previousFlag);
+        }
+    }
+
+    private String checkpoint(Connection connection, String id, String key) throws Exception {
+        try (var query = connection.prepareStatement("select parameters->>? from job where id = ?")) {
+            query.setString(1, key);
+            query.setString(2, id);
+            try (var rows = query.executeQuery()) {
+                assertTrue(rows.next());
+                return rows.getString(1);
+            }
+        }
+    }
+}
diff --git a/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillProcessorTest.java b/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillProcessorTest.java
new file mode 100644
index 000000000000..e637f50faeb8
--- /dev/null
+++ b/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillProcessorTest.java
@@ -0,0 +1,218 @@
+package com.dotcms.storage.binary;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import com.dotcms.jobs.business.api.JobQueueManagerAPI;
+import com.dotcms.jobs.business.error.JobProcessingException;
+import com.dotcms.jobs.business.error.JobValidationException;
+import com.dotcms.jobs.business.job.Job;
+import com.dotcms.jobs.business.job.JobState;
+import com.dotcms.jobs.business.processor.ProgressTracker;
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.business.UserAPI;
+import com.dotmarketing.common.db.DotConnect;
+import com.dotmarketing.db.DbConnectionFactory;
+import com.dotmarketing.exception.DotDataException;
+import com.dotmarketing.util.Config;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.liferay.portal.model.User;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+class BinaryAssetBackfillProcessorTest {
+    private final AtomicReference<Job> persisted = new AtomicReference<>();
+    private final User admin = mock(User.class);
+    private MockedStatic<APILocator> locator;
+    private MockedStatic<DbConnectionFactory> connections;
+    private MockedStatic<BinaryAssetBackfill> copies;
+    private String previousFlag;
+
+    @BeforeEach
+    void configure() throws Exception {
+        previousFlag = Config.getStringProperty(AssetStorageFeature.FLAG, null);
+        Config.setProperty(AssetStorageFeature.FLAG, true);
+        locator = mockStatic(APILocator.class);
+        connections = mockStatic(DbConnectionFactory.class);
+        copies = mockStatic(BinaryAssetBackfill.class);
+        final var users = mock(UserAPI.class);
+        when(users.loadUserById("administrator")).thenReturn(admin);
+        when(admin.isAdmin()).thenReturn(true);
+        when(admin.isActive()).thenReturn(true);
+        locator.when(APILocator::getUserAPI).thenReturn(users);
+        persisted.set(Job.builder().id("backfill-test").queueName(BinaryAssetBackfillProcessor.QUEUE)
+                .state(JobState.RUNNING).parameters(Map.of("userId", "administrator", "batchSize", 1)).build());
+        final var jobs = mock(JobQueueManagerAPI.class);
+        when(jobs.getJob("backfill-test")).thenAnswer(call -> persisted.get());
+        locator.when(APILocator::getJobQueueManagerAPI).thenReturn(jobs);
+    }
+
+    @AfterEach
+    void close() {
+        copies.close();
+        connections.close();
+        locator.close();
+        Config.setProperty(AssetStorageFeature.FLAG, previousFlag);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"copy", "before-commit", "after-commit"})
+    void newWorkerResumesCommittedCheckpointAfterFailure(String failure) throws Exception {
+        final Job originalRequest = persisted.get();
+        copies.when(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()))
+                .thenReturn(new BinaryAssetBackfill.Result("aa", 2, false, List.of()));
+        if (failure.equals("copy")) {
+            copies.when(() -> BinaryAssetBackfill.runBatch(eq("aa"), eq(1), any()))
+                    .thenThrow(new DotDataException("S3 unavailable"))
+                    .thenReturn(new BinaryAssetBackfill.Result("bb", 3, true, List.of()));
+        } else {
+            copies.when(() -> BinaryAssetBackfill.runBatch(eq("aa"), eq(1), any()))
+                    .thenReturn(new BinaryAssetBackfill.Result("bb", 3, true, List.of()));
+        }
+        try (var checkpoints = checkpointStore(failure)) {
+            assertThrows(JobProcessingException.class,
+                    () -> new BinaryAssetBackfillProcessor().process(originalRequest));
+            final String expectedCursor = failure.equals("before-commit") ? "" : "aa";
+            assertEquals(expectedCursor, persisted.get().parameters().getOrDefault("afterInode", ""));
+            final var restarted = new BinaryAssetBackfillProcessor();
+            restarted.process(originalRequest); // Intentionally stale job: the new worker must reload it.
+            assertEquals(Map.of("afterInode", "bb", "verifiedBinaries", 5L, "complete", true,
+                    "skippedCount", 0L, "skippedInodes", List.of()),
+                    restarted.getResultMetadata(originalRequest));
+            assertEquals("administrator", persisted.get().parameters().get("userId"));
+            assertEquals(1, persisted.get().parameters().get("batchSize"));
+            copies.verify(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()),
+                    times(failure.equals("before-commit") ? 2 : 1));
+        }
+    }
+
+    @Test
+    void cancellationKeepsLastCompletedBatchAndDoesNotClaimCompletion() throws Exception {
+        final Job request = persisted.get();
+        final var processor = new BinaryAssetBackfillProcessor();
+        copies.when(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any())).thenAnswer(call -> {
+            processor.cancel(request);
+            return new BinaryAssetBackfill.Result("aa", 2, false, List.of());
+        });
+        try (var checkpoints = checkpointStore("none")) {
+            processor.process(request);
+            assertEquals(Map.of("afterInode", "aa", "verifiedBinaries", 2L, "complete", false,
+                    "skippedCount", 0L, "skippedInodes", List.of()),
+                    processor.getResultMetadata(request));
+            assertEquals("aa", persisted.get().parameters().get("afterInode"));
+            copies.verify(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()));
+            copies.verifyNoMoreInteractions();
+        }
+    }
+
+    @Test
+    void lostCheckpointRaceStopsWithoutAdvancing() throws Exception {
+        copies.when(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()))
+                .thenReturn(new BinaryAssetBackfill.Result("aa", 2, false, List.of()));
+        try (var queries = mockConstruction(DotConnect.class, withSettings().defaultAnswer(RETURNS_SELF),
+                (query, context) -> when(query.loadObjectResults()).thenReturn(List.of()))) {
+            final var processor = new BinaryAssetBackfillProcessor();
+            assertThrows(JobProcessingException.class, () -> processor.process(persisted.get()));
+            assertEquals("", processor.getResultMetadata(persisted.get()).get("afterInode"));
+            assertFalse(persisted.get().parameters().containsKey("afterInode"));
+            copies.verify(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()));
+            copies.verifyNoMoreInteractions();
+        }
+    }
+
+    @Test
+    void skippedInodesDoNotStopTheScanAndSurviveARestart() throws Exception {
+        final Job request = persisted.get();
+        copies.when(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any()))
+                .thenReturn(new BinaryAssetBackfill.Result("aa", 0, false, List.of("aa")));
+        copies.when(() -> BinaryAssetBackfill.runBatch(eq("aa"), eq(1), any()))
+                .thenThrow(new DotDataException("S3 unavailable"))
+                .thenReturn(new BinaryAssetBackfill.Result("bb", 1, false, List.of("bb")));
+        copies.when(() -> BinaryAssetBackfill.runBatch(eq("bb"), eq(1), any()))
+                .thenReturn(new BinaryAssetBackfill.Result("cc", 2, true, List.of()));
+        try (var checkpoints = checkpointStore("none")) {
+            assertThrows(JobProcessingException.class, () -> new BinaryAssetBackfillProcessor().process(request));
+            final var restarted = new BinaryAssetBackfillProcessor();
+            restarted.process(request);
+            assertEquals(Map.of("afterInode", "cc", "verifiedBinaries", 3L, "complete", true,
+                    "skippedCount", 2L, "skippedInodes", List.of("aa", "bb")), restarted.getResultMetadata(request));
+        }
+    }
+
+    @Test
+    void everyVisitedInodeSendsAHeartbeat() throws Exception {
+        final var tracker = mock(ProgressTracker.class);
+        final Job request = Job.builder().from(persisted.get()).progressTracker(tracker).build();
+        copies.when(() -> BinaryAssetBackfill.runBatch(eq(""), eq(1), any())).thenAnswer(call -> {
+            final Runnable afterEachInode = call.getArgument(2);
+            afterEachInode.run();
+            afterEachInode.run();
+            return new BinaryAssetBackfill.Result("aa", 2, true, List.of());
+        });
+        try (var checkpoints = checkpointStore("none")) {
+            new BinaryAssetBackfillProcessor().process(request);
+        }
+        verify(tracker, times(2)).heartbeat();
+    }
+
+    @Test
+    void disabledAndUnauthorizedJobsCannotCopyOrCheckpoint() throws Exception {
+        final var processor = new BinaryAssetBackfillProcessor();
+        Config.setProperty(AssetStorageFeature.FLAG, false);
+        assertThrows(JobValidationException.class, () -> processor.validate(persisted.get().parameters()));
+        assertThrows(JobProcessingException.class, () -> processor.process(persisted.get()));
+        locator.verifyNoInteractions();
+        Config.setProperty(AssetStorageFeature.FLAG, true);
+        when(admin.isAdmin()).thenReturn(false);
+        assertThrows(JobValidationException.class, () -> processor.validate(persisted.get().parameters()));
+        assertThrows(JobProcessingException.class, () -> processor.process(persisted.get()));
+        when(admin.isAdmin()).thenReturn(true);
+        when(admin.isActive()).thenReturn(false);
+        assertThrows(JobProcessingException.class, () -> processor.process(persisted.get()));
+        when(admin.isActive()).thenReturn(true);
+        assertThrows(JobValidationException.class, () -> processor.validate(Map.of("userId", "administrator", "batchSize", 0)));
+        assertThrows(JobValidationException.class, () -> processor.validate(Map.of("userId", "administrator", "afterInode", "../other")));
+        connections.when(DbConnectionFactory::inTransaction).thenReturn(true);
+        assertThrows(JobProcessingException.class, () -> processor.process(persisted.get()));
+        copies.verifyNoInteractions();
+    }
+
+    private MockedConstruction<DotConnect> checkpointStore(String failure) {
+        final AtomicBoolean firstSave = new AtomicBoolean(true);
+        return mockConstruction(DotConnect.class, withSettings().defaultAnswer(RETURNS_SELF), (query, context) -> {
+            final List<String> arguments = new ArrayList<>();
+            when(query.addParam(anyString())).thenAnswer(call -> {
+                arguments.add(call.getArgument(0));
+                return query;
+            });
+            when(query.loadObjectResults()).thenAnswer(call -> {
+                final boolean first = firstSave.getAndSet(false);
+                if (first && failure.equals("before-commit")) {
+                    throw new DotDataException("checkpoint write failed");
+                }
+                assertEquals("backfill-test", arguments.get(1));
+                assertEquals(BinaryAssetBackfillProcessor.QUEUE, arguments.get(2));
+                assertEquals(persisted.get().parameters().getOrDefault("afterInode", ""), arguments.get(3));
+                final var parameters = new HashMap<>(persisted.get().parameters());
+                parameters.putAll(new ObjectMapper().readValue(arguments.get(0), Map.class));
+                persisted.set(Job.builder().from(persisted.get()).parameters(parameters).build());
+                if (first && failure.equals("after-commit")) {
+                    throw new DotDataException("connection lost after checkpoint committed");
+                }
+                return List.of(Map.of("id", "backfill-test"));
+            });
+        });
+    }
+}
diff --git a/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillTest.java b/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillTest.java
new file mode 100644
index 000000000000..78cd63a5fbfc
--- /dev/null
+++ b/dotCMS/src/test/java/com/dotcms/storage/binary/BinaryAssetBackfillTest.java
@@ -0,0 +1,136 @@
+package com.dotcms.storage.binary;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import com.dotcms.content.business.json.ContentletJsonAPI;
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotcms.storage.FileMetadataAPI;
+import com.dotcms.storage.FileStorageAPI;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.common.db.DotConnect;
+import com.dotmarketing.db.DbConnectionFactory;
+import com.dotmarketing.db.HibernateUtil;
+import com.dotmarketing.exception.DotDataException;
+import com.dotmarketing.portlets.contentlet.business.ContentletAPI;
+import com.dotmarketing.portlets.contentlet.model.Contentlet;
+import com.dotmarketing.util.Config;
+import com.dotmarketing.util.ConfigUtils;
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+/** Batch paging and the skip policy shared by the backfill job and starter import, without a database. */
+class BinaryAssetBackfillTest {
+    @TempDir
+    Path assets;
+    private final BinaryAssetStorageAPI binaries = mock(BinaryAssetStorageAPI.class);
+    private final ContentletAPI contents = mock(ContentletAPI.class);
+    private MockedStatic<APILocator> locator;
+    private MockedStatic<ConfigUtils> paths;
+    private MockedStatic<HibernateUtil> transactions;
+    private MockedStatic<DbConnectionFactory> connections;
+    private String previousFlag;
+
+    @BeforeEach
+    void configure() throws Exception {
+        previousFlag = Config.getStringProperty(AssetStorageFeature.FLAG, null);
+        Config.setProperty(AssetStorageFeature.FLAG, true);
+        locator = mockStatic(APILocator.class);
+        paths = mockStatic(ConfigUtils.class);
+        transactions = mockStatic(HibernateUtil.class);
+        connections = mockStatic(DbConnectionFactory.class);
+        paths.when(ConfigUtils::getAssetPath).thenReturn(assets.toString());
+        final var json = mock(ContentletJsonAPI.class);
+        // Legacy rows (no persisted JSON) are serialized from the found content: one Binary field per inode.
+        when(json.toJson(any())).thenAnswer(call -> "{\"fields\":{\"asset\":{\"type\":\"Binary\",\"value\":\""
+                + ((Contentlet) call.getArgument(0)).getInode() + ".txt\"}}}");
+        final var metadata = mock(FileMetadataAPI.class);
+        when(metadata.getFileName(any(), anyString())).thenReturn("/metadata.json");
+        final var files = mock(FileStorageAPI.class);
+        when(files.backfillMetadata(any())).thenReturn(true);
+        when(binaries.acquireCacheLease()).thenReturn(() -> { });
+        when(binaries.backfillBinary(anyString(), anyString(), any())).thenReturn(true);
+        locator.when(APILocator::getContentletJsonAPI).thenReturn(json);
+        locator.when(APILocator::getFileMetadataAPI).thenReturn(metadata);
+        locator.when(APILocator::getFileStorageAPI).thenReturn(files);
+        locator.when(APILocator::getBinaryAssetStorageAPI).thenReturn(binaries);
+        locator.when(APILocator::getContentletAPI).thenReturn(contents);
+    }
+
+    @AfterEach
+    void close() {
+        connections.close();
+        transactions.close();
+        paths.close();
+        locator.close();
+        Config.setProperty(AssetStorageFeature.FLAG, previousFlag);
+    }
+
+    @Test
+    void missingOrUnreadableDataIsSkippedAndReportedWhileTheBatchContinues() throws Exception {
+        found("bb");
+        found("cc");
+        when(binaries.openLocalFile(any())).thenThrow(new NoSuchFileException("absent from S3"));
+        final File local = assets.resolve("c/c/cc/asset/cc.txt").toFile();
+        Files.createDirectories(local.toPath().getParent());
+        Files.writeString(local.toPath(), "local bytes");
+        final AtomicInteger heartbeats = new AtomicInteger();
+
+        try (var queries = rows("aa", "bb", "cc")) {
+            final var result = BinaryAssetBackfill.runBatch("", 3, heartbeats::incrementAndGet);
+
+            // "aa" has no content, "bb" has a binary in neither place, "cc" is copied.
+            assertEquals(new BinaryAssetBackfill.Result("cc", 1, false, List.of("aa", "bb")), result);
+            assertEquals("select inode from contentlet where inode > ? order by inode limit 3",
+                    queries.constructed().get(0).getSQL(), "The page size must be applied by the database");
+            assertEquals(3, heartbeats.get());
+            verify(binaries).backfillBinary("cc", "asset", local);
+            verify(binaries, never()).backfillBinary(eq("bb"), anyString(), any());
+        }
+    }
+
+    @Test
+    void storageErrorsStillFailTheBatchAndNameTheInode() throws Exception {
+        found("bb");
+        when(binaries.openLocalFile(any())).thenThrow(new DotDataException("S3 unavailable"));
+
+        try (var queries = rows("bb")) {
+            final var failure = assertThrows(DotDataException.class,
+                    () -> BinaryAssetBackfill.runBatch("", 3, () -> { }));
+            assertTrue(failure.getMessage().contains("bb"), failure.getMessage());
+            assertEquals("S3 unavailable", failure.getCause().getMessage());
+        }
+    }
+
+    /** Makes a legacy row resolve to content with the given inode. */
+    private void found(final String inode) throws Exception {
+        final Contentlet content = new Contentlet();
+        content.setInode(inode);
+        when(contents.find(eq(inode), any(), eq(false))).thenReturn(content);
+    }
+
+    /** The first query pages the given inodes; each later row lock returns a legacy row without JSON. */
+    private MockedConstruction<DotConnect> rows(final String... inodes) {
+        final List<Map<String, Object>> page = new ArrayList<>();
+        for (final String inode : inodes) {
+            page.add(Map.of("inode", inode));
+        }
+        return mockConstruction(DotConnect.class, withSettings().defaultAnswer(CALLS_REAL_METHODS),
+                (query, context) -> doReturn(context.getCount() == 1 ? page
+                        : List.of(Collections.singletonMap("contentlet_as_json", null)))
+                        .when(query).loadObjectResults());
+    }
+}
diff --git a/dotCMS/src/test/java/com/dotmarketing/util/starter/ExportStarterFailureTest.java b/dotCMS/src/test/java/com/dotmarketing/util/starter/ExportStarterFailureTest.java
new file mode 100644
index 000000000000..10f4ccfacdf9
--- /dev/null
+++ b/dotCMS/src/test/java/com/dotmarketing/util/starter/ExportStarterFailureTest.java
@@ -0,0 +1,100 @@
+package com.dotmarketing.util.starter;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotcms.storage.binary.BinaryAssetStorageAPI;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.common.db.DotConnect;
+import com.dotmarketing.exception.DotRuntimeException;
+import com.dotmarketing.util.Config;
+import com.dotmarketing.util.ConfigUtils;
+import com.dotmarketing.util.ZipUtil;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.zip.ZipException;
+import java.util.zip.ZipFile;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+/**
+ * What a client receives when an asset export fails part way: one file is exported, then reading
+ * the next one fails.
+ */
+class ExportStarterFailureTest {
+    @TempDir
+    Path assets;
+    @TempDir
+    Path downloads;
+    private MockedStatic<APILocator> locator;
+    private MockedStatic<ConfigUtils> paths;
+    private MockedStatic<ZipUtil> zips;
+    private MockedConstruction<DotConnect> queries;
+    private String previousFlag;
+
+    @BeforeEach
+    void configure() throws Exception {
+        previousFlag = Config.getStringProperty(AssetStorageFeature.FLAG, null);
+        Files.createDirectories(assets.resolve("0/0/aa"));
+        Files.writeString(assets.resolve("0/0/aa/one.txt"), "first asset");
+        Files.createDirectories(assets.resolve("0/1/bb"));
+        Files.writeString(assets.resolve("0/1/bb/two.txt"), "second asset");
+        paths = mockStatic(ConfigUtils.class);
+        paths.when(ConfigUtils::getAssetPath).thenReturn(assets.toString());
+        final var binaries = mock(BinaryAssetStorageAPI.class);
+        when(binaries.acquireCacheLease()).thenReturn(() -> { });
+        locator = mockStatic(APILocator.class);
+        locator.when(APILocator::getBinaryAssetStorageAPI).thenReturn(binaries);
+        // No content rows reference binaries, so only the local folder walk exports files.
+        queries = mockConstruction(DotConnect.class, withSettings().defaultAnswer(RETURNS_SELF),
+                (query, context) -> when(query.loadObjectResults()).thenReturn(List.of()));
+        zips = mockStatic(ZipUtil.class, CALLS_REAL_METHODS);
+        zips.when(() -> ZipUtil.addZipEntry(any(), eq("assets/0/1/bb/two.txt"), any(), anyBoolean()))
+                .thenThrow(new IOException("read failed"));
+    }
+
+    @AfterEach
+    void close() {
+        zips.close();
+        queries.close();
+        locator.close();
+        paths.close();
+        Config.setProperty(AssetStorageFeature.FLAG, previousFlag);
+    }
+
+    @Test
+    void failedExportWithS3StorageIsNotAReadableArchive() throws Exception {
+        Config.setProperty(AssetStorageFeature.FLAG, true);
+        final var output = new ByteArrayOutputStream();
+
+        assertThrows(DotRuntimeException.class, () -> new ExportStarterUtil().streamCompressedAssets(output, true, -1));
+
+        assertTrue(output.size() > 0, "The first asset was already streamed to the client");
+        final Path download = Files.write(downloads.resolve("failed.zip"), output.toByteArray());
+        assertThrows(ZipException.class, () -> new ZipFile(download.toFile()).close(),
+                "A partial export must not look like a complete backup");
+    }
+
+    @Test
+    void flagOffExportStillLogsTheFailedFileAndFinishesTheArchive() throws Exception {
+        Config.setProperty(AssetStorageFeature.FLAG, false);
+        final var output = new ByteArrayOutputStream();
+
+        new ExportStarterUtil().streamCompressedAssets(output, true, -1);
+
+        final Path download = Files.write(downloads.resolve("partial.zip"), output.toByteArray());
+        try (var archive = new ZipFile(download.toFile())) {
+            assertEquals(List.of("assets/0/0/aa/one.txt"),
+                    archive.stream().map(entry -> entry.getName()).toList());
+        }
+        locator.verifyNoInteractions();
+    }
+}
diff --git a/dotCMS/src/test/java/com/dotmarketing/util/starter/ImportStarterWorkflowCleanupTest.java b/dotCMS/src/test/java/com/dotmarketing/util/starter/ImportStarterWorkflowCleanupTest.java
new file mode 100644
index 000000000000..d2467aadd605
--- /dev/null
+++ b/dotCMS/src/test/java/com/dotmarketing/util/starter/ImportStarterWorkflowCleanupTest.java
@@ -0,0 +1,150 @@
+package com.dotmarketing.util.starter;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotmarketing.util.Config;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class ImportStarterWorkflowCleanupTest {
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void invalidRulesAbortEnabledImport(boolean enabled,
+            @org.junit.jupiter.api.io.TempDir java.nio.file.Path directory) throws Exception {
+        final String previousFlag = Config.getStringProperty(AssetStorageFeature.FLAG, null);
+        final var invalid = java.nio.file.Files.writeString(directory.resolve("rules.json"), "{").toFile();
+        try {
+            Config.setProperty(AssetStorageFeature.FLAG, enabled);
+            final org.junit.jupiter.api.function.Executable importRules = () ->
+                    com.dotmarketing.portlets.rules.util.RulesImportExportUtil.getInstance().importRules(invalid);
+            if (enabled) {
+                assertThrows(java.io.IOException.class, importRules);
+            } else {
+                assertDoesNotThrow(importRules);
+            }
+        } finally {
+            Config.setProperty(AssetStorageFeature.FLAG, previousFlag);
+        }
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void populatedWorkflowGraphDeletesInDependencyOrderOnlyWhenEnabled(boolean enabled) throws Exception {
+        final String previousFlag = Config.getStringProperty(AssetStorageFeature.FLAG, null);
+        // Match the workflow foreign keys in postgres.sql, including escalation and task references.
+        final List<String> tables = List.of("cms_role", "structure", "folder", "contentlet", "contentlet_version_info", "company", "language",
+                "inode", "identifier", "template", "template_containers", "template_version_info",
+                "dot_containers", "container_version_info", "links", "link_version_info", "category", "relationship",
+                "workflow_scheme", "workflow_action",
+                "workflow_step", "workflow_task", "workflow_history", "workflow_comment", "workflowtask_files",
+                "workflow_action_class", "workflow_action_class_pars", "workflow_action_step",
+                "workflow_scheme_x_structure", "workflow_action_mappings");
+        try (var connection = DriverManager.getConnection("jdbc:h2:mem:" + UUID.randomUUID());
+             var sql = connection.createStatement()) {
+            Config.setProperty(AssetStorageFeature.FLAG, enabled);
+            for (String ddl : """
+                    create table cms_role(id int primary key);
+                    create table language(id int primary key);
+                    create table company(id int primary key, default_language_id int references language(id));
+                    create table inode(inode int primary key);
+                    create table identifier(id int primary key);
+                    create table template(inode int primary key references inode(inode), identifier int references identifier(id));
+                    create table dot_containers(inode int primary key references inode(inode));
+                    create table links(inode int primary key references inode(inode));
+                    create table template_containers(template_id int references identifier(id));
+                    create table template_version_info(working_inode int references template(inode), live_inode int references template(inode));
+                    create table container_version_info(working_inode int references dot_containers(inode), live_inode int references dot_containers(inode));
+                    create table link_version_info(working_inode int references links(inode), live_inode int references links(inode));
+                    create table category(inode int primary key references inode(inode));
+                    create table relationship(inode int primary key references inode(inode));
+                    create table structure(inode int primary key);
+                    create table folder(inode int primary key, default_file_type int references structure(inode));
+                    create table contentlet(inode int primary key, structure_inode int references structure(inode),
+                        language_id int references language(id));
+                    create table contentlet_version_info(working_inode int references contentlet(inode),
+                        live_inode int references contentlet(inode));
+                    create table workflow_scheme(id int primary key);
+                    create table workflow_action(id int primary key, next_assign int not null references cms_role(id));
+                    create table workflow_step(id int primary key, scheme_id int references workflow_scheme(id),
+                        escalation_action int references workflow_action(id));
+                    create table workflow_task(id int primary key, assigned_to int references cms_role(id),
+                        status int references workflow_step(id));
+                    create table workflow_history(id int primary key, workflowtask_id int references workflow_task(id));
+                    create table workflow_comment(id int primary key, workflowtask_id int references workflow_task(id));
+                    create table workflowtask_files(id int primary key, workflowtask_id int references workflow_task(id));
+                    create table workflow_action_class(id int primary key, action_id int references workflow_action(id));
+                    create table workflow_action_class_pars(id int primary key,
+                        workflow_action_class_id int references workflow_action_class(id));
+                    create table workflow_action_step(action_id int references workflow_action(id),
+                        step_id int references workflow_step(id));
+                    create table workflow_scheme_x_structure(id int primary key, scheme_id int references workflow_scheme(id),
+                        structure_id int references structure(inode));
+                    create table workflow_action_mappings(id int primary key);
+                    insert into cms_role values (1);
+                    insert into language values (1);
+                    insert into company values (1, 1);
+                    insert into inode values (1);
+                    insert into identifier values (1);
+                    insert into template values (1, 1);
+                    insert into dot_containers values (1);
+                    insert into links values (1);
+                    insert into template_containers values (1);
+                    insert into template_version_info values (1, 1);
+                    insert into container_version_info values (1, 1);
+                    insert into link_version_info values (1, 1);
+                    insert into category values (1);
+                    insert into relationship values (1);
+                    insert into structure values (1);
+                    insert into folder values (1, 1);
+                    insert into contentlet values (1, 1, 1);
+                    insert into contentlet_version_info values (1, 1);
+                    insert into workflow_scheme values (1);
+                    insert into workflow_action values (1, 1);
+                    insert into workflow_step values (1, 1, 1);
+                    insert into workflow_task values (1, 1, 1);
+                    insert into workflow_history values (1, 1);
+                    insert into workflow_comment values (1, 1);
+                    insert into workflowtask_files values (1, 1);
+                    insert into workflow_action_class values (1, 1);
+                    insert into workflow_action_class_pars values (1, 1);
+                    insert into workflow_action_step values (1, 1);
+                    insert into workflow_scheme_x_structure values (1, 1, 1);
+                    insert into workflow_action_mappings values (1);
+                    """.split(";")) {
+                if (!ddl.isBlank()) {
+                    sql.execute(ddl);
+                }
+            }
+            if (enabled) {
+                deleteTables(sql, tables);
+                for (String table : tables) {
+                    try (var rows = sql.executeQuery("select count(*) from " + table)) {
+                        assertTrue(rows.next());
+                        assertEquals(0, rows.getInt(1), table + " must be empty before import");
+                    }
+                }
+            } else {
+                final SQLException failure = assertThrows(SQLException.class, () -> deleteTables(sql, tables));
+                assertEquals("23503", failure.getSQLState(), "Legacy cleanup still encounters the recorded foreign key");
+                assertFalse(ImportStarterUtil.getTablesToIgnore().contains("workflow_action"),
+                        "Disabled mode must retain the original cleanup list");
+            }
+        } finally {
+            Config.setProperty(AssetStorageFeature.FLAG, previousFlag);
+        }
+    }
+
+    private static void deleteTables(Statement sql, List<String> fixtureTables) throws SQLException {
+        for (String table : ImportStarterUtil.getTablesToIgnore()) {
+            if (fixtureTables.contains(table)) {
+                sql.executeUpdate("delete from " + table);
+            }
+        }
+    }
+}
diff --git a/dotcms-integration/src/test/java/com/dotcms/Junit5Suite1.java b/dotcms-integration/src/test/java/com/dotcms/Junit5Suite1.java
index c93697fcbfdc..476bb7d06687 100644
--- a/dotcms-integration/src/test/java/com/dotcms/Junit5Suite1.java
+++ b/dotcms-integration/src/test/java/com/dotcms/Junit5Suite1.java
@@ -79,7 +79,8 @@
         FolderBulkDuplicateHeartbeatIT.class,
         com.dotcms.storage.binary.BinaryAssetStorageIntegrationTest.class,
         com.dotcms.storage.binary.ContentletBackupStorageTest.class,
-        com.dotcms.storage.binary.SharedAssetStorageIntegrationTest.class
+        com.dotcms.storage.binary.SharedAssetStorageIntegrationTest.class,
+        com.dotcms.storage.binary.BinaryAssetStarterRestoreTest.class
 })
 public class Junit5Suite1 {
 
diff --git a/dotcms-integration/src/test/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityCheckerTest.java b/dotcms-integration/src/test/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityCheckerTest.java
index a29e1dcd51e8..76ca43f7740d 100644
--- a/dotcms-integration/src/test/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityCheckerTest.java
+++ b/dotcms-integration/src/test/java/com/dotcms/integritycheckers/ContentFileAssetIntegrityCheckerTest.java
@@ -36,12 +36,19 @@
  * Simple test to validate the ContentFileAssetIntegrityChecker
  */
 public class ContentFileAssetIntegrityCheckerTest extends IntegrationTestBase implements IntegrityCheckerTest {
-
+    private String repairEndpoint;
 
     @Before
     public void setup() throws Exception {
         //Setting web app environment
         setUpEnvironment();
+        repairEndpoint = UUID.randomUUID().toString();
+    }
+
+    @org.junit.After
+    public void removeRepairResults() throws DotDataException {
+        new DotConnect().setSQL("delete from fileassets_ir where endpoint_id = ?")
+                .addParam(repairEndpoint).loadResult();
     }
 
     /**
@@ -108,7 +115,7 @@ public void TestFixConflictGeneratesContentletAsJson() throws Exception {
 
         final ContentFileAssetIntegrityChecker integrityChecker = new ContentFileAssetIntegrityChecker();
         final Tuple2<String, String> remoteIdentifierAndInode = introduceConflict(fileAsset,
-                endpointId.get());
+                repairEndpoint);
 
         final String remoteIdentifier = remoteIdentifierAndInode._1();
         final String remoteWorkingInode = remoteIdentifierAndInode._2();
@@ -116,7 +123,7 @@ public void TestFixConflictGeneratesContentletAsJson() throws Exception {
         Logger.debug(this, "remoteIdentifier: " + remoteIdentifier + " remoteWorkingInode: "
                 + remoteWorkingInode);
 
-        integrityChecker.executeFix(endpointId.get());
+        integrityChecker.executeFix(repairEndpoint);
         Assert.assertTrue(validateFix(remoteIdentifier));
     }
 
@@ -144,9 +151,9 @@ public void TestFixConflictWhenContentContainsSingleQuotes() throws Exception {
 
         final ContentFileAssetIntegrityChecker integrityChecker = new ContentFileAssetIntegrityChecker();
         final Tuple2<String, String> remoteIdentifierAndInode = introduceConflict(fileAsset,
-                endpointId.get());
+                repairEndpoint);
 
-        integrityChecker.executeFix(endpointId.get());
+        integrityChecker.executeFix(repairEndpoint);
         Assert.assertTrue(validateFix(remoteIdentifierAndInode._1()));
     }
 
@@ -163,11 +170,11 @@ public void test_execute_identifierColumnsNotNull() throws Exception {
         final Contentlet contentlet = FileAssetDataGen.createFileAsset(folder, "text1FileAsset"+UUID.randomUUID().toString(), ".txt");
         final FileAsset fileAsset = APILocator.getFileAssetAPI().fromContentlet(contentlet);
         final ContentFileAssetIntegrityChecker integrityChecker = new ContentFileAssetIntegrityChecker();
-        final Tuple2<String, String> remoteIdentifierAndInode = introduceConflict(fileAsset ,endpointId.get());
+        final Tuple2<String, String> remoteIdentifierAndInode = introduceConflict(fileAsset, repairEndpoint);
         final String remoteIdentifier = remoteIdentifierAndInode._1();
         final String remoteWorkingInode = remoteIdentifierAndInode._2();
 
-        integrityChecker.executeFix(endpointId.get());
+        integrityChecker.executeFix(repairEndpoint);
 
         try {
             final DotConnect dotConnect = new DotConnect();
@@ -197,4 +204,72 @@ public void test_execute_identifierColumnsNotNull() throws Exception {
             DbConnectionFactory.closeSilently();
         }
     }
+
+    @Test
+    public void s3RepairPreservesColdBinaryAndMetadataThroughRollbackAndCleanup() throws Exception {
+        org.junit.Assume.assumeTrue(Boolean.getBoolean("s3.cms.enabled"));
+        Assert.assertTrue(com.dotcms.storage.AssetStorageFeature.isEnabled());
+        final Folder folder = new FolderDataGen().nextPersisted();
+        final java.nio.file.Path temporary = java.nio.file.Files.createTempDirectory("s3-integrity-");
+        final String field = com.dotmarketing.portlets.fileassets.business.FileAssetAPI.BINARY_FIELD;
+        final var binaries = APILocator.getBinaryAssetStorageAPI();
+        final var metadata = APILocator.getFileMetadataAPI();
+        final String endpoint = UUID.randomUUID().toString();
+        try {
+            final File upload = java.nio.file.Files.writeString(temporary.resolve("Repair-MixedCase.Txt"),
+                    "repair original bytes").toFile();
+            final Contentlet source = new FileAssetDataGen(folder, upload).nextPersisted();
+            metadata.putCustomMetadataAttributes(source, Map.of(field, Map.of("credit", "Repair author")));
+            final File original = source.getBinary(field);
+            final var target = introduceConflict(APILocator.getFileAssetAPI().fromContentlet(source), endpoint);
+            final String sourceInode = source.getInode();
+            final String targetInode = target._2();
+            final String originalKey = com.dotcms.storage.binary.BinaryAssetReference.keyOf(original);
+            Assert.assertTrue(binaries.evictLocalFile(original));
+            java.nio.file.Files.delete(java.nio.file.Path.of(com.dotmarketing.util.ConfigUtils.getAssetPath())
+                    .resolve(metadata.getFileName(source, field).substring(1).toLowerCase(java.util.Locale.ROOT)));
+            com.dotmarketing.business.CacheLocator.getMetadataCache().removeMetadata(metadata.getMetadataCacheKey(source, field));
+            final var checker = new ContentFileAssetIntegrityChecker();
+
+            com.dotmarketing.db.HibernateUtil.startTransaction();
+            try {
+                checker.executeFix(endpoint);
+                final var repaired = APILocator.getContentletAPI().find(targetInode, APILocator.systemUser(), false);
+                Assert.assertNotEquals(originalKey, com.dotcms.storage.binary.BinaryAssetReference.keyOf(repaired.getBinary(field)));
+                Assert.assertEquals("Repair author", metadata.getMetadata(repaired, field).getCustomMeta().get("credit"));
+            } finally {
+                com.dotmarketing.db.HibernateUtil.rollbackTransaction();
+                com.dotmarketing.business.CacheLocator.getContentletCache().remove(targetInode);
+                com.dotmarketing.business.CacheLocator.getContentletCache().remove(sourceInode);
+            }
+            Assert.assertEquals(originalKey, com.dotcms.storage.binary.BinaryAssetReference.find(sourceInode, field));
+            Assert.assertTrue("A rolled-back repair must remove the revisions it uploaded",
+                    binaries.listBinaryPaths(targetInode).isEmpty());
+            Assert.assertEquals(0, new DotConnect().setSQL("select count(*) as total from job where queue_name = ? and parameters ->> 'inode' = ?")
+                    .addParam(com.dotcms.storage.binary.BinaryAssetCleanupProcessor.QUEUE).addParam(sourceInode).getInt("total"));
+            Assert.assertTrue(binaries.evictLocalFile(original));
+            Assert.assertEquals("repair original bytes", java.nio.file.Files.readString(source.getBinary(field).toPath()));
+
+            checker.executeFix(endpoint);
+            final var repaired = APILocator.getContentletAPI().find(targetInode, APILocator.systemUser(), false);
+            Assert.assertEquals(target._1(), repaired.getIdentifier());
+            final var jobRows = new DotConnect().setSQL("select id from job where queue_name = ? and parameters ->> 'inode' = ?")
+                    .addParam(com.dotcms.storage.binary.BinaryAssetCleanupProcessor.QUEUE).addParam(sourceInode).loadObjectResults();
+            Assert.assertEquals(1, jobRows.size());
+            new com.dotcms.storage.binary.BinaryAssetCleanupProcessor().process(APILocator.getJobQueueManagerAPI()
+                    .getJob(jobRows.getFirst().get("id").toString()));
+            Assert.assertTrue(binaries.listBinaryPaths(sourceInode).isEmpty());
+            Assert.assertTrue(binaries.evictLocalFile(repaired.getBinary(field)));
+            java.nio.file.Files.delete(java.nio.file.Path.of(com.dotmarketing.util.ConfigUtils.getAssetPath())
+                    .resolve(metadata.getFileName(repaired, field).substring(1).toLowerCase(java.util.Locale.ROOT)));
+            com.dotmarketing.business.CacheLocator.getMetadataCache().removeMetadata(metadata.getMetadataCacheKey(repaired, field));
+            Assert.assertEquals("Repair-MixedCase.Txt", repaired.getBinary(field).getName());
+            Assert.assertEquals("repair original bytes", java.nio.file.Files.readString(repaired.getBinary(field).toPath()));
+            Assert.assertEquals("Repair author", metadata.getMetadata(repaired, field).getCustomMeta().get("credit"));
+        } finally {
+            new DotConnect().setSQL("delete from fileassets_ir where endpoint_id = ?").addParam(endpoint).loadResult();
+            FolderDataGen.remove(folder);
+            org.apache.commons.io.FileUtils.deleteDirectory(temporary.toFile());
+        }
+    }
 }
diff --git a/dotcms-integration/src/test/java/com/dotcms/storage/binary/BinaryAssetStarterRestoreTest.java b/dotcms-integration/src/test/java/com/dotcms/storage/binary/BinaryAssetStarterRestoreTest.java
new file mode 100644
index 000000000000..c9a2101b874c
--- /dev/null
+++ b/dotcms-integration/src/test/java/com/dotcms/storage/binary/BinaryAssetStarterRestoreTest.java
@@ -0,0 +1,260 @@
+package com.dotcms.storage.binary;
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.dotcms.datagen.FileAssetDataGen;
+import com.dotcms.datagen.FolderDataGen;
+import com.dotcms.storage.AssetStorageFeature;
+import com.dotcms.util.IntegrationTestInitService;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.business.CacheLocator;
+import com.dotmarketing.portlets.fileassets.business.FileAssetAPI;
+import com.dotmarketing.util.ConfigUtils;
+import com.dotmarketing.util.starter.ExportStarterUtil;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Locale;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Run alone in two fresh disposable integration harnesses, first with s3.starter.phase=export,
+ * then with s3.starter.phase=restore and starter.run.path pointing at s3.starter.archive.
+ * The S3 bucket and exported archive survive between runs; the database and local cache do not.
+ */
+@EnabledIfSystemProperty(named = "s3.starter.restore.enabled", matches = "true")
+public class BinaryAssetStarterRestoreTest {
+
+    @Test
+    @EnabledIfSystemProperty(named = "s3.starter.phase", matches = "export")
+    void exportColdStarterAndRemoveSourceObjects(
+            @TempDir final Path temporary) throws Exception {
+        IntegrationTestInitService.getInstance().init();
+        assertTrue(AssetStorageFeature.isEnabled());
+        final var client = AmazonS3ClientBuilder.standard()
+                .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
+                        System.getProperty("DOT_STORAGE_FILE_METADATA_S3_ENDPOINT"), "us-east-1"))
+                .withPathStyleAccessEnabled(true)
+                .withCredentials(com.dotmarketing.util.UtilMethods.isSet(System.getProperty("DOT_STORAGE_FILE_METADATA_S3_ACCESS_KEY"))
+                        ? new AWSStaticCredentialsProvider(new BasicAWSCredentials(
+                        System.getProperty("DOT_STORAGE_FILE_METADATA_S3_ACCESS_KEY"),
+                        System.getProperty("DOT_STORAGE_FILE_METADATA_S3_SECRET_ACCESS_KEY")))
+                        : new com.amazonaws.auth.DefaultAWSCredentialsProviderChain())
+                .build();
+        try {
+            final String bucket = System.getProperty("DOT_STORAGE_FILE_METADATA_S3_BUCKET_NAME");
+            final String field = FileAssetAPI.BINARY_FIELD;
+            final var metadata = APILocator.getFileMetadataAPI();
+            final var binaries = APILocator.getBinaryAssetStorageAPI();
+            final var configuration = createConfiguration("Starter", true);
+            final var source = Files.writeString(temporary.resolve("Starter-MixedCase.Txt"), "starter restore original bytes");
+            final var content = new FileAssetDataGen(new FolderDataGen().nextPersisted(), source.toFile()).nextPersisted();
+            metadata.putCustomMetadataAttributes(content, Map.of(field, Map.of("credit", "Starter author")));
+            final String inode = content.getInode();
+            final var binary = content.getBinary(field);
+            final String binaryKey = BinaryAssetReference.keyOf(binary);
+            final String metadataPath = metadata.getFileName(content, field);
+            final String remoteBinary = BinaryAssetStorageAPI.BINARY_ASSETS_GROUP + "/" + binaryKey;
+            final var localMetadata = Path.of(ConfigUtils.getAssetPath())
+                    .resolve(metadataPath.substring(1).toLowerCase(Locale.ROOT));
+            assertNotNull(binaryKey);
+            assertTrue(binaries.evictLocalFile(binary));
+            Files.delete(localMetadata);
+            CacheLocator.getMetadataCache().removeMetadata(metadata.getMetadataCacheKey(content, field));
+
+            final Path archive = Path.of(System.getProperty("s3.starter.archive"));
+            Files.createDirectories(archive.getParent());
+            try (final var output = Files.newOutputStream(archive)) {
+                new ExportStarterUtil().streamCompressedStarter(output, true, true, -1);
+            }
+            assertTrue(Files.size(archive) > 0);
+
+            // Remove only this test asset's durable objects and cache after the archive is complete.
+            // A successful post-import read must therefore come from the import, not the source bucket.
+            binaries.deleteAllBinaries(inode);
+            com.dotcms.storage.StoragePersistenceProvider.INSTANCE.get().getStorage()
+                    .deleteObjectAndReferences(com.dotcms.storage.FileMetadataAPI.DOT_METADATA, metadataPath);
+            Files.deleteIfExists(localMetadata);
+            assertFalse(client.doesObjectExist(bucket, remoteBinary));
+            assertFalse(binary.exists());
+            assertFalse(com.dotcms.storage.StoragePersistenceProvider.INSTANCE.get()
+                    .getStorage(com.dotcms.storage.StorageType.S3)
+                    .existsObject(com.dotcms.storage.FileMetadataAPI.DOT_METADATA, metadataPath));
+            new ObjectMapper().writeValue(archive.resolveSibling("fixture.json").toFile(),
+                    Map.of("inode", inode, "binaryKey", binaryKey, "metadataPath", metadataPath,
+                            "configuration", configuration));
+        } finally {
+            client.shutdown();
+        }
+    }
+
+    @Test
+    @EnabledIfSystemProperty(named = "s3.starter.phase", matches = "restore")
+    void freshStartupRestoresOriginalAndMetadataFromExportedStarter() throws Exception {
+        final Path archive = Path.of(System.getProperty("s3.starter.archive"));
+        assertEquals(archive.toAbsolutePath().toString(), System.getProperty("DOT_STARTER_DATA_LOAD"),
+                "The fresh CMS must actually start from the exported archive");
+        IntegrationTestInitService.getInstance().init();
+        assertRestoredFromArchive(archive);
+    }
+
+    @Test
+    @EnabledIfSystemProperty(named = "s3.starter.phase", matches = "populated")
+    void populatedDatabaseRestoresOriginalAndMetadataFromExportedStarter() throws Exception {
+        final Path archive = Path.of(System.getProperty("s3.starter.archive"));
+        assertNotEquals(archive.toAbsolutePath().toString(), System.getProperty("DOT_STARTER_DATA_LOAD"),
+                "Populate the disposable harness with its normal starter before importing the archive");
+        IntegrationTestInitService.getInstance().init();
+        assertTrue(AssetStorageFeature.isEnabled());
+        assertTrue(new com.dotmarketing.common.db.DotConnect()
+                .setSQL("select count(*) as total from workflow_action").getInt("total") > 0,
+                "This case must exercise an existing workflow graph");
+        final var oldTemplate = new com.dotcms.datagen.TemplateDataGen().nextPersisted();
+        final var oldCategory = new com.dotcms.datagen.CategoryDataGen().nextPersisted();
+        assertNotNull(APILocator.getTemplateAPI().find(oldTemplate.getInode(), APILocator.systemUser(), false));
+        assertNotNull(APILocator.getCategoryAPI().find(oldCategory.getInode(), APILocator.systemUser(), false));
+        final var fixture = new ObjectMapper().readTree(archive.resolveSibling("fixture.json").toFile());
+        final var oldConfiguration = createConfiguration("Destination", false);
+        new com.dotcms.datagen.VariantDataGen()
+                .name(fixture.path("configuration").path("variant").asText())
+                .description("Conflicting destination variant").nextPersisted();
+        assertFalse(com.dotcms.storage.AmazonS3StoragePersistenceAPIImpl.withPlainPaths()
+                .existsObject(BinaryAssetStorageAPI.BINARY_ASSETS_GROUP, fixture.path("binaryKey").asText()),
+                "Run the export phase first: the fixture must be absent from S3 before import");
+        assertFalse(com.dotcms.storage.StoragePersistenceProvider.INSTANCE.get()
+                .getStorage(com.dotcms.storage.StorageType.S3)
+                .existsObject(com.dotcms.storage.FileMetadataAPI.DOT_METADATA, fixture.path("metadataPath").asText()),
+                "The fixture's metadata must also be absent before import");
+        final var importer = new com.dotmarketing.util.starter.ImportStarterUtil(archive.toFile());
+        final int existingContent = new com.dotmarketing.common.db.DotConnect()
+                .setSQL("select count(*) as total from contentlet").getInt("total");
+        final var cleanup = importer.getClass().getDeclaredMethod("deleteDotCMS");
+        cleanup.setAccessible(true);
+        com.dotmarketing.db.HibernateUtil.startTransaction();
+        try {
+            new com.dotmarketing.common.db.DotConnect().setSQL(
+                    "create table starter_cleanup_guard (inode varchar(36) references contentlet(inode))")
+                    .getResult();
+            new com.dotmarketing.common.db.DotConnect().setSQL(
+                    "insert into starter_cleanup_guard select inode from contentlet limit 1").getResult();
+            final var failure = assertThrows(java.lang.reflect.InvocationTargetException.class,
+                    () -> cleanup.invoke(importer));
+            assertTrue(failure.getCause().getMessage().contains("starter_cleanup_guard"),
+                    "Foreign keys must remain enforced during full replacement");
+        } finally {
+            com.dotmarketing.db.HibernateUtil.rollbackTransaction();
+        }
+        assertDeletionTriggersEnabled();
+        assertEquals(existingContent, new com.dotmarketing.common.db.DotConnect()
+                .setSQL("select count(*) as total from contentlet").getInt("total"));
+        com.dotmarketing.db.HibernateUtil.startTransaction();
+        try {
+            cleanup.invoke(importer);
+            assertEquals(0, new com.dotmarketing.common.db.DotConnect()
+                    .setSQL("select count(*) as total from contentlet").getInt("total"));
+            assertDeletionTriggersEnabled();
+        } finally {
+            com.dotmarketing.db.HibernateUtil.rollbackTransaction();
+        }
+        assertEquals(existingContent, new com.dotmarketing.common.db.DotConnect()
+                .setSQL("select count(*) as total from contentlet").getInt("total"),
+                "Rolling back cleanup must retain the populated site");
+        assertDeletionTriggersEnabled();
+        importer.doImport();
+        assertDeletionTriggersEnabled();
+        assertRestoredFromArchive(archive);
+        assertTrue(com.dotmarketing.business.FactoryLocator.getVariantFactory()
+                .get(oldConfiguration.get("variant")).isEmpty());
+        assertTrue(com.dotmarketing.business.FactoryLocator.getExperimentsFactory()
+                .find(oldConfiguration.get("experiment")).isEmpty());
+        assertNull(APILocator.getRulesAPI().getRuleById(oldConfiguration.get("rule"), APILocator.systemUser(), false));
+        assertNull(APILocator.getTemplateAPI().find(oldTemplate.getInode(), APILocator.systemUser(), false),
+                "Restored sites must not serve cached templates from the replaced site");
+        assertNull(APILocator.getCategoryAPI().find(oldCategory.getInode(), APILocator.systemUser(), false));
+        assertEquals(0, new com.dotmarketing.common.db.DotConnect()
+                .setSQL("select count(*) as total from template where inode = ?")
+                .addParam(oldTemplate.getInode()).getInt("total"));
+        assertEquals(0, new com.dotmarketing.common.db.DotConnect()
+                .setSQL("select count(*) as total from category where inode = ?")
+                .addParam(oldCategory.getInode()).getInt("total"));
+        assertTrue(new com.dotmarketing.common.db.DotConnect()
+                .setSQL("select count(*) as total from workflow_action").getInt("total") > 0,
+                "The archive must restore its workflows after cleanup");
+    }
+
+    private static void assertDeletionTriggersEnabled() {
+        assertEquals(2, new com.dotmarketing.common.db.DotConnect().setSQL(
+                "select count(*) as total from pg_trigger where tgenabled = 'O' and "
+                        + "((tgrelid = 'contentlet'::regclass and tgname = 'content_versions_check_trigger') "
+                        + "or (tgrelid = 'identifier'::regclass and tgname = 'check_child_assets_trigger'))")
+                .getInt("total"), "Starter cleanup must restore the deletion hooks");
+    }
+
+    private static Map<String, String> createConfiguration(String prefix, boolean archived) throws Exception {
+        final var variant = new com.dotcms.datagen.VariantDataGen().name(prefix + "_Variant")
+                .description(prefix + " variant").archived(archived).nextPersisted();
+        final var experiment = new com.dotcms.datagen.ExperimentDataGen().name(prefix + " experiment").nextPersisted();
+        final var rule = new com.dotmarketing.portlets.rules.RuleDataGen().name(prefix + " rule").nextPersisted();
+        final var group = new com.dotmarketing.portlets.rules.conditionlet.ConditionGroupDataGen()
+                .rule(rule).nextPersisted();
+        final var condition = new com.dotmarketing.portlets.rules.conditionlet.ConditionDataGen().group(group).next();
+        condition.addValue("comparison", "is");
+        condition.addValue("country", "CA");
+        APILocator.getRulesAPI().saveCondition(condition, APILocator.systemUser(), false);
+        final var action = new com.dotmarketing.portlets.rules.actionlet.RuleActionDataGen().rule(rule).next();
+        action.addParameter(new com.dotmarketing.portlets.rules.model.ParameterModel("sessionKey", "starter"));
+        action.addParameter(new com.dotmarketing.portlets.rules.model.ParameterModel("sessionValue", prefix));
+        APILocator.getRulesAPI().saveRuleAction(action, APILocator.systemUser(), false);
+        // Warm the caches too; cleanup must remove persisted records and their cached representations.
+        assertTrue(com.dotmarketing.business.FactoryLocator.getVariantFactory().get(variant.name()).isPresent());
+        assertTrue(com.dotmarketing.business.FactoryLocator.getExperimentsFactory().find(experiment.id().orElseThrow()).isPresent());
+        assertNotNull(APILocator.getRulesAPI().getRuleById(rule.getId(), APILocator.systemUser(), false));
+        return Map.of("variant", variant.name(), "experiment", experiment.id().orElseThrow(),
+                "rule", rule.getId(), "condition", condition.getId(), "action", action.getId());
+    }
+
+    private void assertRestoredFromArchive(final Path archive) throws Exception {
+        final var fixture = new ObjectMapper().readTree(archive.resolveSibling("fixture.json").toFile());
+        assertTrue(AssetStorageFeature.isEnabled());
+        final String inode = fixture.path("inode").asText();
+        final String field = FileAssetAPI.BINARY_FIELD;
+        final String binaryKey = fixture.path("binaryKey").asText();
+        final String metadataPath = fixture.path("metadataPath").asText();
+        final var configuration = fixture.path("configuration");
+        final var variant = com.dotmarketing.business.FactoryLocator.getVariantFactory()
+                .get(configuration.path("variant").asText()).orElseThrow();
+        assertTrue(variant.archived(), "Restore must replace a conflicting active destination variant");
+        assertEquals("Starter variant", variant.description().orElseThrow());
+        assertTrue(com.dotmarketing.business.FactoryLocator.getVariantFactory().get("DEFAULT").isPresent());
+        assertEquals("Starter experiment", com.dotmarketing.business.FactoryLocator.getExperimentsFactory()
+                .find(configuration.path("experiment").asText()).orElseThrow().name());
+        final var rule = APILocator.getRulesAPI().getRuleById(configuration.path("rule").asText(), APILocator.systemUser(), false);
+        assertNotNull(rule);
+        assertEquals("Starter rule", rule.getName());
+        assertEquals(1, rule.getGroups().size());
+        assertEquals(1, rule.getGroups().get(0).getConditions().size());
+        assertEquals("CA", rule.getGroups().get(0).getConditions().get(0).getParameters().get("country").getValue());
+        assertEquals(1, rule.getRuleActions().size());
+        assertEquals("Starter", rule.getRuleActions().get(0).getParameters().get("sessionValue").getValue());
+        final var metadata = APILocator.getFileMetadataAPI();
+        final var binaries = APILocator.getBinaryAssetStorageAPI();
+        assertFalse(Files.exists(Path.of(ConfigUtils.getBackupPath(), "temp")),
+                "Successful import must clean up its extracted input");
+        final var restored = APILocator.getContentletAPI().find(inode, APILocator.systemUser(), false);
+        assertEquals(binaryKey, BinaryAssetReference.find(inode, field));
+        assertEquals(metadataPath, metadata.getFileName(restored, field));
+        assertTrue(binaries.evictLocalFile(restored.getBinary(field)), "Import must publish a verified S3 copy");
+        Files.delete(Path.of(ConfigUtils.getAssetPath()).resolve(metadataPath.substring(1).toLowerCase(Locale.ROOT)));
+        CacheLocator.getMetadataCache().removeMetadata(metadata.getMetadataCacheKey(restored, field));
+        assertEquals("Starter-MixedCase.Txt", restored.getBinary(field).getName());
+        assertEquals("starter restore original bytes", Files.readString(restored.getBinary(field).toPath()));
+        assertEquals("Starter author", metadata.getMetadata(restored, field).getCustomMeta().get("credit"));
+    }
+}
diff --git a/dotcms-integration/src/test/java/com/dotcms/storage/binary/BinaryAssetStorageIntegrationTest.java b/dotcms-integration/src/test/java/com/dotcms/storage/binary/BinaryAssetStorageIntegrationTest.java
index fbd838370595..8957a8c0f404 100644
--- a/dotcms-integration/src/test/java/com/dotcms/storage/binary/BinaryAssetStorageIntegrationTest.java
+++ b/dotcms-integration/src/test/java/com/dotcms/storage/binary/BinaryAssetStorageIntegrationTest.java
@@ -515,6 +515,241 @@ void s3RegenerationPreservesSnapshotsAndRollback(
         }
     }
 
+    @Test
+    void assetExportIncludesBinaryAndMetadataInBothStorageModes(
+            @org.junit.jupiter.api.io.TempDir java.nio.file.Path uploads) throws Exception {
+        final boolean s3Enabled = Boolean.getBoolean("s3.cms.enabled");
+        final Folder folder = new FolderDataGen().nextPersisted();
+        final String field = FileAssetAPI.BINARY_FIELD;
+        final var metadataAPI = APILocator.getFileMetadataAPI();
+        try {
+            final File upload = Files.writeString(uploads.resolve("ColdExport-MixedCase.Txt"), "cold export original bytes").toFile();
+            final Contentlet asset = new FileAssetDataGen(folder, upload).nextPersisted();
+            metadataAPI.putCustomMetadataAttributes(asset,
+                    java.util.Map.of(field, java.util.Map.of("credit", "Preserved in starter")));
+            final File binary = asset.getBinary(field);
+            final String inode = asset.getInode();
+            final String binaryPath = s3Enabled ? BinaryAssetReference.keyOf(binary)
+                    : inode.charAt(0) + "/" + inode.charAt(1) + "/" + inode + "/" + field + "/" + binary.getName();
+            final String metadataPath = metadataAPI.getFileName(asset, field);
+            final java.nio.file.Path localMetadata = java.nio.file.Path.of(com.dotmarketing.util.ConfigUtils.getAssetPath())
+                    .resolve(metadataPath.substring(1).toLowerCase(java.util.Locale.ROOT));
+            if (s3Enabled) {
+                assertTrue(binaryAssetStorageAPI.evictLocalFile(binary));
+                Files.delete(localMetadata);
+                com.dotmarketing.business.CacheLocator.getMetadataCache().removeMetadata(
+                        metadataAPI.getMetadataCacheKey(asset, field));
+                assertFalse(binary.exists());
+                assertFalse(Files.exists(localMetadata));
+            }
+
+            final java.io.ByteArrayOutputStream output = new java.io.ByteArrayOutputStream();
+            new com.dotmarketing.util.starter.ExportStarterUtil().streamCompressedAssets(output, true, -1);
+            final java.util.Map<String, byte[]> entries = readZipEntries(output.toByteArray());
+            assertArrayEquals(Files.readAllBytes(upload.toPath()), entries.get("assets/" + binaryPath));
+            final byte[] metadataBytes = entries.get("assets/" + metadataPath.substring(1).toLowerCase(java.util.Locale.ROOT));
+            assertNotNull(metadataBytes, "The archive must include metadata restored from S3, including custom attributes");
+            assertEquals("Preserved in starter", new com.fasterxml.jackson.databind.ObjectMapper()
+                    .readTree(metadataBytes).path("dot:credit").asText());
+            assertTrue(binary.exists());
+            if (s3Enabled) {
+                assertEquals(1, new com.dotmarketing.common.db.DotConnect()
+                        .setSQL("select count(*) as draft_count from contentlet_version_info where working_inode = ? and live_inode is null")
+                        .addParam(inode).getInt("draft_count"), "Exercise an unpublished working asset");
+                assertTrue(binaryAssetStorageAPI.evictLocalFile(binary));
+                final java.io.ByteArrayOutputStream workingOnly = new java.io.ByteArrayOutputStream();
+                new com.dotmarketing.util.starter.ExportStarterUtil().streamCompressedAssets(workingOnly, false, -1);
+                assertArrayEquals(Files.readAllBytes(upload.toPath()),
+                        readZipEntries(workingOnly.toByteArray()).get("assets/" + binaryPath),
+                        "Working-only exports must include drafts with no live version");
+                binaryAssetStorageAPI.deleteAllBinaries(inode);
+                assertThrows(com.dotmarketing.exception.DotRuntimeException.class,
+                        () -> new com.dotmarketing.util.starter.ExportStarterUtil()
+                                .streamCompressedAssets(new java.io.ByteArrayOutputStream(), true, -1),
+                        "A referenced but missing S3 binary must fail the export instead of silently omitting it");
+            }
+        } finally {
+            FolderDataGen.remove(folder);
+        }
+    }
+
+    @Test
+    void s3LegacyBackfillResumesWithoutChangingReferencesOrLosingLocalSources(
+            @org.junit.jupiter.api.io.TempDir java.nio.file.Path uploads) throws Exception {
+        org.junit.jupiter.api.Assumptions.assumeTrue(Boolean.getBoolean("s3.cms.enabled"));
+        final Folder folder = new FolderDataGen().nextPersisted();
+        final var client = com.amazonaws.services.s3.AmazonS3ClientBuilder.standard()
+                .withEndpointConfiguration(new com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration(
+                        System.getProperty("DOT_STORAGE_FILE_METADATA_S3_ENDPOINT"), "us-east-1"))
+                .withPathStyleAccessEnabled(true)
+                .withCredentials(com.dotmarketing.util.UtilMethods.isSet(System.getProperty("DOT_STORAGE_FILE_METADATA_S3_ACCESS_KEY"))
+                        ? new com.amazonaws.auth.AWSStaticCredentialsProvider(new com.amazonaws.auth.BasicAWSCredentials(
+                        System.getProperty("DOT_STORAGE_FILE_METADATA_S3_ACCESS_KEY"),
+                        System.getProperty("DOT_STORAGE_FILE_METADATA_S3_SECRET_ACCESS_KEY")))
+                        : new com.amazonaws.auth.DefaultAWSCredentialsProviderChain())
+                .build();
+        final String bucket = System.getProperty("DOT_STORAGE_FILE_METADATA_S3_BUCKET_NAME");
+        try {
+            final String field = FileAssetAPI.BINARY_FIELD;
+            final var metadataAPI = APILocator.getFileMetadataAPI();
+            final File upload = Files.writeString(uploads.resolve("Legacy-MixedCase.Txt"), "legacy imported bytes").toFile();
+            final Contentlet created = new FileAssetDataGen(folder, upload).nextPersisted();
+            metadataAPI.putCustomMetadataAttributes(created,
+                    java.util.Map.of(field, java.util.Map.of("credit", "Legacy author")));
+            final String inode = created.getInode();
+            final String prefix = inode.charAt(0) + "/" + inode.charAt(1) + "/" + inode;
+            final String key = prefix + "/" + field + "/" + upload.getName();
+            final var local = java.nio.file.Path.of(com.dotmarketing.util.ConfigUtils.getAssetPath()).resolve(key);
+            Files.createDirectories(local.getParent());
+            Files.copy(upload.toPath(), local);
+            final String metadataPath = "/" + prefix + "/" + field + "-metadata.json";
+            final var localMetadata = java.nio.file.Path.of(com.dotmarketing.util.ConfigUtils.getAssetPath())
+                    .resolve(metadataPath.substring(1).toLowerCase(java.util.Locale.ROOT));
+            final var mapper = new com.fasterxml.jackson.databind.ObjectMapper();
+            Files.writeString(localMetadata, mapper.writeValueAsString(metadataAPI.getFullMetadataNoCache(created, field).getMap()));
+            final var json = (com.fasterxml.jackson.databind.node.ObjectNode) mapper.readTree(
+                    new com.dotmarketing.common.db.DotConnect().setSQL("select contentlet_as_json from contentlet where inode = ?")
+                            .addParam(inode).getString("contentlet_as_json"));
+            ((com.fasterxml.jackson.databind.node.ObjectNode) json.path("fields").path(field)).remove("storageKey");
+            ((com.fasterxml.jackson.databind.node.ObjectNode) json.path("fields").path(field)).remove("metadataStorageKey");
+            new com.dotmarketing.common.db.DotConnect().setSQL("update contentlet set contentlet_as_json = ?::jsonb where inode = ?")
+                    .addParam(mapper.writeValueAsString(json)).addParam(inode).loadResult();
+            com.dotmarketing.business.CacheLocator.getContentletCache().remove(inode);
+            final Contentlet legacy = contentletAPI.find(inode, user, false);
+            assertNull(BinaryAssetReference.find(inode, field));
+            assertFalse(binaryAssetStorageAPI.evictLocalFile(local.toFile()));
+
+            client.putObject(bucket, binaryObjectKey(key), "conflicting destination bytes");
+            assertThrows(com.dotmarketing.exception.DotDataException.class, BinaryAssetBackfill::runAll);
+            assertEquals("legacy imported bytes", Files.readString(local));
+            assertTrue(Files.exists(localMetadata));
+            assertEquals("conflicting destination bytes", client.getObjectAsString(bucket,
+                    binaryObjectKey(key)));
+            client.deleteObject(bucket, binaryObjectKey(key));
+
+            // Existing remote raw bytes must migrate too, including retired field definitions.
+            client.putObject(bucket, binaryObjectKey(key), upload);
+            assertTrue(binaryAssetStorageAPI.evictLocalFile(local.toFile()));
+            final String retiredKey = prefix + "/retiredBinary/Retired-Mixed.Txt";
+            client.putObject(bucket, binaryObjectKey(retiredKey), "retired field bytes");
+            ((com.fasterxml.jackson.databind.node.ObjectNode) json.path("fields")).putObject("retiredBinary")
+                    .put("type", "Binary").put("value", "Retired-Mixed.Txt");
+            new com.dotmarketing.common.db.DotConnect().setSQL("update contentlet set contentlet_as_json = ?::jsonb where inode = ?")
+                    .addParam(mapper.writeValueAsString(json)).addParam(inode).loadResult();
+
+            var progress = BinaryAssetBackfill.runBatch("", 1, () -> { });
+            int copied = progress.binaries();
+            while (!progress.complete()) {
+                final String cursor = progress.afterInode();
+                progress = BinaryAssetBackfill.runBatch(cursor, 1, () -> { });
+                assertTrue(progress.complete() || !cursor.equals(progress.afterInode()));
+                copied += progress.binaries();
+            }
+            assertTrue(copied > 0);
+            BinaryAssetBackfill.runAll();
+            assertEquals(org.apache.commons.codec.digest.DigestUtils.sha256Hex("legacy imported bytes"),
+                    client.getObjectMetadata(bucket, binaryObjectKey(key))
+                            .getUserMetaDataOf("dotcms-blob-sha256"));
+            assertEquals(org.apache.commons.codec.digest.DigestUtils.sha256Hex("retired field bytes"),
+                    client.getObjectMetadata(bucket, binaryObjectKey(retiredKey))
+                            .getUserMetaDataOf("dotcms-blob-sha256"));
+            assertNull(BinaryAssetReference.find(inode, field), "Backfill must preserve legacy content references");
+            assertTrue(binaryAssetStorageAPI.evictLocalFile(local.toFile()));
+            Files.delete(localMetadata);
+            com.dotmarketing.business.CacheLocator.getMetadataCache().removeMetadata(metadataAPI.getMetadataCacheKey(legacy, field));
+            assertEquals("legacy imported bytes", Files.readString(contentletAPI.find(inode, user, false).getBinary(field).toPath()));
+            assertEquals("Legacy author", metadataAPI.getMetadata(legacy, field).getCustomMeta().get("credit"));
+        } finally {
+            client.shutdown();
+            FolderDataGen.remove(folder);
+        }
+    }
+
+    @Test
+    void legacyImageAndFileBytesSurviveMigrationExportAndRecoveryWithoutChangingFieldTypes(
+            @org.junit.jupiter.api.io.TempDir java.nio.file.Path uploads) throws Exception {
+        org.junit.jupiter.api.Assumptions.assumeTrue(Boolean.getBoolean("s3.cms.enabled"));
+        final ContentType type = new ContentTypeDataGen().nextPersisted();
+        String archiveKey = null;
+        final var backups = ContentletBackupStorage.getInstance();
+        try {
+            for (final var field : java.util.Map.<String, Class<? extends Field>>of(
+                    "title", TextField.class,
+                    "legacyImage", com.dotcms.contenttype.model.field.ImageField.class,
+                    "legacyFile", com.dotcms.contenttype.model.field.FileField.class).entrySet()) {
+                ContentTypeDataGen.addField(new FieldDataGen().velocityVarName(field.getKey())
+                        .contentTypeId(type.id()).type(field.getValue()).nextPersisted());
+            }
+            final Contentlet content = new ContentletDataGen(type).setProperty("title", "Legacy file inventory").nextPersisted();
+            final String inode = content.getInode();
+            final String prefix = inode.charAt(0) + "/" + inode.charAt(1) + "/" + inode;
+            final var mapper = new com.fasterxml.jackson.databind.ObjectMapper();
+            final var json = (com.fasterxml.jackson.databind.node.ObjectNode) mapper.readTree(
+                    new com.dotmarketing.common.db.DotConnect().setSQL("select contentlet_as_json from contentlet where inode = ?")
+                            .addParam(inode).getString("contentlet_as_json"));
+            final var fields = (com.fasterxml.jackson.databind.node.ObjectNode) json.path("fields");
+            fields.putObject("legacyImage").put("type", "Image").put("value", "Legacy-Mixed.PNG");
+            fields.putObject("legacyFile").put("type", "File").put("value", "Document-Mixed.Txt");
+            fields.putObject("retiredBinary").put("type", "Binary").put("value", "Retired-Mixed.Txt");
+            fields.putObject("linkedImage").put("type", "Image").put("value", "other-asset-identifier");
+            fields.putObject("externalFile").put("type", "File").put("value", "https://example.com/file.txt");
+            final var expected = new java.util.HashMap<String, byte[]>();
+            final var localFiles = new java.util.ArrayList<java.nio.file.Path>();
+            for (String field : java.util.List.of("legacyImage", "legacyFile", "retiredBinary")) {
+                final String key = prefix + "/" + field + "/" + fields.path(field).path("value").asText();
+                final var local = java.nio.file.Path.of(com.dotmarketing.util.ConfigUtils.getAssetPath(), key);
+                Files.createDirectories(local.getParent());
+                if (field.equals("legacyImage")) {
+                    javax.imageio.ImageIO.write(new java.awt.image.BufferedImage(20, 10,
+                            java.awt.image.BufferedImage.TYPE_INT_RGB), "png", local.toFile());
+                } else {
+                    Files.writeString(local, "original " + field + " bytes");
+                }
+                expected.put("assets/" + key, Files.readAllBytes(local));
+                localFiles.add(local);
+                assertFalse(binaryAssetStorageAPI.evictLocalFile(local.toFile()), "Unmigrated legacy bytes must stay local");
+            }
+            final String metadataPath = prefix + "/legacyimage-metadata.json";
+            final var localMetadata = java.nio.file.Path.of(com.dotmarketing.util.ConfigUtils.getAssetPath(), metadataPath);
+            Files.writeString(localMetadata, "{\"dot:credit\":\"Legacy image author\"}");
+            final String snapshot = mapper.writeValueAsString(json);
+            new com.dotmarketing.common.db.DotConnect().setSQL("update contentlet set contentlet_as_json = ?::jsonb where inode = ?")
+                    .addParam(snapshot).addParam(inode).loadResult();
+            com.dotmarketing.business.CacheLocator.getContentletCache().remove(inode);
+            BinaryAssetBackfill.runAll();
+            assertEquals(json, mapper.readTree(new com.dotmarketing.common.db.DotConnect()
+                    .setSQL("select contentlet_as_json from contentlet where inode = ?").addParam(inode)
+                    .getString("contentlet_as_json")), "Migration must not turn linked-field strings into binary values");
+            assertEquals("Legacy-Mixed.PNG", contentletAPI.find(inode, user, false).get("legacyImage"));
+            for (var local : localFiles) assertTrue(binaryAssetStorageAPI.evictLocalFile(local.toFile()));
+            Files.delete(localMetadata);
+            final var output = new java.io.ByteArrayOutputStream();
+            new com.dotmarketing.util.starter.ExportStarterUtil().streamCompressedAssets(output, true, -1);
+            final var exported = readZipEntries(output.toByteArray());
+            assertEquals("Legacy-Mixed.PNG", contentletAPI.find(inode, user, false).get("legacyImage"),
+                    "Export must not replace a cached Image field's string with a File");
+            for (var entry : expected.entrySet()) assertArrayEquals(entry.getValue(), exported.get(entry.getKey()), entry.getKey());
+            assertEquals("Legacy image author", mapper.readTree(exported.get("assets/" + metadataPath)).path("dot:credit").asText());
+            for (var local : localFiles) assertTrue(binaryAssetStorageAPI.evictLocalFile(local.toFile()));
+            Files.deleteIfExists(localMetadata);
+            com.dotmarketing.db.HibernateUtil.startTransaction();
+            try {
+                archiveKey = backups.store(content);
+                com.dotmarketing.db.HibernateUtil.commitTransaction();
+            } finally { com.dotmarketing.db.HibernateUtil.rollbackTransaction(); }
+            try (var archive = backups.open(archiveKey)) {
+                final var recovered = readZipEntries(archive.readAllBytes());
+                for (var entry : expected.entrySet()) assertArrayEquals(entry.getValue(), recovered.get(entry.getKey()), entry.getKey());
+                assertEquals(json, mapper.readTree(recovered.get("contentlet.json")));
+                assertEquals("Legacy image author", mapper.readTree(recovered.get("assets/" + metadataPath)).path("dot:credit").asText());
+            }
+        } finally {
+            if (archiveKey != null) com.dotcms.storage.AmazonS3StoragePersistenceAPIImpl.withPlainPaths()
+                    .deleteObjectAndReferences(ContentletBackupStorage.GROUP, archiveKey);
+            ContentTypeDataGen.remove(type);
+        }
+    }
+
     private String binaryObjectKey(String path) {
         final String namespace = System.getProperty("DOT_STORAGE_FILE_METADATA_S3_NAMESPACE", "");
         return (namespace.isEmpty() ? "" : "asset-namespaces/" + namespace + "/")