chore: release v1.4.0

Ships the Portainer backup executor.

Co-Authored-By: Claude <noreply@anthropic.com>

.gitea/workflows/build.yml

@@ -26,7 +26,7 @@ jobs:
       - name: Login to Gitea Registry
         uses: docker/login-action@v3
         with:
-          registry: git.schweitz.internal
+          registry: git.schweitz.net
           username: ${{ secrets.REGISTRY_USER }}
           password: ${{ secrets.REGISTRY_PASSWORD }}
 
@@ -36,8 +36,8 @@ jobs:
           context: .
           push: true
           tags: |
-            git.schweitz.internal/jpmschweitzer/scheduler:latest
-            git.schweitz.internal/jpmschweitzer/scheduler:${{ github.ref_name }}
+            git.schweitz.net/jpmschweitzer/scheduler:latest
+            git.schweitz.net/jpmschweitzer/scheduler:${{ github.ref_name }}
 
       - name: Trigger Watchtower update
         if: success()

CHANGELOG.md

@@ -4,6 +4,46 @@ All notable changes to The Scheduler will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## [Unreleased]
+
+## [1.4.0] - 2026-08-08
+
+### Added
+- **Portainer Backup Executor** (`portainer_backup_executor.py`) — archives Portainer's
+  own state through its `/api/backup` endpoint. Portainer's BoltDB lives in a Docker
+  volume that the daily config backup does not cover, so losing that volume would take
+  every stack definition with it. Uses the API rather than tarring the live volume, and
+  rejects a 200 whose body is not a readable archive.
+
+
+## [1.3.0] - 2026-08-08
+
+### Added
+- **Postgres Retention Executor** (`postgres_retention_executor.py`) — deletes rows
+  past a retention window from a table on the shared Postgres server. Uses the
+  Scheduler's own credentials with only the database name overridden, so the target
+  database grants `scheduler_user` SELECT and DELETE on the table.
+- **Docker Prune Executor** (`docker_prune_executor.py`) — scheduled reclaim of Docker
+  disk usage. Build cache and dangling images are pruned by default; unused images and
+  volumes are opt-in, since volume pruning also removes volumes belonging to stopped
+  containers.
+
+### Fixed
+- `POST /tasks` returned HTTP 500 after successfully creating the task. The response
+  model declared `created_at`/`updated_at` as strings while the database returns
+  timestamps, so every create looked like a failure and retrying hit a duplicate-key
+  error.
+
+### Changed
+- `TASK_REGISTRATION.md` now lists the executors that exist. It previously advertised
+  `shell`, `python` and `docker` executors that were never implemented.
+
+## [1.2.0] - 2026-03-30
+
+### Added
+- **GCS Backup Executor** (`gcs_backup_executor.py`) — offsite backup to Google Cloud
+  Storage.
+
 ## [1.1.3] - 2026-01-08
 
 ### Changed

TASK_REGISTRATION.md

@@ -105,11 +105,83 @@ Calls HTTP endpoints. Supports environment variable substitution in headers/body
 
 ### Other Executors
 
-- `shell`: Execute shell commands
-- `python`: Execute Python scripts
-- `docker`: Docker operations
-- `backup`: Backup operations
-- `doc_sync`: Documentation sync
+The `executor` field is the module name under `src/executors/`. These are the
+modules that actually exist:
+
+- `config_backup_executor`: tar.gz backup of mounted directories, with retention
+- `gcs_backup_executor`: offsite backup to Google Cloud Storage
+- `doc_sync_executor`: mirror upstream docs into Gitea
+- `gitea_release_cleanup_executor`: drop old Gitea releases, keeping the newest N
+- `postgres_retention_executor`: delete rows past a retention window (see below)
+- `docker_prune_executor`: reclaim Docker disk usage (see below)
+- `portainer_backup_executor`: archive Portainer's own state via its backup API (see below)
+- `example_executor`: demo/test
+
+There is **no `shell` or `python` executor**. Earlier revisions of this document
+listed them and they were never implemented; work needing a shell belongs either
+in a purpose-built executor or on a host systemd timer.
+
+#### `postgres_retention_executor`
+
+Connects with the Scheduler's own Postgres credentials, overriding only the
+database name, so the target database must grant `scheduler_user` SELECT and
+DELETE on the table. Table and column names are validated against a strict
+identifier pattern because they cannot be bound as query parameters.
+
+```json
+{
+  "database": "sysmon",
+  "table": "check_history",
+  "timestamp_column": "ts",
+  "retention_days": 30,
+  "dry_run": false
+}
+```
+
+#### `portainer_backup_executor`
+
+Portainer keeps every stack definition, endpoint, user and access-control rule in
+a BoltDB inside the `portainer_data` Docker volume, which lives under
+`/var/lib/docker/volumes/` and is **not** covered by the daily config backup.
+This calls Portainer's `/api/backup` rather than tarring the volume: BoltDB is a
+single memory-mapped file, so copying it live can capture a torn page.
+
+The archive contains TLS certificates and private keys and is written `0600`. A
+200 response whose body is not a readable archive is treated as a failure — an
+archive that will not open is worse than a missing one, because it looks like a
+backup until the day it is needed.
+
+```json
+{
+  "url": "${PORTAINER_URL}",
+  "api_key": "${PORTAINER_API_KEY}",
+  "output_dir": "/backups/portainer",
+  "retention_days": 30
+}
+```
+
+Portainer runs host-networked, so a container name does not resolve; use the
+host address. Requires `/mnt/media/backups/portainer` mounted into the container.
+
+#### `docker_prune_executor`
+
+Uses the docker socket already mounted into the container. Only the two stages
+that discard regenerable data are on by default.
+
+```json
+{
+  "build_cache": true,
+  "dangling_images": true,
+  "unused_images": false,
+  "volumes": false,
+  "build_cache_until_hours": 168,
+  "dry_run": false
+}
+```
+
+**`volumes` removes volumes belonging to merely-stopped containers, not just
+orphaned ones.** Leave it off unless you have checked what is currently
+unattached; on this host it is a plausible way to lose a database.
 
 ## Complete Task Schema
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "the-scheduler"
-version = "1.2.0"
+version = "1.4.0"
 description = "System-wide maintenance orchestration - backups, doc mirroring, cleanup, task automation"
 readme = "README.md"
 requires-python = ">=3.12"

src/executors/docker_prune_executor.py

@@ -0,0 +1,109 @@
+"""
+Docker Prune Executor
+
+Scheduled, non-interactive reclaim of Docker disk usage. The host equivalent is
+system-admin-toj's scripts/disk/prune-docker.sh, which prompts per stage; a cron
+task cannot prompt, so the destructive stages are opt-in instead.
+
+Runs the docker CLI against the socket already mounted into this container.
+
+Config schema:
+    {
+        "build_cache": true,        # safe: cache is rebuilt on demand
+        "dangling_images": true,    # safe: untagged layers nothing references
+        "unused_images": false,     # re-pull on next deploy; costs bandwidth
+        "volumes": false,           # DESTRUCTIVE - see below
+        "build_cache_until_hours": 168,
+        "dry_run": false
+    }
+
+`volumes` is off by default and should stay off unless you have checked what is
+actually unattached. `docker volume prune` removes every volume not bound to a
+*running* container, which includes the data volume of anything merely stopped.
+On this host that is a plausible way to lose a database.
+
+Defaults are the two stages that only ever discard regenerable data.
+"""
+import asyncio
+import logging
+from typing import Any, Dict, List, Tuple
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+COMMAND_TIMEOUT = 900
+
+
+async def _run(args: List[str]) -> Tuple[int, str, str]:
+    proc = await asyncio.create_subprocess_exec(
+        *args,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=COMMAND_TIMEOUT)
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.wait()
+        raise Exception(f"timed out after {COMMAND_TIMEOUT}s: {' '.join(args)}")
+    return proc.returncode, stdout.decode().strip(), stderr.decode().strip()
+
+
+def _reclaimed(output: str) -> str:
+    """Pull the 'Total reclaimed space: X' line out of docker's prune output."""
+    for line in output.splitlines():
+        if "reclaimed space" in line.lower():
+            return line.split(":", 1)[1].strip()
+    return "0B"
+
+
+async def execute(config: Dict[str, Any], settings: Settings) -> str:
+    dry_run = bool(config.get("dry_run", False))
+    until_hours = int(config.get("build_cache_until_hours", 168))
+
+    stages: List[Tuple[str, List[str]]] = []
+    if config.get("build_cache", True):
+        stages.append(
+            ("build cache", ["docker", "builder", "prune", "-f", "--filter", f"until={until_hours}h"])
+        )
+    if config.get("dangling_images", True):
+        stages.append(("dangling images", ["docker", "image", "prune", "-f"]))
+    if config.get("unused_images", False):
+        stages.append(("unused images", ["docker", "image", "prune", "-a", "-f"]))
+    if config.get("volumes", False):
+        logger.warning(
+            "volume pruning is enabled; this removes volumes belonging to stopped "
+            "containers, not just orphaned ones"
+        )
+        stages.append(("volumes", ["docker", "volume", "prune", "-f"]))
+
+    if not stages:
+        return "no prune stages enabled; nothing to do"
+
+    rc, out, err = await _run(["docker", "system", "df"])
+    if rc != 0:
+        raise Exception(f"docker unavailable: {err or out}")
+    before = out
+
+    if dry_run:
+        planned = ", ".join(name for name, _ in stages)
+        logger.info("dry run; would prune: %s", planned)
+        return f"dry run - would prune: {planned}\n{before}"
+
+    results = []
+    for name, args in stages:
+        rc, out, err = await _run(args)
+        if rc != 0:
+            # Report rather than abort: a later stage may still reclaim space, and
+            # a partial reclaim is more useful than none.
+            logger.error("prune stage %r failed: %s", name, err or out)
+            results.append(f"{name}: FAILED ({(err or out).splitlines()[0] if (err or out) else 'unknown'})")
+            continue
+        results.append(f"{name}: {_reclaimed(out)}")
+        logger.info("pruned %s -> %s", name, _reclaimed(out))
+
+    summary = "; ".join(results)
+    if any("FAILED" in r for r in results):
+        raise Exception(f"one or more prune stages failed: {summary}")
+    return f"reclaimed - {summary}"

src/executors/portainer_backup_executor.py

@@ -0,0 +1,140 @@
+"""
+Portainer Backup Executor
+
+Archives Portainer's own state through its `/api/backup` endpoint.
+
+Why it needs backing up separately: Portainer keeps every stack definition,
+endpoint, user and access-control rule in a BoltDB inside the Docker volume
+`portainer_data`, which lives under /var/lib/docker/volumes/. The daily config
+backup covers ~/docker-data and code-server-config only, so that volume is not
+in it. Losing it takes all 24 stack definitions with it.
+
+Why the API rather than tarring the volume: BoltDB is a single memory-mapped
+file, so copying it while Portainer is writing can capture a torn page. The API
+serialises a consistent snapshot.
+
+The archive contains TLS certificates and private keys, so it is written 0600.
+
+Config schema:
+    {
+        "url": "http://172.17.0.1:8001",     # Portainer is host-networked, so a
+                                             # container name does not resolve;
+                                             # use the bridge gateway
+        "api_key": "${PORTAINER_API_KEY}",   # ${VAR} reads the container env
+        "output_dir": "/backups/portainer",
+        "retention_days": 30,
+        "password": ""                       # optional; encrypts the archive
+    }
+"""
+import logging
+import os
+import re
+import tarfile
+import time
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import httpx
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+BACKUP_TIMEOUT = 300
+FILENAME_RE = re.compile(r"^portainer-\d{8}T\d{6}Z\.tar\.gz$")
+
+
+def _substitute_env(value: str) -> str:
+    """Expand ${VAR} against the container environment, as rest_api does."""
+    if not isinstance(value, str):
+        return value
+    for var in re.findall(r"\$\{([A-Z_][A-Z0-9_]*)\}", value):
+        resolved = os.getenv(var, "")
+        if not resolved:
+            logger.warning("environment variable not found: %s", var)
+        value = value.replace(f"${{{var}}}", resolved)
+    return value
+
+
+def _prune(output_dir: Path, retention_days: int) -> int:
+    """Delete archives older than the retention window. Returns how many went."""
+    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
+    removed = 0
+    for path in output_dir.glob("portainer-*.tar.gz"):
+        # Match the exact name this executor writes; never delete a stray file
+        # someone else put here.
+        if not FILENAME_RE.match(path.name):
+            continue
+        if datetime.fromtimestamp(path.stat().st_mtime, timezone.utc) < cutoff:
+            path.unlink()
+            removed += 1
+            logger.info("pruned old portainer backup: %s", path.name)
+    return removed
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    url = _substitute_env(config.get("url", "")).rstrip("/")
+    api_key = _substitute_env(config.get("api_key", ""))
+    output_dir = Path(config.get("output_dir", "/backups/portainer"))
+    retention_days = config.get("retention_days", 30)
+    password = _substitute_env(config.get("password", "") or "")
+
+    if not url:
+        raise ValueError("Missing required config: 'url'")
+    if not api_key:
+        raise ValueError("Missing or unresolved config: 'api_key'")
+    if not isinstance(retention_days, int) or isinstance(retention_days, bool) or retention_days < 1:
+        raise ValueError(f"retention_days must be a positive integer, got {retention_days!r}")
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    final = output_dir / f"portainer-{stamp}.tar.gz"
+    partial = final.with_suffix(".partial")
+
+    started = time.monotonic()
+    try:
+        async with httpx.AsyncClient(timeout=BACKUP_TIMEOUT) as client:
+            response = await client.post(
+                f"{url}/api/backup",
+                headers={"X-API-Key": api_key, "Content-Type": "application/json"},
+                json={"password": password} if password else {},
+            )
+            if response.status_code != 200:
+                raise Exception(
+                    f"Portainer returned HTTP {response.status_code}: {response.text[:200]}"
+                )
+            partial.write_bytes(response.content)
+
+        # A 200 with a truncated body is still a failed backup. An archive that
+        # cannot be opened is worse than a missing one, because it looks like a
+        # backup until the day it is needed.
+        if not password:
+            try:
+                with tarfile.open(partial, "r:gz") as archive:
+                    entries = len(archive.getnames())
+            except Exception as exc:  # noqa: BLE001
+                # Deliberately broad. A truncated archive raises EOFError, which
+                # is neither TarError nor OSError, and any failure to open it
+                # means the same thing regardless of type: this is not a backup.
+                raise Exception(f"response is not a readable archive: {exc}") from exc
+        else:
+            entries = -1  # encrypted; contents cannot be verified here
+
+        partial.replace(final)
+        final.chmod(0o600)  # contains TLS certs and private keys
+    finally:
+        if partial.exists():
+            partial.unlink()
+
+    removed = _prune(output_dir, retention_days)
+    kept = len([p for p in output_dir.glob("portainer-*.tar.gz") if FILENAME_RE.match(p.name)])
+    size_mb = final.stat().st_size / 1_048_576
+    elapsed = time.monotonic() - started
+
+    summary = (
+        f"backed up Portainer to {final.name} "
+        f"({size_mb:.2f} MB{'' if entries < 0 else f', {entries} entries'}, {elapsed:.1f}s); "
+        f"kept {kept}, pruned {removed} older than {retention_days}d"
+    )
+    logger.info(summary)
+    return summary

src/executors/postgres_retention_executor.py

@@ -0,0 +1,109 @@
+"""
+Postgres Retention Executor
+
+Deletes rows older than a retention window from a table on the shared Postgres
+server. Written for sysmon's `check_history`, which grows with every monitoring
+check and had no retention at all, but the executor is table-agnostic.
+
+Connects with the Scheduler's own Postgres credentials and only overrides the
+database name. That keeps a second set of credentials out of the stack; the
+target database grants `scheduler_user` exactly SELECT and DELETE on the table,
+so a bug here can drop old rows but cannot corrupt or forge history.
+
+Config schema:
+    {
+        "database": "sysmon",           # defaults to the Scheduler's own database
+        "table": "check_history",       # required
+        "timestamp_column": "ts",       # required
+        "retention_days": 30,           # required, must be >= 1
+        "dry_run": false                # count what would go, delete nothing
+    }
+
+Table and column names cannot be passed as query parameters, so both are
+validated against a strict identifier pattern before being interpolated.
+
+Autovacuum reclaims the space afterwards; this deliberately does not VACUUM,
+which would need table ownership the Scheduler intentionally does not have.
+"""
+import asyncio
+import logging
+import re
+from typing import Any, Dict
+
+import psycopg2
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+# Deliberately strict: unquoted lowercase identifiers only. Anything needing
+# quoting is out of scope and would be a hole in the interpolation below.
+IDENTIFIER_RE = re.compile(r"^[a-z_][a-z0-9_]*$")
+
+MAX_RETENTION_DAYS = 3650
+
+
+def _validate_identifier(value: str, label: str) -> str:
+    if not isinstance(value, str) or not IDENTIFIER_RE.match(value):
+        raise ValueError(
+            f"invalid {label}: {value!r} (expected an unquoted lowercase identifier)"
+        )
+    return value
+
+
+def _prune(config: Dict[str, Any], settings: Settings) -> str:
+    table = _validate_identifier(config.get("table", ""), "table")
+    column = _validate_identifier(config.get("timestamp_column", ""), "timestamp_column")
+    database = config.get("database") or settings.postgres_db
+    _validate_identifier(database, "database")
+
+    retention_days = config.get("retention_days")
+    if not isinstance(retention_days, int) or isinstance(retention_days, bool):
+        raise ValueError(f"retention_days must be an integer, got {retention_days!r}")
+    # A zero or negative window would delete everything, including the row the
+    # check just wrote. Refuse rather than quietly wipe the table.
+    if retention_days < 1 or retention_days > MAX_RETENTION_DAYS:
+        raise ValueError(
+            f"retention_days must be between 1 and {MAX_RETENTION_DAYS}, got {retention_days}"
+        )
+
+    dry_run = bool(config.get("dry_run", False))
+    cutoff_sql = f"{column} < now() - make_interval(days => %s)"
+
+    conn = psycopg2.connect(
+        host=settings.postgres_host,
+        port=settings.postgres_port,
+        database=database,
+        user=settings.postgres_user,
+        password=settings.postgres_password,
+        connect_timeout=10,
+    )
+    try:
+        with conn:
+            with conn.cursor() as cur:
+                cur.execute(f"SELECT count(*) FROM {table} WHERE {cutoff_sql}", (retention_days,))
+                stale = cur.fetchone()[0]
+
+                if dry_run:
+                    logger.info("dry run: %s rows in %s.%s exceed %sd", stale, database, table, retention_days)
+                    return f"dry run: {stale} rows older than {retention_days}d in {database}.{table}"
+
+                if stale == 0:
+                    return f"nothing to prune in {database}.{table} (retention {retention_days}d)"
+
+                cur.execute(f"DELETE FROM {table} WHERE {cutoff_sql}", (retention_days,))
+                deleted = cur.rowcount
+
+                cur.execute(f"SELECT count(*) FROM {table}")
+                remaining = cur.fetchone()[0]
+    finally:
+        conn.close()
+
+    logger.info("pruned %s rows from %s.%s, %s remain", deleted, database, table, remaining)
+    return f"pruned {deleted} rows older than {retention_days}d from {database}.{table}, {remaining} remain"
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """Delete rows past the retention window. Returns a one-line summary."""
+    # psycopg2 is synchronous; keep it off the scheduler's event loop.
+    return await asyncio.to_thread(_prune, config, settings)

src/models.py

@@ -3,6 +3,7 @@ Pydantic models for The Scheduler API.
 """
 from pydantic import BaseModel, Field
 from typing import Optional, Dict, Any
+from datetime import datetime
 from enum import Enum
 
 
@@ -203,8 +204,13 @@ class TaskUpdate(BaseModel):
 
 class TaskResponse(TaskCreate):
     """Response model for task operations."""
-    created_at: str
-    updated_at: Optional[str] = None
+    # These are `timestamp` columns, so psycopg2 hands back datetime objects.
+    # Declaring them as `str` made Pydantic reject every create response, which
+    # 500'd the endpoint *after* the row had already been inserted and committed.
+    # FastAPI serialises datetime to an ISO 8601 string, so the JSON on the wire
+    # is unchanged — and now matches what GET /tasks/{name} already returned.
+    created_at: datetime
+    updated_at: Optional[datetime] = None
 
     class Config:
         from_attributes = True

tests/test_docker_prune_executor.py

@@ -0,0 +1,135 @@
+"""
+Tests for the docker prune executor.
+
+The important property is which stages run. `volumes` removes volumes belonging
+to merely-stopped containers, so it must never be enabled by accident, and the
+safe stages must stay on by default.
+"""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from src.config import Settings
+from src.executors import docker_prune_executor as prune
+
+
+def _runner(reclaimed="Total reclaimed space: 1.5GB", rc=0):
+    """Fake _run returning docker-shaped output for every invocation."""
+    async def run(args):
+        if args[:3] == ["docker", "system", "df"]:
+            return 0, "TYPE  TOTAL  ACTIVE  SIZE  RECLAIMABLE", ""
+        return rc, reclaimed, "" if rc == 0 else "boom"
+    return run
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestStageSelection:
+
+    @pytest.mark.asyncio
+    async def test_defaults_run_only_the_safe_stages(self, test_settings: Settings):
+        calls = []
+
+        async def run(args):
+            calls.append(args)
+            if args[:3] == ["docker", "system", "df"]:
+                return 0, "df output", ""
+            return 0, "Total reclaimed space: 0B", ""
+
+        with patch.object(prune, "_run", run):
+            await prune.execute({}, test_settings)
+
+        joined = [" ".join(c) for c in calls]
+        assert any("builder prune" in c for c in joined)
+        assert any("image prune -f" in c for c in joined)
+        # The destructive ones must not appear without being asked for.
+        assert not any("volume prune" in c for c in joined)
+        assert not any("image prune -a" in c for c in joined)
+
+    @pytest.mark.asyncio
+    async def test_volumes_only_when_explicitly_enabled(self, test_settings: Settings):
+        calls = []
+
+        async def run(args):
+            calls.append(args)
+            if args[:3] == ["docker", "system", "df"]:
+                return 0, "df output", ""
+            return 0, "Total reclaimed space: 2GB", ""
+
+        with patch.object(prune, "_run", run):
+            await prune.execute({"volumes": True}, test_settings)
+
+        assert any("volume prune" in " ".join(c) for c in calls)
+
+    @pytest.mark.asyncio
+    async def test_all_stages_disabled_is_a_no_op(self, test_settings: Settings):
+        with patch.object(prune, "_run", AsyncMock()) as run:
+            result = await prune.execute(
+                {"build_cache": False, "dangling_images": False}, test_settings
+            )
+        assert "nothing to do" in result
+        run.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_dry_run_executes_no_prune(self, test_settings: Settings):
+        calls = []
+
+        async def run(args):
+            calls.append(args)
+            return 0, "df output", ""
+
+        with patch.object(prune, "_run", run):
+            result = await prune.execute({"dry_run": True}, test_settings)
+
+        assert "dry run" in result
+        assert all("prune" not in " ".join(c) for c in calls)
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestFailureHandling:
+
+    @pytest.mark.asyncio
+    async def test_docker_unavailable_raises(self, test_settings: Settings):
+        async def run(args):
+            return 1, "", "Cannot connect to the Docker daemon"
+
+        with patch.object(prune, "_run", run):
+            with pytest.raises(Exception, match="docker unavailable"):
+                await prune.execute({}, test_settings)
+
+    @pytest.mark.asyncio
+    async def test_failed_stage_surfaces_but_others_still_run(self, test_settings: Settings):
+        attempted = []
+
+        async def run(args):
+            if args[:3] == ["docker", "system", "df"]:
+                return 0, "df output", ""
+            attempted.append(" ".join(args))
+            if "builder" in args:
+                return 1, "", "builder exploded"
+            return 0, "Total reclaimed space: 3MB", ""
+
+        with patch.object(prune, "_run", run):
+            with pytest.raises(Exception, match="one or more prune stages failed"):
+                await prune.execute({}, test_settings)
+
+        # The image stage must still have been attempted after builder failed.
+        assert any("image prune" in a for a in attempted)
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestOutputParsing:
+
+    @pytest.mark.parametrize(
+        "output,expected",
+        [
+            ("Total reclaimed space: 1.5GB", "1.5GB"),
+            ("deleted: sha256:abc\nTotal reclaimed space: 0B", "0B"),
+            ("no such line", "0B"),
+            ("", "0B"),
+        ],
+    )
+    def test_reclaimed_parsing(self, output, expected):
+        assert prune._reclaimed(output) == expected

tests/test_portainer_backup_executor.py

@@ -0,0 +1,175 @@
+"""
+Tests for the Portainer backup executor.
+
+The point of this executor is producing an archive that will still open on the
+day it is needed, so most of these cover the failure paths: a truncated body
+behind a 200, a partial file left on disk, and retention deleting the wrong
+thing.
+"""
+import gzip
+import io
+import tarfile
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from src.config import Settings
+from src.executors import portainer_backup_executor as pbe
+
+
+def _tar_gz_bytes(names=("compose/1/docker-compose.yml", "certs/cert.pem")) -> bytes:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        for n in names:
+            data = b"x"
+            info = tarfile.TarInfo(name=n)
+            info.size = len(data)
+            tar.addfile(info, io.BytesIO(data))
+    return buf.getvalue()
+
+
+def _mock_post(status=200, content=None):
+    response = MagicMock()
+    response.status_code = status
+    response.content = content if content is not None else _tar_gz_bytes()
+    response.text = "error body"
+    client = MagicMock()
+    client.post = AsyncMock(return_value=response)
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=client)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    return ctx, client
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestConfigValidation:
+
+    @pytest.mark.asyncio
+    async def test_missing_url_rejected(self, test_settings: Settings, tmp_path):
+        with pytest.raises(ValueError, match="url"):
+            await pbe.execute({"api_key": "k", "output_dir": str(tmp_path)}, test_settings)
+
+    @pytest.mark.asyncio
+    async def test_missing_api_key_rejected(self, test_settings: Settings, tmp_path):
+        with pytest.raises(ValueError, match="api_key"):
+            await pbe.execute({"url": "http://x", "output_dir": str(tmp_path)}, test_settings)
+
+    @pytest.mark.asyncio
+    async def test_unresolved_env_var_rejected(self, test_settings: Settings, tmp_path, monkeypatch):
+        """${VAR} that expands to nothing must fail, not send an empty key."""
+        monkeypatch.delenv("NOPE_MISSING", raising=False)
+        with pytest.raises(ValueError, match="api_key"):
+            await pbe.execute(
+                {"url": "http://x", "api_key": "${NOPE_MISSING}", "output_dir": str(tmp_path)},
+                test_settings,
+            )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("bad", [0, -1, "30", None, True])
+    async def test_bad_retention_rejected(self, bad, test_settings: Settings, tmp_path):
+        with pytest.raises(ValueError, match="retention_days"):
+            await pbe.execute(
+                {"url": "http://x", "api_key": "k", "output_dir": str(tmp_path),
+                 "retention_days": bad},
+                test_settings,
+            )
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestBackupBehaviour:
+
+    def _config(self, tmp_path, **over):
+        cfg = {"url": "http://portainer:9000", "api_key": "k",
+               "output_dir": str(tmp_path), "retention_days": 30}
+        cfg.update(over)
+        return cfg
+
+    @pytest.mark.asyncio
+    async def test_writes_verified_archive(self, test_settings: Settings, tmp_path):
+        ctx, _ = _mock_post()
+        with patch("httpx.AsyncClient", return_value=ctx):
+            result = await pbe.execute(self._config(tmp_path), test_settings)
+
+        files = list(tmp_path.glob("portainer-*.tar.gz"))
+        assert len(files) == 1
+        assert "2 entries" in result
+        with tarfile.open(files[0], "r:gz") as tar:      # opens = usable backup
+            assert "certs/cert.pem" in tar.getnames()
+
+    @pytest.mark.asyncio
+    async def test_archive_is_not_world_readable(self, test_settings: Settings, tmp_path):
+        """It contains TLS private keys."""
+        ctx, _ = _mock_post()
+        with patch("httpx.AsyncClient", return_value=ctx):
+            await pbe.execute(self._config(tmp_path), test_settings)
+        f = next(tmp_path.glob("portainer-*.tar.gz"))
+        assert oct(f.stat().st_mode)[-3:] == "600"
+
+    @pytest.mark.asyncio
+    async def test_http_error_raises_and_leaves_nothing(self, test_settings: Settings, tmp_path):
+        ctx, _ = _mock_post(status=401)
+        with patch("httpx.AsyncClient", return_value=ctx):
+            with pytest.raises(Exception, match="HTTP 401"):
+                await pbe.execute(self._config(tmp_path), test_settings)
+        assert list(tmp_path.iterdir()) == []
+
+    @pytest.mark.asyncio
+    async def test_truncated_body_behind_200_is_rejected(self, test_settings: Settings, tmp_path):
+        """The dangerous case: a 200 whose body is not a usable archive."""
+        broken = _tar_gz_bytes()[:40]
+        ctx, _ = _mock_post(content=broken)
+        with patch("httpx.AsyncClient", return_value=ctx):
+            with pytest.raises(Exception, match="not a readable archive"):
+                await pbe.execute(self._config(tmp_path), test_settings)
+        # no .partial and no final file left behind
+        assert list(tmp_path.iterdir()) == []
+
+    @pytest.mark.asyncio
+    async def test_gzip_that_is_not_a_tar_is_rejected(self, test_settings: Settings, tmp_path):
+        ctx, _ = _mock_post(content=gzip.compress(b"not a tar"))
+        with patch("httpx.AsyncClient", return_value=ctx):
+            with pytest.raises(Exception, match="not a readable archive"):
+                await pbe.execute(self._config(tmp_path), test_settings)
+        assert list(tmp_path.iterdir()) == []
+
+    @pytest.mark.asyncio
+    async def test_api_key_resolved_from_env(self, test_settings: Settings, tmp_path, monkeypatch):
+        monkeypatch.setenv("PT_KEY", "secret-value")
+        ctx, client = _mock_post()
+        with patch("httpx.AsyncClient", return_value=ctx):
+            await pbe.execute(self._config(tmp_path, api_key="${PT_KEY}"), test_settings)
+        assert client.post.call_args.kwargs["headers"]["X-API-Key"] == "secret-value"
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestRetention:
+
+    def _age(self, path, days):
+        import os
+        old = (datetime.now(timezone.utc) - timedelta(days=days)).timestamp()
+        os.utime(path, (old, old))
+
+    def test_prunes_only_past_the_window(self, tmp_path):
+        fresh = tmp_path / "portainer-20260808T120000Z.tar.gz"
+        stale = tmp_path / "portainer-20260101T120000Z.tar.gz"
+        for f in (fresh, stale):
+            f.write_bytes(b"x")
+        self._age(stale, 45)
+
+        assert pbe._prune(tmp_path, 30) == 1
+        assert fresh.exists() and not stale.exists()
+
+    def test_leaves_unrelated_files_alone(self, tmp_path):
+        """Retention must not touch anything it did not write."""
+        other = tmp_path / "important-database-dump.tar.gz"
+        named_alike = tmp_path / "portainer-backup-manual.tar.gz"
+        for f in (other, named_alike):
+            f.write_bytes(b"x")
+            self._age(f, 400)
+
+        assert pbe._prune(tmp_path, 30) == 0
+        assert other.exists() and named_alike.exists()

tests/test_postgres_retention_executor.py

@@ -0,0 +1,151 @@
+"""
+Tests for the postgres retention executor.
+
+Focus is on the guards. The executor interpolates a table and column name
+straight into SQL (they cannot be bound as parameters), and it issues DELETEs
+against a live table, so the validation in front of both is what keeps a
+malformed config from becoming data loss.
+"""
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from src.config import Settings
+from src.executors import postgres_retention_executor as retention
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestIdentifierValidation:
+    """Table/column/database names are interpolated, so they must be rejected early."""
+
+    @pytest.mark.parametrize(
+        "bad",
+        [
+            "check_history; DROP TABLE users",
+            'check_history"',
+            "check history",
+            "Check_History",   # uppercase would need quoting to resolve
+            "1_history",
+            "",
+            "--comment",
+        ],
+    )
+    def test_rejects_unsafe_identifiers(self, bad):
+        with pytest.raises(ValueError):
+            retention._validate_identifier(bad, "table")
+
+    @pytest.mark.parametrize("good", ["check_history", "ts", "_private", "a1"])
+    def test_accepts_plain_identifiers(self, good):
+        assert retention._validate_identifier(good, "table") == good
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestRetentionGuards:
+    """A bad retention window must never reach the database."""
+
+    def _config(self, **overrides):
+        config = {
+            "database": "sysmon",
+            "table": "check_history",
+            "timestamp_column": "ts",
+            "retention_days": 30,
+        }
+        config.update(overrides)
+        return config
+
+    @pytest.mark.parametrize("days", [0, -1, -30, 3651])
+    def test_rejects_out_of_range_retention(self, days, test_settings: Settings):
+        # 0 or negative would delete every row including the one just written.
+        with patch("psycopg2.connect") as connect:
+            with pytest.raises(ValueError):
+                retention._prune(self._config(retention_days=days), test_settings)
+            connect.assert_not_called()
+
+    @pytest.mark.parametrize("days", ["30", None, 1.5, True])
+    def test_rejects_non_integer_retention(self, days, test_settings: Settings):
+        with patch("psycopg2.connect") as connect:
+            with pytest.raises(ValueError):
+                retention._prune(self._config(retention_days=days), test_settings)
+            connect.assert_not_called()
+
+    def test_rejects_injection_in_table_before_connecting(self, test_settings: Settings):
+        with patch("psycopg2.connect") as connect:
+            with pytest.raises(ValueError):
+                retention._prune(
+                    self._config(table="check_history; DELETE FROM check_history --"),
+                    test_settings,
+                )
+            connect.assert_not_called()
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestRetentionBehaviour:
+    """Behaviour against a mocked cursor."""
+
+    def _mock_conn(self, counts):
+        cursor = MagicMock()
+        cursor.fetchone.side_effect = [(c,) for c in counts]
+        cursor.rowcount = counts[0] if counts else 0
+        conn = MagicMock()
+        conn.cursor.return_value.__enter__.return_value = cursor
+        conn.__enter__.return_value = conn
+        return conn, cursor
+
+    def _config(self, **overrides):
+        config = {
+            "database": "sysmon",
+            "table": "check_history",
+            "timestamp_column": "ts",
+            "retention_days": 30,
+        }
+        config.update(overrides)
+        return config
+
+    def test_dry_run_does_not_delete(self, test_settings: Settings):
+        conn, cursor = self._mock_conn([7])
+        with patch("psycopg2.connect", return_value=conn):
+            result = retention._prune(self._config(dry_run=True), test_settings)
+
+        assert "dry run" in result
+        assert "7" in result
+        executed = " ".join(str(c) for c in cursor.execute.call_args_list)
+        assert "DELETE" not in executed.upper()
+
+    def test_no_stale_rows_skips_delete(self, test_settings: Settings):
+        conn, cursor = self._mock_conn([0])
+        with patch("psycopg2.connect", return_value=conn):
+            result = retention._prune(self._config(), test_settings)
+
+        assert "nothing to prune" in result
+        executed = " ".join(str(c) for c in cursor.execute.call_args_list)
+        assert "DELETE" not in executed.upper()
+
+    def test_deletes_and_reports(self, test_settings: Settings):
+        # count(stale) -> 5, then count(remaining) -> 42
+        conn, cursor = self._mock_conn([5, 42])
+        cursor.rowcount = 5
+        with patch("psycopg2.connect", return_value=conn):
+            result = retention._prune(self._config(), test_settings)
+
+        assert "pruned 5 rows" in result
+        assert "42 remain" in result
+        executed = " ".join(str(c) for c in cursor.execute.call_args_list)
+        assert "DELETE" in executed.upper()
+
+    def test_defaults_to_scheduler_database(self, test_settings: Settings):
+        conn, _ = self._mock_conn([0])
+        config = self._config()
+        del config["database"]
+        with patch("psycopg2.connect", return_value=conn) as connect:
+            retention._prune(config, test_settings)
+        assert connect.call_args.kwargs["database"] == test_settings.postgres_db
+
+    @pytest.mark.asyncio
+    async def test_execute_wraps_prune(self, test_settings: Settings):
+        conn, _ = self._mock_conn([0])
+        with patch("psycopg2.connect", return_value=conn):
+            result = await retention.execute(self._config(), test_settings)
+        assert "nothing to prune" in result