fix(docker): a container that crashes between two samples leaves a die event and its last log lines (#5049)

Agent 1.11.0 sends finished_at (State.FinishedAt) and reads a stopped
container's logs once per stop: since its cursor if it was being read, or
its last 200 lines when first seen already crashed (non-zero exit, stopped
in the last 10 min). Cursors and finals are forgotten once a container is no
longer listed.

Ingest gives a container first seen already exited non-zero a die event, and
dates die/stop events by finished_at when the agent sends it.

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

plugins/docker/ingest.py

@@ -40,6 +40,8 @@ def _derive_events(old_state: dict, new_containers: list) -> list:
 
       start         — a container transitions into running (or a genuinely new
                       container appears already running)
+      die           — also for a new container first seen already exited
+                      non-zero: a task that crashed between two samples (#5049)
       restart       — Docker restarted it (its restart count went up); said
                       instead of a start, since that is what happened
       stop          — running → not-running, or a running container is removed
@@ -64,10 +66,12 @@ def _derive_events(old_state: dict, new_containers: list) -> list:
         old = old_state.get(name)
 
         if old is None:
-            # Newly observed container — only a start is meaningful (we have no
-            # prior state to diff a death against).
+            # Newly observed container: a start, or, when it is already down
+            # with a non-zero exit, the death that happened between samples.
             if new_running:
                 events.append((name, "start", c.get("image") or None))
+            elif new_status in _STOPPED_STATES and new_exit not in (None, 0):
+                events.append((name, "die", f"exit {new_exit}"))
             continue
 
         old_running = (old.get("status") or "").lower() == "running"
@@ -365,10 +369,15 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
         for r in old_rows
     }
     if old_state:
+        # A stop is dated when Docker says it happened (agent ≥ 1.11), not when
+        # the sample that noticed it was taken.
+        finished = {c.get("name"): _parse_started_at(c.get("finished_at"))
+                    for c in latest_containers}
         for name, event, detail in _derive_events(old_state, latest_containers):
+            at = finished.get(name) if event in ("die", "stop") else None
             session.add(DockerEvent(
                 host_id=host.id, container_name=name,
-                event=event, detail=detail, at=latest_at,
+                event=event, detail=detail, at=at or latest_at,
             ))
     await mark("events")
 

plugins/host_agent/agent.py

@@ -20,7 +20,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.10.0"
+AGENT_VERSION = "1.11.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
@@ -456,6 +456,8 @@ DOCKER_DF_INTERVAL_SECONDS = 600.0
 # kept per container in the agent's rate-state.
 DOCKER_LOG_BATCH_MAX_BYTES = 262144   # 256 KiB of log text per push, all containers
 DOCKER_LOG_FIRST_TAIL = 50            # lines seeded on a container's first interval
+DOCKER_LOG_FINAL_TAIL = 200           # lines read once from a container found already crashed
+DOCKER_LOG_FINAL_WINDOW_SECONDS = 600  # ... when it stopped this recently
 
 
 def _dechunk(body: bytes) -> bytes:
@@ -599,12 +601,12 @@ def _docker_inspect_fields(insp: dict):
     return health, restart_count, exit_code, oom_killed
 
 
-def _docker_started_at(insp: dict):
+def _docker_started_at(insp: dict, key: str = "StartedAt"):
     """State.StartedAt as ISO-8601, or None for a container never started
     (Docker's zero time) or an inspect that failed. When the container last
     started, not when it was created: a restarted container's uptime counts
-    from its restart."""
-    raw = (((insp or {}).get("State") or {}).get("StartedAt") or "")
+    from its restart. key="FinishedAt" reads when it last stopped."""
+    raw = (((insp or {}).get("State") or {}).get(key) or "")
     if not raw or raw.startswith("0001-"):
         return None
     ts = _parse_log_ts(raw)
@@ -665,6 +667,9 @@ def _collect_one_container(socket_path: str, c: dict) -> dict:
             pass
 
     started_at = _docker_started_at(insp)
+    # When it last stopped (a running container's is its previous stop, or
+    # None): dates a die event, and tells the log collector a stop is new.
+    finished_at = None if state == "running" else _docker_started_at(insp, "FinishedAt")
     record = {
         "name": name,
         "container_id": cid[:12],
@@ -676,6 +681,7 @@ def _collect_one_container(socket_path: str, c: dict) -> dict:
         "mem_pct": mem_pct,
         "ports": _docker_ports(c.get("Ports", [])),
         "started_at": started_at,
+        "finished_at": finished_at,
         "health": health,
         "restart_count": restart_count,
         "exit_code": exit_code,
@@ -872,39 +878,64 @@ def _parse_container_logs(raw: bytes) -> list:
 
 def collect_docker_logs(socket_path: str, containers: list, state: dict,
                         exclude=None,
-                        max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES) -> list:
-    """New container-log lines since the last interval, for running containers.
+                        max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES,
+                        now: float | None = None) -> list:
+    """New container-log lines since the last interval.
 
     Keeps a per-container since-cursor in `state["docker_log_cursors"]`
     (name → unix ts) so each interval fetches only what's new; a container's
-    first interval seeds from a short tail. The whole batch is capped at
-    `max_bytes` of log text — once exceeded a truncation marker is appended and
-    collection stops (logs must never balloon a push). `exclude` container names
-    are skipped entirely (a per-host bandwidth opt-out). Best-effort: a
-    per-container transport error just contributes nothing this interval.
+    first interval seeds from a short tail. A running container is read every
+    interval. A stopped one is read once per stop (`finished_at`, remembered in
+    `state["docker_log_finals"]`), so the lines it wrote on its way down reach
+    Steward (#5049): since its cursor if it was being read, or, for one first
+    seen already stopped with a non-zero exit in the last
+    DOCKER_LOG_FINAL_WINDOW_SECONDS, its last DOCKER_LOG_FINAL_TAIL lines. That
+    is how a task that crashes between two samples is explained by its own
+    output. Both maps forget a container once it is no longer listed.
+
+    The whole batch is capped at `max_bytes` of log text — once exceeded a
+    truncation marker is appended and collection stops (logs must never balloon
+    a push). `exclude` container names are skipped entirely (a per-host
+    bandwidth opt-out). Best-effort: a per-container transport error just
+    contributes nothing this interval.
     Returns [{"container", "stream", "ts", "line"}, …].
     """
     exclude_set = set(exclude or ())
     cursors = state.setdefault("docker_log_cursors", {})
-    running = {c.get("name") for c in (containers or [])
-               if c.get("name") and c.get("status") == "running"}
-    # Forget cursors for containers no longer running so state can't grow without
-    # bound over the agent's lifetime.
-    for gone in [k for k in cursors if k not in running]:
-        del cursors[gone]
+    finals = state.setdefault("docker_log_finals", {})
+    now = time.time() if now is None else now
+    listed = {c.get("name"): c for c in (containers or []) if c.get("name")}
+    # Forget containers no longer listed so state can't grow without bound
+    # over the agent's lifetime.
+    for book in (cursors, finals):
+        for gone in [k for k in book if k not in listed]:
+            del book[gone]
 
     out: list = []
     total = 0
     truncated = False
-    for c in (containers or []):
-        name = c.get("name")
-        if not name or name not in running or name in exclude_set:
+    for name, c in listed.items():
+        if name in exclude_set:
             continue
         prev = cursors.get(name)
         base = f"/containers/{name}/logs?stdout=1&stderr=1&timestamps=1"
-        # since accepts a fractional Unix ts; we still dedup the boundary line.
-        path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None
-                else f"{base}&since={prev:.9f}")
+        if c.get("status") == "running":
+            # since accepts a fractional Unix ts; we still dedup the boundary line.
+            path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None
+                    else f"{base}&since={prev:.9f}")
+        else:
+            stopped = c.get("finished_at")
+            if not stopped or finals.get(name) == stopped:
+                continue
+            if prev is not None:
+                path = f"{base}&since={prev:.9f}"
+            else:
+                at = _parse_log_ts(stopped)
+                if (c.get("exit_code") in (None, 0) or at is None
+                        or now - at.timestamp() > DOCKER_LOG_FINAL_WINDOW_SECONDS):
+                    continue
+                path = f"{base}&tail={DOCKER_LOG_FINAL_TAIL}"
+            finals[name] = stopped
         try:
             raw = _docker_request_raw(socket_path, path)
         except (OSError, ValueError):

tests/plugins/docker/test_ingest_events.py

@@ -43,6 +43,22 @@ def test_die_on_nonzero_exit():
     assert events == [("job", "die", "exit 137")]
 
 
+@pytest.mark.parametrize("status,exit_code,events", [
+    ("exited", 1, [("codex_db.1.x", "die", "exit 1")]),   # crashed between samples
+    ("dead", 137, [("codex_db.1.x", "die", "exit 137")]),
+    ("exited", 0, []),                                     # a one-shot that finished
+    ("exited", None, []),                                  # inspect failed: no claim
+    ("created", 1, []),                                    # never ran
+])
+def test_new_container_first_seen_down(status, exit_code, events):
+    # #5049: a task that starts and exits between two samples is first seen
+    # already down; a non-zero exit is the death the samples missed.
+    old = {"other": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
+    got = _derive_events(old, [_c("other", "running"),
+                               _c("codex_db.1.x", status, exit_code=exit_code)])
+    assert got == events
+
+
 def test_oom_on_flip_to_true():
     old = {"hog": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
     # OOM-killed → also no longer running with a non-zero exit, so die + oom both fire.

tests/plugins/host_agent/test_agent_docker.py

@@ -634,6 +634,75 @@ def test_collect_docker_logs_forgets_gone_container_cursors(monkeypatch):
     assert "db" in state["docker_log_cursors"]
 
 
+def _final_raw(seen):
+    def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
+        seen.append(path)
+        return (_frame(1, b"2026-10-06T12:00:00.000000000Z booting\n")
+                + _frame(2, b"2026-10-06T12:00:01.000000000Z fatal: no config\n"))
+    return fake_raw
+
+
+def test_collect_docker_logs_reads_a_stopped_container_once(monkeypatch):
+    # #5049: a container that was being read and has stopped is read one last
+    # time since its cursor, then not again until it stops again.
+    seen: list = []
+    monkeypatch.setattr(a, "_docker_request_raw", _final_raw(seen))
+    state = {"docker_log_cursors": {"web": 1791288000.5}}   # 2026-10-06T12:00:00.5Z
+    down = [{"name": "web", "status": "exited", "exit_code": 1,
+             "finished_at": "2026-10-06T12:00:02+00:00"}]
+    out = a.collect_docker_logs("/sock", down, state)
+    assert [r["line"] for r in out] == ["fatal: no config"]   # since the cursor
+    assert "since=" in seen[0]
+    assert a.collect_docker_logs("/sock", down, state) == []  # same stop: not again
+    assert len(seen) == 1
+    again = [{**down[0], "finished_at": "2026-10-06T12:05:00+00:00"}]
+    a.collect_docker_logs("/sock", again, state)               # a new stop is read
+    assert len(seen) == 2
+
+
+@pytest.mark.parametrize("exit_code,stopped_ago,read", [
+    (1, 30, True),        # crashed between two samples: its last lines
+    (0, 30, False),       # finished cleanly: nothing to explain
+    (None, 30, False),    # inspect failed: no claim
+    (1, 3600, False),     # an old crash already on the host when the agent came up
+])
+def test_collect_docker_logs_tails_a_container_first_seen_crashed(monkeypatch, exit_code,
+                                                                   stopped_ago, read):
+    seen: list = []
+    monkeypatch.setattr(a, "_docker_request_raw", _final_raw(seen))
+    stopped = 1791288000.0
+    down = [{"name": "codex_db.1.x", "status": "exited", "exit_code": exit_code,
+             "finished_at": "2026-10-06T12:00:00+00:00"}]
+    out = a.collect_docker_logs("/sock", down, {}, now=stopped + stopped_ago)
+    assert bool(out) is read
+    if read:
+        assert f"tail={a.DOCKER_LOG_FINAL_TAIL}" in seen[0]
+        assert out[-1]["line"] == "fatal: no config"
+
+
+def test_collect_docker_logs_forgets_finals_of_unlisted_containers(monkeypatch):
+    seen: list = []
+    monkeypatch.setattr(a, "_docker_request_raw", _final_raw(seen))
+    state: dict = {}
+    down = [{"name": "job", "status": "exited", "exit_code": 2,
+             "finished_at": "2026-10-06T12:00:00+00:00"}]
+    a.collect_docker_logs("/sock", down, state, now=1791288010.0)
+    assert "job" in state["docker_log_finals"]
+    a.collect_docker_logs("/sock", [], state, now=1791288020.0)
+    assert state["docker_log_finals"] == {} and state["docker_log_cursors"] == {}
+
+
+def test_finished_at_only_for_a_stopped_container(monkeypatch):
+    insp = {"State": {"StartedAt": "2026-10-06T11:00:00Z", "FinishedAt": "2026-10-06T12:00:00Z",
+                      "ExitCode": 1}}
+    monkeypatch.setattr(a, "_docker_request", lambda s, p, timeout=a.DOCKER_API_TIMEOUT: insp)
+    rec = a._collect_one_container("/sock", {"Id": "abc", "Names": ["/job"], "State": "exited"})
+    assert rec["finished_at"].startswith("2026-10-06T12:00:00")
+    insp["State"]["FinishedAt"] = "0001-01-01T00:00:00Z"                # never stopped
+    rec = a._collect_one_container("/sock", {"Id": "abc", "Names": ["/job"], "State": "exited"})
+    assert rec["finished_at"] is None
+
+
 def test_build_sample_includes_logs_when_present(monkeypatch):
     monkeypatch.setattr(a, "collect_docker",
                         lambda _s: ("ok", None, [{"name": "web", "status": "running"}]))