feat(mcp): automation tools — playbooks, runs, schedules, alert actions, docker prune

MCP-6 #4970 "Automation tools: playbooks, runs, schedules, alert actions".

Viewer: list_playbooks, get_playbook (secret defaults masked), list_inventory
(variable names only), list_ansible_runs, get_ansible_run (log tail + grep),
list_schedules, wait_for_run (polls, 120s cap).

Operator: run_playbook (check mode unless check=false; a steward:confirm
playbook applies only with confirmed=true; secret variables reach the run
but are never stored, audited or returned), cancel_ansible_run, and
create/update/delete_schedule (secrets refused, check by default).

Admin: set_alert_action, and the docker plugin's prune_docker, which
previews from the disk report unless check=false (Ansible's check mode
skips the prune commands, so a check run would show nothing).

Every write is audited as "<user> via MCP" with the token id; the audit
tool's user filter includes those rows. prune_extra_vars moves to the
docker service so the page and the tool share it.

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

plugins/docker/mcp_tools.py

@@ -11,11 +11,15 @@ from datetime import datetime, timedelta
 
 from sqlalchemy import select
 
-from steward.mcp._context import mcp_app
+from steward.ansible.runner import trigger_run
+from steward.ansible.sources import BUILTIN_SOURCE_NAME
+from steward.mcp._context import current_caller, mcp_app
+from steward.mcp.tools._audit import audit_mcp
 from steward.mcp.tools._shape import (
     age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, pick, rnd, summarize,
 )
 from steward.mcp.tools.overview import finding
+from steward.models.ansible_inventory import AnsibleTarget
 from steward.models.hosts import Host
 from steward.models.users import UserRole
 
@@ -412,7 +416,81 @@ async def get_docker_disk_usage(host: str = "") -> dict:
     return {"hosts": rows, "total": len(rows)}
 
 
+PRUNE_BUILTIN_PLAYBOOK = "maintenance/docker_prune.yml"
+_STOPPED = ("exited", "created", "dead")
+
+
+def prune_preview(what: str, usage, stopped: int) -> dict:
+    """What a prune of `what` would remove, from the host's latest disk report."""
+    out: dict = {}
+    if what in ("containers", "system"):
+        out["stopped_containers"] = stopped
+    if what == "images":
+        out["unused_images_bytes"] = usage.images_reclaimable if usage else None
+    if what == "system":
+        out["dangling_images"] = "yes (unused but tagged images stay)"
+        out["build_cache_bytes"] = usage.build_cache_size if usage else None
+        out["unused_networks"] = "yes"
+    return out
+
+
+async def prune_docker(host: str, what: str, check: bool = True) -> dict:
+    """Reclaim Docker disk on a host. Previews unless check=false.
+
+    host: the Steward host's name or id. It needs a linked Ansible target
+      (list_inventory shows the links); the prune runs through Ansible as
+      Steward's built-in docker_prune playbook.
+    what: "containers" removes stopped containers; "images" removes ALL
+      unused images, not only dangling ones; "system" is `docker system
+      prune -f`: stopped containers, dangling images, unused networks and
+      build cache. Volumes are never pruned.
+    check: true (default) starts nothing and returns `would_remove`, sized
+      from the host's latest Docker disk report. Ansible's own check mode
+      skips the prune commands, so a check-mode run would show nothing.
+      Pass false to run the prune.
+
+    Admin only, as on the Docker → Disk page. A real prune returns `run_id`
+    (follow it with wait_for_run) and is recorded in the audit log.
+    """
+    what = (what or "").strip()
+    if what not in docker_svc.PRUNE_TARGETS:
+        raise ValueError(f"what {what!r} is not one of {', '.join(docker_svc.PRUNE_TARGETS)}.")
+    async with db_session() as db:
+        h = pick(host, (await db.execute(select(Host))).scalars(), kind="host", list_tool="list_hosts")
+        target = (await db.execute(
+            select(AnsibleTarget).where(AnsibleTarget.host_id == h.id))).scalars().first()
+        usage = await db.get(DockerDiskUsage, h.id)
+        stopped = sum(1 for c in await docker_svc.list_containers(db, h.id)
+                      if (c.status or "").lower() in _STOPPED)
+    if target is None:
+        raise ValueError(f"host {h.name!r} has no linked Ansible target, so Steward cannot reach it "
+                         "to prune. Link one under Ansible → Inventory. Nothing was run.")
+    preview = {
+        "host": h.name,
+        "what": what,
+        "ansible_target": target.name,
+        "would_remove": prune_preview(what, usage, stopped),
+        "disk_report_age_s": age_s(usage.updated_at) if usage else None,
+    }
+    if check:
+        return {**preview, "check": True,
+                "summary": f"Nothing was run. Pass check=false to prune {what} on {h.name}."}
+    caller = current_caller()
+    run, _src, err = await trigger_run(
+        mcp_app(), source_name=BUILTIN_SOURCE_NAME, playbook_path=PRUNE_BUILTIN_PLAYBOOK,
+        inventory_scope=f"steward:target:{target.id}",
+        params={"extra_vars_map": docker_svc.prune_extra_vars(what)},
+        triggered_by=caller.user_id)
+    if err:
+        raise ValueError(f"the prune could not start: {err}.")
+    await audit_mcp("docker.prune.started", "host", h.name,
+                    {"host_id": h.id, "what": what, "run_id": run.id})
+    return {**preview, "check": False, "run_id": run.id,
+            "next": f'wait_for_run(run_id="{run.id}")'}
+
+
 def register_tools(register) -> None:
     for fn in (list_containers, get_container, list_swarm_services,
                search_container_logs, get_docker_disk_usage):
         register(fn, role=UserRole.viewer, read_only=True)
+    register(prune_docker, role=UserRole.admin, read_only=False)

plugins/docker/routes.py

@@ -31,17 +31,6 @@ def _error(status: int, code: str, detail: str | None = None):
     return jsonify(body), status
 
 
-def _prune_extra_vars(prune_target: str) -> dict:
-    """Extra-vars for the bundled docker_prune playbook. "Prune unused images"
-    means ALL unused (docker image prune -a), not just dangling; the full system
-    prune stays conservative (-f, no -a) per the operator's choice (m78). Pure
-    helper so the mapping is unit-tested without the request/DB stack."""
-    extra_vars: dict = {"prune_target": prune_target}
-    if prune_target == "images":
-        extra_vars["prune_all_images"] = True
-    return extra_vars
-
-
 def _human_bytes(n: int | None) -> str:
     """Compact binary size string (e.g. '1.4 GiB', '512 MiB', '0 B')."""
     if n is None:
@@ -544,7 +533,7 @@ async def disk_prune(host_id: str):
 
     form = await request.form
     prune_target = (form.get("target", "") or "").strip()
-    if prune_target not in ("containers", "images", "system"):
+    if prune_target not in docker_svc.PRUNE_TARGETS:
         return _error(400, "bad_target", "Unknown prune target")
 
     async with current_app.db_sessionmaker() as db:
@@ -556,7 +545,7 @@ async def disk_prune(host_id: str):
                       "Link an Ansible target to this host before pruning")
 
     # extra_vars_map outranks the playbook's own `vars:` defaults.
-    extra_vars = _prune_extra_vars(prune_target)
+    extra_vars = docker_svc.prune_extra_vars(prune_target)
 
     actor_role = UserRole(session.get("user_role", "viewer"))
     run, _source, err = await invoke_capability(

plugins/docker/service.py

@@ -183,3 +183,16 @@ async def recent_failures_by_container(db, since) -> dict[tuple[str, str], list]
     for e in rows:
         out.setdefault((e.host_id, e.container_name), []).append(e)
     return out
+
+
+PRUNE_TARGETS = ("containers", "images", "system")
+
+
+def prune_extra_vars(prune_target: str) -> dict:
+    """Extra-vars for the bundled docker_prune playbook. "Prune unused images"
+    means ALL unused (docker image prune -a), not just dangling; the full system
+    prune stays conservative (-f, no -a) per the operator's choice (m78)."""
+    extra_vars: dict = {"prune_target": prune_target}
+    if prune_target == "images":
+        extra_vars["prune_all_images"] = True
+    return extra_vars

steward/mcp/tools/__init__.py

@@ -12,9 +12,9 @@ each argument's empty value means, and what comes back.
 """
 from __future__ import annotations
 
-from . import alerts, audit, hosts, metrics, monitors, overview, session
+from . import alerts, audit, automation, hosts, metrics, monitors, overview, session
 
-_MODULES = (session, overview, hosts, monitors, alerts, metrics, audit)
+_MODULES = (session, overview, hosts, monitors, alerts, metrics, audit, automation)
 
 
 def register_all(register_tool) -> None:

steward/mcp/tools/_audit.py

@@ -0,0 +1,24 @@
+"""Audit trail for changes made through MCP.
+
+Every mutating tool records its change with `audit_mcp`, under the calling
+user's id and the username "<user> via MCP" (an operator ruling on the API
+Surface), so the audit log separates what a person did from what an agent
+did on their token. The token id goes into the detail.
+"""
+from __future__ import annotations
+
+from steward.core.audit import log_audit
+from steward.mcp._context import current_caller, mcp_app
+
+VIA = " via MCP"
+
+
+def mcp_username(username: str) -> str:
+    return (username + VIA)[:64]
+
+
+async def audit_mcp(action: str, entity_type: str | None = None, entity_id: str | None = None,
+                    detail: dict | None = None) -> None:
+    caller = current_caller()
+    await log_audit(mcp_app(), caller.user_id, mcp_username(caller.username), action,
+                    entity_type, entity_id, {**(detail or {}), "token_id": caller.token_id})

steward/mcp/tools/audit.py

@@ -3,11 +3,12 @@ from __future__ import annotations
 
 import json
 
-from sqlalchemy import select
+from sqlalchemy import or_, select
 
 from steward.models.audit import AuditEvent
 from steward.models.users import UserRole
 
+from ._audit import mcp_username
 from ._shape import age_s, bounded, clamp_limit, db_session, iso, lookback
 
 
@@ -17,7 +18,8 @@ async def list_audit_events(action: str = "", user: str = "", entity_type: str =
 
     action: a prefix of the action name, e.g. "monitor." or "ansible.run";
       empty for all.
-    user: a username, exactly; "system" for automated changes.
+    user: a username, exactly; "system" for automated changes. Includes the
+      changes that user made through MCP, which are recorded as "<user> via MCP".
     entity_type: e.g. "monitor", "host", "alert_rule"; empty for all.
     range: "1h", "6h", "24h", "7d" (default) or "30d".
     limit: default 50, max 500.
@@ -33,7 +35,8 @@ async def list_audit_events(action: str = "", user: str = "", entity_type: str =
         esc = action.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
         stmt = stmt.where(AuditEvent.action.like(esc + "%", escape="\\"))
     if user:
-        stmt = stmt.where(AuditEvent.username == user)
+        stmt = stmt.where(or_(AuditEvent.username == user,
+                              AuditEvent.username == mcp_username(user)))
     if entity_type:
         stmt = stmt.where(AuditEvent.entity_type == entity_type)
     async with db_session() as db:

steward/mcp/tools/automation.py

@@ -0,0 +1,657 @@
+"""Automation tools: playbooks, inventory, runs, schedules and alert actions.
+
+Reads are viewer tools. Anything that starts a run or changes what runs is an
+operator tool (set_alert_action is admin, as on the alert rule form), and each
+of those writes an audit row as "<user> via MCP".
+
+Playbook runs default to Ansible check mode (`--check --diff`): the agent sees
+what would change before anything does, and applies only by passing
+check=false. Variables the playbook marks secret reach the run but are never
+stored or echoed back.
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+from collections import deque
+from datetime import timedelta
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from steward.ansible import executor, sources as src_module
+from steward.ansible.runner import trigger_run
+from steward.mcp._context import current_caller, mcp_app
+from steward.models.alerts import AlertRule
+from steward.models.ansible import AnsibleRun, AnsibleRunStatus
+from steward.models.ansible_inventory import AnsibleGroup, AnsibleTarget
+from steward.models.ansible_schedule import AnsibleSchedule
+from steward.models.hosts import Host
+from steward.models.users import User, UserRole
+
+from ._audit import audit_mcp
+from ._shape import age_s, bounded, clamp_limit, db_session, iso, lookback, pick
+
+ACTIVE = (AnsibleRunStatus.running, AnsibleRunStatus.queued)
+RECAP_KEYS = ("ok", "changed", "failed", "unreachable", "skipped")
+CONTENT_CHARS = 20000
+LINE_CHARS = 500
+WAIT_MAX_S = 120
+POLL_S = 2.0
+MASK = "<secret>"
+
+
+# ── sources and playbooks ────────────────────────────────────────────────────
+
+def _sources() -> list[dict]:
+    return src_module.get_sources(mcp_app().config.get("ANSIBLE", {}))
+
+
+def _source(name: str) -> dict:
+    srcs = _sources()
+    for s in srcs:
+        if s["name"] == (name or "").strip():
+            return s
+    raise ValueError(f"no playbook source named {name!r}. Sources: "
+                     f"{', '.join(s['name'] for s in srcs)} (list_playbooks shows their playbooks).")
+
+
+def _playbook(source: dict, path: str) -> str:
+    """The playbook's content; refuses a path the source does not have."""
+    path = (path or "").strip()
+    if path not in src_module.discover_playbooks(source["path"]):
+        raise ValueError(f"source {source['name']!r} has no playbook {path!r}; "
+                         f'list_playbooks(source="{source["name"]}") lists them.')
+    content = src_module.read_playbook(source["path"], path)
+    if content is None:
+        raise ValueError(f"playbook {path!r} in {source['name']!r} could not be read.")
+    return content
+
+
+def public_variables(variables: list[dict]) -> list[dict]:
+    """A playbook's run-time variables with the defaults of secret ones masked."""
+    return [{**v, "default": MASK if v["secret"] and v["default"] not in ("", None) else v["default"]}
+            for v in variables]
+
+
+def split_vars(variables: list[dict], extra_vars: dict | None) -> tuple[dict, dict]:
+    """(plain, secret): the values the run stores, and those it must not.
+
+    Refuses a vars_prompt variable that has no default and no value, since a
+    run cannot stop to ask for it.
+    """
+    extra_vars = dict(extra_vars or {})
+    for k, v in extra_vars.items():
+        if not isinstance(k, str) or not k.strip():
+            raise ValueError("extra_vars keys must be non-empty variable names.")
+        if not isinstance(v, (str, int, float, bool)):
+            raise ValueError(f"extra_vars[{k!r}] must be a string, number or boolean.")
+    secret_names = {v["name"] for v in variables if v["secret"]}
+    missing = [v["name"] for v in variables if v["required"] and v["name"] not in extra_vars]
+    if missing:
+        raise ValueError(f"this playbook prompts for {', '.join(missing)} and has no default; "
+                         "pass a value in extra_vars. Nothing was run.")
+    plain = {k: v for k, v in extra_vars.items() if k not in secret_names}
+    secret = {k: v for k, v in extra_vars.items() if k in secret_names}
+    return plain, secret
+
+
+def build_params(plain: dict, limit: str, tags: str, check: bool) -> dict:
+    params: dict = {}
+    if plain:
+        params["extra_vars_map"] = plain
+    if (limit or "").strip():
+        params["limit"] = limit.strip()
+    if (tags or "").strip():
+        params["tags"] = tags.strip()
+    if check:
+        params["check"] = True
+    return params
+
+
+async def _scope(db, source: dict, scope: str) -> tuple[str, str]:
+    """(stored scope, human description) from the forms the docstrings offer."""
+    s = (scope or "all").strip()
+    if s in ("all", "steward:all"):
+        n = len((await db.execute(select(AnsibleTarget.id))).all())
+        if not n:
+            raise ValueError("Steward's inventory has no targets yet; add one under Ansible → "
+                             "Inventory, or pass scope=\"inventory:<file>\" for a repo inventory.")
+        return "steward:all", f"all {n} inventory targets"
+    kind, _, want = s.removeprefix("steward:").partition(":")
+    if kind == "target":
+        rows = (await db.execute(select(AnsibleTarget))).scalars()
+        t = pick(want, rows, kind="target", list_tool="list_inventory")
+        return f"steward:target:{t.id}", f"target {t.name} ({t.address})"
+    if kind == "group":
+        rows = (await db.execute(select(AnsibleGroup).options(selectinload(AnsibleGroup.targets)))).scalars()
+        g = pick(want, rows, kind="group", list_tool="list_inventory")
+        if not g.targets:
+            raise ValueError(f"group {g.name!r} has no targets. Nothing was run.")
+        return f"steward:group:{g.id}", f"group {g.name} ({len(g.targets)} targets)"
+    if kind in ("inventory", "repo"):
+        inv = want.rsplit(":", 1)[-1] if kind == "repo" else want
+        found = src_module.discover_inventories(source["path"])
+        if inv not in found:
+            raise ValueError(f"source {source['name']!r} has no inventory file {inv!r}"
+                             + (f"; it has {', '.join(found)}." if found else "; it has none."))
+        return f"repo:{source['name']}:{inv}", f"{source['name']}/{inv}"
+    raise ValueError(f"scope {scope!r} is not one of: all, target:<name or id>, "
+                     "group:<name or id>, inventory:<file>. list_inventory shows the choices.")
+
+
+async def list_playbooks(source: str = "", category: str = "", limit: int = 200) -> dict:
+    """The playbooks Steward can run, by source, with what each one says it does.
+
+    source: one source's name; empty for all. "steward-builtin" holds
+      Steward's own maintenance playbooks, "steward-local" those written in
+      Steward's editor; the rest come from configured directories or git repos.
+    category: only playbooks whose `# steward:category:` matches; empty for all.
+    limit: default 200, max 1000.
+
+    `confirm: true` marks a playbook its author flagged as destructive:
+    run_playbook will apply it only with confirmed=true. get_playbook shows a
+    playbook's variables and content.
+    """
+    limit = clamp_limit(limit, 200, 1000)
+    srcs = [_source(source)] if source else _sources()
+    rows, out_sources = [], []
+    for s in srcs:
+        inventories = src_module.discover_inventories(s["path"])
+        out_sources.append({"source": s["name"], "type": s["type"],
+                            "editable": src_module.is_editable_source(s), "inventories": inventories})
+        for path in src_module.discover_playbooks(s["path"]):
+            meta = src_module.discover_playbook_meta(src_module.read_playbook(s["path"], path) or "")
+            if category and meta["category"].lower() != category.strip().lower():
+                continue
+            rows.append({"source": s["name"], "playbook": path, **meta})
+    return {"sources": out_sources,
+            **bounded(rows, limit, key="playbooks", more="Filter by `source` or `category`.")}
+
+
+async def get_playbook(source: str, playbook: str) -> dict:
+    """One playbook: its description, the variables it takes, and its content.
+
+    `variables` are what run_playbook's extra_vars can set: `required` ones
+    (a vars_prompt with no default) must be given; `secret` ones are passed
+    to the run but never stored, and their defaults are shown as "<secret>".
+    `content` is the YAML itself, cut at 20000 characters.
+    """
+    s = _source(source)
+    content = _playbook(s, playbook)
+    out = {
+        "source": s["name"],
+        "playbook": playbook.strip(),
+        **src_module.discover_playbook_meta(content),
+        "variables": public_variables(src_module.discover_playbook_variables(content)),
+        "content": content[:CONTENT_CHARS],
+    }
+    if len(content) > CONTENT_CHARS:
+        out["content_truncated"] = f"{len(content) - CONTENT_CHARS} more characters not shown."
+    return out
+
+
+async def list_inventory() -> dict:
+    """Steward's Ansible inventory: targets, groups, and how to scope a run to them.
+
+    Each target names the Steward host it is linked to (prune_docker needs
+    that link), its groups and the names of its Ansible variables — not their
+    values, which can hold credentials. `scopes` lists what run_playbook's
+    `scope` accepts.
+    """
+    async with db_session() as db:
+        targets = list((await db.execute(
+            select(AnsibleTarget).options(selectinload(AnsibleTarget.groups))
+            .order_by(AnsibleTarget.name))).scalars())
+        groups = list((await db.execute(
+            select(AnsibleGroup).options(selectinload(AnsibleGroup.targets))
+            .order_by(AnsibleGroup.name))).scalars())
+        host_names = dict((await db.execute(select(Host.id, Host.name))).all())
+    return {
+        "targets": [
+            {"target_id": t.id, "name": t.name, "address": t.address, "host_id": t.host_id,
+             "host_name": host_names.get(t.host_id), "groups": sorted(g.name for g in t.groups),
+             "var_names": sorted(t.ansible_vars or {})}
+            for t in targets
+        ],
+        "groups": [
+            {"group_id": g.id, "name": g.name, "targets": sorted(t.name for t in g.targets),
+             "var_names": sorted(g.ansible_vars or {})}
+            for g in groups
+        ],
+        "repo_inventories": [{"source": s["name"], "files": inv} for s in _sources()
+                             if (inv := src_module.discover_inventories(s["path"]))],
+        "scopes": {
+            "all": "every target above",
+            "target:<name or id>": "one target",
+            "group:<name or id>": "every target in a group",
+            "inventory:<file>": "an inventory file in the playbook's own source",
+        },
+    }
+
+
+# ── runs ─────────────────────────────────────────────────────────────────────
+
+def recap_totals(results: dict | None) -> dict | None:
+    hosts = (results or {}).get("hosts") or {}
+    if not hosts:
+        return None
+    return {"hosts": len(hosts),
+            **{k: sum(int((h or {}).get(k) or 0) for h in hosts.values()) for k in RECAP_KEYS}}
+
+
+def run_row(r: AnsibleRun, users: dict) -> dict:
+    end = r.finished_at
+    return {
+        "run_id": r.id,
+        "source": r.source_name,
+        "playbook": r.playbook_path,
+        "scope": r.inventory_scope,
+        "status": r.status.value,
+        "check": bool((r.params or {}).get("check")),
+        "triggered_by": users.get(r.triggered_by, "(deleted user)") if r.triggered_by else "system",
+        "started_at": iso(r.started_at),
+        "age_s": age_s(r.started_at),
+        "finished_at": iso(end),
+        "duration_s": int((end - r.started_at).total_seconds()) if end and r.started_at else None,
+        "recap": recap_totals(r.results),
+    }
+
+
+async def _users(db, ids) -> dict[str, str]:
+    ids = {i for i in ids if i}
+    if not ids:
+        return {}
+    return dict((await db.execute(select(User.id, User.username).where(User.id.in_(ids)))).all())
+
+
+async def list_ansible_runs(status: str = "", playbook: str = "", range: str = "7d",
+                            limit: int = 50) -> dict:
+    """Ansible runs, newest first: manual, scheduled, alert-triggered and MCP-started.
+
+    status: queued, running, success, failed, interrupted or cancelled; empty for all.
+    playbook: a substring of the playbook path; empty for all.
+    range: "1h", "6h", "24h", "7d" (default) or "30d".
+    limit: default 50, max 500.
+
+    `check: true` is a dry run that changed nothing. `recap` sums the play
+    recap over hosts; get_ansible_run shows per-host results and the log.
+    """
+    since, key = lookback(range)
+    limit = clamp_limit(limit, 50, 500)
+    stmt = (select(AnsibleRun).where(AnsibleRun.started_at >= since)
+            .order_by(AnsibleRun.started_at.desc()).limit(limit + 1))
+    if status:
+        try:
+            stmt = stmt.where(AnsibleRun.status == AnsibleRunStatus(status.strip()))
+        except ValueError:
+            raise ValueError(f"status {status!r} is not one of "
+                             f"{', '.join(s.value for s in AnsibleRunStatus)}.") from None
+    if playbook:
+        esc = playbook.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        stmt = stmt.where(AnsibleRun.playbook_path.ilike(f"%{esc}%", escape="\\"))
+    async with db_session() as db:
+        runs = list((await db.execute(stmt)).scalars())
+        users = await _users(db, [r.triggered_by for r in runs])
+    out = {"range": key, **bounded([run_row(r, users) for r in runs], limit, key="runs")}
+    if out["truncated"]:
+        out["total"] = None
+        out["more"] = (f"More than {limit} runs in range. Pass a larger `limit` (max 500), "
+                       "a shorter `range`, or filter by `status` or `playbook`.")
+    return out
+
+
+def _cut(line: str) -> str:
+    return line if len(line) <= LINE_CHARS else line[:LINE_CHARS] + f" [… {len(line) - LINE_CHARS} more]"
+
+
+def read_log(run: AnsibleRun, tail: int, grep: str, max_matches: int = 100) -> dict:
+    """The last `tail` lines of a run's log and the lines containing `grep`."""
+    path = executor.artifact_path(run.id)
+    if os.path.exists(path):
+        with open(path, encoding="utf-8", errors="replace") as f:
+            lines = f.read().splitlines()
+    else:
+        lines = (run.output or "").splitlines()
+    out: dict = {"log_lines": len(lines), "tail": [_cut(x) for x in deque(lines, maxlen=tail)]}
+    if grep:
+        needle = grep.lower()
+        hits = [{"line_no": i + 1, "line": _cut(x)} for i, x in enumerate(lines) if needle in x.lower()]
+        out["matches"] = hits[:max_matches]
+        out["matches_total"] = len(hits)
+    return out
+
+
+async def _load_run(db, run_id: str) -> AnsibleRun:
+    run = await db.get(AnsibleRun, (run_id or "").strip())
+    if run is None:
+        raise ValueError(f"no Ansible run with id {run_id!r}; list_ansible_runs lists them.")
+    return run
+
+
+async def _run_detail(run_id: str, tail: int, grep: str) -> dict:
+    async with db_session() as db:
+        run = await _load_run(db, run_id)
+        users = await _users(db, [run.triggered_by])
+    params = dict(run.params or {})
+    return {
+        **run_row(run, users),
+        "limit": params.get("limit"),
+        "tags": params.get("tags"),
+        "extra_vars": params.get("extra_vars_map") or {},
+        "hosts": (run.results or {}).get("hosts") or {},
+        "failures": ((run.results or {}).get("failures") or [])[:20],
+        **read_log(run, tail, grep),
+    }
+
+
+async def get_ansible_run(run_id: str, tail: int = 200, grep: str = "") -> dict:
+    """One Ansible run: its parameters, per-host recap, failures and log.
+
+    tail: how many of the log's last lines to return; default 200, max 2000.
+    grep: also return every log line containing this text (case-insensitive),
+      with its line number, up to 100.
+
+    `failures` are the task failures Steward parsed out of the log; start
+    there when status is failed. Secret variables were never stored, so they
+    do not appear in `extra_vars`.
+    """
+    return await _run_detail(run_id, clamp_limit(tail, 200, 2000), (grep or "").strip())
+
+
+async def wait_for_run(run_id: str, timeout_s: int = 60) -> dict:
+    """Wait for a run to finish, then return it as get_ansible_run does (last 50 lines).
+
+    timeout_s: how long to wait; default 60, max 120. If the run is still
+    going when it runs out, `finished` is false: call again to keep waiting.
+    """
+    deadline = asyncio.get_running_loop().time() + max(1, min(int(timeout_s or 60), WAIT_MAX_S))
+    while True:
+        async with db_session() as db:
+            status = (await _load_run(db, run_id)).status
+        if status not in ACTIVE or asyncio.get_running_loop().time() >= deadline:
+            break
+        await asyncio.sleep(POLL_S)
+    out = await _run_detail(run_id, 50, "")
+    out["finished"] = status not in ACTIVE
+    return out
+
+
+async def run_playbook(source: str, playbook: str, scope: str = "all",
+                       extra_vars: dict | None = None, limit: str = "", tags: str = "",
+                       check: bool = True, confirmed: bool = False) -> dict:
+    """Start an Ansible playbook run. A dry run (check mode) unless check=false.
+
+    source, playbook: as list_playbooks shows them.
+    scope: which hosts — "all" (default), "target:<name or id>",
+      "group:<name or id>", or "inventory:<file>" for an inventory file in the
+      same source. list_inventory shows the choices.
+    extra_vars: {"name": value}. Variables the playbook marks secret are
+      passed to the run but never stored or returned.
+    limit, tags: Ansible's --limit and --tags; empty for none.
+    check: true (default) runs with --check --diff, which reports what would
+      change and changes nothing. Pass false to apply — after a check run
+      looked right.
+    confirmed: a playbook with `confirm: true` (destructive, by its author's
+      flag) applies only when this is also true.
+
+    Returns at once with `run_id`; wait_for_run follows it to the end. Recorded
+    in the audit log as "<you> via MCP".
+    """
+    s = _source(source)
+    content = _playbook(s, playbook)
+    meta = src_module.discover_playbook_meta(content)
+    if not check and meta["confirm"] and not confirmed:
+        raise ValueError(f"{playbook} is flagged destructive by its author (steward:confirm). "
+                         "Run it with check=true first; to apply, pass check=false and "
+                         "confirmed=true. Nothing was run.")
+    plain, secret = split_vars(src_module.discover_playbook_variables(content), extra_vars)
+    async with db_session() as db:
+        stored_scope, scope_desc = await _scope(db, s, scope)
+    params = build_params(plain, limit, tags, check)
+    caller = current_caller()
+    run, _src, err = await trigger_run(
+        mcp_app(), source_name=s["name"], playbook_path=playbook.strip(),
+        inventory_scope=stored_scope, params=params or None,
+        secret_vars=secret or None, triggered_by=caller.user_id)
+    if err:
+        raise ValueError(f"the run could not start: {err}.")
+    await audit_mcp("ansible.run.started", "ansible_run", run.id, {
+        "source": s["name"], "playbook": playbook.strip(), "scope": stored_scope,
+        "check": check, "limit": params.get("limit"), "tags": params.get("tags"),
+        "extra_vars": sorted(plain), "secret_vars": sorted(secret)})
+    return {
+        "run_id": run.id,
+        "check": check,
+        "scope": scope_desc,
+        "status": run.status.value,
+        "secret_vars_passed": sorted(secret),
+        "summary": (f"{'Check-mode (dry) run' if check else 'Run'} of {playbook.strip()} "
+                    f"on {scope_desc} started."
+                    + (" It changes nothing; pass check=false to apply." if check else "")),
+        "next": f'wait_for_run(run_id="{run.id}")',
+    }
+
+
+async def cancel_ansible_run(run_id: str) -> dict:
+    """Stop a queued or running Ansible run. Recorded in the audit log."""
+    async with db_session() as db:
+        run = await _load_run(db, run_id)
+    if run.status not in ACTIVE:
+        raise ValueError(f"run {run.id} already ended ({run.status.value}); nothing to cancel.")
+    executor.cancel_run(run.id)
+    await audit_mcp("ansible.run.cancelled", "ansible_run", run.id, {"playbook": run.playbook_path})
+    return {"run_id": run.id, "cancelling": True,
+            "next": f'wait_for_run(run_id="{run.id}", timeout_s=30)'}
+
+
+# ── schedules ────────────────────────────────────────────────────────────────
+
+def schedule_row(sc: AnsibleSchedule, last: AnsibleRun | None) -> dict:
+    params = sc.params or {}
+    nxt = sc.last_run_at + timedelta(seconds=sc.interval_seconds) if sc.last_run_at else None
+    return {
+        "schedule_id": sc.id,
+        "name": sc.name,
+        "source": sc.source_name,
+        "playbook": sc.playbook_path,
+        "scope": sc.inventory_scope,
+        "interval_s": sc.interval_seconds,
+        "enabled": sc.enabled,
+        "check": bool(params.get("check")),
+        "extra_vars": params.get("extra_vars_map") or {},
+        "limit": params.get("limit"),
+        "tags": params.get("tags"),
+        "last_run_at": iso(sc.last_run_at),
+        "last_run_id": sc.last_run_id,
+        "last_run_status": last.status.value if last else None,
+        "last_error": sc.last_error,
+        "next_run_at": iso(nxt) if sc.enabled else None,
+    }
+
+
+async def _schedules(db) -> list[AnsibleSchedule]:
+    return list((await db.execute(select(AnsibleSchedule).order_by(AnsibleSchedule.name))).scalars())
+
+
+async def list_schedules() -> dict:
+    """Recurring playbook runs: what runs, how often, and how its last run went.
+
+    `check: true` schedules are dry runs that never change anything.
+    `last_error` is set when a firing could not start at all (a source or
+    playbook that went away); a run that started and failed shows in
+    `last_run_status`, and get_ansible_run(last_run_id) has its log.
+    """
+    async with db_session() as db:
+        scheds = await _schedules(db)
+        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_(ids)))).scalars()} if ids else {}
+    rows = [schedule_row(s, runs.get(s.last_run_id)) for s in scheds]
+    return {"schedules": rows, "total": len(rows)}
+
+
+async def _schedule_fields(db, source: str, playbook: str, scope: str, extra_vars: dict | None,
+                           limit: str, tags: str, check: bool) -> dict:
+    s = _source(source)
+    content = _playbook(s, playbook)
+    plain, secret = split_vars(src_module.discover_playbook_variables(content), extra_vars)
+    if secret:
+        raise ValueError(f"{', '.join(sorted(secret))} {'is' if len(secret) == 1 else 'are'} secret "
+                         "and a schedule would have to store it. Put it in the target's or group's "
+                         "Ansible variables instead. Nothing was saved.")
+    stored_scope, _desc = await _scope(db, s, scope)
+    return {"source_name": s["name"], "playbook_path": playbook.strip(), "inventory_scope": stored_scope,
+            "params": build_params(plain, limit, tags, check) or None}
+
+
+def _interval(interval_s: int) -> int:
+    if not isinstance(interval_s, int) or interval_s <= 0:
+        raise ValueError("interval_s must be a positive number of seconds, e.g. 86400 for daily.")
+    return interval_s
+
+
+async def create_schedule(name: str, source: str, playbook: str, interval_s: int,
+                          scope: str = "all", extra_vars: dict | None = None, limit: str = "",
+                          tags: str = "", check: bool = True, enabled: bool = True) -> dict:
+    """Run a playbook on a recurring interval. Each firing is a "system" run.
+
+    interval_s: seconds between runs, e.g. 3600 hourly, 86400 daily.
+    scope, extra_vars, limit, tags: as run_playbook. Secret variables cannot
+      be scheduled (they would have to be stored); set them as inventory vars.
+    check: true (default) schedules dry runs. Pass false only for a playbook
+      already proven by a check run.
+
+    Recorded in the audit log as "<you> via MCP".
+    """
+    if not (name or "").strip():
+        raise ValueError("a schedule needs a name.")
+    async with db_session() as db, db.begin():
+        fields = await _schedule_fields(db, source, playbook, scope, extra_vars, limit, tags, check)
+        sc = AnsibleSchedule(name=name.strip()[:128], interval_seconds=_interval(interval_s),
+                             enabled=enabled, **fields)
+        db.add(sc)
+    await audit_mcp("ansible.schedule.created", "ansible_schedule", sc.id,
+                    {"name": sc.name, "playbook": sc.playbook_path, "interval_s": sc.interval_seconds,
+                     "check": check})
+    return schedule_row(sc, None)
+
+
+async def update_schedule(schedule: str, name: str | None = None, source: str | None = None,
+                          playbook: str | None = None, interval_s: int | None = None,
+                          scope: str | None = None, extra_vars: dict | None = None,
+                          limit: str | None = None, tags: str | None = None,
+                          check: bool | None = None, enabled: bool | None = None) -> dict:
+    """Change a schedule. Only the arguments you pass change; the rest stay.
+
+    schedule: its name or id (list_schedules). Pass enabled=false to pause it.
+    extra_vars, when passed, replaces the whole set. Recorded in the audit log.
+    """
+    async with db_session() as db, db.begin():
+        sc = pick(schedule, await _schedules(db), kind="schedule", list_tool="list_schedules")
+        params = dict(sc.params or {})
+        cur_scope = sc.inventory_scope
+        if cur_scope.startswith("repo:"):
+            cur_scope = "inventory:" + cur_scope.rsplit(":", 1)[-1]
+        fields = await _schedule_fields(
+            db, source if source is not None else sc.source_name,
+            playbook if playbook is not None else sc.playbook_path,
+            scope if scope is not None else cur_scope,
+            extra_vars if extra_vars is not None else params.get("extra_vars_map"),
+            limit if limit is not None else params.get("limit", ""),
+            tags if tags is not None else params.get("tags", ""),
+            check if check is not None else bool(params.get("check")))
+        for k, v in fields.items():
+            setattr(sc, k, v)
+        if name is not None and name.strip():
+            sc.name = name.strip()[:128]
+        if interval_s is not None:
+            sc.interval_seconds = _interval(interval_s)
+        if enabled is not None:
+            sc.enabled = enabled
+        row = schedule_row(sc, None)
+    changed = sorted(k for k, v in {"name": name, "source": source, "playbook": playbook,
+                                    "interval_s": interval_s, "scope": scope, "extra_vars": extra_vars,
+                                    "limit": limit, "tags": tags, "check": check,
+                                    "enabled": enabled}.items() if v is not None)
+    await audit_mcp("ansible.schedule.updated", "ansible_schedule", row["schedule_id"],
+                    {"name": row["name"], "changed": changed})
+    return row
+
+
+async def delete_schedule(schedule: str) -> dict:
+    """Delete a schedule (its past runs stay). Recorded in the audit log.
+
+    schedule: its name or id. To pause instead, update_schedule(enabled=false).
+    """
+    async with db_session() as db, db.begin():
+        sc = pick(schedule, await _schedules(db), kind="schedule", list_tool="list_schedules")
+        await db.delete(sc)
+    await audit_mcp("ansible.schedule.deleted", "ansible_schedule", sc.id,
+                    {"name": sc.name, "playbook": sc.playbook_path})
+    return {"deleted": sc.name, "schedule_id": sc.id}
+
+
+# ── alert actions ────────────────────────────────────────────────────────────
+
+async def set_alert_action(rule: str, source: str = "", playbook: str = "", inventory: str = "",
+                           extra_vars: dict | None = None, limit: str = "", tags: str = "",
+                           check: bool = True) -> dict:
+    """Set, replace or clear the playbook an alert rule runs when it starts firing.
+
+    rule: the alert rule's name or id (list_alert_rules).
+    playbook: empty clears the rule's action. Otherwise source, playbook and
+      inventory (an inventory file in that source, as list_playbooks shows)
+      are all required.
+    extra_vars: {"name": value}, stored on the rule. The run also receives the
+      alert's own context variables. Secret variables cannot be stored here.
+    check: true (default) makes the action a dry run each time it fires;
+      pass false for a remediation already proven by a check run.
+
+    Admin only, as on the alert rule form. Recorded in the audit log.
+    """
+    async with db_session() as db, db.begin():
+        r = pick(rule, (await db.execute(select(AlertRule))).scalars(),
+                 kind="alert rule", list_tool="list_alert_rules")
+        action = None
+        if (playbook or "").strip():
+            s = _source(source)
+            content = _playbook(s, playbook)
+            inv = (inventory or "").strip()
+            found = src_module.discover_inventories(s["path"])
+            if inv not in found:
+                raise ValueError(f"source {s['name']!r} has no inventory file {inv!r}"
+                                 + (f"; it has {', '.join(found)}." if found else "; it has none.")
+                                 + " Nothing was saved.")
+            plain, secret = split_vars(src_module.discover_playbook_variables(content), extra_vars)
+            if secret:
+                raise ValueError(f"{', '.join(sorted(secret))} is secret and an alert action would "
+                                 "have to store it. Nothing was saved.")
+            action = {"source": s["name"], "playbook": playbook.strip(), "inventory": inv}
+            if plain:
+                action["extra_vars"] = [f"{k}={v}" for k, v in plain.items()]
+            if (limit or "").strip():
+                action["limit"] = limit.strip()
+            if (tags or "").strip():
+                action["tags"] = tags.strip()
+            if check:
+                action["check"] = True
+        r.ansible_action = action
+        name, rid = r.name, r.id
+    await audit_mcp("alert_rule.action_set" if action else "alert_rule.action_cleared",
+                    "alert_rule", name,
+                    {"rule_id": rid, **({k: v for k, v in action.items() if k != "extra_vars"}
+                                        if action else {})})
+    return {"rule_id": rid, "rule_name": name, "action": action,
+            "summary": (f"When {name} starts firing it runs {action['playbook']}"
+                        f"{' in check mode' if action.get('check') else ''}."
+                        if action else f"{name} no longer runs a playbook when it fires.")}
+
+
+def register(register_tool) -> None:
+    for fn in (list_playbooks, get_playbook, list_inventory, list_ansible_runs,
+               get_ansible_run, list_schedules, wait_for_run):
+        register_tool(fn, role=UserRole.viewer, read_only=True)
+    for fn in (run_playbook, cancel_ansible_run, create_schedule, update_schedule, delete_schedule):
+        register_tool(fn, role=UserRole.operator, read_only=False)
+    register_tool(set_alert_action, role=UserRole.admin, read_only=False)

steward/mcp/tools/session.py

@@ -6,8 +6,10 @@ from steward.models.users import UserRole
 
 _ROLE_MEANS = {
     UserRole.viewer: "Read-only: every list_*/get_* tool. Nothing that changes Steward or your hosts.",
-    UserRole.operator: "Viewer tools, plus running playbooks and changing monitors, alerts and schedules.",
+    UserRole.operator: "Viewer tools, plus running playbooks (check mode unless told otherwise) "
-    UserRole.admin: "Everything, including deleting monitors and alert rules.",
+                       "and changing monitors, alerts and schedules.",
+    UserRole.admin: "Everything, including the playbook an alert rule runs, pruning Docker disk, "
+                    "and deleting monitors and alert rules.",
 }
 
 

tests/integration/test_mcp_automation.py

@@ -0,0 +1,196 @@
+"""Integration: the automation tools against real Postgres and a real ansible-playbook.
+
+One local playbook (connection: local) writes a file. A run started through
+run_playbook is check mode unless told otherwise, so the first run must leave
+no file behind and the confirmed apply must write it. Every write is audited as
+"<user> via MCP", and the secret variable reaches the play without being
+stored, audited or echoed.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import textwrap
+import uuid
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+_NEEDS_DB = pytest.mark.skipif(
+    not os.environ.get("STEWARD_DATABASE_URL"),
+    reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
+)
+
+SECRET = "s3cr3t-value-xyz"
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+@_NEEDS_DB
+def test_runs_schedules_and_alert_actions_through_mcp(app, tmp_path):
+    from sqlalchemy import select
+    from steward.core.api_tokens import Caller
+    from steward.mcp import _context
+    from steward.mcp.tools import automation as a
+    from steward.models.alerts import AlertOperator, AlertRule
+    from steward.models.ansible import AnsibleRun
+    from steward.models.ansible_inventory import AnsibleTarget
+    from steward.models.audit import AuditEvent
+    from steward.models.users import User, UserRole
+
+    tag = uuid.uuid4().hex[:8]
+    src, uid, username = f"mcp-src-{tag}", str(uuid.uuid4()), f"mcp-op-{tag}"
+    target_name, rule_name = f"mcp-target-{tag}", f"mcp-rule-{tag}"
+    out_file = tmp_path / "applied.txt"
+
+    play_dir = tmp_path / "playbooks"
+    play_dir.mkdir()
+    (play_dir / "play.yml").write_text(textwrap.dedent("""\
+        # description: Writes a marker file (MCP integration test)
+        # steward:confirm: true
+        - hosts: all
+          connection: local
+          gather_facts: false
+          vars:
+            greeting: hello
+            api_token: changeme
+            out_file: /nonexistent
+          tasks:
+            - debug:
+                msg: "GREETING-{{ greeting }} TOKEN-LEN-{{ api_token | length }}"
+            - copy:
+                dest: "{{ out_file }}"
+                content: applied
+    """))
+    (play_dir / "hosts").write_text("localhost ansible_connection=local\n")
+    app.config["ANSIBLE"] = {"sources": [{"name": src, "type": "local", "path": str(play_dir)}]}
+
+    async def _seed():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add(User(id=uid, username=username, email=f"{tag}@e.test",
+                           password_hash="x", role=UserRole.admin, is_active=True))
+            async with s.begin():
+                s.add(AnsibleTarget(name=target_name, address="127.0.0.1",
+                                    ansible_vars={"ansible_connection": "local"}))
+                s.add(AlertRule(name=rule_name, source_module="host_agent", resource_name="nas",
+                                metric_name="cpu_pct", operator=AlertOperator.gt, threshold=90.0,
+                                created_by=uid))
+
+    async def _audit(action, entity_id=None):
+        async with app.db_sessionmaker() as s:
+            stmt = select(AuditEvent).where(AuditEvent.action == action,
+                                            AuditEvent.user_id == uid)
+            if entity_id:
+                stmt = stmt.where(AuditEvent.entity_id == entity_id)
+            return list((await s.execute(stmt)).scalars())
+
+    async def _go():
+        await _seed()
+        _context._app_ctx.set(app)
+        _context._caller_ctx.set(Caller(user_id=uid, username=username,
+                                        role=UserRole.admin, token_id=42))
+
+        # ── discovery ──────────────────────────────────────────────────────
+        books = (await a.list_playbooks(source=src))["playbooks"]
+        assert [(b["playbook"], b["confirm"]) for b in books] == [("play.yml", True)]
+        pb = await a.get_playbook(src, "play.yml")
+        defaults = {v["name"]: (v["default"], v["secret"]) for v in pb["variables"]}
+        assert defaults["api_token"] == ("<secret>", True)
+        assert defaults["greeting"] == ("hello", False)
+        inv = await a.list_inventory()
+        assert any(t["name"] == target_name and t["var_names"] == ["ansible_connection"]
+                   for t in inv["targets"])
+        with pytest.raises(ValueError, match="has no playbook 'nope.yml'"):
+            await a.get_playbook(src, "nope.yml")
+
+        # ── a check-mode run is the default and changes nothing ───────────
+        extra = {"greeting": "yo", "api_token": SECRET, "out_file": str(out_file)}
+        started = await a.run_playbook(src, "play.yml", scope=f"target:{target_name}",
+                                       extra_vars=extra)
+        assert started["check"] is True and started["secret_vars_passed"] == ["api_token"]
+        assert SECRET not in json.dumps(started)
+        done = await a.wait_for_run(started["run_id"], timeout_s=120)
+        assert done["finished"] and done["status"] == "success", done["tail"]
+        assert done["check"] is True and done["triggered_by"] == username
+        assert not out_file.exists()
+        log = "\n".join(done["tail"])
+        assert "GREETING-yo" in log and f"TOKEN-LEN-{len(SECRET)}" in log
+        assert SECRET not in json.dumps(done)
+        async with app.db_sessionmaker() as s:
+            run = await s.get(AnsibleRun, started["run_id"])
+        assert run.params["check"] is True and SECRET not in json.dumps(run.params)
+
+        rows = await _audit("ansible.run.started", started["run_id"])
+        assert len(rows) == 1 and rows[0].username == f"{username} via MCP"
+        detail = json.loads(rows[0].detail_json)
+        assert detail["check"] is True and detail["token_id"] == 42
+        assert detail["secret_vars"] == ["api_token"] and SECRET not in rows[0].detail_json
+
+        # ── applying a confirm playbook needs confirmed=true ──────────────
+        with pytest.raises(ValueError, match="flagged destructive"):
+            await a.run_playbook(src, "play.yml", scope=f"target:{target_name}",
+                                 extra_vars=extra, check=False)
+        applied = await a.run_playbook(src, "play.yml", scope=f"target:{target_name}",
+                                       extra_vars=extra, check=False, confirmed=True)
+        done = await a.wait_for_run(applied["run_id"], timeout_s=120)
+        assert done["status"] == "success" and done["check"] is False, done["tail"]
+        assert out_file.read_text() == "applied"
+
+        listed = (await a.list_ansible_runs(playbook="play.yml", range="1h", limit=500))["runs"]
+        ours = {r["run_id"]: r for r in listed}
+        assert ours[started["run_id"]]["check"] is True
+        assert ours[applied["run_id"]]["check"] is False
+        grep = await a.get_ansible_run(applied["run_id"], grep="greeting-")
+        assert grep["matches_total"] >= 1
+        with pytest.raises(ValueError, match="already ended"):
+            await a.cancel_ansible_run(applied["run_id"])
+
+        with pytest.raises(ValueError, match="scope 'everything'"):
+            await a.run_playbook(src, "play.yml", scope="everything")
+        with pytest.raises(ValueError, match="no target named"):
+            await a.run_playbook(src, "play.yml", scope=f"target:ghost-{tag}")
+
+        # ── schedules ──────────────────────────────────────────────────────
+        with pytest.raises(ValueError, match="api_token is secret.*Nothing was saved"):
+            await a.create_schedule(f"sched-{tag}", src, "play.yml", 3600,
+                                    scope=f"target:{target_name}", extra_vars=extra)
+        sc = await a.create_schedule(f"sched-{tag}", src, "play.yml", 3600,
+                                     scope=f"target:{target_name}",
+                                     extra_vars={"out_file": str(out_file)})
+        assert sc["check"] is True and sc["enabled"] is True
+        up = await a.update_schedule(f"sched-{tag}", enabled=False, interval_s=86400)
+        assert (up["enabled"], up["interval_s"], up["check"]) == (False, 86400, True)
+        assert up["extra_vars"] == {"out_file": str(out_file)}  # untouched fields stay
+        assert any(s["schedule_id"] == sc["schedule_id"] for s in (await a.list_schedules())["schedules"])
+        await a.delete_schedule(sc["schedule_id"])
+        assert not any(s["schedule_id"] == sc["schedule_id"]
+                       for s in (await a.list_schedules())["schedules"])
+        for action in ("ansible.schedule.created", "ansible.schedule.updated",
+                       "ansible.schedule.deleted"):
+            assert len(await _audit(action, sc["schedule_id"])) == 1, action
+
+        # ── alert actions ──────────────────────────────────────────────────
+        with pytest.raises(ValueError, match="no inventory file 'missing'"):
+            await a.set_alert_action(rule_name, src, "play.yml", "missing")
+        act = await a.set_alert_action(rule_name, src, "play.yml", "hosts",
+                                       extra_vars={"greeting": "alert"})
+        assert act["action"] == {"source": src, "playbook": "play.yml", "inventory": "hosts",
+                                 "extra_vars": ["greeting=alert"], "check": True}
+        cleared = await a.set_alert_action(rule_name)
+        assert cleared["action"] is None
+        async with app.db_sessionmaker() as s:
+            rule = (await s.execute(select(AlertRule).where(AlertRule.name == rule_name))).scalar_one()
+        assert rule.ansible_action is None
+        assert len(await _audit("alert_rule.action_set", rule_name)) == 1
+        assert len(await _audit("alert_rule.action_cleared", rule_name)) == 1
+
+    asyncio.run(_go())

tests/mcp/test_automation.py

@@ -0,0 +1,109 @@
+"""The automation tools' pure parts and their gates: check mode by default,
+secrets kept out of what is stored, roles, and refusals that run nothing.
+
+A real check-mode run, applied run and audit rows are covered against
+Postgres in tests/integration/test_mcp_automation.py.
+"""
+from __future__ import annotations
+
+import inspect
+from types import SimpleNamespace as NS
+
+import pytest
+
+from steward.core.api_tokens import Caller
+from steward.mcp.tools import automation
+from steward.models.users import UserRole
+
+VARS = [
+    {"name": "api_token", "default": "abc", "secret": True, "required": False, "prompt": None},
+    {"name": "greeting", "default": "hi", "secret": False, "required": False, "prompt": None},
+    {"name": "release", "default": "", "secret": False, "required": True, "prompt": "Release?"},
+]
+
+VIEWER_TOOLS = {"list_playbooks", "get_playbook", "list_inventory", "list_ansible_runs",
+                "get_ansible_run", "list_schedules", "wait_for_run"}
+OPERATOR_TOOLS = {"run_playbook", "cancel_ansible_run", "create_schedule", "update_schedule",
+                  "delete_schedule"}
+
+
+def test_split_vars_keeps_secrets_apart_from_what_is_stored():
+    plain, secret = automation.split_vars(VARS, {"api_token": "s3cr3t", "greeting": "yo",
+                                                 "release": "1.2", "other": 5})
+    assert plain == {"greeting": "yo", "release": "1.2", "other": 5}
+    assert secret == {"api_token": "s3cr3t"}
+
+
+def test_split_vars_refuses_a_missing_prompt_and_odd_values():
+    with pytest.raises(ValueError, match="prompts for release.*Nothing was run"):
+        automation.split_vars(VARS, {"greeting": "yo"})
+    with pytest.raises(ValueError, match="string, number or boolean"):
+        automation.split_vars(VARS[:2], {"greeting": ["a"]})
+
+
+def test_public_variables_mask_secret_defaults():
+    out = {v["name"]: v["default"] for v in automation.public_variables(VARS)}
+    assert out == {"api_token": "<secret>", "greeting": "hi", "release": ""}
+
+
+def test_build_params():
+    assert automation.build_params({}, "", "", True) == {"check": True}
+    assert automation.build_params({"a": 1}, " web01 ", "deploy", False) == {
+        "extra_vars_map": {"a": 1}, "limit": "web01", "tags": "deploy"}
+
+
+def test_writes_default_to_check_mode():
+    for fn in (automation.run_playbook, automation.create_schedule, automation.set_alert_action):
+        assert inspect.signature(fn).parameters["check"].default is True, fn.__name__
+
+
+async def test_run_playbook_will_not_apply_a_confirm_playbook_unconfirmed(monkeypatch):
+    monkeypatch.setattr(automation, "_source", lambda name: {"name": name, "path": "/x"})
+    monkeypatch.setattr(automation, "_playbook",
+                        lambda s, p: "# steward:confirm: true\n- hosts: all\n  tasks: []\n")
+
+    def _no_run(*a, **k):
+        raise AssertionError("a run was started")
+    monkeypatch.setattr(automation, "trigger_run", _no_run)
+    with pytest.raises(ValueError, match="flagged destructive.*confirmed=true"):
+        await automation.run_playbook("s", "wipe.yml", check=False)
+
+
+def test_recap_totals_sum_over_hosts():
+    results = {"hosts": {"a": {"ok": 3, "changed": 1}, "b": {"ok": 2, "failed": 1}}}
+    assert automation.recap_totals(results) == {
+        "hosts": 2, "ok": 5, "changed": 1, "failed": 1, "unreachable": 0, "skipped": 0}
+    assert automation.recap_totals(None) is None
+
+
+def test_read_log_tail_and_grep(monkeypatch, tmp_path):
+    monkeypatch.setattr(automation.executor, "artifact_path", lambda rid: str(tmp_path / "none.log"))
+    run = NS(id="r1", output="\n".join(["PLAY [all]", "TASK [x]", "fatal: boom", "x" * 900, "PLAY RECAP"]))
+    out = automation.read_log(run, tail=2, grep="FATAL")
+    assert out["log_lines"] == 5
+    assert out["tail"][1] == "PLAY RECAP" and out["tail"][0].endswith("[… 400 more]")
+    assert out["matches"] == [{"line_no": 3, "line": "fatal: boom"}] and out["matches_total"] == 1
+    (tmp_path / "r1.log").write_text("from the artifact\n")
+    monkeypatch.setattr(automation.executor, "artifact_path", lambda rid: str(tmp_path / f"{rid}.log"))
+    assert automation.read_log(run, tail=10, grep="")["tail"] == ["from the artifact"]
+
+
+def test_automation_tools_are_registered_with_their_roles(app):
+    mcp = app.mcp_server
+    for name in VIEWER_TOOLS:
+        assert mcp.tool_roles[name] is UserRole.viewer, name
+        assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is True, name
+    for name in OPERATOR_TOOLS:
+        assert mcp.tool_roles[name] is UserRole.operator, name
+        assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is False, name
+    assert mcp.tool_roles["set_alert_action"] is UserRole.admin
+    for name in VIEWER_TOOLS | OPERATOR_TOOLS | {"set_alert_action"}:
+        assert (mcp._tool_manager.get_tool(name).description or "").strip(), name
+
+
+def test_a_viewer_token_cannot_start_a_run(app):
+    viewer = Caller(user_id="u", username="v", role=UserRole.viewer, token_id=1)
+    operator = Caller(user_id="u", username="o", role=UserRole.operator, token_id=2)
+    assert "needs the operator role" in app.mcp_server.refusal("run_playbook", viewer)
+    assert app.mcp_server.refusal("run_playbook", operator) is None
+    assert "needs the admin role" in app.mcp_server.refusal("set_alert_action", operator)

tests/plugins/docker/test_mcp_tools.py

@@ -1,8 +1,11 @@
 """The docker plugin's MCP tools: what counts as a problem, and how it is reported."""
+import inspect
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace as NS
 from unittest.mock import MagicMock
 
+import pytest
+
 import plugins.docker as docker_plugin
 from plugins.docker import mcp_tools as t
 
@@ -78,3 +81,21 @@ def test_plugin_exposes_the_mcp_hooks():
     for call in reg.call_args_list:
         assert call.kwargs["read_only"] is True
     assert callable(docker_plugin.mcp_overview)
+
+
+def test_prune_preview_sizes_what_each_target_removes():
+    usage = NS(images_reclaimable=3000, build_cache_size=70)
+    assert t.prune_preview("containers", usage, 4) == {"stopped_containers": 4}
+    assert t.prune_preview("images", usage, 4) == {"unused_images_bytes": 3000}
+    system = t.prune_preview("system", usage, 4)
+    assert system["stopped_containers"] == 4 and system["build_cache_bytes"] == 70
+    assert t.prune_preview("images", None, 0) == {"unused_images_bytes": None}
+
+
+async def test_prune_docker_refuses_an_unknown_target_before_reading_anything(monkeypatch):
+    def _no_db():
+        raise AssertionError("touched the database")
+    monkeypatch.setattr(t, "db_session", _no_db)
+    with pytest.raises(ValueError, match="'volumes' is not one of containers, images, system"):
+        await t.prune_docker("nas", "volumes")
+    assert inspect.signature(t.prune_docker).parameters["check"].default is True

tests/plugins/docker/test_routes.py

@@ -41,17 +41,6 @@ def test_container_log_views_defined():
     assert callable(r.container_logs_lines)
 
 
-def test_prune_extra_vars_mapping():
-    """The prune buttons drive one playbook via prune_target; only 'images'
-    widens to ALL unused images (docker image prune -a), system stays -f."""
-    assert r._prune_extra_vars("containers") == {"prune_target": "containers"}
-    assert r._prune_extra_vars("images") == {
-        "prune_target": "images", "prune_all_images": True,
-    }
-    # System prune stays conservative — no prune_all_images key.
-    assert r._prune_extra_vars("system") == {"prune_target": "system"}
-
-
 def test_human_bytes_formats_binary_units():
     assert r._human_bytes(None) == "—"
     assert r._human_bytes(0) == "0 B"

tests/plugins/docker/test_service.py

@@ -16,3 +16,14 @@ def test_bucket_keeps_short_series_and_averages_long_ones():
     out = svc.bucket([float(i) for i in range(400)], target=40)
     assert len(out) == 40
     assert out[0] == sum(range(10)) / 10
+
+
+def test_prune_extra_vars_mapping():
+    """The prune buttons drive one playbook via prune_target; only 'images'
+    widens to ALL unused images (docker image prune -a), system stays -f."""
+    assert svc.prune_extra_vars("containers") == {"prune_target": "containers"}
+    assert svc.prune_extra_vars("images") == {
+        "prune_target": "images", "prune_all_images": True,
+    }
+    # System prune stays conservative — no prune_all_images key.
+    assert svc.prune_extra_vars("system") == {"prune_target": "system"}