chore: release v1.3.0

Carries the two new pruning executors and the POST /tasks fix, which has
been on main unreleased since the image only rebuilds on a version tag.

Also backfills the missing 1.2.0 changelog entry: that version was tagged
and shipped without one.

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

.gitea/workflows/build.yml

@@ -1,12 +1,25 @@
 name: Build and Push
 
 on:
-  release:
-    types: [published]
+  push:
+    tags:
+      - 'v*'
 
 jobs:
+  release:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Create Gitea Release
+        run: |
+          curl -sf -X POST \
+            -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
+            -H "Content-Type: application/json" \
+            -d '{"tag_name": "${{ github.ref_name }}", "name": "Release ${{ github.ref_name }}", "body": "Automated release for ${{ github.ref_name }}"}' \
+            "${{ github.server_url }}/api/v1/repos/${{ github.repository }}/releases"
+
   build:
     runs-on: ubuntu-latest
+    needs: release
     steps:
       - uses: actions/checkout@v4
 
@@ -25,3 +38,9 @@ jobs:
           tags: |
             git.schweitz.net/jpmschweitzer/scheduler:latest
             git.schweitz.net/jpmschweitzer/scheduler:${{ github.ref_name }}
+
+      - name: Trigger Watchtower update
+        if: success()
+        run: |
+          curl -sf -H "Authorization: Bearer ${{ secrets.WATCHTOWER_TOKEN }}" \
+            http://watchtower:8080/v1/update

AGENTS.md

@@ -0,0 +1,72 @@
+
+# AGENTS.md
+
+> **Start every session by reading this file.**
+> This file outlines the operational protocols, coding standards, and architectural decisions for this FastAPI project.
+
+## 1. Agent Operational Protocols
+
+### 🧠 Work Patterns (Plan-Act-Reflect)
+* **Plan:** Before writing code, briefly outline your plan. Identify which files you will touch and what the side effects might be.
+* **Act:** Execute the changes in small, atomic steps.
+* **Reflect:** After coding, verify your work. Did you break existing tests? Did you add new tests?
+
+### 🛡️ Git Discipline
+* **NEVER commit to `main` or `master` directly.** Always create a feature branch: `feature/your-feature-name` or `fix/issue-description`.
+* **Commit Messages:** Use the [Conventional Commits](https://www.conventionalcommits.org/) format.
+    * `feat: add user login endpoint`
+    * `fix: resolve database connection timeout`
+    * `refactor: split monolith dependency file`
+* **Atomic Commits:** Keep commits small. One logical change = one commit.
+
+### 📝 Changelog Maintenance
+* **Update `CHANGELOG.md`** with every user-facing change.
+* Format: `## [Unreleased] - YYYY-MM-DD` followed by `### Added`, `### Changed`, or `### Fixed`.
+
+### 🚀 Release Flow
+When changes are ready for deployment:
+
+1. **Ask user if deploy cycle is desired **
+
+2. **Update version** in `pyproject.toml`:
+   - Bug fixes: bump patch version (1.8.3 → 1.8.4)
+   - New features: bump minor version (1.8.4 → 1.9.0)
+
+3. **Update CHANGELOG.md**:
+   - Move items from `[Unreleased]` to new version section
+   - Add release date: `## [1.8.4] - 2025-12-16`
+
+4. **Commit and tag**:
+   ```bash
+   git add -A
+   git commit -m "fix: description of changes"
+   git tag v1.8.4
+   git push origin main --tags
+   ```
+
+5. **CI/CD triggers automatically**:
+   - Gitea CI builds Docker image on new tag
+   - Watchtower pulls and deploys to production
+   - Verify deployment: `curl http://192.168.86.149:8000/health`
+
+---
+
+## 2. FastAPI Architecture & Best Practices
+*Reference: [FastAPI Best Practices](https://github.com/zhanymkanov/fastapi-best-practices)*
+
+### 📂 Project Structure (Directory-based, NOT File-type based)
+Do **not** group files by type (e.g., one huge `routers` folder). Group by **domain/module** inside a `src/` directory.
+
+**Correct Structure:**
+```text
+src/
+├── auth/
+│   ├── router.py      # Endpoints
+│   ├── schemas.py     # Pydantic models
+│   ├── service.py     # Business logic (CRUD, etc.)
+│   ├── dependencies.py# Module-specific dependencies
+│   └── config.py      # Module-specific settings
+├── posts/
+│   ├── router.py
+│   └── ...
+└── main.py            # App entry point

CHANGELOG.md

@@ -4,6 +4,87 @@ 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.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
+- Test release to validate CI/CD auto-deploy workflow
+
+## [1.1.2] - 2026-01-03
+
+### Fixed
+- CI: Use curl for release creation (release-action requires Go)
+
+## [1.1.1] - 2026-01-03
+
+### Changed
+- CI: Auto-create Gitea release on version tag push (v*) instead of manual release trigger
+
+## [1.1.0] - 2025-12-14
+
+### Added
+- **Gitea Release Cleanup Executor** (`gitea_release_cleanup_executor.py`)
+  - Automatically cleans up old releases across all Gitea repositories
+  - Configurable retention count (default: 5 releases per repo)
+  - Repository exclusion list support
+  - Dry-run mode for safe testing
+  - Designed to run before Watchtower to prevent image tag accumulation
+- **GITEA_TOKEN setting** in config for API token authentication (separate from password)
+
+## [1.0.4] - 2025-12-14
+
+### Fixed
+- CI/CD: Correct Watchtower port (8080)
+
+## [1.0.3] - 2025-12-14
+
+### Added
+- CI/CD: Trigger Watchtower update after successful Docker build
+
+## [1.0.2] - 2025-12-14
+
+### Fixed
+- Removed unused `setup_database.sql` from Dockerfile (database schema managed externally)
+
+## [1.0.1] - 2025-12-14
+
+### Changed
+- **Version tracking now uses pyproject.toml** as single source of truth
+  - Added `pyproject.toml` with project metadata and dependencies
+  - `config.py` reads version from pyproject.toml using `tomllib`
+  - FastAPI app title and version dynamically loaded from config
+  - Health endpoint now includes version in response
+  - Dockerfile updated to include pyproject.toml
+
 ## [1.0.0] - 2025-12-07
 
 ### Added

Dockerfile

@@ -13,7 +13,7 @@ RUN pip install --no-cache-dir -r requirements.txt
 
 # Copy application
 COPY src/ ./src/
-COPY setup_database.sql .
+COPY pyproject.toml .
 
 # Configure git
 RUN git config --global user.name "The Librarian" && \

TASK_REGISTRATION.md

@@ -105,11 +105,57 @@ 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)
+- `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
+}
+```
+
+#### `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

@@ -0,0 +1,47 @@
+[project]
+name = "the-scheduler"
+version = "1.3.0"
+description = "System-wide maintenance orchestration - backups, doc mirroring, cleanup, task automation"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    # FastAPI ecosystem
+    "fastapi~=0.124.0",
+    "uvicorn[standard]~=0.38.0",
+    "pydantic~=2.12.5",
+    "pydantic-settings~=2.7.0",
+    # Task scheduling
+    "apscheduler~=3.11.1",
+    "sqlalchemy~=2.0.36",
+    # Database
+    "psycopg2-binary~=2.9.11",
+    "redis~=5.2.0",
+    # HTTP client
+    "httpx~=0.28.1",
+    # Web scraping (for doc mirroring)
+    "scrapy~=2.12.0",
+    "beautifulsoup4~=4.12.3",
+    "lxml~=5.1.0",
+    # Git operations
+    "gitpython~=3.1.43",
+    # Security/Auth
+    "python-jose[cryptography]~=3.3.0",
+    # Cloud storage
+    "google-cloud-storage~=2.18.0",
+]
+
+[project.optional-dependencies]
+test = [
+    "pytest~=8.3.4",
+    "pytest-asyncio~=0.25.2",
+    "pytest-cov~=6.0.0",
+    "freezegun~=1.5.1",
+]
+
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["src*"]

requirements.txt

@@ -26,6 +26,9 @@ gitpython~=3.1.43       # Latest stable
 # Security/Auth
 python-jose[cryptography]~=3.3.0  # JWT handling
 
+# Cloud storage
+google-cloud-storage~=2.18.0  # GCS offsite backups
+
 # Testing
 pytest~=8.3.4              # Test framework
 pytest-asyncio~=0.25.2     # Async test support

setup_database.sql

@@ -1,120 +0,0 @@
--- The Scheduler Database Setup
--- Run this from postgres-shared container
-
--- Create database and user
-CREATE DATABASE scheduler;
-CREATE USER scheduler_user WITH PASSWORD 'a/Ph0NhC4pTDDjSpL6q/DtBI+z0nf43ijVHjTo1KrXc=';
-GRANT ALL PRIVILEGES ON DATABASE scheduler TO scheduler_user;
-
--- Connect to the new database
-\c scheduler
-
--- Grant schema permissions
-GRANT ALL ON SCHEMA public TO scheduler_user;
-
--- Create scheduled_tasks table
-CREATE TABLE scheduled_tasks (
-    id SERIAL PRIMARY KEY,
-    task_name VARCHAR(100) NOT NULL UNIQUE,
-    service VARCHAR(50) NOT NULL,           -- Which service owns this task
-    executor VARCHAR(100) NOT NULL,         -- Executor module to run
-    priority INTEGER NOT NULL DEFAULT 30,   -- Lower = higher priority (1-100)
-
-    -- Scheduling (supports wildcards: -1 = any)
-    minute INTEGER DEFAULT -1,              -- 0-59 or -1 (any)
-    hour INTEGER DEFAULT -1,                -- 0-23 or -1 (any)
-    day_of_month INTEGER DEFAULT -1,        -- 1-31 or -1 (any)
-    month INTEGER DEFAULT -1,               -- 1-12 or -1 (any)
-    day_of_week INTEGER DEFAULT -1,         -- 0-6 (0=Monday) or -1 (any)
-
-    enabled BOOLEAN DEFAULT true,
-    description TEXT,
-    config JSONB,                           -- Arguments for executor
-
-    -- Execution tracking
-    last_run TIMESTAMP,
-    last_status VARCHAR(20),                -- success, failed, timeout
-    last_duration_seconds INTEGER,
-    retry_count INTEGER DEFAULT 0,
-    max_retries INTEGER DEFAULT 3,
-    timeout_seconds INTEGER DEFAULT 3600,   -- 1 hour default
-
-    -- Metadata
-    created_at TIMESTAMP DEFAULT NOW(),
-    updated_at TIMESTAMP DEFAULT NOW(),
-    created_by VARCHAR(50),
-
-    -- Constraints
-    CHECK (priority >= 1 AND priority <= 100),
-    CHECK (minute >= -1 AND minute <= 59),
-    CHECK (hour >= -1 AND hour <= 23),
-    CHECK (day_of_month >= -1 AND day_of_month <= 31),
-    CHECK (month >= -1 AND month <= 12),
-    CHECK (day_of_week >= -1 AND day_of_week <= 6)
-);
-
--- Create task_executions table
-CREATE TABLE task_executions (
-    id SERIAL PRIMARY KEY,
-    task_id INTEGER NOT NULL REFERENCES scheduled_tasks(id),
-    task_name VARCHAR(100) NOT NULL,
-    service VARCHAR(50) NOT NULL,
-    executor VARCHAR(100) NOT NULL,
-    priority INTEGER NOT NULL,
-
-    status VARCHAR(20) NOT NULL,            -- pending, running, success, failed, timeout
-    triggered_by VARCHAR(50),               -- scheduler, manual, retry
-    triggered_at TIMESTAMP DEFAULT NOW(),
-    started_at TIMESTAMP,
-    completed_at TIMESTAMP,
-    duration_seconds INTEGER,
-
-    output TEXT,
-    error TEXT,
-    retry_count INTEGER DEFAULT 0,
-    metadata JSONB
-);
-
--- Create doc_sources table
-CREATE TABLE doc_sources (
-    id SERIAL PRIMARY KEY,
-    project_name VARCHAR(100) NOT NULL UNIQUE,
-    current_version VARCHAR(50),
-    last_mirrored TIMESTAMP,
-    last_checked TIMESTAMP,
-    gitea_repo VARCHAR(200),
-    config JSONB
-);
-
--- Indexes for scheduled_tasks
-CREATE INDEX idx_tasks_enabled ON scheduled_tasks(enabled) WHERE enabled = true;
-CREATE INDEX idx_tasks_priority ON scheduled_tasks(priority);
-CREATE INDEX idx_tasks_service ON scheduled_tasks(service);
-CREATE INDEX idx_tasks_schedule ON scheduled_tasks(minute, hour, day_of_month, month, day_of_week) WHERE enabled = true;
-
--- Indexes for task_executions
-CREATE INDEX idx_executions_task_id ON task_executions(task_id);
-CREATE INDEX idx_executions_task_name ON task_executions(task_name);
-CREATE INDEX idx_executions_status ON task_executions(status);
-CREATE INDEX idx_executions_triggered_at ON task_executions(triggered_at DESC);
-CREATE INDEX idx_executions_service ON task_executions(service);
-CREATE INDEX idx_executions_running ON task_executions(task_id) WHERE status = 'running';
-
--- Grant permissions
-GRANT ALL ON ALL TABLES IN SCHEMA public TO scheduler_user;
-GRANT ALL ON ALL SEQUENCES IN SCHEMA public TO scheduler_user;
-
--- Insert example test task
-INSERT INTO scheduled_tasks
-    (task_name, service, executor, priority, minute, hour, description, config, created_by)
-VALUES
-    ('test_example_task', 'scheduler', 'example_executor', 50, -1, -1,
-     'Example task that runs every minute for testing',
-     '{"message": "Scheduler is working!", "delay_seconds": 2}'::jsonb,
-     'setup_script');
-
--- Verify tables created
-\dt
-
--- Show the test task
-SELECT task_name, service, priority, enabled, description FROM scheduled_tasks;

src/config.py

@@ -2,17 +2,37 @@
 Configuration management for The Scheduler.
 Uses Pydantic BaseSettings for type-safe environment variable loading.
 """
+import tomllib
 from functools import lru_cache
+from pathlib import Path
 from pydantic_settings import BaseSettings
 from pydantic import Field
 
 
+def _get_version_from_pyproject() -> tuple[str, str, str]:
+    """Read version info from pyproject.toml."""
+    pyproject_path = Path(__file__).parent.parent / "pyproject.toml"
+    if pyproject_path.exists():
+        with open(pyproject_path, "rb") as f:
+            data = tomllib.load(f)
+            project = data.get("project", {})
+            return (
+                project.get("name", "the-scheduler"),
+                project.get("version", "0.0.0"),
+                project.get("description", ""),
+            )
+    return ("the-scheduler", "0.0.0", "")
+
+
+_PROJECT_NAME, _PROJECT_VERSION, _PROJECT_DESCRIPTION = _get_version_from_pyproject()
+
+
 class Settings(BaseSettings):
     """Application settings loaded from environment variables."""
 
     # Application
-    app_name: str = Field(default="The Scheduler", alias="APP_NAME")
-    app_version: str = Field(default="1.0.0", alias="APP_VERSION")
+    app_name: str = Field(default=_PROJECT_NAME, alias="APP_NAME")
+    app_version: str = Field(default=_PROJECT_VERSION, alias="APP_VERSION")
     debug: bool = Field(default=False, alias="DEBUG")
     host: str = Field(default="0.0.0.0", alias="HOST")
     port: int = Field(default=8090, alias="PORT")
@@ -37,6 +57,7 @@ class Settings(BaseSettings):
     gitea_url: str = Field(default="http://gitea:3000", alias="GITEA_URL")
     gitea_user: str = Field(default="library", alias="GITEA_USER")
     gitea_password: str = Field(default="", alias="GITEA_PASSWORD")
+    gitea_token: str = Field(default="", alias="GITEA_TOKEN")  # API token with write:repository scope
     gitea_ssh_host: str = Field(default="gitea", alias="GITEA_SSH_HOST")
     gitea_ssh_port: int = Field(default=22, alias="GITEA_SSH_PORT")
 
@@ -45,6 +66,9 @@ class Settings(BaseSettings):
     backup_retention_weekly: int = Field(default=4, alias="BACKUP_RETENTION_WEEKLY")
     backup_retention_monthly: int = Field(default=12, alias="BACKUP_RETENTION_MONTHLY")
 
+    # Google Cloud Storage
+    gcs_credentials_file: str = Field(default="", alias="GCS_CREDENTIALS_FILE")
+
     # Documentation Mirroring
     docs_mirror_path: str = Field(default="/docs-mirror", alias="DOCS_MIRROR_PATH")
     docs_check_interval: int = Field(default=21600, alias="DOCS_CHECK_INTERVAL")  # 6 hours

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/gcs_backup_executor.py

@@ -0,0 +1,199 @@
+"""
+Google Cloud Storage Backup Executor
+Backs up data to GCS buckets. Supports multiple modes:
+- git_bundle: Creates a git bundle from a bare repo and uploads it
+"""
+import asyncio
+import logging
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional
+
+from google.cloud import storage
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Execute GCS backup task.
+
+    Config schema:
+        {
+            "mode": "git_bundle",
+            "bucket": "bucket-name",
+            "prefix": "gitea/settled-reach",
+            "credentials_path": "/secrets/gcs-sa-key.json",
+            "repo_path": "/data/docker-data/gitea/data/git/repositories/user/repo.git",
+            "retention_count": 7,
+            "dry_run": false
+        }
+
+    Args:
+        config: Backup configuration
+        settings: Global scheduler settings
+
+    Returns:
+        Summary of backup operation
+
+    Raises:
+        Exception: On backup failure
+    """
+    mode = config.get('mode')
+    if not mode:
+        raise ValueError("Missing required config field: mode")
+
+    if mode == 'git_bundle':
+        return await _mode_git_bundle(config, settings)
+    else:
+        raise ValueError(f"Unknown backup mode: {mode}")
+
+
+async def _mode_git_bundle(config: dict, settings: Settings) -> str:
+    """Create a git bundle from a bare repo and upload to GCS."""
+    bucket_name = config.get('bucket')
+    prefix = config.get('prefix', '').strip('/')
+    credentials_path = config.get('credentials_path', settings.gcs_credentials_file)
+    repo_path = Path(config.get('repo_path', ''))
+    retention_count = config.get('retention_count', 7)
+    dry_run = config.get('dry_run', False)
+
+    # Validate
+    if not bucket_name:
+        raise ValueError("Missing required config field: bucket")
+    if not repo_path or not str(repo_path).strip():
+        raise ValueError("Missing required config field: repo_path")
+    if not repo_path.exists():
+        raise ValueError(f"Repository path does not exist: {repo_path}")
+    if not (repo_path / 'HEAD').exists():
+        raise ValueError(f"Not a valid git repository: {repo_path}")
+
+    repo_name = repo_path.name.removesuffix('.git')
+    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
+    bundle_filename = f"{repo_name}-{timestamp}.bundle"
+    bundle_path = Path(f"/tmp/{bundle_filename}")
+
+    logger.info(f"Starting GCS backup: mode=git_bundle, repo={repo_name}, dry_run={dry_run}")
+
+    try:
+        # Step 1: Create git bundle
+        logger.info(f"Creating git bundle from {repo_path}")
+        start = time.monotonic()
+
+        await _run_command([
+            'git', 'bundle', 'create',
+            str(bundle_path),
+            '--all'
+        ], cwd=repo_path)
+
+        bundle_duration = time.monotonic() - start
+
+        if not bundle_path.exists():
+            raise Exception("Git bundle was not created")
+
+        bundle_size = bundle_path.stat().st_size
+        bundle_size_mb = bundle_size / (1024 * 1024)
+        logger.info(f"Bundle created: {bundle_filename} ({bundle_size_mb:.1f} MB) in {bundle_duration:.1f}s")
+
+        # Step 2: Verify bundle
+        await _run_command([
+            'git', 'bundle', 'verify',
+            str(bundle_path)
+        ], cwd=repo_path)
+        logger.info("Bundle verified OK")
+
+        if dry_run:
+            return (
+                f"[DRY RUN] Would upload {bundle_filename} ({bundle_size_mb:.1f} MB) "
+                f"to gs://{bucket_name}/{prefix}/{bundle_filename}"
+            )
+
+        # Step 3: Upload to GCS
+        gcs_path = f"{prefix}/{bundle_filename}" if prefix else bundle_filename
+        logger.info(f"Uploading to gs://{bucket_name}/{gcs_path}")
+        start = time.monotonic()
+
+        client = storage.Client.from_service_account_json(credentials_path)
+        bucket = client.bucket(bucket_name)
+        blob = bucket.blob(gcs_path)
+        blob.upload_from_filename(str(bundle_path), timeout=3600)
+
+        upload_duration = time.monotonic() - start
+        logger.info(f"Upload complete in {upload_duration:.1f}s")
+
+        # Step 4: Retention cleanup in GCS
+        deleted_count = await _cleanup_gcs_retention(
+            client, bucket_name, prefix, repo_name, retention_count
+        )
+
+        return (
+            f"Backup completed: {bundle_filename} ({bundle_size_mb:.1f} MB). "
+            f"Bundle: {bundle_duration:.1f}s, Upload: {upload_duration:.1f}s. "
+            f"GCS: gs://{bucket_name}/{gcs_path}. "
+            f"Retention: {deleted_count} old bundle(s) removed."
+        )
+
+    finally:
+        # Always clean up the local temp file
+        if bundle_path.exists():
+            bundle_path.unlink()
+            logger.debug(f"Cleaned up temp file: {bundle_path}")
+
+
+async def _cleanup_gcs_retention(
+    client: storage.Client,
+    bucket_name: str,
+    prefix: str,
+    repo_name: str,
+    retention_count: int
+) -> int:
+    """Delete old bundles from GCS, keeping only the most recent retention_count."""
+    if retention_count <= 0:
+        return 0
+
+    bucket = client.bucket(bucket_name)
+    blob_prefix = f"{prefix}/{repo_name}-" if prefix else f"{repo_name}-"
+
+    blobs = list(bucket.list_blobs(prefix=blob_prefix))
+    bundle_blobs = [b for b in blobs if b.name.endswith('.bundle')]
+
+    if len(bundle_blobs) <= retention_count:
+        logger.info(f"Retention OK: {len(bundle_blobs)} bundles (limit: {retention_count})")
+        return 0
+
+    # Sort by name (timestamp in name ensures chronological order)
+    bundle_blobs.sort(key=lambda b: b.name)
+    to_delete = bundle_blobs[:-retention_count]
+
+    for blob in to_delete:
+        logger.info(f"Deleting old bundle: {blob.name}")
+        blob.delete()
+
+    logger.info(f"Retention cleanup: deleted {len(to_delete)} old bundle(s)")
+    return len(to_delete)
+
+
+async def _run_command(
+    cmd: List[str],
+    cwd: Optional[Path] = None,
+) -> str:
+    """Run shell command asynchronously."""
+    logger.debug(f"Running: {' '.join(cmd)} (cwd: {cwd})")
+
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        cwd=cwd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE
+    )
+
+    stdout, stderr = await proc.communicate()
+
+    if proc.returncode != 0:
+        error_msg = stderr.decode() if stderr else "Unknown error"
+        raise Exception(f"Command failed: {' '.join(cmd)}\n{error_msg}")
+
+    return stdout.decode()

src/executors/gitea_release_cleanup_executor.py

@@ -0,0 +1,211 @@
+"""
+Gitea Release Cleanup Executor
+
+Cleans up old releases across all Gitea repositories, keeping only the most recent
+releases per repository. Designed to run before Watchtower to prevent accumulation
+of old container image tags.
+
+Config schema:
+    {
+        "keep_count": 5,           # Number of releases to keep per repo (default: 5)
+        "exclude_repos": [],       # Repository names to skip (default: [])
+        "dry_run": false           # If true, only log what would be deleted (default: false)
+    }
+
+Example config:
+    {
+        "keep_count": 5,
+        "exclude_repos": ["important-repo", "legacy-app"],
+        "dry_run": false
+    }
+
+Required settings:
+    - GITEA_URL: Base URL of Gitea instance
+    - GITEA_TOKEN: API token with write:repository scope
+"""
+import logging
+from typing import Any
+
+import httpx
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+# Default configuration values
+DEFAULT_KEEP_COUNT = 5
+DEFAULT_TIMEOUT = 60
+API_BASE = "/api/v1"
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Clean up old releases across all accessible Gitea repositories.
+
+    Args:
+        config: Task configuration (see module docstring)
+        settings: Global scheduler settings
+
+    Returns:
+        Summary of cleanup actions taken
+
+    Raises:
+        ValueError: On configuration error
+        Exception: On API call failure
+    """
+    # Validate settings
+    if not settings.gitea_url:
+        raise ValueError("GITEA_URL not configured")
+    if not settings.gitea_token:
+        raise ValueError("GITEA_TOKEN not configured - required for release deletion")
+
+    # Parse configuration
+    keep_count = config.get("keep_count", DEFAULT_KEEP_COUNT)
+    exclude_repos = config.get("exclude_repos", [])
+    dry_run = config.get("dry_run", False)
+
+    if keep_count < 1:
+        raise ValueError(f"keep_count must be at least 1, got {keep_count}")
+
+    base_url = settings.gitea_url.rstrip("/")
+    headers = {"Authorization": f"token {settings.gitea_token}"}
+
+    stats = {
+        "repos_scanned": 0,
+        "repos_with_releases": 0,
+        "releases_deleted": 0,
+        "releases_skipped": 0,
+        "errors": [],
+    }
+
+    mode = "DRY RUN" if dry_run else "LIVE"
+    logger.info(f"Starting Gitea release cleanup ({mode}): keeping {keep_count} releases per repo")
+
+    async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+        # Fetch all repositories
+        repos = await _fetch_user_repos(client, base_url, headers)
+        logger.info(f"Found {len(repos)} repositories")
+
+        for repo in repos:
+            owner = repo["owner"]["login"]
+            name = repo["name"]
+            full_name = f"{owner}/{name}"
+
+            # Check exclusion list
+            if name in exclude_repos or full_name in exclude_repos:
+                logger.debug(f"Skipping excluded repo: {full_name}")
+                continue
+
+            stats["repos_scanned"] += 1
+
+            try:
+                # Fetch releases for this repo
+                releases = await _fetch_releases(client, base_url, headers, owner, name)
+
+                if not releases:
+                    continue
+
+                stats["repos_with_releases"] += 1
+
+                # Determine which releases to delete (beyond keep_count)
+                to_delete = releases[keep_count:]
+
+                if not to_delete:
+                    logger.debug(f"{full_name}: {len(releases)} releases, nothing to delete")
+                    continue
+
+                logger.info(f"{full_name}: {len(releases)} releases, deleting {len(to_delete)}")
+
+                # Delete old releases
+                for release in to_delete:
+                    release_id = release["id"]
+                    tag_name = release["tag_name"]
+
+                    if dry_run:
+                        logger.info(f"  [DRY RUN] Would delete: {tag_name} (id={release_id})")
+                        stats["releases_skipped"] += 1
+                    else:
+                        try:
+                            await _delete_release(client, base_url, headers, owner, name, release_id)
+                            logger.info(f"  Deleted: {tag_name}")
+                            stats["releases_deleted"] += 1
+                        except Exception as e:
+                            error_msg = f"{full_name}/{tag_name}: {e}"
+                            logger.warning(f"  Failed to delete {tag_name}: {e}")
+                            stats["errors"].append(error_msg)
+
+            except Exception as e:
+                error_msg = f"{full_name}: {e}"
+                logger.error(f"Error processing {full_name}: {e}")
+                stats["errors"].append(error_msg)
+
+    # Build summary
+    summary = _build_summary(stats, dry_run)
+    logger.info(f"Cleanup complete: {summary}")
+
+    return summary
+
+
+async def _fetch_user_repos(
+    client: httpx.AsyncClient,
+    base_url: str,
+    headers: dict,
+) -> list[dict[str, Any]]:
+    """Fetch all repositories accessible to the authenticated user."""
+    url = f"{base_url}{API_BASE}/user/repos"
+    params = {"limit": 100}
+
+    response = await client.get(url, headers=headers, params=params)
+    response.raise_for_status()
+
+    return response.json()
+
+
+async def _fetch_releases(
+    client: httpx.AsyncClient,
+    base_url: str,
+    headers: dict,
+    owner: str,
+    repo: str,
+) -> list[dict[str, Any]]:
+    """Fetch releases for a repository, sorted newest first (Gitea default)."""
+    url = f"{base_url}{API_BASE}/repos/{owner}/{repo}/releases"
+    params = {"limit": 100}
+
+    response = await client.get(url, headers=headers, params=params)
+    response.raise_for_status()
+
+    return response.json()
+
+
+async def _delete_release(
+    client: httpx.AsyncClient,
+    base_url: str,
+    headers: dict,
+    owner: str,
+    repo: str,
+    release_id: int,
+) -> None:
+    """Delete a specific release."""
+    url = f"{base_url}{API_BASE}/repos/{owner}/{repo}/releases/{release_id}"
+
+    response = await client.delete(url, headers=headers)
+    response.raise_for_status()
+
+
+def _build_summary(stats: dict, dry_run: bool) -> str:
+    """Build a human-readable summary of the cleanup operation."""
+    parts = [
+        f"Scanned {stats['repos_scanned']} repos",
+        f"{stats['repos_with_releases']} with releases",
+    ]
+
+    if dry_run:
+        parts.append(f"{stats['releases_skipped']} releases would be deleted")
+    else:
+        parts.append(f"{stats['releases_deleted']} releases deleted")
+
+    if stats["errors"]:
+        parts.append(f"{len(stats['errors'])} errors")
+
+    return ", ".join(parts)

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/main.py

@@ -102,9 +102,10 @@ async def lifespan(app: FastAPI):
 
 
 # FastAPI app
+settings = get_settings()
 app = FastAPI(
-    title="The Scheduler",
-    version="1.0.0",
+    title=settings.app_name,
+    version=settings.app_version,
     description="System-wide maintenance orchestration - backups, doc mirroring, cleanup, task automation",
     lifespan=lifespan
 )
@@ -134,6 +135,7 @@ async def health(
     """Health check endpoint."""
     return {
         "status": "healthy",
+        "version": settings.app_version,
         "scheduler_running": sched.running,
         "jobs_count": len(sched.get_jobs()),
         "database": settings.postgres_db

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_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