feat(docker): overview findings say when the problem began

#5013 "Overview findings for containers and Swarm services carry since=null"

- Container problems are dated from their own evidence: a crash or restart
  loop from its first death in the hour, an OOM kill or error exit from the
  death that ended it, a failing healthcheck from its last turn to unhealthy.
  list_containers and get_container carry problem_since / problem_age_s.
- Swarm services carry trouble_since: a crash loop's first failure in the
  hour, a short service's oldest troubled task in each failing slot, a
  recovered service's first failure in 24h. list_swarm_services shows it.
- get_overview findings from the docker plugin now have since and age_s,
  so they rank by how long they have stood.

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

plugins/docker/mcp_tools.py

@@ -53,6 +53,29 @@ def container_problems(c, failures_1h: int, failed_24h: bool,
     return out
 
 
+def problem_since(problems: list[str], deaths: list, hour_ago: datetime,
+                  unhealthy_at: datetime | None = None) -> datetime | None:
+    """When a container's worst problem began; None when the retained events
+    cannot say. `deaths` are its die/oom events of the last 24h, newest first.
+
+    A restart or crash loop dates from its first death in the last hour, an
+    OOM kill or error exit from the death that ended it, and a failing
+    healthcheck from when its health last turned unhealthy.
+    """
+    if not problems:
+        return None
+    worst = problems[0]
+    if worst in ("restarting", "crash_looping"):
+        in_hour = [e.at for e in deaths if e.at >= hour_ago]
+        return min(in_hour) if in_hour else (deaths[0].at if deaths else None)
+    if worst == "unhealthy":
+        return unhealthy_at
+    if worst == "oom_killed":
+        ooms = [e.at for e in deaths if e.event == "oom"]
+        return ooms[0] if ooms else (deaths[0].at if deaths else None)
+    return deaths[0].at if deaths else None
+
+
 def cut_middle(text: str, limit: int = LOG_LINE_CHARS) -> str:
     """Keep both ends of an over-long line and say how much of the middle went."""
     if text is None or len(text) <= limit:
@@ -62,7 +85,7 @@ def cut_middle(text: str, limit: int = LOG_LINE_CHARS) -> str:
 
 
 def container_row(c, host_names: dict, problems: list[str], now: datetime,
-                  replaced: bool = False) -> dict:
+                  replaced: bool = False, since: datetime | None = None) -> dict:
     running = (c.status or "").lower() == "running"
     return {
         "container_name": c.name,
@@ -72,6 +95,9 @@ def container_row(c, host_names: dict, problems: list[str], now: datetime,
         "status": c.status,
         "health": c.health,
         "problems": problems,
+        # When the worst problem began, from the container's retained events.
+        "problem_since": iso(since),
+        "problem_age_s": age_s(since, now),
         "exit_code": c.exit_code,
         "oom_killed": bool(c.oom_killed),
         "restart_count": c.restart_count,
@@ -97,6 +123,10 @@ _SAYS = {"restarting": "is restarting", "crash_looping": "keeps dying",
          "exited_nonzero": "exited with an error"}
 
 
+def _from_iso(text: str | None) -> datetime | None:
+    return datetime.fromisoformat(text.replace("Z", "+00:00")) if text else None
+
+
 def container_findings(rows: list[dict]) -> list[dict]:
     out = []
     for r in rows:
@@ -108,6 +138,7 @@ def container_findings(rows: list[dict]) -> list[dict]:
         out.append(finding(
             _SEVERITY[worst], f"container_{worst}", r["container_name"],
             f"container {r['container_name']}{where} {' and '.join(_SAYS[p] for p in r['problems'])}{detail}",
+            since=_from_iso(r["problem_since"]),
             ids={"host_id": r["host_id"], "container_name": r["container_name"]},
             next=f'get_container(container="{r["container_name"]}", host="{r["host_id"]}")'))
     return out
@@ -132,22 +163,25 @@ def service_findings(services: list[dict]) -> list[dict]:
         last = task_error_text(s.get("last_error"))
         tail = f"; last task error: {last}" if last else ""
         ids = {"service_name": s["name"]}
+        since = s.get("trouble_since")
         if s.get("crash_looping"):
             out.append(finding(
                 "critical", "swarm_service_crash_looping", s["name"],
                 f"swarm service {s['name']} keeps failing: {s['failures_1h']} tasks failed in the "
-                f"last hour, {replicas}{tail}", ids=ids, next="list_swarm_services()"))
+                f"last hour, {replicas}{tail}", since=since, ids=ids,
+                next="list_swarm_services()"))
         elif s["state"] != "healthy":
             out.append(finding(
                 "critical" if s["state"] == "down" else "warning", f"swarm_service_{s['state']}",
                 s["name"], f"swarm service {s['name']} has {replicas}{tail}",
-                ids=ids, next="list_swarm_services()"))
+                since=since, ids=ids, next="list_swarm_services()"))
         elif s.get("failures_24h"):
             n = s["failures_24h"]
             out.append(finding(
                 "warning", "swarm_task_failed", s["name"],
                 f"swarm service {s['name']} had {n} failed task{'s' if n != 1 else ''} in the last "
-                f"24h and is now {replicas}{tail}", ids=ids, next="list_swarm_services()"))
+                f"24h and is now {replicas}{tail}", since=since, ids=ids,
+                next="list_swarm_services()"))
     return out
 
 
@@ -175,13 +209,17 @@ async def _rows(db, now: datetime) -> list[dict]:
     hour_ago = now - timedelta(hours=1)
     host_names = await _host_names(db)
     deaths = _crash_loop_deaths()
+    sick = await docker_svc.unhealthy_since(
+        db, [(c.host_id, c.name) for c in containers if (c.health or "") == "unhealthy"])
     rows = []
     for c in containers:
-        gone = (c.host_id, c.name) in replaced
+        key = (c.host_id, c.name)
-        events = failures.get((c.host_id, c.name), [])
+        gone = key in replaced
+        events = failures.get(key, [])
         recent = sum(1 for e in events if e.at >= hour_ago)
         problems = [] if gone else container_problems(c, recent, bool(events), deaths)
-        rows.append(container_row(c, host_names, problems, now, replaced=gone))
+        since = problem_since(problems, events, hour_ago, sick.get(key))
+        rows.append(container_row(c, host_names, problems, now, replaced=gone, since=since))
     return rows
 
 
@@ -222,8 +260,10 @@ async def list_containers(host: str = "", status: str = "", problem: str = "",
     limit: default 100, max 500.
 
     Containers with problems come first. Each row's `problems` lists what is
-    wrong, most severe first. `reported_age_s` is how old the agent's last
+    wrong, most severe first, and `problem_since` / `problem_age_s` say when
-    report is, so a large value means the row may be stale.
+    the worst of them began (null when Steward's retained events cannot say).
+    `reported_age_s` is how old the agent's last report is, so a large value
+    means the row may be stale.
 
     A swarm task's container that Swarm has replaced (a rolling update, or a
     retry after a failure) has `swarm_replaced: true` and no problems: Swarm
@@ -306,14 +346,17 @@ async def get_container(container: str, host: str = "", range: str = "24h", even
         cpu = await docker_svc.container_points(db, c.host_id, c.name, "cpu_pct", since)
         mem = await docker_svc.container_points(db, c.host_id, c.name, "mem_pct", since)
         evs = await docker_svc.container_events(db, c.host_id, c.name, events)
-    recent = sum(1 for e in failures if e.at >= now - timedelta(hours=1))
+        sick = await docker_svc.unhealthy_since(db, [(c.host_id, c.name)])
+    hour_ago = now - timedelta(hours=1)
+    recent = sum(1 for e in failures if e.at >= hour_ago)
     try:
         ports = json.loads(c.ports_json or "[]")
     except ValueError:
         ports = []
     problems = [] if gone else container_problems(c, recent, bool(failures), _crash_loop_deaths())
+    began = problem_since(problems, failures, hour_ago, sick.get((c.host_id, c.name)))
     return {
-        **container_row(c, host_names, problems, now, replaced=gone),
+        **container_row(c, host_names, problems, now, replaced=gone, since=began),
         "container_id": c.container_id,
         "ports": ports,
         "net_rx_bytes": c.net_rx_bytes, "net_tx_bytes": c.net_tx_bytes,
@@ -348,6 +391,10 @@ async def list_swarm_services(limit: int = 100) -> dict:
     - `last_error`: the newest task that failed, was rejected, or carries an
       error while Swarm retries it — `error` is Swarm's own text (e.g. "No
       such image", "task: non-zero exit (1)"), with `exit_code`, `node`, `at`.
+    - `trouble_since` / `trouble_age_s`: when the current trouble began — a
+      crash loop's first failure in the hour, a short service's first failed
+      task in its failing slots, a recovered service's first failure in 24h.
+      Swarm keeps few old tasks, so a long outage reads as a lower bound.
 
     Services with a problem come first. `nodes` lists every swarm node once,
     with its role (manager/worker), availability and status.
@@ -386,6 +433,8 @@ async def list_swarm_services(limit: int = 100) -> dict:
             "crash_looping": s["crash_looping"],
             "failures_1h": s["failures_1h"], "failures_24h": s["failures_24h"],
             "last_error": _error_row(s["last_error"]),
+            "trouble_since": iso(s["trouble_since"]),
+            "trouble_age_s": age_s(s["trouble_since"], now),
             "previous_tasks": s["previous_tasks"], "tasks_known": s["tasks_known"],
             "replicas": [
                 {"ghost": True, "host": r["host"], "count": r["count"]} if r["ghost"] else

plugins/docker/service.py

@@ -6,6 +6,8 @@ Quart's pipeline) call the same code.
 """
 from __future__ import annotations
 
+from datetime import datetime
+
 from sqlalchemy import func, select
 
 from steward.core.time_range import bucket_seconds
@@ -216,6 +218,27 @@ async def recent_failures_by_container(db, since) -> dict[tuple[str, str], list]
     return out
 
 
+async def unhealthy_since(db, keys) -> dict[tuple[str, str], datetime]:
+    """{(host_id, container_name): when it last turned unhealthy} for the given
+    containers, from their health_change events; absent when none is retained."""
+    keys = set(keys)
+    if not keys:
+        return {}
+    rows = (await db.execute(
+        select(DockerEvent)
+        .where(DockerEvent.event == "health_change",
+               DockerEvent.detail.like("%→unhealthy"),
+               DockerEvent.container_name.in_({name for _, name in keys}))
+        .order_by(DockerEvent.at.desc())
+    )).scalars()
+    out: dict[tuple[str, str], datetime] = {}
+    for e in rows:
+        key = (e.host_id, e.container_name)
+        if key in keys and key not in out:
+            out[key] = e.at
+    return out
+
+
 PRUNE_TARGETS = ("containers", "images", "system")
 
 

plugins/docker/swarm_view.py

@@ -120,6 +120,46 @@ def task_summary(tasks: list[dict], now: datetime, crash_loop_deaths: int,
     }
 
 
+def _oldest_failure(tasks: list[dict], since: datetime) -> datetime | None:
+    times = [at for t in tasks if t.get("state") in FAILED_TASK_STATES
+             and (at := task_time(t)) is not None and at >= since]
+    return min(times, default=None)
+
+
+def trouble_since(tasks: list[dict], state: str, crash_looping: bool,
+                  now: datetime) -> datetime | None:
+    """When a service's current trouble began, as far as its retained tasks
+    tell; None when it has none, or the tasks cannot say.
+
+    A crash loop dates from its first failure in the last hour, and a healthy
+    service that lost tasks from its first failure in the last 24h. A service
+    short of replicas dates from its troubled slots (replicated: the slot;
+    global: the node): in each slot whose newest task is not running, the
+    oldest of its newest troubled tasks (failed, rejected, or carrying an
+    error), walking back until a running task interrupts them. Swarm keeps few
+    old tasks, so a long outage reads as the oldest one retained: a lower
+    bound.
+    """
+    if crash_looping:
+        return _oldest_failure(tasks, now - timedelta(hours=1))
+    if state == "healthy":
+        return _oldest_failure(tasks, now - timedelta(hours=24))
+    by_slot: dict = {}
+    for t in tasks:  # newest first, so each slot's list is too
+        by_slot.setdefault(t.get("slot") or t.get("node_id"), []).append(t)
+    began = []
+    for slot_tasks in by_slot.values():
+        oldest = None
+        for t in slot_tasks:
+            if t.get("state") == "running":
+                break
+            if t.get("state") in FAILED_TASK_STATES or t.get("error"):
+                oldest = task_time(t) or oldest
+        if oldest is not None:
+            began.append(oldest)
+    return min(began, default=None)
+
+
 def _slot_key(c) -> str:
     """A task container's slot within its service: "<service>.<slot>.<task>" for
     replicated services, "<service>.<node id>.<task>" for global ones."""
@@ -204,6 +244,7 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d
                       without task data); crash_looping = failures_1h ≥ the
                       Settings → Thresholds crash-loop count
       last_error    = last_task_error(), plus `node` (the node's hostname)
+      trouble_since = trouble_since(): when the current trouble began, or None
     standalone[i] = {host_id, host, containers:[...]}  # non-swarm, by host
     superseded    = the task containers Swarm replaced, so callers can skip them
     """
@@ -282,15 +323,19 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d
         if desired is None:
             desired = len(replicas)
 
+        tasks = tasks_by_svc.get(name, [])
+        state = _service_state(int(running or 0), int(desired or 0))
+        summary = task_summary(tasks, now, crash_loop_deaths, node_hostname)
         out_services.append({
             "name": name,
             "mode": getattr(svc, "mode", "replicated") if svc is not None else "replicated",
             "running": running,
             "desired": desired,
-            "state": _service_state(int(running or 0), int(desired or 0)),
+            "state": state,
             "replicas": replicas,
             "previous_tasks": previous.get(name, 0),
-            **task_summary(tasks_by_svc.get(name, []), now, crash_loop_deaths, node_hostname),
+            **summary,
+            "trouble_since": trouble_since(tasks, state, summary["crash_looping"], now),
         })
 
     # Non-swarm containers grouped by host, running-first order preserved.

tests/integration/test_mcp_plugin_tools.py

@@ -204,6 +204,8 @@ def test_docker_and_agent_tools_over_seeded_rows(app):
         assert (s["failures_1h"], s["failures_24h"], s["crash_looping"]) == (1, 1, False)
         assert s["last_error"]["error"] == "task: non-zero exit (1)"
         assert s["last_error"]["node"] == host_name and s["last_error"]["age_s"] >= 600
+        # Short of replicas since slot 2's task failed, ten minutes ago.
+        assert 600 <= s["trouble_age_s"] < 900
         old = (await dtools.list_containers(host=host_name, status="exited", limit=500))["containers"]
         old = next(c for c in old if c["container_name"] == replaced)
         assert old["swarm_replaced"] is True and old["problems"] == []
@@ -213,11 +215,15 @@ def test_docker_and_agent_tools_over_seeded_rows(app):
 
         # ── overview contributions ─────────────────────────────────────────
         async with app.db_sessionmaker() as db:
-            found = {(f["kind"], f["subject"]) for f in await dtools.mcp_overview(db)}
+            findings = await dtools.mcp_overview(db)
+            found = {(f["kind"], f["subject"]) for f in findings}
             agent_found = {(f["kind"], f["subject"]) for f in await atools.mcp_overview(db)}
         assert ("container_unhealthy", db_) in found
         assert ("container_exited_nonzero", job) in found
         assert ("swarm_service_degraded", svc) in found
+        # Each says how long it has been wrong: the death, the failed task.
+        since = {f["subject"]: f["age_s"] for f in findings}
+        assert 600 <= since[job] < 900 and 600 <= since[svc] < 900
         assert not any(subject == replaced for _, subject in found)
         assert not any(subject == web for _, subject in found)
         assert ("agent_never_reported", quiet_host) in agent_found

tests/plugins/docker/test_mcp_tools.py

@@ -137,3 +137,35 @@ async def test_prune_docker_refuses_an_unknown_target_before_reading_anything(mo
     with pytest.raises(ValueError, match="'volumes' is not one of containers, images, system"):
         await t.prune_docker("nas", "volumes")
     assert inspect.signature(t.prune_docker).parameters["check"].default is True
+
+
+def _death(event, minutes_ago):
+    return NS(event=event, at=NOW - timedelta(minutes=minutes_ago))
+
+
+def test_problem_since_dates_each_problem_from_its_own_evidence():
+    hour_ago = NOW - timedelta(hours=1)
+    deaths = [_death("die", 5), _death("die", 20), _death("die", 50), _death("die", 300)]
+    # A crash loop began with its first death inside the hour, not the older one.
+    assert t.problem_since(["crash_looping"], deaths, hour_ago) == NOW - timedelta(minutes=50)
+    # An error exit dates from the death that ended it.
+    assert t.problem_since(["exited_nonzero"], deaths, hour_ago) == NOW - timedelta(minutes=5)
+    oom = [_death("die", 2), _death("oom", 3)]
+    assert t.problem_since(["oom_killed", "exited_nonzero"], oom, hour_ago) == NOW - timedelta(minutes=3)
+    turned = NOW - timedelta(hours=2)
+    assert t.problem_since(["unhealthy"], [], hour_ago, turned) == turned
+    # No retained evidence: unknown, not now.
+    assert t.problem_since(["unhealthy"], [], hour_ago) is None
+    assert t.problem_since([], deaths, hour_ago) is None
+
+
+def test_findings_carry_when_the_problem_began():
+    began = NOW - timedelta(minutes=40)
+    row = t.container_row(_c(name="db", status="exited", exit_code=1), {}, ["exited_nonzero"],
+                          NOW, since=began)
+    assert row["problem_since"] == "2026-10-06T11:20:00Z" and row["problem_age_s"] == 2400
+    (f,) = t.container_findings([row])
+    assert f["since"] == "2026-10-06T11:20:00Z" and f["age_s"] is not None
+    svc = {"name": "db", "state": "down", "running": 0, "desired": 1, "trouble_since": began}
+    (g,) = t.service_findings([svc])
+    assert g["since"] == "2026-10-06T11:20:00Z"

tests/plugins/docker/test_swarm_view.py

@@ -3,7 +3,9 @@ import json
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 
-from plugins.docker.swarm_view import build_swarm_services, last_task_error, task_failures
+from plugins.docker.swarm_view import (
+    build_swarm_services, last_task_error, task_failures, trouble_since,
+)
 
 NODE_HOSTNAME = {"nA": "docker01", "nB": "docker02", "nC": "docker03"}
 HOST_NAME = {"h1": "docker01", "h2": "docker02"}  # agents run on docker01/02 only
@@ -172,3 +174,40 @@ def test_the_newest_manager_report_wins():
     (svc,) = _view([], [old, new])["services"]
     assert svc["tasks_known"] is True and svc["failures_1h"] == 1
 
+
+
+def _ago(minutes):
+    return NOW - timedelta(minutes=minutes)
+
+
+def test_trouble_since_for_each_kind_of_trouble():
+    loop = [_task("t1", state="failed", ago_min=2), _task("t2", state="failed", ago_min=20),
+            _task("t3", state="failed", ago_min=45), _task("t4", state="failed", ago_min=90)]
+    # A crash loop: its first failure inside the hour.
+    assert trouble_since(loop, "down", True, NOW) == _ago(45)
+    # Down: the oldest of the newest troubled tasks, back to a running one.
+    down = [_task("t1", state="pending", ago_min=1, error="no suitable node"),
+            _task("t2", state="rejected", ago_min=30, error="No such image"),
+            _task("t3", state="running", desired="running", ago_min=600),
+            _task("t4", state="failed", ago_min=900)]
+    assert trouble_since(down, "degraded", False, NOW) == _ago(30)
+    # Another slot's newer, healthy task does not hide the failing one.
+    two = [_task("a2", state="running", desired="running", ago_min=5, slot=1),
+           _task("b1", state="failed", ago_min=8, slot=2),
+           _task("b0", state="failed", ago_min=40, slot=2),
+           _task("a1", state="failed", ago_min=300, slot=1)]
+    assert trouble_since(two, "degraded", False, NOW) == _ago(40)
+    # Healthy again, but a task failed today: dated from that failure.
+    recovered = [_task("t1", state="running", desired="running", ago_min=10),
+                 _task("t2", state="failed", ago_min=60 * 5),
+                 _task("t3", state="failed", ago_min=60 * 30)]
+    assert trouble_since(recovered, "healthy", False, NOW) == _ago(60 * 5)
+    # No task records: unknown.
+    assert trouble_since([], "down", False, NOW) is None
+
+
+def test_the_view_dates_a_down_service():
+    tasks = [_task("t1", state="failed", ago_min=3, exit_code=1),
+             _task("t2", state="failed", ago_min=70, exit_code=1)]
+    (s,) = _view([], [_svc("db", 0, 1, [], tasks=tasks)])["services"]
+    assert s["state"] == "down" and s["trouble_since"] == _ago(70)