fix(hands): watchdog seen moves only on a new peer beat, tracked by message id

beaver_agent/hands/heartbeat.py

@@ -20,9 +20,12 @@
 «что нового с курсора»: курсор отдаёт строку соседа один раз, и следующий
 прогон на живой комнате увидел бы пустое окно.
 
-Опора тревоги - `seen`, то есть когда Я увидел строку: `at=` внутри строки
+Опора тревоги - `seen`, то есть когда Я ВПЕРВЫЕ увидел строку: `at=` внутри
-это показание соседа о себе, и остановившиеся у него часы не должны
+строки это показание соседа о себе, и остановившиеся у него часы не должны
-выглядеть как живая цепь.
+выглядеть как живая цепь. Окно - это последние сообщения комнаты, поэтому
+одна и та же отметка видна в нём много прогонов подряд, и новой её делает
+новый id сообщения, а не текст: двигать `seen` на старой отметке значит
+никогда не поднять тревогу в тихой комнате.
 """
 
 from __future__ import annotations
@@ -264,6 +267,10 @@ class Watch:
     """Последний период, который сосед назвал читаемым числом."""
     every_unknown: bool = False
     """В последней отметке стояло `every=unknown`: порог взят с запасом назад."""
+    beat_id: int = 0
+    """id сообщения с последней зачтённой отметкой: по нему видно, что отметка
+    в окне новая, а не та же. Нуль - поля в файле не было (состояние старое)
+    или отметок не видели ни разу."""
     top: int = 0
     """Максимальный id, который мы видели. Поехал назад - окно смотрит в прошлое."""
     alerted: datetime | None = None
@@ -285,6 +292,7 @@ class Watch:
             seen=_parse_iso(raw.get("seen")),
             every=timedelta(seconds=every) if every else None,
             every_unknown=bool(raw.get("every_unknown")),
+            beat_id=int(raw.get("beat_id", 0)),
             top=int(raw.get("top", 0)),
             alerted=_parse_iso(raw.get("alerted")),
             unarmed_since=_parse_iso(raw.get("unarmed_since")),
@@ -298,6 +306,7 @@ class Watch:
             "seen": _iso(self.seen),
             "every": int(self.every.total_seconds()) if self.every else None,
             "every_unknown": self.every_unknown,
+            "beat_id": self.beat_id,
             "top": self.top,
             "alerted": _iso(self.alerted),
             "unarmed_since": _iso(self.unarmed_since),
@@ -327,6 +336,22 @@ class Report:
     """Строка для лога: почему прогон решил именно так."""
 
 
+def _is_new_beat(state: Watch, eid: int, beat: Beat) -> bool:
+    """Впервые ли мы видим ЭТУ отметку соседа.
+
+    Различитель - id сообщения: `at=` внутри строки это показание соседа о
+    себе, и повторить его он может как угодно. Нуля в `beat_id` два повода:
+    отметок не видели ни разу (тогда любая - новая) и состояние записано до
+    появления поля - там единственное, чем можно сверить отметку со зачтённой,
+    это её `at`, и одного прогона хватает, чтобы запомнить id.
+    """
+    if state.beat_id:
+        return eid > state.beat_id
+    if state.seen is None:
+        return True
+    return beat.at != state.at
+
+
 def look(
     state: Watch,
     events: Sequence[Event],
@@ -350,8 +375,8 @@ def look(
     top = max((e.id for e in events), default=None)
     stale = top is not None and top < state.top
     beats = [
-        beat
+        (event.id, beat)
-        for beat in (parse_beat(e.text) for e in events)
+        for event, beat in ((event, parse_beat(event.text)) for event in events)
         if beat is not None and beat.nick == peer
     ]
     heard = bool(beats) or any(e.nick == peer for e in events)
@@ -360,9 +385,13 @@ def look(
     else:
         if top is not None:
             state.top = max(state.top, top)
+        again = False
         if beats:
-            last = beats[-1]
+            eid, last = beats[-1]
-            state.at, state.seen = last.at, now
+            again = not _is_new_beat(state, eid, last)
+            if not again:
+                state.seen = now
+            state.at, state.beat_id = last.at, eid
             state.every_unknown = last.every is None
             if last.every is not None:
                 state.every = last.every
@@ -371,6 +400,8 @@ def look(
             f"отметок соседа в окне: {len(beats)} "
             f"(сообщений комнаты: {len(events)} из {window}, {read})"
         )
+        if again:
+            note += "; отметка та же, что в прошлый прогон - seen не двигаем"
 
     armed = state.seen is not None
     threshold = state.every * FACTOR if state.every is not None else None

tests/test_heartbeat.py

@@ -1,3 +1,4 @@
+import json
 import logging
 from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
@@ -113,6 +114,59 @@ def test_fresh_beat_arms_and_stays_quiet(tmp_path):
     assert w.seen == NOW and w.top == 100
 
 
+def test_same_beat_seen_twice_does_not_move_seen(tmp_path):
+    w = state(tmp_path)
+    old = hb(NOW - timedelta(hours=1), eid=200)
+    first = look(w, [old], now=NOW, res=None, full=True)
+    assert first.armed and first.alarm is None
+    assert w.seen == NOW and w.beat_id == 200
+
+    # Тот же самый hb в окне следующего прогона - это не новое показание
+    again = look(w, [old], now=NOW + timedelta(hours=6), res=None, full=True)
+    assert w.seen == NOW and again.alarm is None
+    assert "отметка та же" in again.note
+
+    # И через порог 36h та же отметка уже не выглядит живой цепью
+    late = NOW + timedelta(hours=37)
+    loud = look(w, [old], now=late, res=None, full=True)
+    assert loud.alarm == "silence" and w.seen == NOW
+    assert loud.age is not None and loud.age > timedelta(hours=36)
+
+    # Пришла новая отметка - новый id, и seen снова двигается
+    fresh = look(w, [old, hb(late, eid=201)], now=late, res=None, full=True)
+    assert fresh.alarm is None and w.seen == late and w.beat_id == 201
+
+
+def test_state_without_beat_id_trusts_its_own_seen(tmp_path):
+    path = tmp_path / "сторож.json"
+    path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "at": "2026-09-21T05:00:01Z",
+                "seen": "2026-09-22T05:42:01Z",
+                "every": 86400,
+                "every_unknown": False,
+                "top": 300,
+                "alerted": None,
+                "unarmed_since": None,
+                "unarmed_told": False,
+            }
+        ),
+        encoding="utf-8",
+    )
+    w = Watch.load(path)
+    was = datetime(2026, 9, 22, 5, 42, 1, tzinfo=UTC)
+    assert w.beat_id == 0 and w.seen == was
+
+    old = hb(datetime(2026, 9, 21, 5, 0, 1, tzinfo=UTC), eid=305)
+    assert look(w, [old], now=NOW, res=None, full=True).alarm is None
+    assert w.seen == was and w.beat_id == 305
+    # Старое состояние не пере-вооружилось на старой отметке: порог придёт в срок
+    over = datetime(2026, 9, 23, 20, 0, tzinfo=UTC)
+    assert look(w, [old], now=over, res=None, full=True).alarm == "silence"
+
+
 def test_threshold_is_every_times_one_and_a_half(tmp_path):
     # 30 часов при every=24h - ещё не тревога: одна потерянная строка не в счёт
     quiet = state(tmp_path, seen=NOW - timedelta(hours=30), every=timedelta(hours=24))
@@ -265,6 +319,7 @@ def test_state_survives_a_round_trip(tmp_path):
         seen=NOW,
         every=timedelta(hours=24),
         every_unknown=True,
+        beat_id=329,
         top=42,
         alerted=NOW,
         unarmed_since=None,
@@ -273,6 +328,7 @@ def test_state_survives_a_round_trip(tmp_path):
     w.save()
     back = Watch.load(w.path)
     assert back.at == NOW and back.seen == NOW and back.top == 42
+    assert back.beat_id == 329
     assert back.every == timedelta(hours=24) and back.every_unknown
     assert back.alerted == NOW and back.unarmed_told
     assert Watch.load(tmp_path / "нет.json").seen is None