fix(alerts): one silent agent is one alert and one finding (#5721)

VDNT-Docker03's silent agent read as three overview findings (agent offline,
agent stale, the disk rule's no-data) and every drop emailed the disk rule's
silence and return on top of agent offline.

- The host agent registers its heartbeat (seconds_since_seen) and the
  sources it carries (host_agent, docker). While a host's agent is past its
  offline window, its other rules' no_data is recorded "while its agent is
  offline" and not emailed, nor is their quiet return; a return past a
  threshold still fires.
- The overview folds those rules under the host's agent-offline finding
  ("also silent with it: ..."), and a stale agent whose agent-offline alert
  is active is that alert's finding, not a second one.

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

plugins/docker/__init__.py

@@ -48,6 +48,9 @@ def setup(app: "Quart") -> None:
     from steward.core.norms import floor_setting, learn_normal
     learn_normal("docker", "cpu_pct", floor_setting(app, "container_cpu_pct"))
     learn_normal("docker", "mem_pct", floor_setting(app, "container_mem_pct"))
+    # A host's agent carries its containers' figures: they go silent with it (#5721).
+    from steward.core.alerts import agent_fed
+    agent_fed("docker", lambda resource: resource.split("/", 1)[0])
 
 
 def get_scheduled_tasks() -> list:

plugins/host_agent/__init__.py

@@ -19,6 +19,12 @@ def setup(app: "Quart") -> None:
     learn_normal("host_agent", "cpu_pct", floor_setting(app, "host_cpu_pct"), parts=False)
     learn_normal("host_agent", "mem_used_pct", floor_setting(app, "host_mem_pct"), parts=False)
     learn_normal("host_agent", "load_pct", floor_setting(app, "host_load_pct"), parts=False)
+    # "Agent offline" speaks for everything the agent carries: while it is
+    # firing, the host's other rules going silent isn't emailed (#5721).
+    from steward.core.alerts import agent_fed, agent_heartbeat
+    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])
 
 
 def get_scheduled_tasks() -> list:

steward/core/alerts.py

@@ -4,7 +4,7 @@ import logging
 import uuid
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Callable
 
 from sqlalchemy import event as sa_event, select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -35,6 +35,75 @@ def init_alerts(app: "Quart") -> None:
     _app = app
 
 
+# ── One silent agent is one alert (#5721) ───────────────────────────────────
+# A host's agent carries more than its own figures (its containers' too), so
+# when it goes quiet every rule it feeds goes silent with it. That is one
+# cause: "agent offline" says it, and the other rules' silence is recorded but
+# not emailed, nor is their quiet return. A return past a threshold still fires.
+
+AGENT_OFFLINE = "while its agent is offline"
+
+
+@dataclass(frozen=True)
+class Heartbeat:
+    source_module: str
+    metric_name: str
+    offline_after: Callable[[], float]     # seconds; read at each use
+
+
+_HEARTBEAT: Heartbeat | None = None
+_AGENT_FED: dict[str, Callable[[str], str]] = {}
+
+
+def agent_heartbeat(source_module: str, metric_name: str,
+                    offline_after: Callable[[], float]) -> None:
+    """The metric that says how long each host's agent has been quiet; past
+    `offline_after()` the agent is offline. The host agent plugin registers it."""
+    global _HEARTBEAT
+    _HEARTBEAT = Heartbeat(source_module, metric_name, offline_after)
+
+
+def agent_fed(source_module: str, host_of: Callable[[str], str]) -> None:
+    """A source whose samples a host's agent carries; `host_of(resource)`
+    names the host ("nas:/" → "nas", "nas/web" → "nas")."""
+    _AGENT_FED[source_module] = host_of
+
+
+def is_heartbeat(rule) -> bool:
+    """Whether `rule` watches the agent heartbeat ("agent offline")."""
+    return _HEARTBEAT is not None and (rule.source_module, rule.metric_name) == (
+        _HEARTBEAT.source_module, _HEARTBEAT.metric_name)
+
+
+def host_behind(rule) -> str | None:
+    """The host whose agent feeds `rule`, or None: a rule on another source,
+    or the heartbeat itself (agent offline is the one that speaks)."""
+    host_of = _AGENT_FED.get(rule.source_module)
+    if host_of is None or is_heartbeat(rule):
+        return None
+    return host_of(rule.resource_name)
+
+
+async def offline_agents(session) -> set[str]:
+    """Hosts whose agent's last heartbeat was past its offline window."""
+    if _HEARTBEAT is None:
+        return set()
+    return set((await session.execute(
+        select(PluginMetricLatest.resource_name).where(
+            PluginMetricLatest.source_module == _HEARTBEAT.source_module,
+            PluginMetricLatest.metric_name == _HEARTBEAT.metric_name,
+            PluginMetricLatest.value > _HEARTBEAT.offline_after()))).scalars())
+
+
+async def _went_quiet_with_its_agent(session, rule_id: str) -> bool:
+    """Whether a rule's latest silence was recorded while its agent was offline."""
+    note = (await session.execute(
+        select(AlertEvent.note).where(AlertEvent.rule_id == rule_id,
+                                      AlertEvent.to_state == "no_data")
+        .order_by(AlertEvent.transitioned_at.desc()).limit(1))).scalar_one_or_none()
+    return bool(note and AGENT_OFFLINE in note)
+
+
 def alert_extra_vars(rule: AlertRule, value: float) -> dict:
     """Ansible extra-vars describing the alert that fired.
 
@@ -250,15 +319,24 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric], phases
                          key=lambda r: r.recorded_at)
         muted = _muted_by(windows, rule)
         resumed = _track_samples(alert_state, samples, now)
+        # A silence ends when Steward hears of it: an agent's flushed backlog
+        # brings samples hours old, and stamping their events at their own
+        # times put the resume inside the silence it ends (#5720).
+        back = resumed is not None
         for sample in samples:
+            at = max(sample.recorded_at, now) if back else sample.recorded_at
             notifications = await _evaluate_rule(session, rule, alert_state, sample.value,
-                                                 sample.recorded_at, note=muted, resumed=resumed)
+                                                 at, note=muted, resumed=resumed)
             resumed = None  # said once, with the first sample back
             if muted:
                 continue
             for notif_type, event_id, note in notifications:
                 if notif_type == "data_resumed" and not rule.notify_no_data:
                     continue
+                # Back with its agent: the agent's return is the news (#5721).
+                if notif_type == "data_resumed" and host_behind(rule) is not None \
+                        and await _went_quiet_with_its_agent(session, rule.id):
+                    continue
                 _after_commit(session, _dispatch_notification, notif_type, rule, sample.value,
                               event_id, note)
                 # Admin-configured Ansible action runs only on transition-to-firing.
@@ -319,7 +397,13 @@ def _track_samples(state: AlertState, samples: list[PluginMetric], now: datetime
     series had been marked no_data, clear the mark and return the note saying
     so ("data resumed after 15m silent"); the first sample's evaluation
     records it, in one event with whatever that sample does to the rule
-    (#5442)."""
+    (#5442).
+
+    A resumed series counts as last heard `now`, when Steward heard it, not at
+    its newest sample's time: a backlog flushed a few samples a post is hours
+    old, and judging silence by it marked the series silent again between
+    posts, so one silence was resumed twice (#5720). The rest of the backlog
+    then reads as late samples, which teach the cadence nothing."""
     for sample in samples:
         if state.last_sample_at is not None:
             if sample.recorded_at <= state.last_sample_at:
@@ -331,6 +415,7 @@ def _track_samples(state: AlertState, samples: list[PluginMetric], now: datetime
         return None
     note = f"data resumed after {human_duration((now - state.no_data_since).total_seconds())} silent"
     state.no_data_since = None
+    state.last_sample_at = max(state.last_sample_at or now, now)
     return note
 
 
@@ -658,6 +743,7 @@ async def check_silence(app: "Quart") -> None:
                 select(MaintenanceWindow).where(
                     MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now)
             )).scalars().all()
+            offline = await offline_agents(session) if _AGENT_FED else set()
             for state, rule in rows:
                 watching_since = max(rule.created_at, state.last_transition_at)
                 since = silent_since(state.last_sample_at, state.sample_interval_s,
@@ -673,9 +759,11 @@ async def check_silence(app: "Quart") -> None:
                     if state.sample_interval_s:
                         note += f" (usually every {human_duration(state.sample_interval_s)})"
                 muted = _muted_by(windows, rule)
-                event_id = _add_event(session, rule.id, state.state.value, "no_data", None,
-                                      now, note=f"{note}; {muted}" if muted else note)
-                if rule.notify_no_data and not muted:
+                # Silent with its agent: recorded, and "agent offline" is the email (#5721).
+                hushed = AGENT_OFFLINE if host_behind(rule) in offline else None
+                event_id = _add_event(session, rule.id, state.state.value, "no_data", None, now,
+                                      note="; ".join(n for n in (note, hushed, muted) if n))
+                if rule.notify_no_data and not muted and not hushed:
                     notify.append((rule, event_id, note))
     for rule, event_id, note in notify:
         _spawn(_dispatch_notification(app, "no_data", rule, None, event_id, note))

steward/mcp/tools/overview.py

@@ -7,6 +7,7 @@ from datetime import timedelta
 from sqlalchemy import select
 
 from steward.alerts import service as alert_svc
+from steward.core import alerts
 from steward.core import anomalies as anom
 from steward.hosts import service as host_svc
 from steward.mcp._context import mcp_app
@@ -70,8 +71,37 @@ def monitor_findings(monitors, states: dict, host_names: dict, now,
     return out
 
 
-def alert_findings(active) -> list[dict]:
+def fold_agent_silence(active, agent: dict) -> tuple[list, dict[str, list[str]], set[str]]:
+    """One silent agent is one finding (#5721). Returns (the active alerts left
+    to list, {host: names of its rules silent with its agent}, the hosts
+    whose "agent offline" alert is active and so speaks for the agent).
+
+    A rule the agent feeds that has only gone quiet, on a host whose agent is
+    stale, is folded under that host; one firing or pending stays listed,
+    since what it measured before going quiet is news of its own."""
+    offline = {name for name, a in agent.items() if a and not a.get("fresh")}
+    kept, silent, speaking = [], {}, set()
+    for state, rule in active:
+        if alerts.is_heartbeat(rule):
+            speaking.add(rule.resource_name)
+            kept.append((state, rule))
+            continue
+        host = alerts.host_behind(rule)
+        if host in offline and state.no_data_since is not None \
+                and state.state.value not in ("firing", "pending", "acknowledged"):
+            silent.setdefault(host, []).append(rule.name)
+            continue
+        kept.append((state, rule))
+    return kept, silent, speaking
+
+
+def _also_silent(names: list[str] | None) -> str:
+    return f"; also silent with it: {', '.join(sorted(names))}" if names else ""
+
+
+def alert_findings(active, silent: dict[str, list[str]] | None = None) -> list[dict]:
     out = []
+    silent = silent or {}
     for state, rule in active:
         s = state.state.value
         if s == "firing":
@@ -84,6 +114,8 @@ def alert_findings(active) -> list[dict]:
             s, sev, what = "no_data", "warning", "has stopped reporting"
         if state.no_data_since is not None and s != "no_data":
             what += f" and has stopped reporting (no data since {iso(state.no_data_since)})"
+        if alerts.is_heartbeat(rule):
+            what += _also_silent(silent.get(rule.resource_name))
         out.append(finding(
             sev, f"alert_{s}", rule.name,
             f"{rule.metric_name} {rule.operator.value} {rule.threshold:g} on "
@@ -98,18 +130,26 @@ _AGENT_CHECKS = (("cpu_pct", "cpu", "CPU"), ("mem_used_pct", "mem", "memory"),
                  ("disk_used_pct_worst", "disk", "fullest disk"))
 
 
-def agent_findings(hosts, agent: dict, thresholds: dict) -> list[dict]:
+def agent_findings(hosts, agent: dict, thresholds: dict,
+                   silent: dict[str, list[str]] | None = None,
+                   speaking: set[str] | frozenset = frozenset()) -> list[dict]:
+    """Hosts past a CPU, memory or disk cutoff, and agents that stopped
+    reporting. A stale agent whose "agent offline" alert is active (`speaking`)
+    is that alert's finding, not a second one (#5721)."""
     out = []
+    silent = silent or {}
     for h in hosts:
         a = agent.get(h.name)
         if not a:
             continue
         ids = {"host_id": h.id}
         if not a.get("fresh"):
-            out.append(finding(
-                "warning", "agent_stale", h.name,
-                "host agent has stopped reporting; its metrics are out of date",
-                since=a.get("last_at"), ids=ids, next=f'get_host(host="{h.id}")'))
+            if h.name not in speaking:
+                out.append(finding(
+                    "warning", "agent_stale", h.name,
+                    "host agent has stopped reporting; its metrics are out of date"
+                    + _also_silent(silent.get(h.name)),
+                    since=a.get("last_at"), ids=ids, next=f'get_host(host="{h.id}")'))
             continue  # stale values would only add noise
         # CPU and memory within the host's own normal for the hour aren't
         # findings (#5742); the fullest disk keeps its cutoff.
@@ -216,9 +256,10 @@ async def get_overview() -> dict:
         to_review = await anom.to_review_count(db)
 
     thresholds = app.config.get("THRESHOLDS") or {}
+    listed, silent, speaking = fold_agent_silence(active, agent)
     findings = (monitor_findings(monitors, states, host_names, now, thresholds.get("tls_days"))
-                + alert_findings(active)
-                + agent_findings(hosts, agent, thresholds)
+                + alert_findings(listed, silent)
+                + agent_findings(hosts, agent, thresholds, silent, speaking)
                 + run_findings(runs)
                 + anomaly_findings(open_anomalies))
     plugin_found, unavailable = await _plugin_findings(getattr(app, "mcp_server", None))

tests/core/test_alert_no_data.py

@@ -151,3 +151,19 @@ def test_muted_by_names_the_window_that_covers_the_rule():
     assert _muted_by(windows[:1], rule) is None
     assert _muted_by([NS(name="all", scope_source=None, scope_resource=None)], rule) == (
         "notification muted: maintenance window all")
+
+
+def test_a_flushed_backlog_counts_as_heard_when_it_arrives():
+    """#5720: an agent's backlog, flushed a couple of samples a post, is hours
+    old. Judged by its newest sample, the series read silent again between
+    posts and resumed twice. It counts as heard when Steward heard it, and the
+    rest of the backlog reads as late."""
+    st = _state(last_sample_at=NOW - timedelta(hours=3), sample_interval_s=60,
+                no_data_since=NOW - timedelta(hours=2, minutes=55))
+    note = _track_samples(st, _samples((2 * 3600 + 60, 5.0), (2 * 3600, 5.0)), NOW)
+    assert note == "data resumed after 2h 55m silent"
+    assert st.last_sample_at == NOW
+    assert silent_since(st.last_sample_at, st.sample_interval_s, NOW, NOW) is None
+    # The next post of the same backlog is late: no gap, no second resume.
+    assert _track_samples(st, _samples((2 * 3600 - 60, 5.0)), NOW) is None
+    assert st.last_sample_at == NOW

tests/integration/test_alert_no_data.py

@@ -189,3 +189,70 @@ def test_a_series_quiet_only_while_steward_was_down_is_not_silent(app):
     with patch.object(alerts, "_dispatch_notification", fake_dispatch):
         asyncio.run(_go())
 
+
+
+def test_a_rule_silent_with_its_agent_says_nothing_of_its_own(app):
+    """#5721: while a host's agent is offline, "agent offline" is the one
+    email. The host's other rules going silent is recorded, marked as with
+    its agent, and not emailed; their quiet return isn't either, but a
+    return past the threshold still fires."""
+    from sqlalchemy import select
+    from steward.alerts import service as alert_svc
+    from steward.core import alerts
+    from steward.core.watch import watch
+    from steward.models.alerts import AlertEvent, AlertState
+    from steward.models.metrics import PluginMetric, PluginMetricLatest
+    from steward.models.users import User, UserRole
+
+    alerts.agent_heartbeat("host_agent", "seconds_since_seen", lambda: 180.0)
+    alerts.agent_fed("host_agent", lambda resource: resource.split(":", 1)[0])
+    sent = []
+
+    async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
+        sent.append((rule.id, notif_type))
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        host = f"gone-{tag}"
+        now = datetime.now(timezone.utc)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                user = User(username=f"ag-{tag}", email=f"ag-{tag}@test", password_hash="x",
+                            role=UserRole.admin)
+                s.add(user)
+                await alerts.record_metrics(s, [PluginMetric(
+                    source_module="host_agent", resource_name=f"{host}:/", metric_name="disk_used_pct",
+                    value=50.0, recorded_at=now - timedelta(minutes=20))])
+                await s.flush()
+                fields = await alert_svc.rule_fields(
+                    s, name=f"disk {tag}", source_module="host_agent", resource_name=f"{host}:/",
+                    metric_name="disk_used_pct", operator=">", threshold=90, notify_no_data=True)
+                rule_id = (await alert_svc.add_rule(s, fields, created_by=user.id)).id
+                await s.flush()
+                st = (await s.execute(select(AlertState).where(AlertState.rule_id == rule_id))).scalar_one()
+                st.last_sample_at, st.sample_interval_s = now - timedelta(minutes=20), 60.0
+                # The agent has been quiet for 20 minutes: offline.
+                await s.merge(PluginMetricLatest(source_module="host_agent", resource_name=host,
+                                                 metric_name="seconds_since_seen", value=1200.0,
+                                                 recorded_at=now))
+
+        w = watch(app, "alert_silence")
+        w.since, w.last = now - timedelta(days=1), datetime.now(timezone.utc)
+        await alerts.check_silence(app)
+        await asyncio.sleep(0)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                [note] = (await s.execute(select(AlertEvent.note).where(
+                    AlertEvent.rule_id == rule_id, AlertEvent.to_state == "no_data"))).scalars().all()
+        assert alerts.AGENT_OFFLINE in note
+        assert sent == []                         # agent offline is the email
+
+        # The agent comes back and the disk is fine: back quietly.
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alerts.record_metric(s, "host_agent", f"{host}:/", "disk_used_pct", 50.0)
+        await asyncio.sleep(0)
+        assert sent == []
+
+    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
+        asyncio.run(_go())

tests/mcp/test_tools.py

@@ -283,3 +283,37 @@ def test_get_host_groups_parts_by_kind_and_summarises_cores():
     assert got["temperatures"]["items"][0] == {"sensor": "Package id 0", "part": "temp:Package id 0",
                                                "temp_c": 61.0}
     assert "other" not in got
+
+
+def test_one_silent_agent_is_one_finding():
+    """#5721: VDNT-Docker03's silent agent read as three findings. Its agent-
+    offline alert speaks for it, the stale-agent finding goes, and its rules
+    that only went quiet are listed under it; one still firing stays listed."""
+    from steward.core import alerts
+    alerts.agent_heartbeat("host_agent", "seconds_since_seen", lambda: 180.0)
+    alerts.agent_fed("host_agent", lambda resource: resource.split(":", 1)[0])
+
+    def rule(name, metric, resource):
+        return NS(id=name, name=name, metric_name=metric, operator=AlertOperator.gt, threshold=90.0,
+                  source_module="host_agent", resource_name=resource)
+
+    gone = NOW - timedelta(hours=2)
+    active = [
+        (NS(state=NS(value="firing"), last_transition_at=gone, no_data_since=None),
+         rule("Agent offline: d3", "seconds_since_seen", "d3")),
+        (NS(state=NS(value="inactive"), last_transition_at=gone, no_data_since=gone),
+         rule("Disk nearly full: d3", "disk_used_pct", "d3:/")),
+        (NS(state=NS(value="firing"), last_transition_at=gone, no_data_since=gone),
+         rule("Memory high: d3", "mem_used_pct", "d3")),
+    ]
+    agent = {"d3": {"fresh": False, "last_at": gone}}
+    listed, silent, speaking = overview.fold_agent_silence(active, agent)
+    assert [r.name for _s, r in listed] == ["Agent offline: d3", "Memory high: d3"]
+    assert silent == {"d3": ["Disk nearly full: d3"]} and speaking == {"d3"}
+    found = overview.alert_findings(listed, silent) + overview.agent_findings(
+        [NS(id="h3", name="d3")], agent, THRESHOLDS, silent, speaking)
+    assert [f["subject"] for f in found] == ["Agent offline: d3", "Memory high: d3"]
+    assert "also silent with it: Disk nearly full: d3" in found[0]["summary"]
+    # Without its alert, the stale agent is the finding, and says the same.
+    [stale] = overview.agent_findings([NS(id="h3", name="d3")], agent, THRESHOLDS, silent)
+    assert stale["kind"] == "agent_stale" and "Disk nearly full: d3" in stale["summary"]