2026-05-21 03:56:25 +02:00
31 changed files with 1316 additions and 37 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -0,0 +1,37 @@
+# Changelog
+
+## 1.0.0 - 2026-05-21
+
+Initial stable release of the Django-first pobsync control panel.
+
+### Added
+
+- Django control panel for hosts, global settings, schedules, SSH credentials, snapshots, runs, self-checks, and logs.
+- Native systemd installer and updater for production backup servers.
+- SQLite by default, with optional MariaDB support.
+- Scheduler and worker services for queued manual backups and scheduled backups.
+- Manual backup, dry-run, cancellation, verbose rsync logging, and run detail views.
+- Snapshot discovery for existing backup directories and SQL-backed snapshot records.
+- SQL retention planning and apply flow with base snapshot protection and incomplete snapshot visibility.
+- Explicit cleanup flow for incomplete snapshots, separate from normal retention pruning.
+- Purged snapshot audit overview with reason, action source, operator, host, kind, path, and timestamp.
+- Dashboard and host pages with backup health, latest run/snapshot, next run, and storage/stat summaries.
+- Review resolution for failed/warning runs and incomplete snapshot tasks so operational warnings can be acknowledged.
+- Worker heartbeat metadata and stale running-run reconciliation for queued backup workers.
+- SSH key generation, upload, edit, guarded delete, known_hosts management, and per-host key selection.
+- In-app changelog page sourced from this changelog.
+- Restore guidance on snapshot detail pages.
+
+### Changed
+
+- Django and the database are now the source of truth for configuration.
+- Docker Compose is documented as development and disposable test tooling rather than the primary production path.
+- The `pobsync` console entrypoint is now a maintainer layer around Django management commands.
+- Scheduled pruning is evaluated by the pobsync scheduler service and recorded through Django, not host cron.
+- Retention and incomplete cleanup now preserve audit history even after source snapshot records are removed.
+
+### Removed
+
+- Legacy YAML config import/export workflow.
+- Public short aliases for configuration commands.
+- Obsolete global config storage fields.
--- a/2
+++ b/2
@@ -10,7 +10,7 @@ RUN apt-get update \

 WORKDIR /app

-COPY pyproject.toml README.md ./
+COPY pyproject.toml README.md CHANGELOG.md ./
 COPY src ./src
 COPY manage.py ./
 COPY scripts/docker-entrypoint ./scripts/docker-entrypoint
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

 [project]
 name = "pobsync"
-version = "0.1.0"
+version = "1.0.0"
 description = "Pull-based rsync backup tool with hardlinked snapshots"
 requires-python = ">=3.11"
 dependencies = [
--- a/src/pobsync/init.py
+++ b/src/pobsync/init.py
@@ -1,3 +1,2 @@
 __all__ = ["__version__"]
-__version__ = "0.1.0"
-
+__version__ = "1.0.0"
--- a/src/pobsync/cli.py
+++ b/src/pobsync/cli.py
@@ -6,6 +6,8 @@ from typing import Sequence

 from django.core.management import execute_from_command_line

+from pobsync import __version__
+

 COMMAND_ALIASES = {
    "backup": "run_pobsync_backup",
@@ -34,6 +36,9 @@ Configuration is managed from the Django control panel. Use

 def main(argv: Sequence[str] | None = None) -> int:
    args = list(sys.argv[1:] if argv is None else argv)
+    if args and args[0] in {"--version", "version"}:
+        print(f"pobsync {__version__}")
+        return 0
    if not args or args[0] in {"-h", "--help", "help"}:
        print(_usage())
        return 0
--- a/src/pobsync_backend/admin.py
+++ b/src/pobsync_backend/admin.py
@@ -6,7 +6,7 @@ from django.urls import reverse
 from django.utils.html import format_html
 from django.utils.http import urlencode

-from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord, SshCredential
+from .models import BackupRun, GlobalConfig, HostConfig, PurgedSnapshot, ScheduleConfig, SnapshotRecord, SshCredential


@admin.register(SshCredential)
@@ -173,6 +173,16 @@ class SnapshotRecordAdmin(admin.ModelAdmin):
        return format_html('<a href="{}">{}</a>', url, count)


+@admin.register(PurgedSnapshot)
+class PurgedSnapshotAdmin(admin.ModelAdmin):
+    list_display = ("host_name", "kind", "dirname", "action", "reason", "triggered_by", "purged_at")
+    list_filter = ("action", "kind", "purged_at")
+    search_fields = ("host_name", "dirname", "path", "reason", "triggered_by")
+    list_select_related = ("host",)
+    readonly_fields = ("purged_at",)
+    date_hierarchy = "purged_at"
+
+
@admin.register(ScheduleConfig)
 class ScheduleConfigAdmin(admin.ModelAdmin):
    list_display = ("host", "cron_expr", "enabled", "prune", "last_status", "last_started_at", "updated_at")
--- a/src/pobsync_backend/backup_runner.py
+++ b/src/pobsync_backend/backup_runner.py
@@ -1,6 +1,8 @@
 from __future__ import annotations

-from datetime import timedelta
+import os
+import socket
+from datetime import timedelta, timezone as datetime_timezone
 from pathlib import Path

 from django.db import transaction
@@ -107,6 +109,7 @@ def execute_backup_run(
                protect_bases=bool(prune_protect_bases),
                yes=True,
                max_delete=int(prune_max_delete),
+                action=run.run_type,
                acquire_lock=False,
            )
        except Exception as exc:
@@ -158,10 +161,10 @@ def claim_next_queued_run() -> BackupRun | None:
        return run


-def reconcile_running_runs(*, grace_seconds: int = 300) -> int:
+def reconcile_running_runs(*, grace_seconds: int = 300, stale_worker_seconds: int = 24 * 60 * 60) -> int:
    reconciled = 0
    for run in BackupRun.objects.select_related("host").filter(status=BackupRun.Status.RUNNING).order_by("started_at", "id"):
-        if _reconcile_running_run(run=run, grace_seconds=grace_seconds):
+        if _reconcile_running_run(run=run, grace_seconds=grace_seconds, stale_worker_seconds=stale_worker_seconds):
            reconciled += 1
    return reconciled

@@ -176,7 +179,9 @@ def requested_options(run: BackupRun) -> dict[str, object]:
 def _running_result(*, run: BackupRun, dry_run: bool) -> dict[str, object]:
    result = dict(run.result) if isinstance(run.result, dict) else {}
    execution = {
+        **_worker_execution_details(),
        "started_at": (run.started_at or timezone.now()).isoformat(),
+        "heartbeat_at": timezone.now().isoformat(),
    }
    if dry_run:
        execution["log"] = str(dry_run_log_path(run.host.host, run_id=run.id))
@@ -185,24 +190,56 @@ def _running_result(*, run: BackupRun, dry_run: bool) -> dict[str, object]:


 def _run_cancel_requested(run_id: int) -> bool:
-    return BackupRun.objects.filter(id=run_id, status=BackupRun.Status.CANCELLED).exists()
+    try:
+        run = BackupRun.objects.only("id", "status", "result").get(id=run_id)
+    except BackupRun.DoesNotExist:
+        return True
+    if run.status == BackupRun.Status.CANCELLED:
+        return True
+    if run.status == BackupRun.Status.RUNNING:
+        _refresh_run_heartbeat(run)
+    return False


-def _reconcile_running_run(*, run: BackupRun, grace_seconds: int) -> bool:
+def _reconcile_running_run(*, run: BackupRun, grace_seconds: int, stale_worker_seconds: int) -> bool:
    result = run.result if isinstance(run.result, dict) else {}
    requested = result.get("requested") if isinstance(result.get("requested"), dict) else {}
+    stale_worker = _running_worker_timed_out(run=run, stale_worker_seconds=stale_worker_seconds)
    if not requested.get("dry_run"):
+        if stale_worker:
+            result.update(
+                {
+                    "ok": False,
+                    "host": run.host.host,
+                    "failure": {
+                        "category": "worker",
+                        "message": "The worker heartbeat stopped before the run finished.",
+                        "hint": "Check pobsync-worker.service logs before retrying the backup.",
+                    },
+                }
+            )
+            run.status = BackupRun.Status.FAILED
+            run.ended_at = timezone.now()
+            run.result = result
+            run.save(update_fields=["status", "ended_at", "result"])
+            return True
        return False

    log_path = _execution_log_path(result)
    log_tail = _read_log_tail(log_path) if log_path is not None else []
    terminal_log = _terminal_rsync_log(log_tail)
    timed_out = _running_dry_run_timed_out(run=run, grace_seconds=grace_seconds)
-    if not terminal_log and not timed_out:
+    if not terminal_log and not timed_out and not stale_worker:
        return False

-    exit_code = _exit_code_from_log(log_tail) or (124 if timed_out else 255)
+    exit_code = _exit_code_from_log(log_tail) or (124 if timed_out or stale_worker else 255)
    failure = classify_rsync_failure(exit_code, log_tail)
+    if stale_worker and not terminal_log:
+        failure = {
+            "category": "worker",
+            "message": "The worker heartbeat stopped before the dry-run finished.",
+            "hint": "Check pobsync-worker.service logs before retrying the dry-run.",
+        }
    result.update(
        {
            "ok": False,
@@ -226,6 +263,30 @@ def _reconcile_running_run(*, run: BackupRun, grace_seconds: int) -> bool:
    return True


+def _worker_execution_details() -> dict[str, object]:
+    return {
+        "worker_pid": os.getpid(),
+        "worker_host": socket.gethostname(),
+        "claimed_at": timezone.now().isoformat(),
+    }
+
+
+def _refresh_run_heartbeat(run: BackupRun, *, interval_seconds: int = 30) -> None:
+    result = run.result if isinstance(run.result, dict) else {}
+    execution = result.get("execution") if isinstance(result.get("execution"), dict) else {}
+    heartbeat_at = _parse_iso_datetime(execution.get("heartbeat_at"))
+    if heartbeat_at is not None and timezone.now() < heartbeat_at + timedelta(seconds=interval_seconds):
+        return
+    result["execution"] = {
+        **execution,
+        "worker_pid": os.getpid(),
+        "worker_host": socket.gethostname(),
+        "heartbeat_at": timezone.now().isoformat(),
+    }
+    run.result = result
+    run.save(update_fields=["result"])
+
+
 def _execution_log_path(result: dict[str, object]) -> Path | None:
    execution = result.get("execution") if isinstance(result.get("execution"), dict) else {}
    log = execution.get("log") or result.get("log")
@@ -266,3 +327,28 @@ def _running_dry_run_timed_out(*, run: BackupRun, grace_seconds: int) -> bool:
    if not isinstance(timeout_seconds, int) or timeout_seconds <= 0:
        timeout_seconds = DEFAULT_DRY_RUN_TIMEOUT_SECONDS
    return timezone.now() >= run.started_at + timedelta(seconds=timeout_seconds + grace_seconds)
+
+
+def _running_worker_timed_out(*, run: BackupRun, stale_worker_seconds: int) -> bool:
+    if stale_worker_seconds <= 0:
+        return False
+    result = run.result if isinstance(run.result, dict) else {}
+    execution = result.get("execution") if isinstance(result.get("execution"), dict) else {}
+    heartbeat_at = _parse_iso_datetime(execution.get("heartbeat_at"))
+    if heartbeat_at is None:
+        heartbeat_at = run.started_at
+    if heartbeat_at is None:
+        return False
+    return timezone.now() >= heartbeat_at + timedelta(seconds=stale_worker_seconds)
+
+
+def _parse_iso_datetime(value: object):
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        parsed = timezone.datetime.fromisoformat(value)
+    except ValueError:
+        return None
+    if timezone.is_naive(parsed):
+        return timezone.make_aware(parsed, timezone=datetime_timezone.utc)
+    return parsed
--- a/src/pobsync_backend/forms.py
+++ b/src/pobsync_backend/forms.py
@@ -274,6 +274,36 @@ class RetentionApplyForm(forms.Form):
        return value


+class IncompleteCleanupForm(forms.Form):
+    max_delete = forms.IntegerField(min_value=0, initial=0)
+    confirm_delete_count = forms.IntegerField(min_value=0)
+    confirm_host = forms.CharField()
+
+    def __init__(self, *args, host_name: str, expected_delete_count: int, **kwargs) -> None:
+        self.host_name = host_name
+        self.expected_delete_count = expected_delete_count
+        super().__init__(*args, **kwargs)
+        self.fields["confirm_host"].help_text = f"Type {host_name} to confirm incomplete snapshot cleanup."
+        self.fields["confirm_delete_count"].help_text = (
+            f"Type {expected_delete_count} to confirm the current number of incomplete snapshots."
+        )
+        self.fields["max_delete"].help_text = (
+            f"Must be at least {expected_delete_count} for the incomplete snapshots shown here."
+        )
+
+    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
+
+    def clean_confirm_delete_count(self) -> int:
+        value = self.cleaned_data["confirm_delete_count"]
+        if value != self.expected_delete_count:
+            raise forms.ValidationError(f"Type {self.expected_delete_count} to confirm the incomplete count.")
+        return value
+
+
 class ScheduleConfigForm(forms.ModelForm):
    cron_expr = forms.CharField(
        label="Schedule expression",
--- a/src/pobsync_backend/management/commands/run_pobsync_retention.py
+++ b/src/pobsync_backend/management/commands/run_pobsync_retention.py
@@ -36,6 +36,7 @@ class Command(BaseCommand):
                    protect_bases=bool(options["protect_bases"]),
                    yes=True,
                    max_delete=int(options["max_delete"]),
+                    action="cli",
                )
            else:
                result = run_sql_retention_plan(
--- a/src/pobsync_backend/management/commands/run_pobsync_worker.py
+++ b/src/pobsync_backend/management/commands/run_pobsync_worker.py
@@ -19,6 +19,12 @@ class Command(BaseCommand):
        parser.add_argument("--once", action="store_true", help="Process one queued run and exit")
        parser.add_argument("--loop", action="store_true", help="Keep checking for queued runs")
        parser.add_argument("--interval", type=int, default=15, help="Loop interval in seconds")
+        parser.add_argument(
+            "--stale-running-seconds",
+            type=int,
+            default=24 * 60 * 60,
+            help="Mark running runs failed after this many seconds without a worker heartbeat; use 0 to disable",
+        )

    def handle(self, *args: Any, **options: Any) -> None:
        if not options["once"] and not options["loop"]:
@@ -26,14 +32,14 @@ class Command(BaseCommand):

        paths = PobsyncPaths(home=Path(options["prefix"]))
        while True:
-            count = self._run_once(prefix=paths.home)
+            count = self._run_once(prefix=paths.home, stale_running_seconds=int(options["stale_running_seconds"]))
            self.stdout.write(f"Ran {count} queued backup run(s).")
            if options["once"]:
                return
            time.sleep(max(1, int(options["interval"])))

-    def _run_once(self, *, prefix: Path) -> int:
-        reconciled = reconcile_running_runs()
+    def _run_once(self, *, prefix: Path, stale_running_seconds: int = 24 * 60 * 60) -> int:
+        reconciled = reconcile_running_runs(stale_worker_seconds=stale_running_seconds)
        run = claim_next_queued_run()
        if run is None:
            return reconciled
--- a/src/pobsync_backend/migrations/0012_review_state.py
+++ b/src/pobsync_backend/migrations/0012_review_state.py
@@ -0,0 +1,30 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("pobsync_backend", "0011_remove_globalconfig_data"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="backuprun",
+            name="reviewed_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="backuprun",
+            name="reviewed_by",
+            field=models.CharField(blank=True, max_length=150),
+        ),
+        migrations.AddField(
+            model_name="snapshotrecord",
+            name="reviewed_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="snapshotrecord",
+            name="reviewed_by",
+            field=models.CharField(blank=True, max_length=150),
+        ),
+    ]
--- a/src/pobsync_backend/migrations/0013_purgedsnapshot.py
+++ b/src/pobsync_backend/migrations/0013_purgedsnapshot.py
@@ -0,0 +1,50 @@
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("pobsync_backend", "0012_review_state"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="PurgedSnapshot",
+            fields=[
+                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("host_name", models.CharField(max_length=255)),
+                ("kind", models.CharField(max_length=16)),
+                ("dirname", models.CharField(max_length=255)),
+                ("path", models.CharField(max_length=1024)),
+                ("reason", models.CharField(blank=True, max_length=512)),
+                (
+                    "action",
+                    models.CharField(
+                        choices=[
+                            ("manual", "Manual"),
+                            ("scheduled", "Scheduled"),
+                            ("cli", "CLI"),
+                            ("incomplete_cleanup", "Incomplete cleanup"),
+                        ],
+                        max_length=32,
+                    ),
+                ),
+                ("triggered_by", models.CharField(blank=True, max_length=150)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("purged_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "host",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="purged_snapshots",
+                        to="pobsync_backend.hostconfig",
+                    ),
+                ),
+            ],
+            options={
+                "ordering": ["-purged_at", "host_name", "dirname"],
+            },
+        ),
+    ]
--- a/src/pobsync_backend/models.py
+++ b/src/pobsync_backend/models.py
@@ -124,6 +124,8 @@ class BackupRun(models.Model):
    rsync_exit_code = models.IntegerField(null=True, blank=True)
    result = models.JSONField(default=dict, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
+    reviewed_at = models.DateTimeField(null=True, blank=True)
+    reviewed_by = models.CharField(max_length=150, blank=True)

    class Meta:
        ordering = ["-created_at"]
@@ -158,6 +160,8 @@ class SnapshotRecord(models.Model):
    ended_at = models.DateTimeField(null=True, blank=True)
    metadata = models.JSONField(default=dict, blank=True)
    discovered_at = models.DateTimeField(auto_now_add=True)
+    reviewed_at = models.DateTimeField(null=True, blank=True)
+    reviewed_by = models.CharField(max_length=150, blank=True)

    class Meta:
        constraints = [
@@ -169,6 +173,31 @@ class SnapshotRecord(models.Model):
        return f"{self.host}/{self.kind}/{self.dirname}"


+class PurgedSnapshot(models.Model):
+    class Action(models.TextChoices):
+        MANUAL = "manual", "Manual"
+        SCHEDULED = "scheduled", "Scheduled"
+        CLI = "cli", "CLI"
+        INCOMPLETE_CLEANUP = "incomplete_cleanup", "Incomplete cleanup"
+
+    host = models.ForeignKey(HostConfig, on_delete=models.SET_NULL, null=True, blank=True, related_name="purged_snapshots")
+    host_name = models.CharField(max_length=255)
+    kind = models.CharField(max_length=16)
+    dirname = models.CharField(max_length=255)
+    path = models.CharField(max_length=1024)
+    reason = models.CharField(max_length=512, blank=True)
+    action = models.CharField(max_length=32, choices=Action.choices)
+    triggered_by = models.CharField(max_length=150, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+    purged_at = models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        ordering = ["-purged_at", "host_name", "dirname"]
+
+    def __str__(self) -> str:
+        return f"{self.host_name}/{self.kind}/{self.dirname}"
+
+
 class ScheduleConfig(TimestampedModel):
    host = models.OneToOneField(HostConfig, on_delete=models.CASCADE, related_name="schedule")
    cron_expr = models.CharField(max_length=128)
--- a/src/pobsync_backend/retention.py
+++ b/src/pobsync_backend/retention.py
@@ -12,7 +12,7 @@ from pobsync.paths import PobsyncPaths
 from pobsync.retention import Snapshot, apply_base_protection, build_retention_plan
 from pobsync.util import sanitize_host

-from .models import HostConfig, SnapshotRecord
+from .models import HostConfig, PurgedSnapshot, SnapshotRecord


 def run_sql_retention_plan(*, host: str, kind: str, protect_bases: bool) -> dict[str, Any]:
@@ -65,6 +65,8 @@ def run_sql_retention_apply(
    protect_bases: bool,
    yes: bool,
    max_delete: int,
+    action: str = PurgedSnapshot.Action.MANUAL,
+    triggered_by: str = "",
    acquire_lock: bool = True,
 ) -> dict[str, Any]:
    host = sanitize_host(host)
@@ -101,6 +103,7 @@ def run_sql_retention_apply(
                raise ConfigError(f"Refusing to delete unsupported snapshot kind: {snap_kind!r}")

            path = _snapshot_delete_path(path=Path(snap_path), dirname=dirname)
+            reason = str(item.get("reason") or "outside retention policy")
            if not path.exists():
                actions.append(f"skip missing {snap_kind}/{dirname}")
                continue
@@ -108,9 +111,19 @@ def run_sql_retention_apply(
                raise ConfigError(f"Refusing to delete non-directory path: {path}")

            _remove_snapshot_tree(path)
+            _record_purged_snapshot(
+                host_config=_enabled_host_config(host),
+                kind=snap_kind,
+                dirname=dirname,
+                path=path,
+                reason=reason,
+                action=action,
+                triggered_by=triggered_by,
+                metadata={"source": "retention", "protect_bases": bool(protect_bases), "retention_kind": kind},
+            )
            SnapshotRecord.objects.filter(host__host=host, kind=snap_kind, dirname=dirname).delete()
            actions.append(f"deleted {snap_kind} {dirname}")
-            deleted.append({"dirname": dirname, "kind": snap_kind, "path": str(path)})
+            deleted.append({"dirname": dirname, "kind": snap_kind, "path": str(path), "reason": reason})

        return {
            "ok": True,
@@ -131,6 +144,87 @@ def run_sql_retention_apply(
    return _do_apply()


+def run_incomplete_cleanup(
+    *,
+    prefix: Path,
+    host: str,
+    yes: bool,
+    max_delete: int,
+    triggered_by: str = "",
+    acquire_lock: bool = True,
+) -> dict[str, Any]:
+    host = sanitize_host(host)
+    if not yes:
+        raise ConfigError("Refusing to delete incomplete snapshots without --yes")
+    if max_delete < 0:
+        raise ConfigError("--max-delete must be >= 0")
+
+    paths = PobsyncPaths(home=prefix)
+
+    def _do_cleanup() -> dict[str, Any]:
+        host_config = _enabled_host_config(host)
+        incomplete_list = [
+            _snapshot_to_item(snapshot, reasons=["manual incomplete cleanup"])
+            for snapshot in _incomplete_snapshots_for_host(host_config)
+        ]
+        if max_delete == 0 and len(incomplete_list) > 0:
+            raise ConfigError("Incomplete cleanup blocked by --max-delete=0")
+        if len(incomplete_list) > max_delete:
+            raise ConfigError(
+                f"Refusing to delete {len(incomplete_list)} incomplete snapshots (exceeds --max-delete={max_delete})"
+            )
+
+        actions: list[str] = []
+        deleted: list[dict[str, Any]] = []
+
+        for item in incomplete_list:
+            dirname = item["dirname"]
+            snap_path = Path(item["path"])
+            path = _snapshot_delete_path(path=snap_path, dirname=dirname)
+            _validate_incomplete_delete_path(host=host, path=path, dirname=dirname)
+
+            if not path.exists():
+                actions.append(f"skip missing incomplete/{dirname}")
+            elif not path.is_dir():
+                raise ConfigError(f"Refusing to delete non-directory path: {path}")
+            else:
+                _remove_snapshot_tree(path)
+                actions.append(f"deleted incomplete {dirname}")
+
+            _record_purged_snapshot(
+                host_config=host_config,
+                kind=SnapshotRecord.Kind.INCOMPLETE,
+                dirname=dirname,
+                path=path,
+                reason="manual incomplete cleanup",
+                action=PurgedSnapshot.Action.INCOMPLETE_CLEANUP,
+                triggered_by=triggered_by,
+                metadata={"source": "incomplete_cleanup"},
+            )
+            SnapshotRecord.objects.filter(
+                host__host=host,
+                kind=SnapshotRecord.Kind.INCOMPLETE,
+                dirname=dirname,
+            ).delete()
+            deleted.append({"dirname": dirname, "kind": SnapshotRecord.Kind.INCOMPLETE, "path": str(path)})
+
+        return {
+            "ok": True,
+            "host": host,
+            "kind": SnapshotRecord.Kind.INCOMPLETE,
+            "max_delete": max_delete,
+            "source": "sql",
+            "planned_delete_count": len(incomplete_list),
+            "deleted": deleted,
+            "actions": actions,
+        }
+
+    if acquire_lock:
+        with acquire_host_lock(paths.locks_dir, host, command="incomplete-cleanup"):
+            return _do_cleanup()
+    return _do_cleanup()
+
+
 def _enabled_host_config(host: str) -> HostConfig:
    try:
        return HostConfig.objects.get(host=host, enabled=True)
@@ -212,6 +306,39 @@ def _snapshot_delete_path(*, path: Path, dirname: str) -> Path:
    return path


+def _record_purged_snapshot(
+    *,
+    host_config: HostConfig,
+    kind: str,
+    dirname: str,
+    path: Path,
+    reason: str,
+    action: str,
+    triggered_by: str,
+    metadata: dict[str, Any],
+) -> None:
+    PurgedSnapshot.objects.create(
+        host=host_config,
+        host_name=host_config.host,
+        kind=kind,
+        dirname=dirname,
+        path=str(path),
+        reason=reason,
+        action=action,
+        triggered_by=triggered_by,
+        metadata=metadata,
+    )
+
+
+def _validate_incomplete_delete_path(*, host: str, path: Path, dirname: str) -> None:
+    path_parts = path.parts
+    if path.name != dirname or ".incomplete" not in path_parts or host not in path_parts:
+        raise ConfigError(f"Refusing to delete unexpected incomplete snapshot path: {path}")
+    incomplete_index = path_parts.index(".incomplete")
+    if incomplete_index == 0 or path_parts[incomplete_index - 1] != host:
+        raise ConfigError(f"Refusing to delete incomplete snapshot outside host backup root: {path}")
+
+
 def _remove_snapshot_tree(path: Path) -> None:
    _make_directories_user_writable(path)
    shutil.rmtree(path)
--- a/src/pobsync_backend/stats_summary.py
+++ b/src/pobsync_backend/stats_summary.py
@@ -94,6 +94,7 @@ def _run_summary(run: BackupRun) -> dict[str, Any]:
        "snapshot": run.snapshot,
        "snapshot_path": run.snapshot_path,
        "status": run.status,
+        "reviewed_at": run.reviewed_at,
        "has_stats": bool(stats),
        "duration_seconds": _int_at(stats, "duration_seconds"),
        "rsync": stats.get("rsync") if isinstance(stats.get("rsync"), dict) else {},
@@ -130,7 +131,7 @@ def _is_real_run(run: BackupRun) -> bool:

 def _first_run_with_status(runs: list[dict[str, Any]], statuses: set[str]) -> dict[str, Any]:
    for run in runs:
-        if run["status"] in statuses:
+        if run["status"] in statuses and run.get("reviewed_at") is None:
            return run
    return {}

--- a/src/pobsync_backend/templates/pobsync_backend/base.html
+++ b/src/pobsync_backend/templates/pobsync_backend/base.html
@@ -375,6 +375,8 @@
      <a href="{% url 'ssh_credentials' %}">SSH Keys</a>
      <a href="{% url 'self_check' %}">Self Check</a>
      <a href="{% url 'logs' %}">Logs</a>
+      <a href="{% url 'purged_snapshots' %}">Purged</a>
+      <a href="{% url 'changelog' %}">Changelog</a>
      <a href="/api/status/">Status API</a>
      <span class="spacer"></span>
      <span class="muted">{{ request.user.username }}</span>
--- a/src/pobsync_backend/templates/pobsync_backend/changelog.html
+++ b/src/pobsync_backend/templates/pobsync_backend/changelog.html
@@ -0,0 +1,41 @@
+{% extends "pobsync_backend/base.html" %}
+
+{% block title %}Changelog - pobsync{% endblock %}
+
+{% block content %}
+  <h1>Changelog</h1>
+
+  <section class="actions" aria-label="Changelog actions">
+    <a class="button-link secondary" href="{% url 'dashboard' %}">Back to dashboard</a>
+  </section>
+
+  <section class="panel">
+    <div class="stack spaced">
+      <div><strong>Installed version:</strong> {{ app_version }}</div>
+      <div class="muted">Source: {{ changelog_path }}</div>
+      {% if missing %}
+        <div class="status warning">missing</div>
+      {% endif %}
+    </div>
+
+    <div class="stack">
+      {% for block in changelog_blocks %}
+        {% if block.kind == "heading" %}
+          {% if block.level == 1 %}
+            <h2>{{ block.text }}</h2>
+          {% else %}
+            <h3>{{ block.text }}</h3>
+          {% endif %}
+        {% elif block.kind == "list" %}
+          <ul>
+            {% for item in block.items %}
+              <li>{{ item }}</li>
+            {% endfor %}
+          </ul>
+        {% else %}
+          <p>{{ block.text }}</p>
+        {% endif %}
+      {% endfor %}
+    </div>
+  </section>
+{% endblock %}
--- a/src/pobsync_backend/templates/pobsync_backend/dashboard.html
+++ b/src/pobsync_backend/templates/pobsync_backend/dashboard.html
@@ -68,7 +68,7 @@
        {% endif %}
      </div>
    {% elif counts.hosts %}
-      <p><span class="status ok">ok</span> No queued, running, warning, or failed runs.</p>
+      <p><span class="status ok">ok</span> No queued, running, or unreviewed warning/failed runs.</p>
    {% else %}
      <p class="muted">Add a host to start tracking backup status here.</p>
    {% endif %}
@@ -243,6 +243,10 @@
              {% endif %}
              {% if host.retention_warning.incomplete_count %}
                {{ host.retention_warning.incomplete_count }} incomplete snapshot(s) need review.
+                <form method="post" action="{% url 'resolve_host_incomplete_reviews' host.host %}">
+                  {% csrf_token %}
+                  <button type="submit" class="secondary">Mark reviewed</button>
+                </form>
              {% endif %}
              {% if host.retention_warning.error %}
                {{ host.retention_warning.error }}
--- a/src/pobsync_backend/templates/pobsync_backend/host_detail.html
+++ b/src/pobsync_backend/templates/pobsync_backend/host_detail.html
@@ -84,6 +84,10 @@
            {{ retention_warning.incomplete_count }} incomplete snapshot(s) exist. Retention does not delete incomplete
            snapshots automatically; inspect them before cleanup.
          </div>
+          <form method="post" action="{% url 'resolve_host_incomplete_reviews' host.host %}">
+            {% csrf_token %}
+            <button type="submit" class="secondary">Mark incomplete reviewed</button>
+          </form>
        {% endif %}
        {% if retention_warning.error %}
          <div>{{ retention_warning.error }}</div>
--- a/src/pobsync_backend/templates/pobsync_backend/purged_snapshots.html
+++ b/src/pobsync_backend/templates/pobsync_backend/purged_snapshots.html
@@ -0,0 +1,74 @@
+{% extends "pobsync_backend/base.html" %}
+
+{% block title %}Purged Snapshots | pobsync{% endblock %}
+
+{% block content %}
+  <h1>Purged Snapshots</h1>
+
+  <section class="actions" aria-label="Purged snapshot actions">
+    <a class="button-link secondary" href="{% url 'dashboard' %}">Back to dashboard</a>
+  </section>
+
+  <section class="panel">
+    <h2>Filters</h2>
+    <form method="get" class="form-grid">
+      <div class="field">
+        <label for="host">Host</label>
+        <select id="host" name="host">
+          <option value="">All hosts</option>
+          {% for host in hosts %}
+            <option value="{{ host.host }}" {% if selected_host == host.host %}selected{% endif %}>{{ host.host }}</option>
+          {% endfor %}
+        </select>
+      </div>
+      <div class="field">
+        <label for="action">Action</label>
+        <select id="action" name="action">
+          <option value="">All actions</option>
+          {% for value, label in actions %}
+            <option value="{{ value }}" {% if selected_action == value %}selected{% endif %}>{{ label }}</option>
+          {% endfor %}
+        </select>
+      </div>
+      <div class="actions">
+        <button type="submit">Apply filters</button>
+        <a class="button-link secondary" href="{% url 'purged_snapshots' %}">Clear</a>
+      </div>
+    </form>
+  </section>
+
+  <section class="panel">
+    <h2>Purged Snapshot History</h2>
+    <p class="muted">Showing up to 200 of {{ total_count }} purged snapshot record(s).</p>
+    <table>
+      <thead>
+        <tr>
+          <th>Purged</th>
+          <th>Host</th>
+          <th>Kind</th>
+          <th>Dirname</th>
+          <th>Action</th>
+          <th>Reason</th>
+          <th>Triggered by</th>
+          <th>Path</th>
+        </tr>
+      </thead>
+      <tbody>
+        {% for snapshot in purged_snapshots %}
+          <tr>
+            <td>{{ snapshot.purged_at }}</td>
+            <td>{% if snapshot.host %}<a href="{% url 'host_detail' snapshot.host.host %}">{{ snapshot.host_name }}</a>{% else %}{{ snapshot.host_name }}{% endif %}</td>
+            <td>{{ snapshot.kind }}</td>
+            <td>{{ snapshot.dirname }}</td>
+            <td><span class="status skipped">{{ snapshot.get_action_display }}</span></td>
+            <td>{{ snapshot.reason|default:"" }}</td>
+            <td>{{ snapshot.triggered_by|default:"" }}</td>
+            <td class="muted">{{ snapshot.path }}</td>
+          </tr>
+        {% empty %}
+          <tr><td colspan="8" class="muted">No purged snapshots recorded yet.</td></tr>
+        {% endfor %}
+      </tbody>
+    </table>
+  </section>
+{% endblock %}
--- a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html
+++ b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html
@@ -40,6 +40,10 @@
        {{ plan.incomplete|length }} incomplete snapshot(s) exist for this host. Retention does not delete incomplete
        snapshots automatically because they can indicate an interrupted backup that should be inspected first.
      </p>
+      <p>
+        After inspection, use the dedicated cleanup form below to delete only incomplete snapshot directories and their
+        SQL records. Successful scheduled and manual snapshots are not touched by this cleanup.
+      </p>
    </section>
  {% endif %}

@@ -190,6 +194,37 @@
          {% endfor %}
        </tbody>
      </table>
+
+      <h3>Cleanup Incomplete Snapshots</h3>
+      <form method="post" action="{% url 'cleanup_host_incomplete_snapshots' host.host %}" class="form-grid">
+        {% csrf_token %}
+        {{ incomplete_cleanup_form.non_field_errors }}
+
+        <div class="field">
+          {{ incomplete_cleanup_form.max_delete.errors }}
+          <label for="{{ incomplete_cleanup_form.max_delete.id_for_label }}">Max delete</label>
+          {{ incomplete_cleanup_form.max_delete }}
+          <div class="helptext">{{ incomplete_cleanup_form.max_delete.help_text }}</div>
+        </div>
+
+        <div class="field">
+          {{ incomplete_cleanup_form.confirm_host.errors }}
+          <label for="{{ incomplete_cleanup_form.confirm_host.id_for_label }}">Confirm host</label>
+          {{ incomplete_cleanup_form.confirm_host }}
+          <div class="helptext">{{ incomplete_cleanup_form.confirm_host.help_text }}</div>
+        </div>
+
+        <div class="field">
+          {{ incomplete_cleanup_form.confirm_delete_count.errors }}
+          <label for="{{ incomplete_cleanup_form.confirm_delete_count.id_for_label }}">Confirm incomplete count</label>
+          {{ incomplete_cleanup_form.confirm_delete_count }}
+          <div class="helptext">{{ incomplete_cleanup_form.confirm_delete_count.help_text }}</div>
+        </div>
+
+        <div class="actions">
+          <button type="submit">Delete incomplete snapshots</button>
+        </div>
+      </form>
    </section>
  {% endif %}
 {% endblock %}
--- a/src/pobsync_backend/templates/pobsync_backend/run_detail.html
+++ b/src/pobsync_backend/templates/pobsync_backend/run_detail.html
@@ -13,6 +13,14 @@
        <button type="submit" class="secondary">Cancel run</button>
      </form>
    {% endif %}
+    {% if run.status == "failed" or run.status == "warning" %}
+      {% if not run.reviewed_at %}
+        <form method="post" action="{% url 'resolve_run_review' run.id %}">
+          {% csrf_token %}
+          <button type="submit" class="secondary">Mark reviewed</button>
+        </form>
+      {% endif %}
+    {% endif %}
  </section>

  <section class="grid" aria-label="Run summary">
@@ -33,6 +41,16 @@
    </section>
  {% endif %}

+  {% if run.reviewed_at %}
+    <section class="panel highlight success">
+      <h2>Review</h2>
+      <div class="stack">
+        <div><strong>Reviewed:</strong> {{ run.reviewed_at }}</div>
+        <div><strong>Reviewed by:</strong> {{ run.reviewed_by|default:"unknown" }}</div>
+      </div>
+    </section>
+  {% endif %}
+
  {% if dry_run_summary %}
    <section class="panel highlight {{ dry_run_summary.highlight_class }}">
      <h2>Dry Run Summary</h2>
@@ -79,6 +97,10 @@
        <div><strong>Created:</strong> {{ run.created_at }}</div>
        <div><strong>Started:</strong> {{ run.started_at|default:"" }}</div>
        <div><strong>Ended:</strong> {{ run.ended_at|default:"" }}</div>
+        {% if execution %}
+          <div><strong>Worker:</strong> {{ execution.worker_host|default:"unknown" }}{% if execution.worker_pid %} pid {{ execution.worker_pid }}{% endif %}</div>
+          <div><strong>Worker heartbeat:</strong> {{ execution.heartbeat_at|default:"" }}</div>
+        {% endif %}
      </div>
    </section>

--- a/src/pobsync_backend/templates/pobsync_backend/ssh_credential_form.html
+++ b/src/pobsync_backend/templates/pobsync_backend/ssh_credential_form.html
@@ -45,9 +45,21 @@
  {% if credential %}
    <section class="panel">
      <h2>Delete SSH Key</h2>
+      {% if credential.hosts.exists or credential.global_configs.exists %}
+        <p class="muted">
+          This SSH key is still selected by {{ credential.hosts.count }} host(s) or
+          {{ credential.global_configs.count }} global config(s). Select another key there before deleting it.
+        </p>
+      {% else %}
+        <p class="muted">Type <strong>{{ credential.name }}</strong> to confirm deletion.</p>
+      {% endif %}
      <form method="post" action="{% url 'delete_ssh_credential' credential.id %}">
        {% csrf_token %}
-        <button type="submit" class="danger">Delete SSH key</button>
+        <div class="field">
+          <label for="confirm_name">Confirm key name</label>
+          <input id="confirm_name" name="confirm_name" type="text" {% if credential.hosts.exists or credential.global_configs.exists %}disabled{% endif %}>
+        </div>
+        <button type="submit" class="danger" {% if credential.hosts.exists or credential.global_configs.exists %}disabled{% endif %}>Delete SSH key</button>
      </form>
    </section>
  {% endif %}
--- a/src/pobsync_backend/templates/pobsync_backend/ssh_credentials.html
+++ b/src/pobsync_backend/templates/pobsync_backend/ssh_credentials.html
@@ -23,6 +23,7 @@
          <th>Known hosts</th>
          <th>Hosts</th>
          <th>Updated</th>
+          <th>Actions</th>
        </tr>
      </thead>
      <tbody>
@@ -35,9 +36,10 @@
            <td>{{ credential.known_hosts|yesno:"yes,no" }}</td>
            <td>{{ credential.hosts.count }}</td>
            <td>{{ credential.updated_at }}</td>
+            <td><a class="button-link secondary" href="{% url 'edit_ssh_credential' credential.id %}">Edit</a></td>
          </tr>
        {% empty %}
-          <tr><td colspan="7" class="muted">No SSH credentials configured yet.</td></tr>
+          <tr><td colspan="8" class="muted">No SSH credentials configured yet.</td></tr>
        {% endfor %}
      </tbody>
    </table>
--- a/src/pobsync_backend/tests/test_backup_worker.py
+++ b/src/pobsync_backend/tests/test_backup_worker.py
@@ -61,6 +61,9 @@ class BackupWorkerTests(TestCase):
                def fake_run_scheduled(**kwargs):
                    run.refresh_from_db()
                    self.assertIn("execution", run.result)
+                    self.assertIn("worker_pid", run.result["execution"])
+                    self.assertIn("worker_host", run.result["execution"])
+                    self.assertIn("heartbeat_at", run.result["execution"])
                    return {
                        "ok": True,
                        "dry_run": False,
@@ -82,6 +85,57 @@ class BackupWorkerTests(TestCase):
        self.assertEqual(SnapshotRecord.objects.count(), 1)
        self.assertEqual(run.snapshot, SnapshotRecord.objects.get())

+    def test_worker_refreshes_heartbeat_while_run_is_active(self) -> None:
+        with TemporaryDirectory() as tmp:
+            GlobalConfig.objects.create(name="default", backup_root=str(Path(tmp) / "backups"))
+            host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+            run = queue_backup_run(host=host)
+
+            with patch("pobsync_backend.backup_runner.run_scheduled") as run_scheduled:
+                def fake_run_scheduled(**kwargs):
+                    run.refresh_from_db()
+                    old_heartbeat = timezone.now() - timedelta(seconds=120)
+                    run.result["execution"]["heartbeat_at"] = old_heartbeat.isoformat()
+                    run.save(update_fields=["result"])
+
+                    self.assertFalse(kwargs["cancel_check"]())
+                    run.refresh_from_db()
+                    self.assertGreater(
+                        timezone.datetime.fromisoformat(run.result["execution"]["heartbeat_at"]),
+                        old_heartbeat,
+                    )
+                    return {
+                        "ok": True,
+                        "dry_run": False,
+                        "host": host.host,
+                        "snapshot": "",
+                        "base": None,
+                        "rsync": {"exit_code": 0},
+                    }
+
+                run_scheduled.side_effect = fake_run_scheduled
+                Command()._run_once(prefix=Path(tmp) / "home")
+
+    def test_worker_reconciles_stale_real_run_after_heartbeat_timeout(self) -> None:
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        run = queue_backup_run(host=host)
+        run.status = BackupRun.Status.RUNNING
+        run.started_at = timezone.now() - timedelta(seconds=120)
+        run.result["execution"] = {
+            "worker_pid": 123,
+            "worker_host": "backup",
+            "heartbeat_at": (timezone.now() - timedelta(seconds=90)).isoformat(),
+        }
+        run.save(update_fields=["status", "started_at", "result"])
+
+        reconciled = reconcile_running_runs(stale_worker_seconds=30)
+
+        self.assertEqual(reconciled, 1)
+        run.refresh_from_db()
+        self.assertEqual(run.status, BackupRun.Status.FAILED)
+        self.assertEqual(run.result["failure"]["category"], "worker")
+        self.assertIn("heartbeat stopped", run.result["failure"]["message"])
+
    def test_worker_records_dry_run_log_path_while_running(self) -> None:
        with TemporaryDirectory() as tmp:
            GlobalConfig.objects.create(name="default", backup_root=str(Path(tmp) / "backups"))
--- a/src/pobsync_backend/tests/test_console_entrypoint.py
+++ b/src/pobsync_backend/tests/test_console_entrypoint.py
@@ -9,6 +9,14 @@ from pobsync.cli import main


 class ConsoleEntrypointTests(SimpleTestCase):
+    def test_version_prints_package_version(self) -> None:
+        stdout = StringIO()
+        with patch("sys.stdout", stdout):
+            exit_code = main(["--version"])
+
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(stdout.getvalue().strip(), "pobsync 1.0.0")
+
    def test_maps_backup_alias_to_django_command(self) -> None:
        with patch("pobsync.cli.execute_from_command_line") as execute:
            exit_code = main(["backup", "web-01", "--dry-run"])
--- a/src/pobsync_backend/tests/test_run_backup_records_snapshot.py
+++ b/src/pobsync_backend/tests/test_run_backup_records_snapshot.py
@@ -96,6 +96,7 @@ class RunBackupRecordsSnapshotTests(TestCase):
            protect_bases=True,
            yes=True,
            max_delete=3,
+            action=BackupRun.RunType.SCHEDULED,
            acquire_lock=False,
        )
        run = BackupRun.objects.get()
--- a/src/pobsync_backend/tests/test_sql_retention.py
+++ b/src/pobsync_backend/tests/test_sql_retention.py
@@ -11,8 +11,8 @@ from django.core.management import call_command
 from django.test import TestCase

 from pobsync.errors import ConfigError
-from pobsync_backend.models import HostConfig, SnapshotRecord
-from pobsync_backend.retention import run_sql_retention_apply, run_sql_retention_plan
+from pobsync_backend.models import HostConfig, PurgedSnapshot, SnapshotRecord
+from pobsync_backend.retention import run_incomplete_cleanup, run_sql_retention_apply, run_sql_retention_plan


 class SqlRetentionTests(TestCase):
@@ -87,10 +87,26 @@ class SqlRetentionTests(TestCase):
            self.assertTrue(new_dir.exists())
            self.assertTrue(SnapshotRecord.objects.filter(pk=new.pk).exists())
            self.assertFalse(SnapshotRecord.objects.filter(pk=old.pk).exists())
-            self.assertEqual(result["deleted"], [{"dirname": old.dirname, "kind": "scheduled", "path": str(old_dir)}])
+            self.assertEqual(
+                result["deleted"],
+                [
+                    {
+                        "dirname": old.dirname,
+                        "kind": "scheduled",
+                        "path": str(old_dir),
+                        "reason": "outside retention policy",
+                    }
+                ],
+            )
            self.assertEqual(result["planned_delete_count"], 1)
            self.assertEqual(result["max_delete"], 1)
            self.assertEqual(result["incomplete_ignored_count"], 0)
+            purged = PurgedSnapshot.objects.get(dirname=old.dirname)
+            self.assertEqual(purged.host_name, host.host)
+            self.assertEqual(purged.kind, "scheduled")
+            self.assertEqual(purged.path, str(old_dir))
+            self.assertEqual(purged.reason, "outside retention policy")
+            self.assertEqual(purged.action, PurgedSnapshot.Action.MANUAL)

    def test_apply_deletes_snapshot_with_readonly_data_directory(self) -> None:
        with TemporaryDirectory() as tmp:
@@ -126,7 +142,17 @@ class SqlRetentionTests(TestCase):

            self.assertFalse(old_dir.exists())
            self.assertFalse(SnapshotRecord.objects.filter(pk=old.pk).exists())
-            self.assertEqual(result["deleted"], [{"dirname": old.dirname, "kind": "scheduled", "path": str(old_dir)}])
+            self.assertEqual(
+                result["deleted"],
+                [
+                    {
+                        "dirname": old.dirname,
+                        "kind": "scheduled",
+                        "path": str(old_dir),
+                        "reason": "outside retention policy",
+                    }
+                ],
+            )

    def test_apply_respects_max_delete(self) -> None:
        host = HostConfig.objects.create(
@@ -152,6 +178,81 @@ class SqlRetentionTests(TestCase):
                acquire_lock=False,
            )

+    def test_incomplete_cleanup_deletes_directory_and_record(self) -> None:
+        with TemporaryDirectory() as tmp:
+            prefix = Path(tmp) / "home"
+            host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+            incomplete_dir = Path(tmp) / "backups" / host.host / ".incomplete" / "20260519-031500Z__BROKEN01"
+            incomplete_dir.mkdir(parents=True)
+            incomplete_dir.joinpath("partial-file").write_text("interrupted\n")
+            record = SnapshotRecord.objects.create(
+                host=host,
+                kind=SnapshotRecord.Kind.INCOMPLETE,
+                dirname=incomplete_dir.name,
+                path=str(incomplete_dir),
+                status="failed",
+                started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
+            )
+
+            result = run_incomplete_cleanup(
+                prefix=prefix,
+                host=host.host,
+                yes=True,
+                max_delete=1,
+                acquire_lock=False,
+            )
+
+            self.assertFalse(incomplete_dir.exists())
+            self.assertFalse(SnapshotRecord.objects.filter(pk=record.pk).exists())
+            self.assertEqual(
+                result["deleted"],
+                [{"dirname": incomplete_dir.name, "kind": SnapshotRecord.Kind.INCOMPLETE, "path": str(incomplete_dir)}],
+            )
+            self.assertEqual(result["planned_delete_count"], 1)
+            purged = PurgedSnapshot.objects.get(dirname=incomplete_dir.name)
+            self.assertEqual(purged.action, PurgedSnapshot.Action.INCOMPLETE_CLEANUP)
+            self.assertEqual(purged.reason, "manual incomplete cleanup")
+
+    def test_incomplete_cleanup_respects_max_delete(self) -> None:
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        SnapshotRecord.objects.create(
+            host=host,
+            kind=SnapshotRecord.Kind.INCOMPLETE,
+            dirname="20260519-031500Z__BROKEN01",
+            path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
+            status="failed",
+            started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
+        )
+
+        with self.assertRaisesRegex(ConfigError, "blocked by --max-delete=0"):
+            run_incomplete_cleanup(
+                prefix=Path("/tmp/pobsync-test"),
+                host=host.host,
+                yes=True,
+                max_delete=0,
+                acquire_lock=False,
+            )
+
+    def test_incomplete_cleanup_rejects_unexpected_path(self) -> None:
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        SnapshotRecord.objects.create(
+            host=host,
+            kind=SnapshotRecord.Kind.INCOMPLETE,
+            dirname="20260519-031500Z__BROKEN01",
+            path=f"/backups/{host.host}/scheduled/20260519-031500Z__BROKEN01",
+            status="failed",
+            started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
+        )
+
+        with self.assertRaisesRegex(ConfigError, "unexpected incomplete snapshot path"):
+            run_incomplete_cleanup(
+                prefix=Path("/tmp/pobsync-test"),
+                host=host.host,
+                yes=True,
+                max_delete=1,
+                acquire_lock=False,
+            )
+
    def test_management_command_plans_from_sql(self) -> None:
        host = HostConfig.objects.create(
            host="web-01",
--- a/src/pobsync_backend/tests/test_views.py
+++ b/src/pobsync_backend/tests/test_views.py
@@ -12,7 +12,15 @@ from django.test import TestCase, override_settings
 from django.urls import reverse

 from pobsync.util import write_yaml_atomic
-from pobsync_backend.models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord, SshCredential
+from pobsync_backend.models import (
+    BackupRun,
+    GlobalConfig,
+    HostConfig,
+    PurgedSnapshot,
+    ScheduleConfig,
+    SnapshotRecord,
+    SshCredential,
+)


 class ViewTests(TestCase):
@@ -31,6 +39,30 @@ class ViewTests(TestCase):
        self.assertEqual(response.status_code, 302)
        self.assertIn("/admin/login/", response["Location"])

+    def test_changelog_requires_staff_login(self) -> None:
+        response = self.client.get(reverse("changelog"))
+
+        self.assertEqual(response.status_code, 302)
+        self.assertIn("/admin/login/", response["Location"])
+
+    def test_changelog_renders_repository_changelog(self) -> None:
+        self.client.force_login(self.staff_user)
+        with TemporaryDirectory() as tmp:
+            changelog = Path(tmp) / "CHANGELOG.md"
+            changelog.write_text(
+                "# Changelog\n\n## 1.0.0 - 2026-05-21\n\n- Django control panel\n- Native systemd installer\n",
+                encoding="utf-8",
+            )
+
+            with override_settings(BASE_DIR=Path(tmp)):
+                response = self.client.get(reverse("changelog"))
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Installed version:")
+        self.assertContains(response, "1.0.0 - 2026-05-21")
+        self.assertContains(response, "Django control panel")
+        self.assertContains(response, "Native systemd installer")
+
    def test_dashboard_renders_hosts_and_latest_runs(self) -> None:
        self.client.force_login(self.staff_user)
        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
@@ -162,7 +194,7 @@ class ViewTests(TestCase):

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Operational Status")
-        self.assertContains(response, "No queued, running, warning, or failed runs.")
+        self.assertContains(response, "No queued, running, or unreviewed warning/failed runs.")

    def test_dashboard_surfaces_retention_warnings(self) -> None:
        self.client.force_login(self.staff_user)
@@ -192,6 +224,30 @@ class ViewTests(TestCase):
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Scheduled prune would delete 2 snapshot(s), above max 1.")
        self.assertContains(response, "1 incomplete snapshot(s) need review.")
+        self.assertContains(response, "Mark reviewed")
+
+    def test_dashboard_ignores_reviewed_problem_runs(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        BackupRun.objects.create(
+            host=host,
+            status=BackupRun.Status.FAILED,
+            reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
+            reviewed_by="admin",
+        )
+        BackupRun.objects.create(
+            host=host,
+            status=BackupRun.Status.WARNING,
+            reviewed_at=datetime(2026, 5, 19, 4, 20, tzinfo=timezone.utc),
+            reviewed_by="admin",
+        )
+
+        response = self.client.get(reverse("dashboard"))
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "No queued, running, or unreviewed warning/failed runs.")
+        self.assertNotContains(response, "failed 1")
+        self.assertNotContains(response, "warning 1")

    def test_dashboard_links_latest_snapshot_for_each_host(self) -> None:
        self.client.force_login(self.staff_user)
@@ -280,6 +336,61 @@ class ViewTests(TestCase):
        self.assertIn("--since", command)
        self.assertIn("6 hours ago", command)

+    def test_purged_snapshots_view_renders_history(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        PurgedSnapshot.objects.create(
+            host=host,
+            host_name=host.host,
+            kind=SnapshotRecord.Kind.SCHEDULED,
+            dirname="20260518-021500Z__OLDSNAP",
+            path=f"/backups/{host.host}/scheduled/20260518-021500Z__OLDSNAP",
+            reason="outside retention policy",
+            action=PurgedSnapshot.Action.SCHEDULED,
+            triggered_by="pobsync-scheduler",
+        )
+
+        response = self.client.get(reverse("purged_snapshots"))
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Purged Snapshots")
+        self.assertContains(response, "20260518-021500Z__OLDSNAP")
+        self.assertContains(response, "outside retention policy")
+        self.assertContains(response, "Scheduled")
+        self.assertContains(response, "pobsync-scheduler")
+
+    def test_purged_snapshots_view_filters_by_host_and_action(self) -> None:
+        self.client.force_login(self.staff_user)
+        web = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        db = HostConfig.objects.create(host="db-01", address="db-01.example.test")
+        PurgedSnapshot.objects.create(
+            host=web,
+            host_name=web.host,
+            kind=SnapshotRecord.Kind.SCHEDULED,
+            dirname="20260518-021500Z__WEBOLD",
+            path=f"/backups/{web.host}/scheduled/20260518-021500Z__WEBOLD",
+            reason="outside retention policy",
+            action=PurgedSnapshot.Action.MANUAL,
+        )
+        PurgedSnapshot.objects.create(
+            host=db,
+            host_name=db.host,
+            kind=SnapshotRecord.Kind.INCOMPLETE,
+            dirname="20260518-021500Z__DBBROKEN",
+            path=f"/backups/{db.host}/.incomplete/20260518-021500Z__DBBROKEN",
+            reason="manual incomplete cleanup",
+            action=PurgedSnapshot.Action.INCOMPLETE_CLEANUP,
+        )
+
+        response = self.client.get(
+            reverse("purged_snapshots"),
+            {"host": db.host, "action": PurgedSnapshot.Action.INCOMPLETE_CLEANUP},
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "20260518-021500Z__DBBROKEN")
+        self.assertNotContains(response, "20260518-021500Z__WEBOLD")
+
    def test_ssh_credentials_view_creates_key(self) -> None:
        self.client.force_login(self.staff_user)

@@ -358,13 +469,46 @@ class ViewTests(TestCase):
            generate_ssh_key(credential)
            key_path = Path(credential.key_path)

-            response = self.client.post(reverse("delete_ssh_credential", args=[credential.id]), follow=True)
+            response = self.client.post(
+                reverse("delete_ssh_credential", args=[credential.id]),
+                {"confirm_name": credential.name},
+                follow=True,
+            )

            self.assertRedirects(response, reverse("ssh_credentials"))
            self.assertContains(response, "SSH key deleted: generated-key.")
            self.assertFalse(SshCredential.objects.exists())
            self.assertFalse(key_path.exists())

+    def test_ssh_credentials_view_requires_delete_confirmation(self) -> None:
+        self.client.force_login(self.staff_user)
+        credential = SshCredential.objects.create(name="backup-key")
+
+        response = self.client.post(
+            reverse("delete_ssh_credential", args=[credential.id]),
+            {"confirm_name": "wrong"},
+            follow=True,
+        )
+
+        self.assertRedirects(response, reverse("edit_ssh_credential", args=[credential.id]))
+        self.assertContains(response, "Type backup-key to confirm SSH key deletion.")
+        self.assertTrue(SshCredential.objects.filter(pk=credential.pk).exists())
+
+    def test_ssh_credentials_view_blocks_delete_when_key_is_in_use(self) -> None:
+        self.client.force_login(self.staff_user)
+        credential = SshCredential.objects.create(name="backup-key")
+        HostConfig.objects.create(host="web-01", address="web-01.example.test", ssh_credential=credential)
+
+        response = self.client.post(
+            reverse("delete_ssh_credential", args=[credential.id]),
+            {"confirm_name": credential.name},
+            follow=True,
+        )
+
+        self.assertRedirects(response, reverse("edit_ssh_credential", args=[credential.id]))
+        self.assertContains(response, "SSH key backup-key is still in use and cannot be deleted.")
+        self.assertTrue(SshCredential.objects.filter(pk=credential.pk).exists())
+
    def test_ssh_credentials_view_rejects_invalid_key(self) -> None:
        self.client.force_login(self.staff_user)

@@ -428,7 +572,7 @@ class ViewTests(TestCase):
            response = self.client.post(
                reverse("edit_ssh_credential", args=[credential.id]),
                {
-                    "name": "backup-key",
+                    "name": "renamed-backup-key",
                    "private_key": "UPDATED KEY",
                    "public_key": "",
                    "known_hosts": "",
@@ -439,6 +583,7 @@ class ViewTests(TestCase):

        self.assertRedirects(response, reverse("ssh_credentials"))
        credential.refresh_from_db()
+        self.assertEqual(credential.name, "renamed-backup-key")
        self.assertEqual(credential.private_key, "UPDATED KEY\n")
        self.assertEqual(credential.public_key, "UPDATED PUBLIC KEY")
        self.assertEqual(credential.notes, "rotated")
@@ -1286,6 +1431,34 @@ class ViewTests(TestCase):
        self.assertContains(response, "Incomplete ignored")
        self.assertContains(response, "deleted scheduled 20260518-021500Z__OLD")

+    def test_run_review_action_marks_problem_run_reviewed(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        run = BackupRun.objects.create(host=host, status=BackupRun.Status.FAILED, result={"ok": False})
+
+        response = self.client.post(reverse("resolve_run_review", args=[run.id]), follow=True)
+
+        run.refresh_from_db()
+        self.assertIsNotNone(run.reviewed_at)
+        self.assertEqual(run.reviewed_by, self.staff_user.username)
+        self.assertRedirects(response, reverse("run_detail", args=[run.id]))
+        self.assertContains(response, f"Run {run.id} marked reviewed.")
+        self.assertContains(response, "Review")
+        self.assertContains(response, self.staff_user.username)
+        self.assertNotContains(response, "Mark reviewed")
+
+    def test_run_review_action_ignores_successful_run(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        run = BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, result={"ok": True})
+
+        response = self.client.post(reverse("resolve_run_review", args=[run.id]), follow=True)
+
+        run.refresh_from_db()
+        self.assertIsNone(run.reviewed_at)
+        self.assertRedirects(response, reverse("run_detail", args=[run.id]))
+        self.assertContains(response, f"Run {run.id} does not need review.")
+
    def test_run_detail_surfaces_host_retention_warnings(self) -> None:
        self.client.force_login(self.staff_user)
        host = HostConfig.objects.create(
@@ -1349,6 +1522,29 @@ class ViewTests(TestCase):
        self.assertContains(response, "Cancel run")
        self.assertContains(response, reverse("cancel_run", args=[run.id]))

+    def test_run_detail_renders_worker_execution_metadata(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        run = BackupRun.objects.create(
+            host=host,
+            status=BackupRun.Status.RUNNING,
+            result={
+                "execution": {
+                    "worker_host": "backup-01",
+                    "worker_pid": 4242,
+                    "heartbeat_at": "2026-05-21T10:30:00+00:00",
+                }
+            },
+        )
+
+        response = self.client.get(reverse("run_detail", args=[run.id]))
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Worker:")
+        self.assertContains(response, "backup-01")
+        self.assertContains(response, "pid 4242")
+        self.assertContains(response, "Worker heartbeat:")
+
    def test_cancel_run_marks_queued_run_cancelled(self) -> None:
        self.client.force_login(self.staff_user)
        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
@@ -1573,6 +1769,68 @@ class ViewTests(TestCase):
        self.assertContains(response, "Incomplete Snapshots")
        self.assertContains(response, "20260519-031500Z__BROKEN01")
        self.assertContains(response, "excluded from retention cleanup")
+        self.assertContains(response, "Delete incomplete snapshots")
+        self.assertContains(response, "Type 1 to confirm the current number of incomplete snapshots.")
+
+    def test_incomplete_cleanup_deletes_incomplete_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")
+            incomplete_dir = Path(tmp) / "backups" / host.host / ".incomplete" / "20260519-031500Z__BROKEN01"
+            incomplete_dir.mkdir(parents=True)
+            incomplete_dir.joinpath("partial-file").write_text("interrupted\n")
+            record = SnapshotRecord.objects.create(
+                host=host,
+                kind=SnapshotRecord.Kind.INCOMPLETE,
+                dirname=incomplete_dir.name,
+                path=str(incomplete_dir),
+                status="failed",
+                started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
+            )
+
+            with override_settings(POBSYNC_HOME=str(home)):
+                response = self.client.post(
+                    reverse("cleanup_host_incomplete_snapshots", args=[host.host]),
+                    {
+                        "max_delete": "1",
+                        "confirm_host": host.host,
+                        "confirm_delete_count": "1",
+                    },
+                    follow=True,
+                )
+
+            self.assertFalse(incomplete_dir.exists())
+
+        self.assertRedirects(response, reverse("host_retention_plan", args=[host.host]))
+        self.assertContains(response, "Deleted 1 incomplete snapshot(s) for web-01.")
+        self.assertFalse(SnapshotRecord.objects.filter(pk=record.pk).exists())
+
+    def test_incomplete_cleanup_rejects_bad_confirmation(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        SnapshotRecord.objects.create(
+            host=host,
+            kind=SnapshotRecord.Kind.INCOMPLETE,
+            dirname="20260519-031500Z__BROKEN01",
+            path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
+            status="failed",
+            started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
+        )
+
+        response = self.client.post(
+            reverse("cleanup_host_incomplete_snapshots", args=[host.host]),
+            {
+                "max_delete": "1",
+                "confirm_host": "wrong",
+                "confirm_delete_count": "1",
+            },
+            follow=True,
+        )
+
+        self.assertRedirects(response, reverse("host_retention_plan", args=[host.host]))
+        self.assertContains(response, "Incomplete cleanup confirmation is invalid.")
+        self.assertEqual(SnapshotRecord.objects.filter(kind=SnapshotRecord.Kind.INCOMPLETE).count(), 1)

    def test_host_detail_surfaces_retention_warnings(self) -> None:
        self.client.force_login(self.staff_user)
@@ -1595,6 +1853,46 @@ class ViewTests(TestCase):
        self.assertContains(response, "Retention Warnings")
        self.assertContains(response, "Scheduled pruning would delete 2 snapshot(s), above max delete")

+    def test_host_detail_can_mark_incomplete_snapshots_reviewed(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        incomplete = SnapshotRecord.objects.create(
+            host=host,
+            kind=SnapshotRecord.Kind.INCOMPLETE,
+            dirname="20260519-031500Z__BROKEN01",
+            path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
+            status="failed",
+            started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
+        )
+
+        response = self.client.post(reverse("resolve_host_incomplete_reviews", args=[host.host]), follow=True)
+
+        incomplete.refresh_from_db()
+        self.assertIsNotNone(incomplete.reviewed_at)
+        self.assertEqual(incomplete.reviewed_by, self.staff_user.username)
+        self.assertRedirects(response, reverse("host_detail", args=[host.host]))
+        self.assertContains(response, "Marked 1 incomplete snapshot(s) reviewed for web-01.")
+        self.assertNotContains(response, "Retention Warnings")
+
+    def test_host_detail_does_not_warn_for_reviewed_incomplete_snapshots(self) -> None:
+        self.client.force_login(self.staff_user)
+        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
+        SnapshotRecord.objects.create(
+            host=host,
+            kind=SnapshotRecord.Kind.INCOMPLETE,
+            dirname="20260519-031500Z__BROKEN01",
+            path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
+            status="failed",
+            started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
+            reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
+            reviewed_by="admin",
+        )
+
+        response = self.client.get(reverse("host_detail", args=[host.host]))
+
+        self.assertEqual(response.status_code, 200)
+        self.assertNotContains(response, "Retention Warnings")
+
    def test_retention_plan_rejects_invalid_kind(self) -> None:
        self.client.force_login(self.staff_user)
        host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
@@ -1646,6 +1944,10 @@ class ViewTests(TestCase):
        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())
+        purged = PurgedSnapshot.objects.get(dirname=old_snapshot.dirname)
+        self.assertEqual(purged.action, PurgedSnapshot.Action.MANUAL)
+        self.assertEqual(purged.triggered_by, self.staff_user.username)
+        self.assertEqual(purged.reason, "outside retention policy")

    def test_retention_apply_rejects_bad_confirmation(self) -> None:
        self.client.force_login(self.staff_user)
--- a/src/pobsync_backend/views.py
+++ b/src/pobsync_backend/views.py
@@ -15,6 +15,7 @@ from django.shortcuts import get_object_or_404, redirect, render
 from django.utils import timezone
 from django.views.decorators.http import require_POST

+from pobsync import __version__
 from pobsync.errors import PobsyncError
 from pobsync.config.defaults import DEFAULT_EXCLUDES, DEFAULT_RSYNC_ARGS

@@ -24,6 +25,7 @@ from .forms import (
    CreateHostConfigForm,
    GlobalConfigForm,
    HostConfigForm,
+    IncompleteCleanupForm,
    ManualBackupForm,
    RetentionApplyForm,
    SshCredentialGenerateForm,
@@ -31,9 +33,9 @@ from .forms import (
    SshCredentialForm,
 )
 from .host_ops import ensure_host_directories
-from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord, SshCredential
+from .models import BackupRun, GlobalConfig, HostConfig, PurgedSnapshot, ScheduleConfig, SnapshotRecord, SshCredential
 from .preflight import collect_backup_gate, effective_host_config_preview, run_remote_preflight
-from .retention import run_sql_retention_apply, run_sql_retention_plan
+from .retention import run_incomplete_cleanup, run_sql_retention_apply, run_sql_retention_plan
 from .self_check import collect_self_checks, summarize_self_checks
 from .scheduler import next_due_after
 from .snapshot_discovery import discover_snapshots, inspect_snapshot_discovery
@@ -51,8 +53,16 @@ def dashboard(request):
            run_count=Count("runs", distinct=True),
            queued_run_count=Count("runs", filter=Q(runs__status=BackupRun.Status.QUEUED), distinct=True),
            running_run_count=Count("runs", filter=Q(runs__status=BackupRun.Status.RUNNING), distinct=True),
-            warning_run_count=Count("runs", filter=Q(runs__status=BackupRun.Status.WARNING), distinct=True),
-            failed_run_count=Count("runs", filter=Q(runs__status=BackupRun.Status.FAILED), distinct=True),
+            warning_run_count=Count(
+                "runs",
+                filter=Q(runs__status=BackupRun.Status.WARNING, runs__reviewed_at__isnull=True),
+                distinct=True,
+            ),
+            failed_run_count=Count(
+                "runs",
+                filter=Q(runs__status=BackupRun.Status.FAILED, runs__reviewed_at__isnull=True),
+                distinct=True,
+            ),
        )
        .order_by("host")
    )
@@ -81,13 +91,41 @@ def dashboard(request):
            "runs": BackupRun.objects.count(),
            "queued_runs": BackupRun.objects.filter(status=BackupRun.Status.QUEUED).count(),
            "running_runs": BackupRun.objects.filter(status=BackupRun.Status.RUNNING).count(),
-            "warning_runs": BackupRun.objects.filter(status=BackupRun.Status.WARNING).count(),
-            "failed_runs": BackupRun.objects.filter(status=BackupRun.Status.FAILED).count(),
+            "warning_runs": BackupRun.objects.filter(
+                status=BackupRun.Status.WARNING,
+                reviewed_at__isnull=True,
+            ).count(),
+            "failed_runs": BackupRun.objects.filter(
+                status=BackupRun.Status.FAILED,
+                reviewed_at__isnull=True,
+            ).count(),
        },
    }
    return render(request, "pobsync_backend/dashboard.html", context)


+@staff_member_required
+def changelog(request):
+    changelog_path = Path(settings.BASE_DIR) / "CHANGELOG.md"
+    try:
+        changelog_text = changelog_path.read_text(encoding="utf-8")
+        missing = False
+    except FileNotFoundError:
+        changelog_text = "CHANGELOG.md was not found in this installation."
+        missing = True
+
+    return render(
+        request,
+        "pobsync_backend/changelog.html",
+        {
+            "app_version": __version__,
+            "changelog_blocks": _parse_changelog(changelog_text),
+            "changelog_path": changelog_path,
+            "missing": missing,
+        },
+    )
+
+
@staff_member_required
 def self_check(request):
    checks = collect_self_checks()
@@ -107,6 +145,27 @@ def logs(request):
    return render(request, "pobsync_backend/logs.html", context)


+@staff_member_required
+def purged_snapshots(request):
+    host = request.GET.get("host", "").strip()
+    action = request.GET.get("action", "").strip()
+    purged = PurgedSnapshot.objects.select_related("host").order_by("-purged_at", "host_name", "dirname")
+    if host:
+        purged = purged.filter(host_name=host)
+    if action:
+        purged = purged.filter(action=action)
+
+    context = {
+        "purged_snapshots": purged[:200],
+        "hosts": HostConfig.objects.order_by("host"),
+        "actions": PurgedSnapshot.Action.choices,
+        "selected_host": host,
+        "selected_action": action,
+        "total_count": purged.count(),
+    }
+    return render(request, "pobsync_backend/purged_snapshots.html", context)
+
+
@staff_member_required
 def ssh_credentials(request):
    context = {
@@ -201,6 +260,9 @@ def delete_ssh_credential(request, credential_id: int):
    if credential.hosts.exists() or credential.global_configs.exists():
        messages.error(request, f"SSH key {credential.name} is still in use and cannot be deleted.")
        return redirect("edit_ssh_credential", credential_id=credential.id)
+    if request.POST.get("confirm_name", "").strip() != credential.name:
+        messages.error(request, f"Type {credential.name} to confirm SSH key deletion.")
+        return redirect("edit_ssh_credential", credential_id=credential.id)

    name = credential.name
    try:
@@ -307,7 +369,10 @@ def host_detail(request, host: str):
            "runs": host_config.runs.count(),
            "queued_runs": queued_runs.count(),
            "running_runs": running_runs.count(),
-            "failed_runs": host_config.runs.filter(status=BackupRun.Status.FAILED).count(),
+            "failed_runs": host_config.runs.filter(
+                status=BackupRun.Status.FAILED,
+                reviewed_at__isnull=True,
+            ).count(),
            "incomplete_snapshots": host_config.snapshots.filter(kind=SnapshotRecord.Kind.INCOMPLETE).count(),
        },
    }
@@ -425,6 +490,7 @@ def run_detail(request, run_id: int):
    rsync_result = result.get("rsync") if isinstance(result.get("rsync"), dict) else {}
    failure = result.get("failure") if isinstance(result.get("failure"), dict) else {}
    prune_result = result.get("prune") if isinstance(result.get("prune"), dict) else {}
+    execution = result.get("execution") if isinstance(result.get("execution"), dict) else {}
    rsync_log_path = _run_rsync_log_path(run)
    rsync_log_tail = _run_rsync_log_tail(rsync_result, rsync_log_path)
    requested = result.get("requested") if isinstance(result.get("requested"), dict) else {}
@@ -432,6 +498,7 @@ def run_detail(request, run_id: int):
        "run": run,
        "can_cancel": run.status in {BackupRun.Status.QUEUED, BackupRun.Status.RUNNING},
        "requested": requested,
+        "execution": execution,
        "stats": run_stats if isinstance(run_stats, dict) else {},
        "rsync": rsync_result,
        "rsync_command": _run_rsync_command(rsync_result),
@@ -489,6 +556,40 @@ def cancel_run(request, run_id: int):
    return redirect("run_detail", run_id=run.id)


+@staff_member_required
+@require_POST
+def resolve_run_review(request, run_id: int):
+    run = get_object_or_404(BackupRun.objects.select_related("host"), id=run_id)
+    if run.status not in {BackupRun.Status.FAILED, BackupRun.Status.WARNING}:
+        messages.warning(request, f"Run {run.id} does not need review.")
+        return redirect("run_detail", run_id=run.id)
+    if run.reviewed_at:
+        messages.info(request, f"Run {run.id} was already marked reviewed.")
+        return redirect("run_detail", run_id=run.id)
+
+    run.reviewed_at = timezone.now()
+    run.reviewed_by = request.user.get_username()
+    run.save(update_fields=["reviewed_at", "reviewed_by"])
+    messages.success(request, f"Run {run.id} marked reviewed.")
+    return redirect("run_detail", run_id=run.id)
+
+
+@staff_member_required
+@require_POST
+def resolve_host_incomplete_reviews(request, host: str):
+    host_config = get_object_or_404(HostConfig, host=host)
+    reviewed_count = host_config.snapshots.filter(
+        kind=SnapshotRecord.Kind.INCOMPLETE,
+        reviewed_at__isnull=True,
+    ).update(reviewed_at=timezone.now(), reviewed_by=request.user.get_username())
+
+    if reviewed_count:
+        messages.success(request, f"Marked {reviewed_count} incomplete snapshot(s) reviewed for {host_config.host}.")
+    else:
+        messages.info(request, f"No incomplete snapshots needed review for {host_config.host}.")
+    return redirect("host_detail", host=host_config.host)
+
+
@staff_member_required
 def snapshot_detail(request, snapshot_id: int):
    snapshot = get_object_or_404(
@@ -544,6 +645,7 @@ def host_retention_plan(request, host: str):
    schedule = _schedule_for_host(host_config)
    scheduled_prune_limit = schedule.prune_max_delete if schedule and schedule.prune else None
    delete_count = len(plan["delete"])
+    incomplete_count = len(plan["incomplete"])
    context = {
        "host": host_config,
        "kind": kind,
@@ -562,6 +664,14 @@ def host_retention_plan(request, host: str):
                "confirm_delete_count": delete_count,
            },
        ),
+        "incomplete_cleanup_form": IncompleteCleanupForm(
+            host_name=host_config.host,
+            expected_delete_count=incomplete_count,
+            initial={
+                "max_delete": incomplete_count,
+                "confirm_delete_count": incomplete_count,
+            },
+        ),
    }
    return render(request, "pobsync_backend/retention_plan.html", context)

@@ -604,6 +714,8 @@ def apply_host_retention(request, host: str):
            protect_bases=protect_bases,
            yes=True,
            max_delete=form.cleaned_data["max_delete"],
+            action=PurgedSnapshot.Action.MANUAL,
+            triggered_by=request.user.get_username(),
        )
    except PobsyncError as exc:
        messages.error(request, str(exc))
@@ -618,6 +730,41 @@ def apply_host_retention(request, host: str):
    return target


+@staff_member_required
+@require_POST
+def cleanup_host_incomplete_snapshots(request, host: str):
+    host_config = get_object_or_404(HostConfig, host=host)
+    try:
+        plan = run_sql_retention_plan(host=host_config.host, kind="all", protect_bases=True)
+    except PobsyncError as exc:
+        messages.error(request, str(exc))
+        return redirect("host_retention_plan", host=host_config.host)
+
+    incomplete_count = len(plan.get("incomplete") or [])
+    form = IncompleteCleanupForm(
+        request.POST,
+        host_name=host_config.host,
+        expected_delete_count=incomplete_count,
+    )
+    if not form.is_valid():
+        messages.error(request, "Incomplete cleanup confirmation is invalid.")
+        return redirect("host_retention_plan", host=host_config.host)
+
+    try:
+        result = run_incomplete_cleanup(
+            prefix=Path(settings.POBSYNC_HOME),
+            host=host_config.host,
+            yes=True,
+            max_delete=form.cleaned_data["max_delete"],
+            triggered_by=request.user.get_username(),
+        )
+    except PobsyncError as exc:
+        messages.error(request, str(exc))
+    else:
+        messages.success(request, f"Deleted {len(result['deleted'])} incomplete snapshot(s) for {host_config.host}.")
+    return redirect("host_retention_plan", host=host_config.host)
+
+
@staff_member_required
 def edit_host_config(request, host: str):
    host_config = get_object_or_404(HostConfig, host=host)
@@ -695,7 +842,10 @@ def _next_run_for_schedule(schedule: ScheduleConfig | None, host_config: HostCon


 def _retention_warning_for_host(host_config: HostConfig, schedule: ScheduleConfig | None) -> dict[str, object]:
-    incomplete_count = host_config.snapshots.filter(kind=SnapshotRecord.Kind.INCOMPLETE).count()
+    incomplete_count = host_config.snapshots.filter(
+        kind=SnapshotRecord.Kind.INCOMPLETE,
+        reviewed_at__isnull=True,
+    ).count()
    warning: dict[str, object] = {
        "has_warning": incomplete_count > 0,
        "incomplete_count": incomplete_count,
@@ -793,6 +943,49 @@ def _pretty_json(value: object) -> str:
    return json.dumps(value or {}, indent=2, sort_keys=True)


+def _parse_changelog(text: str) -> list[dict[str, object]]:
+    blocks: list[dict[str, object]] = []
+    paragraph: list[str] = []
+    list_items: list[str] = []
+
+    def flush_paragraph() -> None:
+        if paragraph:
+            blocks.append({"kind": "paragraph", "text": " ".join(paragraph)})
+            paragraph.clear()
+
+    def flush_list() -> None:
+        if list_items:
+            blocks.append({"kind": "list", "items": list(list_items)})
+            list_items.clear()
+
+    for raw_line in text.splitlines():
+        line = raw_line.strip()
+        if not line:
+            flush_paragraph()
+            flush_list()
+            continue
+
+        if line.startswith("#"):
+            flush_paragraph()
+            flush_list()
+            marker, _space, heading = line.partition(" ")
+            level = min(max(len(marker), 1), 3)
+            blocks.append({"kind": "heading", "level": level, "text": heading.strip() or line.lstrip("#").strip()})
+            continue
+
+        if line.startswith("- "):
+            flush_paragraph()
+            list_items.append(line[2:].strip())
+            continue
+
+        flush_list()
+        paragraph.append(line)
+
+    flush_paragraph()
+    flush_list()
+    return blocks
+
+
 def _snapshot_restore_guidance(snapshot: SnapshotRecord) -> dict[str, str]:
    source_path = Path(snapshot.path) / "data"
    destination_path = Path("/restore") / snapshot.host.host
--- a/src/pobsync_server/urls.py
+++ b/src/pobsync_server/urls.py
@@ -8,8 +8,10 @@ from pobsync_backend import api, views

 urlpatterns = [
    path("", views.dashboard, name="dashboard"),
+    path("changelog/", views.changelog, name="changelog"),
    path("self-check/", views.self_check, name="self_check"),
    path("logs/", views.logs, name="logs"),
+    path("purged-snapshots/", views.purged_snapshots, name="purged_snapshots"),
    path("config/global/", views.edit_global_config, name="edit_global_config"),
    path("ssh-credentials/", views.ssh_credentials, name="ssh_credentials"),
    path("ssh-credentials/new/", views.create_ssh_credential, name="create_ssh_credential"),
@@ -26,10 +28,21 @@ urlpatterns = [
    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>/incomplete-cleanup/",
+        views.cleanup_host_incomplete_snapshots,
+        name="cleanup_host_incomplete_snapshots",
+    ),
    path("hosts/<str:host>/schedule/", views.edit_host_schedule, name="edit_host_schedule"),
    path("runs/<int:run_id>/", views.run_detail, name="run_detail"),
    path("runs/<int:run_id>/rsync-log/", views.run_rsync_log, name="run_rsync_log"),
    path("runs/<int:run_id>/cancel/", views.cancel_run, name="cancel_run"),
+    path("runs/<int:run_id>/resolve-review/", views.resolve_run_review, name="resolve_run_review"),
+    path(
+        "hosts/<str:host>/resolve-incomplete-reviews/",
+        views.resolve_host_incomplete_reviews,
+        name="resolve_host_incomplete_reviews",
+    ),
    path("snapshots/<int:snapshot_id>/", views.snapshot_detail, name="snapshot_detail"),
    path("api/", api.api_index),
    path("api/status/", api.status),