fix(mcp): get_anomaly keeps the events nearest the occurrence and says how many it left out (#5861)

Each source in an anomaly's `around` was cut to its oldest 40 with
nothing in the reply to say so; on a busy host the events nearest the
occurrence were the ones dropped. Each source now keeps the 40 nearest
the occurrence's start, and `around_left_out` counts the rest. Sources
return everything their bounded query read (Docker and the host agent
no longer cut at 40 and 20 themselves), so the count is honest.

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

docs/plugins/writing-a-plugin.md

@@ -403,7 +403,7 @@ async def mcp_events(db, host_id, host_name, since, until) -> list[dict]:
             for r in await _jobs_on(db, host_id, since, until)]
 ```
 
-- Return what happened between `since` and `until`, oldest first: something that lasted gives `until`. Keep it to tens of rows; the caller keeps the first 40.
+- Return what happened between `since` and `until`, oldest first: something that lasted gives `until`. Bound your query (a few hundred rows); Steward keeps the 40 nearest the occurrence's start and tells the caller how many it left out (`around_left_out`).
 - It is called only while the plugin is enabled, with its own session. If it raises, the anomaly is still returned without the plugin's events. The host agent (maintenance jobs, reboots) and Docker (container starts, deaths, OOM kills) are the examples to read.
 
 A resource with no host (a Traefik router, say) has no host to read events by. A plugin that knows what happened to one of its own resources exports `mcp_resource_events(db, source, resource, since, until)` instead, same return shape, and returns `[]` for a source that isn't its own. Traefik's is the example: it gives a router's swarm service's container events, on whichever hosts they ran.

plugins/docker/mcp_tools.py

@@ -758,7 +758,7 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic
     from steward.mcp.tools._events import event
     rows = (await db.execute(select(DockerEvent).where(
         DockerEvent.host_id == host_id, DockerEvent.at >= since, DockerEvent.at <= until)
-        .order_by(DockerEvent.at).limit(40))).scalars()
+        .order_by(DockerEvent.at).limit(500))).scalars()
     return [event(e.at, "container", f"{e.container_name} {e.event}" + (f" ({e.detail})" if e.detail else ""),
                   subject=f"{host_name}/{e.container_name}") for e in rows]
 

plugins/host_agent/mcp_tools.py

@@ -303,7 +303,7 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic
     jobs = (await db.execute(select(HostMaintenanceJob).where(
         HostMaintenanceJob.host_id == host_id, HostMaintenanceJob.created_at <= until,
         (HostMaintenanceJob.finished_at.is_(None)) | (HostMaintenanceJob.finished_at >= since))
-        .order_by(HostMaintenanceJob.created_at).limit(20))).scalars()
+        .order_by(HostMaintenanceJob.created_at).limit(200))).scalars()
     for job in jobs:
         out.append(event(job.created_at, "maintenance", f"Maintenance: {maintenance.summary(job)}",
                          until=job.finished_at, ids={"job_id": job.id}))
@@ -311,7 +311,7 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic
     rows = (await db.execute(select(HostDowntime).where(
         HostDowntime.host_id == host_id, HostDowntime.started_at <= until,
         (HostDowntime.ended_at.is_(None)) | (HostDowntime.ended_at >= since))
-        .order_by(HostDowntime.started_at).limit(20))).scalars()
+        .order_by(HostDowntime.started_at).limit(200))).scalars()
     for row in rows:
         seen = downtime.describe(row, now)
         out.append(event(row.started_at, "downtime",

plugins/traefik/mcp_tools.py

@@ -323,7 +323,7 @@ async def service_events(db, router: str, since, until) -> list[dict]:
     host: a router anomaly has no host of its own to read events by (#5862).
     [] without the Docker plugin, or when no single service matches."""
     from steward.core.plugin_manager import loaded_plugin_submodule
-    from steward.mcp.tools._events import PER_SOURCE, event
+    from steward.mcp.tools._events import event
     from steward.models.hosts import Host
     models = loaded_plugin_submodule("docker", "models")
     if models is None:
@@ -335,7 +335,7 @@ async def service_events(db, router: str, since, until) -> list[dict]:
     rows = (await db.execute(
         select(e, Host.name).join(Host, Host.id == e.host_id)
         .where(e.container_name.startswith(f"{service}.", autoescape=True), e.at >= since, e.at <= until)
-        .order_by(e.at).limit(PER_SOURCE))).all()
+        .order_by(e.at).limit(500))).all()
     return [event(ev.at, "container", f"{ev.container_name} {ev.event}" + (f" ({ev.detail})" if ev.detail else ""),
                   subject=f"{host}/{ev.container_name}") for ev, host in rows]
 

steward/mcp/tools/_events.py

@@ -34,7 +34,8 @@ from ._shape import db_session, iso
 
 logger = logging.getLogger(__name__)
 
-PER_SOURCE = 40     # events any one source contributes to one window
+PER_SOURCE = 40     # events any one source contributes to one window: the
+                    # nearest the occurrence's start, the rest counted (#5861)
 # Steward samples its own health once a minute (core/self_monitor.py); a
 # longer break between two samples is a stretch it wasn't recording.
 STEWARD_GAP_S = 90.0
@@ -98,7 +99,7 @@ async def alert_events(db, place: Place, since: datetime, until: datetime) -> li
         note = f" ({ev.note})" if ev.note else ""
         out.append(event(ev.transitioned_at, "alert", f"{rule.name}: {ev.from_state} → {ev.to_state}{note}",
                          subject=rule.resource_name, ids={"rule_id": rule.id}))
-    return out[:PER_SOURCE]
+    return out
 
 
 async def ansible_runs(db, place: Place, since: datetime, until: datetime) -> list[dict]:
@@ -119,7 +120,7 @@ async def ansible_runs(db, place: Place, since: datetime, until: datetime) -> li
         out.append(event(r.started_at, "ansible_run",
                          f"Ansible {r.playbook_path}{check}: {r.status.value}",
                          until=r.finished_at, ids={"run_id": r.id}))
-    return out[:PER_SOURCE]
+    return out
 
 
 def _same_host(source: str, resource: str, place: Place) -> bool:
@@ -148,7 +149,7 @@ async def other_anomalies(db, place: Place, anomaly_id: str, since: datetime,
                               until=span.ended_at, subject=a.resource_name, ids={"anomaly_id": a.id}))
         else:
             elsewhere += 1
-    return here[:PER_SOURCE], elsewhere
+    return here, elsewhere
 
 
 def gaps_in(times: list[datetime], since: datetime, until: datetime) -> list[dict]:
@@ -179,9 +180,10 @@ async def steward_gaps(db, since: datetime, until: datetime) -> list[dict]:
     return gaps_in(list(rows), since, until)
 
 
-async def plugin_events(place: Place, since: datetime, until: datetime) -> list[dict]:
-    """What enabled plugins know happened on the host (`mcp_events`), each
-    in its own session so one that fails can't spoil the others' reads."""
+async def plugin_events(place: Place, since: datetime, until: datetime) -> list[list[dict]]:
+    """What enabled plugins know happened on the host (`mcp_events`), one
+    list per plugin, each read in its own session so one that fails can't
+    spoil the others' reads."""
     server = getattr(mcp_app(), "mcp_server", None)
     if server is None or not place.host_id:
         return []
@@ -191,15 +193,16 @@ async def plugin_events(place: Place, since: datetime, until: datetime) -> list[
             continue
         try:
             async with db_session() as db:
-                out.extend((await source(db, place.host_id, place.host_name, since, until))[:PER_SOURCE])
+                out.append(await source(db, place.host_id, place.host_name, since, until))
         except Exception:
             logger.exception("MCP events source of plugin %r failed", plugin)
     return out
 
 
-async def resource_events(place: Place, since: datetime, until: datetime) -> list[dict]:
+async def resource_events(place: Place, since: datetime, until: datetime) -> list[list[dict]]:
     """What enabled plugins know happened to the resource itself
-    (`mcp_resource_events`), for one with no host to read events by."""
+    (`mcp_resource_events`), for one with no host to read events by: one
+    list per plugin."""
     server = getattr(mcp_app(), "mcp_server", None)
     if server is None:
         return []
@@ -209,18 +212,37 @@ async def resource_events(place: Place, since: datetime, until: datetime) -> lis
             continue
         try:
             async with db_session() as db:
-                out.extend((await source(db, place.source, place.resource, since, until))[:PER_SOURCE])
+                out.append(await source(db, place.source, place.resource, since, until))
         except Exception:
             logger.exception("MCP resource events source of plugin %r failed", plugin)
     return out
 
 
-async def around(db, place: Place, anomaly_id: str, since: datetime,
-                 until: datetime) -> tuple[list[dict], int]:
-    """Everything above for one window, oldest first, and the count of
-    anomalies elsewhere."""
+def _at(e: dict) -> datetime | None:
+    return datetime.fromisoformat(e["at"].replace("Z", "+00:00")) if e.get("at") else None
+
+
+def nearest(events: list[dict], focus: datetime, keep: int = PER_SOURCE) -> tuple[list[dict], int]:
+    """The `keep` events nearest `focus` (an occurrence's start), and how
+    many were left out: a busy host's window is cut where it matters least."""
+    if len(events) <= keep:
+        return events, 0
+    by_distance = sorted(events, key=lambda e: abs((_at(e) - focus).total_seconds()) if _at(e) else float("inf"))
+    return by_distance[:keep], len(events) - keep
+
+
+async def around(db, place: Place, anomaly_id: str, since: datetime, until: datetime,
+                 focus: datetime) -> tuple[list[dict], int, int]:
+    """Everything above for one window, oldest first; the count of anomalies
+    elsewhere; and how many events were left out, each source keeping its
+    PER_SOURCE nearest `focus`."""
     anomalies, elsewhere = await other_anomalies(db, place, anomaly_id, since, until)
-    found = (await alert_events(db, place, since, until) + await ansible_runs(db, place, since, until)
-             + anomalies + await plugin_events(place, since, until)
-             + await resource_events(place, since, until) + await steward_gaps(db, since, until))
-    return sorted(found, key=lambda e: e["at"] or ""), elsewhere
+    groups = [await alert_events(db, place, since, until), await ansible_runs(db, place, since, until),
+              anomalies, *await plugin_events(place, since, until),
+              *await resource_events(place, since, until), await steward_gaps(db, since, until)]
+    found, left_out = [], 0
+    for group in groups:
+        kept, cut = nearest(group, focus)
+        found += kept
+        left_out += cut
+    return sorted(found, key=lambda e: e["at"] or ""), elsewhere, left_out

steward/mcp/tools/anomalies.py

@@ -163,7 +163,9 @@ async def get_anomaly(anomaly_id: str, occurrences: int = 5) -> dict:
       stops; for a Traefik router, its swarm service's containers on any
       host), and kind "steward": a stretch Steward itself wasn't recording
       (a restart or deploy), around which any reading can be wrong. From 30
-      minutes before to 30 minutes after; and
+      minutes before to 30 minutes after. Each kind keeps the 40 events
+      nearest the occurrence's start; `around_left_out` counts the rest
+      (0 when nothing was cut). And
       `anomalies_elsewhere`, how many anomalies on other hosts overlapped it.
       Many at once points at something shared (network, storage, Steward).
     - `series`: the newest shown occurrence with time either side (as long
@@ -194,8 +196,8 @@ async def get_anomaly(anomaly_id: str, occurrences: int = 5) -> dict:
         shown = []
         for span in spans[:limit]:
             usual = _usual_at(kind, slots, span.started_at)
-            near, elsewhere = await events.around(db, place, a.id, span.started_at - AROUND,
-                                                  (span.ended_at or now) + AROUND)
+            near, elsewhere, left_out = await events.around(db, place, a.id, span.started_at - AROUND,
+                                                            (span.ended_at or now) + AROUND, span.started_at)
             shown.append({
                 "started_at": iso(span.started_at), "ended_at": iso(span.ended_at),
                 "ongoing": span.ended_at is None,
@@ -206,6 +208,7 @@ async def get_anomaly(anomaly_id: str, occurrences: int = 5) -> dict:
                            "flag_above": rnd(flag_above(usual.usual_hold, floor), 3) if floor is not None else None}
                           if usual else None),
                 "around": near,
+                "around_left_out": left_out,
                 "anomalies_elsewhere": elsewhere,
             })
         series = None

tests/mcp/test_anomaly_review.py

@@ -119,6 +119,16 @@ def test_a_plugin_adds_a_hostless_resources_events_through_its_hook(app):
         mcp.resource_event_sources.pop("events_test", None)
 
 
+def test_a_busy_window_keeps_the_events_nearest_the_occurrence_and_counts_the_rest():
+    """#5861: 50 container events over an hour; the occurrence starts at :45."""
+    evs = [_events.event(NOW + timedelta(minutes=m), "container", f"c{m} start") for m in range(50)]
+    kept, cut = _events.nearest(evs, NOW + timedelta(minutes=45))
+    assert len(kept) == 40 and cut == 10
+    assert {e["what"] for e in kept} >= {"c45 start", "c49 start", "c20 start"}
+    assert "c0 start" not in {e["what"] for e in kept}
+    assert _events.nearest(evs[:5], NOW) == (evs[:5], 0)
+
+
 def test_steward_not_recording_is_an_event_beside_every_anomaly():
     """A Steward restart between two of its once-a-minute samples (#5862)."""
     ticks = [NOW + timedelta(minutes=m) for m in (0, 1, 2)] + [NOW + timedelta(minutes=4, seconds=10)]