diff --git a/backend/backup/engine.py b/backend/backup/engine.py
index 1babed3..922c91c 100644
--- a/backend/backup/engine.py
+++ b/backend/backup/engine.py
@@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select
 from app.models import Job, JobExecution, Backup
 from backup.adapters import get_adapter
+from backup.retention import RetentionPolicy
 
 class BackupEngine:
     def __init__(self, db: AsyncSession):
@@ -109,6 +110,13 @@ class BackupEngine:
                 execution.bytes_processed = total_processed
                 execution.bytes_backed_up = total_backed_up
                 
+                # Apply retention policy
+                retention = RetentionPolicy(self.db)
+                keep_count = getattr(job, 'retention_count', None)
+                keep_days = getattr(job, 'retention_days', None)
+                if keep_count or keep_days:
+                    await retention.apply_retention_for_job(job_id, keep_count, keep_days)
+                
             finally:
                 await adapter.disconnect()
             
diff --git a/backend/backup/retention.py b/backend/backup/retention.py
new file mode 100644
index 0000000..efb421d
--- /dev/null
+++ b/backend/backup/retention.py
@@ -0,0 +1,90 @@
+import shutil
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+from typing import List, Optional
+from sqlalchemy import select
+
+from app.models import Backup
+
+
+class RetentionPolicy:
+    def __init__(self, db):
+        self.db = db
+
+    async def apply_retention_for_job(
+        self,
+        job_id: int,
+        keep_count: Optional[int] = None,
+        keep_days: Optional[int] = None
+    ) -> List[Backup]:
+        """
+        Apply retention policy for a job's backups.
+
+        Args:
+            job_id: The job ID to apply retention for
+            keep_count: Maximum number of backups to keep (oldest deleted first)
+            keep_days: Delete backups older than this many days
+
+        Returns:
+            List of deleted backups
+        """
+        deleted_backups = []
+
+        result = await self.db.execute(
+            select(Backup)
+            .where(Backup.execution.has(job_id=job_id))
+            .order_by(Backup.created_at.asc())
+        )
+        backups = result.scalars().all()
+
+        if not backups:
+            return deleted_backups
+
+        backups_to_delete = set()
+
+        if keep_count is not None and len(backups) > keep_count:
+            backups_to_delete.update(backups[:-keep_count])
+
+        if keep_days is not None:
+            cutoff_date = datetime.now(timezone.utc) - timedelta(days=keep_days)
+            for backup in backups:
+                if backup.created_at < cutoff_date:
+                    backups_to_delete.add(backup)
+
+        for backup in list(backups_to_delete):
+            await self._delete_backup(backup)
+            deleted_backups.append(backup)
+
+        await self.db.commit()
+        return deleted_backups
+
+    async def _delete_backup(self, backup: Backup):
+        """Delete a backup and its storage."""
+        try:
+            storage_path = Path(backup.storage_path)
+            if storage_path.exists():
+                shutil.rmtree(storage_path)
+        except Exception:
+            pass
+
+        await self.db.delete(backup)
+
+    async def cleanup_orphaned_backups(self) -> int:
+        """
+        Remove backup records whose storage no longer exists.
+
+        Returns:
+            Number of orphaned backups removed
+        """
+        result = await self.db.execute(select(Backup))
+        backups = result.scalars().all()
+
+        removed_count = 0
+        for backup in backups:
+            storage_path = Path(backup.storage_path)
+            if not storage_path.exists():
+                await self.db.delete(backup)
+                removed_count += 1
+
+        await self.db.commit()
+        return removed_count