fix(observer): checkpoint by timestamp not lexical path — lexically-smaller-but-newer events were silently skipped forever (poisoned node)

Per-node checkpoint now stores the last-processed event TIMESTAMP (int epoch)
instead of a file path compared lexically. A file is "new" iff its timestamp
(parsed from evt-<node>-<unixts>-<type>-<svc>.json, mtime fallback) exceeds the
node's checkpoint; processing is ordered by timestamp, not path.

Root cause (PIHA dead ~34d, 2026-07-12): a stray evt-unknown-<ts>-… file landed
in events/piha/, lexically greater than every evt-piha-… name. The lexical
checkpoint pinned there, so every genuinely newer piha event sorted "before" it
and was skipped forever. Event backlog grew to 7344 files, last_seen frozen,
shadow-read logged false SHADOW_LIVENESS_MISMATCH event=dead prom=up.

- _event_ts_from_path: filename epoch, mtime fallback; NEVER returns 0 for an
  existing file (0 == "older than checkpoint" == the poison).
- _checkpoint_ts_from_value: graceful migration of pre-fix path-string
  checkpoints (and the older last_processed_file format) to int epochs;
  unparseable → 0 (reprocess all — safe, process_event is idempotent on
  last_seen/world_state; bias to reprocess, never to skip).
- Preserved: quarantine of bad events, observer-source re-ingest guard.
- Regression tests (test_incident_lifecycle.py section 9): lexically-smaller-
  but-newer processed, unparseable name falls back to mtime (not wedged),
  ts-not-path ordering, both checkpoint-format migrations, helper units.

Separate bug filed in backlog (not fixed here): ha-diag-agent emits node=
"unknown" events (config.py node_name default) into another node's dir when
NODE_NAME reaches the compose volume path but not the app env — the source of
the poison file.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

docs/backlog.md

@@ -579,11 +579,61 @@ więc KAŻDY nowy event był uznawany za starszy niż checkpoint i pomijany.
 observera → 7344 eventy przetworzone, `last_seen_age` spadł z 2 082 036 s (~24 dni)
 do 19 s, status=online/fresh, mismatch zniknął.
 
-**Fix systemowy (DO ZROBIENIA).** Porównanie po ścieżce jest z gruntu kruche —
+**Fix systemowy (ZROBIONE 2026-07-14, `task/fix-observer-checkpoint`).** Checkpoint
-wystarczy JEDEN plik o "większej" nazwie (inny node w katalogu, inny prefiks) żeby
+per-węzeł trzyma teraz **TIMESTAMP** (int epoch), nie ścieżkę. „Nowy event" =
-trwale zablokować węzeł. Checkpoint powinien opierać się na **timestampie** (jest
+`ts_z_nazwy_pliku > checkpoint_ts_węzła`; kolejność przetwarzania sortowana po
-w nazwie: `evt-<node>-<unixts>-<type>`) albo mtime pliku, nie na kolejności alfabetycznej.
+timestampie, nie leksykalnie. Timestamp parsowany z nazwy `evt-<node>-<unixts>-…`
-Dodatkowo: zbadać, czemu eventy z node="unknown" (HA) trafiają do katalogu innego węzła.
+(regex `-(\d{9,11})-`, ten sam co `operator_ui._event_file_ts`); **fallback na
+mtime** gdy nazwa nie pasuje — nieparsowalna nazwa NIGDY nie zwraca 0 (0 = leksykalne
+„starszy niż checkpoint" = dokładnie ten poison). Migracja starych checkpointów
+(ścieżka→ts) przy starcie; nieparsowalna wartość → 0 (reprocess wszystkiego —
+bezpieczne, `process_event` jest idempotentne na `last_seen`/`world_state`; lepiej
+przetworzyć duplikaty niż zgubić węzeł). Testy regresyjne w
+`test_incident_lifecycle.py` (sekcja 9). Znany, akceptowalny warunek brzegowy:
+strict `>` może pominąć event o `ts == checkpoint` dostarczony w PÓŹNIEJSZYM cyklu
+niż inne eventy z tej samej sekundy — nierealne przy cadence shippingu (rsync co
+60 s wysyła całą partię danej sekundy razem; kolejne partie są ~60 s od siebie).
+
+**Uwaga do idempotencji (zbadane).** Reprocess tego samego eventu NIE psuje
+world_state (status/last_seen deterministyczne, resolve incydentu guardowany na
+`status=="active"`), ALE `_handle_incident`/`deployment_*` inkrementują
+`occurrence_count` i dopisują do `events[]` przy każdym przetworzeniu — reprocess
+(np. jednorazowo po migracji) zawyża te liczniki. To kosmetyka, nie korupcja stanu.
+Docelowo można dedupować po `event.id` w `events[]` — osobny, drobny task.
+
+## Bug: ha-diag-agent emituje eventy z node="unknown" do katalogu innego węzła (2026-07-14)
+
+**Kontekst.** To był plik-truciciel z buga checkpointu wyżej:
+`evt-unknown-1781254800-ha_update_available-homeassistant-951.json` w
+`events/piha/`. Node w evencie = `unknown`, ale plik wylądował w katalogu `piha/`.
+Sufiks `-951` to `_seq` emittera → agent nachodził długo, wyemitował 951 eventów,
+wszystkie jako `node="unknown"`.
+
+**Root cause (config-wiring).** Tożsamość agenta (`node_name`) i KATALOG eventów
+pochodzą z DWÓCH niezależnych źródeł:
+- `services/ha-diag-agent/src/ha_diag/config.py:20` → `node_name: str = "unknown"`
+  (domyślne, gdy env `NODE_NAME` nie dotrze do procesu w kontenerze).
+- `services/ha-diag-agent/docker-compose.yml:12` → wolumen
+  `/opt/homelab/events/${NODE_NAME:-ha-diag}:/events` — `${NODE_NAME}` jest
+  interpolowane po stronie HOSTA (compose), a katalog jest dodatkowo twardo
+  przypięty do `piha` w `hosts/piha/runtime/ha-diag-agent/docker-compose.override.yml`.
+
+Jeśli `NODE_NAME` trafi do interpolacji wolumenu/override (→ `piha`), ale NIE do
+`environment:` procesu (albo `Settings.load()` przez `os.environ.setdefault` go nie
+nadpisze), aplikacja czyta `node_name="unknown"` i pisze eventy `node="unknown"`
+do katalogu `events/piha/`. Rozjazd między nazwą w evencie a katalogiem docelowym.
+
+**Skutek.** Poza zatruciem checkpointu (już naprawione osobno): eventy `node="unknown"`
+są bezużyteczne dla world_state (observer tworzy węzeł-widmo `unknown`, potem prune go
+kasuje bo nie ma go w topologii) — realny sygnał z ha-diag na piha przepada.
+
+**Fix (DO ZROBIENIA — osobny task).** Wymusić spójność: albo (a) `node_name` NIGDY
+nie może być `"unknown"` w produkcji — fail-fast/loud log gdy `NODE_NAME` nieustawione,
+zamiast cichego defaultu; albo (b) wyprowadzać `node_name` z tej samej zmiennej co
+ścieżka wolumenu i zweryfikować, że override piha faktycznie wstrzykuje `NODE_NAME=piha`
+do `environment:` kontenera (jest w override — sprawdzić, czemu w praktyce nie zadziałało:
+kolejność `env_file` vs `environment`, restart bez recreate, stary kontener). Docelowo
+observer/emitter powinien odrzucać/kwarantannować event, którego `node` != katalog.
 
 ## Bug: deploy-local.sh control-plane pada na ghost-kontenerach i zostawia mózg rozłożony (2026-07-12)
 

scripts/observer/observer.py

@@ -1,4 +1,5 @@
 import os
+import re
 import sys
 import json
 import time
@@ -55,6 +56,69 @@ def _parse_ts(ts) -> float:
     except Exception:
         return 0.0
 
+
+# Event filenames follow evt-<node>-<unixts>-<type>-<svc>.json (node_agent.py,
+# ha_diag/event_emitter.py, and the observer's own _emit_node_transition).  The
+# <unixts> is the authoritative ordering key for checkpointing — matched the same
+# way operator_ui.py::_event_file_ts does (a 9–11 digit run flanked by dashes;
+# real epochs are 10 digits and stay so until year 2286).
+_EVENT_TS_RE = re.compile(r"-(\d{9,11})-")
+
+
+def _ts_from_event_name(name) -> "int | None":
+    """Parse the embedded <unixts> from an event filename or full path.
+
+    Returns the int epoch, or None when the name does not carry one (foreign
+    prefix, events.py naming, junk file) so the caller can fall back to mtime.
+    """
+    m = _EVENT_TS_RE.search(Path(name).stem)
+    return int(m.group(1)) if m else None
+
+
+def _event_ts_from_path(file_path) -> int:
+    """Ordering/checkpoint timestamp for an event file.
+
+    Primary: the <unixts> embedded in the filename.  Fallback (name doesn't
+    parse): the file's mtime.  NEVER returns 0 for an existing file — a value of
+    0 would make the file compare as "older than the checkpoint" and be skipped
+    forever, which is exactly the lexical-path poisoning this fix removes.  If
+    even stat() fails, fall back to now() so the file is treated as new and gets
+    a chance to be processed (and quarantined if truly unreadable).
+    """
+    ts = _ts_from_event_name(file_path)
+    if ts is not None:
+        return ts
+    try:
+        return int(os.stat(file_path).st_mtime)
+    except OSError:
+        return int(time.time())
+
+
+def _checkpoint_ts_from_value(value) -> int:
+    """Coerce a stored checkpoint value into an int epoch (migration helper).
+
+    Historical formats of observer_checkpoint.json values:
+      - int/float           → already a timestamp (current format) — kept as-is
+      - path/name string    → the pre-fix lexical checkpoint; parse the embedded
+                              <unixts> out of it so the node resumes near where it
+                              left off instead of reprocessing everything
+      - anything else / unparseable string → 0 (reprocess all events for that
+        node).  Reprocessing is safe: process_event is idempotent w.r.t.
+        last_seen/world_state, so re-ingesting duplicates cannot corrupt state —
+        whereas guessing too high a checkpoint could silently drop events (the
+        failure mode being fixed).  Bias to reprocess, never to skip.
+    """
+    if isinstance(value, bool):  # bool is an int subclass — exclude explicitly
+        return 0
+    if isinstance(value, (int, float)):
+        return int(value)
+    if isinstance(value, str) and value:
+        ts = _ts_from_event_name(value)
+        if ts is not None:
+            return ts
+    return 0
+
+
 # Constants and Paths
 RUNTIME_PATH = os.getenv("RUNTIME_PATH", "/opt/homelab")
 EVENTS_DIR = Path(RUNTIME_PATH) / "events"
@@ -83,9 +147,13 @@ logger = logging.getLogger("observer")
 
 class Observer:
     def __init__(self):
-        # Per-node-directory checkpoint: {"vps": "last/file/path", "piha": "last/file/path"}
+        # Per-node-directory checkpoint keyed on the last-processed event
-        # Replaces the old single last_processed_file which silently skipped event dirs
+        # TIMESTAMP (int epoch): {"vps": 1784000000, "piha": 1784000123}.
-        # that sort alphabetically before the checkpoint (e.g. piha/ < vps/).
+        # A file is "new" iff its event timestamp > the node's checkpoint.
+        # This replaces the earlier lexical-PATH comparison, which permanently
+        # poisoned a node the moment a single file with a lexically-larger name
+        # landed in its dir (e.g. evt-unknown-… > evt-piha-…): every genuinely
+        # newer event then sorted "before" the checkpoint and was skipped forever.
         self.node_checkpoints: dict = {}
         self.world_state = {
             "nodes": {},
@@ -168,15 +236,28 @@ class Observer:
                     checkpoint = json.load(f)
 
                 if "node_checkpoints" in checkpoint:
-                    # New format: per-directory checkpoints.
+                    # Per-directory checkpoints.  Values may be int epochs (current
-                    self.node_checkpoints = checkpoint["node_checkpoints"]
+                    # format) OR pre-fix path strings — coerce every value to an
+                    # int timestamp so a checkpoint file written by the old
+                    # lexical-path observer migrates transparently on first start.
+                    raw = checkpoint["node_checkpoints"] or {}
+                    self.node_checkpoints = {
+                        node: _checkpoint_ts_from_value(val)
+                        for node, val in raw.items()
+                    }
+                    if any(not isinstance(v, (int, float)) for v in raw.values()):
+                        logger.info(
+                            "Migrated path-based node_checkpoints → timestamps: %s",
+                            self.node_checkpoints,
+                        )
                 elif "last_processed_file" in checkpoint:
-                    # Migrate old single-file checkpoint: extract node dir from path.
+                    # Migrate the very old single-file checkpoint: extract node dir
+                    # from the path and the timestamp from the filename.
                     old = checkpoint["last_processed_file"]
                     if old:
                         try:
                             node_dir = Path(old).relative_to(EVENTS_DIR).parts[0]
-                            self.node_checkpoints = {node_dir: old}
+                            self.node_checkpoints = {node_dir: _checkpoint_ts_from_value(old)}
                             logger.info(f"Migrated old checkpoint → node_checkpoints: {self.node_checkpoints}")
                         except Exception:
                             pass  # Bad path — start fresh
@@ -687,11 +768,12 @@ class Observer:
         except Exception as e:
             logger.error(f"Failed to touch heartbeat file: {e}")
 
-        # Collect all event files grouped by node directory.
+        # Collect all event files grouped by node directory.  A file is "new"
-        # Per-node checkpoints are compared within each directory independently,
+        # when its event TIMESTAMP (from the filename, mtime fallback) is greater
-        # so late-arriving events from remote nodes (sorted earlier in the path)
+        # than the node's checkpoint timestamp — never a lexical path compare, so
-        # are never skipped just because another node's checkpoint is further ahead.
+        # a lexically-smaller-but-newer filename (evt-piha-… after a stray
-        all_files = sorted(glob.glob(str(EVENTS_DIR / "**" / "*.json"), recursive=True))
+        # evt-unknown-…) can no longer poison a node into skipping every event.
+        all_files = glob.glob(str(EVENTS_DIR / "**" / "*.json"), recursive=True)
 
         new_files = []
         for file_path in all_files:
@@ -699,9 +781,15 @@ class Observer:
                 node_dir = str(Path(file_path).relative_to(EVENTS_DIR).parts[0])
             except (IndexError, ValueError):
                 node_dir = "__unknown__"
-            last_for_node = self.node_checkpoints.get(node_dir, "")
+            ev_ts = _event_ts_from_path(file_path)
-            if file_path > last_for_node:
+            last_for_node = self.node_checkpoints.get(node_dir, 0)
-                new_files.append((node_dir, file_path))
+            if ev_ts > last_for_node:
+                new_files.append((ev_ts, node_dir, file_path))
+
+        # Process oldest-first (tie-break on path for determinism) so the
+        # checkpoint advances monotonically in time and a mid-batch crash resumes
+        # from the right place.
+        new_files.sort(key=lambda t: (t[0], t[2]))
 
         if not new_files:
             # Even if no new events, prune stale entries and refresh summary freshness.
@@ -710,8 +798,8 @@ class Observer:
             return
 
         logger.info(f"Processing {len(new_files)} new events across "
-                    f"{len({n for n, _ in new_files})} node(s)")
+                    f"{len({n for _, n, _ in new_files})} node(s)")
-        for node_dir, file_path in new_files:
+        for ev_ts, node_dir, file_path in new_files:
             try:
                 with open(file_path, "r") as f:
                     event = json.load(f)
@@ -721,13 +809,13 @@ class Observer:
                 # node we just declared dead.  They are still consumed by the
                 # supervisor (alerting) and the panel event feed.
                 if event.get("source") == "observer":
-                    if file_path > self.node_checkpoints.get(node_dir, ""):
+                    if ev_ts > self.node_checkpoints.get(node_dir, 0):
-                        self.node_checkpoints[node_dir] = file_path
+                        self.node_checkpoints[node_dir] = ev_ts
                     continue
                 self.process_event(event)
-                # Advance per-node checkpoint (only forward — no regression).
+                # Advance per-node checkpoint by timestamp (only forward).
-                if file_path > self.node_checkpoints.get(node_dir, ""):
+                if ev_ts > self.node_checkpoints.get(node_dir, 0):
-                    self.node_checkpoints[node_dir] = file_path
+                    self.node_checkpoints[node_dir] = ev_ts
             except Exception as e:
                 logger.error(
                     "Error processing node_dir=%s file=%s (%s: %s)",

services/control-plane/tests/test_incident_lifecycle.py

@@ -346,7 +346,11 @@ def test_run_once_quarantines_bad_event_and_processes_next_for_same_node(tmp_pat
     assert quarantined.exists()
     assert not bad_event.exists()
     assert obs.world_state["nodes"]["lustro"]["status"] == "online"
-    assert obs.node_checkpoints["lustro"] == str(good_event)
+    # Checkpoint is now a timestamp, not a path.  The filenames here carry no
+    # 10-digit epoch (evt-lustro-2-good) so _event_ts_from_path falls back to
+    # mtime — a positive int, and the node is not wedged.
+    assert isinstance(obs.node_checkpoints["lustro"], int)
+    assert obs.node_checkpoints["lustro"] > 0
 
 
 # ---------------------------------------------------------------------------
@@ -538,5 +542,215 @@ def test_run_once_skips_observer_emitted_events(tmp_path):
 
     # Not ingested: last_seen stays old → vps stays dead/offline.
     assert obs.world_state["nodes"]["vps"]["status"] == "offline"
-    # Checkpoint still advanced over the skipped file.
+    # Checkpoint still advanced past the skipped file — now by timestamp
-    assert obs.node_checkpoints.get("vps") == str(ev_path)
+    # (the 9999999999 epoch embedded in the filename), not the path string.
+    assert obs.node_checkpoints.get("vps") == 9999999999
+
+
+# ---------------------------------------------------------------------------
+# 9. Checkpoint by TIMESTAMP, not lexical path
+# ---------------------------------------------------------------------------
+# Regression suite for the "poisoned node" bug (2026-07-12): PIHA was dead to
+# the observer for ~34 days because its checkpoint compared event PATHS
+# lexically.  A stray evt-unknown-<ts>-… file (lexically > every real
+# evt-piha-… name) pinned the checkpoint, so every genuinely newer piha event
+# sorted "before" it and was skipped forever.  The fix keys the checkpoint on
+# the timestamp embedded in the filename (mtime fallback).
+
+from observer.observer import (  # noqa: E402
+    _ts_from_event_name,
+    _event_ts_from_path,
+    _checkpoint_ts_from_value,
+)
+
+
+def _write_event(dir_path: Path, name: str, *, node: str, ts: int,
+                 etype: str = "node_health", source: str | None = None) -> Path:
+    dir_path.mkdir(parents=True, exist_ok=True)
+    body = {
+        "id": name[:-5],
+        "timestamp": ts,
+        "date": "2026-07-14T00:00:00Z",
+        "type": etype,
+        "severity": "info",
+        "node": node,
+        "service": "",
+        "message": "test",
+        "payload": {"disk_pct": 1, "mem_pct": 2, "cpu_pct": 3},
+    }
+    if source:
+        body["source"] = source
+    p = dir_path / name
+    p.write_text(json.dumps(body))
+    return p
+
+
+def _topology_with(obs, *nodes: str) -> None:
+    import observer.observer as obs_mod
+    lines = ["nodes:"]
+    for n in nodes:
+        lines.append(f"  {n}:\n    roles: [infra]\n    connectivity: {{}}")
+    obs_mod.INVENTORY_TOPOLOGY.write_text("\n".join(lines) + "\n")
+    obs.inventory = obs._load_inventory()
+
+
+def test_lexically_smaller_but_newer_event_is_processed(tmp_path):
+    """THE regression: a stray evt-unknown-… must not wedge piha forever.
+
+    An evt-unknown-<older-ts>-… file lands in the piha/ dir (lexically larger
+    than any evt-piha-… name).  A genuinely NEWER evt-piha-… event, which is
+    lexically SMALLER, must still be processed.
+    """
+    obs = _make_observer_simple(tmp_path)
+    import observer.observer as obs_mod
+    _topology_with(obs, "vps", "piha")
+
+    piha_dir = obs_mod.EVENTS_DIR / "piha"
+    # The poison: HA event with node="unknown" that landed in piha/ (older ts).
+    _write_event(
+        piha_dir,
+        "evt-unknown-1781254800-ha_update_available-homeassistant-951.json",
+        node="unknown", ts=1781254800, etype="ha_update_available",
+    )
+    obs.run_once()
+    # Checkpoint is the stray event's ts (unknown node itself is pruned).
+    assert obs.node_checkpoints["piha"] == 1781254800
+
+    # A newer real piha heartbeat — lexically "evt-piha-…" < "evt-unknown-…".
+    _write_event(
+        piha_dir, "evt-piha-1784000000-node_health-node.json",
+        node="piha", ts=1784000000,
+    )
+    obs.run_once()
+
+    # Under the old lexical logic this was skipped forever.  Now it is processed.
+    assert "piha" in obs.world_state["nodes"]
+    assert obs.world_state["nodes"]["piha"]["last_seen"] == 1784000000
+    assert obs.node_checkpoints["piha"] == 1784000000
+
+
+def test_unparseable_name_does_not_block_node(tmp_path):
+    """A file whose name carries no epoch falls back to mtime, never ts=0.
+
+    ts=0 would compare as "older than the checkpoint" and be skipped forever —
+    the exact poisoning mechanism.  The event must be ingested instead.
+    """
+    obs = _make_observer_simple(tmp_path)
+    import observer.observer as obs_mod
+    _topology_with(obs, "vps", "solaria")
+
+    solaria_dir = obs_mod.EVENTS_DIR / "solaria"
+    # No 10-digit epoch in the name → mtime fallback (≈ now, a positive int).
+    # Use a fresh event timestamp so read-time liveness keeps it online.
+    fresh = int(time.time())
+    _write_event(
+        solaria_dir, "weird-legacy-name.json",
+        node="solaria", ts=fresh,
+    )
+    obs.run_once()
+
+    # Ingested (not skipped): the node exists with the event's last_seen, and
+    # the checkpoint advanced to a positive int via the mtime fallback.
+    assert "solaria" in obs.world_state["nodes"]
+    assert obs.world_state["nodes"]["solaria"]["last_seen"] == fresh
+    assert obs.world_state["nodes"]["solaria"]["status"] == "online"
+    assert isinstance(obs.node_checkpoints["solaria"], int)
+    assert obs.node_checkpoints["solaria"] > 0
+
+
+def test_checkpoint_governed_by_timestamp_not_path(tmp_path):
+    """A lexically-larger but temporally-OLDER file must be skipped.
+
+    Proves the ordering key is the timestamp, not the path: an event whose name
+    sorts after the checkpoint yet whose ts predates it does not regress state.
+    """
+    obs = _make_observer_simple(tmp_path)
+    import observer.observer as obs_mod
+    _topology_with(obs, "vps")
+
+    obs.world_state["nodes"]["vps"] = {
+        "status": "online", "last_seen": 1784000000, "roles": ["infra"],
+    }
+    obs.node_checkpoints["vps"] = 1784000000
+
+    vps_dir = obs_mod.EVENTS_DIR / "vps"
+    # Lexically HUGE name ("evt-zzzz…" > "evt-vps-…") but an OLDER epoch.
+    _write_event(
+        vps_dir, "evt-zzzzzzzz-1000000000-node_health-node.json",
+        node="vps", ts=1000000000,
+    )
+    obs.run_once()
+    # Skipped by ts (1e9 < checkpoint) despite the larger path → last_seen intact.
+    assert obs.world_state["nodes"]["vps"]["last_seen"] == 1784000000
+    assert obs.node_checkpoints["vps"] == 1784000000
+
+    # A genuinely newer event advances the checkpoint by ts.
+    _write_event(
+        vps_dir, "evt-vps-1785000000-node_health-node.json",
+        node="vps", ts=1785000000,
+    )
+    obs.run_once()
+    assert obs.world_state["nodes"]["vps"]["last_seen"] == 1785000000
+    assert obs.node_checkpoints["vps"] == 1785000000
+
+
+def test_migration_path_checkpoint_to_timestamp(tmp_path):
+    """A pre-fix checkpoint file holding PATH strings migrates to int epochs."""
+    obs = _make_observer_simple(tmp_path)  # constructs to establish paths
+    import observer.observer as obs_mod
+
+    old_path = str(
+        obs_mod.EVENTS_DIR / "piha"
+        / "evt-unknown-1781254800-ha_update_available-homeassistant-951.json"
+    )
+    obs_mod.OBSERVER_STATE_FILE.write_text(json.dumps({
+        "node_checkpoints": {
+            "piha": old_path,                      # old lexical path string
+            "vps": 1785000000,                     # already a timestamp int
+            "solaria": "junk-no-epoch-here.json",  # unparseable → reprocess (0)
+        }
+    }))
+
+    migrated = Observer()  # re-loads the checkpoint we just wrote
+    assert migrated.node_checkpoints["piha"] == 1781254800
+    assert migrated.node_checkpoints["vps"] == 1785000000
+    assert migrated.node_checkpoints["solaria"] == 0
+
+
+def test_migration_old_single_file_checkpoint(tmp_path):
+    """The very old {"last_processed_file": <path>} format also migrates."""
+    obs = _make_observer_simple(tmp_path)
+    import observer.observer as obs_mod
+
+    old = str(obs_mod.EVENTS_DIR / "piha" / "evt-piha-1784000000-node_health-node.json")
+    obs_mod.OBSERVER_STATE_FILE.write_text(json.dumps({"last_processed_file": old}))
+
+    migrated = Observer()
+    assert migrated.node_checkpoints == {"piha": 1784000000}
+
+
+def test_ts_helpers_units():
+    """Unit-level coverage of the timestamp parsing helpers."""
+    assert _ts_from_event_name("evt-piha-1784000000-node_health-node.json") == 1784000000
+    assert _ts_from_event_name(
+        "evt-unknown-1781254800-ha_update_available-homeassistant-951.json"
+    ) == 1781254800
+    assert _ts_from_event_name("no-epoch-here.json") is None
+
+    assert _checkpoint_ts_from_value(1784000000) == 1784000000
+    assert _checkpoint_ts_from_value(1784000000.9) == 1784000000
+    assert _checkpoint_ts_from_value(
+        "/opt/homelab/events/piha/evt-piha-1784000000-node_health-node.json"
+    ) == 1784000000
+    assert _checkpoint_ts_from_value("garbage") == 0
+    assert _checkpoint_ts_from_value(None) == 0
+    assert _checkpoint_ts_from_value(True) == 0  # bool is an int subclass — excluded
+
+
+def test_event_ts_from_path_mtime_fallback(tmp_path):
+    """A real file with a no-epoch name returns its mtime, never 0."""
+    p = tmp_path / "weird-name.json"
+    p.write_text("{}")
+    ts = _event_ts_from_path(str(p))
+    assert isinstance(ts, int)
+    assert ts > 0