fix(host_agent): a node the operator keeps drained stays drained through its maintenance (#5777)

#5777 "Maintenance would make a node the operator keeps drained active after
its reboot". A swarm node already set to drain or pause, such as a manager
kept free of work, is no longer drained before its reboot or made active
after it. Its job only updates and reboots it, and back-in-service needs it
ready but not active.

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

plugins/host_agent/cadence.py

@@ -74,19 +74,21 @@ def is_due(plan, now: datetime) -> datetime | None:
     return slot
 
 
-def back_in_service(*, last_seen_at, node, services, job_finished_at, now: datetime) -> str | None:
+def back_in_service(*, last_seen_at, node, services, job_finished_at, now: datetime,
+                    expect_active: bool = True) -> str | None:
     """None when the host is back in service, else why not yet. `node` is the
     freshest swarm-node row for the job's node (None outside a swarm, or when
     the job has no node); `services` the freshest row per swarm service. Only
     rows reported after the job finished count, so a stale "all fine" from
-    before the reboot never passes."""
+    before the reboot never passes. `expect_active` is False for a node the
+    job didn't drain (the operator keeps it drained): it need only be ready."""
     if last_seen_at is None or now - last_seen_at > POSTING_WITHIN:
         return "its agent isn't posting yet"
     if node is None:
         return None
     if node.updated_at < job_finished_at:
         return "no swarm report since it came back"
-    if node.status != "ready" or node.availability != "active":
+    if node.status != "ready" or (expect_active and node.availability != "active"):
         return f"its swarm node is {node.status}/{node.availability}"
     fresh = [s for s in services if s.updated_at >= job_finished_at]
     if not fresh:
@@ -184,7 +186,8 @@ async def _back(session, host, job, reg, now: datetime) -> str | None:
                     freshest[s.service_name] = s
             services = list(freshest.values())
     return back_in_service(last_seen_at=reg.last_seen_at if reg else None, node=node,
-                           services=services, job_finished_at=job.finished_at or now, now=now)
+                           services=services, job_finished_at=job.finished_at or now, now=now,
+                           expect_active=bool(job.drain_host_id))
 
 
 async def advance(session, run, plan, now: datetime) -> str:

plugins/host_agent/maintenance.py

@@ -283,14 +283,18 @@ class CannotStart(Exception):
 
 
 def pick_drainer(node_hostname_to_id: dict, managers: dict, host_name: str,
-                 host_id: str, helpers: set) -> tuple:
+                 host_id: str, helpers: set, availability: str = "active") -> 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."""
+    drains itself when it is a manager, else another manager does. A node the
+    operator already set to drain or pause (a manager kept free of work) is
+    left as it is: no drain before, and no making it active after."""
     node_id = node_hostname_to_id.get(host_name.lower())
     if node_id is None:
         return None, None
+    if availability != "active":
+        return node_id, None
     if host_id in managers and host_id in helpers:
         return node_id, host_id
     for mgr in sorted(managers):
@@ -316,10 +320,12 @@ async def plan_job(session, host, reg, reboot_policy: str = "if_required") -> tu
         raise CannotStart("a maintenance job is already running on it")
     if reboot_policy == "never":
         return None, None
-    nodes, managers = await _swarm(session)
+    nodes, managers, availability = 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)
+    node_id = nodes.get(host.name.lower())
+    return pick_drainer(nodes, managers, host.name, host.id, helpers,
+                        availability.get(node_id, "active"))
 
 
 async def create_job(session, host, reg, user_id: str | None,
@@ -394,12 +400,13 @@ def docker_models():
 
 
 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."""
+    """({node hostname: node id}, {manager host id: its node id},
+    {node id: availability}) 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."""
     models = docker_models()
     if models is None:
-        return {}, {}
+        return {}, {}, {}
     DockerSwarmNode = models.DockerSwarmNode
     from sqlalchemy import select
     from steward.models.hosts import Host
@@ -409,4 +416,6 @@ async def _swarm(session) -> tuple:
     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
+    availability = {n.node_id: (n.availability or "active")
+                    for n in sorted(rows, key=lambda n: n.updated_at)}
+    return nodes, managers, availability

tests/plugins/host_agent/test_cadence.py

@@ -66,6 +66,11 @@ def test_a_host_is_back_once_posting_its_node_active_and_its_services_at_replica
                              services=[_svc("web", 1, 2), _svc("agent", 2, 3, mode="global")],
                              **kw) == "services below their replicas: web"
     assert c.back_in_service(last_seen_at=NOW, node=_node(), services=[_svc("web", 2, 2)], **kw) is None
+    # A node the job didn't drain stays drained, as the operator keeps it.
+    assert c.back_in_service(last_seen_at=NOW, node=_node(availability="drain"),
+                             services=[_svc("web", 1, 1)], expect_active=False, **kw) is None
+    assert "down" in c.back_in_service(last_seen_at=NOW, node=_node(status="down", availability="drain"),
+                                       services=[_svc("web", 1, 1)], expect_active=False, **kw)
 
 
 def _run(**kw):

tests/plugins/host_agent/test_maintenance.py

@@ -121,6 +121,16 @@ def test_a_manager_drains_itself_and_a_worker_is_drained_by_a_manager():
         m.pick_drainer(nodes, managers, "Docker03", "h3", {"h3"})
 
 
+def test_a_node_the_operator_keeps_drained_is_left_drained():
+    # A manager kept free of work (availability drain) is neither drained
+    # before nor made active after: the job only updates and reboots it.
+    nodes = {"util01": "c" * 25, "docker02": "a" * 25}
+    managers = {"h1": "c" * 25, "h2": "a" * 25}
+    assert m.pick_drainer(nodes, managers, "Util01", "h1", {"h1", "h2"}, "drain") == ("c" * 25, None)
+    assert m.pick_drainer(nodes, managers, "Util01", "h1", {"h1", "h2"}, "pause") == ("c" * 25, None)
+    assert m.pick_drainer(nodes, managers, "Util01", "h1", {"h1", "h2"}, "active") == ("c" * 25, "h1")
+
+
 # ── the agent's side ─────────────────────────────────────────────────────────
 
 def test_the_agent_leaves_each_handed_step_once(tmp_path):