diff --git a/README.md b/README.md
index 17271cc..87760fe 100644
--- a/README.md
+++ b/README.md
@@ -126,6 +126,7 @@ See `docs/plugins/` for a full plugin development guide.
 | `docs/plugins/writing-a-plugin.md` | Step-by-step plugin development guide |
 | `docs/plugins/traefik.md` | Traefik plugin: config, metrics, alert rule examples |
 | `docs/plugins/prometheus.md` | Prometheus plugin: scraping any /metrics endpoint, how series are stored, the Fabled Archives recipe |
+| `docs/plugins/homeassistant.md` | Home Assistant plugin: waiting updates, backup age, and installing updates after a confirmed backup |
 | `docs/plugins/syslog.md` | Syslog plugin: sources, patterns and alerts, publishing port 5514 (ingress or host mode under Swarm) |
 | `docs/plugins/prometheus-widgets.md` | Prometheus widget templates: writing, sharing and copying one in, every block, the format reference |
 | `docs/reference/code-map.md` | Where every key function, model, and route lives in the codebase |
diff --git a/docs/plugins/homeassistant.md b/docs/plugins/homeassistant.md
new file mode 100644
index 0000000..3611c29
--- /dev/null
+++ b/docs/plugins/homeassistant.md
@@ -0,0 +1,27 @@
+# Home Assistant plugin
+
+Shows which Home Assistant updates are waiting and how old its last backup is, and installs Core, OS and add-on updates, each only after a fresh backup is confirmed. Each update goes on the timeline.
+
+Steward never holds a credential that can act on your house. It reads with a read-only token, and acts only through a webhook whose automation can take a backup or install an allowed update.
+
+## Setting it up
+
+1. In Home Assistant, turn on automatic backups (Settings → System → Backups).
+2. Make a local-only user in the read-only group (`system-read-only`) and a long-lived token for it.
+3. On Steward's Home Assistant page, choose Connect, enter Home Assistant's local address and the token, and save.
+4. Copy the automation the connection page shows into Home Assistant (Settings → Automations → Create → Edit in YAML). Steward reads whether it is there and on.
+
+Steward must reach Home Assistant by a local address, because the webhook only answers local requests. Home Assistant bans an address after a few refused requests, so Steward stops asking after one refusal until you save the connection or choose Try again.
+
+## Installing an update
+
+Install asks once, then Steward:
+
+1. Asks for a backup and waits until Home Assistant's last successful backup time moves. If it doesn't, nothing is installed.
+2. Asks for the update and follows it until the new version is installed, or until a deadline: 20 minutes for an add-on, 30 for Core, 45 for the OS.
+
+One runs at a time, and it carries on if Steward restarts. Firmware and custom integrations are left to Home Assistant, and Supervisor updates itself.
+
+## Metrics
+
+Under source `homeassistant`, resource `homeassistant`, read every 5 minutes: `is_up`, `updates_waiting` and `backup_age_hours`. An alert rule on `backup_age_hours` above 26 warns when a daily backup has been missed; a rule's no-data warning covers Home Assistant going quiet.
diff --git a/plugins/homeassistant/__init__.py b/plugins/homeassistant/__init__.py
new file mode 100644
index 0000000..50dbea9
--- /dev/null
+++ b/plugins/homeassistant/__init__.py
@@ -0,0 +1,45 @@
+# plugins/homeassistant/__init__.py
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from quart import Quart
+
+_app: "Quart | None" = None
+
+
+def setup(app: "Quart") -> None:
+    global _app
+    _app = app
+    from .models import HomeAssistantConnection, HomeAssistantUpdate  # noqa: F401 — registers the tables
+    from .timeline import register
+    register()
+
+
+def get_scheduled_tasks() -> list:
+    from .scheduler import make_tick_task
+    return [make_tick_task(_app)]
+
+
+def get_blueprint():
+    from .routes import homeassistant_bp
+    return homeassistant_bp
+
+
+def get_nav() -> list[dict]:
+    return [{"label": "Home Assistant", "href": "/plugins/homeassistant/"}]
+
+
+def get_alert_metrics() -> dict[str, list[str]]:
+    """What the poll records, all under the resource "homeassistant"."""
+    return {"homeassistant": ["is_up", "backup_age_hours", "updates_waiting"]}
+
+
+def register_mcp_tools(register) -> None:
+    from .mcp_tools import register_tools
+    register_tools(register)
+
+
+async def mcp_overview(db) -> list[dict]:
+    from .mcp_tools import mcp_overview as findings
+    return await findings(db)
diff --git a/plugins/homeassistant/client.py b/plugins/homeassistant/client.py
new file mode 100644
index 0000000..689bf64
--- /dev/null
+++ b/plugins/homeassistant/client.py
@@ -0,0 +1,76 @@
+# plugins/homeassistant/client.py
+"""Talking to Home Assistant. Every request but the webhook carries the token,
+because Home Assistant bans an address after a few refused requests, token-less
+ones included. Every failure comes back as one sentence a page can show."""
+from __future__ import annotations
+
+import json
+
+import httpx
+
+TIMEOUT_SECONDS = 15.0
+MAX_BYTES = 16 * 1024 * 1024
+
+
+class Unreachable(Exception):
+    """Home Assistant didn't answer usefully; it may be restarting."""
+
+
+class Refused(Exception):
+    """Home Assistant refused the token, or has banned this address."""
+
+
+async def _get(url: str, path: str, token: str, verify_tls: bool, transport) -> list | dict | None:
+    """The parsed JSON of a GET, or None for a 404."""
+    headers = {"Authorization": f"Bearer {token}"}
+    try:
+        async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS, verify=verify_tls, transport=transport,
+                                     follow_redirects=False) as client:
+            async with client.stream("GET", url.rstrip("/") + path, headers=headers) as response:
+                if response.status_code == 401:
+                    raise Refused("Home Assistant refused the token. Check it, or make a new one.")
+                if response.status_code == 403:
+                    raise Refused("Home Assistant answered 403: it may have banned Steward's address "
+                                  "after refused requests. Look in its ip_bans.yaml.")
+                if response.status_code == 404:
+                    return None
+                if response.status_code != 200:
+                    raise Unreachable(f"Home Assistant answered {response.status_code}.")
+                body = bytearray()
+                async for chunk in response.aiter_bytes():
+                    body.extend(chunk)
+                    if len(body) > MAX_BYTES:
+                        raise Unreachable("Home Assistant's answer was too large to read.")
+                return json.loads(body)
+    except httpx.TimeoutException:
+        raise Unreachable(f"Home Assistant didn't answer within {TIMEOUT_SECONDS:.0f} s.") from None
+    except httpx.HTTPError as exc:
+        raise Unreachable(f"Couldn't reach Home Assistant: {exc}") from None
+    except ValueError:
+        raise Unreachable("Home Assistant's answer wasn't JSON.") from None
+
+
+async def read_states(url: str, token: str, verify_tls: bool, transport=None) -> list[dict]:
+    states = await _get(url, "/api/states", token, verify_tls, transport)
+    if not isinstance(states, list):
+        raise Unreachable("Home Assistant's answer wasn't a list of entities.")
+    return states
+
+
+async def read_state(url: str, entity_id: str, token: str, verify_tls: bool, transport=None) -> dict | None:
+    return await _get(url, f"/api/states/{entity_id}", token, verify_tls, transport)
+
+
+async def send_webhook(url: str, webhook_id: str, body: dict, verify_tls: bool, transport=None) -> None:
+    """Hand the maintenance automation a request. Home Assistant answers 200
+    whether or not the id is known, so what the request did is read afterwards."""
+    try:
+        async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS, verify=verify_tls, transport=transport,
+                                     follow_redirects=False) as client:
+            response = await client.post(f"{url.rstrip('/')}/api/webhook/{webhook_id}", json=body)
+    except httpx.TimeoutException:
+        raise Unreachable(f"Home Assistant didn't answer within {TIMEOUT_SECONDS:.0f} s.") from None
+    except httpx.HTTPError as exc:
+        raise Unreachable(f"Couldn't reach Home Assistant: {exc}") from None
+    if response.status_code != 200:
+        raise Unreachable(f"Home Assistant answered {response.status_code} to the webhook.")
diff --git a/plugins/homeassistant/mcp_tools.py b/plugins/homeassistant/mcp_tools.py
new file mode 100644
index 0000000..544fb3b
--- /dev/null
+++ b/plugins/homeassistant/mcp_tools.py
@@ -0,0 +1,111 @@
+"""The Home Assistant plugin's MCP tools and its get_overview findings.
+
+get_home_assistant reads what the poll stored. update_home_assistant starts an
+update as the page's Install does. The token and webhook id are never returned.
+"""
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from steward.mcp._context import current_caller, mcp_app
+from steward.mcp.tools._audit import audit_mcp, mcp_username
+from steward.mcp.tools._shape import age_s, db_session, iso, now_utc, redact_url
+from steward.mcp.tools.overview import finding
+from steward.models.users import UserRole
+
+from .models import HomeAssistantConnection, HomeAssistantUpdate
+from .routes import connection_says
+from .service import STALE_BACKUP_HOURS, backup_age_hours, backup_verdict, installable, parse_moment
+from .updates import CannotStart, start_update
+
+RECENT_UPDATES = 10
+
+
+def _run_view(r: HomeAssistantUpdate) -> dict:
+    return {"update_id": r.id, "entity_id": r.entity_id, "title": r.title, "kind": r.kind,
+            "from_version": r.from_version, "to_version": r.to_version, "state": r.state, "message": r.message,
+            "started_by": r.started_by, "started_at": iso(r.started_at), "backup_at": iso(r.backup_at),
+            "finished_at": iso(r.finished_at)}
+
+
+async def get_home_assistant() -> dict:
+    """Home Assistant as of the last poll: whether it answers, which updates are
+    waiting, how old its last backup is, and the updates Steward was asked to install.
+
+    `installable` is what update_home_assistant takes (Core, OS, add-ons);
+    `others_waiting` is firmware and custom integrations, which Steward leaves
+    to Home Assistant. `automation` is the Steward maintenance automation's
+    state: "on" is needed to install, null means it wasn't found.
+    """
+    now = now_utc()
+    async with db_session() as db:
+        c = (await db.execute(select(HomeAssistantConnection))).scalars().first()
+        runs = list((await db.execute(select(HomeAssistantUpdate).order_by(
+            HomeAssistantUpdate.started_at.desc()).limit(RECENT_UPDATES))).scalars())
+    if c is None:
+        return {"connected": False, "says": "No connection yet: an admin adds it on the Home Assistant page."}
+    reach, says = connection_says(c, now)
+    backup = c.backup or {}
+    age = backup_age_hours(backup, now)
+    return {"connected": True, "url": redact_url(c.url), "reach": reach, "says": says,
+            "polled_at": iso(c.last_polled_at), "ok_age_s": age_s(c.last_ok_at, now), "automation": c.automation,
+            "installable": [{k: u[k] for k in ("entity_id", "title", "kind", "installed", "latest", "in_progress")}
+                            for u in installable(c)],
+            "others_waiting": [u["title"] for u in c.updates or [] if u["waiting"] and not u["kind"]],
+            "backup": {"verdict": backup_verdict(backup, now)[1], "last_ok": iso(parse_moment(backup.get("last_ok"))),
+                       "age_hours": round(age, 1) if age is not None else None, "stale_after_hours": STALE_BACKUP_HOURS,
+                       "next": iso(parse_moment(backup.get("next"))), "last_result": backup.get("last_result")},
+            "updates": [_run_view(r) for r in runs]}
+
+
+async def update_home_assistant(entity_id: str) -> dict:
+    """Install one Core, OS or add-on update, as the page's Install does: Steward
+    takes a backup, confirms it landed, installs, and follows it to its end.
+
+    entity_id: from get_home_assistant's `installable`.
+
+    An update restarts Home Assistant or the add-on, and an OS update reboots the
+    machine. Refused while another is under way, or if nothing is waiting there.
+    Admin only, as on the page. Recorded in the audit log.
+    """
+    caller = current_caller()
+    try:
+        run = await start_update(mcp_app(), entity_id, mcp_username(caller.username))
+    except CannotStart as exc:
+        raise ValueError(f"Nothing was started: {exc}.") from None
+    await audit_mcp("homeassistant.update_started", "homeassistant_update", run.id,
+                    {"entity_id": entity_id, "title": run.title, "to": run.to_version})
+    return {**_run_view(run), "next": "get_home_assistant follows it: `updates` shows its state."}
+
+
+def overview_findings(c, latest_run, now) -> list[dict]:
+    """A connection that isn't answering, stale backups, and a newest update that
+    failed. Updates waiting aren't a problem to raise."""
+    if c is None:
+        return []
+    found = []
+    reach, says = connection_says(c, now)
+    if reach == "down":
+        found.append(finding("warning", "home_assistant_down", "Home Assistant",
+                             f"Home Assistant isn't answering: {says}", since=c.last_ok_at, next="get_home_assistant()"))
+    tone, backup_says = backup_verdict(c.backup, now)
+    if c.last_ok_at is not None and tone == "warn":
+        found.append(finding("warning", "home_assistant_backup", "Home Assistant", backup_says,
+                             since=parse_moment((c.backup or {}).get("last_ok")), next="get_home_assistant()"))
+    if latest_run is not None and latest_run.state == "failed":
+        found.append(finding("warning", "home_assistant_update_failed", latest_run.title,
+                             f"The update of {latest_run.title} failed: {latest_run.message}",
+                             since=latest_run.finished_at, next="get_home_assistant()"))
+    return found
+
+
+async def mcp_overview(db) -> list[dict]:
+    c = (await db.execute(select(HomeAssistantConnection))).scalars().first()
+    latest = (await db.execute(select(HomeAssistantUpdate).order_by(
+        HomeAssistantUpdate.started_at.desc()).limit(1))).scalars().first()
+    return overview_findings(c, latest, now_utc())
+
+
+def register_tools(register) -> None:
+    register(get_home_assistant, role=UserRole.viewer, read_only=True)
+    register(update_home_assistant, role=UserRole.admin)
diff --git a/plugins/homeassistant/migrations/__init__.py b/plugins/homeassistant/migrations/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/plugins/homeassistant/migrations/versions/__init__.py b/plugins/homeassistant/migrations/versions/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/plugins/homeassistant/migrations/versions/homeassistant_001_connection.py b/plugins/homeassistant/migrations/versions/homeassistant_001_connection.py
new file mode 100644
index 0000000..71b3b94
--- /dev/null
+++ b/plugins/homeassistant/migrations/versions/homeassistant_001_connection.py
@@ -0,0 +1,59 @@
+"""Home Assistant: the connection and the updates Steward was asked to install
+
+Revision ID: homeassistant_001_connection
+Revises: (none — a branch off core via depends_on)
+Create Date: 2026-10-31
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "homeassistant_001_connection"
+down_revision: Union[str, None] = None
+branch_labels: Union[str, Sequence[str], None] = "homeassistant"
+depends_on: Union[str, Sequence[str], None] = ("0004_app_settings",)
+
+
+def upgrade() -> None:
+    op.create_table(
+        "homeassistant_connection",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("url", sa.String(500), nullable=False),
+        sa.Column("token", sa.Text, nullable=False),
+        sa.Column("webhook_id", sa.Text, nullable=False),
+        sa.Column("verify_tls", sa.Boolean, nullable=False, server_default=sa.true()),
+        sa.Column("refused", sa.Boolean, nullable=False, server_default=sa.false()),
+        sa.Column("last_polled_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_ok_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_error", sa.Text, nullable=True),
+        sa.Column("updates", sa.JSON, nullable=False, server_default="[]"),
+        sa.Column("backup", sa.JSON, nullable=False, server_default="{}"),
+        sa.Column("automation", sa.String(20), nullable=True),
+    )
+    op.create_table(
+        "homeassistant_updates",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("entity_id", sa.String(255), nullable=False),
+        sa.Column("title", sa.String(255), nullable=False),
+        sa.Column("kind", sa.String(10), nullable=False),
+        sa.Column("from_version", sa.String(100), nullable=True),
+        sa.Column("to_version", sa.String(100), nullable=True),
+        sa.Column("state", sa.String(20), nullable=False, server_default="backing_up"),
+        sa.Column("message", sa.Text, nullable=True),
+        sa.Column("started_by", sa.String(100), nullable=False, server_default=""),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("backup_before", sa.String(64), nullable=True),
+        sa.Column("backup_event_before", sa.String(64), nullable=True),
+        sa.Column("triggered_before", sa.String(64), nullable=True),
+        sa.Column("backup_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("install_sent_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.execute("CREATE UNIQUE INDEX uq_homeassistant_update_under_way ON homeassistant_updates ((true)) "
+               "WHERE state IN ('backing_up', 'installing')")
+
+
+def downgrade() -> None:
+    op.drop_table("homeassistant_updates")
+    op.drop_table("homeassistant_connection")
diff --git a/plugins/homeassistant/models.py b/plugins/homeassistant/models.py
new file mode 100644
index 0000000..e3634a6
--- /dev/null
+++ b/plugins/homeassistant/models.py
@@ -0,0 +1,68 @@
+# plugins/homeassistant/models.py
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from sqlalchemy import JSON, Boolean, DateTime, Index, String, Text, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from steward.core.ids import time_ordered_id
+from steward.models.base import Base
+
+
+class HomeAssistantConnection(Base):
+    """The Home Assistant instance Steward looks after; at most one row.
+
+    The token is a read-only user's and the webhook id can only reach the
+    maintenance automation, so nothing stored here can act on the house. Both
+    are encrypted. The last_* columns and `updates` and `backup` are the newest
+    poll's outcome, which the pages read. `refused` stops all polling until
+    someone saves the connection or tries again: Home Assistant bans an address
+    after a few refused requests.
+    """
+    __tablename__ = "homeassistant_connection"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id)
+    url: Mapped[str] = mapped_column(String(500), nullable=False)
+    token: Mapped[str] = mapped_column(Text, nullable=False)
+    webhook_id: Mapped[str] = mapped_column(Text, nullable=False)
+    verify_tls: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    refused: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    last_polled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    last_ok_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    updates: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
+    backup: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
+    automation: Mapped[str | None] = mapped_column(String(20), nullable=True)
+
+
+class HomeAssistantUpdate(Base):
+    """One update Steward was asked to install, from the click to its end. It
+    moves through backing_up and installing to done or failed, and its row
+    carries everything needed to carry on after Steward restarts. The *_before
+    columns are Home Assistant's own readings at the start, so "a newer backup"
+    is judged on its clock alone. One can be under way at a time.
+    """
+    __tablename__ = "homeassistant_updates"
+    __table_args__ = (
+        Index("uq_homeassistant_update_under_way", text("(true)"), unique=True,
+              postgresql_where=text("state IN ('backing_up', 'installing')")),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id)
+    entity_id: Mapped[str] = mapped_column(String(255), nullable=False)
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    kind: Mapped[str] = mapped_column(String(10), nullable=False)
+    from_version: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    to_version: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    state: Mapped[str] = mapped_column(String(20), nullable=False, default="backing_up")
+    message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    started_by: Mapped[str] = mapped_column(String(100), nullable=False, default="")
+    started_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
+    backup_before: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    backup_event_before: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    triggered_before: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    backup_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    install_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
diff --git a/plugins/homeassistant/plugin.yaml b/plugins/homeassistant/plugin.yaml
new file mode 100644
index 0000000..12efeea
--- /dev/null
+++ b/plugins/homeassistant/plugin.yaml
@@ -0,0 +1,17 @@
+# plugins/homeassistant/plugin.yaml
+name: homeassistant
+version: "1.0.0"
+description: "Shows which Home Assistant updates are waiting and how old its last backup is, and installs Core, OS and add-on updates after a confirmed backup."
+author: "Steward"
+license: "MIT"
+min_app_version: "0.1.0"
+repository_url: "https://git.fabledsword.com/bvandeusen/FabledSteward"
+homepage: "https://git.fabledsword.com/bvandeusen/FabledSteward/src/branch/dev/plugins/homeassistant"
+tags:
+  - home-assistant
+  - updates
+  - backups
+
+# The connection is added and edited on the Home Assistant page
+# (/plugins/homeassistant/).
+config: {}
diff --git a/plugins/homeassistant/routes.py b/plugins/homeassistant/routes.py
new file mode 100644
index 0000000..bc739e0
--- /dev/null
+++ b/plugins/homeassistant/routes.py
@@ -0,0 +1,206 @@
+"""The Home Assistant page. Every GET reads Steward's own records, which the poll
+fills. Only Test and Install, an admin's explicit actions, ask Home Assistant
+anything (System 178's ruling)."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from quart import Blueprint, current_app, redirect, render_template, request, session
+from sqlalchemy import select
+
+from steward.auth.middleware import require_role
+from steward.core.audit import log_audit
+from steward.core.crypto import decrypt_secret, encrypt_secret
+from steward.models.users import UserRole
+
+from . import client
+from .models import HomeAssistantConnection, HomeAssistantUpdate
+from .service import (STALE_BACKUP_HOURS, automation_yaml, backup_age_hours, backup_verdict, installable,
+                      new_webhook_id, no_spaces, parse_moment, read_snapshot)
+from .updates import CannotStart, start_update
+
+homeassistant_bp = Blueprint("homeassistant", __name__, template_folder="templates")
+
+RECENT_UPDATES = 10
+
+
+def _actor() -> tuple[str | None, str]:
+    return session.get("user_id"), session.get("username", "")
+
+
+async def _connection(db) -> HomeAssistantConnection | None:
+    return (await db.execute(select(HomeAssistantConnection))).scalars().first()
+
+
+def connection_says(connection, now: datetime) -> tuple[str, str]:
+    """(badge state, sentence) for whether Home Assistant is answering."""
+    if connection.refused:
+        return "down", f"{connection.last_error} Steward has stopped asking until you save the connection or try again."
+    if connection.last_polled_at is None:
+        return "pending", "Not read yet: the first read runs within seconds."
+    if connection.last_error:
+        return "down", connection.last_error
+    return "up", "Answering."
+
+
+async def _render_index(error: str | None = None, status: int = 200):
+    async with current_app.db_sessionmaker() as db:
+        connection = await _connection(db)
+        runs = list((await db.execute(select(HomeAssistantUpdate).order_by(
+            HomeAssistantUpdate.started_at.desc()).limit(RECENT_UPDATES))).scalars())
+    if connection is None:
+        return await render_template("homeassistant/index.html", connection=None), status
+    now = datetime.now(timezone.utc)
+    reach, says = connection_says(connection, now)
+    updates = connection.updates or []
+    backup = connection.backup or {}
+    tone, backup_says = backup_verdict(backup, now)
+    return await render_template(
+        "homeassistant/index.html", connection=connection, error=error, reach=reach, says=says,
+        installable=installable(connection),
+        others=[u for u in updates if u["waiting"] and not u["kind"]],
+        current=[u for u in updates if not u["waiting"] and u["kind"]],
+        runs=runs, under_way=next((r for r in runs if r.state in ("backing_up", "installing")), None),
+        backup=backup, backup_tone=tone, backup_says=backup_says,
+        backup_at=parse_moment(backup.get("last_ok")), backup_next=parse_moment(backup.get("next")),
+        backup_age=backup_age_hours(backup, now), stale_hours=STALE_BACKUP_HOURS), status
+
+
+@homeassistant_bp.get("/")
+@require_role(UserRole.viewer)
+async def index():
+    return await _render_index()
+
+
+@homeassistant_bp.post("/install")
+@require_role(UserRole.admin)
+async def install():
+    form = await request.form
+    entity_id = (form.get("entity_id") or "").strip()
+    user_id, username = _actor()
+    try:
+        run = await start_update(current_app, entity_id, username)
+    except CannotStart as exc:
+        return await _render_index(f"Nothing was started: {exc}.", 409)
+    await log_audit(current_app, user_id, username, "homeassistant.update_started",
+                    entity_type="homeassistant_update", entity_id=run.id,
+                    detail={"entity_id": entity_id, "title": run.title, "to": run.to_version})
+    return redirect("/plugins/homeassistant/")
+
+
+# ── The connection ───────────────────────────────────────────────────────────
+
+def _values(connection, form=None) -> dict:
+    if form is not None:
+        return {"url": form.get("url", ""), "verify_tls": bool(form.get("verify_tls")),
+                "has_token": bool(connection), "has_webhook": bool(connection)}
+    return {"url": connection.url if connection else "", "verify_tls": connection.verify_tls if connection else True,
+            "has_token": bool(connection), "has_webhook": bool(connection)}
+
+
+async def _render_form(connection, values: dict, error: str | None = None, status: int = 200):
+    yaml_text = automation_yaml(decrypt_secret(connection.webhook_id, context="Home Assistant webhook id")) \
+        if connection is not None else None
+    return await render_template("homeassistant/form.html", connection=connection, v=values, error=error,
+                                 automation=yaml_text), status
+
+
+@homeassistant_bp.get("/connection")
+@require_role(UserRole.admin)
+async def connection_form():
+    async with current_app.db_sessionmaker() as db:
+        connection = await _connection(db)
+    return await _render_form(connection, _values(connection))
+
+
+@homeassistant_bp.post("/connection")
+@require_role(UserRole.admin)
+async def save_connection():
+    form = await request.form
+    url = (form.get("url") or "").strip().rstrip("/")
+    token, webhook = no_spaces(form.get("token")), no_spaces(form.get("webhook_id"))
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            connection = await _connection(db)
+            error = None
+            if not url.startswith(("http://", "https://")):
+                error = "Enter Home Assistant's address: http://… or https://…"
+            elif not token and connection is None:
+                error = "Enter a token: a long-lived token of a read-only Home Assistant user."
+            if error:
+                return await _render_form(connection, _values(connection, form), error, 422)
+            if connection is None:
+                connection = HomeAssistantConnection(url=url, token=encrypt_secret(token),
+                                                     webhook_id=encrypt_secret(webhook or new_webhook_id()))
+                db.add(connection)
+            else:
+                connection.url = url
+                if token:
+                    connection.token = encrypt_secret(token)
+                if webhook:
+                    connection.webhook_id = encrypt_secret(webhook)
+            connection.verify_tls = bool(form.get("verify_tls"))
+            connection.refused, connection.last_error, connection.last_polled_at = False, None, None
+            await db.flush()
+            connection_id = connection.id
+    await log_audit(current_app, *_actor(), "homeassistant.connection_saved",
+                    entity_type="homeassistant_connection", entity_id=connection_id, detail={"url": url})
+    return redirect("/plugins/homeassistant/connection")
+
+
+@homeassistant_bp.post("/connection/retry")
+@require_role(UserRole.admin)
+async def retry_connection():
+    """Ask again after a refusal, once the token or a ban has been dealt with."""
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            connection = await _connection(db)
+            if connection is not None:
+                connection.refused, connection.last_polled_at = False, None
+    return redirect("/plugins/homeassistant/")
+
+
+@homeassistant_bp.post("/connection/delete")
+@require_role(UserRole.admin)
+async def delete_connection():
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            connection = await _connection(db)
+            if connection is None:
+                return redirect("/plugins/homeassistant/")
+            connection_id = connection.id
+            await db.delete(connection)
+    await log_audit(current_app, *_actor(), "homeassistant.connection_removed",
+                    entity_type="homeassistant_connection", entity_id=connection_id)
+    return redirect("/plugins/homeassistant/")
+
+
+@homeassistant_bp.post("/connection/test")
+@require_role(UserRole.admin)
+async def test_connection():
+    """Read Home Assistant once with what's typed and say what Steward would see.
+    An admin's explicit action, so it may contact it."""
+    form = await request.form
+    url = (form.get("url") or "").strip().rstrip("/")
+    token = no_spaces(form.get("token"))
+    if not token:
+        async with current_app.db_sessionmaker() as db:
+            connection = await _connection(db)
+        if connection is not None:
+            token = decrypt_secret(connection.token, context="Home Assistant token")
+    found: dict = {}
+    if not url.startswith(("http://", "https://")):
+        found["error"] = "Enter the address first: http://… or https://…"
+    elif not token:
+        found["error"] = "Enter a token first. A request without one counts toward Home Assistant banning this address."
+    else:
+        try:
+            states = await client.read_states(url, token, bool(form.get("verify_tls")))
+        except (client.Unreachable, client.Refused) as exc:
+            found["error"] = str(exc)
+        else:
+            snapshot = read_snapshot(states)
+            found = {"entities": len(states), "installable": sum(1 for u in snapshot["updates"] if u["waiting"] and u["kind"]),
+                     "updates": sum(1 for u in snapshot["updates"] if u["kind"]),
+                     "backup": snapshot["backup"]["known"], "automation": snapshot["automation"]}
+    return await render_template("homeassistant/_test.html", found=found)
diff --git a/plugins/homeassistant/scheduler.py b/plugins/homeassistant/scheduler.py
new file mode 100644
index 0000000..f7e84ac
--- /dev/null
+++ b/plugins/homeassistant/scheduler.py
@@ -0,0 +1,95 @@
+# plugins/homeassistant/scheduler.py
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+from sqlalchemy import select
+
+from steward.core.crypto import decrypt_secret
+from steward.core.scheduler import ScheduledTask
+
+from . import client
+from .models import HomeAssistantConnection
+from .service import RESOURCE, SOURCE, backup_age_hours, installable, read_snapshot
+from .updates import advance
+
+if TYPE_CHECKING:
+    from quart import Quart
+
+logger = logging.getLogger(__name__)
+
+# An update under way is followed each tick, with one small read of its entity.
+TICK_SECONDS = 15
+# The whole entity list is a few hundred KB, and updates and backups change rarely.
+POLL_SECONDS = 300
+# A step past this is given up until the next tick, whatever the client's own
+# timeouts did: a slow drip of bytes can keep each read inside its timeout for ever.
+STEP_DEADLINE_SECONDS = 60
+
+
+def make_tick_task(app: "Quart") -> ScheduledTask:
+    busy = asyncio.Lock()
+
+    async def tick() -> None:
+        if busy.locked():
+            return
+        async with busy:
+            for step in (poll, advance):
+                try:
+                    await asyncio.wait_for(step(app), STEP_DEADLINE_SECONDS)
+                except TimeoutError:
+                    logger.warning("Home Assistant %s gave up after %d s", step.__name__, STEP_DEADLINE_SECONDS)
+
+    return ScheduledTask(name="homeassistant_tick", coro_factory=tick,
+                         interval_seconds=TICK_SECONDS, run_on_startup=True)
+
+
+def is_due(connection, now: datetime) -> bool:
+    """Whether to poll. Never after a refusal: another refused request counts
+    toward Home Assistant banning this address."""
+    if connection.refused:
+        return False
+    if connection.last_polled_at is None:
+        return True
+    return (now - connection.last_polled_at).total_seconds() >= POLL_SECONDS - TICK_SECONDS / 2
+
+
+async def poll(app: "Quart", transport=None) -> None:
+    from steward.core.alerts import MetricBatch
+
+    now = datetime.now(timezone.utc)
+    async with app.db_sessionmaker() as session:
+        connection = (await session.execute(select(HomeAssistantConnection))).scalars().first()
+    if connection is None or not is_due(connection, now):
+        return
+    token = decrypt_secret(connection.token, context="Home Assistant token")
+    snapshot, error, refused = None, None, False
+    try:
+        snapshot = read_snapshot(await client.read_states(connection.url, token, connection.verify_tls, transport))
+    except client.Refused as exc:
+        error, refused = str(exc), True
+    except client.Unreachable as exc:
+        error = str(exc)
+    async with app.db_sessionmaker() as session:
+        async with session.begin():
+            row = await session.get(HomeAssistantConnection, connection.id)
+            if row is None:         # removed while it was being polled
+                return
+            batch = MetricBatch()
+            row.last_polled_at, row.last_error, row.refused = now, error, refused
+            if snapshot is None:
+                batch.add(source_module=SOURCE, resource_name=RESOURCE, metric_name="is_up", value=0.0, recorded_at=now)
+            else:
+                row.last_ok_at = now
+                row.updates, row.backup, row.automation = snapshot["updates"], snapshot["backup"], snapshot["automation"]
+                batch.add(source_module=SOURCE, resource_name=RESOURCE, metric_name="is_up", value=1.0, recorded_at=now)
+                batch.add(source_module=SOURCE, resource_name=RESOURCE, metric_name="updates_waiting",
+                          value=len(installable(row)), recorded_at=now)
+                age = backup_age_hours(snapshot["backup"], now)
+                if age is not None:
+                    batch.add(source_module=SOURCE, resource_name=RESOURCE, metric_name="backup_age_hours",
+                              value=age, recorded_at=now)
+            await batch.record(session)
diff --git a/plugins/homeassistant/service.py b/plugins/homeassistant/service.py
new file mode 100644
index 0000000..0f1f145
--- /dev/null
+++ b/plugins/homeassistant/service.py
@@ -0,0 +1,169 @@
+# plugins/homeassistant/service.py
+"""Reading what Home Assistant reports: which updates it knows of, which of them
+Steward may install, and how its backups stand."""
+from __future__ import annotations
+
+import secrets
+from datetime import datetime, timezone
+
+SOURCE = "homeassistant"
+RESOURCE = "homeassistant"
+# Backups run daily, so past a day and a bit one has been missed.
+STALE_BACKUP_HOURS = 26
+AUTOMATION_ID = "steward_maintenance"
+
+CORE = "update.home_assistant_core_update"
+OS = "update.home_assistant_operating_system_update"
+SUPERVISOR = "update.home_assistant_supervisor_update"
+LAST_BACKUP = "sensor.backup_last_successful_automatic_backup"
+NEXT_BACKUP = "sensor.backup_next_scheduled_automatic_backup"
+BACKUP_MANAGER = "sensor.backup_backup_manager_state"
+BACKUP_EVENT = "event.backup_automatic_backup"
+
+_KIND_WORDS = {"core": "Core", "os": "Operating system", "addon": "Add-on"}
+_KIND_ORDER = {"core": 0, "os": 1, "addon": 2, None: 3}
+
+# The automation that gives Steward its only way to act on Home Assistant. It
+# takes a backup or installs a Core, OS or add-on update, and does nothing else.
+AUTOMATION_YAML = """\
+id: steward_maintenance
+alias: Steward maintenance
+description: Lets Steward take a backup or install a Core, OS or add-on update. Does nothing else.
+mode: single
+triggers:
+  - trigger: webhook
+    webhook_id: {webhook_id}
+    allowed_methods:
+      - POST
+    local_only: true
+variables:
+  request: "{{{{ trigger.json | default({{}}) }}}}"
+  allowed: >-
+    {{{{ integration_entities('hassio') | select('match', 'update\\\\.')
+       | reject('eq', 'update.home_assistant_supervisor_update') | list }}}}
+actions:
+  - choose:
+      - conditions:
+          - condition: template
+            value_template: "{{{{ request.get('action') == 'backup' }}}}"
+        sequence:
+          - action: backup.create_automatic
+      - conditions:
+          - condition: template
+            value_template: "{{{{ request.get('action') == 'install' and request.get('entity_id') in allowed }}}}"
+        sequence:
+          - action: update.install
+            target:
+              entity_id: "{{{{ request.get('entity_id') }}}}"
+"""
+
+
+def new_webhook_id() -> str:
+    return "steward_" + secrets.token_urlsafe(32)
+
+
+def automation_yaml(webhook_id: str) -> str:
+    return AUTOMATION_YAML.format(webhook_id=webhook_id)
+
+
+def no_spaces(typed: str | None) -> str:
+    """A pasted token or id with a terminal's wrapping taken back out."""
+    return "".join((typed or "").split())
+
+
+def classify(entity: dict) -> str | None:
+    """"core", "os" or "addon" for an update Steward may install; None for the
+    rest: Supervisor updates itself, and firmware and custom integrations are
+    Home Assistant's own to offer."""
+    entity_id = entity.get("entity_id")
+    if entity_id == CORE:
+        return "core"
+    if entity_id == OS:
+        return "os"
+    if entity_id == SUPERVISOR:
+        return None
+    picture = (entity.get("attributes") or {}).get("entity_picture") or ""
+    return "addon" if picture.startswith("/api/hassio/addons/") else None
+
+
+def _update(entity: dict) -> dict:
+    a = entity.get("attributes") or {}
+    kind = classify(entity)
+    url = a.get("release_url") or ""
+    return {"entity_id": entity["entity_id"], "kind": kind, "kind_word": _KIND_WORDS.get(kind, "Other"),
+            "title": a.get("title") or a.get("friendly_name") or entity["entity_id"],
+            "installed": a.get("installed_version"), "latest": a.get("latest_version"),
+            "waiting": entity.get("state") == "on", "in_progress": bool(a.get("in_progress")),
+            "auto_update": bool(a.get("auto_update")),
+            "release_url": url if url.startswith(("http://", "https://")) else None}
+
+
+def _state(entity: dict | None) -> str | None:
+    value = (entity or {}).get("state")
+    return None if value in (None, "unknown", "unavailable") else value
+
+
+def parse_moment(value: str | None) -> datetime | None:
+    try:
+        moment = datetime.fromisoformat(value) if value else None
+    except ValueError:
+        return None
+    if moment is not None and moment.tzinfo is None:
+        moment = moment.replace(tzinfo=timezone.utc)
+    return moment
+
+
+def is_newer(value: str | None, than: str | None) -> bool:
+    """Whether reading `value` is a later moment than `than`, where `than` may be never."""
+    moment, before = parse_moment(value), parse_moment(than)
+    return moment is not None and (before is None or moment > before)
+
+
+def read_snapshot(states: list[dict]) -> dict:
+    """What a poll keeps of Home Assistant's entities."""
+    by_id = {e.get("entity_id"): e for e in states}
+    updates = [_update(e) for i, e in by_id.items() if (i or "").startswith("update.") and e.get("state") != "unavailable"]
+    updates.sort(key=lambda u: (not u["waiting"], _KIND_ORDER[u["kind"]], u["title"].lower()))
+    event = by_id.get(BACKUP_EVENT) or {}
+    event_attrs = event.get("attributes") or {}
+    automation = next((e for e in states if (e.get("entity_id") or "").startswith("automation.")
+                       and (e.get("attributes") or {}).get("id") == AUTOMATION_ID), None)
+    return {
+        "updates": updates,
+        "backup": {"last_ok": _state(by_id.get(LAST_BACKUP)), "next": _state(by_id.get(NEXT_BACKUP)),
+                   "manager": _state(by_id.get(BACKUP_MANAGER)), "last_result": event_attrs.get("event_type"),
+                   "failed_reason": event_attrs.get("failed_reason"), "known": LAST_BACKUP in by_id,
+                   "automation_entity": automation.get("entity_id") if automation else None},
+        "automation": automation.get("state") if automation else None,
+    }
+
+
+def backup_age_hours(backup: dict | None, now: datetime) -> float | None:
+    last = parse_moment((backup or {}).get("last_ok"))
+    return None if last is None else max(0.0, (now - last).total_seconds() / 3600)
+
+
+def installable(connection) -> list[dict]:
+    """The waiting updates Steward may install."""
+    return [u for u in (connection.updates or []) if u["waiting"] and u["kind"]]
+
+
+def backup_verdict(backup: dict | None, now: datetime) -> tuple[str, str]:
+    """(tone, sentence) for how the backups stand."""
+    backup = backup or {}
+    if not backup.get("known"):
+        return "warn", "Home Assistant reports no automatic backups. Turn them on under Settings → System → Backups."
+    age = backup_age_hours(backup, now)
+    if age is None:
+        return "warn", "Home Assistant has no successful automatic backup yet."
+    if backup.get("last_result") == "failed":
+        reason = backup.get("failed_reason")
+        return "warn", f"The newest backup attempt failed{f' ({reason})' if reason else ''}."
+    if age > STALE_BACKUP_HOURS:
+        return "warn", f"The last good backup is {age:.0f} hours old, past the {STALE_BACKUP_HOURS} hours expected."
+    return "ok", "Backups are up to date."
+
+
+def automation_entity_id(connection) -> str:
+    """The automation's entity, as a poll found it."""
+    return (connection.backup or {}).get("automation_entity") or f"automation.{AUTOMATION_ID}"
diff --git a/plugins/homeassistant/templates/homeassistant/_test.html b/plugins/homeassistant/templates/homeassistant/_test.html
new file mode 100644
index 0000000..325c09c
--- /dev/null
+++ b/plugins/homeassistant/templates/homeassistant/_test.html
@@ -0,0 +1,6 @@
+{# plugins/homeassistant/templates/homeassistant/_test.html — what Test found. #}
+{% if found.error %}<p class="tone-crit" style="margin-bottom:0;">{{ found.error }}</p>
+{% else %}<p class="tone-ok" style="margin-bottom:0.25rem;">Answered: {{ found.entities }} entities, {{ found.updates }} Core, OS and add-on update entities, {{ found.installable }} waiting.</p>
+<p class="{{ 'field-hint' if found.backup else 'tone-warn' }}" style="margin:0;">{{ "Automatic backups are readable." if found.backup else "No automatic backup sensor: turn on automatic backups in Home Assistant." }}</p>
+<p class="{{ 'field-hint' if found.automation == 'on' else 'tone-warn' }}" style="margin:0;">{% if found.automation == 'on' %}The Steward maintenance automation is on.{% elif found.automation %}The Steward maintenance automation is {{ found.automation }}.{% else %}The Steward maintenance automation isn't there yet.{% endif %}</p>
+{% endif %}
diff --git a/plugins/homeassistant/templates/homeassistant/form.html b/plugins/homeassistant/templates/homeassistant/form.html
new file mode 100644
index 0000000..908062b
--- /dev/null
+++ b/plugins/homeassistant/templates/homeassistant/form.html
@@ -0,0 +1,80 @@
+{# plugins/homeassistant/templates/homeassistant/form.html — where Home Assistant
+   is, the read-only token Steward reads it with, and the automation to add so
+   Steward can install updates. A secret is never sent back in a field; a blank
+   one keeps what's saved. #}
+{% extends "base.html" %}
+{% from "_macros.html" import crumbs, page_header %}
+{% block title %}Home Assistant connection — Steward{% endblock %}
+{% block breadcrumb %}{{ crumbs([("Home Assistant", "/plugins/homeassistant/"), ("Connection", "")]) }}{% endblock %}
+{% block content %}
+{{ page_header("Home Assistant connection") }}
+
+<div class="form-page">
+{% if error %}<p class="field-error" role="alert">{{ error }}</p>{% endif %}
+<form method="post" id="ha-form" action="/plugins/homeassistant/connection">{{ csrf_field() }}
+  <section class="card form-section" aria-labelledby="ha-where">
+    <h2 id="ha-where" class="section-title">Reading it</h2>
+    <div class="form-pair">
+      <div class="field">
+        <label class="field-label" for="ha-url">Address</label>
+        <input class="input mono" id="ha-url" name="url" required maxlength="500" value="{{ v.url or '' }}" placeholder="http://homeassistant.local:8123" autocomplete="off">
+        <span class="field-hint">Its address on your local network, not a public one: the webhook only answers local requests.</span>
+      </div>
+      <div class="field">
+        <label class="field-label" for="ha-token">Read-only token</label>
+        <input class="input" id="ha-token" name="token" type="password" autocomplete="new-password"
+               placeholder="{{ 'kept; type to change' if v.has_token else 'a long-lived access token' }}">
+        <span class="field-hint">Use a long-lived token of a user in Home Assistant's read-only group, so nothing Steward holds can act on your house.</span>
+      </div>
+    </div>
+    <label style="display:flex;gap:0.5rem;align-items:center;cursor:pointer;margin-bottom:0.9rem;">
+      <input type="checkbox" name="verify_tls" value="on"{% if v.verify_tls %} checked{% endif %}> Check its HTTPS certificate
+    </label>
+    <div style="display:flex;align-items:center;gap:0.75rem;flex-wrap:wrap;">
+      <button type="button" class="btn btn-sm btn-ghost"
+              hx-post="/plugins/homeassistant/connection/test" hx-include="#ha-form" hx-target="#ha-test"
+              hx-indicator="#ha-testing">Test</button>
+      <span id="ha-testing" class="field-hint htmx-indicator">Reading…</span>
+      <span class="field-hint">Reads Home Assistant once with what's typed here. Nothing is saved.</span>
+    </div>
+    <div id="ha-test" aria-live="polite"></div>
+  </section>
+
+  <section class="card form-section" aria-labelledby="ha-install">
+    <h2 id="ha-install" class="section-title">Installing updates</h2>
+    <div class="field">
+      <label class="field-label" for="ha-webhook">Webhook id</label>
+      <input class="input" id="ha-webhook" name="webhook_id" type="password" autocomplete="new-password"
+             placeholder="{{ 'kept; type to change' if v.has_webhook else 'blank makes a new one' }}">
+      <span class="field-hint">Steward makes one when you first save. Type one only if the automation in Home Assistant already has its own.</span>
+    </div>
+    {% if automation %}
+    <p class="field-hint">Add this automation in Home Assistant (Settings → Automations → Create → Edit in YAML). It can take a backup or install a Core, OS or add-on update, and nothing else. It contains the webhook id, so treat it like a password.</p>
+    <pre class="mono" id="ha-automation" style="overflow-x:auto;font-size:0.8rem;">{{ automation }}</pre>
+    <button type="button" class="btn btn-sm btn-ghost"
+            onclick="navigator.clipboard.writeText(document.getElementById('ha-automation').textContent).then(()=>this.textContent='Copied')">Copy</button>
+    {% else %}
+    <p class="field-hint" style="margin-bottom:0;">Once saved, this shows the automation to add to Home Assistant.</p>
+    {% endif %}
+  </section>
+
+  <div class="form-actions">
+    <button type="submit" class="btn">Save</button>
+    <a href="/plugins/homeassistant/" class="btn btn-ghost">Cancel</a>
+  </div>
+</form>
+
+{% if connection %}
+<section class="card danger-zone" aria-labelledby="ha-remove">
+  <div>
+    <h2 id="ha-remove" class="section-title">Remove this connection</h2>
+    <p class="field-hint">Steward stops reading Home Assistant and forgets its token. Remove the automation in Home Assistant too.</p>
+  </div>
+  <form method="post" action="/plugins/homeassistant/connection/delete"
+        data-confirm="Remove the Home Assistant connection? Steward stops reading it and forgets its token.">{{ csrf_field() }}
+    <button class="btn btn-danger" type="submit">Remove connection</button>
+  </form>
+</section>
+{% endif %}
+</div>
+{% endblock %}
diff --git a/plugins/homeassistant/templates/homeassistant/index.html b/plugins/homeassistant/templates/homeassistant/index.html
new file mode 100644
index 0000000..2e1b475
--- /dev/null
+++ b/plugins/homeassistant/templates/homeassistant/index.html
@@ -0,0 +1,97 @@
+{# plugins/homeassistant/templates/homeassistant/index.html — whether Home
+   Assistant answers, which updates are waiting with Install beside each, how its
+   backups stand, and the updates Steward has run. Refreshes itself while one is
+   under way. #}
+{% extends "base.html" %}
+{% from "_macros.html" import page_header, status_badge, stat_tile, empty_state %}
+{% block title %}Home Assistant — Steward{% endblock %}
+{% block content %}
+{% set admin = session.user_role == 'admin' %}
+{% call page_header("Home Assistant", connection.url if connection else none) %}
+  {% if admin and connection %}<a class="btn btn-sm btn-ghost" href="/plugins/homeassistant/connection">Connection</a>{% endif %}
+{% endcall %}
+
+{% if not connection %}
+<div class="card">{{ empty_state("Not connected. Connect a Home Assistant instance to see its waiting updates and backups, and to install updates after a confirmed backup.",
+                                 "Connect" if admin else none, "/plugins/homeassistant/connection") }}</div>
+{% else %}
+{% if error %}<p class="field-error" role="alert">{{ error }}</p>{% endif %}
+<div id="ha-body"{% if under_way %} hx-get="/plugins/homeassistant/" hx-select="#ha-body" hx-trigger="every 15s" hx-swap="outerHTML"{% endif %}>
+<p class="page-verdict">
+  {% if under_way %}{{ status_badge('pending', 'Updating') }}
+    {{ under_way.title }}: {{ "taking a backup first" if under_way.state == 'backing_up' else "installing " ~ under_way.to_version }}. Home Assistant may stop answering for a while.
+  {% elif reach == 'up' %}{{ status_badge('up', 'Answering') }}
+    {% if installable %}{{ installable|length }} update{{ "s" if installable|length != 1 }} waiting.{% else %}Everything Steward can update is up to date.{% endif %}
+    Read {{ connection.last_polled_at|ago }}.
+  {% elif reach == 'down' %}{{ status_badge('down', 'Not answering') }} <span class="tone-crit">{{ says }}</span>
+    {% if connection.last_ok_at %}Last answered {{ connection.last_ok_at|ago }}; what's below is from then.{% endif %}
+    {% if admin and connection.refused %}<form method="post" action="/plugins/homeassistant/connection/retry" style="display:inline;">{{ csrf_field() }}<button class="btn btn-sm btn-ghost" type="submit">Try again</button></form>{% endif %}
+  {% else %}{{ status_badge('pending', 'Not read yet') }} {{ says }}{% endif %}
+</p>
+
+{% if connection.last_ok_at and connection.automation != 'on' %}
+<p class="tone-warn">{% if connection.automation %}The Steward maintenance automation is {{ connection.automation }} in Home Assistant{% else %}The Steward maintenance automation isn't in Home Assistant{% endif %}, so updates can't be installed from here.
+  {% if admin %}<a href="/plugins/homeassistant/connection">See what to add.</a>{% endif %}</p>
+{% endif %}
+
+<div class="stat-strip">
+  {{ stat_tile("Updates waiting", installable|length if connection.last_ok_at else "—") }}
+  {{ stat_tile("Last good backup", (backup_at|ago) if backup_at else "—", state="degraded" if backup_tone == 'warn' else none) }}
+  {{ stat_tile("Next backup", (backup_next|ago) if backup_next else "—") }}
+</div>
+<p class="{{ 'tone-warn' if backup_tone == 'warn' else 'field-hint' }}">{{ backup_says }}</p>
+
+<section class="card" aria-labelledby="ha-updates">
+  <h2 id="ha-updates" class="section-title">Updates</h2>
+  {% if installable %}
+  <table class="table">
+    <thead><tr><th scope="col">What</th><th scope="col">Now</th><th scope="col">New</th>{% if admin %}<th scope="col"></th>{% endif %}</tr></thead>
+    <tbody>
+    {% for u in installable %}
+      <tr>
+        <td>{{ u.title }}<div class="cell-sub">{{ u.kind_word }}{% if u.auto_update %}, updates itself{% endif %}</div></td>
+        <td class="mono">{{ u.installed or "—" }}</td>
+        <td class="mono">{% if u.release_url %}<a href="{{ u.release_url }}" rel="noopener noreferrer" target="_blank">{{ u.latest or "—" }}</a>{% else %}{{ u.latest or "—" }}{% endif %}</td>
+        {% if admin %}<td style="text-align:right;">
+          {% if connection.automation == 'on' and not under_way and not connection.refused and not u.in_progress %}
+          <form method="post" action="/plugins/homeassistant/install"
+                data-confirm="Install {{ u.title }} {{ u.latest }}? Steward takes a backup first and won't install without it. {{ 'Home Assistant restarts.' if u.kind == 'core' else ('The machine reboots.' if u.kind == 'os' else 'The add-on restarts.') }}">{{ csrf_field() }}
+            <input type="hidden" name="entity_id" value="{{ u.entity_id }}">
+            <button class="btn btn-sm" type="submit">Install</button>
+          </form>
+          {% endif %}
+        </td>{% endif %}
+      </tr>
+    {% endfor %}
+    </tbody>
+  </table>
+  {% else %}
+  <p class="field-hint" style="margin:0;">Nothing waiting.</p>
+  {% endif %}
+  {% if current %}<p class="field-hint">Up to date: {% for u in current %}{{ u.title }} {{ u.installed }}{{ ", " if not loop.last }}{% endfor %}.</p>{% endif %}
+  {% if others %}<p class="field-hint" style="margin-bottom:0;">Waiting, for Home Assistant to install: {% for u in others %}{{ u.title }}{{ ", " if not loop.last }}{% endfor %}. Steward installs Core, OS and add-on updates only.</p>{% endif %}
+</section>
+
+{% if runs %}
+<section class="card" aria-labelledby="ha-runs">
+  <h2 id="ha-runs" class="section-title">Updates run from Steward</h2>
+  <table class="table">
+    <thead><tr><th scope="col">What</th><th scope="col">Versions</th><th scope="col">Outcome</th><th scope="col">Started</th></tr></thead>
+    <tbody>
+    {% for r in runs %}
+      <tr>
+        <td>{{ r.title }}<div class="cell-sub">by {{ r.started_by or "Steward" }}</div></td>
+        <td class="mono">{{ r.from_version or "—" }} → {{ r.to_version or "—" }}</td>
+        <td>{% if r.state == 'done' %}{{ status_badge('up', 'Done') }}
+          {% elif r.state == 'failed' %}{{ status_badge('down', 'Failed') }}<div class="cell-sub">{{ r.message }}</div>
+          {% else %}{{ status_badge('pending', 'Taking a backup' if r.state == 'backing_up' else 'Installing') }}{% endif %}</td>
+        <td>{{ r.started_at|ago }}</td>
+      </tr>
+    {% endfor %}
+    </tbody>
+  </table>
+</section>
+{% endif %}
+</div>
+{% endif %}
+{% endblock %}
diff --git a/plugins/homeassistant/timeline.py b/plugins/homeassistant/timeline.py
new file mode 100644
index 0000000..ff52da5
--- /dev/null
+++ b/plugins/homeassistant/timeline.py
@@ -0,0 +1,34 @@
+# plugins/homeassistant/timeline.py
+"""What Home Assistant maintenance puts on the timeline: each update once it ends."""
+from __future__ import annotations
+
+from steward.timeline import catalog
+from steward.timeline.record import record_or_update
+
+TYPES = (
+    ("homeassistant.updated", "Home Assistant updated", "automation", "service", True,
+     "Steward installed a Home Assistant update after a confirmed backup."),
+    ("homeassistant.update_failed", "Home Assistant update failed", "automation", "service", True,
+     "A Home Assistant update Steward was asked to install didn't finish."),
+)
+
+
+def register() -> None:
+    for t in TYPES:
+        catalog.register_type(*t)
+    from .models import HomeAssistantUpdate
+    catalog.register_ref("homeassistant_update", HomeAssistantUpdate)
+
+
+async def update_entry(session, run) -> None:
+    done = run.state == "done"
+    versions = f"{run.from_version} to {run.to_version}" if run.to_version else "a new version"
+    summary = (f"{run.title} updated from {versions}" if done
+               else f"{run.title} update failed: {run.message}")
+    await record_or_update(
+        session, ("homeassistant_update", run.id),
+        type="homeassistant.updated" if done else "homeassistant.update_failed",
+        at=run.started_at, until=run.finished_at, summary=summary, subject=run.title,
+        detail={"entity_id": run.entity_id, "from": run.from_version, "to": run.to_version,
+                "url": "/plugins/homeassistant/"},
+        actor=run.started_by or "Steward")
diff --git a/plugins/homeassistant/updates.py b/plugins/homeassistant/updates.py
new file mode 100644
index 0000000..74a75ff
--- /dev/null
+++ b/plugins/homeassistant/updates.py
@@ -0,0 +1,214 @@
+# plugins/homeassistant/updates.py
+"""Installing one update: take a backup, confirm it landed, install, and follow
+it to its end. Its row holds where it has got to, so Steward restarting, or Home
+Assistant restarting as an update does, only delays the next step."""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from steward.core.crypto import decrypt_secret
+
+from . import client
+from .models import HomeAssistantConnection, HomeAssistantUpdate
+from .service import BACKUP_EVENT, LAST_BACKUP, automation_entity_id, is_newer, parse_moment
+from .timeline import update_entry
+
+logger = logging.getLogger(__name__)
+
+UNDER_WAY = ("backing_up", "installing")
+BACKUP_DEADLINE_S = 30 * 60
+# How long the automation has to show it ran before the webhook is called wrong.
+TRIGGER_GRACE_S = 90
+# An OS update reboots the machine, and a Pi is slow to come back.
+INSTALL_DEADLINE_S = {"addon": 20 * 60, "core": 30 * 60, "os": 45 * 60}
+
+
+class CannotStart(Exception):
+    """Why an update wasn't started, said to the person."""
+
+
+def _secrets(connection) -> tuple[str, str]:
+    return (decrypt_secret(connection.token, context="Home Assistant token"),
+            decrypt_secret(connection.webhook_id, context="Home Assistant webhook id"))
+
+
+async def _mark_refused(app, connection_id: str, said: str) -> None:
+    async with app.db_sessionmaker() as session:
+        async with session.begin():
+            row = await session.get(HomeAssistantConnection, connection_id)
+            if row is not None:
+                row.refused, row.last_error = True, said
+
+
+async def under_way(session) -> HomeAssistantUpdate | None:
+    return (await session.execute(select(HomeAssistantUpdate).where(
+        HomeAssistantUpdate.state.in_(UNDER_WAY)))).scalars().first()
+
+
+async def start_update(app, entity_id: str, username: str, transport=None) -> HomeAssistantUpdate:
+    """Start installing `entity_id`, an explicit action that may contact Home
+    Assistant. Raises CannotStart, with nothing started, unless the update is
+    still waiting and the automation is there to install it."""
+    async with app.db_sessionmaker() as session:
+        connection = (await session.execute(select(HomeAssistantConnection))).scalars().first()
+        if connection is None:
+            raise CannotStart("Home Assistant isn't connected")
+        if connection.refused:
+            raise CannotStart("Home Assistant refused Steward's token")
+        if await under_way(session) is not None:
+            raise CannotStart("another update is under way")
+        if connection.automation != "on":
+            raise CannotStart("the Steward maintenance automation isn't on in Home Assistant")
+    url, verify = connection.url, connection.verify_tls
+    token, webhook_id = _secrets(connection)
+    known = next((u for u in connection.updates or [] if u["entity_id"] == entity_id), None)
+    if known is None or not known["kind"]:
+        raise CannotStart("that isn't a Core, OS or add-on update")
+    try:
+        current = await client.read_state(url, entity_id, token, verify, transport)
+        backup = await client.read_state(url, LAST_BACKUP, token, verify, transport)
+        event = await client.read_state(url, BACKUP_EVENT, token, verify, transport)
+        automation = await client.read_state(url, automation_entity_id(connection), token, verify, transport)
+    except client.Refused as exc:
+        await _mark_refused(app, connection.id, str(exc))
+        raise CannotStart(str(exc).rstrip(".")) from None
+    except client.Unreachable as exc:
+        raise CannotStart(str(exc).rstrip(".")) from None
+    if backup is None:
+        raise CannotStart("Home Assistant has no automatic backups to confirm one with")
+    if current is None or current.get("state") != "on":
+        raise CannotStart("that update is no longer waiting")
+    attrs = current.get("attributes") or {}
+    run = HomeAssistantUpdate(
+        entity_id=entity_id, title=known["title"], kind=known["kind"],
+        from_version=attrs.get("installed_version"), to_version=attrs.get("latest_version"),
+        started_by=username, backup_before=backup.get("state"),
+        backup_event_before=(event or {}).get("state"),
+        triggered_before=((automation or {}).get("attributes") or {}).get("last_triggered"))
+    async with app.db_sessionmaker() as session:
+        try:
+            async with session.begin():
+                session.add(run)
+                await session.flush()
+                run_id = run.id
+        except IntegrityError:
+            raise CannotStart("another update is under way") from None
+    try:
+        await client.send_webhook(url, webhook_id, {"action": "backup"}, verify, transport)
+    except client.Unreachable as exc:
+        await _finish(app, run_id, "failed", f"The backup couldn't be started: {exc} Nothing was installed.")
+        raise CannotStart(str(exc).rstrip(".")) from None
+    async with app.db_sessionmaker() as session:
+        return await session.get(HomeAssistantUpdate, run_id)
+
+
+async def _finish(app, run_id: str, state: str, message: str | None) -> None:
+    async with app.db_sessionmaker() as session:
+        async with session.begin():
+            run = await session.get(HomeAssistantUpdate, run_id)
+            run.state, run.message, run.finished_at = state, message, datetime.now(timezone.utc)
+            # The page should show the new versions now, not at the next poll.
+            connection = (await session.execute(select(HomeAssistantConnection))).scalars().first()
+            if connection is not None:
+                connection.last_polled_at = None
+            try:
+                async with session.begin_nested():
+                    await update_entry(session, run)
+            except Exception:
+                logger.exception("Home Assistant update %s: its timeline entry could not be written", run_id)
+
+
+async def advance(app, transport=None) -> None:
+    """Move the update under way one step on, if there is one."""
+    async with app.db_sessionmaker() as session:
+        run = await under_way(session)
+        connection = (await session.execute(select(HomeAssistantConnection))).scalars().first()
+    if run is None:
+        return
+    if connection is None:
+        await _finish(app, run.id, "failed", "The Home Assistant connection was removed.")
+        return
+    try:
+        if run.state == "backing_up":
+            await _advance_backup(app, run, connection, transport)
+        else:
+            await _advance_install(app, run, connection, transport)
+    except client.Refused as exc:
+        await _mark_refused(app, connection.id, str(exc))
+        await _finish(app, run.id, "failed", str(exc))
+    except client.Unreachable as exc:
+        # Expected while Home Assistant restarts: the next tick asks again.
+        logger.info("Home Assistant update %s: %s", run.id, exc)
+        await _past_deadline(app, run)
+
+
+def _elapsed(since: datetime) -> float:
+    return (datetime.now(timezone.utc) - since).total_seconds()
+
+
+async def _past_deadline(app, run) -> bool:
+    if run.state == "backing_up" and _elapsed(run.started_at) > BACKUP_DEADLINE_S:
+        await _finish(app, run.id, "failed",
+                      f"No new backup within {BACKUP_DEADLINE_S // 60} minutes. Nothing was installed.")
+        return True
+    if run.state == "installing":
+        limit = INSTALL_DEADLINE_S.get(run.kind, INSTALL_DEADLINE_S["os"])
+        if _elapsed(run.install_sent_at or run.backup_at or run.started_at) > limit:
+            await _finish(app, run.id, "failed",
+                          f"Still not on {run.to_version} after {limit // 60} minutes. Look in Home Assistant.")
+            return True
+    return False
+
+
+async def _advance_backup(app, run, connection, transport) -> None:
+    token, _webhook = _secrets(connection)
+    url, verify = connection.url, connection.verify_tls
+    backup = await client.read_state(url, LAST_BACKUP, token, verify, transport)
+    value = (backup or {}).get("state")
+    if is_newer(value, run.backup_before):
+        async with app.db_sessionmaker() as session:
+            async with session.begin():
+                row = await session.get(HomeAssistantUpdate, run.id)
+                row.state, row.backup_at = "installing", parse_moment(value)
+        return
+    event = await client.read_state(url, BACKUP_EVENT, token, verify, transport)
+    if event and event.get("state") != run.backup_event_before \
+            and (event.get("attributes") or {}).get("event_type") == "failed":
+        reason = (event.get("attributes") or {}).get("failed_reason")
+        await _finish(app, run.id, "failed",
+                      f"The backup failed{f' ({reason})' if reason else ''}. Nothing was installed.")
+        return
+    if _elapsed(run.started_at) > TRIGGER_GRACE_S:
+        automation = await client.read_state(url, automation_entity_id(connection), token, verify, transport)
+        triggered = ((automation or {}).get("attributes") or {}).get("last_triggered")
+        if not is_newer(triggered, run.triggered_before):
+            await _finish(app, run.id, "failed",
+                          "Home Assistant didn't run the maintenance automation, so its webhook id may not "
+                          "match Steward's, or Steward isn't on its local network. Nothing was installed.")
+            return
+    await _past_deadline(app, run)
+
+
+async def _advance_install(app, run, connection, transport) -> None:
+    token, webhook_id = _secrets(connection)
+    url, verify = connection.url, connection.verify_tls
+    if run.install_sent_at is None:
+        await client.send_webhook(url, webhook_id, {"action": "install", "entity_id": run.entity_id},
+                                  verify, transport)
+        async with app.db_sessionmaker() as session:
+            async with session.begin():
+                row = await session.get(HomeAssistantUpdate, run.id)
+                row.install_sent_at = datetime.now(timezone.utc)
+        return
+    current = await client.read_state(url, run.entity_id, token, verify, transport)
+    attrs = (current or {}).get("attributes") or {}
+    installed = attrs.get("installed_version")
+    if installed and (installed == run.to_version
+                      or (current.get("state") == "off" and installed != run.from_version)):
+        await _finish(app, run.id, "done", None)
+        return
+    await _past_deadline(app, run)
diff --git a/steward/audit/service.py b/steward/audit/service.py
index 0bc0669..edff4d8 100644
--- a/steward/audit/service.py
+++ b/steward/audit/service.py
@@ -214,6 +214,9 @@ ACTIONS = {
     "prometheus.template_created": "added Prometheus widget template",
     "prometheus.template_updated": "changed Prometheus widget template",
     "prometheus.template_deleted": "deleted Prometheus widget template",
+    "homeassistant.connection_saved": "saved the Home Assistant connection",
+    "homeassistant.connection_removed": "removed the Home Assistant connection",
+    "homeassistant.update_started": "started a Home Assistant update",
     "syslog.source_added": "added syslog source",
     "syslog.source_updated": "changed syslog source",
     "syslog.source_removed": "removed syslog source",
@@ -244,7 +247,7 @@ AREAS = (("auth.", "Sign-ins"), ("account.", "Accounts"), ("user.", "Users"), ("
          ("ansible.run", "Playbook runs"), ("ansible.", "Ansible (all)"),
          ("settings.", "Settings"), ("plugin.", "Plugins"), ("api_token.", "AI access"),
          ("docker.", "Docker"), ("snmp.", "SNMP"), ("prometheus.", "Prometheus"),
-         ("syslog.", "Syslog"), ("backup.", "Backups"), ("timeline.", "Timeline"),
+         ("syslog.", "Syslog"), ("homeassistant.", "Home Assistant"), ("backup.", "Backups"), ("timeline.", "Timeline"),
          ("ticket.", "Tickets"))
 
 
diff --git a/steward/backup/contents.py b/steward/backup/contents.py
index 95993f2..27f3ffb 100644
--- a/steward/backup/contents.py
+++ b/steward/backup/contents.py
@@ -29,6 +29,7 @@ KEPT: frozenset[str] = frozenset({
     # Syslog senders: which were added, under what name, and which ignored.
     "syslog_sources",
     "syslog_patterns",
+    "homeassistant_connection",
 })
 
 # Rows that are measurements, logs, history or in-flight work: left out.
@@ -67,6 +68,7 @@ EXCLUDED: frozenset[str] = frozenset({
     "unifi_events", "unifi_fw_rules", "unifi_networks", "unifi_port_forwards",
     "unifi_speedtest_results", "unifi_wan_stats", "unifi_wlan_stats",
     "syslog_lines",
+    "homeassistant_updates",
 })
 
 # ── What a check compares (#5823) ────────────────────────────────────────────
@@ -91,6 +93,8 @@ ACTIVITY: dict[str, frozenset[str]] = {
     "snmp_devices": frozenset({"last_polled_at", "last_ok_at", "last_error", "missing"}),
     "prometheus_targets": frozenset({"last_scraped_at", "last_ok_at", "last_error", "last_kept",
                                      "last_dropped", "last_capped"}),
+    "homeassistant_connection": frozenset({"refused", "last_polled_at", "last_ok_at", "last_error",
+                                           "updates", "backup", "automation"}),
     # What the sender last sent; its address, name, state and host are setup.
     "syslog_sources": frozenset({"hostname_seen", "first_seen_at", "last_seen_at", "lines_total",
                                  "last_severity", "last_line"}),
@@ -121,5 +125,6 @@ AREAS: tuple[tuple[str, tuple[str, ...]], ...] = (
     ("Ansible", ("ansible_targets", "ansible_groups", "ansible_target_groups", "ansible_schedules")),
     ("Dashboards", ("dashboards", "dashboard_widgets", "dashboard_share_tokens")),
     ("Plugins", ("snmp_devices", "prometheus_targets", "prometheus_widget_templates",
-                 "unifi_known_clients", "syslog_sources", "syslog_patterns")),
+                 "unifi_known_clients", "syslog_sources", "syslog_patterns",
+                 "homeassistant_connection")),
 )
diff --git a/steward/core/formatting.py b/steward/core/formatting.py
index 859b971..8e6c20b 100644
--- a/steward/core/formatting.py
+++ b/steward/core/formatting.py
@@ -298,6 +298,8 @@ def metric_value(name: str, value: Any) -> str:
 # its underscores spaced (SNMP's OID labels are the operator's own words).
 METRICS: dict[str, tuple[str, str]] = {
     "is_up": ("Up or down", "whether the check passed"),
+    "backup_age_hours": ("Age of last backup (hours)", "hours since Home Assistant's last successful automatic backup"),
+    "updates_waiting": ("Updates waiting", "Core, OS and add-on updates Home Assistant has waiting"),
     "is_healthy": ("Health check", "the container's own HEALTHCHECK result"),
     "response_ms": ("Response time", "how long the check took"),
     "ingest_posts": ("Agent posts per minute", "host-agent reports Steward took in the last minute"),
@@ -421,6 +423,7 @@ SOURCES: dict[str, str] = {
     "http": "HTTP monitors", "steward": "Steward itself", "host_agent": "Host agent",
     "docker": "Docker containers", "traefik": "Traefik routers", "unifi": "UniFi",
     "snmp": "SNMP devices", "prometheus": "Prometheus targets", "syslog": "Syslog patterns",
+    "homeassistant": "Home Assistant",
 }
 
 
diff --git a/steward/core/settings.py b/steward/core/settings.py
index 6212769..d703643 100644
--- a/steward/core/settings.py
+++ b/steward/core/settings.py
@@ -181,6 +181,7 @@ DEFAULTS: dict[str, Any] = {
     "plugin.snmp": {"enabled": True},
     "plugin.prometheus": {"enabled": True},
     "plugin.syslog": {"enabled": True},
+    "plugin.homeassistant": {"enabled": True},
     # OIDC single-sign-on
     "oidc.enabled": False,
     "oidc.discovery_url": "",
diff --git a/steward/timeline/catalog.py b/steward/timeline/catalog.py
index c8a9a83..594c65f 100644
--- a/steward/timeline/catalog.py
+++ b/steward/timeline/catalog.py
@@ -118,7 +118,7 @@ for _t in (
 # A source not listed here is Monitoring.
 _SOURCE_AREAS = {
     "icmp": "network", "dns": "network", "tcp": "service", "http": "service",
-    "docker": "service", "prometheus": "service",
+    "docker": "service", "prometheus": "service", "homeassistant": "service",
     "traefik": "network", "unifi": "network",
     "snmp": "hardware",
 }
@@ -154,6 +154,8 @@ _AUDIT_TYPES = (
     ("account.time_zone", None), ("account.email", None),
     ("host_agent.maintenance_plan", "steward.config"), ("host_agent.maintenance", None),
     ("docker.prune.", None),
+    # An update is on the timeline as itself, once it ends (homeassistant.timeline).
+    ("homeassistant.update", None),
     # A change recorded by hand is on the timeline as itself (timeline.changes).
     ("timeline.change", None),
     # A ticket's opening, resolution and reopening are on the timeline as
diff --git a/tests/core/test_metric_catalog.py b/tests/core/test_metric_catalog.py
index a2168ae..1304d80 100644
--- a/tests/core/test_metric_catalog.py
+++ b/tests/core/test_metric_catalog.py
@@ -9,7 +9,7 @@ import steward
 from steward.alerts.service import CORE_METRICS
 
 _REPO = pathlib.Path(steward.__file__).parent.parent
-_PLUGINS = ("traefik", "unifi", "docker", "host_agent", "snmp", "prometheus")
+_PLUGINS = ("traefik", "unifi", "docker", "host_agent", "snmp", "prometheus", "homeassistant")
 
 
 def catalog() -> dict[str, set[str]]:
diff --git a/tests/integration/test_homeassistant.py b/tests/integration/test_homeassistant.py
new file mode 100644
index 0000000..6d62b54
--- /dev/null
+++ b/tests/integration/test_homeassistant.py
@@ -0,0 +1,290 @@
+"""Integration: Steward reads Home Assistant, and installs one update through its
+webhook only after a newer backup, following it to its end. A fake Home Assistant
+answers over httpx's mock transport. Requires STEWARD_DATABASE_URL."""
+from __future__ import annotations
+
+import asyncio
+import importlib
+import json
+import os
+import uuid
+
+import httpx
+import pytest
+from sqlalchemy import delete, select
+
+pytestmark = pytest.mark.integration
+
+ADDON = "update.example_update"
+BEFORE, AFTER = "2026-10-30T03:00:00+00:00", "2026-10-31T03:00:00+00:00"
+
+
+@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)
+
+
+def _ha(sub):
+    from steward.core.plugin_manager import get_loaded_plugin_modules
+    loaded = get_loaded_plugin_modules().get("homeassistant")
+    base = loaded.__name__ if loaded is not None else "plugins.homeassistant"
+    return importlib.import_module(f"{base}.{sub}")
+
+
+class FakeHomeAssistant:
+    """Its entities, the webhook calls it received, and what it refuses."""
+
+    def __init__(self):
+        self.calls = []
+        self.tokens = set()
+        self.refuse = False
+        self.down = False
+        self.hook_runs = True
+        self.entities = {
+            ADDON: self._update("on", "1.0"),
+            "sensor.backup_last_successful_automatic_backup": {"state": BEFORE, "attributes": {}},
+            "event.backup_automatic_backup": {"state": BEFORE, "attributes": {"event_type": "completed"}},
+            "automation.steward_maintenance": {"state": "on", "attributes": {"id": "steward_maintenance",
+                                                                             "last_triggered": None}},
+        }
+
+    def _update(self, state, installed):
+        return {"state": state, "attributes": {"title": "Example", "installed_version": installed,
+                                               "latest_version": "1.1", "in_progress": False,
+                                               "entity_picture": "/api/hassio/addons/example/icon"}}
+
+    def backup_lands(self):
+        self.entities["sensor.backup_last_successful_automatic_backup"]["state"] = AFTER
+
+    def update_lands(self):
+        self.entities[ADDON] = self._update("off", "1.1")
+
+    def handler(self, request: httpx.Request) -> httpx.Response:
+        if self.down:
+            raise httpx.ConnectError("connection refused")
+        path = request.url.path
+        if path.startswith("/api/webhook/"):
+            assert "authorization" not in request.headers
+            self.calls.append(json.loads(request.content))
+            if self.hook_runs:
+                self.entities["automation.steward_maintenance"]["attributes"]["last_triggered"] = AFTER
+            return httpx.Response(200)
+        self.tokens.add(request.headers.get("authorization"))
+        if self.refuse:
+            return httpx.Response(401)
+        if path == "/api/states":
+            return httpx.Response(200, json=[{"entity_id": k, **v} for k, v in self.entities.items()])
+        entity = self.entities.get(path.removeprefix("/api/states/"))
+        return httpx.Response(200, json=entity) if entity else httpx.Response(404)
+
+    @property
+    def transport(self):
+        return httpx.MockTransport(self.handler)
+
+
+async def _clean(app, **connection):
+    from steward.core.crypto import encrypt_secret
+    m = _ha("models")
+    async with app.db_sessionmaker() as s:
+        async with s.begin():
+            await s.execute(delete(m.HomeAssistantUpdate))
+            await s.execute(delete(m.HomeAssistantConnection))
+            s.add(m.HomeAssistantConnection(url="http://ha.test:8123", token=encrypt_secret("read-only"),
+                                            webhook_id=encrypt_secret(f"hook{uuid.uuid4().hex}"), **connection))
+
+
+async def _state(app):
+    m = _ha("models")
+    async with app.db_sessionmaker() as s:
+        connection = (await s.execute(select(m.HomeAssistantConnection))).scalars().first()
+        runs = (await s.execute(select(m.HomeAssistantUpdate))).scalars().all()
+    return connection, runs
+
+
+def test_a_poll_stores_what_it_read_and_records_its_metrics(app):
+    from steward.models.metrics import PluginMetricLatest
+    sched = _ha("scheduler")
+    ha = FakeHomeAssistant()
+
+    async def go():
+        await _clean(app)
+        await sched.poll(app, transport=ha.transport)
+        connection, _ = await _state(app)
+        async with app.db_sessionmaker() as s:
+            rows = (await s.execute(select(PluginMetricLatest).where(
+                PluginMetricLatest.source_module == "homeassistant"))).scalars().all()
+        return connection, {r.metric_name: r.value for r in rows}
+
+    connection, metrics = asyncio.run(go())
+    assert ha.tokens == {"Bearer read-only"}
+    assert connection.automation == "on" and connection.last_error is None and not connection.refused
+    assert [u["entity_id"] for u in connection.updates] == [ADDON]
+    assert connection.backup["last_ok"] == BEFORE
+    assert metrics["is_up"] == 1.0 and metrics["updates_waiting"] == 1.0 and metrics["backup_age_hours"] >= 0
+
+
+def test_a_refused_token_is_asked_once_and_not_again(app):
+    sched = _ha("scheduler")
+    ha = FakeHomeAssistant()
+    ha.refuse = True
+    seen = []
+    real = ha.handler
+
+    def counted(request):
+        seen.append(request.url.path)
+        return real(request)
+    transport = httpx.MockTransport(counted)
+
+    async def go():
+        await _clean(app)
+        await sched.poll(app, transport=transport)
+        await sched.poll(app, transport=transport)
+        return (await _state(app))[0]
+
+    connection = asyncio.run(go())
+    assert seen == ["/api/states"]
+    assert connection.refused and "refused the token" in connection.last_error
+
+
+def test_an_update_waits_for_a_newer_backup_then_installs_and_ends_on_the_timeline(app):
+    from steward.models.timeline import TimelineEntry
+    sched, upd = _ha("scheduler"), _ha("updates")
+    ha = FakeHomeAssistant()
+
+    async def go():
+        await _clean(app)
+        await sched.poll(app, transport=ha.transport)
+        run = await upd.start_update(app, ADDON, "someone", transport=ha.transport)
+        assert ha.calls == [{"action": "backup"}]
+        assert run.state == "backing_up" and run.from_version == "1.0" and run.to_version == "1.1"
+
+        # No newer backup yet: nothing is installed, however often it is asked.
+        await upd.advance(app, transport=ha.transport)
+        await upd.advance(app, transport=ha.transport)
+        assert ha.calls == [{"action": "backup"}]
+
+        ha.backup_lands()
+        await upd.advance(app, transport=ha.transport)
+        _, [row] = await _state(app)
+        assert row.state == "installing" and row.backup_at is not None
+        await upd.advance(app, transport=ha.transport)
+        assert ha.calls == [{"action": "backup"}, {"action": "install", "entity_id": ADDON}]
+
+        # Home Assistant restarting is expected, and isn't the end of the update.
+        ha.down = True
+        await upd.advance(app, transport=ha.transport)
+        _, [row] = await _state(app)
+        assert row.state == "installing"
+        ha.down = False
+
+        ha.update_lands()
+        await upd.advance(app, transport=ha.transport)
+        connection, [row] = await _state(app)
+        assert row.state == "done" and row.finished_at is not None
+        assert connection.last_polled_at is None        # the page shows the new version at the next tick
+        await upd.advance(app, transport=ha.transport)
+        assert len(ha.calls) == 2
+        async with app.db_sessionmaker() as s:
+            entries = (await s.execute(select(TimelineEntry).where(
+                TimelineEntry.ref_type == "homeassistant_update", TimelineEntry.ref_id == row.id))).scalars().all()
+        return entries
+
+    [entry] = asyncio.run(go())
+    assert entry.type == "homeassistant.updated" and entry.actor == "someone"
+    assert "1.0 to 1.1" in entry.summary and entry.area == "service"
+
+
+def test_a_failed_backup_installs_nothing(app):
+    sched, upd = _ha("scheduler"), _ha("updates")
+    ha = FakeHomeAssistant()
+
+    async def go():
+        await _clean(app)
+        await sched.poll(app, transport=ha.transport)
+        await upd.start_update(app, ADDON, "someone", transport=ha.transport)
+        ha.entities["event.backup_automatic_backup"] = {
+            "state": AFTER, "attributes": {"event_type": "failed", "failed_reason": "upload_failed"}}
+        await upd.advance(app, transport=ha.transport)
+        return (await _state(app))[1]
+
+    [run] = asyncio.run(go())
+    assert run.state == "failed" and "upload_failed" in run.message and "Nothing was installed" in run.message
+    assert ha.calls == [{"action": "backup"}]
+
+
+def test_a_webhook_that_never_ran_the_automation_says_so(app):
+    from datetime import timedelta
+    from sqlalchemy import update
+    sched, upd, m = _ha("scheduler"), _ha("updates"), _ha("models")
+    ha = FakeHomeAssistant()
+    ha.hook_runs = False
+
+    async def go():
+        await _clean(app)
+        await sched.poll(app, transport=ha.transport)
+        run = await upd.start_update(app, ADDON, "someone", transport=ha.transport)
+        await upd.advance(app, transport=ha.transport)
+        assert (await _state(app))[1][0].state == "backing_up"      # within its grace
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await s.execute(update(m.HomeAssistantUpdate).where(m.HomeAssistantUpdate.id == run.id).values(
+                    started_at=run.started_at - timedelta(seconds=upd.TRIGGER_GRACE_S + 1)))
+        await upd.advance(app, transport=ha.transport)
+        return (await _state(app))[1]
+
+    [run] = asyncio.run(go())
+    assert run.state == "failed" and "webhook id" in run.message and "Nothing was installed" in run.message
+
+
+def test_an_install_past_its_deadline_fails_and_only_one_runs_at_a_time(app):
+    from datetime import timedelta
+    from sqlalchemy import update
+    sched, upd, m = _ha("scheduler"), _ha("updates"), _ha("models")
+    ha = FakeHomeAssistant()
+
+    async def go():
+        await _clean(app)
+        await sched.poll(app, transport=ha.transport)
+        run = await upd.start_update(app, ADDON, "someone", transport=ha.transport)
+        with pytest.raises(upd.CannotStart, match="under way"):
+            await upd.start_update(app, ADDON, "someone else", transport=ha.transport)
+        ha.backup_lands()
+        await upd.advance(app, transport=ha.transport)
+        await upd.advance(app, transport=ha.transport)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                row = await s.get(m.HomeAssistantUpdate, run.id)
+                row.install_sent_at -= timedelta(seconds=upd.INSTALL_DEADLINE_S["addon"] + 1)
+        await upd.advance(app, transport=ha.transport)
+        return (await _state(app))[1]
+
+    [run] = asyncio.run(go())
+    assert run.state == "failed" and "Still not on 1.1" in run.message
+    assert len(ha.calls) == 2
+
+
+@pytest.mark.parametrize("entity,change,says", [
+    ("update.unknown", None, "isn't a Core, OS or add-on"),
+    (ADDON, "no_longer_waiting", "no longer waiting"),
+    (ADDON, "no_automation", "automation isn't on"),
+])
+def test_nothing_starts_unless_the_update_can_run(app, entity, change, says):
+    sched, upd = _ha("scheduler"), _ha("updates")
+    ha = FakeHomeAssistant()
+
+    async def go():
+        await _clean(app)
+        if change == "no_automation":
+            del ha.entities["automation.steward_maintenance"]
+        await sched.poll(app, transport=ha.transport)
+        if change == "no_longer_waiting":
+            ha.update_lands()
+        with pytest.raises(upd.CannotStart, match=says):
+            await upd.start_update(app, entity, "someone", transport=ha.transport)
+        return (await _state(app))[1]
+
+    assert asyncio.run(go()) == []
+    assert ha.calls == []
diff --git a/tests/plugins/homeassistant/__init__.py b/tests/plugins/homeassistant/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/plugins/homeassistant/test_service.py b/tests/plugins/homeassistant/test_service.py
new file mode 100644
index 0000000..444c4cb
--- /dev/null
+++ b/tests/plugins/homeassistant/test_service.py
@@ -0,0 +1,189 @@
+"""Reading what Home Assistant reports: which updates Steward may install, how
+its backups stand, the automation to add, and what the pages and MCP show."""
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace as NS
+
+import yaml
+
+import plugins.homeassistant as homeassistant_plugin
+from plugins.homeassistant import mcp_tools, scheduler, service as s
+from steward.models.users import UserRole
+from tests._templates import render_plugin_template
+from tests.test_style_tokens import hex_colours
+
+NOW = datetime(2026, 10, 31, 12, 0, tzinfo=timezone.utc)
+
+
+def _entity(entity_id, state="on", **attrs):
+    return {"entity_id": entity_id, "state": state, "attributes": attrs}
+
+
+def _addon(slug="core_example", state="on"):
+    return _entity(f"update.{slug}_update", state, title="Example add-on", installed_version="1.0",
+                   latest_version="1.1", entity_picture=f"/api/hassio/addons/{slug}/icon")
+
+
+def test_only_core_os_and_add_ons_can_be_installed():
+    assert s.classify(_entity(s.CORE)) == "core"
+    assert s.classify(_entity(s.OS)) == "os"
+    assert s.classify(_addon()) == "addon"
+    assert s.classify(_entity(s.SUPERVISOR, entity_picture="/api/brands/integration/hassio/icon.png")) is None
+    assert s.classify(_entity("update.switch_firmware", device_class="firmware",
+                              entity_picture="/api/brands/integration/unifi/icon.png")) is None
+    assert s.classify(_entity("update.custom_locks_update",
+                              entity_picture="https://brands.home-assistant.io/_/locks/icon.png")) is None
+
+
+def test_a_snapshot_keeps_waiting_updates_first_and_finds_the_automation_by_its_id():
+    snapshot = s.read_snapshot([
+        _addon("core_quiet", "off"), _entity("update.switch_firmware", title="Switch"),
+        _addon(), _entity(s.CORE, title="Home Assistant Core"),
+        _entity("update.gone", "unavailable"),
+        _entity(s.LAST_BACKUP, "2026-10-31T03:00:00+00:00"), _entity(s.NEXT_BACKUP, "unknown"),
+        _entity(s.BACKUP_EVENT, "2026-10-31T03:00:00+00:00", event_type="completed"),
+        _entity("automation.renamed_by_someone", "on", id=s.AUTOMATION_ID, last_triggered=None),
+        _entity("light.kitchen", "on"),
+    ])
+    assert [u["entity_id"] for u in snapshot["updates"]] == [
+        s.CORE, "update.core_example_update", "update.switch_firmware", "update.core_quiet_update"]
+    assert snapshot["automation"] == "on"
+    backup = snapshot["backup"]
+    assert backup["last_ok"] == "2026-10-31T03:00:00+00:00" and backup["next"] is None
+    assert backup["last_result"] == "completed"
+    assert s.automation_entity_id(NS(backup=backup)) == "automation.renamed_by_someone"
+    assert s.read_snapshot([])["automation"] is None
+
+
+def test_a_release_link_is_only_kept_when_it_is_a_web_address():
+    assert s.read_snapshot([_entity(s.CORE, release_url="javascript:alert(1)")])["updates"][0]["release_url"] is None
+    assert s.read_snapshot([_entity(s.CORE, release_url="https://example.com/r")])["updates"][0]["release_url"]
+
+
+def test_newer_is_judged_on_the_readings_alone():
+    assert s.is_newer("2026-10-31T03:00:01+00:00", "2026-10-31T03:00:00+00:00")
+    assert not s.is_newer("2026-10-31T03:00:00+00:00", "2026-10-31T03:00:00+00:00")
+    assert not s.is_newer("2026-10-30T03:00:00+00:00", "2026-10-31T03:00:00+00:00")
+    assert s.is_newer("2026-10-31T03:00:00+00:00", "unknown")
+    assert not s.is_newer("unavailable", "2026-10-31T03:00:00+00:00")
+
+
+def test_how_the_backups_read():
+    ok = {"known": True, "last_ok": (NOW - timedelta(hours=25)).isoformat(), "last_result": "completed"}
+    assert s.backup_verdict(ok, NOW)[0] == "ok"
+    assert abs(s.backup_age_hours(ok, NOW) - 25) < 0.01
+    stale = {**ok, "last_ok": (NOW - timedelta(hours=27)).isoformat()}
+    assert s.backup_verdict(stale, NOW) == ("warn", "The last good backup is 27 hours old, past the 26 hours expected.")
+    failed = {**ok, "last_result": "failed", "failed_reason": "upload_failed"}
+    assert "failed (upload_failed)" in s.backup_verdict(failed, NOW)[1]
+    assert "no successful" in s.backup_verdict({"known": True, "last_ok": None}, NOW)[1]
+    assert "reports no automatic backups" in s.backup_verdict({}, NOW)[1]
+
+
+def test_the_automation_to_add_only_takes_a_backup_or_installs_what_it_allows():
+    text = s.automation_yaml("steward_abc")
+    config = yaml.safe_load(text)
+    assert config["id"] == s.AUTOMATION_ID and config["mode"] == "single"
+    [trigger] = config["triggers"]
+    assert trigger == {"trigger": "webhook", "webhook_id": "steward_abc", "allowed_methods": ["POST"], "local_only": True}
+    assert "integration_entities('hassio')" in config["variables"]["allowed"]
+    assert "'update\\\\.'" in config["variables"]["allowed"]
+    assert s.SUPERVISOR in config["variables"]["allowed"]
+    branches = config["actions"][0]["choose"]
+    assert [b["sequence"][0]["action"] for b in branches] == ["backup.create_automatic", "update.install"]
+    assert "in allowed" in branches[1]["conditions"][0]["value_template"]
+
+
+def test_a_pasted_secret_loses_a_terminals_wrapping():
+    assert s.no_spaces("  eyJabc\n  def ghi\n") == "eyJabcdefghi"
+    assert s.no_spaces(None) == ""
+
+
+def test_a_refused_token_stops_polling_until_someone_acts():
+    fresh = NS(refused=False, last_polled_at=NOW - timedelta(seconds=60))
+    assert not scheduler.is_due(fresh, NOW)
+    assert scheduler.is_due(NS(refused=False, last_polled_at=NOW - timedelta(seconds=300)), NOW)
+    assert scheduler.is_due(NS(refused=False, last_polled_at=None), NOW)
+    assert not scheduler.is_due(NS(refused=True, last_polled_at=None), NOW)
+
+
+def _connection(**over):
+    return NS(**{"url": "http://ha.example:8123", "token": "tok-s3cret", "webhook_id": "hook-s3cret",
+                 "refused": False, "last_error": None, "last_polled_at": NOW, "last_ok_at": NOW,
+                 "automation": "on", "backup": {"known": True, "last_ok": NOW.isoformat(), "last_result": "completed"},
+                 "updates": [], **over})
+
+
+def test_the_overview_names_what_needs_a_look_and_no_more():
+    assert mcp_tools.overview_findings(None, None, NOW) == []
+    assert mcp_tools.overview_findings(_connection(), None, NOW) == []
+    waiting = _connection(updates=[{"waiting": True, "kind": "core", "title": "Core"}])
+    assert mcp_tools.overview_findings(waiting, None, NOW) == []
+    down = mcp_tools.overview_findings(_connection(last_error="Couldn't reach Home Assistant."), None, NOW)
+    assert [f["kind"] for f in down] == ["home_assistant_down"]
+    stale = _connection(backup={"known": True, "last_ok": (NOW - timedelta(days=2)).isoformat()})
+    assert [f["kind"] for f in mcp_tools.overview_findings(stale, None, NOW)] == ["home_assistant_backup"]
+    failed = NS(state="failed", title="Core", message="No new backup.", finished_at=NOW)
+    [f] = mcp_tools.overview_findings(_connection(), failed, NOW)
+    assert f["kind"] == "home_assistant_update_failed" and "No new backup." in f["summary"]
+
+
+def test_reading_is_open_to_a_viewer_and_installing_is_for_an_admin():
+    seen = {}
+    homeassistant_plugin.register_mcp_tools(lambda fn, role, read_only=False: seen.update({fn.__name__: (role, read_only)}))
+    assert seen == {"get_home_assistant": (UserRole.viewer, True), "update_home_assistant": (UserRole.admin, False)}
+
+
+def _page_values(**over):
+    update = {"entity_id": s.CORE, "title": "Core", "kind": "core", "kind_word": "Core", "installed": "1.0",
+              "latest": "1.1", "waiting": True, "in_progress": False, "auto_update": False,
+              "release_url": "https://example.com/r"}
+    return {"connection": _connection(), "reach": "up", "says": "Answering.", "installable": [update], "others": [],
+            "current": [], "runs": [], "under_way": None, "backup": {}, "backup_tone": "ok",
+            "backup_says": "Backups are up to date.", "backup_at": NOW, "backup_next": None, "backup_age": 1.0,
+            "stale_hours": 26, **over}
+
+
+async def _render(app, role, **values):
+    from quart import session
+    async with app.test_request_context("/plugins/homeassistant/"):
+        session["user_role"] = role
+        page = await render_plugin_template(app, "homeassistant", "homeassistant/index.html", **values)
+    return " ".join(page.split())
+
+
+async def test_an_admin_is_asked_before_an_install_and_a_viewer_cant_install(app):
+    page = await _render(app, "admin", **_page_values())
+    assert "1 update waiting" in page and 'action="/plugins/homeassistant/install"' in page
+    assert "Steward takes a backup first" in page and "Home Assistant restarts." in page
+    assert "tok-s3cret" not in page and "hook-s3cret" not in page
+    main = page[page.index("<main"):]
+    assert not hex_colours(main), hex_colours(main)
+    assert 'action="/plugins/homeassistant/install"' not in await _render(app, "viewer", **_page_values())
+
+
+async def test_install_is_held_back_while_it_couldnt_work_and_the_page_says_why(app):
+    run = NS(id="r", title="Core", state="backing_up", from_version="1.0", to_version="1.1", started_by="someone",
+             started_at=NOW, message=None)
+    busy = await _render(app, "admin", **_page_values(under_way=run, runs=[run]))
+    assert 'action="/plugins/homeassistant/install"' not in busy and "taking a backup first" in busy
+    assert 'hx-trigger="every 15s"' in busy
+    gone = await _render(app, "admin", **_page_values(connection=_connection(automation=None)))
+    assert 'action="/plugins/homeassistant/install"' not in gone
+    assert "maintenance automation isn't in Home Assistant" in gone
+
+
+async def test_an_unconnected_page_offers_to_connect(app):
+    page = await _render(app, "admin", connection=None)
+    assert 'href="/plugins/homeassistant/connection"' in page
+
+
+async def test_the_form_shows_the_automation_but_never_a_saved_secret_in_a_field(app):
+    from quart import session
+    async with app.test_request_context("/plugins/homeassistant/connection"):
+        session["user_role"] = "admin"
+        page = await render_plugin_template(
+            app, "homeassistant", "homeassistant/form.html", connection=_connection(),
+            v={"url": "http://ha.example:8123", "verify_tls": True, "has_token": True, "has_webhook": True},
+            error=None, automation=s.automation_yaml("generated-id"))
+    assert "generated-id" in page and "tok-s3cret" not in page
+    assert "kept; type to change" in page