diff --git a/instances/jobs.py b/instances/jobs.py
index 76ae56a4..b16f606f 100644
--- a/instances/jobs.py
+++ b/instances/jobs.py
@@ -1,11 +1,20 @@
 """Jobs of the instances app (operations framework, spec section 5)."""
 
+import re
+
 from django.core.exceptions import PermissionDenied
 from django.utils.translation import gettext_noop
-from libvirt import VIR_DOMAIN_SNAPSHOT_LIST_EXTERNAL, VIR_ERR_NO_DOMAIN, libvirtError
+from django.utils.translation import gettext_noop as _
+from libvirt import (VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE, VIR_DOMAIN_SNAPSHOT_LIST_EXTERNAL, VIR_ERR_NO_DOMAIN,
+                     libvirtError)
 from vrtManager import util
+from vrtManager.connection import wvmConnect
 from vrtManager.instance import wvmInstance
 
+from accounts.models import UserInstance
+from appsettings.settings import app_settings
+from computes.models import Compute
+from instances import utils
 from instances.models import Instance
 from operations import claims
 from operations.jobs import Kind, Outcome, Plan, register
@@ -33,6 +42,122 @@ def _instance(pk):
     return Instance.objects.select_related("compute").get(pk=pk)
 
 
+CLONE_POST_KEY_RE = re.compile(r"^(clone-net-mac-\d+|disk-[a-z0-9]+|meta-[a-z0-9]+)$")
+CLONE_FIELDS = ("name", "clone-title", "clone-description")
+
+
+def clone_data(user, instance, fields):
+    """The clone's settings from the form fields and the source as it is now;
+    OperationError with the reason when it cannot be made."""
+    from instances.views import VOLUME_NAME_RE, quota_refused
+
+    data = dict()
+    data["name"] = fields.get("name", "").strip()
+    data["clone-title"] = fields.get("clone-title", "").strip()
+    data["clone-description"] = fields.get("clone-description", "").strip()
+
+    # the clone copies the disks of the persistent definition
+    disk_sum = sum(int(disk.get("size") or 0) >> 30 for disk in instance.config_disks)
+    quota_msg = utils.check_user_quota(user, 1, instance.vcpu, instance.memory, disk_sum)
+
+    data["disk_owner_uid"] = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID)
+    data["disk_owner_gid"] = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID)
+
+    # Only superusers may choose disk names and MAC addresses; the form only
+    # shows those fields to them.
+    if user.is_superuser:
+        for key, value in fields.items():
+            if CLONE_POST_KEY_RE.fullmatch(key):
+                data[key] = value.strip()
+
+    if app_settings.CLONE_INSTANCE_AUTO_NAME == "True" and not data["name"]:
+        auto_vname = utils.get_clone_free_names()[0]
+        data["name"] = auto_vname
+        data["clone-net-mac-0"] = utils.get_dhcp_mac_address(auto_vname)
+        for disk in instance.config_disks:
+            data[f"disk-{disk['dev']}"] = utils.get_clone_disk_name(disk, instance.name, auto_vname)
+
+    if not user.is_superuser:
+        for disk in instance.config_disks:
+            data[f"disk-{disk['dev']}"] = utils.get_clone_disk_name(disk, instance.name, data["name"])
+        for num in range(max(1, len(instance.config_networks))):
+            key = f"clone-net-mac-{num}"
+            if not data.get(key):
+                data[key] = (utils.get_dhcp_mac_address(data["name"]) if num == 0 else "") or utils.get_random_mac_address()
+
+    check_instance = Instance.objects.filter(name=data["name"])
+    invalid_macs = [
+        value for key, value in data.items()
+        if key.startswith("clone-net-mac-") and not utils.MAC_ADDRESS_RE.fullmatch(value)
+    ]
+    invalid_disks = [
+        value for key, value in data.items()
+        # None: a disk without a volume, which clone_instance does not copy.
+        if key.startswith("disk-") and value is not None and not VOLUME_NAME_RE.fullmatch(value)
+    ]
+
+    if not user.is_superuser and quota_msg:
+        raise util.OperationError(quota_refused(quota_msg, _(
+            "User '%(quota_msg)s' quota reached, cannot create '%(clone_name)s'!"
+        ) % {"quota_msg": quota_msg, "clone_name": data["name"]}))
+    if check_instance:
+        raise util.OperationError(_("Instance '%(clone_name)s' already exists!") % {"clone_name": data["name"]})
+    if not re.match(r"^[a-zA-Z0-9-]+$", data["name"]):
+        raise util.OperationError(
+            _("Instance name '%(clone_name)s' contains invalid characters!") % {"clone_name": data["name"]})
+    if "clone-net-mac-0" not in data or invalid_macs:
+        raise util.OperationError(_("Instance MAC '%(clone_mac)s' invalid format!") % {"clone_mac": ", ".join(invalid_macs)})
+    if invalid_disks:
+        raise util.OperationError(
+            _("Disk name '%(disk_name)s' contains invalid characters!") % {"disk_name": ", ".join(str(v) for v in invalid_disks)})
+    return data
+
+
+def vm_claims(compute, name):
+    """A job that adds a VM: other VMs stay usable; the VM list reconcile waits."""
+    return {claims.compute_key(compute.pk): Claim.SHARED, claims.reconcile_key(compute.pk): Claim.EXCLUSIVE,
+            claims.name_key(compute.pk, name): Claim.EXCLUSIVE}
+
+
+def record_vm(op, owner):
+    """The panel's records of the new VM, as the synchronous views wrote them."""
+    vm = Instance.objects.get_or_create(compute_id=op.intent["compute"], uuid=op.intent["uuid"],
+                                        defaults={"name": op.intent["name"]})[0]
+    if owner and op.user_id:
+        UserInstance.objects.get_or_create(instance=vm, user_id=op.user_id,
+                                           defaults={"is_delete": True, "is_change": True, "is_vnc": True})
+
+
+def vm_outcome(op, owner, volumes):
+    """The VM is defined last: if it exists, the job succeeded. If it does not and
+    none of its planned volumes exists either, the job failed and changed nothing."""
+    c = Compute.objects.get(pk=op.intent["compute"])
+    conn = wvmConnect(c.hostname, c.login, c.password, c.type).wvm
+    try:
+        conn.lookupByUUIDString(op.intent["uuid"])
+    except libvirtError as err:
+        if err.get_error_code() != VIR_ERR_NO_DOMAIN:
+            raise  # unreachable: recovery waits and retries
+        held = set()
+        for pool in conn.listAllStoragePools(VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE):
+            held.update(pool.listVolumes())
+        left = sorted(held & set(volumes))
+        if not left:
+            return Outcome(Operation.FAILED, error="The VM was not defined and no volume of it was left.")
+        return Outcome(Operation.UNCERTAIN,
+                       error=f"The VM was not defined; check these volumes, then resolve this job: {', '.join(left)}")
+    record_vm(op, owner)
+    return Outcome(Operation.SUCCEEDED)
+
+
+def _clonable(user, pk):
+    from instances.views import _clone_source
+
+    if user is None or not user.has_perm("instances.clone_instances"):
+        raise PermissionDenied
+    return _clone_source(user, pk)
+
+
 def _merged(jobs):
     """(done, total) bytes of the snapshot's block jobs."""
     return sum(j["cur"] for j in jobs), sum(j["end"] for j in jobs)
@@ -111,3 +236,48 @@ class DeleteExternalSnapshot(Kind):
         compute, _sep, vm = op.resource_label.rpartition("/")  # a VM name has no "/"
         message = f"Delete external snapshot: {op.params['name']}"
         return compute, vm, message if state == Operation.SUCCEEDED else f"{message} ({state})"
+
+
+@register
+class CloneInstance(Kind):
+    name = "instance.clone"
+    label = gettext_noop("Clone VM")
+    time_limit = 12 * 3600  # a copy cannot be stopped: over time is only noted
+
+    def normalize(self, raw):
+        fields = {k: str(v) for k, v in raw.items() if k in CLONE_FIELDS or CLONE_POST_KEY_RE.fullmatch(k)}
+        return {"instance": int(raw["instance"]), "fields": fields}
+
+    def plan(self, user, params):
+        instance = _clonable(user, params["instance"])
+        if instance.proxy.get_status() != 5:
+            raise util.OperationError("Shut the VM down before cloning it")
+        data = clone_data(user, instance, params["fields"])
+        keys = vm_claims(instance.compute, data["name"])
+        keys[claims.instance_key(instance.uuid)] = Claim.SHARED
+        return Plan(claims=keys, resource=claims.instance_key(instance.uuid),
+                    label=f"{instance.compute.name}/{instance.name}",
+                    intent={"compute": instance.compute_id, "uuid": util.randomUUID(), "name": data["name"],
+                            "clone": data})
+
+    def precheck(self, op, ctx):
+        instance = _clonable(op.user, op.params["instance"])
+        if instance.compute_id != op.intent["compute"]:
+            raise util.OperationError("The VM moved to another compute after the job was queued")
+
+    def run(self, op, ctx):
+        instance = _instance(op.params["instance"])
+        ctx.record_intent()
+        ctx.set_phase(gettext_noop("Copying disks"))
+        instance.proxy.clone_instance(op.intent["clone"], op.intent["uuid"])
+        record_vm(op, owner=True)
+        return {}
+
+    def reconcile(self, op):
+        disks = [v for k, v in op.intent["clone"].items() if k.startswith("disk-") and v]
+        return vm_outcome(op, owner=True, volumes=disks)
+
+    def log_entry(self, op, state):
+        compute, _sep, source = op.resource_label.rpartition("/")
+        message = f"Create a clone of '{source}'"
+        return compute, op.intent["name"], message if state == Operation.SUCCEEDED else f"{message} ({state})"
diff --git a/instances/templates/instances/settings_tab.html b/instances/templates/instances/settings_tab.html
index 70f561b5..aebce899 100644
--- a/instances/templates/instances/settings_tab.html
+++ b/instances/templates/instances/settings_tab.html
@@ -1,6 +1,7 @@
 {% load i18n %}
 {% load django_bootstrap5 %}
 {% load bootstrap_icons %}
+{% load operations %}
 <div class="tab-pane" id="settings" role="tabpanel">
     <!-- Nav tabs -->
     <ul class="nav nav-tabs" role="tablist" aria-label="Instance settings">
@@ -720,7 +721,7 @@
         {% if can_clone %}
         <div role="tabpanel" class="tab-pane tab-pane-bordered" id="clone">
             <p class="fw-bold">{% trans "Create a clone" %}</p>
-            <form class="form" action="{% url 'instances:clone' instance.id %}" method="post" role="form" aria-label="Create clone form">{% csrf_token %}
+            <form class="form" action="{% url 'instances:clone' instance.id %}" method="post" role="form" aria-label="Create clone form">{% csrf_token %}{% submission_key %}
                 <div class="row">
                     <label class="col-sm-3 col-form-label fw-normal">{% trans "Clone Name" %}</label>
                     {% if request.user.is_superuser %}
@@ -805,7 +806,7 @@
                     </div>
                 </div>
                 {% if instance.status == 5 %}
-                    <button type="submit" class="btn btn-lg btn-success float-end" name="clone" onclick="showPleaseWaitDialog();">{% trans "Clone" %}</button>
+                    <button type="submit" class="btn btn-lg btn-success float-end" name="clone">{% trans "Clone" %}</button>
                 {% else %}
                     <button class="btn btn-lg btn-success float-end disabled" disabled name="clone">{% trans "Clone" %}</button>
                 {% endif %}
diff --git a/instances/tests/test_clone_disks.py b/instances/tests/test_clone_disks.py
index 5c2a60e1..e3f3b110 100644
--- a/instances/tests/test_clone_disks.py
+++ b/instances/tests/test_clone_disks.py
@@ -11,6 +11,7 @@ from accounts.models import UserInstance
 from appsettings.models import AppSettings
 from computes.models import Compute
 from instances.models import Instance
+from operations.models import Operation
 
 LIVE = [{"dev": "vda", "path": "/pool/live.qcow2", "storage": "pool", "image": "live.qcow2", "size": 1 << 30}]
 SAVED = [{"dev": "vdb", "path": "/pool/saved.qcow2", "storage": "pool", "image": "saved.qcow2", "size": 20 << 30}]
@@ -30,16 +31,16 @@ class CloneDisksTestCase(TestCase):
             wvm.return_value.get_vcpu.return_value = 1
             wvm.return_value.get_memory.return_value = 512
             wvm.return_value.get_net_devices.return_value = []
-            wvm.return_value.clone_instance.return_value = "abcdefab-0000-1111-2222-333333333333"
-            response = self.client.post(reverse("instances:clone", args=[self.vm.id]), post)
+            wvm.return_value.get_status.return_value = 5
+            response = self.client.post(reverse("instances:clone", args=[self.vm.id]), {"submission_key": "key-clone-disk1", **post})
             self.messages = [str(m) for m in response.wsgi_request._messages]
-        return quota, wvm.return_value.clone_instance
+        return quota
 
     def test_the_quota_counts_the_disks_the_clone_copies(self):
         owner = get_user_model().objects.create_user("cd_owner", password="x")
         owner.user_permissions.add(Permission.objects.get(codename="clone_instances"))
         UserInstance.objects.create(instance=self.vm, user=owner, is_change=True)
-        quota, _ = self.clone(owner, name="cd-clone")
+        quota = self.clone(owner, name="cd-clone")
         self.assertEqual(quota.call_args.args[4], 20)
 
     def test_automatic_disk_names_are_for_the_disks_the_clone_copies(self):
@@ -48,7 +49,7 @@ class CloneDisksTestCase(TestCase):
         with patch("instances.views.utils.get_clone_free_names", return_value=["auto-1"]), patch(
             "instances.views.utils.get_dhcp_mac_address", return_value="52:54:00:00:00:01"
         ):
-            _, clone_instance = self.clone(admin)
-        data = clone_instance.call_args.args[0]
+            self.clone(admin)
+        data = Operation.objects.get().intent["clone"]
         self.assertIn("disk-vdb", data)
         self.assertNotIn("disk-vda", data)
diff --git a/instances/tests/test_clone_job.py b/instances/tests/test_clone_job.py
new file mode 100644
index 00000000..e4f5e12c
--- /dev/null
+++ b/instances/tests/test_clone_job.py
@@ -0,0 +1,112 @@
+"""VM clone runs as a job: the page queues it, the worker clones and records it."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import libvirt
+from django.contrib.auth import get_user_model
+from django.core.exceptions import PermissionDenied
+from django.test import TransactionTestCase
+from django.urls import reverse
+
+from accounts.models import UserInstance
+from appsettings.settings import get_settings
+from computes.models import Compute
+from instances.jobs import CloneInstance
+from instances.models import Instance
+from operations.jobs import Context
+from vrtManager.util import OperationError
+from operations.models import Claim, Operation, acquire
+
+
+def proxy():
+    p = MagicMock()
+    p.get_status.return_value = 5
+    p.get_disk_devices.return_value = [{"dev": "vda", "image": "src.qcow2", "size": 1 << 30}]
+    p.get_net_devices.return_value = [{"mac": "52:54:00:aa:bb:cc"}]
+    p.get_vcpu.return_value = 1
+    p.get_memory.return_value = 512
+    return p
+
+
+def pool(volumes):
+    p = MagicMock()
+    p.listVolumes.return_value = volumes
+    return p
+
+
+class CloneJobTestCase(TransactionTestCase):
+    def setUp(self):
+        get_settings()
+        self.admin = get_user_model().objects.create_superuser("clone_admin", "a@example.com", "pw")
+        self.client.force_login(self.admin)
+        self.compute = Compute.objects.create(name="clone-c", hostname="192.0.2.30", login="root", password="", type=1)
+        self.source = Instance.objects.create(compute=self.compute, name="src", uuid="12121212-3434-5656-7878-909090909090")
+        patcher = patch("instances.models.wvmInstance", return_value=proxy())
+        self.proxy = patcher.start().return_value
+        self.addCleanup(patcher.stop)
+
+    def test_the_page_queues_the_clone_and_the_worker_makes_and_records_it(self):
+        response = self.client.post(reverse("instances:clone", args=[self.source.pk]),
+                                    {"name": "copy", "disk-vda": "copy.qcow2", "clone-net-mac-0": "52:54:00:00:00:01",
+                                     "submission_key": "key-clone001"})
+        self.assertRedirects(response, reverse("instances:instance", args=[self.source.pk]),
+                             fetch_redirect_response=False)
+        op = Operation.objects.get()
+        c = self.compute.pk
+        self.assertEqual(set(Claim.objects.values_list("resource", "mode")), {
+            (f"compute:{c}", "shared"), (f"reconcile:{c}", "exclusive"),
+            (f"instance:{self.source.uuid}", "shared"), (f"name:{c}:copy", "exclusive")})
+        self.proxy.clone_instance.assert_not_called()
+
+        acquire("wvc-test:1:a", "wvc-test")
+        Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING, worker="wvc-test:1:a")
+        op.refresh_from_db()
+        CloneInstance().run(op, Context(op, SimpleNamespace(identity="wvc-test:1:a")))
+        self.proxy.clone_instance.assert_called_once_with(op.intent["clone"], op.intent["uuid"])
+        clone = Instance.objects.get(uuid=op.intent["uuid"])
+        self.assertTrue(UserInstance.objects.filter(instance=clone, user=self.admin).exists())
+
+    def queued(self):
+        self.client.post(reverse("instances:clone", args=[self.source.pk]),
+                         {"name": "copy", "disk-vda": "copy.qcow2", "clone-net-mac-0": "52:54:00:00:00:01",
+                          "submission_key": "key-clone002"})
+        return Operation.objects.get()
+
+    def test_the_worker_refuses_a_clone_whose_access_was_revoked(self):
+        op = self.queued()
+        self.admin.is_superuser = False
+        self.admin.save()
+        with self.assertRaises(PermissionDenied):
+            CloneInstance().precheck(op, None)
+
+    def test_the_worker_refuses_a_clone_whose_source_moved(self):
+        op = self.queued()
+        other = Compute.objects.create(name="clone-d", hostname="192.0.2.31", login="root", password="", type=1)
+        Instance.objects.filter(pk=self.source.pk).update(compute=other)
+        with self.assertRaises(OperationError):
+            CloneInstance().precheck(op, None)
+
+    @patch("instances.jobs.wvmConnect")
+    def test_recovery(self, conn):
+        op = Operation(user=self.admin, intent={"compute": self.compute.pk, "uuid": "abababab-0000-1111-2222-333333333333",
+                                                "name": "copy", "clone": {"name": "copy", "disk-vda": "copy.qcow2"}})
+        self.assertEqual(CloneInstance().reconcile(op).state, Operation.SUCCEEDED)  # the VM is defined
+        self.assertTrue(Instance.objects.filter(uuid=op.intent["uuid"]).exists())
+        missing = libvirt.libvirtError("no domain")
+        missing.err = (libvirt.VIR_ERR_NO_DOMAIN, libvirt.VIR_FROM_QEMU, "no domain")
+        conn.return_value.wvm.lookupByUUIDString.side_effect = missing
+        conn.return_value.wvm.listAllStoragePools.return_value = [pool(["copy.qcow2", "other.qcow2"])]
+        outcome = CloneInstance().reconcile(op)
+        self.assertEqual(outcome.state, Operation.UNCERTAIN)
+        self.assertIn("copy.qcow2", outcome.error)
+
+    @patch("instances.jobs.wvmConnect")
+    def test_a_clone_that_left_nothing_failed(self, conn):
+        op = Operation(user=self.admin, intent={"compute": self.compute.pk, "uuid": "abababab-0000-1111-2222-333333333333",
+                                                "name": "copy", "clone": {"name": "copy", "disk-vda": "copy.qcow2"}})
+        missing = libvirt.libvirtError("no domain")
+        missing.err = (libvirt.VIR_ERR_NO_DOMAIN, libvirt.VIR_FROM_QEMU, "no domain")
+        conn.return_value.wvm.lookupByUUIDString.side_effect = missing
+        conn.return_value.wvm.listAllStoragePools.return_value = [pool(["other.qcow2"])]
+        self.assertEqual(CloneInstance().reconcile(op).state, Operation.FAILED)
diff --git a/instances/tests/test_clone_permission.py b/instances/tests/test_clone_permission.py
index f49b7cb4..59a802bb 100644
--- a/instances/tests/test_clone_permission.py
+++ b/instances/tests/test_clone_permission.py
@@ -17,6 +17,7 @@ from appsettings.settings import app_settings
 from computes.models import Compute
 from instances.models import Instance
 from instances.utils import get_dhcp_mac_address
+from operations.models import Operation
 
 SRC_DISK = {
     "dev": "vda",
@@ -71,9 +72,9 @@ class ClonePermissionTestCase(TestCase):
             proxy = mock_wvm.return_value
             proxy.get_disk_devices.return_value = [SRC_DISK]
             proxy.get_net_devices.return_value = [{"mac": "52:54:00:aa:bb:cc"}]
+            proxy.get_status.return_value = 5
             proxy.get_vcpu.return_value = 1
             proxy.get_memory.return_value = 512
-            proxy.clone_instance.return_value = "abababab-cdcd-efef-0101-232323232323"
             yield proxy
 
     def _clone(self, user, data):
@@ -91,50 +92,49 @@ class ClonePermissionTestCase(TestCase):
             "disk-vda": "clone-vm.qcow2",
             "clone-title": "",
             "clone-description": "",
+            "submission_key": "key-clone-perm1",
         }
         data.update(extra)
         return data
 
     def test_viewer_cannot_clone_someone_elses_vm(self):
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             res = self._clone(self.viewer, self._valid_post())
         self.assertEqual(res.status_code, 403)
-        proxy.clone_instance.assert_not_called()
+        self.assertFalse(Operation.objects.exists())
 
     def test_owner_without_is_change_cannot_clone(self):
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             res = self._clone(self.owner_readonly, self._valid_post())
         self.assertEqual(res.status_code, 403)
-        proxy.clone_instance.assert_not_called()
+        self.assertFalse(Operation.objects.exists())
 
     def test_a_clone_stays_on_the_host_of_its_vm(self):
         # the removed "VM Clone Auto Migrate" setting sent it to a random host
         other = Compute.objects.create(name="other-host", hostname="198.51.100.20", login="", password="", type=4)
-        with self.mocked_libvirt() as proxy, \
+        with self.mocked_libvirt(), \
              patch("instances.views.utils.migrate_instance") as migrate, \
              patch.object(app_settings, "CLONE_INSTANCE_AUTO_MIGRATE", "True", create=True), \
              patch("instances.views.Compute.objects.order_by") as random_host:
             random_host.return_value.first.return_value = other
             self._clone(self.owner_change, self._valid_post())
-        proxy.clone_instance.assert_called_once()
         migrate.assert_not_called()
-        self.assertEqual(Instance.objects.get(name="clone-vm").compute, self.instance.compute)
+        self.assertEqual(Operation.objects.get().intent["compute"], self.instance.compute_id)
 
     def test_viewer_can_clone_a_template(self):
         self.instance.is_template = True
         self.instance.save(update_fields=["is_template"])
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             self._clone(self.viewer, self._valid_post())
-        proxy.clone_instance.assert_called_once()
+        self.assertEqual(Operation.objects.count(), 1)
 
     def test_posted_disk_owner_and_unknown_keys_are_ignored(self):
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             self._clone(
                 self.owner_change,
                 self._valid_post(disk_owner_uid="4242", disk_owner_gid="4242", evil="x"),
             )
-        proxy.clone_instance.assert_called_once()
-        clone_data = proxy.clone_instance.call_args[0][0]
+        clone_data = Operation.objects.get().intent["clone"]
         self.assertEqual(
             clone_data["disk_owner_uid"],
             int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID),
@@ -146,53 +146,37 @@ class ClonePermissionTestCase(TestCase):
         self.assertNotIn("evil", clone_data)
 
     def test_non_superuser_disk_names_are_derived_by_the_server(self):
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             self._clone(
                 self.owner_change,
                 self._valid_post(**{"disk-vda": "../../../etc/passwd"}),
             )
-        proxy.clone_instance.assert_called_once()
-        clone_data = proxy.clone_instance.call_args[0][0]
+        clone_data = Operation.objects.get().intent["clone"]
         self.assertEqual(clone_data["disk-vda"], "clone-vm.qcow2")
 
     def test_superuser_unsafe_disk_name_is_rejected(self):
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             self._clone(
                 self.superuser,
                 self._valid_post(**{"disk-vda": "x</name><evil/>"}),
             )
-        proxy.clone_instance.assert_not_called()
+        self.assertFalse(Operation.objects.exists())
 
     def test_superuser_invalid_second_mac_is_rejected(self):
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             self._clone(
                 self.superuser,
                 self._valid_post(**{"clone-net-mac-1": "not-a-mac"}),
             )
-        proxy.clone_instance.assert_not_called()
-
-    def test_clone_holds_the_compute_exclusively(self):
-        # Clones of different VMs must not race for the destination name.
-        calls = []
-
-        @contextmanager
-        def record(compute, *, shared=False):
-            calls.append(shared)
-            yield
-
-        with self.mocked_libvirt() as proxy, patch("instances.views.libvirt_compute_lock", record):
-            self._clone(self.superuser, self._valid_post())
-        proxy.clone_instance.assert_called_once()
-        self.assertEqual(calls, [False])
+        self.assertFalse(Operation.objects.exists())
 
     def test_superuser_valid_clone_passes_posted_values(self):
-        with self.mocked_libvirt() as proxy:
+        with self.mocked_libvirt():
             self._clone(
                 self.superuser,
                 self._valid_post(**{"disk-vda": "custom-name.qcow2", "meta-vda": "true"}),
             )
-        proxy.clone_instance.assert_called_once()
-        clone_data = proxy.clone_instance.call_args[0][0]
+        clone_data = Operation.objects.get().intent["clone"]
         self.assertEqual(clone_data["disk-vda"], "custom-name.qcow2")
         self.assertEqual(clone_data["meta-vda"], "true")
         self.assertEqual(clone_data["clone-net-mac-0"], "52:54:00:12:34:56")
@@ -203,8 +187,7 @@ class ClonePermissionTestCase(TestCase):
         with self.mocked_libvirt() as proxy:
             proxy.get_disk_devices.return_value = [SRC_DISK, empty_disk]
             self._clone(self.owner_change, self._valid_post())
-        proxy.clone_instance.assert_called_once()
-        clone_data = proxy.clone_instance.call_args[0][0]
+        clone_data = Operation.objects.get().intent["clone"]
         self.assertEqual(clone_data["disk-vda"], "clone-vm.qcow2")
         self.assertIsNone(clone_data["disk-vdb"])
 
@@ -213,9 +196,9 @@ class ClonePermissionTestCase(TestCase):
         with self.mocked_libvirt() as proxy:
             proxy.get_disk_devices.return_value = [odd_disk]
             self._clone(self.owner_change, self._valid_post())
-        proxy.clone_instance.assert_called_once()
+        self.assertEqual(Operation.objects.count(), 1)
         self.assertEqual(
-            proxy.clone_instance.call_args[0][0]["disk-vda"], "my-disk-2-clone.qcow2"
+            Operation.objects.get().intent["clone"]["disk-vda"], "my-disk-2-clone.qcow2"
         )
 
 
diff --git a/instances/tests/test_live_clone.py b/instances/tests/test_live_clone.py
index 9b7d65aa..49265276 100644
--- a/instances/tests/test_live_clone.py
+++ b/instances/tests/test_live_clone.py
@@ -68,7 +68,7 @@ class LiveCloneTestCase(SimpleTestCase):
         before = self.pool_volumes()
 
         with self.assertRaises(Exception):
-            vm.clone_instance(self.clone_data("clone-dst"))
+            vm.clone_instance(self.clone_data("clone-dst"), "abababab-0000-1111-2222-333333333333")
 
         self.assertEqual(self.pool_volumes(), before, "copies left behind or a volume deleted")
         self.assertTrue(livetest.volume_exists(self.conn, taken))
@@ -81,7 +81,7 @@ class LiveCloneTestCase(SimpleTestCase):
         before = self.pool_volumes()
 
         with self.assertRaises(Exception):
-            vm.clone_instance(self.clone_data("clone-run-copy"))
+            vm.clone_instance(self.clone_data("clone-run-copy"), "abababab-0000-1111-2222-333333333333")
 
         self.assertEqual(self.pool_volumes(), before)
 
@@ -94,7 +94,7 @@ class LiveCloneTestCase(SimpleTestCase):
         source = vm.get_nvram()
 
         uuid = vm.clone_instance(
-            {"name": P + "clone-uefi-copy", "disk-vda": P + "clone-uefi-copy-a.qcow2", "disk_owner_uid": 0, "disk_owner_gid": 0}
+            {"name": P + "clone-uefi-copy", "disk-vda": P + "clone-uefi-copy-a.qcow2", "disk_owner_uid": 0, "disk_owner_gid": 0}, "abababab-0000-1111-2222-333333333333"
         )
 
         clone = self.conn.lookupByUUIDString(uuid)
@@ -111,7 +111,7 @@ class LiveCloneTestCase(SimpleTestCase):
 
     def test_clone_of_a_stopped_vm(self):
         _, vm = self.source("clone-ok")
-        uuid = vm.clone_instance(self.clone_data("clone-ok-copy"))
+        uuid = vm.clone_instance(self.clone_data("clone-ok-copy"), "abababab-0000-1111-2222-333333333333")
         clone = self.conn.lookupByUUIDString(uuid)
         self.assertEqual(clone.name(), P + "clone-ok-copy")
         self.assertTrue({P + "clone-ok-copy-a.qcow2", P + "clone-ok-copy-b.qcow2"} <= self.pool_volumes())
diff --git a/instances/tests/test_moved_while_waiting.py b/instances/tests/test_moved_while_waiting.py
index ebd413b3..64c70fb0 100644
--- a/instances/tests/test_moved_while_waiting.py
+++ b/instances/tests/test_moved_while_waiting.py
@@ -69,74 +69,3 @@ class MovedWhileWaitingTestCase(TestCase):
         self.assertTrue(Instance.objects.filter(pk=self.vm.pk).exists())
         messages = [str(m) for m in response.wsgi_request._messages]
         self.assertIn("The VM was moved meanwhile; try again.", messages)
-
-    def clone(self, user, change):
-        @contextmanager
-        def compute_lock_meanwhile(compute, *, shared=False):
-            change()
-            yield
-
-        self.client.force_login(user)
-        with patch("instances.views.libvirt_compute_lock", compute_lock_meanwhile), patch(
-            "instances.models.wvmInstance"
-        ) as wvm, patch.object(Compute, "status", new_callable=PropertyMock, return_value=True), patch(
-            "instances.views.utils.check_user_quota", return_value=""  # not what this tests
-        ):
-            wvm.return_value.get_disk_devices.return_value = []
-            wvm.return_value.get_vcpu.return_value = 1
-            wvm.return_value.get_memory.return_value = 512
-            response = self.client.post(reverse("instances:clone", args=[self.vm.id]), {"name": "moved-clone"})
-        return response, wvm.return_value.clone_instance.called
-
-    def test_a_clone_of_a_vm_moved_while_it_waited_is_not_made(self):
-        from accounts.models import UserInstance
-        from django.contrib.auth.models import Permission
-
-        # an owner who may clone it (a superuser would also choose MACs)
-        owner = get_user_model().objects.create_user("clone_owner", password="x")
-        owner.user_permissions.add(Permission.objects.get(codename="clone_instances"))
-        UserInstance.objects.create(instance=self.vm, user=owner, is_change=True)
-        response, cloned = self.clone(
-            owner, lambda: Instance.objects.filter(pk=self.vm.pk).update(compute=self.target)
-        )
-        self.assertFalse(cloned)
-        self.assertIn("The VM was moved meanwhile; try again.", [str(m) for m in response.wsgi_request._messages])
-
-    def test_a_template_unmarked_while_the_request_waited_is_not_cloned_by_a_viewer(self):
-        from django.contrib.auth.models import Permission
-
-        viewer = get_user_model().objects.create_user("clone_viewer", password="x")
-        for codename in ("view_instances", "clone_instances"):
-            viewer.user_permissions.add(Permission.objects.get(codename=codename))
-        Instance.objects.filter(pk=self.vm.pk).update(is_template=True)
-        response, cloned = self.clone(viewer, lambda: Instance.objects.filter(pk=self.vm.pk).update(is_template=False))
-        self.assertFalse(cloned)
-        self.assertEqual(response.status_code, 403)
-
-    def test_a_clone_checks_the_quota_against_the_source_as_it_is_under_the_lock(self):
-        from accounts.models import UserInstance
-        from django.contrib.auth.models import Permission
-
-        owner = get_user_model().objects.create_user("clone_quota", password="x")
-        owner.user_permissions.add(Permission.objects.get(codename="clone_instances"))
-        UserInstance.objects.create(instance=self.vm, user=owner, is_change=True)
-        self.client.force_login(owner)
-        with patch("instances.models.wvmInstance") as wvm:
-            wvm.return_value.get_disk_devices.return_value = []
-            wvm.return_value.get_vcpu.return_value = 1
-            wvm.return_value.get_memory.return_value = 512
-
-            @contextmanager
-            def compute_lock_after_a_resize(compute, *, shared=False):
-                wvm.return_value.get_vcpu.return_value = 8  # resized meanwhile
-                yield
-
-            with patch("instances.views.libvirt_compute_lock", compute_lock_after_a_resize), patch.object(
-                Compute, "status", new_callable=PropertyMock, return_value=True
-            ), patch(
-                "instances.views.utils.check_user_quota",
-                side_effect=lambda user, count, cpu, memory, disk: "cpu" if cpu > 4 else "",
-            ):
-                response = self.client.post(reverse("instances:clone", args=[self.vm.id]), {"name": "grown-clone"})
-        self.assertFalse(wvm.return_value.clone_instance.called)
-        self.assertTrue(any("quota" in str(m) for m in response.wsgi_request._messages))
diff --git a/instances/views.py b/instances/views.py
index dff02567..52857f63 100755
--- a/instances/views.py
+++ b/instances/views.py
@@ -1850,7 +1850,6 @@ def del_owner(request, pk):
 # Buses for the empty CD-ROM of a new VM: fdc and virtio have no ejectable
 # media, and libvirt refuses an empty usb disk.
 CDROM_BUSES = ("ide", "sata", "scsi")
-CLONE_POST_KEY_RE = re.compile(r"^(clone-net-mac-\d+|disk-[a-z0-9]+|meta-[a-z0-9]+)$")
 
 
 def _clone_source(user, pk):
@@ -1862,145 +1861,22 @@ def _clone_source(user, pk):
     return instance
 
 
-def _clone_plan(request, instance):
-    """The clone's settings from the request and the source as it is now,
-    and why it cannot be made, or None."""
-    clone_data = dict()
-    clone_data["name"] = request.POST.get("name", "").strip()
-    clone_data["clone-title"] = request.POST.get("clone-title", "").strip()
-    clone_data["clone-description"] = request.POST.get("clone-description", "").strip()
-
-    # the clone copies the disks of the persistent definition
-    disk_sum = sum(int(disk.get("size") or 0) >> 30 for disk in instance.config_disks)
-    quota_msg = utils.check_user_quota(
-        request.user, 1, instance.vcpu, instance.memory, disk_sum
-    )
-
-    clone_data["disk_owner_uid"] = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID)
-    clone_data["disk_owner_gid"] = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID)
-
-    # Only superusers may choose disk names and MAC addresses; the form only
-    # shows those fields to them.
-    if request.user.is_superuser:
-        for key, value in request.POST.items():
-            if CLONE_POST_KEY_RE.fullmatch(key):
-                clone_data[key] = value.strip()
-
-    if app_settings.CLONE_INSTANCE_AUTO_NAME == "True" and not clone_data["name"]:
-        auto_vname = utils.get_clone_free_names()[0]
-        clone_data["name"] = auto_vname
-        clone_data["clone-net-mac-0"] = utils.get_dhcp_mac_address(auto_vname)
-        for disk in instance.config_disks:
-            disk_dev = f"disk-{disk['dev']}"
-            disk_name = utils.get_clone_disk_name(disk, instance.name, auto_vname)
-            clone_data[disk_dev] = disk_name
-
-    if not request.user.is_superuser:
-        for disk in instance.config_disks:
-            clone_data[f"disk-{disk['dev']}"] = utils.get_clone_disk_name(
-                disk, instance.name, clone_data["name"]
-            )
-        for num in range(max(1, len(instance.config_networks))):
-            key = f"clone-net-mac-{num}"
-            if not clone_data.get(key):
-                clone_data[key] = (
-                    utils.get_dhcp_mac_address(clone_data["name"]) if num == 0 else ""
-                ) or utils.get_random_mac_address()
-
-    check_instance = Instance.objects.filter(name=clone_data["name"])
-    invalid_macs = [
-        value
-        for key, value in clone_data.items()
-        if key.startswith("clone-net-mac-") and not utils.MAC_ADDRESS_RE.fullmatch(value)
-    ]
-    invalid_disks = [
-        value
-        for key, value in clone_data.items()
-        # None: a disk without a volume, which clone_instance does not copy.
-        if key.startswith("disk-")
-        and value is not None
-        and not VOLUME_NAME_RE.fullmatch(value)
-    ]
-
-    if not request.user.is_superuser and quota_msg:
-        msg = quota_refused(quota_msg, _(
-            "User '%(quota_msg)s' quota reached, cannot create '%(clone_name)s'!"
-        ) % {
-            "quota_msg": quota_msg,
-            "clone_name": clone_data["name"],
-        })
-        return clone_data, msg
-    if check_instance:
-        msg = _("Instance '%(clone_name)s' already exists!") % {
-            "clone_name": clone_data["name"]
-        }
-        return clone_data, msg
-    if not re.match(r"^[a-zA-Z0-9-]+$", clone_data["name"]):
-        msg = _("Instance name '%(clone_name)s' contains invalid characters!") % {
-            "clone_name": clone_data["name"]
-        }
-        return clone_data, msg
-    if "clone-net-mac-0" not in clone_data or invalid_macs:
-        msg = _("Instance MAC '%(clone_mac)s' invalid format!") % {
-            "clone_mac": ", ".join(invalid_macs)
-        }
-        return clone_data, msg
-    if invalid_disks:
-        msg = _("Disk name '%(disk_name)s' contains invalid characters!") % {
-            "disk_name": ", ".join(str(v) for v in invalid_disks)
-        }
-        return clone_data, msg
-    return clone_data, None
-
-
 @require_POST
 @permission_required("instances.clone_instances", raise_exception=True)
 @serialize_user_quota
 def clone(request, pk):
-    instance = _clone_source(request.user, pk)
+    _clone_source(request.user, pk)  # 403/404 first
     try:
-        # The whole compute: the destination name and disk names must not be
-        # taken by a concurrent clone of another VM while volumes are copied.
-        with libvirt_compute_lock(instance.compute):
-            # Before this request took the claim, a migration may have moved the
-            # source, or its template mark, size or the user's rights changed:
-            # the clone is checked against the source as it is now.
-            if _moved_meanwhile(pk, instance):
-                return _moved(request, pk)
-            instance = _clone_source(request.user, pk)
-            clone_data, refusal = _clone_plan(request, instance)
-            if refusal:
-                messages.error(request, refusal)
-                return get_safe_redirect(request, default=reverse("instances:instance", args=[pk]))
-            new_uuid = instance.proxy.clone_instance(clone_data)
-            new_instance = Instance.objects.get_or_create(
-                compute=instance.compute,
-                uuid=new_uuid,
-                defaults={"name": clone_data["name"]},
-            )[0]
-            if new_instance.name != clone_data["name"]:
-                new_instance.name = clone_data["name"]
-                new_instance.save(update_fields=["name"])
-            UserInstance.objects.get_or_create(
-                instance_id=new_instance.id,
-                user_id=request.user.id,
-                defaults={"is_delete": True, "is_change": True, "is_vnc": True},
-            )
-        msg = _("Create a clone of '%(instance_name)s'") % {
-            "instance_name": instance.name
-        }
-        messages.success(request, msg)
-        addlogmsg(
-            request.user.username, instance.compute.name, new_instance.name, msg
-        )
-
-        return redirect(reverse("instances:instance", args=[new_instance.id]))
+        op = jobs.submit("instance.clone", request.user, {"instance": pk, **request.POST.dict()},
+                         request.POST.get("submission_key", ""))
     except claims.Busy as busy:
         return _busy(request, pk, busy)
-    except (libvirtError, ValueError, OSError) as e:
-        messages.error(request, error_text(request, e))
-
-    return get_safe_redirect(request, default=reverse("instances:instance", args=[pk]))
+    except (jobs.SubmissionConflict, libvirtError, ValueError) as err:
+        messages.error(request, error_text(request, err))
+    else:
+        messages.info(request, _("The clone is being made; follow it in the job panel."))
+        jobs.remember_job(request, op)
+    return redirect(reverse("instances:instance", args=[pk]))
 
 
 @require_POST
diff --git a/operations/claims.py b/operations/claims.py
index 446b6eda..108ab358 100644
--- a/operations/claims.py
+++ b/operations/claims.py
@@ -48,6 +48,10 @@ def volume_key(compute_id, pool, name):
     return f"volume:{compute_id}:{pool}/{name}"
 
 
+def name_key(compute_id, name):
+    return f"name:{compute_id}:{name}"
+
+
 def quota_key(user_id):
     return f"quota:user:{user_id}"
 
diff --git a/vrtManager/instance.py b/vrtManager/instance.py
index c7350bcf..571fc73a 100644
--- a/vrtManager/instance.py
+++ b/vrtManager/instance.py
@@ -1923,7 +1923,7 @@ class wvmInstance(wvmConnect):
         mac_tuples = [mac[i : i + n] for i in range(0, len(mac), n)]
         return ":".join(mac_tuples)
 
-    def clone_instance(self, clone_data):
+    def clone_instance(self, clone_data, uuid):
         """Copy a shut-off VM, with its disks and UEFI variables (NVRAM).
         Inputs are checked before anything is allocated, and on failure only
         the volumes this call created are removed. Returns the clone's UUID."""
@@ -1935,10 +1935,10 @@ class wvmInstance(wvmConnect):
 
         tree = etree.fromstring(self._XMLDesc(PERSISTENT_XML))
         tree.find("name").text = clone_data["name"]
-        uuid = tree.find("uuid")
-        if uuid is None:
-            uuid = etree.SubElement(tree, "uuid")
-        uuid.text = util.randomUUID()
+        uuid_el = tree.find("uuid")
+        if uuid_el is None:
+            uuid_el = etree.SubElement(tree, "uuid")
+        uuid_el.text = uuid
         self._set_options(
             tree,
             {
diff --git a/vrtManager/tests/test_clone_nvram.py b/vrtManager/tests/test_clone_nvram.py
index 37328b7c..12f650c6 100644
--- a/vrtManager/tests/test_clone_nvram.py
+++ b/vrtManager/tests/test_clone_nvram.py
@@ -41,7 +41,7 @@ def proxy(nvram):
 class CloneNvramTestCase(unittest.TestCase):
     def test_the_copy_keeps_the_format_owner_and_mode(self):
         p = proxy("<nvram format='qcow2'>/var/lib/libvirt/qemu/nvram/vm_VARS.qcow2</nvram>")
-        p.clone_instance({"name": "copy"})
+        p.clone_instance({"name": "copy"}, "abababab-0000-1111-2222-333333333333")
         xml, original, _ = p.pool.createXMLFrom.call_args.args
         volume = etree.fromstring(xml)
         self.assertEqual(volume.findtext("name"), "copy_VARS.qcow2")
@@ -54,7 +54,7 @@ class CloneNvramTestCase(unittest.TestCase):
 
     def test_an_nvram_source_element_points_at_the_copy(self):
         p = proxy("<nvram type='file'>\n  <source file='/var/lib/libvirt/qemu/nvram/vm_VARS.qcow2'/>\n</nvram>")
-        p.clone_instance({"name": "copy"})
+        p.clone_instance({"name": "copy"}, "abababab-0000-1111-2222-333333333333")
         p.pool.storageVolLookupByName.assert_called_once_with("vm_VARS.qcow2")
         (defined,), _ = p._defineXML.call_args
         self.assertEqual(
@@ -64,5 +64,5 @@ class CloneNvramTestCase(unittest.TestCase):
     def test_an_nvram_that_is_not_a_file_is_refused_before_any_copy(self):
         p = proxy("<nvram type='network'><source protocol='iscsi' name='x'/></nvram>")
         with self.assertRaises(util.OperationError):
-            p.clone_instance({"name": "copy"})
+            p.clone_instance({"name": "copy"}, "abababab-0000-1111-2222-333333333333")
         p.pool.createXMLFrom.assert_not_called()
diff --git a/vrtManager/tests/test_xml_safety.py b/vrtManager/tests/test_xml_safety.py
index 06bca9ee..9cf7a40b 100644
--- a/vrtManager/tests/test_xml_safety.py
+++ b/vrtManager/tests/test_xml_safety.py
@@ -142,7 +142,7 @@ class TestCloneRollback(unittest.TestCase):
         inst._defineXML = MagicMock(side_effect=libvirt.libvirtError("bad definition"))
 
         with self.assertRaises(libvirt.libvirtError):
-            inst.clone_instance({"name": "copy", "disk-vda": "copy.qcow2", "disk_owner_uid": 0, "disk_owner_gid": 0})
+            inst.clone_instance({"name": "copy", "disk-vda": "copy.qcow2", "disk_owner_uid": 0, "disk_owner_gid": 0}, "abababab-0000-1111-2222-333333333333")
         copy.delete.assert_called_once_with(0)
         source_vol.delete.assert_not_called()
 
@@ -153,7 +153,7 @@ class TestCloneRollback(unittest.TestCase):
         inst.get_status = MagicMock(return_value=1)
         inst._XMLDesc = MagicMock()
         with self.assertRaisesRegex(libvirt.libvirtError, "Shut the VM down"):
-            inst.clone_instance({"name": "copy"})
+            inst.clone_instance({"name": "copy"}, "abababab-0000-1111-2222-333333333333")
         inst._XMLDesc.assert_not_called()
 
 
@@ -352,7 +352,7 @@ class TestEmptyDiskClone(unittest.TestCase):
         # clone is still defined and looked up through the real test driver.
         inst._XMLDesc = MagicMock(return_value=etree.tostring(config).decode())
         inst.get_status = MagicMock(return_value=5)  # only shut-off VMs are cloned
-        clone_uuid = inst.clone_instance({"name": "empty-disk-clone"})
+        clone_uuid = inst.clone_instance({"name": "empty-disk-clone"}, "abababab-0000-1111-2222-333333333333")
         clone = conn.lookupByUUIDString(clone_uuid)
         self.addCleanup(clone.undefine)
         clone_config = etree.fromstring(clone.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE))
@@ -371,5 +371,5 @@ class TestEmptyDiskClone(unittest.TestCase):
         inst._XMLDesc = MagicMock()
         inst.get_status = MagicMock(return_value=5)
         with self.assertRaisesRegex(ValueError, "already exists"):
-            inst.clone_instance({"name": "test"})
+            inst.clone_instance({"name": "test"}, "abababab-0000-1111-2222-333333333333")
         inst._XMLDesc.assert_not_called()