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>

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