diff --git a/docs/plugins/writing-a-plugin.md b/docs/plugins/writing-a-plugin.md
index 7e93f1f..5653414 100644
--- a/docs/plugins/writing-a-plugin.md
+++ b/docs/plugins/writing-a-plugin.md
@@ -356,6 +356,25 @@ async def list_targets(limit: int = 50) -> dict:
 - Refuse by raising `ValueError("what was wrong, and what to do instead")`. The agent reads that text. Any other exception reaches it only as a generic failure.
 - Tool names are global across core and every plugin; a name already taken fails registration and the plugin contributes no tools (logged).
 - Hot-reloading the plugin replaces its tools.
+- Follow the response conventions the core tools use (`steward/mcp/tools/_shape.py`): times via `iso()` plus `age_s()`, lists through `bounded()` so a cut is never silent, series through `summarize()`, and lookups by name or id through `pick()`.
+
+### Contributing to get_overview
+
+`get_overview` is the agent's first call: what is wrong right now, ranked. A plugin adds its own problems to it by exporting an optional `mcp_overview(db)` that returns findings built with `steward.mcp.tools.overview.finding`:
+
+```python
+# plugins/myplugin/__init__.py
+async def mcp_overview(db) -> list[dict]:
+    from steward.mcp.tools.overview import finding
+    return [
+        finding("warning", "target_unreachable", t.name, f"{t.name} has not answered for {t.misses} scans",
+                since=t.last_ok_at, ids={"target_id": t.id}, next=f'get_target(target="{t.id}")')
+        for t in await _unreachable_targets(db)
+    ]
+```
+
+- `severity` is `"critical"` (something is down or about to be) or `"warning"` (degraded, or heading that way). Report problems only; a healthy plugin returns `[]`.
+- It is called only while the plugin is enabled, with its own session. If it raises, the overview still returns, and names the plugin under `unavailable`.
 
 ---
 
diff --git a/steward/core/metric_series.py b/steward/core/metric_series.py
new file mode 100644
index 0000000..97a0c2e
--- /dev/null
+++ b/steward/core/metric_series.py
@@ -0,0 +1,122 @@
+"""Reads over the generic metric store (plugin_metrics + its hourly rollup).
+
+Any source that records into PluginMetric — host_agent, traefik, unifi, snmp,
+the monitor types, Steward's self-monitor — is readable through these without
+importing the plugin that wrote it.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import func, or_, select, text
+
+from steward.core.time_range import bucket_seconds
+from steward.models.metrics import PluginMetric, PluginMetricHourly
+
+# Raw rows older than this are rolled into plugin_metrics_hourly and deleted
+# (see core/metrics_retention.py), so a range reaching past it reads the rollup.
+RAW_DAYS = 7
+
+# Loose index scans: step through the distinct leading values of the
+# (source_module, resource_name, …) index one probe at a time instead of
+# scanning every sample. Postgres has no native skip scan before 18.
+_SOURCES_SQL = text("""
+    WITH RECURSIVE s(v) AS (
+        SELECT min(source_module) FROM plugin_metrics
+        UNION ALL
+        SELECT (SELECT min(source_module) FROM plugin_metrics WHERE source_module > s.v)
+        FROM s WHERE s.v IS NOT NULL
+    )
+    SELECT v FROM s WHERE v IS NOT NULL
+""")
+
+_RESOURCES_SQL = text("""
+    WITH RECURSIVE r(v) AS (
+        SELECT min(resource_name) FROM plugin_metrics WHERE source_module = :src
+        UNION ALL
+        SELECT (SELECT min(resource_name) FROM plugin_metrics
+                WHERE source_module = :src AND resource_name > r.v)
+        FROM r WHERE r.v IS NOT NULL
+    )
+    SELECT v FROM r WHERE v IS NOT NULL
+""")
+
+
+async def sources(db) -> list[str]:
+    """Every source_module with raw samples on record."""
+    return [v for (v,) in await db.execute(_SOURCES_SQL)]
+
+
+async def resources(db, source_module: str) -> list[str]:
+    """Every resource a source has raw samples for, sorted."""
+    return [v for (v,) in await db.execute(_RESOURCES_SQL, {"src": source_module})]
+
+
+async def latest_values(db, source_module: str, resource: str, *,
+                        include_children: bool = False) -> list[PluginMetric]:
+    """The newest sample of each metric for a resource, sorted by (resource, metric).
+
+    include_children also returns the resource's sub-resources, named
+    "<resource>:<part>" (a host's mounts, interfaces, sensors).
+    """
+    match = PluginMetric.resource_name == resource
+    if include_children:
+        like = resource.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + ":%"
+        match = or_(match, PluginMetric.resource_name.like(like, escape="\\"))
+    return list((await db.execute(
+        select(PluginMetric)
+        .where(PluginMetric.source_module == source_module, match)
+        .distinct(PluginMetric.resource_name, PluginMetric.metric_name)
+        .order_by(PluginMetric.resource_name, PluginMetric.metric_name,
+                  PluginMetric.recorded_at.desc())
+    )).scalars())
+
+
+async def points(db, source_module: str, resource: str, metric: str, since: datetime,
+                 *, target: int = 120) -> list[tuple[datetime, float]]:
+    """(time, average) points for one metric since `since`, oldest first.
+
+    The part of the range older than the raw window comes from the hourly
+    rollup; the rest is bucket-averaged in SQL (buckets no wider than an hour
+    when both are present, so the series is evenly grained across the seam).
+    """
+    now = datetime.now(timezone.utc)
+    raw_cutoff = (now - timedelta(days=RAW_DAYS)).replace(minute=0, second=0, microsecond=0)
+    out: list[tuple[datetime, float]] = []
+
+    if since < raw_cutoff:
+        hourly = await db.execute(
+            select(PluginMetricHourly.bucket, PluginMetricHourly.value_avg)
+            .where(
+                PluginMetricHourly.source_module == source_module,
+                PluginMetricHourly.resource_name == resource,
+                PluginMetricHourly.metric_name == metric,
+                PluginMetricHourly.bucket >= since,
+                PluginMetricHourly.bucket < raw_cutoff,
+            )
+            .order_by(PluginMetricHourly.bucket)
+        )
+        out.extend((b, float(v)) for b, v in hourly)
+
+    raw_since = max(since, raw_cutoff)
+    width_s = bucket_seconds(raw_since, target)
+    if since < raw_cutoff:
+        width_s = min(width_s, 3600)
+    bucket = func.date_bin(
+        func.make_interval(0, 0, 0, 0, 0, 0, width_s),
+        PluginMetric.recorded_at,
+        func.to_timestamp(0),
+    ).label("bucket")
+    raw = await db.execute(
+        select(bucket, func.avg(PluginMetric.value))
+        .where(
+            PluginMetric.source_module == source_module,
+            PluginMetric.resource_name == resource,
+            PluginMetric.metric_name == metric,
+            PluginMetric.recorded_at >= raw_since,
+        )
+        .group_by(bucket)
+        .order_by(bucket)
+    )
+    out.extend((b, float(v)) for b, v in raw)
+    return out
diff --git a/steward/hosts/service.py b/steward/hosts/service.py
index da6f2bb..dd7b966 100644
--- a/steward/hosts/service.py
+++ b/steward/hosts/service.py
@@ -13,6 +13,11 @@ from steward.core.status import _heartbeat_title, _last_n_results, _uptime_by_mo
 from steward.models.monitors import Monitor, MonitorResult
 
 
+# Host-level agent metrics read for the at-a-glance views (plus the root
+# mount's disk_used_pct, read from the "<host>:/" sub-resource).
+AGENT_GLANCE_METRICS = ("cpu_pct", "mem_used_pct", "disk_used_pct", "disk_used_pct_worst")
+
+
 def agent_stale_after_seconds(config) -> int:
     """The host_agent plugin's freshness window from app config (default 180s)."""
     return int(((config.get("PLUGINS") or {}).get("host_agent") or {})
@@ -161,7 +166,10 @@ async def agent_metrics_by_host(db, host_names: list[str]) -> dict[str, dict]:
 async def agent_overview_by_host(
     db, host_names: list[str], stale_after_seconds: int = 180,
 ) -> dict[str, dict]:
-    """Per-host agent glance (cpu/mem/root-disk + freshness) from PluginMetric.
+    """Per-host agent glance (cpu/mem/disk + freshness) from PluginMetric.
+
+    {host_name: {cpu_pct, mem_used_pct, disk_used_pct_worst, disk_root,
+                 last_at, fresh}} for each host with any agent samples.
 
     Core-safe (no host_agent import). Freshness compares the latest host-level
     sample against the host_agent plugin's stale window (stale_after_seconds;
@@ -181,7 +189,7 @@ async def agent_overview_by_host(
         .where(
             PluginMetric.source_module == "host_agent",
             PluginMetric.resource_name.in_(resources),
-            PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct", "disk_used_pct")),
+            PluginMetric.metric_name.in_(AGENT_GLANCE_METRICS),
         )
         .group_by(PluginMetric.resource_name, PluginMetric.metric_name)
     ).subquery()
@@ -204,6 +212,7 @@ async def agent_overview_by_host(
                 d["_ts"] = r.recorded_at
     for d in out.values():
         ts = d.pop("_ts", None)
+        d["last_at"] = ts
         d["fresh"] = bool(ts and (now - ts).total_seconds() <= stale_after)
     return out
 
diff --git a/steward/mcp/server.py b/steward/mcp/server.py
index 9c8c2b4..f221e86 100644
--- a/steward/mcp/server.py
+++ b/steward/mcp/server.py
@@ -77,6 +77,8 @@ class StewardMCPServer(MCPServer):
         self._steward_app = app
         self.tool_roles: dict[str, UserRole] = {}
         self.tool_plugins: dict[str, str] = {}
+        # plugin name -> its `mcp_overview(db)`: findings it adds to get_overview.
+        self.overview_sources: dict[str, object] = {}
 
     # ── registration ─────────────────────────────────────────────────────────
     def register_tool(self, fn, *, role: UserRole, read_only: bool, plugin: str | None = None) -> None:
@@ -100,13 +102,18 @@ class StewardMCPServer(MCPServer):
             self.tool_plugins[name] = plugin
 
     def register_plugin(self, name: str, module) -> None:
-        """Add a plugin's tools via its optional `register_mcp_tools(register)`.
+        """Add a plugin's tools via its optional `register_mcp_tools(register)`,
+        and its optional `mcp_overview(db)` as a get_overview findings source.
 
         Idempotent per plugin (hot-reload drops its previous tools first) and
         tolerant: a raising hook is logged and leaves the plugin with no tools,
         never a broken server.
         """
         self._drop_plugin_tools(name)
+        self.overview_sources.pop(name, None)
+        overview = getattr(module, "mcp_overview", None)
+        if callable(overview):
+            self.overview_sources[name] = overview
         hook = getattr(module, "register_mcp_tools", None)
         if hook is None:
             return
@@ -124,7 +131,7 @@ class StewardMCPServer(MCPServer):
             self.tool_plugins.pop(tool_name, None)
 
     # ── gating ───────────────────────────────────────────────────────────────
-    def _plugin_enabled(self, plugin: str) -> bool:
+    def plugin_enabled(self, plugin: str) -> bool:
         cfg = (self._steward_app.config.get("PLUGINS") or {}).get(plugin)
         return isinstance(cfg, dict) and bool(cfg.get("enabled"))
 
@@ -134,7 +141,7 @@ class StewardMCPServer(MCPServer):
         if required is None:
             return f"There is no Steward tool named {name!r}."
         plugin = self.tool_plugins.get(name)
-        if plugin and not self._plugin_enabled(plugin):
+        if plugin and not self.plugin_enabled(plugin):
             return (f"{name} belongs to the {plugin} plugin, which is disabled in this "
                     "Steward. An admin can enable it under Settings → Plugins.")
         if caller is None:
diff --git a/steward/mcp/tools/__init__.py b/steward/mcp/tools/__init__.py
index 8f8005b..eb055b6 100644
--- a/steward/mcp/tools/__init__.py
+++ b/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 session
+from . import alerts, audit, hosts, metrics, monitors, overview, session
 
-_MODULES = (session,)
+_MODULES = (session, overview, hosts, monitors, alerts, metrics, audit)
 
 
 def register_all(register_tool) -> None:
diff --git a/steward/mcp/tools/_shape.py b/steward/mcp/tools/_shape.py
new file mode 100644
index 0000000..6971d81
--- /dev/null
+++ b/steward/mcp/tools/_shape.py
@@ -0,0 +1,149 @@
+"""Response shaping every tool shares: the conventions INSTRUCTIONS promise.
+
+- Times are ISO-8601 UTC with a `Z`, and anything with a "when" also carries
+  `age_s` so the agent never does clock arithmetic.
+- Every list is bounded: a tool returns `total`, `truncated`, and when cut, a
+  `more` line naming the argument that fetches the rest. Nothing is cut
+  silently.
+- A time series comes back as a summary plus a downsampled series, never raw
+  rows.
+- An entity is looked up by id or by name, and a miss raises ValueError naming
+  the tool that lists the valid choices (the server turns it into a readable
+  ToolError).
+"""
+from __future__ import annotations
+
+import difflib
+import math
+from datetime import datetime, timezone
+from typing import Callable, Iterable, TypeVar
+from urllib.parse import urlsplit, urlunsplit
+
+from steward.core.time_range import RANGES
+from steward.mcp._context import mcp_app
+
+T = TypeVar("T")
+
+SERIES_POINTS = 60  # max points in any returned series
+
+
+def db_session():
+    """A session on the calling Steward's database; use as `async with db_session() as db`."""
+    return mcp_app().db_sessionmaker()
+
+
+def now_utc() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _aware(dt: datetime) -> datetime:
+    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
+
+
+def iso(dt: datetime | None) -> str | None:
+    if dt is None:
+        return None
+    return _aware(dt).astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def age_s(dt: datetime | None, now: datetime | None = None) -> int | None:
+    if dt is None:
+        return None
+    return max(0, int(((now or now_utc()) - _aware(dt)).total_seconds()))
+
+
+def rnd(value, digits: int = 2):
+    return None if value is None else round(float(value), digits)
+
+
+def clamp_limit(limit: int, default: int, maximum: int) -> int:
+    """`limit` <= 0 means the default; anything above `maximum` is capped."""
+    if not limit or limit <= 0:
+        return default
+    return min(int(limit), maximum)
+
+
+def bounded(items: list, limit: int, *, key: str = "items", more: str = "") -> dict:
+    """`{key: items[:limit], total, truncated}`, plus `more` when something was cut."""
+    total = len(items)
+    out = {key: items[:limit], "total": total, "truncated": total > limit}
+    if total > limit:
+        out["more"] = f"Showing {limit} of {total}. " + (more or "Pass a larger `limit`.")
+    return out
+
+
+def lookback(range_: str) -> tuple[datetime, str]:
+    """A lookback window as (since, key). Unlike core parse_range, an unknown key
+    is refused rather than quietly read as 24h: the agent should learn its typo."""
+    key = (range_ or "24h").strip()
+    if key not in RANGES:
+        raise ValueError(f"range {range_!r} is not one of {', '.join(RANGES)}.")
+    return now_utc() - RANGES[key], key
+
+
+def downsample(points: list[tuple[datetime, float]], n: int = SERIES_POINTS) -> list[tuple[datetime, float]]:
+    """Average consecutive points into at most n, each stamped with its first time."""
+    if len(points) <= n:
+        return points
+    size = math.ceil(len(points) / n)
+    out = []
+    for i in range(0, len(points), size):
+        chunk = points[i:i + size]
+        out.append((chunk[0][0], sum(v for _, v in chunk) / len(chunk)))
+    return out
+
+
+def summarize(points: list[tuple[datetime, float]], n: int = SERIES_POINTS) -> dict:
+    """min/avg/max/p95/last over oldest-first (time, value) points, plus a ≤n series."""
+    if not points:
+        return {"points": 0, "series": []}
+    values = [v for _, v in points]
+    ordered = sorted(values)
+    p95 = ordered[max(0, math.ceil(0.95 * len(ordered)) - 1)]
+    return {
+        "points": len(values),
+        "min": rnd(ordered[0]),
+        "avg": rnd(sum(values) / len(values)),
+        "max": rnd(ordered[-1]),
+        "p95": rnd(p95),
+        "last": rnd(values[-1]),
+        "last_at": iso(points[-1][0]),
+        "series": [[iso(t), rnd(v)] for t, v in downsample(points, n)],
+    }
+
+
+def pick(want: str, rows: Iterable[T], *, kind: str, list_tool: str,
+         name: Callable[[T], str] = lambda r: r.name,
+         ident: Callable[[T], str] = lambda r: r.id) -> T:
+    """The row whose id is `want`, else the one whose name matches (case-insensitive)."""
+    rows = list(rows)
+    want = (want or "").strip()
+    if not want:
+        raise ValueError(f"pass a {kind} name or id; {list_tool} lists them.")
+    for r in rows:
+        if ident(r) == want:
+            return r
+    same = [r for r in rows if (name(r) or "").lower() == want.lower()]
+    if len(same) == 1:
+        return same[0]
+    if same:
+        ids = ", ".join(f"{ident(r)}" for r in same)
+        raise ValueError(f"{len(same)} {kind}s are named {want!r}; pass one of these ids instead: {ids}.")
+    close = difflib.get_close_matches(want.lower(), [(name(r) or "").lower() for r in rows], n=3, cutoff=0.5)
+    hint = f" Did you mean {', '.join(repr(c) for c in close)}?" if close else ""
+    raise ValueError(f"no {kind} named or with id {want!r}.{hint} {list_tool} lists them.")
+
+
+def redact_url(url: str) -> str:
+    """Drop a password from a URL's userinfo; the rest stays readable."""
+    try:
+        parts = urlsplit(url)
+    except ValueError:
+        return url
+    if parts.password is None:
+        return url
+    host = parts.hostname or ""
+    if parts.port:
+        host = f"{host}:{parts.port}"
+    netloc = f"{parts.username}:***@{host}" if parts.username else f"***@{host}"
+    return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
diff --git a/steward/mcp/tools/alerts.py b/steward/mcp/tools/alerts.py
new file mode 100644
index 0000000..99c9c3a
--- /dev/null
+++ b/steward/mcp/tools/alerts.py
@@ -0,0 +1,191 @@
+"""Alert tools: what is firing, what fired before, the rules, and maintenance windows."""
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from steward.alerts import service as alert_svc
+from steward.models.alerts import AlertEvent, AlertRule, AlertState
+from steward.models.users import User, UserRole
+
+from ._shape import (
+    age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, pick, rnd,
+)
+
+
+def _condition(rule: AlertRule) -> str:
+    n = rule.consecutive_failures_required or 1
+    tail = f" for {n} consecutive checks" if n > 1 else ""
+    return f"{rule.metric_name} {rule.operator.value} {rule.threshold:g}{tail}"
+
+
+def _action(rule: AlertRule) -> dict | None:
+    """The playbook a rule runs when it fires (alerts/routes.py _parse_ansible_action)."""
+    a = rule.ansible_action
+    if not a:
+        return None
+    return {k: a.get(k) for k in ("source", "playbook", "inventory", "limit", "tags",
+                                  "extra_vars", "check") if a.get(k) is not None}
+
+
+async def _usernames(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_alerts() -> dict:
+    """Alerts that need attention now: firing, pending, or acknowledged but still breaching.
+
+    pending: the condition holds but not yet for the rule's required number of
+    consecutive checks. firing: it has. acknowledged: someone has seen it and
+    it still holds. Each carries the rule's condition, the resource and metric
+    it watches, and `since` (its latest state change). Firing, unacknowledged
+    alerts come first. An alert that cleared is not here; get_alert_history
+    shows the transitions.
+    """
+    async with db_session() as db:
+        active = await alert_svc.active_alerts(db)
+        names = await _usernames(db, [s.acknowledged_by for s, _ in active])
+    order = {"firing": 0, "pending": 1, "acknowledged": 2}
+    rows = [
+        {
+            "rule_id": rule.id,
+            "rule_name": rule.name,
+            "state": state.state.value,
+            "source_module": rule.source_module,
+            "resource": rule.resource_name,
+            "metric": rule.metric_name,
+            "condition": _condition(rule),
+            "since": iso(state.last_transition_at),
+            "age_s": age_s(state.last_transition_at),
+            "acknowledged_by": names.get(state.acknowledged_by) if state.acknowledged_by else None,
+            "acknowledged_at": iso(state.acknowledged_at),
+        }
+        for state, rule in active
+    ]
+    rows.sort(key=lambda r: (order.get(r["state"], 3), -(r["age_s"] or 0)))
+    return {"alerts": rows, "total": len(rows)}
+
+
+async def get_alert_history(rule: str = "", range: str = "7d", limit: int = 50) -> dict:
+    """State transitions of alert rules, newest first: when each fired, cleared or was acked.
+
+    rule: one rule's name or id (list_alert_rules shows them); empty for every rule.
+    range: "1h", "6h", "24h", "7d" (default) or "30d".
+    limit: default 50, max 500.
+
+    Each transition carries the metric value that caused it and whether the
+    email/webhook notification went out (with the error if not), plus the id
+    of any Ansible run the rule's action triggered.
+    """
+    since, key = lookback(range)
+    limit = clamp_limit(limit, 50, 500)
+    async with db_session() as db:
+        rules = await alert_svc.list_rules(db)
+        stmt = (select(AlertEvent, AlertRule)
+                .join(AlertRule, AlertEvent.rule_id == AlertRule.id)
+                .where(AlertEvent.transitioned_at >= since)
+                .order_by(AlertEvent.transitioned_at.desc())
+                .limit(limit + 1))
+        chosen = None
+        if rule:
+            chosen = pick(rule, rules, kind="alert rule", list_tool="list_alert_rules")
+            stmt = stmt.where(AlertEvent.rule_id == chosen.id)
+        rows = (await db.execute(stmt)).all()
+    events = [
+        {
+            "rule_id": r.id,
+            "rule_name": r.name,
+            "from_state": e.from_state,
+            "to_state": e.to_state,
+            "metric_value": rnd(e.metric_value),
+            "at": iso(e.transitioned_at),
+            "age_s": age_s(e.transitioned_at),
+            "email_sent": e.email_sent,
+            "email_error": e.email_error,
+            "webhook_sent": e.webhook_sent,
+            "webhook_error": e.webhook_error,
+            "ansible_run_id": e.ansible_run_id,
+        }
+        for e, r in rows
+    ]
+    out = {"range": key, "rule": chosen.name if chosen else None}
+    out.update(bounded(events, limit, key="events",
+                       more="Pass a larger `limit` (max 500), a shorter `range`, or one `rule`."))
+    if out["truncated"]:
+        # Fetched limit+1 rows to detect the cut; the true total isn't known.
+        out["total"] = None
+        out["more"] = f"More than {limit} transitions in range. Pass a larger `limit` (max 500), a shorter `range`, or one `rule`."
+    return out
+
+
+async def list_alert_rules(enabled_only: bool = False, limit: int = 200) -> dict:
+    """Every alert rule: what it watches, its condition, current state and automatic action.
+
+    enabled_only: true to leave out disabled rules.
+    limit: default 200, max 500.
+
+    `state` is the rule's current state: "inactive" (condition not met),
+    "pending", "firing", "acknowledged", "resolved" (cleared since it last
+    fired), or null if it has never been evaluated. `ansible_action` is the
+    playbook the rule runs when it fires, if any (`check: true` = dry run).
+    """
+    limit = clamp_limit(limit, 200, 500)
+    async with db_session() as db:
+        rules = await alert_svc.list_rules(db)
+        states = {s.rule_id: s for s in (await db.execute(select(AlertState))).scalars()}
+    if enabled_only:
+        rules = [r for r in rules if r.enabled]
+    rows = [
+        {
+            "rule_id": r.id,
+            "rule_name": r.name,
+            "enabled": r.enabled,
+            "source_module": r.source_module,
+            "resource": r.resource_name,
+            "metric": r.metric_name,
+            "condition": _condition(r),
+            "state": states[r.id].state.value if r.id in states else None,
+            "state_since": iso(states[r.id].last_transition_at) if r.id in states else None,
+            "ansible_action": _action(r),
+        }
+        for r in rules
+    ]
+    return bounded(rows, limit, key="rules")
+
+
+async def list_maintenance_windows(include_past: bool = False, limit: int = 50) -> dict:
+    """Maintenance windows, during which matching alerts are held back from notifying.
+
+    include_past: true to include windows that have ended (default: current
+    and upcoming only).
+    limit: default 50, max 500.
+
+    `scope` is what a window silences: a source module and/or one resource, or
+    everything when both are null. `active` is true while it is in effect.
+    """
+    limit = clamp_limit(limit, 50, 500)
+    now = now_utc()
+    async with db_session() as db:
+        windows = await alert_svc.list_maintenance_windows(db)
+    rows = [
+        {
+            "window_id": w.id,
+            "name": w.name,
+            "start_at": iso(w.start_at),
+            "end_at": iso(w.end_at),
+            "active": w.start_at <= now <= w.end_at,
+            "scope": {"source_module": w.scope_source, "resource": w.scope_resource},
+        }
+        for w in windows
+        if include_past or w.end_at >= now
+    ]
+    return bounded(rows, limit, key="windows")
+
+
+def register(register_tool) -> None:
+    register_tool(list_alerts, role=UserRole.viewer, read_only=True)
+    register_tool(get_alert_history, role=UserRole.viewer, read_only=True)
+    register_tool(list_alert_rules, role=UserRole.viewer, read_only=True)
+    register_tool(list_maintenance_windows, role=UserRole.viewer, read_only=True)
diff --git a/steward/mcp/tools/audit.py b/steward/mcp/tools/audit.py
new file mode 100644
index 0000000..3428f3b
--- /dev/null
+++ b/steward/mcp/tools/audit.py
@@ -0,0 +1,71 @@
+"""The audit log: who changed what in Steward, including changes made through MCP."""
+from __future__ import annotations
+
+import json
+
+from sqlalchemy import select
+
+from steward.models.audit import AuditEvent
+from steward.models.users import UserRole
+
+from ._shape import age_s, bounded, clamp_limit, db_session, iso, lookback
+
+
+async def list_audit_events(action: str = "", user: str = "", entity_type: str = "",
+                            range: str = "7d", limit: int = 50) -> dict:
+    """Changes recorded in Steward's audit log, newest first.
+
+    action: a prefix of the action name, e.g. "monitor." or "ansible.run";
+      empty for all.
+    user: a username, exactly; "system" for automated changes.
+    entity_type: e.g. "monitor", "host", "alert_rule"; empty for all.
+    range: "1h", "6h", "24h", "7d" (default) or "30d".
+    limit: default 50, max 500.
+
+    Answers "what changed before this broke?": line an event's `at` up against
+    a monitor's down_since or an alert's since.
+    """
+    since, key = lookback(range)
+    limit = clamp_limit(limit, 50, 500)
+    stmt = (select(AuditEvent).where(AuditEvent.timestamp >= since)
+            .order_by(AuditEvent.timestamp.desc()).limit(limit + 1))
+    if action:
+        esc = action.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        stmt = stmt.where(AuditEvent.action.like(esc + "%", escape="\\"))
+    if user:
+        stmt = stmt.where(AuditEvent.username == user)
+    if entity_type:
+        stmt = stmt.where(AuditEvent.entity_type == entity_type)
+    async with db_session() as db:
+        rows = list((await db.execute(stmt)).scalars())
+
+    def _detail(raw):
+        if not raw:
+            return None
+        try:
+            return json.loads(raw)
+        except ValueError:
+            return raw[:500]
+
+    events = [
+        {
+            "at": iso(e.timestamp),
+            "age_s": age_s(e.timestamp),
+            "user": e.username,
+            "action": e.action,
+            "entity_type": e.entity_type,
+            "entity_id": e.entity_id,
+            "detail": _detail(e.detail_json),
+        }
+        for e in rows
+    ]
+    out = {"range": key, **bounded(events, limit, key="events")}
+    if out["truncated"]:
+        out["total"] = None
+        out["more"] = (f"More than {limit} events in range. Pass a larger `limit` (max 500), "
+                       "a shorter `range`, or filter by `action`, `user` or `entity_type`.")
+    return out
+
+
+def register(register_tool) -> None:
+    register_tool(list_audit_events, role=UserRole.viewer, read_only=True)
diff --git a/steward/mcp/tools/hosts.py b/steward/mcp/tools/hosts.py
new file mode 100644
index 0000000..0ea4d5e
--- /dev/null
+++ b/steward/mcp/tools/hosts.py
@@ -0,0 +1,209 @@
+"""Host tools: the fleet, one host in depth, and a host's agent metrics."""
+from __future__ import annotations
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from steward.alerts import service as alert_svc
+from steward.core import metric_series
+from steward.hosts import service as host_svc
+from steward.mcp._context import mcp_app
+from steward.models.ansible_inventory import AnsibleTarget
+from steward.models.hosts import Host
+from steward.models.users import UserRole
+
+from ._shape import (
+    age_s, bounded, clamp_limit, db_session, iso, lookback, pick, rnd, summarize,
+)
+
+AGENT_SOURCE = "host_agent"
+
+
+async def _all_hosts(db) -> list[Host]:
+    return list((await db.execute(select(Host).order_by(Host.name))).scalars())
+
+
+def _agent_glance(a: dict | None) -> dict | None:
+    if not a:
+        return None
+    return {
+        "reporting": a.get("fresh", False),
+        "last_seen_at": iso(a.get("last_at")),
+        "last_seen_age_s": age_s(a.get("last_at")),
+        "cpu_pct": rnd(a.get("cpu_pct")),
+        "mem_used_pct": rnd(a.get("mem_used_pct")),
+        "disk_used_pct_worst": rnd(a.get("disk_used_pct_worst")),
+        "disk_root_used_pct": rnd(a.get("disk_root")),
+    }
+
+
+async def list_hosts(limit: int = 100) -> dict:
+    """Every registered host with its health at a glance.
+
+    For each host: `status` rolled up from its linked monitors ("down" if any
+    monitor's latest check failed, "up", "pending" when none has run yet,
+    "unmonitored" when it has no monitors), the best latency, uptime % over
+    24h/7d/30d, and — when a host agent reports for it — CPU, memory and disk
+    utilisation and whether the agent is still reporting (`agent.reporting`;
+    false means its last sample is older than the stale window).
+
+    Use get_host for one host's monitors, alerts and Ansible link, and
+    get_host_metrics for a metric's history. limit: default 100, max 500.
+    """
+    limit = clamp_limit(limit, 100, 500)
+    async with db_session() as db:
+        hosts = await _all_hosts(db)
+        status = await host_svc.host_status(db, [h.id for h in hosts])
+        uptime = await host_svc.compute_uptime(db)
+        agent = await host_svc.agent_overview_by_host(
+            db, [h.name for h in hosts], host_svc.agent_stale_after_seconds(mcp_app().config))
+    rows = []
+    for h in hosts:
+        st = status.get(h.id)
+        rows.append({
+            "host_id": h.id,
+            "host_name": h.name,
+            "address": h.address,
+            "status": st["state"] if st else "unmonitored",
+            "monitor_count": st["count"] if st else 0,
+            "best_latency_ms": rnd(st["latency_ms"]) if st else None,
+            "uptime_pct": uptime.get(h.id) or {"24h": None, "7d": None, "30d": None},
+            "agent": _agent_glance(agent.get(h.name)),
+        })
+    return bounded(rows, limit, key="hosts")
+
+
+async def get_host(host: str) -> dict:
+    """One host in depth: its monitors, agent snapshot, active alerts and Ansible link.
+
+    host: the host's name (case-insensitive) or id, as list_hosts shows.
+
+    Returns the host's monitors (latest result, error, uptime), the agent's
+    latest values — host-level metrics plus each sub-resource such as a mount
+    or interface under `agent.parts` — alerts currently pending, firing or
+    acknowledged on this host's resources, and the Ansible inventory target
+    linked to it with its groups (`ansible_target`: null when unlinked).
+    """
+    async with db_session() as db:
+        h = pick(host, await _all_hosts(db), kind="host", list_tool="list_hosts")
+        monitors = await host_svc.host_monitors(db, h.id)
+        uptime = (await host_svc.compute_uptime(db)).get(h.id)
+        status = (await host_svc.host_status(db, [h.id])).get(h.id)
+        glance = (await host_svc.agent_overview_by_host(
+            db, [h.name], host_svc.agent_stale_after_seconds(mcp_app().config))).get(h.name)
+        samples = await metric_series.latest_values(db, AGENT_SOURCE, h.name, include_children=True)
+        active = await alert_svc.active_alerts(db)
+        target = (await db.execute(
+            select(AnsibleTarget).where(AnsibleTarget.host_id == h.id)
+            .options(selectinload(AnsibleTarget.groups))
+        )).scalar_one_or_none()
+
+    agent = None
+    if samples:
+        host_level: dict = {}
+        parts: dict[str, dict] = {}
+        for s in samples:
+            if s.resource_name == h.name:
+                host_level[s.metric_name] = rnd(s.value)
+            else:
+                parts.setdefault(s.resource_name[len(h.name) + 1:], {})[s.metric_name] = rnd(s.value)
+        agent = {
+            **(_agent_glance(glance) or {}),
+            "metrics": host_level,
+            "parts": dict(list(sorted(parts.items()))[:40]),
+            "parts_total": len(parts),
+        }
+
+    prefix = h.name + ":"
+    alerts = [
+        {
+            "rule_id": rule.id, "rule_name": rule.name, "state": state.state.value,
+            "resource": rule.resource_name, "metric": rule.metric_name,
+            "condition": f"{rule.metric_name} {rule.operator.value} {rule.threshold:g}",
+            "since": iso(state.last_transition_at), "age_s": age_s(state.last_transition_at),
+        }
+        for state, rule in active
+        if rule.resource_name == h.name or rule.resource_name.startswith(prefix)
+    ]
+
+    return {
+        "host_id": h.id,
+        "host_name": h.name,
+        "address": h.address,
+        "status": status["state"] if status else "unmonitored",
+        "uptime_pct": uptime or {"24h": None, "7d": None, "30d": None},
+        "monitors": [
+            {
+                "monitor_id": d["monitor"].id,
+                "monitor_name": d["monitor"].name,
+                "type": d["monitor"].type,
+                "target": d["monitor"].target,
+                "enabled": d["monitor"].enabled,
+                "state": ("pending" if d["latest"] is None
+                          else "up" if d["latest"].is_up else "down"),
+                "last_checked_at": iso(d["latest"].checked_at) if d["latest"] else None,
+                "response_ms": rnd(d["latest"].response_ms) if d["latest"] else None,
+                "error": d["latest"].error_msg if d["latest"] and not d["latest"].is_up else None,
+                "uptime_24h_pct": d["uptime_pct"],
+            }
+            for d in monitors
+        ],
+        "agent": agent,
+        "active_alerts": alerts,
+        "ansible_target": {
+            "target_id": target.id,
+            "target_name": target.name,
+            "address": target.address,
+            "groups": sorted(g.name for g in target.groups),
+        } if target else None,
+    }
+
+
+async def get_host_metrics(host: str, metric: str = "", range: str = "24h") -> dict:
+    """A host agent metric over time: summary statistics plus a ≤60-point series.
+
+    host: name or id, as list_hosts shows.
+    metric: e.g. "cpu_pct", "mem_used_pct", "disk_used_pct_worst", "load_1m",
+      "net_rx_bps", "temp_c_max". Empty lists the metrics this host reports,
+      with their latest values. For a mount or interface, use query_metrics
+      with source_module="host_agent" and resource="<host>:<part>" (get_host
+      lists the parts).
+    range: "1h", "6h", "24h" (default), "7d" or "30d".
+
+    The summary carries min, avg, max, p95 and last over the range; each series
+    point is [time, value].
+    """
+    since, key = lookback(range)
+    async with db_session() as db:
+        h = pick(host, await _all_hosts(db), kind="host", list_tool="list_hosts")
+        if not metric:
+            latest = await metric_series.latest_values(db, AGENT_SOURCE, h.name)
+            if not latest:
+                raise ValueError(f"no agent metrics recorded for host {h.name!r}; "
+                                 "is a host agent installed and reporting? list_hosts shows `agent`.")
+            return {
+                "host_id": h.id, "host_name": h.name,
+                "metrics": [
+                    {"metric": s.metric_name, "last": rnd(s.value),
+                     "last_at": iso(s.recorded_at), "age_s": age_s(s.recorded_at)}
+                    for s in latest
+                ],
+                "next": "Pass one of these as `metric` for its history.",
+            }
+        pts = await metric_series.points(db, AGENT_SOURCE, h.name, metric, since)
+        if not pts:
+            names = [s.metric_name for s in await metric_series.latest_values(db, AGENT_SOURCE, h.name)]
+            if metric not in names:
+                hint = (f" It reports: {', '.join(names)}." if names
+                        else " No agent metrics are recorded for it at all.")
+                raise ValueError(f"host {h.name!r} has no metric {metric!r}.{hint}")
+    return {
+        "host_id": h.id, "host_name": h.name, "metric": metric, "range": key,
+        **summarize(pts),
+    }
+
+
+def register(register_tool) -> None:
+    register_tool(list_hosts, role=UserRole.viewer, read_only=True)
+    register_tool(get_host, role=UserRole.viewer, read_only=True)
+    register_tool(get_host_metrics, role=UserRole.viewer, read_only=True)
diff --git a/steward/mcp/tools/metrics.py b/steward/mcp/tools/metrics.py
new file mode 100644
index 0000000..6c87731
--- /dev/null
+++ b/steward/mcp/tools/metrics.py
@@ -0,0 +1,71 @@
+"""Generic metric access: anything recorded in Steward's metric store."""
+from __future__ import annotations
+
+from steward.core import metric_series
+from steward.models.users import UserRole
+
+from ._shape import age_s, bounded, clamp_limit, db_session, iso, lookback, rnd, summarize
+
+
+async def query_metrics(source_module: str = "", resource: str = "", metric: str = "",
+                        range: str = "24h", limit: int = 200) -> dict:
+    """Read Steward's metric store, narrowing one argument at a time.
+
+    Every plugin and monitor type records into one store keyed by
+    (source_module, resource, metric): host_agent, docker, traefik, unifi,
+    snmp, the monitor types (icmp/tcp/dns/http) and Steward's own self-monitor.
+
+    - No arguments: the source modules that have data.
+    - source_module only: its resources (hosts, routers, containers, devices).
+      A host agent's sub-resources are named "<host>:<part>", e.g. "nas:/".
+    - source_module + resource: the latest value of each metric it reports.
+    - all three: that metric over `range` ("1h", "6h", "24h" default, "7d",
+      "30d") as min/avg/max/p95/last plus a ≤60-point [time, value] series.
+
+    limit bounds the resource list (default 200, max 1000).
+    """
+    limit = clamp_limit(limit, 200, 1000)
+    async with db_session() as db:
+        if not source_module:
+            return {"source_modules": await metric_series.sources(db),
+                    "next": "Pass one as `source_module` to list its resources."}
+
+        if not resource:
+            names = await metric_series.resources(db, source_module)
+            if not names:
+                known = await metric_series.sources(db)
+                raise ValueError(f"no metrics recorded for source_module {source_module!r}. "
+                                 f"Sources with data: {', '.join(known) or 'none'}.")
+            return {"source_module": source_module,
+                    **bounded(names, limit, key="resources",
+                              more="Pass a larger `limit` (max 1000)."),
+                    "next": "Pass one as `resource` to see its metrics."}
+
+        if not metric:
+            latest = await metric_series.latest_values(db, source_module, resource)
+            if not latest:
+                raise ValueError(f"no metrics for resource {resource!r} in {source_module!r}; "
+                                 "call query_metrics with only source_module to list its resources.")
+            return {
+                "source_module": source_module, "resource": resource,
+                "metrics": [
+                    {"metric": s.metric_name, "last": rnd(s.value),
+                     "last_at": iso(s.recorded_at), "age_s": age_s(s.recorded_at)}
+                    for s in latest
+                ],
+                "next": "Pass one as `metric` for its history.",
+            }
+
+        since, key = lookback(range)
+        pts = await metric_series.points(db, source_module, resource, metric, since)
+        if not pts:
+            names = [s.metric_name for s in await metric_series.latest_values(db, source_module, resource)]
+            if metric not in names:
+                raise ValueError(f"{resource!r} in {source_module!r} has no metric {metric!r}. "
+                                 f"It reports: {', '.join(names) or 'nothing'}.")
+    return {"source_module": source_module, "resource": resource, "metric": metric,
+            "range": key, **summarize(pts)}
+
+
+def register(register_tool) -> None:
+    register_tool(query_metrics, role=UserRole.viewer, read_only=True)
diff --git a/steward/mcp/tools/monitors.py b/steward/mcp/tools/monitors.py
new file mode 100644
index 0000000..db4376b
--- /dev/null
+++ b/steward/mcp/tools/monitors.py
@@ -0,0 +1,142 @@
+"""Monitor tools: every uptime check, and one check's failures in depth."""
+from __future__ import annotations
+
+import json
+
+from sqlalchemy import select
+
+from steward.models.hosts import Host
+from steward.models.monitors import MONITOR_TYPE_VALUES, Monitor
+from steward.models.users import UserRole
+from steward.monitors import service as monitor_svc
+
+from ._shape import (
+    age_s, bounded, clamp_limit, db_session, iso, now_utc, pick, redact_url, rnd,
+)
+
+_STATES = ("up", "down", "pending", "disabled")
+
+
+def _state(m: Monitor, latest) -> str:
+    if not m.enabled:
+        return "disabled"
+    if latest is None:
+        return "pending"
+    return "up" if latest.is_up else "down"
+
+
+def _tls_days(latest) -> float | None:
+    if latest is None or latest.tls_expires_at is None:
+        return None
+    return rnd((latest.tls_expires_at - now_utc()).total_seconds() / 86400.0, 1)
+
+
+def _row(m: Monitor, st: dict, host_names: dict[str, str]) -> dict:
+    latest = st["latest"]
+    return {
+        "monitor_id": m.id,
+        "monitor_name": m.name,
+        "type": m.type,
+        "target": redact_url(m.target),
+        "host_id": m.host_id,
+        "host_name": host_names.get(m.host_id) if m.host_id else None,
+        "state": _state(m, latest),
+        "down_since": iso(st["down_since"]),
+        "down_for_s": age_s(st["down_since"]),
+        "last_checked_at": iso(latest.checked_at) if latest else None,
+        "response_ms": rnd(latest.response_ms) if latest else None,
+        "error": latest.error_msg if latest is not None and not latest.is_up else None,
+        "uptime_pct": st["uptime"] or {"24h": None, "7d": None, "30d": None},
+        "tls_expires_in_days": _tls_days(latest),
+    }
+
+
+async def _load(db):
+    monitors = list((await db.execute(select(Monitor).order_by(Monitor.name))).scalars())
+    host_names = dict((await db.execute(select(Host.id, Host.name))).all())
+    return monitors, host_names
+
+
+async def list_monitors(state: str = "", type: str = "", limit: int = 100) -> dict:
+    """Every uptime monitor (ping, TCP, DNS, HTTP) with its current state.
+
+    state: "down", "up", "pending" (never checked yet) or "disabled"; empty for all.
+    type: "icmp", "tcp", "dns" or "http"; empty for all.
+    limit: default 100, max 500.
+
+    Down monitors come first, each with `down_since` (when its current run of
+    failures began) and the latest `error`. `uptime_pct` covers 24h/7d/30d;
+    `tls_expires_in_days` is set for HTTPS checks. get_monitor shows one
+    monitor's configuration and recent failures.
+    """
+    if state and state not in _STATES:
+        raise ValueError(f"state {state!r} is not one of {', '.join(_STATES)}.")
+    if type and type not in MONITOR_TYPE_VALUES:
+        raise ValueError(f"type {type!r} is not one of {', '.join(MONITOR_TYPE_VALUES)}.")
+    limit = clamp_limit(limit, 100, 500)
+    async with db_session() as db:
+        monitors, host_names = await _load(db)
+        if type:
+            monitors = [m for m in monitors if m.type == type]
+        states = await monitor_svc.monitor_states(db, monitors)
+    rows = [_row(m, states[m.id], host_names) for m in monitors]
+    if state:
+        rows = [r for r in rows if r["state"] == state]
+    order = {"down": 0, "pending": 1, "up": 2, "disabled": 3}
+    rows.sort(key=lambda r: (order[r["state"]], r["monitor_name"].lower()))
+    counts = {s: 0 for s in _STATES}
+    for r in rows:
+        counts[r["state"]] += 1
+    return {"counts": counts, **bounded(rows, limit, key="monitors",
+                                        more="Pass a larger `limit`, or filter by `state` or `type`.")}
+
+
+def _config(m: Monitor) -> dict:
+    try:
+        cfg = json.loads(m.config_json or "{}")
+    except ValueError:
+        return {}
+    if isinstance(cfg.get("headers"), dict):
+        # Header values are often credentials (Authorization, API keys).
+        cfg["headers"] = {k: "<redacted>" for k in cfg["headers"]}
+    return cfg
+
+
+async def get_monitor(monitor: str, failures: int = 10) -> dict:
+    """One monitor in depth: configuration, current state, uptime and recent failures.
+
+    monitor: the monitor's name (case-insensitive) or id, as list_monitors shows.
+    failures: how many of its latest failed checks to return (default 10, max 50),
+      newest first, each with its error message.
+
+    `heartbeat` is its last 30 checks oldest-first as a string, "+" up and "-"
+    down, to show a flapping check at a glance. Header values in an HTTP
+    monitor's configuration are redacted.
+    """
+    failures = clamp_limit(failures, 10, 50)
+    async with db_session() as db:
+        monitors, host_names = await _load(db)
+        m = pick(monitor, monitors, kind="monitor", list_tool="list_monitors")
+        st = (await monitor_svc.monitor_states(db, [m]))[m.id]
+        failed = await monitor_svc.recent_failures(db, m.id, failures)
+    return {
+        **_row(m, st, host_names),
+        "check_interval_s": m.check_interval_seconds or None,
+        "config": _config(m),
+        "heartbeat": "".join("+" if r.is_up else "-" for r in st["recent"]),
+        "recent_failures": [
+            {
+                "checked_at": iso(r.checked_at),
+                "age_s": age_s(r.checked_at),
+                "error": r.error_msg,
+                "status_code": r.status_code,
+                "response_ms": rnd(r.response_ms),
+            }
+            for r in failed
+        ],
+    }
+
+
+def register(register_tool) -> None:
+    register_tool(list_monitors, role=UserRole.viewer, read_only=True)
+    register_tool(get_monitor, role=UserRole.viewer, read_only=True)
diff --git a/steward/mcp/tools/overview.py b/steward/mcp/tools/overview.py
new file mode 100644
index 0000000..761986c
--- /dev/null
+++ b/steward/mcp/tools/overview.py
@@ -0,0 +1,221 @@
+"""get_overview: what is wrong right now, ranked, across everything Steward watches."""
+from __future__ import annotations
+
+import logging
+from datetime import timedelta
+
+from sqlalchemy import select
+
+from steward.alerts import service as alert_svc
+from steward.hosts import service as host_svc
+from steward.mcp._context import mcp_app
+from steward.models.ansible import AnsibleRun, AnsibleRunStatus
+from steward.models.hosts import Host
+from steward.models.monitors import Monitor
+from steward.models.users import UserRole
+from steward.monitors import service as monitor_svc
+
+from ._shape import age_s, db_session, iso, now_utc, rnd
+
+logger = logging.getLogger(__name__)
+
+TLS_WARN_DAYS = 14
+FINDINGS_CAP = 50
+_SEVERITY = {"critical": 0, "warning": 1}
+
+
+def finding(severity: str, kind: str, subject: str, summary: str, *, since=None,
+            ids: dict | None = None, next: str | None = None) -> dict:
+    """One overview finding. Plugins' `mcp_overview(db)` return a list of these."""
+    return {
+        "severity": severity, "kind": kind, "subject": subject, "summary": summary,
+        "since": iso(since), "age_s": age_s(since), "ids": ids or {}, "next": next,
+    }
+
+
+def rank(findings: list[dict]) -> list[dict]:
+    """Critical first; within a severity, the longest-standing first."""
+    return sorted(findings, key=lambda f: (_SEVERITY.get(f.get("severity"), 2),
+                                           -(f.get("age_s") or 0), f.get("subject", "")))
+
+
+def monitor_findings(monitors, states: dict, host_names: dict, now) -> list[dict]:
+    out = []
+    for m in monitors:
+        if not m.enabled:
+            continue
+        st = states.get(m.id) or {}
+        latest = st.get("latest")
+        if latest is None:
+            continue
+        where = f" on {host_names[m.host_id]}" if m.host_id in host_names else ""
+        ids = {"monitor_id": m.id, "host_id": m.host_id}
+        if not latest.is_up:
+            out.append(finding(
+                "critical", "monitor_down", m.name,
+                f"{m.type} check of {m.target}{where} is failing: {latest.error_msg or 'down'}",
+                since=st.get("down_since") or latest.checked_at, ids=ids,
+                next=f'get_monitor(monitor="{m.id}")'))
+        elif latest.tls_expires_at is not None:
+            days = (latest.tls_expires_at - now).total_seconds() / 86400.0
+            if days <= TLS_WARN_DAYS:
+                out.append(finding(
+                    "critical" if days <= 3 else "warning", "tls_expiring", m.name,
+                    f"TLS certificate for {m.target} expires in {days:.1f} days",
+                    ids=ids, next=f'get_monitor(monitor="{m.id}")'))
+    return out
+
+
+def alert_findings(active) -> list[dict]:
+    out = []
+    for state, rule in active:
+        s = state.state.value
+        if s == "firing":
+            sev, what = "critical", "is firing"
+        elif s == "pending":
+            sev, what = "warning", "is pending (breaching, not yet for long enough to fire)"
+        else:
+            sev, what = "warning", "is acknowledged but still breaching"
+        out.append(finding(
+            sev, f"alert_{s}", rule.name,
+            f"{rule.metric_name} {rule.operator.value} {rule.threshold:g} on "
+            f"{rule.source_module}/{rule.resource_name} {what}",
+            since=state.last_transition_at, ids={"rule_id": rule.id},
+            next=f'get_alert_history(rule="{rule.id}")'))
+    return out
+
+
+_AGENT_CHECKS = (("cpu_pct", "cpu", "CPU"), ("mem_used_pct", "mem", "memory"),
+                 ("disk_used_pct_worst", "disk", "fullest disk"))
+
+
+def agent_findings(hosts, agent: dict, thresholds: dict) -> list[dict]:
+    out = []
+    for h in hosts:
+        a = agent.get(h.name)
+        if not a:
+            continue
+        ids = {"host_id": h.id}
+        if not a.get("fresh"):
+            out.append(finding(
+                "warning", "agent_stale", h.name,
+                "host agent has stopped reporting; its metrics are out of date",
+                since=a.get("last_at"), ids=ids, next=f'get_host(host="{h.id}")'))
+            continue  # stale values would only add noise
+        for metric, kind, label in _AGENT_CHECKS:
+            value = a.get(metric)
+            th = thresholds.get(kind) or {}
+            if value is None or th.get("warn") is None:
+                continue
+            if th.get("crit") is not None and value >= th["crit"]:
+                sev = "critical"
+            elif value >= th["warn"]:
+                sev = "warning"
+            else:
+                continue
+            out.append(finding(
+                sev, f"host_{kind}_high", h.name,
+                f"{label} at {rnd(value, 1)}% (warn {th['warn']}%, crit {th.get('crit')}%)",
+                ids=ids, next=f'get_host_metrics(host="{h.id}", metric="{metric}")'))
+    return out
+
+
+def run_findings(runs) -> list[dict]:
+    out = []
+    for r in runs:
+        failures = ((r.results or {}).get("failures") or [])[:3]
+        detail = f": {'; '.join(str(f)[:200] for f in failures)}" if failures else ""
+        out.append(finding(
+            "warning", f"ansible_run_{r.status.value}", r.playbook_path,
+            f"Ansible run of {r.playbook_path} ({r.source_name}) {r.status.value}{detail}",
+            since=r.finished_at or r.started_at, ids={"run_id": r.id}))
+    return out
+
+
+async def _plugin_findings(server) -> tuple[list[dict], list[str]]:
+    found, unavailable = [], []
+    for plugin, source in sorted((getattr(server, "overview_sources", None) or {}).items()):
+        if not server.plugin_enabled(plugin):
+            continue
+        try:
+            async with db_session() as db:
+                found.extend(await source(db))
+        except Exception:
+            logger.exception("MCP overview source of plugin %r failed", plugin)
+            unavailable.append(plugin)
+    return found, unavailable
+
+
+async def get_overview() -> dict:
+    """What is wrong right now, across everything Steward watches. Start here.
+
+    Returns `findings` ranked critical first, then longest-standing first:
+    failing monitors (with the error and since when), firing, pending and
+    still-breaching acknowledged alerts, TLS certificates expiring within
+    14 days, host agents that stopped reporting, hosts over the CPU, memory or
+    disk thresholds set under Settings → Thresholds, Ansible runs that failed
+    in the last 24h, and anything enabled plugins add (unhealthy containers,
+    for one). Each finding carries the ids to follow and `next`, the call
+    that shows its evidence.
+
+    `counts` sizes the fleet, and `active_maintenance` names windows
+    currently silencing alerts. `all_clear` is true when nothing is wrong.
+    """
+    app = mcp_app()
+    now = now_utc()
+    async with db_session() as db:
+        hosts = list((await db.execute(select(Host).order_by(Host.name))).scalars())
+        host_names = {h.id: h.name for h in hosts}
+        monitors = list((await db.execute(select(Monitor).order_by(Monitor.name))).scalars())
+        states = await monitor_svc.monitor_states(db, [m for m in monitors if m.enabled])
+        active = await alert_svc.active_alerts(db)
+        agent = await host_svc.agent_overview_by_host(
+            db, [h.name for h in hosts], host_svc.agent_stale_after_seconds(app.config))
+        runs = list((await db.execute(
+            select(AnsibleRun)
+            .where(AnsibleRun.status.in_((AnsibleRunStatus.failed, AnsibleRunStatus.interrupted)),
+                   AnsibleRun.started_at >= now - timedelta(hours=24))
+            .order_by(AnsibleRun.started_at.desc())
+            .limit(20)
+        )).scalars())
+        windows = await alert_svc.list_maintenance_windows(db)
+
+    findings = (monitor_findings(monitors, states, host_names, now)
+                + alert_findings(active)
+                + agent_findings(hosts, agent, app.config.get("THRESHOLDS") or {})
+                + run_findings(runs))
+    plugin_found, unavailable = await _plugin_findings(getattr(app, "mcp_server", None))
+    findings = rank(findings + plugin_found)
+
+    down = sum(1 for m in monitors if m.enabled and (st := states.get(m.id))
+               and st["latest"] is not None and not st["latest"].is_up)
+    out = {
+        "generated_at": iso(now),
+        "all_clear": not findings,
+        "counts": {
+            "hosts": len(hosts),
+            "monitors": len(monitors),
+            "monitors_down": down,
+            "alerts_active": len(active),
+            "critical": sum(1 for f in findings if f["severity"] == "critical"),
+            "warning": sum(1 for f in findings if f["severity"] == "warning"),
+        },
+        "active_maintenance": [
+            {"window_id": w.id, "name": w.name, "end_at": iso(w.end_at),
+             "scope": {"source_module": w.scope_source, "resource": w.scope_resource}}
+            for w in windows if w.start_at <= now <= w.end_at
+        ],
+        "findings": findings[:FINDINGS_CAP],
+        "truncated": len(findings) > FINDINGS_CAP,
+    }
+    if out["truncated"]:
+        out["more"] = (f"{len(findings) - FINDINGS_CAP} lower-ranked findings left out; "
+                       "list_monitors(state=\"down\"), list_alerts and list_hosts show everything.")
+    if unavailable:
+        out["unavailable"] = [f"{p} plugin's findings could not be read (see Steward's log)"
+                              for p in unavailable]
+    return out
+
+
+def register(register_tool) -> None:
+    register_tool(get_overview, role=UserRole.viewer, read_only=True)
diff --git a/steward/monitors/service.py b/steward/monitors/service.py
index 2c3d92e..e2653e0 100644
--- a/steward/monitors/service.py
+++ b/steward/monitors/service.py
@@ -8,11 +8,11 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from steward.core.status import _heartbeat_title, _last_n_results, _uptime_by_monitor
 from steward.core.time_range import parse_range
-from steward.models.monitors import Monitor
+from steward.models.monitors import Monitor, MonitorResult
 
 
 async def monitor_rows(db, range_key: str):
@@ -50,3 +50,54 @@ async def monitor_rows(db, range_key: str):
     down = sum(1 for d in data if d["latest"] and not d["latest"].is_up)
     pending = sum(1 for d in data if not d["latest"])
     return data, up, down, pending, range_key
+
+
+async def down_since(db, monitor_id: str):
+    """When the monitor's current run of failures began, or None if its latest check is up.
+
+    The first failed check after the last success; if it has never succeeded,
+    its first check.
+    """
+    last_up = (await db.execute(
+        select(func.max(MonitorResult.checked_at))
+        .where(MonitorResult.monitor_id == monitor_id, MonitorResult.is_up.is_(True))
+    )).scalar()
+    stmt = select(func.min(MonitorResult.checked_at)).where(MonitorResult.monitor_id == monitor_id)
+    if last_up is not None:
+        stmt = stmt.where(MonitorResult.checked_at > last_up)
+    return (await db.execute(stmt)).scalar()
+
+
+async def monitor_states(db, monitors: list[Monitor]) -> dict[str, dict]:
+    """{monitor_id: {latest, recent, uptime, down_since}} for the given monitors.
+
+    `recent` is the last HEARTBEAT_COUNT results oldest-first; `uptime` is
+    {"24h","7d","30d"} percentages; `down_since` is set only for a monitor
+    whose latest check failed.
+    """
+    ids = [m.id for m in monitors]
+    recent = await _last_n_results(db, ids)
+    uptime = await _uptime_by_monitor(db, ids)
+    out: dict[str, dict] = {}
+    for m in monitors:
+        rows = recent.get(m.id, [])
+        latest = rows[-1] if rows else None
+        out[m.id] = {
+            "latest": latest,
+            "recent": rows,
+            "uptime": uptime.get(m.id, {}),
+            "down_since": None,
+        }
+        if latest is not None and not latest.is_up:
+            out[m.id]["down_since"] = await down_since(db, m.id)
+    return out
+
+
+async def recent_failures(db, monitor_id: str, limit: int = 10) -> list[MonitorResult]:
+    """The monitor's latest failed checks, newest first."""
+    return list((await db.execute(
+        select(MonitorResult)
+        .where(MonitorResult.monitor_id == monitor_id, MonitorResult.is_up.is_(False))
+        .order_by(MonitorResult.checked_at.desc())
+        .limit(limit)
+    )).scalars())
diff --git a/tests/integration/test_mcp_tools.py b/tests/integration/test_mcp_tools.py
new file mode 100644
index 0000000..9e38b3e
--- /dev/null
+++ b/tests/integration/test_mcp_tools.py
@@ -0,0 +1,201 @@
+"""Integration: the MCP diagnostic tools against real Postgres.
+
+Seeds one small, uniquely named slice of a fleet — a host with a failing ping
+monitor, an agent reporting high CPU, a firing alert, an HTTPS monitor whose
+certificate is about to expire, a failed Ansible run, an active maintenance
+window and an audit event — then calls each tool as a viewer and checks that
+its slice comes back shaped as the tool promises. The database is shared with
+other tests, so assertions look for the seeded rows rather than exact totals.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+
+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)",
+)
+
+
+@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_diagnostic_tools_over_a_seeded_fleet(app):
+    from steward.core.api_tokens import Caller
+    from steward.core.audit import log_audit
+    from steward.mcp import _context
+    from steward.mcp.tools import alerts, audit, hosts, metrics, monitors, overview
+    from steward.models.alerts import (
+        AlertEvent, AlertOperator, AlertRule, AlertState, AlertStateEnum, MaintenanceWindow,
+    )
+    from steward.models.ansible import AnsibleRun, AnsibleRunStatus
+    from steward.models.hosts import Host
+    from steward.models.metrics import PluginMetric
+    from steward.models.monitors import Monitor, MonitorResult
+    from steward.models.users import User, UserRole
+
+    now = datetime.now(timezone.utc)
+    tag = uuid.uuid4().hex[:8]
+    uid, hid, rule_id = str(uuid.uuid4()), str(uuid.uuid4()), str(uuid.uuid4())
+    ping_id, web_id, run_id = str(uuid.uuid4()), str(uuid.uuid4()), str(uuid.uuid4())
+    host_name, username = f"mcp-host-{tag}", f"mcp-{tag}"
+
+    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))
+                s.add(Host(id=hid, name=host_name, address="10.9.9.9"))
+            async with s.begin():
+                s.add(Monitor(id=ping_id, name=f"ping-{tag}", type="icmp",
+                              target="10.9.9.9", host_id=hid, config_json="{}", enabled=True))
+                s.add(Monitor(id=web_id, name=f"web-{tag}", type="http",
+                              target="https://bob:hunter2@web.test/health",
+                              config_json=json.dumps({"headers": {"Authorization": "Bearer s3cret"}}),
+                              enabled=True))
+            async with s.begin():
+                s.add(MonitorResult(monitor_id=ping_id, checked_at=now - timedelta(minutes=10),
+                                    is_up=True, response_ms=1.0))
+                s.add(MonitorResult(monitor_id=ping_id, checked_at=now - timedelta(minutes=5),
+                                    is_up=False, error_msg="timeout"))
+                s.add(MonitorResult(monitor_id=ping_id, checked_at=now - timedelta(minutes=1),
+                                    is_up=False, error_msg="timeout"))
+                s.add(MonitorResult(monitor_id=web_id, checked_at=now - timedelta(minutes=1),
+                                    is_up=True, status_code=200, response_ms=40.0,
+                                    tls_expires_at=now + timedelta(days=5)))
+                for i, v in enumerate((10.0, 20.0, 95.0)):
+                    s.add(PluginMetric(source_module="host_agent", resource_name=host_name,
+                                       metric_name="cpu_pct", value=v,
+                                       recorded_at=now - timedelta(minutes=40 - i * 20, seconds=30)))
+                s.add(PluginMetric(source_module="host_agent", resource_name=host_name,
+                                   metric_name="mem_used_pct", value=50.0,
+                                   recorded_at=now - timedelta(seconds=30)))
+                s.add(PluginMetric(source_module="host_agent", resource_name=host_name + ":/",
+                                   metric_name="disk_used_pct", value=41.0,
+                                   recorded_at=now - timedelta(seconds=30)))
+                s.add(AlertRule(id=rule_id, name=f"cpu hot {tag}", source_module="host_agent",
+                                resource_name=host_name, metric_name="cpu_pct",
+                                operator=AlertOperator.gt, threshold=90.0,
+                                consecutive_failures_required=1, enabled=True, created_by=uid))
+            async with s.begin():
+                s.add(AlertState(rule_id=rule_id, state=AlertStateEnum.firing,
+                                 last_transition_at=now - timedelta(minutes=2)))
+                s.add(AlertEvent(rule_id=rule_id, from_state="inactive", to_state="firing",
+                                 metric_value=95.0, transitioned_at=now - timedelta(minutes=2)))
+                s.add(MaintenanceWindow(name=f"patching {tag}", start_at=now - timedelta(hours=1),
+                                        end_at=now + timedelta(hours=1), scope_source="host_agent",
+                                        scope_resource=host_name, created_by=uid))
+                s.add(AnsibleRun(id=run_id, playbook_path=f"site-{tag}.yml", source_name="t",
+                                 status=AnsibleRunStatus.failed, started_at=now - timedelta(hours=1),
+                                 finished_at=now - timedelta(minutes=55),
+                                 results={"failures": ["web01: apt lock held"]}))
+        await log_audit(app, uid, username, "host.created", "host", hid, {"name": host_name})
+
+    async def _go():
+        await _seed()
+        _context._app_ctx.set(app)
+        _context._caller_ctx.set(Caller(user_id=uid, username=username,
+                                        role=UserRole.viewer, token_id=0))
+
+        # ── get_overview ───────────────────────────────────────────────────
+        ov = await overview.get_overview()
+        assert ov["all_clear"] is False
+        mine = {(f["kind"], f["subject"]): f for f in ov["findings"]
+                if tag in f["subject"] or f["subject"] == host_name}
+        down = mine[("monitor_down", f"ping-{tag}")]
+        assert down["severity"] == "critical" and "timeout" in down["summary"]
+        assert down["ids"]["monitor_id"] == ping_id
+        # down_since is the first failure after the last success, not the latest check.
+        assert 250 <= down["age_s"] <= 400
+        assert mine[("tls_expiring", f"web-{tag}")]["severity"] == "warning"
+        assert mine[("alert_firing", f"cpu hot {tag}")]["ids"]["rule_id"] == rule_id
+        assert mine[("host_cpu_high", host_name)]["severity"] == "critical"
+        assert "apt lock held" in mine[("ansible_run_failed", f"site-{tag}.yml")]["summary"]
+        assert any(w["name"] == f"patching {tag}" for w in ov["active_maintenance"])
+        sev = [f["severity"] for f in ov["findings"]]
+        assert sev == sorted(sev, key=lambda s: {"critical": 0, "warning": 1}[s])
+
+        # ── hosts ──────────────────────────────────────────────────────────
+        listed = {h["host_name"]: h for h in (await hosts.list_hosts(limit=500))["hosts"]}
+        h = listed[host_name]
+        assert h["status"] == "down" and h["agent"]["reporting"] is True
+        assert h["agent"]["cpu_pct"] == 95.0
+
+        detail = await hosts.get_host(host_name.upper())  # names match case-insensitively
+        assert detail["host_id"] == hid
+        assert [m["state"] for m in detail["monitors"]] == ["down"]
+        assert detail["monitors"][0]["error"] == "timeout"
+        assert detail["agent"]["parts"]["/"]["disk_used_pct"] == 41.0
+        assert [a["rule_id"] for a in detail["active_alerts"]] == [rule_id]
+        assert detail["ansible_target"] is None
+
+        series = await hosts.get_host_metrics(hid, "cpu_pct", "1h")
+        assert series["max"] == 95.0 and series["points"] >= 1 and series["series"]
+        listing = await hosts.get_host_metrics(host_name)
+        assert {m["metric"] for m in listing["metrics"]} == {"cpu_pct", "mem_used_pct"}
+        with pytest.raises(ValueError, match="no metric 'nope'"):
+            await hosts.get_host_metrics(host_name, "nope")
+        with pytest.raises(ValueError, match="list_hosts"):
+            await hosts.get_host(f"no-such-{tag}")
+
+        # ── monitors ───────────────────────────────────────────────────────
+        down_list = await monitors.list_monitors(state="down", limit=500)
+        ping = next(m for m in down_list["monitors"] if m["monitor_id"] == ping_id)
+        assert ping["host_name"] == host_name and ping["down_since"]
+        web = await monitors.get_monitor(f"web-{tag}")
+        assert "hunter2" not in web["target"] and "bob:***@" in web["target"]
+        assert web["config"]["headers"] == {"Authorization": "<redacted>"}
+        assert 4 <= web["tls_expires_in_days"] <= 5.1
+        ping_detail = await monitors.get_monitor(ping_id, failures=5)
+        assert ping_detail["heartbeat"] == "+--"
+        assert [f["error"] for f in ping_detail["recent_failures"]] == ["timeout", "timeout"]
+        with pytest.raises(ValueError, match="is not one of"):
+            await monitors.list_monitors(state="broken")
+
+        # ── alerts ─────────────────────────────────────────────────────────
+        firing = [a for a in (await alerts.list_alerts())["alerts"] if a["rule_id"] == rule_id]
+        assert firing[0]["state"] == "firing" and firing[0]["condition"] == "cpu_pct > 90"
+        hist = await alerts.get_alert_history(rule=f"cpu hot {tag}", range="24h")
+        assert [(e["from_state"], e["to_state"]) for e in hist["events"]] == [("inactive", "firing")]
+        rules = {r["rule_id"]: r for r in (await alerts.list_alert_rules(limit=500))["rules"]}
+        assert rules[rule_id]["state"] == "firing"
+        windows = (await alerts.list_maintenance_windows(limit=500))["windows"]
+        assert any(w["name"] == f"patching {tag}" and w["active"] for w in windows)
+
+        # ── query_metrics, one argument at a time ──────────────────────────
+        assert "host_agent" in (await metrics.query_metrics())["source_modules"]
+        res = (await metrics.query_metrics("host_agent", limit=1000))["resources"]
+        assert host_name in res and host_name + ":/" in res
+        latest = await metrics.query_metrics("host_agent", host_name)
+        assert {m["metric"] for m in latest["metrics"]} == {"cpu_pct", "mem_used_pct"}
+        cpu = await metrics.query_metrics("host_agent", host_name, "cpu_pct", "1h")
+        assert cpu["last"] == 95.0
+        with pytest.raises(ValueError, match="no metrics recorded"):
+            await metrics.query_metrics(f"nothing-{tag}")
+        with pytest.raises(ValueError, match="is not one of"):
+            await metrics.query_metrics("host_agent", host_name, "cpu_pct", "2w")
+
+        # ── audit ──────────────────────────────────────────────────────────
+        events = (await audit.list_audit_events(action="host.", user=username))["events"]
+        assert [(e["action"], e["entity_id"]) for e in events] == [("host.created", hid)]
+        assert events[0]["detail"] == {"name": host_name}
+
+        # ── through the SDK: arguments validate and results serialise ──────
+        await app.mcp_server.call_tool("list_hosts", {"limit": 5})
+        await app.mcp_server.call_tool("get_overview", {})
+
+    asyncio.run(_go())
diff --git a/tests/mcp/test_tools.py b/tests/mcp/test_tools.py
new file mode 100644
index 0000000..51efb8f
--- /dev/null
+++ b/tests/mcp/test_tools.py
@@ -0,0 +1,236 @@
+"""The diagnostic tools' pure parts: response shaping, lookup, redaction, ranking.
+
+The queries behind each tool run against real Postgres in
+tests/integration/test_mcp_tools.py; these cover the logic that decides what
+an agent is told.
+"""
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace as NS
+from unittest.mock import MagicMock
+
+import pytest
+
+from steward.mcp.tools import _shape, overview
+from steward.mcp.tools.monitors import _config
+from steward.models.alerts import AlertOperator
+from steward.models.ansible import AnsibleRunStatus
+from steward.models.users import UserRole
+
+NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)
+
+MCP4_TOOLS = {
+    "get_overview", "list_hosts", "get_host", "get_host_metrics", "query_metrics",
+    "list_monitors", "get_monitor", "list_alerts", "get_alert_history",
+    "list_alert_rules", "list_maintenance_windows", "list_audit_events",
+}
+
+
+# ── shaping ──────────────────────────────────────────────────────────────────
+
+def test_iso_is_utc_with_z_and_naive_times_are_read_as_utc():
+    assert _shape.iso(NOW) == "2026-10-06T12:00:00Z"
+    assert _shape.iso(datetime(2026, 10, 6, 12, 0)) == "2026-10-06T12:00:00Z"
+    assert _shape.iso(None) is None
+
+
+def test_age_s_never_negative():
+    assert _shape.age_s(NOW - timedelta(seconds=90), NOW) == 90
+    assert _shape.age_s(NOW + timedelta(seconds=5), NOW) == 0
+
+
+def test_clamp_limit():
+    assert _shape.clamp_limit(0, 50, 500) == 50
+    assert _shape.clamp_limit(-3, 50, 500) == 50
+    assert _shape.clamp_limit(10, 50, 500) == 10
+    assert _shape.clamp_limit(9999, 50, 500) == 500
+
+
+def test_bounded_says_when_and_how_it_cut():
+    whole = _shape.bounded([1, 2], 5, key="xs")
+    assert whole == {"xs": [1, 2], "total": 2, "truncated": False}
+    cut = _shape.bounded(list(range(10)), 3, key="xs", more="Filter by `state`.")
+    assert cut["xs"] == [0, 1, 2] and cut["total"] == 10 and cut["truncated"] is True
+    assert cut["more"] == "Showing 3 of 10. Filter by `state`."
+
+
+def test_lookback_refuses_an_unknown_range_instead_of_guessing():
+    since, key = _shape.lookback("7d")
+    assert key == "7d"
+    with pytest.raises(ValueError, match="'2w' is not one of"):
+        _shape.lookback("2w")
+
+
+def test_summarize_and_downsample():
+    pts = [(NOW + timedelta(minutes=i), float(i)) for i in range(1, 101)]
+    s = _shape.summarize(pts, n=10)
+    assert (s["points"], s["min"], s["max"], s["last"]) == (100, 1.0, 100.0, 100.0)
+    assert s["avg"] == 50.5 and s["p95"] == 95.0
+    assert len(s["series"]) == 10
+    assert s["series"][0] == ["2026-10-06T12:01:00Z", 5.5]
+    assert _shape.summarize([]) == {"points": 0, "series": []}
+
+
+def _rows(*names):
+    return [NS(id=f"id-{i}", name=n) for i, n in enumerate(names)]
+
+
+def test_pick_by_id_then_case_insensitive_name():
+    rows = _rows("nas", "web01")
+    assert _shape.pick("id-1", rows, kind="host", list_tool="list_hosts").name == "web01"
+    assert _shape.pick("NAS", rows, kind="host", list_tool="list_hosts").id == "id-0"
+
+
+def test_pick_explains_a_miss_and_an_ambiguity():
+    with pytest.raises(ValueError, match=r"Did you mean 'web01'\?.*list_hosts"):
+        _shape.pick("web1", _rows("nas", "web01"), kind="host", list_tool="list_hosts")
+    with pytest.raises(ValueError, match="2 hosts are named 'nas'.*id-0, id-1"):
+        _shape.pick("nas", _rows("nas", "NAS"), kind="host", list_tool="list_hosts")
+    with pytest.raises(ValueError, match="pass a host name or id"):
+        _shape.pick("", _rows("nas"), kind="host", list_tool="list_hosts")
+
+
+def test_redact_url_hides_only_the_password():
+    assert _shape.redact_url("https://bob:pw@h.test:8443/x?q=1") == "https://bob:***@h.test:8443/x?q=1"
+    assert _shape.redact_url("https://h.test/x") == "https://h.test/x"
+    assert _shape.redact_url("10.0.0.1") == "10.0.0.1"
+
+
+def test_monitor_config_redacts_header_values():
+    m = NS(config_json=json.dumps({"method": "GET", "headers": {"Authorization": "Bearer x"}}))
+    assert _config(m) == {"method": "GET", "headers": {"Authorization": "<redacted>"}}
+    assert _config(NS(config_json="not json")) == {}
+
+
+# ── overview findings ────────────────────────────────────────────────────────
+
+def _monitor(mid, *, enabled=True, host_id=None):
+    return NS(id=mid, name=f"mon-{mid}", type="http", target="https://x.test",
+              enabled=enabled, host_id=host_id)
+
+
+def _result(*, up, error=None, tls_days=None):
+    return NS(is_up=up, error_msg=error, checked_at=NOW - timedelta(minutes=1),
+              tls_expires_at=NOW + timedelta(days=tls_days) if tls_days is not None else None)
+
+
+def test_monitor_findings():
+    mons = [_monitor("a", host_id="h1"), _monitor("b"), _monitor("c"), _monitor("d", enabled=False),
+            _monitor("e")]
+    states = {
+        "a": {"latest": _result(up=False, error="503"), "down_since": NOW - timedelta(hours=2)},
+        "b": {"latest": _result(up=True, tls_days=2)},
+        "c": {"latest": _result(up=True, tls_days=10)},
+        "d": {"latest": _result(up=False)},
+        "e": {"latest": _result(up=True, tls_days=60)},
+    }
+    out = {f["subject"]: f for f in overview.monitor_findings(mons, states, {"h1": "nas"}, NOW)}
+    assert set(out) == {"mon-a", "mon-b", "mon-c"}  # disabled and healthy left out
+    down = out["mon-a"]
+    assert down["severity"] == "critical" and down["kind"] == "monitor_down"
+    assert "on nas" in down["summary"] and "503" in down["summary"]
+    assert down["since"] == "2026-10-06T10:00:00Z"
+    assert down["next"] == 'get_monitor(monitor="a")'
+    assert (out["mon-b"]["severity"], out["mon-c"]["severity"]) == ("critical", "warning")
+
+
+def test_alert_findings_severity_by_state():
+    rule = NS(id="r1", name="cpu", metric_name="cpu_pct", operator=AlertOperator.gt,
+              threshold=90.0, source_module="host_agent", resource_name="nas")
+    rows = [(NS(state=NS(value=s), last_transition_at=NOW), rule)
+            for s in ("firing", "pending", "acknowledged")]
+    sev = {f["kind"]: f["severity"] for f in overview.alert_findings(rows)}
+    assert sev == {"alert_firing": "critical", "alert_pending": "warning",
+                   "alert_acknowledged": "warning"}
+
+
+THRESHOLDS = {"cpu": {"warn": 80, "crit": 90}, "mem": {"warn": 80, "crit": 90},
+              "disk": {"warn": 80, "crit": 90}}
+
+
+def test_agent_findings_use_the_configured_thresholds():
+    hosts = [NS(id="h1", name="nas"), NS(id="h2", name="web"), NS(id="h3", name="db")]
+    agent = {
+        "nas": {"fresh": True, "cpu_pct": 95.0, "mem_used_pct": 85.0, "disk_used_pct_worst": 10.0},
+        "web": {"fresh": False, "cpu_pct": 99.0, "last_at": NOW - timedelta(minutes=30)},
+    }
+    out = overview.agent_findings(hosts, agent, THRESHOLDS)
+    got = {(f["subject"], f["kind"]): f["severity"] for f in out}
+    assert got == {
+        ("nas", "host_cpu_high"): "critical",
+        ("nas", "host_mem_high"): "warning",
+        # A stale agent's old numbers are not reported as current load.
+        ("web", "agent_stale"): "warning",
+    }
+
+
+def test_run_findings_quote_the_failures():
+    run = NS(id="r", playbook_path="site.yml", source_name="infra",
+             status=AnsibleRunStatus.failed, started_at=NOW, finished_at=NOW,
+             results={"failures": ["web01: apt lock held"]})
+    f = overview.run_findings([run])[0]
+    assert f["kind"] == "ansible_run_failed" and "apt lock held" in f["summary"]
+
+
+def test_rank_puts_critical_first_then_oldest():
+    fs = [overview.finding("warning", "k", "w-old", "", since=NOW - timedelta(days=1)),
+          overview.finding("critical", "k", "c-new", "", since=NOW - timedelta(minutes=1)),
+          overview.finding("critical", "k", "c-old", "", since=NOW - timedelta(hours=1))]
+    assert [f["subject"] for f in overview.rank(fs)] == ["c-old", "c-new", "w-old"]
+
+
+async def test_plugin_overview_sources_skip_disabled_plugins_and_survive_failures(monkeypatch):
+    calls = []
+
+    async def good(db):
+        calls.append("good")
+        return [overview.finding("warning", "container_unhealthy", "redis", "unhealthy")]
+
+    async def broken(db):
+        raise RuntimeError("boom")
+
+    async def off(db):
+        calls.append("off")
+        return []
+
+    class _CM:
+        async def __aenter__(self):
+            return object()
+
+        async def __aexit__(self, *a):
+            return False
+
+    monkeypatch.setattr(overview, "db_session", lambda: _CM())
+    server = MagicMock()
+    server.overview_sources = {"docker": good, "snmp": broken, "unifi": off}
+    server.plugin_enabled = lambda p: p != "unifi"
+    found, unavailable = await overview._plugin_findings(server)
+    assert [f["subject"] for f in found] == ["redis"]
+    assert unavailable == ["snmp"] and calls == ["good"]
+
+
+def test_register_plugin_picks_up_an_overview_source():
+    from steward.mcp.server import StewardMCPServer
+    app = MagicMock()
+    app.config = {"PLUGINS": {}}
+    mcp = StewardMCPServer(app)
+
+    async def mcp_overview(db):
+        return []
+    mcp.register_plugin("docker", NS(mcp_overview=mcp_overview))
+    assert mcp.overview_sources == {"docker": mcp_overview}
+    mcp.register_plugin("docker", NS())  # reloaded without it
+    assert mcp.overview_sources == {}
+
+
+# ── registration ─────────────────────────────────────────────────────────────
+
+def test_diagnostic_tools_are_registered_read_only_for_viewers(app):
+    mcp = app.mcp_server
+    assert MCP4_TOOLS <= set(mcp.tool_roles)
+    for name in MCP4_TOOLS:
+        assert mcp.tool_roles[name] is UserRole.viewer, name
+        assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is True, name
+        assert (mcp._tool_manager.get_tool(name).description or "").strip(), f"{name} has no docstring"