diff --git a/admin/frontend/dashboard/src/components/sites/BackupConfigDialog.vue b/admin/frontend/dashboard/src/components/sites/BackupConfigDialog.vue
index 13edfb4cb..828b76239 100644
--- a/admin/frontend/dashboard/src/components/sites/BackupConfigDialog.vue
+++ b/admin/frontend/dashboard/src/components/sites/BackupConfigDialog.vue
@@ -105,7 +105,7 @@ const open = async () => {
   show.value = true
   try {
     const data = await sitesApi.backups.schedule.get(props.siteName)
-    isEnabled.value = !!data.schedule
+    isEnabled.value = !!data.schedule || !!data.retention?.calendar_rotation
     applySchedule(data.schedule)
     applyRetention(data.retention)
   } catch (e) {
diff --git a/admin/frontend/dashboard/src/components/sites/Backups.vue b/admin/frontend/dashboard/src/components/sites/Backups.vue
index 619979416..357279348 100644
--- a/admin/frontend/dashboard/src/components/sites/Backups.vue
+++ b/admin/frontend/dashboard/src/components/sites/Backups.vue
@@ -41,13 +41,15 @@ const error = ref('')
 
 const configRef = ref<InstanceType<typeof BackupConfigDialog> | null>(null)
 const config = ref<BackupSchedule | null>(null)
-const enabled = computed(() => !!config.value?.schedule)
+const isDefault = computed(() => !!config.value?.retention?.calendar_rotation)
+const enabled = computed(() => !!config.value?.schedule || isDefault.value)
 
-const scheduleSummary = computed(() =>
-  enabled.value
+const scheduleSummary = computed(() => {
+  if (isDefault.value) return 'Daily, with the default retention policy.'
+  return enabled.value
     ? `${cronToLabel(config.value?.schedule)}.`
-    : 'Manual backups are kept until you delete them.',
-)
+    : 'Manual backups are kept until you delete them.'
+})
 
 const loadConfig = async () => {
   try {
diff --git a/docs/configuration.md b/docs/configuration.md
index 5409e89ce..1d4ab2115 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -262,6 +262,12 @@ Set `proxy.cloudflare = true` when site domains use Cloudflare's proxy. nginx th
 
 `[frappe_cloud] url` is the Frappe Cloud that a site restore copies backups from. The default is `https://cloud.frappe.io`. Set it to use another Frappe Cloud, for example a local Press during development.
 
+### Scheduled Backups
+
+Each bench has one hourly `0 * * * *` cron job for scheduled backups. It is installed with the first enrolled site. Sites without a custom backup schedule use this default. New sites are enrolled, including sites created from a backup, and the `enroll_sites_in_scheduled_backups` patch enrolls existing sites that have no backup schedule. Each run reads the last 24 hours of the bench audit log once, then queues a logical backup of the database and site config for each enrolled site with no successful backup in the last 23 hours. Measuring 23 hours keeps each site's backup at the same hour every day. A backup also includes public and private files when the site has no file backup that day (UTC). Physical backups are not scheduled. After six failed backups in 24 hours, the job skips the site until the count drops. The sixth failure since the last successful backup adds a "Backups of <site> are failing" notification to the bench feed. Scheduled backups stay on disk in `sites/<site>/backups`, outside the directory that Frappe prunes. When S3 is configured, each backup is also uploaded and expired runs are deleted from S3. The host must have cron and the Pilot task worker running. A scheduled backup does not start while another task holds the site.
+
+After each backup, Pilot clears expired runs with the default GFS policy: every run is kept for 7 days, Sunday runs for 28 days, runs on the first day of a month for 366 days, and January 1 runs for 3,653 days. The newest run is always kept. The `backup_retention` object in `site_config.json` marks this policy with `scheme: "gfs"` and `calendar_rotation: true`. A custom schedule set in the dashboard or through the backup schedule API replaces the default. It installs a cron job for the site and stores a GFS or FIFO retention record without `calendar_rotation`, so the bench job skips the site and the count-based policy clears its runs. Turning off automated backups removes the schedule and the retention record, and the site then keeps only manual backups.
+
 ### Telemetry
 
 `[telemetry]` is one credential for both shippers. `endpoint` is the base URL of the region's Datum, with no path: metrics append `/v1/ingest` and logs append `/v1/logs/ingest`. `token` is the JWT Central mints for the region, which Datum tells apart by route rather than by token, so one serves both. `logs_enabled` and `metrics_enabled` switch each shipper independently.
diff --git a/pilot/config/backup.py b/pilot/config/backup.py
index 5c3644f7c..620447962 100644
--- a/pilot/config/backup.py
+++ b/pilot/config/backup.py
@@ -14,6 +14,7 @@ class BackupConfig:
     keep_weekly: int = 5  # one per week across the current month
     keep_monthly: int = 12  # one per month, past year
     keep_yearly: int = 5  # one per year, kept a few years
+    calendar_rotation: bool = False
 
     @property
     def counts(self) -> dict[str, int]:
diff --git a/pilot/core/bench/__init__.py b/pilot/core/bench/__init__.py
index abd03667b..d6bdcf095 100644
--- a/pilot/core/bench/__init__.py
+++ b/pilot/core/bench/__init__.py
@@ -12,6 +12,7 @@
 if TYPE_CHECKING:
     from pilot.config import S3Config, SiteConfig
     from pilot.core.app import App, NewAppOptions, RevisionPin
+    from pilot.core.bench.backups import BenchBackups
     from pilot.core.bench.hostname_aliases import HostnameAliases
     from pilot.core.bench.migration.store import MigrationStore
     from pilot.core.bench.processes import BenchProcesses
@@ -103,6 +104,12 @@ def processes(self) -> "BenchProcesses":
 
         return BenchProcesses(self)
 
+    @cached_property
+    def backups(self) -> "BenchBackups":
+        from pilot.core.bench.backups import BenchBackups
+
+        return BenchBackups(self)
+
     @cached_property
     def notifications(self) -> "NotificationStore":
         from pilot.core.notification import NotificationStore
diff --git a/pilot/core/bench/audit_log.py b/pilot/core/bench/audit_log.py
index 3c32873ea..d9eaf6eb2 100644
--- a/pilot/core/bench/audit_log.py
+++ b/pilot/core/bench/audit_log.py
@@ -4,6 +4,7 @@
 
 from collections.abc import Callable
 from dataclasses import dataclass
+from datetime import datetime
 
 from pilot.internal.jsonl_log import JsonlLog, utc_now
 
@@ -67,10 +68,14 @@ def __init__(self, bench) -> None:
     def append(self, entry_type: str, entry: dict) -> None:
         self._log.append({"type": entry_type, "logged_at": utc_now(), **entry})
 
-    def entries(self, entry_type=None, site=None, status=None, jti=None, limit=None) -> list[dict]:
+    def entries(
+        self, entry_type=None, site=None, status=None, jti=None, limit=None, since: datetime | None = None
+    ) -> list[dict]:
         """Return matching records newest first across weekly files."""
         matched = []
         for record in self._log.read_newest_first():
+            if since is not None and datetime.fromisoformat(record["logged_at"]) < since:
+                break
             if self._matches(record, entry_type, site, status, jti):
                 matched.append(record)
                 if limit is not None and len(matched) >= limit:
diff --git a/pilot/core/bench/backups.py b/pilot/core/bench/backups.py
new file mode 100644
index 000000000..523d6561e
--- /dev/null
+++ b/pilot/core/bench/backups.py
@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import sys
+from collections import defaultdict
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from pilot.core.bench.audit_log import AuditLog
+
+if TYPE_CHECKING:
+    from pilot.core.bench import Bench
+
+CRON_JOB_KEY = "scheduled-backups"
+CRON_SCHEDULE = "0 * * * *"
+
+
+class BenchBackups:
+    """The hourly scheduled backup poll for every site on a bench."""
+
+    def __init__(self, bench: Bench) -> None:
+        self.bench = bench
+
+    def install_schedule(self) -> None:
+        from pilot.managers.cron import CronManager, cron_module_command
+
+        command = cron_module_command(
+            "pilot.core.bench.backups", [self.bench.path], self.bench.logs_path / "scheduled-backups.log"
+        )
+        CronManager(self.bench.path).set_schedule(CRON_JOB_KEY, CRON_SCHEDULE, command)
+
+    def get_recent_history(self, now: datetime) -> dict[str, list[tuple[datetime, dict]]]:
+        """Backup records from the last 24 hours, newest first, by site. One audit log read."""
+        history = defaultdict(list)
+        for entry in AuditLog(self.bench).entries(entry_type="backup", since=now - timedelta(days=1)):
+            if entry.get("event") == "backup":
+                history[entry["site"]].append((datetime.fromisoformat(entry["logged_at"]), entry))
+        return history
+
+    def queue_scheduled(self, now: datetime | None = None) -> None:
+        now = now or datetime.now(UTC)
+        history = self.get_recent_history(now)
+        for site in self.bench.sites():
+            site.backups.queue_scheduled(now, history.get(site.config.name, []))
+
+
+if __name__ == "__main__":
+    from pilot.core.bench import Bench
+
+    BenchBackups(Bench(Path(sys.argv[1]))).queue_scheduled()
diff --git a/pilot/core/notification/events.py b/pilot/core/notification/events.py
index 03aaeee5a..67701ffa5 100644
--- a/pilot/core/notification/events.py
+++ b/pilot/core/notification/events.py
@@ -107,6 +107,22 @@ def record_alert(
         return False
 
 
+def backups_failing(bench: "Bench", site: str, failures: int) -> None:
+    """Raised once when a site reaches the failure limit since its last successful backup."""
+    try:
+        bench.notifications.create(
+            f"Backups of {site} are failing",
+            category="Sites",
+            event="backups_failing",
+            severity="Error",
+            message=f"{failures} backup attempts failed since the last successful backup.",
+            site=site,
+            action_route=f"/sites/{site}",
+        )
+    except Exception as exc:
+        logging.warning("Notification skipped for failing backups of %s: %s", site, exc)
+
+
 def _readable(command: str) -> str:
     """`backup-site` -> `Backup site`, so the title reads like a sentence."""
     return command.replace("-", " ").capitalize()
diff --git a/pilot/core/site/__init__.py b/pilot/core/site/__init__.py
index a2afd6b9f..6d47d20ab 100644
--- a/pilot/core/site/__init__.py
+++ b/pilot/core/site/__init__.py
@@ -329,11 +329,15 @@ def provision(
         admin_password: str,
         db_type: str | None = None,
         on_progress: Callable[[str], None] = lambda message: None,
+        *,
+        schedule_backups: bool = True,
     ) -> "Site":
         """Create a site, install apps, reload nginx, and issue TLS if needed."""
         from pilot.core.site.provisioning import SiteProvisioner
 
-        return SiteProvisioner(bench, name, apps, admin_password, db_type).provision(on_progress)
+        return SiteProvisioner(bench, name, apps, admin_password, db_type).provision(
+            on_progress, schedule_backups=schedule_backups
+        )
 
     def set_ssl(self, enabled: bool) -> None:
         set_site_ssl_flag(self.bench.sites_path, self.config.name, enabled)
diff --git a/pilot/core/site/backups.py b/pilot/core/site/backups.py
index 30b438934..b8491c3a4 100644
--- a/pilot/core/site/backups.py
+++ b/pilot/core/site/backups.py
@@ -2,9 +2,12 @@
 
 import re
 from dataclasses import asdict
+from datetime import datetime, timedelta
+from hashlib import sha256
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from pilot.config.backup import BackupConfig
 from pilot.config.site_backup import clear_retention, read_retention, write_retention
 from pilot.core.site.retention import BackupRetentionPolicy
 from pilot.integrations.s3.backups import OffsiteBackup
@@ -13,6 +16,7 @@
 if TYPE_CHECKING:
     from pilot.core.site import Site
 
+MAX_DAILY_FAILURES = 6
 _TS_RE = re.compile(r"^(\d{8}_\d{6})")
 
 
@@ -153,7 +157,7 @@ def schedule(self) -> dict:
         retention = read_retention(self.site.path / "site_config.json")
         return {"schedule": schedule, "retention": asdict(retention) if retention else None}
 
-    def set_schedule(self, schedule: str, retention) -> dict:
+    def set_schedule(self, schedule: str, retention: BackupConfig) -> dict:
         from pilot.managers.cron import CronManager
 
         CronManager(self.site.bench.path).set_schedule(
@@ -164,6 +168,68 @@ def set_schedule(self, schedule: str, retention) -> dict:
         write_retention(self.site.path / "site_config.json", retention)
         return self.schedule()
 
+    def enable_default_schedule(self) -> None:
+        """Default backups for a site without a custom schedule. `set_schedule` replaces them.
+        The bench job goes in first, so a failed install leaves the site unenrolled for a retry."""
+        self.site.bench.backups.install_schedule()
+        write_retention(
+            self.site.path / "site_config.json",
+            BackupConfig(keep_weekly=4, keep_yearly=10, calendar_rotation=True),
+        )
+
+    @property
+    def is_default_schedule_enabled(self) -> bool:
+        retention = read_retention(self.site.path / "site_config.json")
+        return retention is not None and retention.calendar_rotation
+
+    def get_recent_backup_history(self, now: datetime) -> list[tuple[datetime, dict]]:
+        return self.site.bench.backups.get_recent_history(now).get(self.site.config.name, [])
+
+    def get_failures_since_last_success(self, now: datetime) -> int:
+        failures = 0
+        for _, entry in self.get_recent_backup_history(now):
+            if entry["status"] == "success":
+                break
+            failures += entry["status"] == "failed"
+        return failures
+
+    def get_scheduled_backup(
+        self, now: datetime, history: list[tuple[datetime, dict]] | None = None
+    ) -> dict[str, bool] | None:
+        """Daily logical backup, due 23 hours after the last success so it keeps its hour. Files once a day."""
+        if not self.is_default_schedule_enabled:
+            return None
+
+        backups = self.get_recent_backup_history(now) if history is None else history
+        successful = [(when, entry) for when, entry in backups if entry["status"] == "success"]
+        failures = sum(entry["status"] == "failed" for _, entry in backups)
+        if failures >= MAX_DAILY_FAILURES or any(when > now - timedelta(hours=23) for when, _ in successful):
+            return None
+
+        has_file_backup_today = any(
+            entry.get("with_files") and when.date() == now.date() for when, entry in successful
+        )
+        return {"with_files": not has_file_backup_today}
+
+    def queue_scheduled(self, now: datetime, history: list[tuple[datetime, dict]] | None = None) -> None:
+        from pilot.exceptions import TaskConflictError
+        from pilot.tasks.backup_site import BackupSiteTask
+
+        if self.get_scheduled_backup(now, history) is None:
+            return
+
+        slot = f"{self.site.config.name}:{now:%Y%m%d%H}"
+        try:
+            BackupSiteTask.queue(
+                self.site.bench,
+                site=self.site.config.name,
+                automatic=True,
+                idempotency_key=f"scheduled-backup:{sha256(slot.encode()).hexdigest()}",
+                resource_key=f"site:{self.site.config.name.lower()}",
+            )
+        except TaskConflictError:
+            print(f"Skipped scheduled backup: {self.site.config.name} has an active site task.")
+
     def move_schedule(self, old_name: str) -> None:
         """Carry a renamed site's backup schedule over from its old name."""
         from pilot.managers.cron import CronManager
diff --git a/pilot/core/site/provisioning.py b/pilot/core/site/provisioning.py
index 6b0280942..40561ca16 100644
--- a/pilot/core/site/provisioning.py
+++ b/pilot/core/site/provisioning.py
@@ -29,7 +29,7 @@ def __init__(
         self.admin_password = admin_password
         self.db_type = db_type
 
-    def provision(self, on_progress: Callable[[str], None]) -> "Site":
+    def provision(self, on_progress: Callable[[str], None], *, schedule_backups: bool = True) -> "Site":
         from pilot.core.site import Site
 
         via_wildcard = validate_new_site(self.bench, self.name, self.apps)
@@ -57,12 +57,14 @@ def provision(self, on_progress: Callable[[str], None]) -> "Site":
         self.write_pilot_communication_config(site)
         self.bench.write_common_site_config()
         site.enable_scheduler()
-        on_progress(f"\nSite '{self.name}' created successfully.")
         self.build_missing_assets()
         self.add_to_hosts(site)
         self.reload_nginx()
         if origin_tls:
             self.obtain_cert(site, on_progress)
+        if schedule_backups:
+            site.backups.enable_default_schedule()
+        on_progress(f"\nSite '{self.name}' created successfully.")
         return site
 
     def write_email_sender_name(self, site: "Site") -> None:
@@ -203,12 +205,13 @@ def provision_from_backup(
     from pilot.core.site import Site
     from pilot.core.site.apps import SiteApps
 
-    site = Site.provision(bench, name, [], admin_password, on_progress=on_progress)
+    site = Site.provision(bench, name, [], admin_password, on_progress=on_progress, schedule_backups=False)
     on_progress(f"Restoring backup: {db_file}")
     site.restore(db_file, public_files, private_files)
     # The restored database decides which apps the site runs.
     SiteApps(site).restore_installed_apps_copy()
     bench.processes.apply()
+    site.backups.enable_default_schedule()
     return site
 
 
diff --git a/pilot/core/site/retention.py b/pilot/core/site/retention.py
index e08c8b2ea..09f968ca9 100644
--- a/pilot/core/site/retention.py
+++ b/pilot/core/site/retention.py
@@ -1,6 +1,6 @@
 """Decide which backup runs to prune. Pure logic, no I/O."""
 
-from datetime import datetime
+from datetime import date, datetime
 
 from pilot.config import SCHEME_FIFO, BackupConfig
 
@@ -13,17 +13,37 @@ class BackupRetentionPolicy:
     def __init__(self, config: BackupConfig) -> None:
         self.config = config
 
-    def select_deletions(self, timestamps: list[str]) -> list[str]:
+    def select_deletions(self, timestamps: list[str], today: date | None = None) -> list[str]:
         runs = self._parse(timestamps)
         if len(runs) <= 1:
             return []
-        keep = self._keep_fifo(runs) if self.config.scheme == SCHEME_FIFO else self._keep_gfs(runs)
+        if self.config.scheme == SCHEME_FIFO:
+            keep = self._keep_fifo(runs)
+        elif self.config.calendar_rotation:
+            keep = self._keep_calendar_gfs(runs, today or date.today())
+        else:
+            keep = self._keep_gfs(runs)
         keep.add(runs[0][1])  # newest run is never pruned
         return [ts for _, ts in runs if ts not in keep]
 
     def _keep_fifo(self, runs: list[tuple[datetime, str]]) -> set[str]:
         return {ts for _, ts in runs[: self.config.keep_last]}
 
+    def _keep_calendar_gfs(self, runs: list[tuple[datetime, str]], today: date) -> set[str]:
+        """Keep every run on an anchor day while its calendar window is open."""
+        keep = set()
+        for when, timestamp in runs:
+            age = (today - when.date()).days
+            tiers = (
+                (7, True),
+                (28, when.weekday() == 6),
+                (366, when.day == 1),
+                (3653, when.month == 1 and when.day == 1),
+            )
+            if any(anchor and age <= window for window, anchor in tiers):
+                keep.add(timestamp)
+        return keep
+
     def _keep_gfs(self, runs: list[tuple[datetime, str]]) -> set[str]:
         config = self.config
         keep: set[str] = set()
diff --git a/pilot/managers/cron.py b/pilot/managers/cron.py
index 02a0a366b..7c3dedcea 100644
--- a/pilot/managers/cron.py
+++ b/pilot/managers/cron.py
@@ -56,6 +56,8 @@ def set_schedule(self, job_key: str, cron_expr: str, command: str) -> None:
             lines = self._read_crontab()
             marker = self._marker(job_key)
             entry = f"{cron_expr} {command}"
+            if marker in lines and lines[lines.index(marker) + 1 :][:1] == [entry]:
+                return
             try:
                 i = lines.index(marker)
             except ValueError:
diff --git a/pilot/patches/enroll_sites_in_scheduled_backups.py b/pilot/patches/enroll_sites_in_scheduled_backups.py
new file mode 100644
index 000000000..2b653bb65
--- /dev/null
+++ b/pilot/patches/enroll_sites_in_scheduled_backups.py
@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+PATCH_NAME = Path(__file__).stem
+
+
+def run(benches_root: Path) -> None:
+    """Give existing sites the default scheduled backups that new sites get. A site with a
+    retention record has a custom schedule or the default already, and keeps it."""
+    from pilot.config.site_backup import read_retention
+    from pilot.core.bench import Bench
+    from pilot.internal.patch_state import is_applied, mark_applied
+
+    for bench_dir in sorted(benches_root.glob("*")):
+        if not bench_dir.is_dir() or not (bench_dir / "bench.toml").exists():
+            continue
+        if is_applied(bench_dir, PATCH_NAME):
+            continue
+        bench = Bench(bench_dir)
+        for site in bench.sites():
+            if read_retention(site.path / "site_config.json") is not None:
+                continue
+            site.backups.enable_default_schedule()
+            print(f"Site '{site.config.name}' now has scheduled backups.")
+        mark_applied(bench_dir, PATCH_NAME)
+
+
+if __name__ == "__main__":
+    import sys
+
+    sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+    from pilot.utils import benches_dir
+
+    run(benches_dir())
diff --git a/pilot/patches/patches.txt b/pilot/patches/patches.txt
index 52a907aa4..8e4355be0 100644
--- a/pilot/patches/patches.txt
+++ b/pilot/patches/patches.txt
@@ -15,3 +15,4 @@ keep_mariadb_out_of_swap
 check_realtime_sessions_locally
 enable_database_at_boot
 add_mariadb_boot_sizing
+enroll_sites_in_scheduled_backups
diff --git a/pilot/tasks/backup_site.py b/pilot/tasks/backup_site.py
index a288ca479..9cb7168c3 100644
--- a/pilot/tasks/backup_site.py
+++ b/pilot/tasks/backup_site.py
@@ -18,27 +18,42 @@ class BackupSiteTask(Task):
 
     site: str
     with_files: bool = False
+    automatic: bool = False
 
     @cached_property
     def site_record(self) -> Site:
         return self.bench.site(self.site)
 
     def run(self) -> None:
+        if self.automatic:
+            options = self.site_record.backups.get_scheduled_backup(datetime.now(UTC))
+            if options is None:
+                return
+            self.with_files = options["with_files"]
         self.backup()
 
     @step("backup", lambda self: f"Backup site {self.site}")
     def backup(self) -> None:
         try:
             timestamp, backup_files = self.site_record.backups.take(with_files=self.with_files)
-        except BenchError as error:
+        except (BenchError, OSError) as error:
             print(error)
             self.record(status="failed", timestamp="", files={}, offsite=False, pruned=[])
+            self.notify_if_failing()
             sys.exit(getattr(error, "returncode", None) or 1)
         files = {path.name: path.stat().st_size for path in backup_files}
         offsite = self.upload(timestamp, backup_files)
         pruned = self.prune()
         self.record(status="success", timestamp=timestamp, files=files, offsite=offsite, pruned=pruned)
 
+    def notify_if_failing(self) -> None:
+        from pilot.core.notification.events import backups_failing
+        from pilot.core.site.backups import MAX_DAILY_FAILURES
+
+        failures = self.site_record.backups.get_failures_since_last_success(datetime.now(UTC))
+        if failures == MAX_DAILY_FAILURES:
+            backups_failing(self.bench, self.site, failures)
+
     def upload(self, timestamp: str, backup_files: list[Path]) -> bool:
         if not self.bench.config.s3.is_configured:
             return False
diff --git a/tests/admin/backend/api/test_site_backups_view.py b/tests/admin/backend/api/test_site_backups_view.py
index fb448458a..bd13d3f14 100644
--- a/tests/admin/backend/api/test_site_backups_view.py
+++ b/tests/admin/backend/api/test_site_backups_view.py
@@ -281,6 +281,28 @@ def test_backup_schedule_put_rejects_invalid_cron(tmp_path: Path) -> None:
     assert response.status_code == 422
 
 
+def test_a_custom_schedule_replaces_the_default_policy(tmp_path: Path) -> None:
+    from pilot.config import BackupConfig
+    from pilot.config.site_backup import read_retention, write_retention
+
+    bench_root = tmp_path / "benches" / "current"
+    _make_site(bench_root, "site.localhost")
+    site_config = bench_root / "sites" / "site.localhost" / "site_config.json"
+    write_retention(site_config, BackupConfig(keep_weekly=4, keep_yearly=10, calendar_rotation=True))
+    client = _client(bench_root)
+    with (
+        patch("pilot.managers.cron.CronManager.get_schedule", return_value="0 2 * * *"),
+        patch("pilot.managers.cron.CronManager.set_schedule"),
+    ):
+        response = client.put(
+            "/api/v1/sites/site.localhost/backup-schedule",
+            json={"schedule": "0 2 * * *", "retention": {"scheme": "fifo", "keep_last": 3}},
+        )
+
+    assert response.status_code == 200
+    assert read_retention(site_config) == BackupConfig(scheme="fifo", keep_last=3)
+
+
 def test_backup_schedule_delete_returns_no_content(tmp_path: Path) -> None:
     bench_root = tmp_path / "benches" / "current"
     _make_site(bench_root, "site.localhost")
diff --git a/tests/pilot/core/test_audit_log.py b/tests/pilot/core/test_audit_log.py
index c7c54016e..345e8a6f8 100644
--- a/tests/pilot/core/test_audit_log.py
+++ b/tests/pilot/core/test_audit_log.py
@@ -1,5 +1,5 @@
 import stat
-from datetime import UTC
+from datetime import UTC, datetime
 from types import SimpleNamespace
 
 from pilot.core.bench.audit_log import AuditLog
@@ -31,6 +31,19 @@ def test_filter_by_type_status_and_site(tmp_path) -> None:
     assert log.entries(limit=1)[0]["type"] == "activity"
 
 
+def test_since_limits_polling_to_recent_history_and_includes_the_boundary(tmp_path) -> None:
+    log = AuditLog(_bench(tmp_path))
+    for day in (15, 16, 17):
+        log.append("backup", {"site": "a", "logged_at": f"2026-10-{day:02d}T00:00:00+00:00"})
+
+    records = log.entries(entry_type="backup", site="a", since=datetime(2026, 10, 16, tzinfo=UTC))
+
+    assert [record["logged_at"] for record in records] == [
+        "2026-10-17T00:00:00+00:00", "2026-10-16T00:00:00+00:00",
+    ]
+    assert len(log.entries()) == 3
+
+
 def test_filter_by_jti_matches_subject_or_actor(tmp_path) -> None:
     log = AuditLog(_bench(tmp_path))
     log.append("session", {"event": "issued", "jti": "a"})
diff --git a/tests/pilot/core/test_backup_retention.py b/tests/pilot/core/test_backup_retention.py
index 1b34e5299..e57f816fc 100644
--- a/tests/pilot/core/test_backup_retention.py
+++ b/tests/pilot/core/test_backup_retention.py
@@ -1,5 +1,7 @@
 from datetime import date, timedelta
 
+import pytest
+
 from pilot.config import BackupConfig
 from pilot.core.site.retention import BackupRetentionPolicy
 
@@ -72,3 +74,40 @@ def test_gfs_yearly_tier_keeps_one_run_per_year() -> None:
 def test_backup_defaults_are_gfs() -> None:
     config = BackupConfig()
     assert config.scheme == "gfs"
+
+
+@pytest.mark.parametrize(
+    ("today", "timestamp", "kept"),
+    [
+        (date(2026, 10, 18), "20261011_000000", True),
+        (date(2026, 10, 18), "20261010_235959", False),
+        (date(2026, 10, 18), "20260920_060000", True),
+        (date(2026, 10, 19), "20260920_060000", False),
+        (date(2026, 10, 2), "20251001_000000", True),
+        (date(2026, 10, 3), "20251001_000000", False),
+        (date(2026, 1, 1), "20160101_000000", True),
+        (date(2026, 1, 2), "20160101_000000", False),
+    ],
+)
+def test_calendar_gfs_window_boundaries(today: date, timestamp: str, kept: bool) -> None:
+    newest = today.strftime("%Y%m%d_230000")
+    policy = BackupRetentionPolicy(BackupConfig(keep_weekly=4, keep_yearly=10, calendar_rotation=True))
+
+    deleted = policy.select_deletions([timestamp, newest], today=today)
+
+    assert (timestamp not in deleted) is kept
+    assert newest not in deleted
+
+
+def test_calendar_gfs_keeps_all_runs_on_an_anchor_day() -> None:
+    runs = ["20260920_000000", "20260920_060000", "20260920_120000", "20261018_000000"]
+    policy = BackupRetentionPolicy(BackupConfig(keep_weekly=4, keep_yearly=10, calendar_rotation=True))
+
+    assert policy.select_deletions(runs, today=date(2026, 10, 18)) == []
+
+
+def test_calendar_gfs_clears_expired_history_but_keeps_the_newest() -> None:
+    runs = ["20140101_000000", "20150101_000000", "20160101_000000"]
+    policy = BackupRetentionPolicy(BackupConfig(keep_weekly=4, keep_yearly=10, calendar_rotation=True))
+
+    assert set(policy.select_deletions(runs, today=date(2026, 10, 18))) == set(runs[:2])
diff --git a/tests/pilot/core/test_scheduled_backups.py b/tests/pilot/core/test_scheduled_backups.py
new file mode 100644
index 000000000..fe128cbc7
--- /dev/null
+++ b/tests/pilot/core/test_scheduled_backups.py
@@ -0,0 +1,114 @@
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from pilot.config import BackupConfig
+from pilot.config.site_backup import write_retention
+from pilot.core.bench.audit_log import AuditLog
+from pilot.core.site import Site
+from pilot.internal.tasks.store import TaskStore
+from pilot.tasks.backup_site import BackupSiteTask
+from tests.pilot.commands.test_commands import make_bench
+
+_NOW = datetime(2026, 10, 18, 8, tzinfo=UTC)
+
+
+def _site(tmp_path: Path) -> Site:
+    bench = make_bench(tmp_path)
+    bench.create_directories()
+    bench.config.write(bench.path)
+    site = bench.site("test.localhost")
+    site.path.mkdir(parents=True)
+    write_retention(site.path / "site_config.json", BackupConfig(calendar_rotation=True))
+    return site
+
+
+def _record(site: Site, when: datetime, **fields) -> None:
+    AuditLog(site.bench).append(
+        "backup",
+        {
+            "site": site.config.name, "event": "backup", "status": "success",
+            "finished_at": when.isoformat(), "logged_at": when.isoformat(), **fields,
+        },
+    )
+
+
+@pytest.mark.parametrize(
+    ("now", "history", "expected"),
+    [
+        (_NOW, [], {"with_files": True}),
+        (_NOW, [(timedelta(hours=1), {})], None),
+        (_NOW, [(timedelta(hours=22, minutes=55), {})], None),
+        (_NOW, [(timedelta(hours=23), {})], {"with_files": True}),
+        (_NOW.replace(hour=23, minute=30), [(timedelta(hours=23, minutes=1), {})], {"with_files": False}),
+        (_NOW, [(timedelta(minutes=10 * n), {"status": "failed"}) for n in range(5)], {"with_files": True}),
+        (_NOW, [(timedelta(minutes=10 * n), {"status": "failed"}) for n in range(6)], None),
+    ],
+    ids=["first", "recent", "under-23h", "23h", "files-done-today", "5-failures", "6-failures"],
+)
+def test_when_a_scheduled_backup_is_due(tmp_path: Path, now: datetime, history: list, expected: dict | None) -> None:
+    site = _site(tmp_path)
+    for age, fields in reversed(history):
+        _record(site, now - age, **{"with_files": True, **fields})
+
+    assert site.backups.get_scheduled_backup(now) == expected
+
+
+@pytest.mark.parametrize("manual_backup", [False, True])
+def test_polls_do_not_overlap_a_queued_site_backup(tmp_path: Path, manual_backup: bool) -> None:
+    site = _site(tmp_path)
+    with patch("pilot.managers.task.control.TaskWorkerControl.wake"):
+        if manual_backup:
+            BackupSiteTask.queue(
+                site.bench, site=site.config.name, with_files=True, resource_key=f"site:{site.config.name}"
+            )
+        site.bench.backups.queue_scheduled(_NOW)
+        site.bench.backups.queue_scheduled(_NOW + timedelta(hours=1))
+
+    task_dirs = list((site.bench.path / "tasks").glob("*/meta.json"))
+    assert len(task_dirs) == 1
+    metadata = TaskStore(site.bench.path).read_metadata(task_dirs[0].parent.name)
+    assert metadata["command"] == "backup-site"
+    assert metadata["resource_keys"] == [f"site:{site.config.name}"]
+
+
+@pytest.mark.parametrize(("earlier_failures", "notified"), [(4, False), (5, True), (6, False)])
+def test_the_sixth_failure_since_the_last_success_notifies_once(
+    tmp_path: Path, earlier_failures: int, notified: bool
+) -> None:
+    site = _site(tmp_path)
+    now = datetime.now(UTC)
+    _record(site, now - timedelta(hours=10), status="failed")
+    _record(site, now - timedelta(hours=9))
+    for attempt in range(earlier_failures):
+        _record(site, now - timedelta(hours=attempt + 1), status="failed")
+    task = BackupSiteTask(bench=site.bench, bench_root=site.bench.path, site=site.config.name, automatic=True)
+
+    with patch("pilot.core.site.backups.SiteBackups.take", side_effect=OSError("disk full")), pytest.raises(SystemExit):
+        task.backup()
+
+    events = [notification.event for notification in site.bench.notifications.list(50)]
+    assert ("backups_failing" in events) is notified
+
+
+def test_the_bench_poll_reads_history_once_and_queues_only_due_sites(tmp_path: Path) -> None:
+    site = _site(tmp_path)
+    backed_up = site.bench.site("backed-up.localhost")
+    backed_up.path.mkdir(parents=True)
+    write_retention(backed_up.path / "site_config.json", BackupConfig(calendar_rotation=True))
+    unenrolled = site.bench.site("manual.localhost")
+    unenrolled.path.mkdir(parents=True)
+    _record(backed_up, _NOW - timedelta(hours=2), with_files=True)
+
+    with (
+        patch("pilot.managers.task.control.TaskWorkerControl.wake"),
+        patch.object(AuditLog, "entries", wraps=AuditLog(site.bench).entries) as entries,
+    ):
+        site.bench.backups.queue_scheduled(_NOW)
+
+    entries.assert_called_once()
+    task_dirs = list((site.bench.path / "tasks").glob("*/meta.json"))
+    queued = [TaskStore(site.bench.path).read_metadata(path.parent.name)["args"]["site"] for path in task_dirs]
+    assert queued == [site.config.name]
diff --git a/tests/pilot/core/test_site_provisioning.py b/tests/pilot/core/test_site_provisioning.py
index 4c0f7260f..786f98a4f 100644
--- a/tests/pilot/core/test_site_provisioning.py
+++ b/tests/pilot/core/test_site_provisioning.py
@@ -6,6 +6,8 @@
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 from pilot.config import BenchConfig, RoutePolicy, SiteConfig
 from pilot.config.site import email_sender_name
 from pilot.core.bench import Bench
@@ -95,6 +97,7 @@ def create_site(site: Site, db_type: str | None = None) -> None:
         patch.object(provisioner, "obtain_cert") as obtain_cert,
         patch.object(provisioner.bench, "write_common_site_config"),
         patch.object(Site, "enable_scheduler") as enable_scheduler,
+        patch("pilot.core.bench.backups.BenchBackups.install_schedule"),
     ):
         site = provisioner.provision(lambda _: None)
 
@@ -108,6 +111,66 @@ def create_site(site: Site, db_type: str | None = None) -> None:
     obtain_cert.assert_not_called()
 
 
+def test_only_a_successful_new_site_gets_default_backups(tmp_path: Path) -> None:
+    from pilot.config.site_backup import read_retention
+    from pilot.exceptions import BenchError
+
+    provisioner = _provisioner(tmp_path)
+
+    def create_site(site: Site, db_type: str | None = None) -> None:
+        site.path.mkdir(parents=True)
+        (site.path / "site_config.json").write_text('{"db_name": "site1"}')
+
+    with (
+        patch("pilot.core.site.provisioning.validate_new_site", return_value=False),
+        patch.object(Site, "create", create_site),
+        patch.object(Site, "enable_scheduler"),
+        patch.object(provisioner, "install_apps") as install_apps,
+        patch.object(provisioner, "write_pilot_communication_config"),
+        patch.object(provisioner, "build_missing_assets"),
+        patch.object(provisioner, "add_to_hosts"),
+        patch.object(provisioner, "reload_nginx"),
+        patch.object(provisioner.bench, "write_common_site_config"),
+        patch("pilot.core.bench.backups.BenchBackups.install_schedule") as install_schedule,
+    ):
+        site = provisioner.provision(lambda _: None)
+        assert read_retention(site.path / "site_config.json").calendar_rotation is True
+        install_schedule.assert_called_once()
+
+        install_schedule.reset_mock()
+        provisioner.name = "failed.example.com"
+        install_apps.side_effect = BenchError("app installation failed")
+        with pytest.raises(BenchError, match="app installation failed"):
+            provisioner.provision(lambda _: None)
+        install_schedule.assert_not_called()
+
+
+@pytest.mark.parametrize("restore_fails", [False, True])
+def test_restore_only_enables_backups_after_the_database_and_apps_are_ready(restore_fails: bool) -> None:
+    from pilot.core.site.provisioning import provision_from_backup
+    from pilot.exceptions import BenchError
+
+    bench = MagicMock()
+    site = MagicMock()
+    operations = MagicMock()
+    site.restore = operations.restore
+    site.backups.enable_default_schedule = operations.schedule
+    if restore_fails:
+        operations.restore.side_effect = BenchError("restore failed")
+    with (
+        patch.object(Site, "provision", return_value=site) as provision,
+        patch("pilot.core.site.apps.SiteApps.restore_installed_apps_copy", operations.apps),
+    ):
+        if restore_fails:
+            with pytest.raises(BenchError, match="restore failed"):
+                provision_from_backup(bench, "new.example.com", "database.sql.gz", "admin")
+            operations.schedule.assert_not_called()
+        else:
+            provision_from_backup(bench, "new.example.com", "database.sql.gz", "admin")
+            assert [call[0] for call in operations.mock_calls] == ["restore", "apps", "schedule"]
+        assert provision.call_args.kwargs["schedule_backups"] is False
+
+
 def test_a_new_site_calls_pilot_over_loopback_and_gives_the_browser_the_admin_domain(tmp_path: Path) -> None:
     provisioner = _provisioner(tmp_path)
     provisioner.bench.config.admin.domain = "admin.local"
diff --git a/tests/pilot/managers/test_cron.py b/tests/pilot/managers/test_cron.py
index e6219aabd..ae0f90c22 100644
--- a/tests/pilot/managers/test_cron.py
+++ b/tests/pilot/managers/test_cron.py
@@ -169,3 +169,14 @@ def test_move_schedule_rekeys_the_entry_and_keeps_its_schedule() -> None:
 
     written = mock_run.call_args_list[-1].kwargs["input"].splitlines()
     assert written == ["0 */6 * * * someone-elses-job", manager._marker("new"), "0 3 * * * run-new"]
+
+
+def test_set_schedule_does_not_rewrite_an_unchanged_entry() -> None:
+    manager = make_manager()
+    existing = f"{manager._marker('job1')}\n0 3 * * * run-it\n"
+
+    with patch("subprocess.run") as mock_run:
+        mock_run.return_value = _crontab_result(existing)
+        manager.set_schedule("job1", "0 3 * * *", "run-it")
+
+    assert all(call.kwargs.get("input") is None for call in mock_run.call_args_list)
diff --git a/tests/pilot/patches/test_enroll_sites_in_scheduled_backups.py b/tests/pilot/patches/test_enroll_sites_in_scheduled_backups.py
new file mode 100644
index 000000000..22370023b
--- /dev/null
+++ b/tests/pilot/patches/test_enroll_sites_in_scheduled_backups.py
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from pilot.config import BackupConfig
+from pilot.config.site_backup import read_retention, write_retention
+from pilot.patches.enroll_sites_in_scheduled_backups import run
+
+
+def _bench(tmp_path: Path, *sites: str) -> Path:
+    bench = tmp_path / "benches" / "one"
+    (bench / "sites").mkdir(parents=True)
+    for site in sites:
+        (bench / "sites" / site).mkdir(parents=True)
+        (bench / "sites" / site / "site_config.json").write_text(json.dumps({"db_name": site}))
+    (bench / "bench.toml").write_text('[bench]\nname = "one"\npython = "3.14"\n')
+    return bench
+
+
+def test_sites_without_a_policy_are_enrolled_once_and_custom_ones_kept(tmp_path: Path) -> None:
+    bench = _bench(tmp_path, "a.localhost", "custom.localhost")
+    custom = bench / "sites" / "custom.localhost" / "site_config.json"
+    write_retention(custom, BackupConfig(scheme="fifo", keep_last=3))
+
+    with patch("pilot.core.bench.backups.BenchBackups.install_schedule") as install:
+        run(tmp_path / "benches")
+        run(tmp_path / "benches")
+
+    install.assert_called_once()
+    assert read_retention(bench / "sites" / "a.localhost" / "site_config.json").calendar_rotation is True
+    assert read_retention(custom) == BackupConfig(scheme="fifo", keep_last=3)
+
+
+def test_a_failed_cron_install_is_retried_on_the_next_run(tmp_path: Path) -> None:
+    bench = _bench(tmp_path, "a.localhost")
+    site_config = bench / "sites" / "a.localhost" / "site_config.json"
+
+    with (
+        patch("pilot.core.bench.backups.BenchBackups.install_schedule", side_effect=OSError("crontab failed")),
+        pytest.raises(OSError),
+    ):
+        run(tmp_path / "benches")
+    assert read_retention(site_config) is None
+
+    with patch("pilot.core.bench.backups.BenchBackups.install_schedule") as install:
+        run(tmp_path / "benches")
+
+    install.assert_called_once()
+    assert read_retention(site_config).calendar_rotation is True
+
diff --git a/tests/pilot/tasks/test_backup_site_task.py b/tests/pilot/tasks/test_backup_site_task.py
index a6939151a..86ac30cd5 100644
--- a/tests/pilot/tasks/test_backup_site_task.py
+++ b/tests/pilot/tasks/test_backup_site_task.py
@@ -5,6 +5,7 @@
 
 from pilot.config import SiteConfig
 from pilot.core.bench.audit_log import AuditLog
+from pilot.core.bench.backups import BenchBackups
 from pilot.core.site import Site
 from pilot.exceptions import CommandError
 from pilot.tasks.backup_site import BackupSiteTask
@@ -12,6 +13,7 @@
 
 def _task(tmp_path):
     bench = SimpleNamespace(
+        path=tmp_path,
         sites_path=tmp_path / "sites",
         logs_path=tmp_path / "logs",
         frappe_call=["python", "-m", "frappe"],
@@ -19,6 +21,7 @@ def _task(tmp_path):
     )
     bench.site = lambda name: Site(SiteConfig(name=name, apps=[]), bench)
     bench.audit_action = lambda category, fields: AuditLog(bench).append(category, fields)
+    bench.backups = BenchBackups(bench)
     (bench.sites_path / "site1" / "backups").mkdir(parents=True)
     return BackupSiteTask(bench=bench, bench_root=tmp_path, site="site1", with_files=False), bench
 
@@ -55,6 +58,20 @@ def fail(*a, **k):
     assert entries[0]["status"] == "failed"
 
 
+def test_a_disk_write_failure_is_recorded_for_the_scheduler_retry_limit(tmp_path, monkeypatch) -> None:
+    task, bench = _task(tmp_path)
+
+    def fail(*args, **kwargs):
+        raise OSError("No space left on device")
+
+    monkeypatch.setattr("pilot.utils.run_command", fail)
+    with pytest.raises(SystemExit) as failure:
+        task.run()
+
+    assert failure.value.code == 1
+    assert AuditLog(bench).entries()[0]["status"] == "failed"
+
+
 def test_backup_is_written_outside_private_backups(tmp_path, monkeypatch) -> None:
     """Frappe prunes private/backups on its own schedule."""
     task, bench = _task(tmp_path)