fix(ingestor): a retained status message is not observer liveness (#1885)

## Problem

The broker replays every retained `status` message on subscribe, so each
ingestor restart pushes all of them through the status path in
`handleMessage`. That path stamps `last_seen` with `time.Now()`
(deliberately, per #1465).

Observed on a live deployment on 2026-08-11: **23 observers all carried
`last_seen = 2026-08-06T08:20:09Z`** — 15 seconds after container start
— and 18 of them had sent no actual packet in over a month. Their
retained publish dates lined up almost 1:1 with `last_packet_at`, i.e.
the replay was their only sign of "life":

| observer | last real packet | retained status published |
|---|---|---|
| ON8AR - Observer | never | 2026-03-20 |
| BE-BGS-RRY120-RES | never | 2026-04-02 |
| A3BEF374 | 2026-04-30 | 2026-04-30 |
| BE-BGS-RRY120-RUDY | 2026-05-18 | 2026-05-18 |
| BE-JBE-ETG-O1 | 2026-06-10 | 2026-06-10 |

That makes dead observers immortal, three ways per restart:

1. `last_seen` jumps forward, so `RemoveStaleObservers` can never age
them out as long as a restart happens inside `observerDays`.
2. The unconditional `inactive = 0` reactivation at the end of
`UpsertObserverAt` undoes any soft-delete that did land.
3. A metrics sample is filed at ingest time, dating a months-old reading
as a present-tense measurement.

`UpsertObserverAt`'s docstring already claimed retained replays were a
no-op for `last_seen` thanks to the `MAX` guard. That held only while
the caller passed the envelope timestamp; #1465 switched it to ingest
time, which defeats the guard.

## Fix

The retained path now updates metadata only, via a new
`UpsertObserverRetained`:

- no `last_seen` advance
- no `inactive = 0` reactivation
- no `packet_count` bump
- no metrics sample
- **no INSERT** — a retained-only observer the analyzer has never heard
from live describes a past that may be months old and does not belong in
the list. A live message from the same observer creates the row through
the normal path moments later.

Live status handling is unchanged.

## Tests

Seven tests in `cmd/ingestor/retained_status_test.go`, written before
the fix:

- 4 that failed on the bug: `last_seen` advance, reactivation of a
soft-deleted row, creation of a never-seen observer, metrics-sample
insert
- 2 regression guards pinning live (non-retained) behaviour: `last_seen`
still advances, unknown observer still created
- 1 asserting retained metadata is still applied — the snapshot is the
observer's last known state, only the liveness signal is suppressed

`mockMessage` gained a `retained` field so `Retained()` is controllable.

Meta flattening is extracted to `observerMetaColumns` so both write
paths bind identical args.

Full `cmd/ingestor` suite passes.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

cmd/ingestor/db.go

@@ -1103,8 +1103,68 @@ func (s *Store) UpsertObserverAt(id, name, iata string, meta *ObserverMeta, last
 	}
 	normalizedIATA := strings.TrimSpace(strings.ToUpper(iata))
 
-	var model, firmware, clientVersion, radio interface{}
-	var batteryMv, uptimeSecs, noiseFloor, canRelay interface{}
+	model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay := observerMetaColumns(meta)
+
+	_, err := s.stmtUpsertObserver.Exec(
+		id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
+		name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
+	)
+	if err != nil {
+		s.Stats.WriteErrors.Add(1)
+		return err
+	}
+	s.Stats.ObserverUpserts.Add(1)
+
+	// Reactivate if this observer was previously marked inactive
+	s.db.Exec(`UPDATE observers SET inactive = 0 WHERE id = ? AND inactive = 1`, id)
+	return nil
+}
+
+// UpsertObserverRetained applies the metadata from a RETAINED status message
+// without treating it as a sign of life. The broker replays retained messages
+// on every subscribe, so an ingestor restart would otherwise stamp last_seen
+// with the restart time for every observer that ever published one — making
+// dead observers permanently un-ageable by RemoveStaleObservers, and undoing
+// any inactive flag it did manage to set.
+//
+// Deliberately narrower than UpsertObserverAt: it updates metadata columns on
+// an existing row only. It does not advance last_seen, does not clear
+// inactive, does not bump packet_count, and does not INSERT — a retained-only
+// observer the analyzer has never heard from live describes a past that may be
+// months old and does not belong in the list. A live message from the same
+// observer arrives moments later and creates the row through the normal path.
+func (s *Store) UpsertObserverRetained(id, name, iata string, meta *ObserverMeta) error {
+	normalizedIATA := strings.TrimSpace(strings.ToUpper(iata))
+	model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay := observerMetaColumns(meta)
+
+	_, err := s.db.Exec(`
+		UPDATE observers SET
+			name = COALESCE(?, name),
+			iata = COALESCE(?, iata),
+			model = COALESCE(?, model),
+			firmware = COALESCE(?, firmware),
+			client_version = COALESCE(?, client_version),
+			radio = COALESCE(?, radio),
+			battery_mv = COALESCE(?, battery_mv),
+			uptime_secs = COALESCE(?, uptime_secs),
+			noise_floor = COALESCE(?, noise_floor),
+			can_relay = COALESCE(?, can_relay),
+			can_relay_seen = CASE WHEN ? IS NULL THEN can_relay_seen ELSE 1 END
+		WHERE id = ?`,
+		name, normalizedIATA, model, firmware, clientVersion, radio,
+		batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay, id,
+	)
+	if err != nil {
+		s.Stats.WriteErrors.Add(1)
+		return err
+	}
+	return nil
+}
+
+// observerMetaColumns flattens an *ObserverMeta into the driver args the
+// observer upserts bind. A nil field stays nil so the COALESCE in the SQL
+// leaves the existing column untouched (#1290).
+func observerMetaColumns(meta *ObserverMeta) (model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay interface{}) {
 	if meta != nil {
 		if meta.Model != nil {
 			model = *meta.Model
@@ -1139,20 +1199,7 @@ func (s *Store) UpsertObserverAt(id, name, iata string, meta *ObserverMeta, last
 			}
 		}
 	}
-
-	_, err := s.stmtUpsertObserver.Exec(
-		id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
-		name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
-	)
-	if err != nil {
-		s.Stats.WriteErrors.Add(1)
-		return err
-	}
-	s.Stats.ObserverUpserts.Add(1)
-
-	// Reactivate if this observer was previously marked inactive
-	s.db.Exec(`UPDATE observers SET inactive = 0 WHERE id = ? AND inactive = 1`, id)
-	return nil
+	return model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay
 }
 
 // Close checkpoints the WAL and closes the database.

cmd/ingestor/main.go

@@ -604,6 +604,22 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message,
 		name, _ := msg["origin"].(string)
 		iata := parts[1]
 		meta := extractObserverMeta(msg)
+		// A replayed status message is the broker handing us the observer's
+		// last published snapshot — it is not evidence the observer is alive
+		// now. Stamping last_seen from it resurrects dead observers, so the
+		// replay path only refreshes metadata.
+		//
+		// Two ways to recognise one: the retain flag (our own subscribe), and
+		// the payload itself (a replay that reached us through the mosquitto
+		// bridge, where the flag does not survive the hop — see
+		// statusIsLiveness).
+		if m.Retained() || !statusIsLiveness(msg, time.Now().UTC()) {
+			if err := store.UpsertObserverRetained(observerID, name, iata, meta); err != nil {
+				log.Printf("MQTT [%s] retained observer status error: %v", tag, err)
+			}
+			log.Print(formatStatusLog(tag, firstNonEmpty(name, observerID), iata))
+			return
+		}
 		// observer.last_seen is "when did the analyzer last hear from this
 		// observer" — fundamentally an ingest-time question. Passing "" makes
 		// UpsertObserverAt use time.Now(), independent of the envelope timestamp

cmd/ingestor/main_test.go

@@ -105,13 +105,14 @@ func TestUnixTime(t *testing.T) {
 
 // mockMessage implements mqtt.Message for testing handleMessage
 type mockMessage struct {
-	topic   string
-	payload []byte
+	topic    string
+	payload  []byte
+	retained bool
 }
 
 func (m *mockMessage) Duplicate() bool  { return false }
 func (m *mockMessage) Qos() byte        { return 0 }
-func (m *mockMessage) Retained() bool   { return false }
+func (m *mockMessage) Retained() bool   { return m.retained }
 func (m *mockMessage) Topic() string     { return m.topic }
 func (m *mockMessage) MessageID() uint16 { return 0 }
 func (m *mockMessage) Payload() []byte   { return m.payload }

cmd/ingestor/retained_status_test.go

@@ -0,0 +1,171 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+// A retained MQTT status message is a snapshot the broker replays on every
+// subscribe — it says what the observer last published, not that the observer
+// is alive now. Treating the replay as liveness makes dead observers immortal:
+// last_seen jumps to the ingestor's restart time, RemoveStaleObservers can
+// never age them out, and any row it did mark gets reactivated on the next
+// restart. Observed on live 2026-08-11: 23 observers, all stamped 15s after
+// container start, 18 of them with no real packet in over a month.
+
+const retainedStatusTopic = "meshcore/LAX/obs-retained/status"
+
+func retainedStatusMsg(topic string, payload string) *mockMessage {
+	return &mockMessage{topic: topic, payload: []byte(payload), retained: true}
+}
+
+// seedObserver inserts an observer and forces last_seen to ageDays ago.
+func seedObserver(t *testing.T, s *Store, id string, ageDays int) string {
+	t.Helper()
+	if err := s.UpsertObserver(id, id, "LAX", nil); err != nil {
+		t.Fatal(err)
+	}
+	old := time.Now().UTC().AddDate(0, 0, -ageDays).Format(time.RFC3339)
+	if _, err := s.db.Exec(`UPDATE observers SET last_seen = ? WHERE id = ?`, old, id); err != nil {
+		t.Fatal(err)
+	}
+	return old
+}
+
+func observerLastSeen(t *testing.T, s *Store, id string) string {
+	t.Helper()
+	var got string
+	if err := s.db.QueryRow(`SELECT last_seen FROM observers WHERE id = ?`, id).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	return got
+}
+
+func TestRetainedStatusDoesNotAdvanceLastSeen(t *testing.T) {
+	store := newTestStore(t)
+	want := seedObserver(t, store, "obs-retained", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","noise_floor":-95.5}`),
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-retained"); got != want {
+		t.Errorf("last_seen = %s, want %s (retained replay is not liveness)", got, want)
+	}
+}
+
+func TestRetainedStatusDoesNotReactivateInactiveObserver(t *testing.T) {
+	store := newTestStore(t)
+	seedObserver(t, store, "obs-retained", 60)
+	if _, err := store.db.Exec(`UPDATE observers SET inactive = 1 WHERE id = ?`, "obs-retained"); err != nil {
+		t.Fatal(err)
+	}
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","noise_floor":-95.5}`),
+		nil, nil, &Config{})
+
+	var inactive int
+	if err := store.db.QueryRow(`SELECT inactive FROM observers WHERE id = ?`, "obs-retained").Scan(&inactive); err != nil {
+		t.Fatal(err)
+	}
+	if inactive != 1 {
+		t.Errorf("inactive = %d, want 1 (retained replay must not resurrect a soft-deleted observer)", inactive)
+	}
+}
+
+// A retained snapshot from an observer the analyzer has never heard from live
+// describes a past that may be months old. Creating a row for it puts a
+// permanently dead observer in the list.
+func TestRetainedStatusDoesNotCreateUnknownObserver(t *testing.T) {
+	store := newTestStore(t)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		retainedStatusMsg("meshcore/LAX/obs-never-seen/status", `{"origin":"Ghost","noise_floor":-95.5}`),
+		nil, nil, &Config{})
+
+	var count int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, "obs-never-seen").Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Errorf("observers row count = %d, want 0 (retained-only observer must not be created)", count)
+	}
+}
+
+// The metrics sample is stamped with ingest time, so a replay would file a
+// months-old reading as a present-tense measurement.
+func TestRetainedStatusDoesNotInsertMetricsSample(t *testing.T) {
+	store := newTestStore(t)
+	seedObserver(t, store, "obs-retained", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","noise_floor":-95.5,"battery_mv":3900}`),
+		nil, nil, &Config{})
+
+	var count int
+	if err := store.db.QueryRow(
+		`SELECT COUNT(*) FROM observer_metrics WHERE observer_id = ?`, "obs-retained",
+	).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Errorf("observer_metrics rows = %d, want 0 (retained replay is not a new sample)", count)
+	}
+}
+
+// The retained snapshot is still the observer's last known state, so its
+// metadata is worth keeping — only the liveness signal is suppressed.
+func TestRetainedStatusStillUpdatesMetadata(t *testing.T) {
+	store := newTestStore(t)
+	seedObserver(t, store, "obs-retained", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","firmware":"v1.2.3","noise_floor":-95.5}`),
+		nil, nil, &Config{})
+
+	var firmware string
+	var noiseFloor float64
+	if err := store.db.QueryRow(
+		`SELECT firmware, noise_floor FROM observers WHERE id = ?`, "obs-retained",
+	).Scan(&firmware, &noiseFloor); err != nil {
+		t.Fatal(err)
+	}
+	if firmware != "v1.2.3" {
+		t.Errorf("firmware = %q, want %q", firmware, "v1.2.3")
+	}
+	if noiseFloor != -95.5 {
+		t.Errorf("noise_floor = %v, want -95.5", noiseFloor)
+	}
+}
+
+// Regression guard for #1465: a live status message must still stamp last_seen
+// with ingest time. Only the retained flag changes the behaviour.
+func TestLiveStatusStillAdvancesLastSeen(t *testing.T) {
+	store := newTestStore(t)
+	before := seedObserver(t, store, "obs-live", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		&mockMessage{topic: "meshcore/LAX/obs-live/status", payload: []byte(`{"origin":"obs-live","noise_floor":-95.5}`)},
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-live"); got == before {
+		t.Errorf("last_seen = %s, want it advanced past %s (live status is liveness)", got, before)
+	}
+}
+
+func TestLiveStatusStillCreatesUnknownObserver(t *testing.T) {
+	store := newTestStore(t)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		&mockMessage{topic: "meshcore/LAX/obs-fresh/status", payload: []byte(`{"origin":"Fresh","noise_floor":-95.5}`)},
+		nil, nil, &Config{})
+
+	var count int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, "obs-fresh").Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 1 {
+		t.Errorf("observers row count = %d, want 1 (live status creates the observer)", count)
+	}
+}

cmd/ingestor/status_liveness.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"strings"
+	"time"
+)
+
+// statusLivenessMaxAge is how far a status payload's own timestamp may lag
+// before the message stops counting as evidence the observer is alive now.
+//
+// Deliberately generous: observer clocks are naive (#1478) and can sit a whole
+// timezone off, so anything under a day has to be accepted. The replays this
+// guards against are months stale, so the margin costs nothing.
+const statusLivenessMaxAge = 24 * time.Hour
+
+// statusIsLiveness reports whether a status message proves the observer was
+// alive when we received it.
+//
+// The retain flag alone cannot answer that. Between EMQX and the ingestor sits
+// a mosquitto bridge, and MQTT only sets RETAIN on messages delivered in
+// response to a *new* subscription. The bridge runs cleansession, so every
+// reconnect resubscribes and pulls EMQX's entire retained set — which it then
+// republishes into its own broker as ordinary traffic. The ingestor is already
+// subscribed there, so those months-old snapshots arrive with RETAIN cleared,
+// indistinguishable from live status. On 2026-08-13 an EMQX restart replayed
+// them and recreated 17 long-dead observers in two seconds.
+//
+// So judge the payload instead of the transport:
+//   - a status that is not "online" is the observer's own offline notice (its
+//     LWT, or a replay of it) and is never proof of life;
+//   - a timestamp older than statusLivenessMaxAge describes a past that may be
+//     months old.
+//
+// A payload with no timestamp, or one we cannot parse, is unjudgeable and
+// keeps the previous behaviour — this guard removes false liveness, it does
+// not invent new grounds to discard observers.
+func statusIsLiveness(msg map[string]interface{}, now time.Time) bool {
+	if s, ok := msg["status"].(string); ok && s != "" && !strings.EqualFold(s, "online") {
+		return false
+	}
+	raw, _ := msg["timestamp"].(string)
+	if raw == "" {
+		return true
+	}
+	t, _, err := parseEnvelopeTime(raw)
+	if err != nil {
+		return true
+	}
+	return !t.Before(now.Add(-statusLivenessMaxAge))
+}

cmd/ingestor/status_liveness_test.go

@@ -0,0 +1,203 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+// The retain flag alone does not identify a replay. Our live topology puts a
+// mosquitto bridge between EMQX and the ingestor, and MQTT only sets RETAIN on
+// messages delivered in response to a *new* subscription. When the bridge
+// reconnects (it runs cleansession, so every reconnect resubscribes), EMQX
+// replays its whole retained set to the bridge, which republishes it into its
+// own broker — and the ingestor, already subscribed, receives those months-old
+// snapshots with RETAIN cleared, indistinguishable from live traffic.
+//
+// Observed on live 2026-08-13: EMQX restarted at 13:09 UTC and 17 long-dead
+// observers were recreated inside two seconds, among them BE-BGS-RRY120-RES
+// whose retained payload was published 2026-04-02 and says "offline".
+//
+// So judge the payload, not the transport: a status message is evidence of
+// life only if it claims to be online and carries a timestamp from roughly now.
+
+func liveStatusMsg(topic, payload string) *mockMessage {
+	return &mockMessage{topic: topic, payload: []byte(payload), retained: false}
+}
+
+func statusPayload(origin, status string, ts time.Time) string {
+	return fmt.Sprintf(`{"origin":%q,"status":%q,"timestamp":%q,"noise_floor":-95.5}`,
+		origin, status, ts.Format("2006-01-02T15:04:05.000000"))
+}
+
+func TestStaleStatusPayloadDoesNotAdvanceLastSeen(t *testing.T) {
+	store := newTestStore(t)
+	want := seedObserver(t, store, "obs-zombie", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-zombie/status",
+			statusPayload("obs-zombie", "online", time.Now().UTC().AddDate(0, -4, 0))),
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-zombie"); got != want {
+		t.Errorf("last_seen = %s, want %s (a 4-month-old payload is a replay, not liveness)", got, want)
+	}
+}
+
+// The exact live case: an observer purged from the DB must not come back when
+// the bridge replays its months-old retained snapshot with RETAIN cleared.
+func TestStaleStatusPayloadDoesNotCreateObserver(t *testing.T) {
+	store := newTestStore(t)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/OST/obs-purged/status",
+			statusPayload("BE-BGS-RRY120-RES", "offline", time.Now().UTC().AddDate(0, -4, 0))),
+		nil, nil, &Config{})
+
+	var count int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, "obs-purged").Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Errorf("observers row count = %d, want 0 (replayed snapshot must not resurrect a purged observer)", count)
+	}
+}
+
+// An offline notice is the observer's own death certificate — the broker
+// replaying it, or the LWT firing, is never proof of life.
+func TestOfflineStatusIsNotLiveness(t *testing.T) {
+	store := newTestStore(t)
+	want := seedObserver(t, store, "obs-offline", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-offline/status",
+			statusPayload("obs-offline", "offline", time.Now().UTC())),
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-offline"); got != want {
+		t.Errorf("last_seen = %s, want %s (an offline status is not liveness)", got, want)
+	}
+}
+
+func TestStaleStatusPayloadDoesNotReactivateInactiveObserver(t *testing.T) {
+	store := newTestStore(t)
+	seedObserver(t, store, "obs-zombie", 60)
+	if _, err := store.db.Exec(`UPDATE observers SET inactive = 1 WHERE id = ?`, "obs-zombie"); err != nil {
+		t.Fatal(err)
+	}
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-zombie/status",
+			statusPayload("obs-zombie", "online", time.Now().UTC().AddDate(0, -4, 0))),
+		nil, nil, &Config{})
+
+	var inactive int
+	if err := store.db.QueryRow(`SELECT inactive FROM observers WHERE id = ?`, "obs-zombie").Scan(&inactive); err != nil {
+		t.Fatal(err)
+	}
+	if inactive != 1 {
+		t.Errorf("inactive = %d, want 1 (a replay must not undo the soft-delete)", inactive)
+	}
+}
+
+// ─── The guard must not swallow real observers ──────────────────────────────
+
+func TestFreshStatusStillCountsAsLiveness(t *testing.T) {
+	store := newTestStore(t)
+	stale := seedObserver(t, store, "obs-live", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-live/status",
+			statusPayload("obs-live", "online", time.Now().UTC())),
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-live"); got == stale {
+		t.Errorf("last_seen = %s, want it advanced (a fresh online status is liveness)", got)
+	}
+}
+
+// Observer clocks are naive and can sit a whole timezone off (#1478). A UTC-12
+// observer must not be mistaken for a replay.
+func TestNaiveClockSkewStillCountsAsLiveness(t *testing.T) {
+	store := newTestStore(t)
+	stale := seedObserver(t, store, "obs-skewed", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-skewed/status",
+			statusPayload("obs-skewed", "online", time.Now().UTC().Add(-12*time.Hour))),
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-skewed"); got == stale {
+		t.Errorf("last_seen = %s, want it advanced (12h naive skew is a timezone, not a replay)", got)
+	}
+}
+
+func TestStatusWithoutTimestampStillCountsAsLiveness(t *testing.T) {
+	store := newTestStore(t)
+	stale := seedObserver(t, store, "obs-nots", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-nots/status", `{"origin":"obs-nots","noise_floor":-95.5}`),
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-nots"); got == stale {
+		t.Errorf("last_seen = %s, want it advanced (no timestamp to judge — keep the old behaviour)", got)
+	}
+}
+
+func TestStatusWithUnparseableTimestampStillCountsAsLiveness(t *testing.T) {
+	store := newTestStore(t)
+	stale := seedObserver(t, store, "obs-badts", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-badts/status",
+			`{"origin":"obs-badts","status":"online","timestamp":"not-a-time","noise_floor":-95.5}`),
+		nil, nil, &Config{})
+
+	if got := observerLastSeen(t, store, "obs-badts"); got == stale {
+		t.Errorf("last_seen = %s, want it advanced (unjudgeable timestamp — keep the old behaviour)", got)
+	}
+}
+
+// A replayed snapshot is still the observer's last known state: metadata is
+// kept, only the liveness signal is suppressed.
+func TestStaleStatusPayloadStillUpdatesMetadata(t *testing.T) {
+	store := newTestStore(t)
+	seedObserver(t, store, "obs-zombie", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-zombie/status",
+			fmt.Sprintf(`{"origin":"obs-zombie","status":"online","timestamp":%q,"firmware":"v1.14.1","noise_floor":-95.5}`,
+				time.Now().UTC().AddDate(0, -4, 0).Format("2006-01-02T15:04:05.000000"))),
+		nil, nil, &Config{})
+
+	var firmware string
+	if err := store.db.QueryRow(`SELECT firmware FROM observers WHERE id = ?`, "obs-zombie").Scan(&firmware); err != nil {
+		t.Fatal(err)
+	}
+	if firmware != "v1.14.1" {
+		t.Errorf("firmware = %q, want %q (metadata from a replay is still the last known state)", firmware, "v1.14.1")
+	}
+}
+
+// No metrics sample either — a months-old reading filed at ingest time reads
+// as a present-tense measurement.
+func TestStaleStatusPayloadDoesNotInsertMetricsSample(t *testing.T) {
+	store := newTestStore(t)
+	seedObserver(t, store, "obs-zombie", 60)
+
+	handleMessage(store, "test", MQTTSource{Name: "test"},
+		liveStatusMsg("meshcore/LAX/obs-zombie/status",
+			statusPayload("obs-zombie", "online", time.Now().UTC().AddDate(0, -4, 0))),
+		nil, nil, &Config{})
+
+	var count int
+	if err := store.db.QueryRow(
+		`SELECT COUNT(*) FROM observer_metrics WHERE observer_id = ?`, "obs-zombie",
+	).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Errorf("observer_metrics rows = %d, want 0 (a replayed reading is not a new sample)", count)
+	}
+}