diff --git a/src/pobsync_backend/forms.py b/src/pobsync_backend/forms.py
index c57379f..3895084 100644
--- a/src/pobsync_backend/forms.py
+++ b/src/pobsync_backend/forms.py
@@ -133,6 +133,24 @@ class ManualBackupForm(forms.Form):
     )
 
 
+class RetentionApplyForm(forms.Form):
+    kind = forms.ChoiceField(choices=(("scheduled", "Scheduled"), ("manual", "Manual"), ("all", "All")))
+    protect_bases = forms.BooleanField(required=False)
+    max_delete = forms.IntegerField(min_value=0, initial=10)
+    confirm_host = forms.CharField()
+
+    def __init__(self, *args, host_name: str, **kwargs) -> None:
+        self.host_name = host_name
+        super().__init__(*args, **kwargs)
+        self.fields["confirm_host"].help_text = f"Type {host_name} to confirm deletion."
+
+    def clean_confirm_host(self) -> str:
+        value = self.cleaned_data["confirm_host"].strip()
+        if value != self.host_name:
+            raise forms.ValidationError(f"Type {self.host_name} to confirm.")
+        return value
+
+
 class ScheduleConfigForm(forms.ModelForm):
     cron_expr = forms.CharField(
         label="Cron expression",
diff --git a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html
index dac85a7..1b1f9ec 100644
--- a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html
+++ b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html
@@ -59,6 +59,41 @@
     </table>
   </section>
 
+  {% if plan.delete %}
+    <section class="panel">
+      <h2>Apply Retention</h2>
+      <form method="post" action="{% url 'apply_host_retention' host.host %}" class="form-grid">
+        {% csrf_token %}
+        {{ apply_form.non_field_errors }}
+        {{ apply_form.kind.as_hidden }}
+
+        <div class="field">
+          {{ apply_form.max_delete.errors }}
+          <label for="{{ apply_form.max_delete.id_for_label }}">Max delete</label>
+          {{ apply_form.max_delete }}
+          <div class="helptext">Must be at least the number of snapshots shown in Would Delete.</div>
+        </div>
+
+        <div class="field">
+          {{ apply_form.protect_bases.errors }}
+          <label for="{{ apply_form.protect_bases.id_for_label }}">Protect bases</label>
+          {{ apply_form.protect_bases }}
+        </div>
+
+        <div class="field">
+          {{ apply_form.confirm_host.errors }}
+          <label for="{{ apply_form.confirm_host.id_for_label }}">Confirm host</label>
+          {{ apply_form.confirm_host }}
+          <div class="helptext">{{ apply_form.confirm_host.help_text }}</div>
+        </div>
+
+        <div class="actions">
+          <button type="submit">Apply retention</button>
+        </div>
+      </form>
+    </section>
+  {% endif %}
+
   <section class="panel">
     <h2>Keep Reasons</h2>
     <table>
diff --git a/src/pobsync_backend/tests/test_views.py b/src/pobsync_backend/tests/test_views.py
index 31feeb0..d2bd253 100644
--- a/src/pobsync_backend/tests/test_views.py
+++ b/src/pobsync_backend/tests/test_views.py
@@ -5,7 +5,7 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 
 from django.contrib.auth import get_user_model
-from django.test import TestCase
+from django.test import TestCase, override_settings
 from django.urls import reverse
 
 from pobsync.util import write_yaml_atomic
@@ -458,6 +458,75 @@ class ViewTests(TestCase):
         self.assertRedirects(response, reverse("host_detail", args=[host.host]))
         self.assertContains(response, "Retention kind must be scheduled, manual, or all.")
 
+    def test_retention_apply_deletes_planned_snapshot_after_confirmation(self) -> None:
+        self.client.force_login(self.staff_user)
+        with TemporaryDirectory() as tmp:
+            home = Path(tmp) / "home"
+            host = HostConfig.objects.create(
+                host="web-01",
+                address="web-01.example.test",
+                retention_daily=0,
+                retention_weekly=0,
+                retention_monthly=0,
+                retention_yearly=0,
+            )
+            old_dir = Path(tmp) / "backups" / host.host / "scheduled" / "20260518-021500Z__OLDSNAP"
+            new_dir = Path(tmp) / "backups" / host.host / "scheduled" / "20260519-021500Z__NEWSNAP"
+            old_dir.mkdir(parents=True)
+            new_dir.mkdir(parents=True)
+            old_snapshot = self._snapshot(host, old_dir.name)
+            old_snapshot.path = str(old_dir)
+            old_snapshot.save(update_fields=["path"])
+            new_snapshot = self._snapshot(host, new_dir.name)
+            new_snapshot.path = str(new_dir)
+            new_snapshot.save(update_fields=["path"])
+
+            with override_settings(POBSYNC_HOME=str(home)):
+                response = self.client.post(
+                    reverse("apply_host_retention", args=[host.host]),
+                    {
+                        "kind": "scheduled",
+                        "max_delete": "1",
+                        "confirm_host": host.host,
+                    },
+                    follow=True,
+                )
+
+            self.assertFalse(old_dir.exists())
+            self.assertTrue(new_dir.exists())
+
+        self.assertRedirects(response, f"{reverse('host_retention_plan', args=[host.host])}?kind=scheduled")
+        self.assertContains(response, "Retention deleted 1 snapshot(s) for web-01.")
+        self.assertFalse(SnapshotRecord.objects.filter(pk=old_snapshot.pk).exists())
+        self.assertTrue(SnapshotRecord.objects.filter(pk=new_snapshot.pk).exists())
+
+    def test_retention_apply_rejects_bad_confirmation(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        self._snapshot(host, "20260518-021500Z__OLDSNAP")
+
+        response = self.client.post(
+            reverse("apply_host_retention", args=[host.host]),
+            {
+                "kind": "scheduled",
+                "max_delete": "1",
+                "confirm_host": "wrong",
+            },
+            follow=True,
+        )
+
+        self.assertRedirects(response, reverse("host_retention_plan", args=[host.host]))
+        self.assertContains(response, "Retention apply confirmation is invalid.")
+        self.assertEqual(SnapshotRecord.objects.count(), 1)
+
+    def test_retention_apply_requires_post(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+
+        response = self.client.get(reverse("apply_host_retention", args=[host.host]))
+
+        self.assertEqual(response.status_code, 405)
+
     def test_schedule_form_renders_defaults_for_new_schedule(self) -> None:
         self.client.force_login(self.staff_user)
         host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
diff --git a/src/pobsync_backend/views.py b/src/pobsync_backend/views.py
index bde74bb..a838456 100644
--- a/src/pobsync_backend/views.py
+++ b/src/pobsync_backend/views.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+from pathlib import Path
 
 from django.contrib import messages
 from django.contrib.admin.views.decorators import staff_member_required
@@ -9,12 +10,19 @@ from django.db.models import Count
 from django.shortcuts import get_object_or_404, redirect, render
 from django.views.decorators.http import require_POST
 
-from pobsync.errors import ConfigError
+from pobsync.errors import PobsyncError
 
 from .backup_runner import queue_backup_run
-from .forms import CreateHostConfigForm, GlobalConfigForm, HostConfigForm, ManualBackupForm, ScheduleConfigForm
+from .forms import (
+    CreateHostConfigForm,
+    GlobalConfigForm,
+    HostConfigForm,
+    ManualBackupForm,
+    RetentionApplyForm,
+    ScheduleConfigForm,
+)
 from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord
-from .retention import run_sql_retention_plan
+from .retention import run_sql_retention_apply, run_sql_retention_plan
 from .snapshot_discovery import discover_snapshots, inspect_snapshot_discovery
 
 
@@ -196,7 +204,7 @@ def host_retention_plan(request, host: str):
     protect_bases = request.GET.get("protect_bases") in {"1", "true", "on", "yes"}
     try:
         plan = run_sql_retention_plan(host=host_config.host, kind=kind, protect_bases=protect_bases)
-    except ConfigError as exc:
+    except PobsyncError as exc:
         messages.error(request, str(exc))
         return redirect("host_detail", host=host_config.host)
     context = {
@@ -204,10 +212,51 @@ def host_retention_plan(request, host: str):
         "kind": kind,
         "protect_bases": protect_bases,
         "plan": plan,
+        "apply_form": RetentionApplyForm(
+            host_name=host_config.host,
+            initial={
+                "kind": kind,
+                "protect_bases": protect_bases,
+                "max_delete": len(plan["delete"]),
+            },
+        ),
     }
     return render(request, "pobsync_backend/retention_plan.html", context)
 
 
+@staff_member_required
+@require_POST
+def apply_host_retention(request, host: str):
+    host_config = get_object_or_404(HostConfig, host=host)
+    form = RetentionApplyForm(request.POST, host_name=host_config.host)
+    if not form.is_valid():
+        messages.error(request, "Retention apply confirmation is invalid.")
+        return redirect("host_retention_plan", host=host_config.host)
+
+    kind = form.cleaned_data["kind"]
+    protect_bases = bool(form.cleaned_data["protect_bases"])
+    try:
+        result = run_sql_retention_apply(
+            prefix=Path(settings.POBSYNC_HOME),
+            host=host_config.host,
+            kind=kind,
+            protect_bases=protect_bases,
+            yes=True,
+            max_delete=form.cleaned_data["max_delete"],
+        )
+    except PobsyncError as exc:
+        messages.error(request, str(exc))
+    else:
+        messages.success(request, f"Retention deleted {len(result['deleted'])} snapshot(s) for {host_config.host}.")
+
+    target = redirect("host_retention_plan", host=host_config.host)
+    query = f"kind={kind}"
+    if protect_bases:
+        query += "&protect_bases=1"
+    target["Location"] = f"{target['Location']}?{query}"
+    return target
+
+
 @staff_member_required
 def edit_host_config(request, host: str):
     host_config = get_object_or_404(HostConfig, host=host)
diff --git a/src/pobsync_server/urls.py b/src/pobsync_server/urls.py
index 31a6199..6088e15 100644
--- a/src/pobsync_server/urls.py
+++ b/src/pobsync_server/urls.py
@@ -14,6 +14,7 @@ urlpatterns = [
     path("hosts/<str:host>/config/", views.edit_host_config, name="edit_host_config"),
     path("hosts/<str:host>/queue-backup/", views.queue_manual_backup, name="queue_manual_backup"),
     path("hosts/<str:host>/discover-snapshots/", views.discover_host_snapshots, name="discover_host_snapshots"),
+    path("hosts/<str:host>/retention-apply/", views.apply_host_retention, name="apply_host_retention"),
     path("hosts/<str:host>/retention-plan/", views.host_retention_plan, name="host_retention_plan"),
     path("hosts/<str:host>/schedule/", views.edit_host_schedule, name="edit_host_schedule"),
     path("runs/<int:run_id>/", views.run_detail, name="run_detail"),