diff --git a/backend/src/backup_tool/api/app.py b/backend/src/backup_tool/api/app.py
index 491c910..723620d 100644
--- a/backend/src/backup_tool/api/app.py
+++ b/backend/src/backup_tool/api/app.py
@@ -29,6 +29,7 @@ from backup_tool.db.models import (
     Source,
     User,
 )
+from backup_tool.execution import EnqueueError, enqueue
 from backup_tool.repository import (
     RepositoryError,
     initialize,
@@ -750,6 +751,32 @@ def create_app(settings: Settings) -> FastAPI:
             "state": job.state,
         }
 
+    @app.post("/api/v2/jobs/{job_id}/executions", status_code=202)
+    async def enqueue_execution(
+        job_id: str,
+        db: Annotated[AsyncSession, Depends(session)],
+        _: Annotated[tuple[User, set[str], bool], Depends(require)],
+    ) -> dict[str, Any]:
+        try:
+            execution = await enqueue(db, job_id)
+        except EnqueueError as error:
+            if error.code == "execution_active":
+                raise Problem(
+                    409,
+                    "execution_active",
+                    f"Job already has active execution {error.active_execution_id}.",
+                ) from error
+            if error.code == "job_disabled":
+                raise Problem(409, "job_disabled", str(error)) from error
+            raise Problem(404, "resource_not_found", str(error)) from error
+        return {
+            "id": execution.id,
+            "job_id": execution.job_id,
+            "state": execution.state,
+            "trigger": execution.trigger,
+            "attempt": execution.attempt,
+        }
+
     @app.get("/api/v2/audit")
     async def list_audit(
         db: Annotated[AsyncSession, Depends(session)],
diff --git a/backend/src/backup_tool/execution.py b/backend/src/backup_tool/execution.py
index a3479ac..d64bd38 100644
--- a/backend/src/backup_tool/execution.py
+++ b/backend/src/backup_tool/execution.py
@@ -1,6 +1,13 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backup_tool.db.models import Execution, Job
 
 ACTIVE_STATES = frozenset({"queued", "preparing", "running", "verifying", "cancelling"})
 TERMINAL_STATES = frozenset({"committed", "cancelled", "failed"})
@@ -21,8 +28,97 @@ class TransitionError(ValueError):
     code = "invalid_transition"
 
 
+class EnqueueError(ValueError):
+    def __init__(self, code: str, detail: str, active_execution_id: str | None = None):
+        super().__init__(detail)
+        self.code = code
+        self.active_execution_id = active_execution_id
+
+
 def transition(current: str, target: str) -> str:
     """Validate and return one durable execution-state transition."""
     if target not in _ALLOWED.get(current, frozenset()):
         raise TransitionError(f"Cannot transition execution from {current!r} to {target!r}.")
     return target
+
+
+async def enqueue(db: AsyncSession, job_id: str, trigger: str = "manual") -> Execution:
+    """Create exactly one active execution for an enabled active job."""
+    job = await db.get(Job, job_id)
+    if job is None:
+        raise EnqueueError("resource_not_found", "Job was not found.")
+    if job.state != "active" or not job.enabled:
+        raise EnqueueError("job_disabled", "Job is disabled or archived.")
+    job_identifier = job.id
+    execution = Execution(job_id=job_identifier, trigger=trigger, progress={})
+    db.add(execution)
+    try:
+        await db.commit()
+    except IntegrityError as error:
+        await db.rollback()
+        existing = await db.scalar(
+            select(Execution.id).where(
+                Execution.job_id == job_identifier, Execution.state.in_(ACTIVE_STATES)
+            )
+        )
+        raise EnqueueError(
+            "execution_active", "Job already has an active execution.", existing
+        ) from error
+    await db.refresh(execution)
+    return execution
+
+
+async def claim(
+    db: AsyncSession, execution_id: str, owner: str, lease_seconds: int = 60
+) -> Execution | None:
+    """Atomically lease a queued execution; a stale lease may be recovered."""
+    now = datetime.now(UTC)
+    expires = now + timedelta(seconds=lease_seconds)
+    result = await db.execute(
+        update(Execution)
+        .where(
+            Execution.id == execution_id,
+            Execution.state.in_({"queued", "preparing"}),
+            (Execution.lease_expires_at.is_(None)) | (Execution.lease_expires_at < now),
+        )
+        .values(state="preparing", lease_owner=owner, lease_expires_at=expires, heartbeat_at=now)
+    )
+    if getattr(result, "rowcount", 0) != 1:
+        await db.rollback()
+        return None
+    await db.commit()
+    return await db.get(Execution, execution_id)
+
+
+async def heartbeat(
+    db: AsyncSession, execution_id: str, owner: str, lease_seconds: int = 60
+) -> bool:
+    now = datetime.now(UTC)
+    result = await db.execute(
+        update(Execution)
+        .where(
+            Execution.id == execution_id,
+            Execution.lease_owner == owner,
+            Execution.state.in_(ACTIVE_STATES),
+        )
+        .values(heartbeat_at=now, lease_expires_at=now + timedelta(seconds=lease_seconds))
+    )
+    await db.commit()
+    return getattr(result, "rowcount", 0) == 1
+
+
+async def recover_stale(db: AsyncSession) -> int:
+    now = datetime.now(UTC)
+    result = await db.execute(
+        update(Execution)
+        .where(Execution.state.in_({"preparing", "running"}), Execution.lease_expires_at < now)
+        .values(
+            state="queued",
+            lease_owner=None,
+            lease_expires_at=None,
+            heartbeat_at=None,
+            reason_code="worker_lost",
+        )
+    )
+    await db.commit()
+    return getattr(result, "rowcount", 0) or 0
diff --git a/tests/integration/test_sources_jobs.py b/tests/integration/test_sources_jobs.py
index 4c2ffe3..cb4130c 100644
--- a/tests/integration/test_sources_jobs.py
+++ b/tests/integration/test_sources_jobs.py
@@ -84,6 +84,16 @@ async def test_local_source_probe_archive_and_repository_targeted_job(
         )
         assert job.status_code == 201
         assert "destination_path" not in job.json()
+        execution = await client.post(
+            f"/api/v2/jobs/{job.json()['id']}/executions", headers=headers
+        )
+        assert execution.status_code == 202
+        assert execution.json()["state"] == "queued"
+        duplicate = await client.post(
+            f"/api/v2/jobs/{job.json()['id']}/executions", headers=headers
+        )
+        assert duplicate.status_code == 409
+        assert duplicate.json()["code"] == "execution_active"
         archived = await client.delete(f"/api/v2/sources/{source_id}", headers=headers)
         assert archived.status_code == 204
         assert (
diff --git a/tests/unit/test_execution_lifecycle.py b/tests/unit/test_execution_lifecycle.py
index ca85851..9671632 100644
--- a/tests/unit/test_execution_lifecycle.py
+++ b/tests/unit/test_execution_lifecycle.py
@@ -1,6 +1,11 @@
 from __future__ import annotations
 
-from backup_tool.execution import ACTIVE_STATES, TERMINAL_STATES, TransitionError, transition
+from backup_tool.execution import (
+    ACTIVE_STATES,
+    TERMINAL_STATES,
+    TransitionError,
+    transition,
+)
 
 
 def test_transitions_are_monotonic_and_terminal_states_cannot_change() -> None: