feat(host_agent): a reboot reads as a reboot, and lost contact as its span (#5756)

#5756 "The agent says when its host rebooted and when it lost contact, so
downtime is recorded as what it was". Agent 1.14.0 posts events with the
next post that gets through: started (with the host's boot time), stopping
(on SIGTERM, with whether systemd says the host is going down) and
lost_contact (from its first unanswered post to the one carrying it).

Steward keeps them as the host's downtime (host_downtime): reboot,
agent_stop, unclean (a boot after the agent was last heard that nobody
announced) or lost_contact. Retried posts apply once.

An announced stop explains the silence for reboot_grace_seconds (Settings
-> Plugins -> host_agent, 10 minutes): seconds_since_seen reads 0 until
then, and the host's other rules go quiet with it (agents_away beside
#5721's registry). Past the grace, agent offline fires.

The host's Agent card lists the last 7 days of downtime. Rows are pruned
at data.retention_days.

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

plugins/host_agent/__init__.py

@@ -25,6 +25,15 @@ def setup(app: "Quart") -> None:
     from steward.hosts.service import agent_stale_after_seconds
     agent_heartbeat("host_agent", "seconds_since_seen", lambda: agent_stale_after_seconds(app.config))
     agent_fed("host_agent", lambda resource: resource.split(":", 1)[0])
+    # A reboot the agent announced is silence with a reason, for a while (#5756).
+    from datetime import datetime, timezone
+    from steward.core.alerts import agents_away
+    from .downtime import hosts_away, restart_grace_seconds
+    agents_away(lambda session: hosts_away(session, datetime.now(timezone.utc),
+                                           restart_grace_seconds(app.config)))
+    from steward.core.cleanup import prune_by_age
+    from .models import HostDowntime
+    prune_by_age(HostDowntime.started_at)
 
 
 def get_scheduled_tasks() -> list:

plugins/host_agent/agent.py

@@ -12,6 +12,7 @@ import re
 import shutil
 import signal
 import socket
+import subprocess
 import sys
 import time
 import urllib.error
@@ -20,7 +21,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.13.0"
+AGENT_VERSION = "1.14.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
@@ -1330,16 +1331,70 @@ def build_sample(mounts: list[str], state: dict,
     return sample
 
 
-def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict:
+def build_payload(samples: list[dict], hostname: str, metadata: dict,
+                  events: list | None = None) -> dict:
     # sent_at is this host's clock at the moment of sending: Steward compares
     # it with its own on arrival to say how far apart the two clocks are.
-    return {
+    payload = {
         "agent_version": AGENT_VERSION,
         "hostname": hostname,
         "metadata": metadata,
         "sent_at": datetime.now(timezone.utc).isoformat(),
         "samples": samples,
     }
+    if events:
+        payload["events"] = events
+    return payload
+
+
+# ─── what happened to the host (#5756) ───────────────────────────────────────
+# Events ride the next post that gets through, so Steward records a reboot as a
+# reboot and a lost connection as the span it lasted, not as the agent gone.
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def started_event() -> dict:
+    """The agent starting, with when the host booted: Steward tells a host
+    that rebooted from an agent that only restarted by it."""
+    event = {"kind": "started", "at": _now_iso()}
+    try:
+        booted = time.time() - float(_read_file(UPTIME_PATH).split()[0])
+        event["boot_at"] = datetime.fromtimestamp(booted, timezone.utc).isoformat()
+    except Exception:
+        pass
+    return event
+
+
+def host_shutting_down():
+    """Whether the host itself is going down, not only this agent: systemd says
+    "stopping" while it shuts down or reboots. None when it can't tell."""
+    try:
+        out = subprocess.run(["systemctl", "is-system-running"], stdout=subprocess.PIPE,
+                             stderr=subprocess.DEVNULL, universal_newlines=True,
+                             timeout=3).stdout.strip()
+    except (OSError, subprocess.SubprocessError):
+        return None
+    return out == "stopping" if out else None
+
+
+def stopping_event() -> dict:
+    return {"kind": "stopping", "at": _now_iso(), "host_shutdown": host_shutting_down()}
+
+
+def lost_contact_event(since: str) -> dict:
+    """The span this agent could not reach Steward: from its first failed post
+    to now, the post that is carrying it."""
+    return {"kind": "lost_contact", "from": since, "to": _now_iso()}
+
+
+def lost_contact(status) -> bool:
+    """Whether a failed post means Steward couldn't be reached: no answer, or
+    a proxy's 5xx in front of a Steward that is down. A 400 or 401 is Steward
+    answering."""
+    return status is None or status >= 500
 
 
 # ─── POST + backoff ──────────────────────────────────────────────────────────
@@ -1441,6 +1496,9 @@ def main_loop(conf_path: str) -> int:
     docker_log_exclude = cfg.get("docker_log_exclude") or []
     buffer = RingBuffer(maxlen=20)
     backoff = 0
+    # Said once each, with the next post that gets through (#5756).
+    events: list = [started_event()]
+    lost_since = None
     # Carries previous net/disk counters + monotonic ts for rate computation.
     rate_state: dict = {}
 
@@ -1470,10 +1528,14 @@ def main_loop(conf_path: str) -> int:
             samples=send + [sample],
             hostname=hostname,
             metadata=metadata,
+            events=events + ([lost_contact_event(lost_since)] if lost_since else []),
         )
         ok, status = post_payload(cfg["url"], cfg["token"], payload)
 
         if ok:
+            if lost_since:
+                _log("INFO", f"reached Steward again (unreachable since {lost_since})")
+            events, lost_since = [], None
             for s in held:
                 buffer.push(s)
             if send:
@@ -1482,6 +1544,8 @@ def main_loop(conf_path: str) -> int:
             backoff = 0
             sleep_for = cfg["interval_seconds"]
         else:
+            if lost_since is None and lost_contact(status):
+                lost_since = sample["ts"]
             if status == 400:
                 _log("ERROR", "server rejected payload (400) — dropping sample")
             elif status == 401:
@@ -1501,10 +1565,12 @@ def main_loop(conf_path: str) -> int:
             slept += 1.0
 
     _log("INFO", "SIGTERM — flushing and exiting")
-    final = buffer.drain()
-    if final:
-        post_payload(cfg["url"], cfg["token"],
-                     build_payload(final, hostname, metadata))
+    # Say it is stopping, and whether the host is going down with it, so Steward
+    # reads the silence that follows as a reboot rather than the agent lost.
+    final_events = events + ([lost_contact_event(lost_since)] if lost_since else [])
+    final_events.append(stopping_event())
+    post_payload(cfg["url"], cfg["token"],
+                 build_payload(buffer.drain(), hostname, metadata, events=final_events))
     return 0
 
 

plugins/host_agent/downtime.py

@@ -0,0 +1,160 @@
+# plugins/host_agent/downtime.py
+"""What a host's agent says happened to it while it was away (#5756).
+
+An agent ≥ 1.14.0 tells Steward when it stops (and whether its host is going
+down with it), when it starts (and when its host booted), and how long it
+couldn't reach Steward. Each becomes a HostDowntime row, so a reboot reads as
+a reboot with its length, and "agent offline" is left for silence nobody
+explained. A post can be retried, so every event is applied idempotently.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+
+from .models import HostDowntime
+
+# A stop the agent announced: the silence after it is explained, for a while.
+ANNOUNCED = ("reboot", "agent_stop")
+
+LABELS = {
+    "reboot": "Rebooted",
+    "agent_stop": "Agent stopped",
+    "unclean": "Restarted without shutting down",
+    "lost_contact": "Lost contact",
+}
+
+
+def restart_grace_seconds(config) -> int:
+    """How long an announced stop explains the agent's silence (Settings →
+    Plugins → host_agent, default 10 minutes): past it, agent offline fires."""
+    return int(((config.get("PLUGINS") or {}).get("host_agent") or {})
+               .get("reboot_grace_seconds", 600))
+
+
+def _at(value) -> datetime | None:
+    if not isinstance(value, str):
+        return None
+    try:
+        got = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return got if got.tzinfo else got.replace(tzinfo=timezone.utc)
+
+
+def _when(event: dict) -> datetime | None:
+    return _at(event.get("at") or event.get("from"))
+
+
+async def record_events(session, host_id: str, events, last_seen_at: datetime | None) -> int:
+    """Apply one post's events to the host's downtime, oldest first; returns
+    how many rows were added or changed. `last_seen_at` is when the agent was
+    last heard before this post: a boot after it that nobody announced is a
+    restart without a shutdown."""
+    if not isinstance(events, list):
+        return 0
+    timed = sorted(((t, e) for e in events if isinstance(e, dict)
+                    for t in [_when(e)] if t is not None), key=lambda te: te[0])
+    changed = 0
+    for at, event in timed:
+        kind = event.get("kind")
+        if kind == "stopping":
+            changed += await _stopping(session, host_id, at, event.get("host_shutdown"))
+        elif kind == "started":
+            changed += await _started(session, host_id, at, _at(event.get("boot_at")), last_seen_at)
+        elif kind == "lost_contact":
+            changed += await _lost_contact(session, host_id, at, _at(event.get("to")))
+    return changed
+
+
+async def _stopping(session, host_id: str, at: datetime, host_shutdown) -> int:
+    seen = (await session.execute(select(HostDowntime.id).where(
+        HostDowntime.host_id == host_id, HostDowntime.kind.in_(ANNOUNCED),
+        HostDowntime.started_at == at))).first()
+    if seen:
+        return 0
+    session.add(HostDowntime(host_id=host_id, kind="reboot" if host_shutdown else "agent_stop",
+                             started_at=at))
+    return 1
+
+
+async def _started(session, host_id: str, at: datetime, boot_at: datetime | None,
+                   last_seen_at: datetime | None) -> int:
+    stop = (await session.execute(select(HostDowntime).where(
+        HostDowntime.host_id == host_id, HostDowntime.kind.in_(ANNOUNCED),
+        HostDowntime.ended_at.is_(None), HostDowntime.started_at <= at)
+        .order_by(HostDowntime.started_at.desc()).limit(1))).scalar_one_or_none()
+    if stop is not None:
+        stop.ended_at = at
+        # The boot time settles what the stop was: the host came back from a
+        # boot after it, or it stayed up while only the agent restarted.
+        if boot_at is not None:
+            stop.kind = "reboot" if boot_at > stop.started_at else "agent_stop"
+        return 1
+    if boot_at is None or last_seen_at is None or boot_at <= last_seen_at:
+        return 0
+    seen = (await session.execute(select(HostDowntime.id).where(
+        HostDowntime.host_id == host_id, HostDowntime.kind == "unclean",
+        HostDowntime.ended_at == at))).first()
+    if seen:
+        return 0
+    session.add(HostDowntime(host_id=host_id, kind="unclean", started_at=last_seen_at, ended_at=at))
+    return 1
+
+
+async def _lost_contact(session, host_id: str, since: datetime, until: datetime | None) -> int:
+    if until is None or until <= since:
+        return 0
+    row = (await session.execute(select(HostDowntime).where(
+        HostDowntime.host_id == host_id, HostDowntime.kind == "lost_contact",
+        HostDowntime.started_at == since))).scalar_one_or_none()
+    if row is None:
+        session.add(HostDowntime(host_id=host_id, kind="lost_contact", started_at=since, ended_at=until))
+        return 1
+    if row.ended_at is None or until > row.ended_at:
+        row.ended_at = until
+        return 1
+    return 0
+
+
+async def announced_stops(session, now: datetime, grace_seconds: int) -> dict[str, datetime]:
+    """{host id: when the silence its agent announced stops being explained}
+    for every announced stop still going and inside its grace."""
+    rows = (await session.execute(select(HostDowntime.host_id, HostDowntime.started_at).where(
+        HostDowntime.kind.in_(ANNOUNCED), HostDowntime.ended_at.is_(None),
+        HostDowntime.started_at > now - timedelta(seconds=grace_seconds)))).all()
+    out: dict[str, datetime] = {}
+    for host_id, started in rows:
+        until = started + timedelta(seconds=grace_seconds)
+        out[host_id] = max(out.get(host_id, until), until)
+    return out
+
+
+async def hosts_away(session, now: datetime, grace_seconds: int) -> set[str]:
+    """Names of the hosts in announced_stops: their silence has a reason."""
+    from steward.models.hosts import Host
+    ids = list(await announced_stops(session, now, grace_seconds))
+    if not ids:
+        return set()
+    return set((await session.execute(select(Host.name).where(Host.id.in_(ids)))).scalars())
+
+
+def describe(row: HostDowntime, now: datetime) -> dict:
+    """One downtime row as the host page shows it."""
+    label = LABELS.get(row.kind, row.kind)
+    if row.kind == "agent_stop" and row.ended_at is not None:
+        label = "Agent restarted"
+    end = row.ended_at or now
+    return {"kind": row.kind, "label": label, "started_at": row.started_at,
+            "ended_at": row.ended_at, "ongoing": row.ended_at is None,
+            "seconds": max(0.0, (end - row.started_at).total_seconds())}
+
+
+async def recent(session, host_id: str, now: datetime, days: int = 7, limit: int = 6) -> list[dict]:
+    """The host's downtime in the last `days`, newest first."""
+    rows = (await session.execute(select(HostDowntime).where(
+        HostDowntime.host_id == host_id,
+        HostDowntime.started_at > now - timedelta(days=days))
+        .order_by(HostDowntime.started_at.desc()).limit(limit))).scalars().all()
+    return [describe(r, now) for r in rows]

plugins/host_agent/migrations/versions/host_agent_004_downtime.py

@@ -0,0 +1,32 @@
+# plugins/host_agent/migrations/versions/host_agent_004_downtime.py
+"""host_agent: each host's downtime and what it was (#5756)
+
+Revision ID: host_agent_004_downtime
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "host_agent_004_downtime"
+down_revision: Union[str, None] = "host_agent_003_clock_offset"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "host_downtime",
+        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("kind", sa.String(16), nullable=False),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_host_downtime_host_id", "host_downtime", ["host_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_host_downtime_host_id", table_name="host_downtime")
+    op.drop_table("host_downtime")

plugins/host_agent/models.py

@@ -37,3 +37,22 @@ class HostAgentRegistration(Base):
     created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow)
     updated_at = Column(DateTime(timezone=True), nullable=False,
                         default=_utcnow, onupdate=_utcnow)
+
+
+class HostDowntime(Base):
+    """A span a host's agent was away, and what the agent said it was (#5756).
+
+    kind is "reboot" (the host went down and came back), "agent_stop" (the
+    agent stopped; the host stayed up), "unclean" (the host came back from a
+    boot nobody announced: power loss or a crash) or "lost_contact" (the agent
+    ran but couldn't reach Steward). ended_at is None while it is going on.
+    """
+    __tablename__ = "host_downtime"
+
+    id = Column(String(36), primary_key=True, default=_uuid)
+    host_id = Column(String(36), ForeignKey("hosts.id", ondelete="CASCADE"),
+                     nullable=False, index=True)
+    kind = Column(String(16), nullable=False)
+    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)

plugins/host_agent/plugin.yaml

@@ -16,4 +16,7 @@ tags:
 
 config:
   stale_after_seconds: 180
+  # How long a reboot or restart the agent announced explains its silence
+  # before "agent offline" fires (agent >= 1.14.0, #5756).
+  reboot_grace_seconds: 600
   default_interval_seconds: 30

plugins/host_agent/routes.py

@@ -24,6 +24,7 @@ from steward.core.time_range import parse_range, RANGE_OPTIONS
 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 .models import HostAgentRegistration
 from .scheduler import arrivals
 # Query helpers live in a model-free module so integration tests can import them
@@ -232,7 +233,10 @@ async def ingest():
         return _error(400, "malformed_payload", "missing 'samples'")
 
     samples = payload.get("samples") or []
-    if not isinstance(samples, list) or not samples:
+    # What happened to the host (agent ≥ 1.14.0, #5756). A stopping agent's last
+    # post may carry its events alone, with no sample left to send.
+    events = payload.get("events") if isinstance(payload.get("events"), list) else []
+    if not isinstance(samples, list) or not (samples or events):
         return _error(400, "malformed_payload", "'samples' must be a non-empty list")
 
     token_hash = _hash_token(raw)
@@ -305,7 +309,7 @@ async def ingest():
                 if latest_ts is None or recorded_at > latest_ts:
                     latest_ts = recorded_at
 
-            if accepted == 0:
+            if accepted == 0 and not events:
                 await session.rollback()
                 return _error(400, "malformed_payload", "no valid samples")
             # How far this host's clock is from Steward's, either way, so an
@@ -346,6 +350,11 @@ async def ingest():
                             host.name)
                     phases.mark("docker")
 
+            # Before last_seen_at moves: a boot after it that nobody announced
+            # was a restart without a shutdown.
+            if events:
+                await record_events(session, host.id, events, reg.last_seen_at)
+
             md = payload.get("metadata") or {}
             changed = False
             if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at):
@@ -818,6 +827,7 @@ async def host_panel(host_id: str):
             HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
         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 []
 
     ls = reg.last_seen_at if reg else None
     # "Deployed" is gated on the agent actually checking in (last_seen_at), NOT on
@@ -829,7 +839,7 @@ async def host_panel(host_id: str):
     ansible_cfg = current_app.config.get("ANSIBLE", {})
     return await render_template(
         "panel.html",
-        host=host, reg=reg, reporting=reporting, stale=stale, target=target,
+        host=host, reg=reg, reporting=reporting, stale=stale, target=target, downtime=downtime,
         ansible_available=has_capability("ansible.run_playbook"),
         managed_key_set=bool((ansible_cfg.get("ssh_public_key") or "").strip()),
     )

plugins/host_agent/scheduler.py

@@ -12,6 +12,7 @@ from steward.core.watch import watch
 from steward.hosts.service import agent_stale_after_seconds
 from steward.models.hosts import Host
 from steward.models.metrics import PluginMetricLatest
+from .downtime import announced_stops, restart_grace_seconds
 from .models import HostAgentRegistration
 
 logger = logging.getLogger(__name__)
@@ -36,6 +37,7 @@ def seconds_since_seen(regs: Iterable, host_names: dict[str, str],
                        now: datetime, watching_since: datetime | None = None,
                        arrived: dict[str, datetime] | None = None,
                        already_silent: set[str] | frozenset = frozenset(),
+                       explained_until: dict[str, datetime] | None = None,
                        ) -> list[tuple[str, float]]:
     """(host name, seconds since its agent last reported) per registration.
 
@@ -52,6 +54,11 @@ def seconds_since_seen(regs: Iterable, host_names: dict[str, str],
     whole silence counts. Counting them from `watching_since` read them as
     just seen, which resolved their alert on every restart and fired it again
     three minutes later (#5719).
+
+    `explained_until` ({host id: time}) is a stop the agent announced, a reboot
+    or its own restart, plus its grace (#5756): until then the agent reads as
+    just seen. A host still silent past it reads its whole silence, so agent
+    offline fires as the grace runs out, for silence nobody explained.
     """
     out = []
     for r in regs:
@@ -62,6 +69,8 @@ def seconds_since_seen(regs: Iterable, host_names: dict[str, str],
             continue
         if watching_since is not None and name not in already_silent:
             seen = max(seen, watching_since)
+        if r.host_id in (explained_until or {}):
+            seen = max(seen, explained_until[r.host_id])
         out.append((name, max(0.0, (now - seen).total_seconds())))
     return out
 
@@ -88,9 +97,10 @@ async def record_seconds_since_seen(app) -> None:
                     PluginMetricLatest.metric_name == "seconds_since_seen",
                     PluginMetricLatest.resource_name.in_(list(host_names.values())),
                     PluginMetricLatest.value > stale_after))).scalars())
+            explained = await announced_stops(session, now, restart_grace_seconds(app.config))
             batch = MetricBatch()
             for name, seconds in seconds_since_seen(regs, host_names, now, watching_since,
-                                                    arrivals(app), already_silent):
+                                                    arrivals(app), already_silent, explained):
                 batch.add(source_module="host_agent", resource_name=name,
                           metric_name="seconds_since_seen", value=seconds, recorded_at=now)
             await batch.record(session)

plugins/host_agent/templates/panel.html

@@ -19,6 +19,20 @@
      "Manage agent" (#5278). #}
   <a href="/plugins/host_agent/{{ host.id }}/" class="btn btn-sm btn-ghost">Full metrics →</a>
 
+  {# What the agent said happened while it was away, last 7 days (#5756). #}
+  {% if downtime %}
+  <div style="margin-top:0.8rem;">
+    <div style="font-size:0.78rem;color:var(--text-muted);margin-bottom:0.3rem;">Downtime, last 7 days</div>
+    {% for d in downtime %}
+    <div style="display:flex;gap:0.6rem;align-items:baseline;font-size:0.82rem;padding:0.15rem 0;">
+      <span class="{{ 'tone-warn' if d.ongoing or d.kind == 'unclean' else '' }}" style="min-width:12rem;">{{ d.label }}</span>
+      <span style="color:var(--text-muted);">{{ d.started_at|when }}</span>
+      <span style="color:var(--text-muted);margin-left:auto;">{{ 'going on, ' if d.ongoing }}{{ d.seconds|duration }}</span>
+    </div>
+    {% endfor %}
+  </div>
+  {% endif %}
+
   {% if session.user_role == 'admin' %}
   <details style="margin-top:0.7rem;">
     <summary style="cursor:pointer;font-size:0.82rem;color:var(--text-muted);">Manage agent</summary>