fix: Correlate planned Docker stops before reporting crashes

Dockerfile

@@ -11,7 +11,7 @@ RUN pip install --no-cache-dir \
     "cryptography>=42.0"
 
 # Copy agent
-COPY watchdog.py .
+COPY watchdog.py lifecycle.py .
 
 # UID 1000 account for install.sh's non-root hardening path (docker-compose.yaml
 # defaults to root; install.sh instead sets WATCHDOG_UID/GID to run as this user).

README.md

@@ -59,7 +59,11 @@ Alerts are deduplicated via a **cooldown** — once an alert fires for a contain
 
 Two additional deduplication rules suppress redundant alerts at the event level:
 - **OOM → die suppression**: when a container is OOM-killed, Docker emits both an `oom` event and an immediately following `die` event. The watchdog fires the `oom` CRITICAL alert and suppresses the subsequent `crashed` alert to avoid a duplicate for the same incident.
-- **Restart-loop incident coalescing**: Docker `die`/OOM events provide an immediate first failure alert. Only periodic Docker `RestartCount` increments determine the number of automatic restarts and whether the `restart_threshold` has been reached within `restart_window_minutes`; event deliveries are **never** added to the restart total. One HIGH `restart-loop` alert then suppresses per-cycle crashes, OOMs and failed Docker, manual or auto-discovered probes until sustained recovery, regardless of cooldown. Docker IDs identify distinct container instances even when names are reused.
+- **Restart-loop incident coalescing**: Docker `die`/OOM events initiate the first failure alert (nonzero `die` events may be held for the short stop-correlation window). Only periodic Docker `RestartCount` increments determine the number of automatic restarts and whether the `restart_threshold` has been reached within `restart_window_minutes`; event deliveries are **never** added to the restart total. One HIGH `restart-loop` alert then suppresses per-cycle crashes, OOMs and failed Docker, manual or auto-discovered probes until sustained recovery, regardless of cooldown. Docker IDs identify distinct container instances even when names are reused.
+
+**Docker-requested stops and exit 137:** An exit status of 137 means SIGKILL; it does **not** prove an OOM. Docker may deliver a stop event shortly after die, so nonzero exits are briefly held for lifecycle_detection.correlation_window_seconds (default 3 seconds). A matching stop from the same container ID suppresses that instance's misleading CRASH and is logged instead; a direct docker kill, an unconfirmed stop, or a genuine OOM still produces the normal CRITICAL incident. A start event prevents the following run's stop from masking an earlier crash. This correlation only provides evidence of a **Docker-requested stop**, not evidence that stopping the service was authorized. The existing restart-loop counting, recovery rules, and alert delivery remain authoritative. Set lifecycle_detection.enabled to false to disable the correlation.
+
+Planned-stop correlation does not implement an availability policy: long-running services stopped intentionally will not produce a downtime alert unless another monitoring rule detects their outage. Optional expected-running monitoring and HA-specific planned-shutdown handling are tracked separately.
 
 **Recovery alerts**: once a previously-alarmed container is healthy and running, the watchdog sends a one-time INFO `recovered` notification (if `alert_on_recovery: true`). For an auto-restarting container recovering from `crashed`, `oom`, or `restart-loop`, a fleeting `running` state is *not* recovery: Docker's `State.StartedAt` must prove at least `restart_recovery_stability_seconds` continuous uptime (default 120 seconds), with no intervening failure. On recovery, restart history is reset to avoid immediate re-alerting. Docker health events cannot bypass this check. Other container recovery checks preserve their existing behavior.
 
@@ -72,6 +76,7 @@ Two additional deduplication rules suppress redundant alerts at the event level:
 ```
 .
 ├── watchdog.py                  # Main monitor script
+├── lifecycle.py                 # Bounded Docker stop/die evidence correlation
 ├── watchdog-config.yaml         # All configuration (no secrets)
 ├── docker-compose.yaml          # Production / offline runtime (image only)
 ├── docker-compose.build.yaml    # Developer / online build

lifecycle.py

@@ -0,0 +1,86 @@
+"""Short-lived Docker stop/die evidence, not another alert-state machine.
+
+Docker can deliver the final stop event just after die. Buffer only unexpected
+nonzero exits for a bounded time; when no stop evidence arrives, report the
+failure through Watchdog's existing incident engine.
+"""
+from dataclasses import dataclass
+import threading
+import time
+
+
+@dataclass
+class PendingDeath:
+    container_id: str
+    name: str
+    container: object
+    exit_code: int
+    event_time: float | None
+    happened_at: float
+    stopped: bool = False
+    oom: bool = False
+    new_run_started: bool = False
+
+
+class DockerLifecycleCorrelator:
+    def __init__(self, window_seconds: float = 3.0):
+        self.window = float(window_seconds)
+        self._lock = threading.Lock()
+        self._pending: list[PendingDeath] = []
+        self._recent_stops: dict[str, float] = {}
+
+    def record(self, action: str, container_id: str, *, name: str = "",
+               container: object = None, exit_code: int = 1,
+               event_time: float | None = None, now: float | None = None) -> bool:
+        """Return whether an OOM event matched an already pending death."""
+        if not container_id:
+            return False
+        now = time.monotonic() if now is None else now
+        with self._lock:
+            self._recent_stops = {
+                cid: when for cid, when in self._recent_stops.items()
+                if 0 <= now - when <= self.window
+            }
+            if action == "start":
+                self._recent_stops.pop(container_id, None)
+                # A later run's stop cannot excuse a prior run's failure.
+                for death in self._pending:
+                    if death.container_id == container_id:
+                        death.new_run_started = True
+            elif action == "stop":
+                self._recent_stops[container_id] = now
+                for death in reversed(self._pending):
+                    if (death.container_id == container_id and not death.new_run_started
+                            and 0 <= now - death.happened_at <= self.window):
+                        death.stopped = True
+                        break  # One stop must not mask several restart-cycle deaths.
+            elif action == "oom":
+                matched = False
+                for death in reversed(self._pending):
+                    if (death.container_id == container_id and not death.new_run_started
+                            and 0 <= now - death.happened_at <= self.window):
+                        death.oom = True
+                        matched = True
+                        break
+                return matched
+            elif action == "die":
+                stopped_at = self._recent_stops.get(container_id)
+                stopped = stopped_at is not None and 0 <= now - stopped_at <= self.window
+                self._pending.append(PendingDeath(
+                    container_id, name, container, exit_code, event_time, now,
+                    stopped=stopped,
+                ))
+        return False
+
+    def drain_ready(self, *, now: float | None = None, force: bool = False) -> list[PendingDeath]:
+        now = time.monotonic() if now is None else now
+        with self._lock:
+            ready, waiting = [], []
+            for death in self._pending:
+                if (force or death.stopped or death.oom or death.new_run_started
+                        or now - death.happened_at >= self.window):
+                    ready.append(death)
+                else:
+                    waiting.append(death)
+            self._pending = waiting
+            return ready

tests/test_event_missing.py

@@ -12,6 +12,7 @@ def make_subject(self):
         monitor = object.__new__(watchdog.Watchdog)
         monitor._excluded = set()
         monitor._missing_events = {}
+        monitor._lifecycle_enabled = False
         monitor.client = SimpleNamespace(containers=Mock())
         monitor.client.containers.get.side_effect = docker.errors.NotFound("removed")
         return monitor

tests/test_lifecycle.py

@@ -0,0 +1,254 @@
+"""Integration regression: Docker stop evidence feeds ONE PR #5 incident engine."""
+import threading
+import time
+import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from lifecycle import DockerLifecycleCorrelator
+import watchdog
+
+CID = "a" * 64
+OTHER = "b" * 64
+
+
+class CorrelationTests(unittest.TestCase):
+    def test_die_before_stop(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("die", CID, name="svc", exit_code=137, now=10)
+        tracker.record("stop", CID, now=10.1)
+        [death] = tracker.drain_ready(now=10.2)
+        self.assertTrue(death.stopped)
+
+    def test_stop_before_die(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("stop", CID, now=10)
+        tracker.record("die", CID, name="svc", exit_code=137, now=10.1)
+        [death] = tracker.drain_ready(now=10.2)
+        self.assertTrue(death.stopped)
+
+    def test_late_stop_not_suppressed(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("die", CID, name="svc", exit_code=137, now=10)
+        tracker.record("stop", CID, now=14)
+        [death] = tracker.drain_ready(now=14)
+        self.assertFalse(death.stopped)
+
+    def test_wrong_id_not_suppressed(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("die", CID, name="svc", exit_code=137, now=10)
+        tracker.record("stop", OTHER, now=10.1)
+        [death] = tracker.drain_ready(now=14)
+        self.assertFalse(death.stopped)
+
+    def test_start_clears_old_stop(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("stop", CID, now=10)
+        tracker.record("start", CID, now=10.1)
+        tracker.record("die", CID, name="svc", exit_code=137, now=10.2)
+        [death] = tracker.drain_ready(now=14)
+        self.assertFalse(death.stopped)
+
+    def test_next_run_stop_cannot_hide_previous_run_crash(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("die", CID, name="svc", exit_code=1, now=10)
+        tracker.record("start", CID, now=10.1)
+        tracker.record("stop", CID, now=10.2)
+        [death] = tracker.drain_ready(now=10.3)
+        self.assertFalse(death.stopped)
+
+    def test_single_stop_can_suppress_only_latest_die(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("die", CID, name="svc", exit_code=2, now=10)
+        tracker.record("die", CID, name="svc", exit_code=137, now=10.2)
+        tracker.record("stop", CID, now=10.3)
+        deaths = tracker.drain_ready(force=True)
+        self.assertEqual([d.stopped for d in deaths], [False, True])
+
+    def test_oom_after_die_is_correlated(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("die", CID, name="svc", exit_code=137, now=10)
+        self.assertTrue(tracker.record("oom", CID, now=10.1))
+        [death] = tracker.drain_ready(now=10.2)
+        self.assertTrue(death.oom)
+
+    def test_oom_without_pending_die_is_not_matched(self):
+        tracker = DockerLifecycleCorrelator(3)
+        self.assertFalse(tracker.record("oom", CID, now=10))
+
+    def test_no_premature_resolution(self):
+        tracker = DockerLifecycleCorrelator(3)
+        tracker.record("die", CID, name="svc", exit_code=1, now=10)
+        self.assertEqual(tracker.drain_ready(now=12.99), [])
+        self.assertEqual(len(tracker.drain_ready(now=13.01)), 1)
+
+
+class FakeContainer:
+    def __init__(self, cid=CID, name="svc"):
+        self.id, self.name = cid, name
+        self.status = "running"
+        self.labels = {}
+        self.attrs = {
+            "Config": {"Image": "test/svc", "Labels": {}},
+            "HostConfig": {"AutoRemove": False, "RestartPolicy": {"Name": "always"}},
+            "State": {"OOMKilled": False,
+                      "StartedAt": (datetime.now(timezone.utc)-timedelta(seconds=5)).isoformat()},
+            "Mounts": [], "RestartCount": 0,
+            "NetworkSettings": {"Networks": {}},
+        }
+        self.image = SimpleNamespace(tags=["test/svc:latest"])
+
+    def logs(self, **kwargs):
+        return b"failure in test container"
+
+    def stats(self, **kwargs):
+        return {"memory_stats": {}}
+
+    def reload(self):
+        pass
+
+
+class WatchdogLifecycleTests(unittest.TestCase):
+    def setUp(self):
+        self.container = FakeContainer()
+        wd = object.__new__(watchdog.Watchdog)
+        wd.cfg = {"alert_on_recovery": True}
+        wd.state = watchdog.WatchdogState("")
+        wd.host = "lab"
+        wd.runbook_base = ""
+        wd._excluded = set()
+        wd._ignored_events = []
+        wd._discovered_checks = {}
+        wd._oom_containers = set()
+        wd._missing_events = {}
+        wd._lifecycle_enabled = True
+        wd._lifecycle = DockerLifecycleCorrelator(3)
+        wd._lifecycle_wake = threading.Event()
+        wd._restart_threshold = 3
+        wd._restart_window_secs = 600
+        wd._restart_recovery_stability_secs = 120
+        wd._unhealthy_threshold = 3
+        wd._cooldown_minutes = 5
+        wd._auto_cfg = {"enabled": False}
+        wd._manual_checks = {}
+        wd._boot_time = 0
+        wd._grace_until = 0
+        wd._folded = []
+        wd._folded_lock = threading.Lock()
+        wd.client = SimpleNamespace(
+            containers=SimpleNamespace(get=lambda cid: self.container,
+                                       list=lambda all=True: [self.container]))
+        self.alerts = []
+        wd._dispatch = self.alerts.append
+        self.wd = wd
+
+    def event(self, action, code=None, cid=CID, timeNano=None):
+        attrs = {"name": self.container.name}
+        if code is not None:
+            attrs["exitCode"] = str(code)
+        event = {"Action": action, "Actor": {"ID": cid, "Attributes": attrs}}
+        if timeNano:
+            event["timeNano"] = timeNano
+        return event
+
+    def resolve(self):
+        for death in self.wd._lifecycle.drain_ready(force=True):
+            self.wd._resolve_lifecycle_death(death)
+
+    def reasons(self):
+        return [alert.failure_type for alert in self.alerts]
+
+    def test_october_forced_stop_does_not_crash(self):
+        self.wd._handle_event(self.event("die", 137))
+        self.wd._handle_event(self.event("stop"))
+        self.resolve()
+        self.assertEqual(self.reasons(), [])
+        self.assertEqual(self.wd.state.get("svc").alerted_for, "")
+
+    def test_stop_first_suppresses_forced_exit(self):
+        self.wd._handle_event(self.event("stop"))
+        self.wd._handle_event(self.event("die", 137))
+        self.resolve()
+        self.assertEqual(self.reasons(), [])
+
+    def test_direct_kill_without_stop_is_a_crash(self):
+        self.wd._handle_event(self.event("die", 137))
+        self.resolve()
+        self.assertEqual(self.reasons(), ["crashed"])
+        self.assertNotIn("likely OOM", self.alerts[0].recommended_action)
+
+    def test_nonzero_exit_without_stop_is_a_crash(self):
+        self.wd._handle_event(self.event("die", 2))
+        self.resolve()
+        self.assertEqual(self.reasons(), ["crashed"])
+
+    def test_clean_exit_is_not_a_crash(self):
+        self.wd._handle_event(self.event("die", 0))
+        self.resolve()
+        self.assertEqual(self.reasons(), [])
+
+    def test_oom_before_die_no_duplicate(self):
+        self.wd._handle_event(self.event("oom"))
+        self.wd._handle_event(self.event("die", 137))
+        self.resolve()
+        self.assertEqual(self.reasons(), ["oom"])
+
+    def test_oom_after_die_no_duplicate_or_stale_marker(self):
+        self.wd._handle_event(self.event("die", 137))
+        self.wd._handle_event(self.event("oom"))
+        self.resolve()
+        self.assertEqual(self.reasons(), ["oom"])
+        self.assertEqual(self.wd._oom_containers, set())
+
+    def test_pending_die_does_not_erase_active_restart_loop(self):
+        state = self.wd.state.get("svc")
+        state.alerted_for = "restart-loop"
+        self.wd._handle_event(self.event("die", 2))
+        self.resolve()
+        self.assertEqual(self.reasons(), [])
+        self.assertEqual(state.alerted_for, "restart-loop")
+
+    def test_multiple_pending_dies_still_one_crash(self):
+        for _ in range(4):
+            self.wd._handle_event(self.event("die", 2))
+        self.resolve()
+        self.assertEqual(self.reasons(), ["crashed"])
+
+    def test_pending_death_from_replaced_container_is_ignored(self):
+        self.wd._handle_event(self.event("die", 137))
+        new = FakeContainer(cid=OTHER)
+        self.wd._sync_container_identity(self.wd.state.get("svc"), new)
+        self.resolve()
+        self.assertEqual(self.reasons(), [])
+
+    def test_delayed_old_event_after_recovery_is_ignored(self):
+        state = self.wd.state.get("svc")
+        state.last_recovery_time = time.time()
+        self.wd._handle_event(self.event("die", 2, timeNano=int((time.time()-5)*1e9)))
+        self.resolve()
+        self.assertEqual(self.reasons(), [])
+
+    def test_stop_evidence_skips_container_lookup(self):
+        self.wd.client.containers.get = Mock(side_effect=AssertionError("unexpected inspect"))
+        self.wd._handle_event(self.event("stop"))
+        self.wd.client.containers.get.assert_not_called()
+
+    def test_lifecycle_can_be_disabled(self):
+        self.wd._lifecycle_enabled = False
+        self.wd._handle_event(self.event("die", 2))
+        self.assertEqual(self.reasons(), ["crashed"])
+
+    def test_event_subscription_does_not_need_kill(self):
+        events = []
+        self.wd._stop_event = threading.Event()
+        self.wd.client.events = lambda **kw: events.append(kw) or iter([])
+        self.wd._event_listener()
+        filters = events[0]["filters"]["event"]
+        self.assertIn("start", filters)
+        self.assertIn("stop", filters)
+        self.assertNotIn("kill", filters)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_restart_loop.py

@@ -48,6 +48,7 @@ def make_watchdog(container, threshold=3, stable=120):
     wd._discovered_checks = {}
     wd._oom_containers = set()
     wd._missing_events = {}  # PR #6 event cache (normal Watchdog init owns it)
+    wd._lifecycle_enabled = False
     wd._auto_cfg = {"enabled": False}
     wd._manual_checks = {}
     wd._boot_time = 1