feat(host_agent): Steward hands a host a maintenance job and follows it through the reboot (#5767)

#5767 "Steward hands a host a maintenance job and follows it to the end,
through its own restarts", milestone 514.

A host_maintenance_jobs row per job: queued → updating → [draining →]
rebooting → [undraining →] done, or failed / timed_out. Steward's answer to
an ingest post carries the step that host's agent should carry out now; the
agent (1.18.0) leaves each step id for the root helper once, and the helper
refuses an id it already ran, so repeats are harmless. Results come back as
the #5766 maintenance events, matched by step id; the reboot completes when
the host's `started` event has a boot_at after the reboot was handed out.

The job reboots only when the update says one is needed (or never, if asked).
A swarm node is drained first by itself if it's a manager or by a manager
with the helper, and made active after; with no manager to drain it the job
is refused. A minute tick times out a job past its state's deadline. A
maintenance window over host_agent/<host> lasts the rest of the job.

Everything is on the row, so a Steward restart, or the reboot of the host
Steward runs on, resumes from the next post. Ingest runs it in its own
savepoint so a job going wrong never costs the samples.

The Agent panel leads with a running job and says what a finished one did;
Manage agent has Update now (with "never reboot"), or says why a job can't
start. Integration test: hand off, Steward restarts, the post-boot events
close the job.

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

docs/plugins/host-agent-design.md

@@ -174,7 +174,52 @@ host that has an older agent.
 
 So a stolen agent token can make its host update, reboot or drain, and nothing
 else. Steward handing a host a job and following it through the reboot is
-#5767; the cadence that does it one host at a time is #5768.
+#5767 (below); the cadence that does it one host at a time is #5768.
+
+### A maintenance job, followed through the reboot (1.18.0, #5767)
+
+Steward keeps one `host_maintenance_jobs` row per job and moves it on only from
+what the agents post, so a restart of Steward, or a reboot of the host Steward
+runs on, loses nothing. `plugins/host_agent/maintenance.py` has the states and
+transitions (model-free, unit-tested) and the database side.
+
+```
+queued → updating → [draining →] rebooting → [undraining →] done
+                  ↘ done (no reboot needed, or the job may never reboot)
+any state → failed (a step's result said so) | timed_out (its deadline passed)
+```
+
+- **Hand-off.** Steward's answer to an ingest post carries
+  `"maintenance": [{"id", "action", ["node"]}]`, the step that host's agent
+  should carry out now. It is repeated on every post until a result for that
+  step id arrives; the agent leaves each id for the helper once
+  (`take_maintenance`), and the helper refuses an id it already has a result
+  for, so a repeat, or a re-hand after the agent restarts, never runs twice.
+  A step id is `<job id>-<action>`.
+- **Results** are the `maintenance` events of #5766, matched to the job by step
+  id, so a drain reported by a manager reaches the job of the node it drained.
+- **The reboot completes** when the host's agent sends `started` with a
+  `boot_at` after the reboot was handed out (60 s of clock slack). An agent
+  restart on a host that stayed up has an old `boot_at` and changes nothing.
+- **Reboot only when required**: the update's result says whether a reboot is
+  needed; a job made with "never reboot" stops after the update.
+- **Swarm nodes** are drained first and made active after. The node is found
+  by hostname in the Docker plugin's swarm nodes; a manager drains itself, a
+  worker is drained by a manager whose agent has the helper. With no such
+  manager the job is refused rather than rebooting an undrained node.
+- **Deadlines** (`DEADLINES`): queued 1 h, updating 90 min, draining 10 min,
+  rebooting 30 min, undraining 10 min. A minute tick (`host_agent_maintenance`)
+  times out a job past its deadline.
+- **A maintenance window** scoped to the host's agent metrics
+  (`host_agent/<host>`) opens when the job starts, lasts as long as the rest of
+  the job could take, and ends when it does.
+- Ingest applies all of this in a savepoint of its own: a job going wrong is
+  logged and never costs the host its samples.
+
+On the host's Agent panel: a running job leads what the host has waiting, with
+since-when; a finished one says what it did ("Done: 12 packages updated,
+rebooted"). Under Manage agent, **Update now** (with "never reboot") starts a
+job, or the panel says why one can't start.
 
 ### What happened to the host (1.14.0, #5756)
 

plugins/host_agent/__init__.py

@@ -37,8 +37,8 @@ def setup(app: "Quart") -> None:
 
 
 def get_scheduled_tasks() -> list:
-    from .scheduler import make_task
-    return [make_task(_app)]
+    from .scheduler import make_maintenance_task, make_task
+    return [make_task(_app), make_maintenance_task(_app)]
 
 
 def get_blueprint():

plugins/host_agent/agent.py

@@ -22,7 +22,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.17.0"
+AGENT_VERSION = "1.18.0"
 
 # Default path to the local Docker Engine socket. Overridable via the
 # `docker_socket` config key; collection is silently skipped if it's absent or
@@ -1484,6 +1484,27 @@ def request_maintenance(job_id: str, action: str, node: str | None = None,
     os.replace(tmp, os.path.join(requests_dir, f"{job_id}.json"))
 
 
+def take_maintenance(steps, requested: set, requests_dir: str = MAINT_REQUESTS) -> list:
+    """Leave the helper each step Steward handed over that this agent hasn't
+    already left it; returns the ids left. Steward hands a step on every post
+    until its result arrives, so a step is taken once here, and the helper
+    also refuses an id it has a result for (after a reboot this set is empty)."""
+    taken = []
+    for step in steps if isinstance(steps, list) else []:
+        if not isinstance(step, dict) or step.get("id") in requested:
+            continue
+        try:
+            request_maintenance(str(step.get("id") or ""), str(step.get("action") or ""),
+                                step.get("node"), requests_dir=requests_dir)
+        except (ValueError, OSError) as exc:
+            _log("WARN", f"maintenance step {step.get('id')!r} not taken: {exc}")
+            continue
+        requested.add(step["id"])
+        taken.append(step["id"])
+        _log("INFO", f"maintenance: {step['action']} requested ({step['id']})")
+    return taken
+
+
 def maintenance_results(seen: set, results_dir: str = MAINT_RESULTS,
                         now: float | None = None) -> tuple:
     """(events, keys): a `maintenance` event for each result whose (id, state)
@@ -1774,7 +1795,15 @@ def next_backoff(current: int) -> int:
     return min(current * 2, BACKOFF_CAP)
 
 
-def post_payload(url: str, token: str, payload: dict) -> tuple[bool, int | None]:
+# Steward's answer to a post is small ({"ok", counts, maybe a maintenance
+# step}); anything past this is not read.
+ANSWER_MAX_BYTES = 65536
+
+
+def post_payload(url: str, token: str, payload: dict,
+                 answer: dict | None = None) -> tuple[bool, int | None]:
+    """POST the payload. When `answer` is given, Steward's JSON reply is put
+    in it (#5767: it carries any maintenance step for this host)."""
     body = json.dumps(payload).encode("utf-8")
     req = urllib.request.Request(
         url.rstrip("/") + "/plugins/host_agent/ingest",
@@ -1788,6 +1817,13 @@ def post_payload(url: str, token: str, payload: dict) -> tuple[bool, int | None]
     )
     try:
         with urllib.request.urlopen(req, timeout=10) as resp:
+            if answer is not None:
+                try:
+                    got = json.loads(resp.read(ANSWER_MAX_BYTES).decode("utf-8"))
+                    if isinstance(got, dict):
+                        answer.update(got)
+                except (ValueError, UnicodeDecodeError, OSError):
+                    pass                # the post landed; only the reply is lost
             return (200 <= resp.status < 300, resp.status)
     except urllib.error.HTTPError as e:
         return (False, e.code)
@@ -1842,6 +1878,7 @@ def main_loop(conf_path: str) -> int:
     # Carries previous net/disk counters + monotonic ts for rate computation.
     rate_state: dict = {}
     maint_seen: set = set()
+    maint_requested: set = set()
 
     _log("INFO", f"steward-host-agent {AGENT_VERSION} starting "
                  f"(url={cfg['url']}, interval={cfg['interval_seconds']}s)")
@@ -1873,9 +1910,11 @@ def main_loop(conf_path: str) -> int:
             events=events + ([lost_contact_event(lost_since)] if lost_since else [])
             + maint_events,
         )
-        ok, status = post_payload(cfg["url"], cfg["token"], payload)
+        answer: dict = {}
+        ok, status = post_payload(cfg["url"], cfg["token"], payload, answer)
 
         if ok:
+            take_maintenance(answer.get("maintenance"), maint_requested)
             if lost_since:
                 _log("INFO", f"reached Steward again (unreachable since {lost_since})")
             events, lost_since = [], None

plugins/host_agent/maintenance.py

@@ -0,0 +1,384 @@
+# plugins/host_agent/maintenance.py
+"""A host's maintenance job, followed to its end through any restart (#5767).
+
+Steward plans and the host's agent acts: a job is one host's update and, only
+when the update says one is needed, its reboot, with the host's swarm node
+drained first and made active again after. Each action is one step the agent
+hands to its root helper (maint.py, #5766). Everything lives on the job's row,
+so Steward restarting, or the host Steward runs on rebooting, loses nothing:
+the next post from the agent carries the step's result, and the `started`
+event a host sends after booting is what completes a reboot.
+
+The transitions here are model-free (a job is anything with the attributes),
+so they're unit-tested without a database; routes.ingest and the scheduler's
+tick apply them.
+
+States: queued → updating → [draining →] rebooting → [undraining →] done,
+or failed / timed_out. A job with no reboot needed goes updating → done.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+ACTIVE = ("queued", "updating", "draining", "rebooting", "undraining")
+FINISHED = ("done", "failed", "timed_out")
+# The helper action each handed state asks for.
+STEP_ACTION = {"updating": "update", "draining": "drain", "rebooting": "reboot",
+               "undraining": "undrain"}
+# How long a state may last before the job is given up. Queued counts from the
+# job's creation (the host may be silent); the rest from their hand-off.
+DEADLINES = {
+    "queued": timedelta(hours=1),
+    "updating": timedelta(minutes=90),     # the helper itself stops at 60
+    "draining": timedelta(minutes=10),
+    "rebooting": timedelta(minutes=30),
+    "undraining": timedelta(minutes=10),
+}
+# A host's clock and Steward's differ by seconds; a boot this far before the
+# reboot was handed out is still taken as that reboot.
+BOOT_SLACK = timedelta(seconds=60)
+DETAIL_MAX = 500
+
+
+def step_id(job, state: str) -> str:
+    """The helper request id for a job's step: unique per job and step, and in
+    the helper's id alphabet (letters, digits, dashes)."""
+    return f"{job.id}-{STEP_ACTION[state]}"
+
+
+def _enter(job, state: str, now: datetime) -> None:
+    job.state, job.state_at = state, now
+    job.step_id = step_id(job, state) if state in STEP_ACTION else None
+    job.step_seen = False               # a new step is handed out until its result arrives
+    if state in FINISHED:
+        job.finished_at = now
+
+
+def _finish(job, state: str, now: datetime, detail: str | None = None) -> None:
+    if detail:
+        job.detail = detail[:DETAIL_MAX]
+    _enter(job, state, now)
+
+
+def start(job, now: datetime) -> None:
+    """The host's agent is about to be handed the job: update first."""
+    if job.state == "queued":
+        _enter(job, "updating", now)
+
+
+def executor(job) -> str:
+    """The host whose agent carries out the current step: the job's own host,
+    except a drain or undrain, which a swarm manager does."""
+    if job.state in ("draining", "undraining") and job.drain_host_id:
+        return job.drain_host_id
+    return job.host_id
+
+
+def handoff(job) -> dict | None:
+    """What to hand the executing agent now, or None. Handed on every post
+    until a result for the step arrives; the agent and the helper both ignore
+    an id they've already taken, so a repeat is harmless."""
+    if job.state not in STEP_ACTION or job.step_seen:
+        return None
+    out = {"id": job.step_id, "action": STEP_ACTION[job.state]}
+    if job.state in ("draining", "undraining") and job.node_id:
+        out["node"] = job.node_id
+    return out
+
+
+def _after_update(job, now: datetime) -> None:
+    if not (job.reboot_needed and job.reboot_policy != "never"):
+        _enter(job, "done", now)
+    elif job.node_id and job.drain_host_id:
+        _enter(job, "draining", now)
+    else:
+        _enter(job, "rebooting", now)
+
+
+def apply_result(job, event: dict, now: datetime) -> bool:
+    """Apply a `maintenance` event for this job's current step; True when the
+    job changed. A result for a step the job has moved past is ignored, so a
+    repeated or late report does nothing."""
+    if job.state not in STEP_ACTION or event.get("id") != job.step_id:
+        return False
+    state = event.get("state")
+    if state in ("running", "rebooting"):
+        if job.step_seen:
+            return False
+        job.step_seen = True
+        return True
+    if state not in ("done", "failed"):
+        return False
+    if state == "failed" or event.get("ok") is False:
+        what = STEP_ACTION[job.state]
+        detail = f"{what} failed: {event.get('detail') or 'no detail'}"
+        if job.state == "undraining":
+            detail += " (the node is still drained)"
+        _finish(job, "failed", now, detail)
+        return True
+    if job.state == "updating":
+        if isinstance(event.get("packages"), int):
+            job.packages = event["packages"]
+        job.reboot_needed = event.get("reboot_required") is True
+        _after_update(job, now)
+    elif job.state == "draining":
+        _enter(job, "rebooting", now)
+    elif job.state == "undraining":
+        _enter(job, "done", now)
+    else:                               # rebooting: the boot completes it, not this
+        return False
+    return True
+
+
+def apply_boot(job, boot_at: datetime | None, now: datetime) -> bool:
+    """A `started` event from the job's host. A boot after the reboot was
+    handed out is that reboot: the host is back."""
+    if job.state != "rebooting" or boot_at is None or job.state_at is None:
+        return False
+    if boot_at < job.state_at - BOOT_SLACK:
+        return False                    # the agent restarted; the host didn't
+    job.back_at = now
+    if job.node_id and job.drain_host_id:
+        _enter(job, "undraining", now)
+    else:
+        _enter(job, "done", now)
+    return True
+
+
+def check_deadline(job, now: datetime) -> bool:
+    """Give up on a job whose state has lasted past its deadline."""
+    limit = DEADLINES.get(job.state)
+    since = job.state_at or job.created_at
+    if limit is None or since is None or now - since <= limit:
+        return False
+    waiting = {"queued": "the host's agent didn't check in",
+               "updating": "the update didn't finish",
+               "draining": "the node wasn't drained",
+               "rebooting": "the host didn't come back",
+               "undraining": "the node wasn't made active again"}[job.state]
+    _finish(job, "timed_out", now, f"{waiting} within {int(limit.total_seconds() // 60)} min")
+    return True
+
+
+def window_end(job, now: datetime) -> datetime:
+    """When the job's maintenance window should end: the longest the rest of
+    the job could take, or now once it has finished."""
+    if job.state in FINISHED:
+        return now
+    left = sum((DEADLINES[s] for s in ACTIVE[ACTIVE.index(job.state):]), timedelta())
+    return now + left
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+# ── applied to the database: ingest, the tick, and starting a job ────────────
+
+async def _window(session, job, host_name: str, now: datetime) -> None:
+    """Keep the job's maintenance window over the host until the job ends: one
+    scoped to this host's agent metrics, so an alert the job causes waits."""
+    from steward.models.alerts import MaintenanceWindow
+    if job.window_id:
+        w = await session.get(MaintenanceWindow, job.window_id)
+        if w is not None:
+            w.end_at = max(w.start_at, window_end(job, now))
+        return
+    if job.created_by is None or job.state in FINISHED:
+        return                          # a window needs an owner; none for a finished job
+    w = MaintenanceWindow(name=f"Maintenance: {host_name}"[:128], start_at=now,
+                          end_at=window_end(job, now), scope_source="host_agent",
+                          scope_resource=host_name, created_by=job.created_by)
+    session.add(w)
+    await session.flush()
+    job.window_id = w.id
+
+
+async def _host_name(session, host_id: str) -> str:
+    from steward.models.hosts import Host
+    host = await session.get(Host, host_id)
+    return host.name if host is not None else host_id
+
+
+async def apply_events(session, host_id: str, events, now: datetime) -> int:
+    """Apply one post's `maintenance` results and `started` boots to the jobs
+    they belong to; returns how many jobs changed. A result is matched by its
+    step id, so a drain reported by a manager reaches the job of the node it
+    drained."""
+    from sqlalchemy import select
+    from .downtime import _at
+    from .models import HostMaintenanceJob
+    if not isinstance(events, list):
+        return 0
+    changed = set()
+    for event in events:
+        if not isinstance(event, dict):
+            continue
+        if event.get("kind") == "maintenance" and isinstance(event.get("id"), str):
+            job = (await session.execute(select(HostMaintenanceJob).where(
+                HostMaintenanceJob.step_id == event["id"],
+                HostMaintenanceJob.state.in_(ACTIVE)))).scalar_one_or_none()
+            if job is not None and apply_result(job, event, now):
+                changed.add(job)
+        elif event.get("kind") == "started":
+            boot_at = _at(event.get("boot_at"))
+            for job in (await session.execute(select(HostMaintenanceJob).where(
+                    HostMaintenanceJob.host_id == host_id,
+                    HostMaintenanceJob.state == "rebooting"))).scalars():
+                if apply_boot(job, boot_at, now):
+                    changed.add(job)
+    for job in changed:
+        await _window(session, job, await _host_name(session, job.host_id), now)
+    return len(changed)
+
+
+async def handoffs(session, host_id: str, host_name: str, now: datetime) -> list[dict]:
+    """The steps this host's agent should carry out now. A queued job for the
+    host starts here, when its agent is known to be listening."""
+    from sqlalchemy import or_, select
+    from .models import HostMaintenanceJob
+    jobs = (await session.execute(select(HostMaintenanceJob).where(
+        HostMaintenanceJob.state.in_(ACTIVE),
+        or_(HostMaintenanceJob.host_id == host_id,
+            HostMaintenanceJob.drain_host_id == host_id)
+    ).order_by(HostMaintenanceJob.created_at))).scalars().all()
+    out = []
+    for job in jobs:
+        if job.state == "queued" and job.host_id == host_id:
+            start(job, now)
+            await _window(session, job, host_name, now)
+        if executor(job) == host_id:
+            step = handoff(job)
+            if step is not None:
+                out.append(step)
+    return out
+
+
+async def tick(session, now: datetime) -> int:
+    """Give up on jobs past their deadline; returns how many."""
+    from sqlalchemy import select
+    from .models import HostMaintenanceJob
+    gave_up = 0
+    for job in (await session.execute(select(HostMaintenanceJob).where(
+            HostMaintenanceJob.state.in_(ACTIVE)))).scalars():
+        if check_deadline(job, now):
+            gave_up += 1
+            await _window(session, job, await _host_name(session, job.host_id), now)
+    return gave_up
+
+
+class CannotStart(Exception):
+    """Why a host can't take a maintenance job now, in words for the page."""
+
+
+def pick_drainer(node_hostname_to_id: dict, managers: dict, host_name: str,
+                 host_id: str, helpers: set) -> tuple:
+    """(node_id, drain_host_id) for a host in a swarm, (None, None) for one
+    that isn't. `managers` maps a manager's host id to its swarm node id;
+    `helpers` is the hosts whose agent has the maintenance helper. The host
+    drains itself when it is a manager, else another manager does."""
+    node_id = node_hostname_to_id.get(host_name.lower())
+    if node_id is None:
+        return None, None
+    if host_id in managers and host_id in helpers:
+        return node_id, host_id
+    for mgr in sorted(managers):
+        if mgr != host_id and mgr in helpers:
+            return node_id, mgr
+    raise CannotStart("it is a swarm node and no swarm manager has the maintenance "
+                      "helper to drain it: run Update agent on a manager")
+
+
+async def plan_job(session, host, reg, reboot_policy: str = "if_required") -> tuple:
+    """(node_id, drain_host_id) for a job on `host`, or CannotStart saying why
+    it can't have one now. The panel asks this before offering the button."""
+    from sqlalchemy import select
+    from .models import HostAgentRegistration, HostMaintenanceJob
+    if reg is None or not reg.maint_helper:
+        raise CannotStart("its agent has no maintenance helper: run Update agent first")
+    if reboot_policy not in ("if_required", "never"):
+        raise CannotStart(f"unknown reboot policy {reboot_policy!r}")
+    busy = (await session.execute(select(HostMaintenanceJob.id).where(
+        HostMaintenanceJob.host_id == host.id,
+        HostMaintenanceJob.state.in_(ACTIVE)))).first()
+    if busy is not None:
+        raise CannotStart("a maintenance job is already running on it")
+    if reboot_policy == "never":
+        return None, None
+    nodes, managers = await _swarm(session)
+    helpers = set((await session.execute(select(HostAgentRegistration.host_id).where(
+        HostAgentRegistration.maint_helper.is_(True)))).scalars())
+    return pick_drainer(nodes, managers, host.name, host.id, helpers)
+
+
+async def create_job(session, host, reg, user_id: str | None,
+                     reboot_policy: str = "if_required", now: datetime | None = None):
+    """Queue a maintenance job for `host`; it starts with the agent's next post."""
+    from .models import HostMaintenanceJob
+    now = now or utcnow()
+    node_id, drain_host_id = await plan_job(session, host, reg, reboot_policy)
+    job = HostMaintenanceJob(host_id=host.id, state="queued", state_at=now,
+                             reboot_policy=reboot_policy, node_id=node_id,
+                             drain_host_id=drain_host_id, created_by=user_id, created_at=now)
+    session.add(job)
+    await session.flush()
+    return job
+
+
+async def latest_job(session, host_id: str, now: datetime, days: int = 7):
+    """The host's running job, else its last one in `days`, for the panel."""
+    from sqlalchemy import select
+    from .models import HostMaintenanceJob
+    return (await session.execute(select(HostMaintenanceJob).where(
+        HostMaintenanceJob.host_id == host_id,
+        HostMaintenanceJob.created_at >= now - timedelta(days=days),
+    ).order_by(HostMaintenanceJob.state.in_(ACTIVE).desc(),
+               HostMaintenanceJob.created_at.desc()).limit(1))).scalar_one_or_none()
+
+
+LABELS = {
+    "queued": "Waiting for the agent",
+    "updating": "Updating",
+    "draining": "Draining its swarm node",
+    "rebooting": "Rebooting",
+    "undraining": "Making its swarm node active again",
+    "done": "Done",
+    "failed": "Failed",
+    "timed_out": "Timed out",
+}
+
+
+def summary(job) -> str:
+    """One line on what the job did, for the panel."""
+    if job.state not in FINISHED:
+        return LABELS[job.state]
+    parts = []
+    if job.packages is not None:
+        parts.append(f"{job.packages} package{'s' if job.packages != 1 else ''} updated")
+    if job.back_at is not None:
+        parts.append("rebooted")
+    elif job.reboot_needed and job.reboot_policy == "never":
+        parts.append("reboot needed, not allowed by this job")
+    elif job.state == "done" and job.reboot_needed is False:
+        parts.append("no reboot needed")
+    head = LABELS[job.state]
+    return f"{head}: {', '.join(parts)}" if parts else head
+
+
+async def _swarm(session) -> tuple:
+    """({node hostname: node id}, {manager host id: its node id}) from what the
+    Docker plugin last stored; empty when it isn't installed. A manager is a
+    host whose agent reported the swarm's nodes."""
+    try:
+        from plugins.docker.models import DockerSwarmNode
+    except ImportError:
+        return {}, {}
+    from sqlalchemy import select
+    from steward.models.hosts import Host
+    rows = (await session.execute(select(DockerSwarmNode))).scalars().all()
+    nodes = {(n.hostname or "").lower(): n.node_id for n in rows if n.hostname}
+    reporters = {n.host_id for n in rows}
+    names = dict((await session.execute(select(Host.id, Host.name).where(
+        Host.id.in_(reporters)))).all()) if reporters else {}
+    managers = {hid: nodes.get(name.lower()) for hid, name in names.items()}
+    return nodes, managers

plugins/host_agent/migrations/versions/host_agent_007_maint_jobs.py

@@ -0,0 +1,46 @@
+"""host_agent: a host's maintenance job, followed through its reboot (#5767)
+
+Revision ID: host_agent_007_maint_jobs
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "host_agent_007_maint_jobs"
+down_revision: Union[str, None] = "host_agent_006_maint_helper"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "host_maintenance_jobs",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="CASCADE"),
+                  nullable=False),
+        sa.Column("state", sa.String(16), nullable=False),
+        sa.Column("state_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("reboot_policy", sa.String(16), nullable=False),
+        sa.Column("step_id", sa.String(64), nullable=True),
+        sa.Column("step_seen", sa.Boolean(), nullable=False, server_default=sa.false()),
+        sa.Column("node_id", sa.String(64), nullable=True),
+        sa.Column("drain_host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="SET NULL"),
+                  nullable=True),
+        sa.Column("packages", sa.Integer(), nullable=True),
+        sa.Column("reboot_needed", sa.Boolean(), nullable=True),
+        sa.Column("back_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("detail", sa.Text(), nullable=True),
+        sa.Column("window_id", sa.String(36),
+                  sa.ForeignKey("maintenance_windows.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("created_by", sa.String(36), sa.ForeignKey("users.id", ondelete="SET NULL"),
+                  nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_host_maintenance_jobs_host_id", "host_maintenance_jobs", ["host_id"])
+    op.create_index("ix_host_maintenance_jobs_state", "host_maintenance_jobs", ["state"])
+    op.create_index("ix_host_maintenance_jobs_step_id", "host_maintenance_jobs", ["step_id"])
+
+
+def downgrade() -> None:
+    op.drop_table("host_maintenance_jobs")

plugins/host_agent/models.py

@@ -73,3 +73,36 @@ class HostDowntime(Base):
     started_at = Column(DateTime(timezone=True), nullable=False)
     ended_at = Column(DateTime(timezone=True), nullable=True)
     created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow)
+
+
+class HostMaintenanceJob(Base):
+    """One host's update, and its reboot when the update needs one (#5767).
+
+    Steward keeps the job here and the host's agent carries out each step
+    through its root helper; maintenance.py has the states and transitions.
+    step_id is the helper request id of the current step, and step_seen says a
+    result for it has arrived, so it stops being handed out. node_id and
+    drain_host_id are set when the host is a swarm node: the node to drain and
+    the manager whose agent drains it.
+    """
+    __tablename__ = "host_maintenance_jobs"
+
+    id = Column(String(36), primary_key=True, default=_uuid)
+    host_id = Column(String(36), ForeignKey("hosts.id", ondelete="CASCADE"),
+                     nullable=False, index=True)
+    state = Column(String(16), nullable=False, default="queued", index=True)
+    state_at = Column(DateTime(timezone=True), nullable=True)
+    reboot_policy = Column(String(16), nullable=False, default="if_required")
+    step_id = Column(String(64), nullable=True, index=True)
+    step_seen = Column(Boolean, nullable=False, default=False)
+    node_id = Column(String(64), nullable=True)
+    drain_host_id = Column(String(36), ForeignKey("hosts.id", ondelete="SET NULL"), nullable=True)
+    packages = Column(Integer, nullable=True)
+    reboot_needed = Column(Boolean, nullable=True)
+    back_at = Column(DateTime(timezone=True), nullable=True)
+    detail = Column(Text, nullable=True)
+    window_id = Column(String(36), ForeignKey("maintenance_windows.id", ondelete="SET NULL"),
+                       nullable=True)
+    created_by = Column(String(36), ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow)
+    finished_at = Column(DateTime(timezone=True), nullable=True)

plugins/host_agent/routes.py

@@ -17,6 +17,7 @@ from steward.models.users import UserRole
 from sqlalchemy import select, or_, and_
 from datetime import timedelta
 
+from steward.core.audit import audit_request
 from steward.core.formatting import worst_reading
 from steward.core.health import ingest as ingest_tally
 from steward.core.metric_series import points as metric_points
@@ -26,6 +27,7 @@ from steward.hosts import service as host_svc
 from steward.models.hosts import Host
 from steward.models.metrics import PluginMetric
 from .downtime import recent as recent_downtime, record_events
+from . import maintenance
 from .models import HostAgentRegistration
 from .scheduler import arrivals
 # Query helpers live in a model-free module so integration tests can import them
@@ -347,6 +349,7 @@ async def ingest():
 
     token_hash = _hash_token(raw)
     phases = _Phases()
+    maint_steps: list = []
     samples, dropped = newest_samples(samples, MAX_SAMPLES_PER_INGEST)
     if dropped:
         current_app.logger.warning(
@@ -461,6 +464,20 @@ async def ingest():
             # was a restart without a shutdown.
             if events:
                 await record_events(session, host.id, events, reg.last_seen_at)
+            # Maintenance jobs (#5767): results and boots move a job on, then
+            # whatever step is due for this host's agent goes back in the answer.
+            # Each in its own savepoint: a job going wrong never costs the samples.
+            try:
+                async with session.begin_nested():
+                    if events:
+                        await maintenance.apply_events(session, host.id, events, received_at)
+                    if reg.maint_helper:
+                        maint_steps = await maintenance.handoffs(
+                            session, host.id, host.name, received_at)
+            except Exception:
+                maint_steps = []
+                current_app.logger.exception(
+                    "host_agent ingest: maintenance jobs failed for host=%s", host.name)
 
             md = payload.get("metadata") or {}
             changed = False
@@ -518,7 +535,10 @@ async def ingest():
     phases.report(host.name, accepted, len(metric_rows),
                   sum(len(c) for _, c in docker_snapshots),
                   sum(len(lines) for _, lines in docker_log_batches))
-    return jsonify({"ok": True, "samples_accepted": accepted, "samples_dropped": dropped}), 200
+    answer = {"ok": True, "samples_accepted": accepted, "samples_dropped": dropped}
+    if maint_steps:
+        answer["maintenance"] = maint_steps
+    return jsonify(answer), 200
 
 
 # The agent gives up on a post after 10 s (agent.py); an ingest past this
@@ -964,6 +984,14 @@ async def host_panel(host_id: str):
         target = (await db.execute(select(AnsibleTarget).where(
             AnsibleTarget.host_id == host_id))).scalar_one_or_none()
         downtime = await recent_downtime(db, host_id, datetime.now(timezone.utc)) if reg else []
+        # Maintenance (#5767): the running or latest job, and whether one can start.
+        maint_job = await maintenance.latest_job(db, host_id, datetime.now(timezone.utc)) if reg else None
+        maint_blocker = None
+        if reg is not None:
+            try:
+                await maintenance.plan_job(db, host, reg)
+            except maintenance.CannotStart as exc:
+                maint_blocker = str(exc)
 
     ls = reg.last_seen_at if reg else None
     # "Deployed" is gated on the agent actually checking in (last_seen_at), NOT on
@@ -976,6 +1004,8 @@ async def host_panel(host_id: str):
     return await render_template(
         "panel.html",
         host=host, reg=reg, reporting=reporting, stale=stale, target=target, downtime=downtime,
+        maint_job=maint_job, maint_blocker=maint_blocker,
+        maint_summary=maintenance.summary(maint_job) if maint_job else None,
         ansible_available=has_capability("ansible.run_playbook"),
         managed_key_set=bool((ansible_cfg.get("ssh_public_key") or "").strip()),
     )
@@ -1060,6 +1090,30 @@ async def rotate_token(host_id: str):
     return redirect(url_for("host_agent.fleet") + f"?new_token={raw}&host_id={host_id}")
 
 
+@host_agent_bp.post("/fleet/<host_id>/maintain")
+@require_role(UserRole.admin)
+async def start_maintenance(host_id: str):
+    """Update this host now, rebooting it only if the update needs it (#5767).
+    The job starts with the agent's next post and is followed on the panel."""
+    form = await request.form
+    policy = "never" if form.get("reboot_policy") == "never" else "if_required"
+    user_id = session.get("user_id")
+    async with current_app.db_sessionmaker() as db:
+        host = await db.get(Host, host_id)
+        if host is None:
+            return _error(404, "not_found")
+        reg = (await db.execute(select(HostAgentRegistration).where(
+            HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
+        try:
+            job = await maintenance.create_job(db, host, reg, user_id, policy)
+        except maintenance.CannotStart as exc:
+            return _error(409, "cannot_start", f"{host.name} can't be updated now: {exc}")
+        await db.commit()
+    await audit_request("host_agent.maintenance", "host", host_id,
+                        {"job_id": job.id, "reboot_policy": policy})
+    return redirect(f"/hosts/{host_id}")
+
+
 @host_agent_bp.post("/fleet/<host_id>/delete")
 @require_role(UserRole.admin)
 async def delete_registration(host_id: str):

plugins/host_agent/scheduler.py

@@ -119,3 +119,25 @@ def make_task(app) -> ScheduledTask:
         interval_seconds=60,
         run_on_startup=False,
     )
+
+
+def make_maintenance_task(app) -> ScheduledTask:
+    """Gives up on maintenance jobs past their deadline (#5767). Every other
+    move a job makes comes from its agent's posts."""
+    async def _tick():
+        from . import maintenance
+        try:
+            async with app.db_sessionmaker() as session:
+                async with session.begin():
+                    gave_up = await maintenance.tick(session, datetime.now(timezone.utc))
+            if gave_up:
+                logger.warning("host_agent: %d maintenance job(s) timed out", gave_up)
+        except Exception:
+            logger.exception("host_agent: maintenance tick failed")
+
+    return ScheduledTask(
+        name="host_agent_maintenance",
+        coro_factory=_tick,
+        interval_seconds=60,
+        run_on_startup=False,
+    )

plugins/host_agent/templates/panel.html

@@ -21,8 +21,16 @@
 
   {# What the host has waiting, from the agent's hourly check (#5765): a
      needed reboot first, then the updates. #}
-  {% if reg.updates_checked_at or reg.reboot_required %}
+  {% if reg.updates_checked_at or reg.reboot_required or maint_job %}
   <div style="margin-top:0.8rem;font-size:0.82rem;display:flex;flex-direction:column;gap:0.2rem;">
+    {# Steward's own maintenance job on this host (#5767): a running one first. #}
+    {% if maint_job %}
+    {% set running = maint_job.state not in ('done', 'failed', 'timed_out') %}
+    <div class="{{ 'tone-warn' if running or maint_job.state != 'done' else '' }}" title="{{ maint_job.detail or '' }}">
+      Maintenance: {{ maint_summary }}
+      <span style="color:var(--text-muted);">· {% if running %}since {{ maint_job.state_at|ago }}{% else %}{{ maint_job.finished_at|ago }}{% endif %}{% if maint_job.state in ('failed', 'timed_out') and maint_job.detail %} · {{ maint_job.detail|truncate(90) }}{% endif %}</span>
+    </div>
+    {% endif %}
     {% if reg.reboot_required %}
     <div class="tone-warn" title="{{ reg.reboot_packages or '' }}">
       Reboot required{% if reg.reboot_required_since %} since {{ reg.reboot_required_since|ago }}{% endif %}{% if reg.reboot_packages %}
@@ -78,6 +86,20 @@
       Link an Ansible target (below) to update the agent from here.
     </p>
     {% endif %}
+    {# Update now (#5767): the agent's helper updates the host and reboots it
+       only if the update needs it, draining its swarm node first. #}
+    {% if maint_blocker %}
+    <p style="font-size:0.8rem;margin:0.6rem 0 0;color:var(--text-dim);">Can't update from here: {{ maint_blocker }}.</p>
+    {% else %}
+    <form method="post" action="/plugins/host_agent/fleet/{{ host.id }}/maintain"
+          style="display:flex;gap:0.6rem;align-items:center;flex-wrap:wrap;margin:0.6rem 0 0;"
+          data-confirm="Update {{ host.name }} now? It reboots only if the update needs it, after its swarm node is drained.">{{ csrf_field() }}
+      <button type="submit" class="btn btn-sm">Update now</button>
+      <label style="font-size:0.8rem;color:var(--text-muted);display:flex;gap:0.3rem;align-items:center;">
+        <input type="checkbox" name="reboot_policy" value="never"> never reboot
+      </label>
+    </form>
+    {% endif %}
     {# The root helper that carries out scheduled updates and reboots (#5766). #}
     <p style="font-size:0.8rem;margin:0.6rem 0 0;color:var(--text-dim);">
       {% if reg.maint_helper %}Maintenance helper installed: Steward can update and reboot this host.

tests/core/test_host_page.py

@@ -279,3 +279,32 @@ async def test_manage_agent_says_whether_the_maintenance_helper_is_installed(app
             reg=SimpleNamespace(**{**vars(reg), "maint_helper": None}), **kw)
     assert "Maintenance helper installed" in installed
     assert "No maintenance helper yet: Update agent installs it" in missing
+
+
+async def test_the_agent_card_follows_a_maintenance_job_and_says_why_it_cant_start_one(app):
+    """#5767: a running job leads what the host has waiting; Update now is
+    offered only when a job can start, else the reason is shown."""
+    from datetime import datetime, timedelta, timezone
+    from plugins.host_agent.maintenance import summary
+    from tests._templates import render_plugin_template
+    now = datetime.now(timezone.utc)
+    host = SimpleNamespace(id="h1", name="box", address="10.0.0.5")
+    reg = SimpleNamespace(agent_version="1.18.0", last_seen_at=now, maint_helper=True,
+                          updates_checked_at=now, updates_pending=4, updates_security=0,
+                          reboot_required=False)
+    job = SimpleNamespace(state="rebooting", state_at=now - timedelta(minutes=2), finished_at=None,
+                          detail=None, packages=4, back_at=None, reboot_needed=True,
+                          reboot_policy="if_required")
+    kw = dict(host=host, reg=reg, reporting=True, stale=False, target=None, ansible_available=False,
+              managed_key_set=False, downtime=[])
+    async with app.test_request_context("/plugins/host_agent/panel/h1"):
+        session["user_role"] = "admin"
+        busy = await render_plugin_template(
+            app, "host_agent", "panel.html", maint_job=job, maint_summary=summary(job),
+            maint_blocker="a maintenance job is already running on it", **kw)
+        free = await render_plugin_template(app, "host_agent", "panel.html", maint_job=None,
+                                            maint_blocker=None, **kw)
+    assert busy.index("Maintenance: Rebooting") < busy.index("4 updates waiting")
+    assert "Can't update from here: a maintenance job is already running on it." in busy
+    assert ">Update now</button>" not in busy
+    assert ">Update now</button>" in free and 'name="reboot_policy" value="never"' in free

tests/integration/test_host_maintenance_job.py

@@ -0,0 +1,118 @@
+"""Integration (#5767): Steward hands a host a maintenance job and follows it
+to the end, through a restart of its own, from the agent's posts alone.
+
+Rows are seeded and read with plain SQL: importing the plugin's models again
+after the app loaded the plugin raises "Table already defined" (see
+test_host_agent_alerts)."""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+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)
+
+
+async def _seed(app):
+    from sqlalchemy import text
+    from steward.models.hosts import Host
+    from steward.models.users import User
+    tag = uuid.uuid4().hex[:8]
+    token = f"tok-{tag}-{uuid.uuid4().hex}"
+    job_id = str(uuid.uuid4())
+    async with app.db_sessionmaker() as s:
+        async with s.begin():
+            user = User(username=f"mj-{tag}", email=f"mj-{tag}@test", password_hash="x")
+            host = Host(name=f"maint-{tag}", address="10.9.9.7")
+            s.add_all([user, host])
+            await s.flush()
+            await s.execute(text(
+                "INSERT INTO host_agent_registrations "
+                "(id, host_id, token_hash, token_created_at, last_seen_at, maint_helper, "
+                " created_at, updated_at) "
+                "VALUES (:id, :host, :hash, now(), now(), true, now(), now())"),
+                {"id": str(uuid.uuid4()), "host": host.id,
+                 "hash": hashlib.sha256(token.encode()).hexdigest()})
+            await s.execute(text(
+                "INSERT INTO host_maintenance_jobs "
+                "(id, host_id, state, state_at, reboot_policy, step_seen, created_by, created_at) "
+                "VALUES (:id, :host, 'queued', now(), 'if_required', false, :user, now())"),
+                {"id": job_id, "host": host.id, "user": user.id})
+            host_id = host.id
+    return host_id, token, job_id
+
+
+async def _post(app, token, events=(), sample=True):
+    now = datetime.now(timezone.utc).isoformat()
+    resp = await app.test_client().post(
+        "/plugins/host_agent/ingest",
+        json={"samples": [{"ts": now, "cpu_pct": 2.0}] if sample else [],
+              "events": list(events)},
+        headers={"Authorization": f"Bearer {token}"})
+    assert resp.status_code == 200, await resp.get_data(as_text=True)
+    return (await resp.get_json()).get("maintenance", [])
+
+
+async def _job(app, job_id):
+    from sqlalchemy import text
+    async with app.db_sessionmaker() as s:
+        async with s.begin():
+            return (await s.execute(text(
+                "SELECT j.state, j.packages, j.reboot_needed, j.back_at, w.end_at "
+                "FROM host_maintenance_jobs j LEFT JOIN maintenance_windows w ON w.id = j.window_id "
+                "WHERE j.id = :id"), {"id": job_id})).one()
+
+
+def test_a_job_is_followed_through_a_steward_restart_and_the_host_reboot():
+    app = _app()
+
+    async def _first_half():
+        host_id, token, job_id = await _seed(app)
+        # The agent's next post starts the job: update first, under a window.
+        assert await _post(app, token) == [{"id": f"{job_id}-update", "action": "update"}]
+        state, *_, window_end = await _job(app, job_id)
+        assert state == "updating" and window_end > datetime.now(timezone.utc)
+        # The helper took it: the step is no longer handed out.
+        running = {"kind": "maintenance", "id": f"{job_id}-update", "action": "update",
+                   "state": "running"}
+        assert await _post(app, token, [running]) == []
+        return token, job_id
+
+    token, job_id = asyncio.run(_first_half())
+
+    # Steward restarts: a new process, nothing in memory but the database.
+    app = _app()
+
+    async def _second_half():
+        done = {"kind": "maintenance", "id": f"{job_id}-update", "action": "update",
+                "state": "done", "ok": True, "packages": 12, "reboot_required": True}
+        assert await _post(app, token, [done]) == [{"id": f"{job_id}-reboot", "action": "reboot"}]
+        assert (await _job(app, job_id))[:3] == ("rebooting", 12, True)
+        # The agent restarting on a host that stayed up doesn't count as the reboot.
+        old_boot = datetime.now(timezone.utc) - timedelta(days=3)
+        await _post(app, token, [{"kind": "started", "at": datetime.now(timezone.utc).isoformat(),
+                                  "boot_at": old_boot.isoformat()}])
+        assert (await _job(app, job_id))[0] == "rebooting"
+        # After the reboot: the helper's result and the agent's boot, in one post.
+        boot = datetime.now(timezone.utc)
+        rebooted = {"kind": "maintenance", "id": f"{job_id}-reboot", "action": "reboot",
+                    "state": "rebooting", "ok": True}
+        started = {"kind": "started", "at": (boot + timedelta(seconds=20)).isoformat(),
+                   "boot_at": boot.isoformat()}
+        assert await _post(app, token, [rebooted, started]) == []
+        state, packages, reboot_needed, back_at, window_end = await _job(app, job_id)
+        assert state == "done" and back_at is not None
+        assert window_end <= datetime.now(timezone.utc)
+
+    asyncio.run(_second_half())

tests/plugins/host_agent/test_maintenance.py

@@ -0,0 +1,154 @@
+# tests/plugins/host_agent/test_maintenance.py
+"""#5767: a host's maintenance job, step by step, from the agent's reports.
+
+The transitions are model-free, so a job here is a SimpleNamespace with the
+row's attributes; the database side is tests/integration/test_host_maintenance_job.py.
+"""
+import json
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from plugins.host_agent import agent as a
+from plugins.host_agent import maintenance as m
+
+T0 = datetime(2026, 10, 14, 23, 0, tzinfo=timezone.utc)
+
+
+def _job(**kw):
+    base = dict(id="job-1", host_id="h1", state="queued", state_at=T0, created_at=T0,
+                reboot_policy="if_required", step_id=None, step_seen=False, node_id=None,
+                drain_host_id=None, packages=None, reboot_needed=None, back_at=None,
+                detail=None, finished_at=None)
+    base.update(kw)
+    return SimpleNamespace(**base)
+
+
+def _result(job, state, **kw):
+    return {"kind": "maintenance", "id": job.step_id, "state": state, **kw}
+
+
+def test_an_update_with_no_reboot_needed_is_done_after_the_update():
+    job = _job()
+    m.start(job, T0)
+    assert m.handoff(job) == {"id": "job-1-update", "action": "update"}
+    assert m.apply_result(job, _result(job, "running"), T0)
+    assert m.handoff(job) is None                       # taken: not handed again
+    m.apply_result(job, _result(job, "done", ok=True, packages=3, reboot_required=False), T0)
+    assert (job.state, job.packages, job.finished_at) == ("done", 3, T0)
+    assert m.summary(job) == "Done: 3 packages updated, no reboot needed"
+
+
+def test_a_swarm_node_is_drained_rebooted_and_made_active_again():
+    job = _job(node_id="n" * 25, drain_host_id="mgr")
+    m.start(job, T0)
+    m.apply_result(job, _result(job, "done", ok=True, packages=12, reboot_required=True), T0)
+    assert job.state == "draining" and m.executor(job) == "mgr"
+    assert m.handoff(job) == {"id": "job-1-drain", "action": "drain", "node": "n" * 25}
+    m.apply_result(job, _result(job, "done", ok=True), T0)
+    assert job.state == "rebooting" and m.executor(job) == "h1"
+    assert m.handoff(job) == {"id": "job-1-reboot", "action": "reboot"}
+    t1 = T0 + timedelta(minutes=4)
+    assert m.apply_boot(job, T0 + timedelta(minutes=2), t1)
+    assert job.state == "undraining" and job.back_at == t1 and m.executor(job) == "mgr"
+    m.apply_result(job, _result(job, "done", ok=True), t1)
+    assert job.state == "done"
+    assert m.summary(job) == "Done: 12 packages updated, rebooted"
+
+
+def test_a_job_that_may_never_reboot_stops_after_the_update():
+    job = _job(reboot_policy="never")
+    m.start(job, T0)
+    m.apply_result(job, _result(job, "done", ok=True, packages=1, reboot_required=True), T0)
+    assert job.state == "done"
+    assert "reboot needed, not allowed" in m.summary(job)
+
+
+def test_only_a_boot_after_the_reboot_was_handed_out_completes_it():
+    job = _job(state="rebooting", step_id="job-1-reboot", state_at=T0)
+    assert not m.apply_boot(job, T0 - timedelta(days=3), T0)    # the agent restarted, not the host
+    assert not m.apply_boot(job, None, T0)
+    assert job.state == "rebooting"
+    assert m.apply_boot(job, T0 - timedelta(seconds=30), T0)     # within the clock slack
+    assert job.state == "done"
+
+
+def test_a_failed_step_fails_the_job_and_says_which():
+    job = _job(state="undraining", step_id="job-1-undrain", node_id="n" * 25, drain_host_id="mgr")
+    m.apply_result(job, _result(job, "failed", ok=False, detail="This node is not a swarm manager."), T0)
+    assert job.state == "failed"
+    assert job.detail.startswith("undrain failed: This node is not a swarm manager.")
+    assert "still drained" in job.detail
+
+
+def test_a_late_or_repeated_result_does_nothing():
+    job = _job()
+    m.start(job, T0)
+    stale = {"kind": "maintenance", "id": "job-1-drain", "state": "done", "ok": True}
+    assert not m.apply_result(job, stale, T0)
+    assert m.apply_result(job, _result(job, "running"), T0)
+    assert not m.apply_result(job, _result(job, "running"), T0)
+    m.apply_result(job, _result(job, "done", ok=True, reboot_required=False), T0)
+    assert not m.apply_result(job, {"kind": "maintenance", "id": "job-1-update", "state": "done"}, T0)
+
+
+@pytest.mark.parametrize("state,minutes,expected", [
+    ("queued", 59, False), ("queued", 61, True),
+    ("updating", 91, True), ("rebooting", 29, False), ("rebooting", 31, True),
+])
+def test_a_job_is_given_up_past_its_state_deadline(state, minutes, expected):
+    job = _job(state=state, step_id=f"job-1-{m.STEP_ACTION.get(state, 'x')}")
+    assert m.check_deadline(job, T0 + timedelta(minutes=minutes)) is expected
+    if expected:
+        assert job.state == "timed_out" and "within" in job.detail
+
+
+def test_the_window_lasts_the_rest_of_the_job_and_ends_with_it():
+    job = _job(state="rebooting", step_id="job-1-reboot")
+    assert m.window_end(job, T0) == T0 + timedelta(minutes=30 + 10)
+    job.state = "done"
+    assert m.window_end(job, T0) == T0
+
+
+def test_a_manager_drains_itself_and_a_worker_is_drained_by_a_manager():
+    nodes = {"docker02": "a" * 25, "docker03": "b" * 25}
+    managers = {"h2": "a" * 25}
+    assert m.pick_drainer(nodes, managers, "Docker02", "h2", {"h2", "h3"}) == ("a" * 25, "h2")
+    assert m.pick_drainer(nodes, managers, "Docker03", "h3", {"h2", "h3"}) == ("b" * 25, "h2")
+    assert m.pick_drainer(nodes, managers, "Util01", "h4", {"h2"}) == (None, None)
+    with pytest.raises(m.CannotStart, match="no swarm manager"):
+        m.pick_drainer(nodes, managers, "Docker03", "h3", {"h3"})
+
+
+# ── the agent's side ─────────────────────────────────────────────────────────
+
+def test_the_agent_leaves_each_handed_step_once(tmp_path):
+    requested: set = set()
+    steps = [{"id": "job-1-update", "action": "update"}, {"id": "job-1-x", "action": "shell"}, "junk"]
+    assert a.take_maintenance(steps, requested, requests_dir=str(tmp_path)) == ["job-1-update"]
+    assert json.loads((tmp_path / "job-1-update.json").read_text()) == {
+        "id": "job-1-update", "action": "update"}
+    (tmp_path / "job-1-update.json").unlink()          # the helper took it
+    assert a.take_maintenance(steps, requested, requests_dir=str(tmp_path)) == []
+    assert not (tmp_path / "job-1-update.json").exists()
+    assert a.take_maintenance(None, requested, requests_dir=str(tmp_path)) == []
+
+
+def test_the_agent_reads_stewards_answer_when_asked(monkeypatch):
+    class Resp:
+        status = 200
+
+        def read(self, n):
+            return b'{"ok": true, "maintenance": [{"id": "job-1-reboot", "action": "reboot"}]}'
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+    monkeypatch.setattr(a.urllib.request, "urlopen", lambda req, timeout: Resp())
+    answer: dict = {}
+    assert a.post_payload("https://x", "tok", {}, answer) == (True, 200)
+    assert answer["maintenance"] == [{"id": "job-1-reboot", "action": "reboot"}]