diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
index c957be325..96e2e91ce 100644
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -31,6 +31,10 @@ jobs:
       run: |
         sudo apt install tox
 
+    - name: Install busybox for the test instances' root filesystem
+      shell: bash
+      run: sudo apt-get install -y busybox-static
+
     - name: Build write_data resource binary
       shell: bash
       run: |
diff --git a/coriolis/tests/integration/README.md b/coriolis/tests/integration/README.md
index 6ec090f5c..31efcace3 100644
--- a/coriolis/tests/integration/README.md
+++ b/coriolis/tests/integration/README.md
@@ -40,10 +40,11 @@ devices.
 | `dd`, `sync`, `cmp` | Test-pattern writes and device comparison |
 | `docker` | MariaDB database container; data-minion container image |
 | `ssh-keygen` | Generates the ephemeral SSH key pair used by the test provider |
+| `busybox` (statically linked) | Minimal root filesystem of the core test provider's instances, which `exec_test_vm` chroots into |
 
 On Ubuntu / Debian:
 ```bash
-sudo apt-get install util-linux coreutils
+sudo apt-get install util-linux coreutils busybox-static
 ```
 
 ### Docker image - data-minion
diff --git a/coriolis/tests/integration/base.py b/coriolis/tests/integration/base.py
index 47e389753..424378a3e 100644
--- a/coriolis/tests/integration/base.py
+++ b/coriolis/tests/integration/base.py
@@ -26,6 +26,7 @@
 
 from coriolis import constants, context, exception
 from coriolis.db import api as db_api
+from coriolis.providers import backup_writers
 from coriolis.providers import factory as providers_factory
 from coriolis.tests import test_base
 from coriolis.tests.integration import harness
@@ -414,6 +415,17 @@ def _dst_device(self):
 
         return None
 
+    @property
+    def _transfer_disk_ids(self):
+        """The "disk_id" of every disk in the transfer's volumes_info."""
+        ctxt = self._get_db_context()
+
+        transfer = db_api.get_transfer(ctxt, self._transfer.id, include_task_info=True)
+        info = transfer.get("info", {}).get(self._instance_name, {})
+        return [
+            vol["disk_id"] for vol in info.get("volumes_info", []) if vol.get("disk_id")
+        ]
+
     def _cleanup_provider_dst_devices(self, transfer_id):
         """Remove any devices the provider allocated for this transfer."""
         ctxt = self._get_db_context()
@@ -444,6 +456,35 @@ def _execute_and_wait(self, transfer_id, timeout=600):
         )
         self.assertExecutionCompleted(execution.id, timeout=timeout)
 
+    def _execute_and_spy_on_writer(self, transfer_id):
+        """Execute *transfer_id*, recording what the HTTP backup writer did.
+
+        Returns a ``(checksummed_disk_ids, bytes_written)`` tuple.
+        """
+        checksummed_disks = []
+        written = []
+        writer_cls = backup_writers.HTTPBackupWriterImpl
+        original_get_disk_checksum = writer_cls.get_disk_checksum
+        original_write = writer_cls.write
+
+        def _record_get_disk_checksum(writer, *args, **kwargs):
+            checksummed_disks.append(writer._disk_id)
+            return original_get_disk_checksum(writer, *args, **kwargs)
+
+        def _record_write(writer, data, encoding=None, uncompressed_size=None):
+            written.append(uncompressed_size or len(data))
+            return original_write(writer, data, encoding, uncompressed_size)
+
+        with (
+            mock.patch.object(
+                writer_cls, "get_disk_checksum", _record_get_disk_checksum
+            ),
+            mock.patch.object(writer_cls, "write", _record_write),
+        ):
+            self._execute_and_wait(transfer_id)
+
+        return checksummed_disks, sum(written)
+
     def _start_executions(self, transfer_ids):
         """Start one execution per transfer id without waiting on any."""
         return [
diff --git a/coriolis/tests/integration/osmorphing_utils.py b/coriolis/tests/integration/osmorphing_utils.py
index 2f80e4c4c..b98bad671 100644
--- a/coriolis/tests/integration/osmorphing_utils.py
+++ b/coriolis/tests/integration/osmorphing_utils.py
@@ -10,6 +10,8 @@
 import contextlib
 import json
 import os
+import shutil
+import stat
 import subprocess
 import tempfile
 from typing import Iterator
@@ -70,6 +72,60 @@ def write_os_image_to_disk(device_path, container_image):
         _run(["docker", "rm", "-f", container_id], check=False)
 
 
+def write_busybox_rootfs_to_device(device_path):
+    """Write a minimal, chroot-able busybox rootfs to *device_path*.
+
+    Much smaller than a full OS image: a tiny ext4 filesystem holding only the
+    statically linked ``busybox`` of the host, and its applet links.
+    """
+    busybox = shutil.which("busybox")
+    _run(
+        [
+            "mkfs.ext4",
+            "-F",
+            "-q",
+            "-m",
+            "0",
+            "-N",
+            "1024",
+            "-O",
+            "^has_journal",
+            device_path,
+        ]
+    )
+
+    with mounted(device_path) as mount_point:
+        bin_dir = os.path.join(mount_point, "bin")
+        os.makedirs(bin_dir)
+        shutil.copy(busybox, os.path.join(bin_dir, "busybox"))
+
+        # Relative links, as "busybox --install -s" would point them at the
+        # host's mount point, which does not exist inside the chroot.
+        applets = _run([busybox, "--list"]).stdout.decode().split()
+        for applet in applets:
+            if applet != "busybox":
+                os.symlink("busybox", os.path.join(bin_dir, applet))
+
+        for name in ("dev", "proc", "tmp", "etc"):
+            os.makedirs(os.path.join(mount_point, name))
+
+        for name, minor in (("null", 3), ("zero", 5), ("urandom", 9)):
+            os.mknod(
+                os.path.join(mount_point, "dev", name),
+                0o666 | stat.S_IFCHR,
+                os.makedev(1, minor),
+            )
+
+
+def chroot_exec(device_path, command):
+    """Run the shell *command* chrooted in the filesystem of *device_path*.
+
+    Returns the ``subprocess.CompletedProcess``, without raising on failure.
+    """
+    with mounted(device_path) as mount_point:
+        return _run(["chroot", mount_point, "/bin/sh", "-c", command], check=False)
+
+
 def _fixup_luks_inner_os(mapper_path, luks_uuid):
     """Patch the OS image inside a LUKS mapper to work with OS morphing.
 
diff --git a/coriolis/tests/integration/provider_test_base.py b/coriolis/tests/integration/provider_test_base.py
index b0235c682..7b93b4444 100644
--- a/coriolis/tests/integration/provider_test_base.py
+++ b/coriolis/tests/integration/provider_test_base.py
@@ -19,9 +19,13 @@
 from oslo_log import log as logging
 
 from coriolis import context as coriolis_context
+from coriolis import exception
 
 LOG = logging.getLogger(__name__)
 
+# Writes random data to a file in the current (home) directory, and flushes it.
+_MUTATE_DATA_COMMAND = "dd if=/dev/urandom of=coriolis-marker bs=512 count=8 && sync"
+
 
 class BaseTestExportProvider(abc.ABC):
     """Base test provider to be used by export providers.
@@ -107,6 +111,41 @@ def get_test_instance_device(self, instance_name: str):
         """
         return None
 
+    def can_exec_test_vm(self) -> bool:
+        """Whether ``exec_test_vm`` is implemented.
+
+        The default is False; tests that need to run commands inside the source
+        instance (e.g.: to change its data) are skipped when this is False.
+        """
+        return False
+
+    def exec_test_vm(self, instance_name: str, command: str):
+        """Run the shell *command* inside *instance_name*.
+
+        Returns an ``ssh_utils.ExecResult`` (``returncode``, ``stdout``, ``stderr``).
+        Only called when ``can_exec_test_vm()`` is True. Providers are expected to
+        SSH into the instance, or to otherwise enter its filesystem if it is not
+        reachable (e.g.: chroot for the core test provider).
+        """
+        raise NotImplementedError(
+            "%s does not support executing commands in test instances"
+            % type(self).__name__
+        )
+
+    def mutate_test_instance_data(self, instance_name: str):
+        """Change some data on *instance_name*'s primary disk.
+
+        Called between two executions of the same transfer to force a detectable
+        difference for an incremental sync to pick up. Only called when
+        ``can_exec_test_vm()`` is True.
+        """
+        result = self.exec_test_vm(instance_name, _MUTATE_DATA_COMMAND)
+        if result.returncode != 0:
+            raise exception.CoriolisException(
+                "Failed to mutate the data of instance '%s' (exit code %s): %s"
+                % (instance_name, result.returncode, result.stderr)
+            )
+
 
 class BaseTestImportProvider(abc.ABC):
     def initialize(self, connection_info: dict, source_config: dict):
diff --git a/coriolis/tests/integration/ssh_utils.py b/coriolis/tests/integration/ssh_utils.py
new file mode 100644
index 000000000..1e1b6494d
--- /dev/null
+++ b/coriolis/tests/integration/ssh_utils.py
@@ -0,0 +1,59 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""
+SSH helpers for test providers which need to run commands inside test VMs.
+"""
+
+import collections
+
+import paramiko
+from oslo_log import log as logging
+
+from coriolis import utils
+
+LOG = logging.getLogger(__name__)
+
+# Default guest login, matching the well-known Cirros test image. Can be overridden
+# through ``source.guest_ssh`` in providers.yaml.
+DEFAULT_GUEST_SSH = {"username": "cirros", "password": "gocubsgo"}
+
+ExecResult = collections.namedtuple("ExecResult", ["returncode", "stdout", "stderr"])
+
+
+def get_guest_ssh_config(source_config: dict) -> dict:
+    """Return the guest SSH login: the defaults, updated by ``guest_ssh``.
+
+    Recognized keys are ``username``, ``password``, ``pkey_path`` and ``port``.
+    """
+    return {**DEFAULT_GUEST_SSH, **source_config.get("guest_ssh", {})}
+
+
+def exec_ssh_command(
+    ip: str, command: str, guest_ssh: dict, connect_attempts: int = 10
+) -> ExecResult:
+    """Run *command* on *ip* over SSH, waiting for the guest's SSH port to open.
+
+    *guest_ssh* is a dict as returned by ``get_guest_ssh_config``.
+    """
+    port = guest_ssh.get("port", 22)
+    utils.wait_for_port_connectivity(ip, port)
+
+    ssh = paramiko.SSHClient()
+    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    try:
+        utils.retry_on_error(max_attempts=connect_attempts, sleep_seconds=5)(
+            ssh.connect
+        )(
+            hostname=ip,
+            port=port,
+            username=guest_ssh["username"],
+            password=guest_ssh.get("password"),
+            key_filename=guest_ssh.get("pkey_path"),
+            timeout=10,
+        )
+        _, stdout, stderr = ssh.exec_command(command)
+        returncode = stdout.channel.recv_exit_status()
+        return ExecResult(returncode, stdout.read().decode(), stderr.read().decode())
+    finally:
+        ssh.close()
diff --git a/coriolis/tests/integration/test_provider/exp.py b/coriolis/tests/integration/test_provider/exp.py
index 047a0504d..ebd33560d 100644
--- a/coriolis/tests/integration/test_provider/exp.py
+++ b/coriolis/tests/integration/test_provider/exp.py
@@ -11,6 +11,7 @@
 import csv
 import io
 import os
+import shutil
 import unittest
 import uuid
 
@@ -29,7 +30,7 @@
     BaseSourceMinionPoolProvider,
     BaseUpdateSourceReplicaProvider,
 )
-from coriolis.tests.integration import provider_test_base
+from coriolis.tests.integration import osmorphing_utils, provider_test_base, ssh_utils
 from coriolis.tests.integration import utils as test_utils
 from coriolis.tests.integration.test_provider import common
 
@@ -103,6 +104,12 @@ def teardown(self, connection_info: dict):
         )
 
     def check_prerequisites(self):
+        if not shutil.which("busybox"):
+            raise unittest.SkipTest(
+                "A statically linked 'busybox' binary is required on the host "
+                "(e.g.: the busybox-static package)"
+            )
+
         if not test_utils.container_image_exists(test_utils.DATA_MINION_IMAGE):
             raise unittest.SkipTest(
                 "Docker image '%s' not found; build it with: "
@@ -113,7 +120,7 @@ def check_prerequisites(self):
 
     def get_test_instance(self, size_mb: int):
         device = test_utils.create_loop_device(size_mb * 1024 * 1024)
-        test_utils.write_test_pattern(device, 8192)
+        osmorphing_utils.write_busybox_rootfs_to_device(device)
 
         instance_name = "%s-%s" % (os.path.basename(device), uuid.uuid4().hex[:8])
         self._test_devices[instance_name] = device
@@ -130,6 +137,18 @@ def delete_test_instance(self, instance_name: str):
     def get_test_instance_device(self, instance_name: str):
         return self._test_devices.get(instance_name)
 
+    def can_exec_test_vm(self):
+        return True
+
+    def exec_test_vm(self, instance_name: str, command: str):
+        """Run *command* chrooted in the instance's busybox rootfs."""
+        result = osmorphing_utils.chroot_exec(
+            self._test_devices[instance_name], command
+        )
+        return ssh_utils.ExecResult(
+            result.returncode, result.stdout.decode(), result.stderr.decode()
+        )
+
     def _event_manager(self):
         return events.EventManager(self._event_handler)
 
@@ -350,7 +369,6 @@ def replicate_disks(
         repl_state = _extract_repl_state(volumes_info) if incremental else None
 
         disk_mappings = source_resources.get("disk_mappings")
-        reused_minion = disk_mappings is None
         if disk_mappings is None:
             # Minion pool case: "source_resources" only carries the pool minion's
             # container_id. the block devices were never attached to the container,
@@ -369,11 +387,13 @@ def replicate_disks(
             source_conn_info, self._event_manager(), volumes_info, repl_state
         )
         replicator.init_replicator()
-        if reused_minion and incremental:
-            # The pool minion's replicator process persists across executions and it
-            # only computes whole-disk checksums once, at process startup. Without
-            # restarting it, "verify_disk_integrity" would keep comparing against that
-            # stale, pre-sync checksum on every subsequent execution against the minion.
+        if incremental:
+            # The replicator process is already running, either started by
+            # deploy_replica_source_resources() or persisting on a pool minion across
+            # executions, and it only loads the state file and computes whole-disk
+            # checksums at startup. Without restarting it, the previous replica state
+            # would be ignored (resulting in full syncs), and "verify_disk_integrity"
+            # would compare against a stale, pre-sync checksum on a pool minion.
             replicator.update_state(repl_state or [], restart=True)
         replicator.wait_for_chunks()
 
diff --git a/coriolis/tests/integration/test_provider/imp.py b/coriolis/tests/integration/test_provider/imp.py
index fa45f3b3c..6d8d9d356 100644
--- a/coriolis/tests/integration/test_provider/imp.py
+++ b/coriolis/tests/integration/test_provider/imp.py
@@ -160,11 +160,21 @@ def deploy_replica_disks(
         export_info,
         volumes_info,
     ):
-        """Allocate disks and return volumes_info."""
+        """Allocate disks and return volumes_info.
+
+        Disks allocated by a previous execution are reused, along with their
+        "replica_state", so that subsequent executions are incremental.
+        """
         src_disks = export_info.get("devices", {}).get("disks", [])
+        existing_vols = {vol["disk_id"]: vol for vol in volumes_info or []}
 
         result = []
         for disk in src_disks:
+            existing_vol = existing_vols.get(disk["id"])
+            if existing_vol and existing_vol.get("volume_dev"):
+                result.append(existing_vol)
+                continue
+
             owner = disk.get("owner")
             if owner and owner != instance_name:
                 # Shared disk owned by another instance of a clustered
diff --git a/coriolis/tests/integration/transfers/test_transfer.py b/coriolis/tests/integration/transfers/test_transfer.py
index 1eb1dfa28..4a1331947 100644
--- a/coriolis/tests/integration/transfers/test_transfer.py
+++ b/coriolis/tests/integration/transfers/test_transfer.py
@@ -63,45 +63,27 @@ def test_transfer(self):
     def test_incremental_replica_transfer(self):
         """Full transfer followed by incremental after source modification.
 
-        - Write a known byte pattern to the source loop device.
         - Create source / destination endpoints and a Replica transfer via the
           Coriolis REST API (using coriolisclient).
-        - Execute the transfer and wait for it to complete.
-        - Overwrite a single chunk on the source device.
         - Update the transfer to enable disk integrity verification.
+        - Execute the transfer and wait for it to complete.
+        - Mutate some data on the source instance's primary disk.
         - Execute a second transfer run (incremental=True).
-        - Assert that the destination disk checksum was computed.
+        - Assert that the disk checksums were verified on both runs, and that
+          the second run wrote less data than the first one.
 
-        The content is verified only if the test import provider is being used.
+        The execution fails if the source and destination checksums differ, so the
+        content is verified for any provider pair. The devices are also compared
+        directly if the test import provider is being used. Requires a source provider
+        that implements ``can_exec_test_vm()``, and which supports disk checksumming.
         """
-        if not self._harness.uses_core_test_export_provider():
+        if not self._exp_provider.can_exec_test_vm():
             self.skipTest(
-                "Incremental transfer verification requires direct access "
-                "to the source device"
-            )
-
-        # First run: full transfer
-        self._execute_and_wait(self._transfer.id)
-
-        if self._harness.uses_core_test_import_provider():
-            self.assertTrue(
-                test_utils.devices_match(self._src_device, self._dst_device),
-                "Devices do not match after initial full transfer",
+                "%s does not support incremental transfer verification"
+                % type(self._exp_provider).__name__
             )
 
-        # Mutate source: write a different pattern at the second chunk
-        test_utils.write_bytes_at_offset(
-            self._src_device,
-            offset=4096,
-            data=b"\xff\xfe\xfd\xfc" * 1024,
-        )
-        if self._harness.uses_core_test_import_provider():
-            self.assertFalse(
-                test_utils.devices_match(self._src_device, self._dst_device),
-                "Devices should differ after mutating the source",
-            )
-
-        # Enable disk integrity verification for the incremental run.
+        # Enable disk integrity verification for both runs.
         # Note that the HTTP backup writer is the only one implementing
         # get_disk_checksum().
         execution = self._client.transfers.update(
@@ -117,25 +99,48 @@ def test_incremental_replica_transfer(self):
         )
         self.assertExecutionCompleted(execution.id)
 
-        checksummed_disks = []
-        original = backup_writers.HTTPBackupWriterImpl.get_disk_checksum
+        # First run: full transfer
+        checksummed_disks, full_bytes = self._execute_and_spy_on_writer(
+            self._transfer.id
+        )
+
+        self.assertEqual(
+            self._transfer_disk_ids,
+            checksummed_disks,
+            "The checksum was not computed after the initial full transfer",
+        )
+        if self._harness.uses_core_test_import_provider():
+            self.assertTrue(
+                test_utils.devices_match(self._src_device, self._dst_device),
+                "Devices do not match after initial full transfer",
+            )
 
-        def _recording_get_disk_checksum(writer, *args, **kwargs):
-            checksummed_disks.append(writer._disk_id)
-            return original(writer, *args, **kwargs)
+        # Mutate source.
+        self._exp_provider.mutate_test_instance_data(self._instance_name)
+
+        if self._harness.uses_core_test_import_provider():
+            self.assertFalse(
+                test_utils.devices_match(self._src_device, self._dst_device),
+                "Devices should differ after mutating the source",
+            )
 
         # Second run: incremental
-        with mock.patch.object(
-            backup_writers.HTTPBackupWriterImpl,
-            "get_disk_checksum",
-            _recording_get_disk_checksum,
-        ):
-            self._execute_and_wait(self._transfer.id)
+        checksummed_disks, incremental_bytes = self._execute_and_spy_on_writer(
+            self._transfer.id
+        )
 
         self.assertEqual(
-            [os.path.basename(self._src_device)],
+            self._transfer_disk_ids,
             checksummed_disks,
-            "The checksum was not computed for the transferred disk",
+            "The checksum was not computed after the incremental transfer",
+        )
+        self.assertGreater(
+            incremental_bytes, 0, "The incremental transfer wrote no data"
+        )
+        self.assertLess(
+            incremental_bytes,
+            full_bytes,
+            "The incremental transfer wrote as much data as the full one",
         )
 
         if self._harness.uses_core_test_import_provider():