feat(ansible): scheduled recurring playbook runs

Adds cron-like recurring runs (the engine the maintenance-automation work
needs). New AnsibleSchedule model + migration 0018; a core ScheduledTask
(ansible_scheduled_runs, 60s) fires due schedules, each creating a
system-triggered AnsibleRun (triggered_by=None). Centralises the
resolve-inventory → create-run → launch flow in ansible/runner.trigger_run,
shared by the manual route (refactored to use it) and the scheduler.

Schedules UI under /ansible/schedules: create/edit/pause/delete/run-now,
with interval presets, scope targeting (all / group / target), extra-vars /
limit / tags / dry-run, and last-run status (resolved via last_run_id) +
next-run. Unit test for the due-check.

Task #549 (milestone #37).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

steward/ansible/routes.py

@@ -1,20 +1,28 @@
 # steward/ansible/routes.py
 from __future__ import annotations
-import asyncio
-import uuid
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 from quart import (
-    Blueprint, Response, current_app, render_template,
-    request, session,
+    Blueprint, Response, current_app, redirect, render_template,
+    request, session, url_for,
 )
 from sqlalchemy import select
 
 from steward.ansible import executor, sources as src_module
 from steward.auth.middleware import require_role
 from steward.models.ansible import AnsibleRun, AnsibleRunStatus
+from steward.models.ansible_schedule import AnsibleSchedule
 from steward.models.users import User, UserRole
 
+INTERVAL_PRESETS = [
+    (300, "Every 5 minutes"),
+    (3600, "Hourly"),
+    (21600, "Every 6 hours"),
+    (43200, "Every 12 hours"),
+    (86400, "Daily"),
+    (604800, "Weekly"),
+]
+
 ansible_bp = Blueprint("ansible", __name__, url_prefix="/ansible")
 
 
@@ -85,11 +93,34 @@ async def view_playbook(source_name: str, playbook_path: str):
     )
 
 
+def _parse_run_params(form) -> tuple[dict | None, str | None]:
+    """Build the executor params dict from form fields. Returns (params, error)."""
+    extra_vars: list[str] = []
+    for line in (form.get("extra_vars", "") or "").splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        if "=" not in line:
+            return None, f"Invalid extra var (expected key=value): {line!r}"
+        extra_vars.append(line)
+    params: dict = {}
+    if extra_vars:
+        params["extra_vars"] = extra_vars
+    limit = (form.get("limit", "") or "").strip()
+    if limit:
+        params["limit"] = limit
+    tags = (form.get("tags", "") or "").strip()
+    if tags:
+        params["tags"] = tags
+    if "check" in form:
+        params["check"] = True
+    return (params or None), None
+
+
 @ansible_bp.post("/runs")
 @require_role(UserRole.operator)
 async def create_run():
-    import json as _json
-    from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory
+    from steward.ansible.runner import trigger_run
 
     form = await request.form
     playbook_path = form.get("playbook_path", "").strip()
@@ -99,82 +130,22 @@ async def create_run():
     if not playbook_path:
         return "playbook_path is required", 400
 
-    all_sources = _get_sources()
-    source = next((s for s in all_sources if s["name"] == source_name), None)
-    if source is None:
-        return "Source not found", 404
+    params_or_none, err = _parse_run_params(form)
+    if err:
+        return err, 400
 
-    # Resolve inventory for this scope.
-    inventory_content: str | None = None
-    inventory_path: str | None = None
-    if inventory_scope.startswith("steward:"):
-        async with current_app.db_sessionmaker() as db:
-            targets = await fetch_scope_targets(db, inventory_scope)
-        inventory_content = _json.dumps(generate_inventory(targets))
-    elif inventory_scope.startswith("repo:"):
-        parts = inventory_scope.split(":", 2)
-        inventory_path = parts[2] if len(parts) == 3 else ""
-    else:
-        return "Invalid inventory_scope", 400
-
-    # Optional run parameters (all passed as argv by the executor — no shell).
-    extra_vars: list[str] = []
-    for line in (form.get("extra_vars", "") or "").splitlines():
-        line = line.strip()
-        if not line:
-            continue
-        if "=" not in line:
-            return f"Invalid extra var (expected key=value): {line!r}", 400
-        extra_vars.append(line)
-    limit = (form.get("limit", "") or "").strip()
-    tags = (form.get("tags", "") or "").strip()
-    check = "check" in form
-
-    params: dict = {}
-    if extra_vars:
-        params["extra_vars"] = extra_vars
-    if limit:
-        params["limit"] = limit
-    if tags:
-        params["tags"] = tags
-    if check:
-        params["check"] = True
-    params_or_none = params or None
-
-    run_id = str(uuid.uuid4())
-    now = datetime.now(timezone.utc)
-
-    run = AnsibleRun(
-        id=run_id,
-        playbook_path=playbook_path,
-        inventory_path=inventory_path,
-        inventory_scope=inventory_scope,
+    run, source, err = await trigger_run(
+        current_app._get_current_object(),  # type: ignore[attr-defined]
         source_name=source_name,
-        triggered_by=session["user_id"],
-        status=AnsibleRunStatus.running,
-        started_at=now,
+        playbook_path=playbook_path,
+        inventory_scope=inventory_scope,
         params=params_or_none,
+        triggered_by=session["user_id"],
     )
-    async with current_app.db_sessionmaker() as db:
-        async with db.begin():
-            db.add(run)
-
-    task = asyncio.create_task(
-        executor.start_run(
-            current_app._get_current_object(),  # type: ignore[attr-defined]
-            run_id,
-            playbook_path,
-            inventory_path or "",
-            source["path"],
-            params_or_none,
-            inventory_content,
-        )
-    )
-    task.add_done_callback(
-        lambda t: t.exception() and current_app.logger.error(
-            "Ansible run %s raised: %s", run_id, t.exception()
-        )
-    )
+    if err == "Source not found":
+        return err, 404
+    if err:
+        return err, 400
 
     return await render_template(
         "ansible/run_started.html",
@@ -233,3 +204,153 @@ async def run_detail(run_id: str):
             triggered_label = res.scalar_one_or_none() or "(deleted user)"
     return await render_template(
         "ansible/run_detail.html", run=run, triggered_label=triggered_label)
+
+
+# ── Scheduled recurring runs ──────────────────────────────────────────────────
+
+@ansible_bp.get("/schedules")
+@require_role(UserRole.viewer)
+async def schedules():
+    from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
+
+    editing_id = request.args.get("edit")
+    async with current_app.db_sessionmaker() as db:
+        scheds = (await db.execute(
+            select(AnsibleSchedule).order_by(AnsibleSchedule.name))).scalars().all()
+        groups = (await db.execute(
+            select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()
+        targets = (await db.execute(
+            select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
+        run_ids = [s.last_run_id for s in scheds if s.last_run_id]
+        runs = {
+            r.id: r for r in (await db.execute(
+                select(AnsibleRun).where(AnsibleRun.id.in_(run_ids)))).scalars().all()
+        } if run_ids else {}
+
+    source_data = [
+        {"name": s["name"], "playbooks": src_module.discover_playbooks(s["path"])}
+        for s in _get_sources()
+    ]
+    all_playbooks = sorted({p for sd in source_data for p in sd["playbooks"]})
+
+    rows = []
+    editing = None
+    for s in scheds:
+        rows.append({
+            "s": s,
+            "last_run": runs.get(s.last_run_id) if s.last_run_id else None,
+            "next_run": (s.last_run_at + timedelta(seconds=s.interval_seconds)) if s.last_run_at else None,
+        })
+        if editing_id and s.id == editing_id:
+            editing = s
+
+    return await render_template(
+        "ansible/schedules.html",
+        rows=rows, groups=groups, targets=targets,
+        source_data=source_data, all_playbooks=all_playbooks,
+        editing=editing, interval_presets=INTERVAL_PRESETS,
+    )
+
+
+def _schedule_form_fields(form) -> tuple[dict | None, str | None]:
+    """Validate + extract schedule fields from a form. Returns (fields, error)."""
+    name = (form.get("name", "") or "").strip()
+    source_name = (form.get("source_name", "") or "").strip()
+    playbook_path = (form.get("playbook_path", "") or "").strip()
+    inventory_scope = (form.get("inventory_scope", "steward:all") or "steward:all").strip()
+    try:
+        interval = int(form.get("interval_seconds", "0"))
+    except (TypeError, ValueError):
+        interval = 0
+    if not name or not source_name or not playbook_path or interval <= 0:
+        return None, "name, source, playbook, and a positive interval are required"
+    params, err = _parse_run_params(form)
+    if err:
+        return None, err
+    return {
+        "name": name, "source_name": source_name, "playbook_path": playbook_path,
+        "inventory_scope": inventory_scope, "interval_seconds": interval, "params": params,
+    }, None
+
+
+@ansible_bp.post("/schedules")
+@require_role(UserRole.operator)
+async def create_schedule():
+    form = await request.form
+    fields, err = _schedule_form_fields(form)
+    if err:
+        return err, 400
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            db.add(AnsibleSchedule(enabled=True, **fields))
+    return redirect(url_for("ansible.schedules"))
+
+
+@ansible_bp.post("/schedules/<schedule_id>")
+@require_role(UserRole.operator)
+async def update_schedule(schedule_id: str):
+    form = await request.form
+    fields, err = _schedule_form_fields(form)
+    if err:
+        return err, 400
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            sched = await db.get(AnsibleSchedule, schedule_id)
+            if sched is None:
+                return "Not found", 404
+            for k, v in fields.items():
+                setattr(sched, k, v)
+    return redirect(url_for("ansible.schedules"))
+
+
+@ansible_bp.post("/schedules/<schedule_id>/toggle")
+@require_role(UserRole.operator)
+async def toggle_schedule(schedule_id: str):
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            sched = await db.get(AnsibleSchedule, schedule_id)
+            if sched is None:
+                return "Not found", 404
+            sched.enabled = not sched.enabled
+    return redirect(url_for("ansible.schedules"))
+
+
+@ansible_bp.post("/schedules/<schedule_id>/delete")
+@require_role(UserRole.operator)
+async def delete_schedule(schedule_id: str):
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            sched = await db.get(AnsibleSchedule, schedule_id)
+            if sched is not None:
+                await db.delete(sched)
+    return redirect(url_for("ansible.schedules"))
+
+
+@ansible_bp.post("/schedules/<schedule_id>/run-now")
+@require_role(UserRole.operator)
+async def run_schedule_now(schedule_id: str):
+    from steward.ansible.runner import trigger_run
+
+    async with current_app.db_sessionmaker() as db:
+        sched = await db.get(AnsibleSchedule, schedule_id)
+        if sched is None:
+            return "Not found", 404
+        source_name, playbook_path = sched.source_name, sched.playbook_path
+        inventory_scope, params = sched.inventory_scope, sched.params
+
+    run, _source, err = await trigger_run(
+        current_app._get_current_object(),  # type: ignore[attr-defined]
+        source_name=source_name, playbook_path=playbook_path,
+        inventory_scope=inventory_scope, params=params,
+        triggered_by=session["user_id"],
+    )
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            fresh = await db.get(AnsibleSchedule, schedule_id)
+            if fresh:
+                fresh.last_run_at = datetime.now(timezone.utc)
+                fresh.last_run_id = run.id if run else None
+                fresh.last_error = err
+    if run:
+        return redirect(url_for("ansible.run_detail", run_id=run.id))
+    return redirect(url_for("ansible.schedules"))

steward/ansible/runner.py

@@ -0,0 +1,78 @@
+# steward/ansible/runner.py
+"""Shared playbook-run launcher used by the manual route, alerts, and schedules.
+
+Centralises the resolve-inventory → create AnsibleRun → launch executor flow so
+manual, alert-triggered, and scheduled runs all take the exact same path.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from datetime import datetime, timezone
+
+from steward.ansible import executor, sources as src_module
+from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory
+from steward.models.ansible import AnsibleRun, AnsibleRunStatus
+
+
+async def trigger_run(
+    app,
+    *,
+    source_name: str,
+    playbook_path: str,
+    inventory_scope: str = "steward:all",
+    params: dict | None = None,
+    triggered_by: str | None = None,
+):
+    """Resolve inventory for the scope, persist an AnsibleRun, and launch it.
+
+    triggered_by=None marks a system/automated run (alerts, schedules).
+    Returns (run, source, error): on success error is None; on failure run is
+    None and error is a short human-readable reason.
+    """
+    sources = src_module.get_sources(app.config.get("ANSIBLE", {}))
+    source = next((s for s in sources if s["name"] == source_name), None)
+    if source is None:
+        return None, None, "Source not found"
+
+    inventory_content: str | None = None
+    inventory_path: str | None = None
+    if inventory_scope.startswith("steward:"):
+        async with app.db_sessionmaker() as db:
+            targets = await fetch_scope_targets(db, inventory_scope)
+        inventory_content = json.dumps(generate_inventory(targets))
+    elif inventory_scope.startswith("repo:"):
+        parts = inventory_scope.split(":", 2)
+        inventory_path = parts[2] if len(parts) == 3 else ""
+    else:
+        return None, source, "Invalid inventory_scope"
+
+    run_id = str(uuid.uuid4())
+    run = AnsibleRun(
+        id=run_id,
+        playbook_path=playbook_path,
+        inventory_path=inventory_path,
+        inventory_scope=inventory_scope,
+        source_name=source_name,
+        triggered_by=triggered_by,
+        status=AnsibleRunStatus.running,
+        started_at=datetime.now(timezone.utc),
+        params=params or None,
+    )
+    async with app.db_sessionmaker() as db:
+        async with db.begin():
+            db.add(run)
+
+    task = asyncio.create_task(
+        executor.start_run(
+            app, run_id, playbook_path, inventory_path or "",
+            source["path"], params or None, inventory_content,
+        )
+    )
+    task.add_done_callback(
+        lambda t: t.exception() and app.logger.error(
+            "Ansible run %s raised: %s", run_id, t.exception()
+        )
+    )
+    return run, source, None

steward/ansible/scheduler.py

@@ -0,0 +1,66 @@
+# steward/ansible/scheduler.py
+"""Fire due Ansible schedules. Driven by the core ScheduledTask loop (~60s)."""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+
+from steward.ansible.runner import trigger_run
+from steward.models.ansible_schedule import AnsibleSchedule
+
+logger = logging.getLogger(__name__)
+
+
+def is_due(schedule: AnsibleSchedule, now: datetime) -> bool:
+    """True when an enabled schedule has never run or its interval has elapsed.
+
+    Cadence is measured from the previous *fire* time (last_run_at), not run
+    completion, so a long-running playbook doesn't cause catch-up storms.
+    """
+    if not schedule.enabled:
+        return False
+    if schedule.last_run_at is None:
+        return True
+    return (now - schedule.last_run_at).total_seconds() >= schedule.interval_seconds
+
+
+async def run_due_schedules(app) -> int:
+    """Trigger every due schedule once. Returns the number launched."""
+    now = datetime.now(timezone.utc)
+    async with app.db_sessionmaker() as db:
+        schedules = (await db.execute(
+            select(AnsibleSchedule).where(AnsibleSchedule.enabled.is_(True))
+        )).scalars().all()
+
+    fired = 0
+    for s in schedules:
+        if not is_due(s, now):
+            continue
+        run = None
+        err: str | None = None
+        try:
+            run, _source, err = await trigger_run(
+                app,
+                source_name=s.source_name,
+                playbook_path=s.playbook_path,
+                inventory_scope=s.inventory_scope,
+                params=s.params,
+                triggered_by=None,
+            )
+        except Exception:
+            logger.exception("Schedule %s (%s) failed to launch", s.id, s.name)
+            err = "launch error"
+
+        async with app.db_sessionmaker() as db:
+            async with db.begin():
+                fresh = await db.get(AnsibleSchedule, s.id)
+                if fresh:
+                    fresh.last_run_at = now
+                    fresh.last_run_id = run.id if run else None
+                    fresh.last_error = err
+        if run:
+            fired += 1
+            logger.info("Schedule %s (%s) fired run %s", s.id, s.name, run.id)
+    return fired