feat(ingestor): full-packet RF observations from mobile clients (#1905)

## What A CoreDrive RX drive already carries far more RF information than reaches CoreScope, and it was being discarded twice: once in the mobile app (every packet it could not attribute to a directly-heard node was dropped before queueing) and once here (the ingestor decodes the *complete* packet, then keeps only `heard_key`/`snr`/`rssi`/`lat`/`lon`). This captures what was being thrown away, at **zero extra airtime** — nothing new is transmitted. - **`transmissions.code1` / `code2`** — the transport codes were decoded on every packet and used only to derive `scope_name`, then dropped. Storing them turns "which repeater forwards which scope" from a re-parse into a query. - **An async backfill** re-parses the `raw_hex` already on disk, so months of scope history become queryable with no new data collection. - **`client_rx_observations`** — a new diagnostic table holding every decodable packet a phone heard, with route type, transport codes, scope name, path-hash size, the full forwarder chain and the forwarder. ## Why it is safe for existing deployments Both halves are **opt-in and default off** (`clientRxObservations.enabled`, and `fullRfLog` on the app side), so an existing deployment sees no behaviour change and no volume change on upgrade. The coverage invariant is untouched: `client_receptions` keeps its rule — 0-hop advert pubkey or FLOOD `path[last]`, ≥2-byte hash — and an unattributable packet writes **zero** coverage rows. `deriveHeardKey`, `buildClientReception` and `InsertClientReception` are unmodified except for one guard described below. ## Performance justification (touches the ingest hot path) - **Backfill:** keyset-paginated by `id` in 5000-row batches, a single forward scan, `rows.Close()` before `Begin()` so it never deadlocks against `SetMaxOpenConns(1)`, and commits per batch so live ingest interleaves. Termination is driven by rows *scanned*, not rows decoded — an earlier count-based loop would have stopped at the first batch containing an undecodable row and then written its completion guard, permanently stranding the rest. - **Guard row is written if and only if the loop ran to genuine exhaustion.** Every error path leaves it unwritten so the next startup retries. - **Per-packet cost:** one extra INSERT on the client topic when enabled, gated behind an opt-in flag. No new work on the observer path. - **New indexes** cover the prune (`rx_at`), the flood-grouping (`pkt_hash, rx_at`), the per-repeater query (`forwarder, rx_at`) and the scope query (`scope_name, rx_at`). Retention has its own shorter window — this table is diagnostic, not archival. ## Two firmware-derived correctness points - **`pkt_hash` is `ComputeContentHash()`**, byte-identical to `transmissions.hash`, so dark-traffic queries are a plain equality join rather than a translation layer. - **TRACE packets are refused.** TRACE repurposes the header path bytes as per-hop SNR values, so deriving a `heard_key` from them invents a node that never existed. `packetpath.PathBytesAreHops` existed but was never wired into the client path; it became reachable only because the app half now publishes packets it previously dropped locally. ## Testing Full ingestor suite green. Notable coverage: a FLOOD-routed TRACE writes zero coverage rows and NULL `forwarder`; a `direction: "tx"` message writes no observation; a DIRECT route never sets `forwarder`; two forwarder copies of one flood remain two rows; the backfill's multi-batch path is exercised with an undecodable row in the first page; and a forced error asserts the migration guard stays unwritten. --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-11-11 11:57:57 +00:00 · 2026-09-02 18:35:20 +02:00
1 parent 376c3e9f4a
commit 9e13e0b05f
@@ -1,6 +1,7 @@
 package main

 import (
+	"encoding/json"
 	"log"
 	"regexp"
 	"strings"
@@ -21,7 +22,7 @@ var clientPubkeyRe = regexp.MustCompile(`^[0-9a-f]{2,64}$`)
 // companion reports WHERE it directly heard a node, so we write a
 // client_receptions row and never touch the observers/observations tables.
 // rxPubkey is the companion pubkey from the topic (ACL-bound by the broker).
-func handleClientPacket(store *Store, tag, rxPubkey string, msg map[string]interface{}, channelKeys map[string]string) {
+func handleClientPacket(store *Store, cfg *Config, tag, rxPubkey string, msg map[string]interface{}, channelKeys map[string]string, regionKeys map[string][]byte) {
 	// The companion identity IS the (ACL-bound) topic pubkey. Reject non-hex
 	// topic segments so a no-ACL broker can't pollute the coverage tables, and
 	// never fall back to a payload-supplied id (that would defeat the ACL trust
@@ -72,13 +73,35 @@ func handleClientPacket(store *Store, tag, rxPubkey string, msg map[string]inter
 		rssiPtr = &v
 	}

-	rxAt, _ := resolveRxTime(msg, tag)
+	// Resolved once via resolveRxTimeCore and formatted twice (RFC3339 for
+	// coverage, rxTimeMillisLayout for observations) rather than reparsing the
+	// envelope timestamp per format — a second parse would re-validate/re-log
+	// the same timestamp and take its own independent time.Now() reading, so
+	// the coverage row and the observation for the SAME packet could straddle
+	// a second boundary on a fallback path.
+	rxTime, _ := resolveRxTimeCore(msg, tag)
+	rxAt := rxTime.Format(time.RFC3339)
+	ingestedAt := time.Now().UTC().Format(time.RFC3339)
 	isAdvert := decoded.Header.PayloadTypeName == "ADVERT"

+	if cfg.ClientRxObservationsEnabled() {
+		// rxAtMillis: UNIQUE(rx_pubkey, pkt_hash, rx_at) needs sub-second
+		// resolution to keep distinct forwarder copies of the same flood as
+		// separate rows — resolveRxTime's second-resolution RFC3339 would
+		// collapse them and ON CONFLICT DO NOTHING would silently drop all but
+		// the first.
+		rxAtMillis := rxTime.Format(rxTimeMillisLayout)
+		if obs := buildClientRxObservation(direction, rxPubkey, rawHex, rxAtMillis, ingestedAt, decoded, regionKeys, snrPtr, rssiPtr, lat, lon, accPtr); obs != nil {
+			if _, err := store.InsertClientRxObservation(obs); err != nil {
+				log.Printf("MQTT [%s] client observation insert: %v", tag, err)
+			}
+		}
+	}
+
 	rec, ok := buildClientReception(
 		rxPubkey,
-		direction, decoded.Header.RouteType, decoded.Path.Hops, decoded.Payload.PubKey, isAdvert,
-		snrPtr, rssiPtr, lat, lon, accPtr, rxAt, time.Now().UTC().Format(time.RFC3339),
+		direction, decoded.Header.RouteType, decoded.Header.PayloadType, decoded.Path.Hops, decoded.Payload.PubKey, isAdvert,
+		snrPtr, rssiPtr, lat, lon, accPtr, rxAt, ingestedAt,
 	)
 	if !ok {
 		return
@@ -89,7 +112,7 @@ func handleClientPacket(store *Store, tag, rxPubkey string, msg map[string]inter
 	// Remember the companion's self-reported name (sent as "origin") so the
 	// leaderboard can show a name even if this companion never advertised.
 	if name := stringField(msg, "origin"); name != "" {
-		if err := store.UpsertClientObserver(rec.RxPubkey, name, time.Now().UTC().Format(time.RFC3339)); err != nil {
+		if err := store.UpsertClientObserver(rec.RxPubkey, name, ingestedAt); err != nil {
 			log.Printf("MQTT [%s] client_observer upsert: %v", tag, err)
 		}
 	}
@@ -146,6 +169,8 @@ type ClientReception struct {
 // deriveHeardKey applies the RX capture HARD RULE: record only what the
 // companion heard itself and directly.
 //   - direction must be "rx".
+//   - payload type must not repurpose the header path bytes (TRACE stores
+//     per-hop SNR there, not node hashes — packetpath.PathBytesAreHops).
 //   - hops present AND a FLOOD route → the directly-heard node is the LAST hop
 //     (path[len-1] = the forwarder that just transmitted; each FLOOD forwarder
 //     appends its hash to the end). 1-byte (2 hex char) prefixes are rejected.
@@ -156,10 +181,13 @@ type ClientReception struct {
 //   - otherwise → not attributable (ok=false).
 //
 // Returns (heardKey lowercased, keylenBytes, src, ok).
-func deriveHeardKey(direction string, routeType int, hops []string, advertPubkey string, isAdvert bool) (string, int, string, bool) {
+func deriveHeardKey(direction string, routeType, payloadType int, hops []string, advertPubkey string, isAdvert bool) (string, int, string, bool) {
 	if !strings.EqualFold(direction, "rx") {
 		return "", 0, "", false
 	}
+	if !packetpath.PathBytesAreHops(byte(payloadType)) {
+		return "", 0, "", false
+	}
 	if len(hops) > 0 {
 		// FLOOD routes (TRANSPORT_FLOOD 0, FLOOD 1) APPEND each forwarder's hash to
 		// the END of the path, so path[last] is the immediate RF transmitter. DIRECT
@@ -185,7 +213,7 @@ func deriveHeardKey(direction string, routeType int, hops []string, advertPubkey
 // buildClientReception validates inputs and assembles a ClientReception, or
 // returns ok=false when the packet is not attributable / out of range.
 func buildClientReception(
-	rxPubkey, direction string, routeType int, hops []string, advertPubkey string, isAdvert bool,
+	rxPubkey, direction string, routeType, payloadType int, hops []string, advertPubkey string, isAdvert bool,
 	snr *float64, rssi *int, lat, lon float64, posAccM *float64, rxAt, ingestedAt string,
 ) (*ClientReception, bool) {
 	if rxPubkey == "" || rxAt == "" {
@@ -194,7 +222,7 @@ func buildClientReception(
 	if lat < -90 || lat > 90 || lon < -180 || lon > 180 {
 		return nil, false
 	}
-	heardKey, keylen, src, ok := deriveHeardKey(direction, routeType, hops, advertPubkey, isAdvert)
+	heardKey, keylen, src, ok := deriveHeardKey(direction, routeType, payloadType, hops, advertPubkey, isAdvert)
 	if !ok {
 		return nil, false
 	}
@@ -221,3 +249,121 @@ func (s *Store) InsertClientReception(r *ClientReception) (bool, error) {
 	n, _ := res.RowsAffected()
 	return n > 0, nil
 }
+
+// ClientRxObservation is one decodable packet a mobile client heard, whether or
+// not it was attributable to a directly-heard node. Diagnostic only: this table
+// is never a coverage source (see client_receptions for that).
+type ClientRxObservation struct {
+	RxPubkey    string
+	RxAt        string
+	IngestedAt  string
+	PktHash     string
+	RouteType   int
+	PayloadType int
+	HashSize    int
+	HopCount    int
+	Code1       *string
+	Code2       *string
+	ScopeName   *string
+	PathJSON    *string
+	Forwarder   *string
+	SNR         *float64
+	RSSI        *int
+	Lat         float64
+	Lon         float64
+	PosAccM     *float64
+}
+
+// buildClientRxObservation assembles a ClientRxObservation from a decoded
+// client packet, regardless of whether it is attributable to a directly-heard
+// node (deriveHeardKey/buildClientReception decide that separately, and this
+// function has no opinion on it). It does, however, require direction "rx"
+// (case-insensitively, matching deriveHeardKey's own check on the coverage
+// path) and returns nil otherwise: a companion's own outgoing transmission is
+// not RF it observed, and recording it would inflate this table's headline
+// signal — per-flood row multiplicity meant to measure forwarder
+// amplification of traffic actually heard over the air.
+func buildClientRxObservation(
+	direction, rxPubkey, rawHex, rxAt, ingestedAt string, decoded *DecodedPacket, regionKeys map[string][]byte,
+	snr *float64, rssi *int, lat, lon float64, posAccM *float64,
+) *ClientRxObservation {
+	if !strings.EqualFold(direction, "rx") {
+		return nil
+	}
+	obs := &ClientRxObservation{
+		RxPubkey:    rxPubkey,
+		RxAt:        rxAt,
+		IngestedAt:  ingestedAt,
+		PktHash:     ComputeContentHash(rawHex),
+		RouteType:   decoded.Header.RouteType,
+		PayloadType: decoded.Header.PayloadType,
+		HashSize:    decoded.Path.HashSize,
+		HopCount:    decoded.Path.HashCount, // declared count; len(Hops) can be short on a truncated path
+		SNR:         snr,
+		RSSI:        rssi,
+		Lat:         lat,
+		Lon:         lon,
+		PosAccM:     posAccM,
+	}
+	if decoded.TransportCodes != nil {
+		obs.Code1 = &decoded.TransportCodes.Code1
+		obs.Code2 = &decoded.TransportCodes.Code2
+		if decoded.TransportCodes.Code1 != "0000" {
+			sn := matchScope(regionKeys, byte(decoded.Header.PayloadType), decoded.payloadRaw, decoded.TransportCodes.Code1)
+			obs.ScopeName = &sn
+		}
+	}
+	// TRACE repurposes the header path bytes as per-hop SNR values, not node
+	// hashes (decoder.go), so they must not be recorded as a forwarder or a hop
+	// chain — same guard as deriveHeardKey. The row is still written (route
+	// type, transport codes and signal are all still valid diagnostics); only
+	// forwarder/path_json stay NULL.
+	if len(decoded.Path.Hops) > 0 && packetpath.PathBytesAreHops(byte(decoded.Header.PayloadType)) {
+		// The decoder emits hops as uppercase hex (decoder.go strings.ToUpper),
+		// but every other identifier in this schema is lowercase (rxPubkey,
+		// heard_key, ComputeContentHash output). Lowercase every hop here so
+		// path_json and forwarder agree in case within this table and stay
+		// joinable against them — otherwise a case-sensitive comparison (e.g.
+		// json_extract(path_json,'$[n]') = forwarder) silently matches zero
+		// rows instead of erroring.
+		lowerHops := make([]string, len(decoded.Path.Hops))
+		for i, h := range decoded.Path.Hops {
+			lowerHops[i] = strings.ToLower(h)
+		}
+		if b, err := json.Marshal(lowerHops); err == nil {
+			j := string(b)
+			obs.PathJSON = &j
+		}
+		// FLOOD routes accumulate forwarders at the END of the path, so
+		// path[last] is the node that actually transmitted. DIRECT routes
+		// consume from the FRONT, so their path[last] is the route's far
+		// end — never the transmitter. Same rule as deriveHeardKey.
+		if decoded.Header.RouteType == packetpath.RouteTransportFlood || decoded.Header.RouteType == packetpath.RouteFlood {
+			f := lowerHops[len(lowerHops)-1]
+			obs.Forwarder = &f
+		}
+	}
+	return obs
+}
+
+// InsertClientRxObservation writes one diagnostic observation. Idempotent via
+// UNIQUE(rx_pubkey, pkt_hash, rx_at); returns ins=false when the row existed.
+// Several rows per pkt_hash are EXPECTED — each is one forwarder's copy of the
+// same flood, and that multiplicity is the flood-amplification signal.
+func (s *Store) InsertClientRxObservation(o *ClientRxObservation) (bool, error) {
+	res, err := s.db.Exec(`
+		INSERT INTO client_rx_observations
+			(rx_pubkey, rx_at, ingested_at, pkt_hash, route_type, payload_type,
+			 code1, code2, scope_name, hash_size, hop_count, path_json, forwarder,
+			 snr, rssi, lat, lon, pos_acc_m)
+		VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+		ON CONFLICT(rx_pubkey, pkt_hash, rx_at) DO NOTHING`,
+		o.RxPubkey, o.RxAt, o.IngestedAt, o.PktHash, o.RouteType, o.PayloadType,
+		o.Code1, o.Code2, o.ScopeName, o.HashSize, o.HopCount, o.PathJSON, o.Forwarder,
+		o.SNR, o.RSSI, o.Lat, o.Lon, o.PosAccM)
+	if err != nil {
+		return false, err
+	}
+	n, _ := res.RowsAffected()
+	return n > 0, nil
+}
@@ -2,6 +2,7 @@ package main

 import (
 	"database/sql"
+	"encoding/hex"
 	"strings"
 	"testing"
 	"time"
@@ -45,6 +46,77 @@ func TestPruneOldClientReceptions(t *testing.T) {
 	}
 }

+// TestPruneClientRxObservationsUsesIndex verifies the diagnostic observations
+// reaper's DELETE ... WHERE rx_at < ? seeks idx_cro_prune rather than
+// full-scanning under the writer lock, mirroring
+// TestClientReceptionsRetentionUsesRxAtIndex for client_receptions below. May
+// pass immediately since idx_cro_prune already exists (Task 3) — in that case
+// this is a regression guard.
+func TestPruneClientRxObservationsUsesIndex(t *testing.T) {
+	s := newTestStore(t)
+	rows, err := s.db.Query(`EXPLAIN QUERY PLAN DELETE FROM client_rx_observations WHERE rx_at < ?`, "2026-01-01T00:00:00Z")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	plan := ""
+	for rows.Next() {
+		var id, parent, notused int
+		var detail string
+		if err := rows.Scan(&id, &parent, &notused, &detail); err != nil {
+			t.Fatal(err)
+		}
+		plan += detail + "\n"
+	}
+	if !strings.Contains(plan, "idx_cro_prune") {
+		t.Fatalf("retention DELETE should use idx_cro_prune, plan was:\n%s", plan)
+	}
+}
+
+// TestPruneOldClientRxObservations verifies the diagnostic observations reaper
+// deletes rows older than the window and keeps recent ones, and that days=0
+// disables it — mirroring TestPruneOldClientReceptions for the sibling table.
+func TestPruneOldClientRxObservations(t *testing.T) {
+	s := newTestStore(t)
+	now := time.Now().UTC()
+	recent := now.AddDate(0, 0, -1).Format(rxTimeMillisLayout)
+	old := now.AddDate(0, 0, -40).Format(rxTimeMillisLayout)
+
+	mk := func(rxAt, pktHash string) *ClientRxObservation {
+		return &ClientRxObservation{
+			RxPubkey: testCompanionPK, RxAt: rxAt, IngestedAt: rxAt, PktHash: pktHash,
+			RouteType: 1, PayloadType: 5, HashSize: 1, HopCount: 0, Lat: 51.0, Lon: 3.7,
+		}
+	}
+	if _, err := s.InsertClientRxObservation(mk(recent, "hash-recent")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.InsertClientRxObservation(mk(old, "hash-old")); err != nil {
+		t.Fatal(err)
+	}
+
+	if n, _ := s.PruneOldClientRxObservations(0); n != 0 {
+		t.Fatalf("days=0 must be a no-op, got %d", n)
+	}
+	n, err := s.PruneOldClientRxObservations(7)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Fatalf("expected 1 old observation pruned, got %d", n)
+	}
+	var remaining int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&remaining)
+	if remaining != 1 {
+		t.Fatalf("expected 1 observation remaining (recent), got %d", remaining)
+	}
+	var remainingHash string
+	s.db.QueryRow(`SELECT pkt_hash FROM client_rx_observations`).Scan(&remainingHash)
+	if remainingHash != "hash-recent" {
+		t.Fatalf("expected the recent row to survive, got pkt_hash=%q", remainingHash)
+	}
+}
+
 func TestClientReceptionsTableExists(t *testing.T) {
 	s := newTestStore(t)
 	cols := map[string]bool{}
@@ -170,46 +242,56 @@ func TestRxLeaderboardQueryIsIndexBacked(t *testing.T) {

 func TestDeriveHeardKey(t *testing.T) {
 	full := "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"
-	k, l, src, ok := deriveHeardKey("rx", packetpath.RouteFlood, nil, strings.ToUpper(full), true)
+	k, l, src, ok := deriveHeardKey("rx", packetpath.RouteFlood, PayloadADVERT, nil, strings.ToUpper(full), true)
 	if !ok || l != 32 || src != "advert" || k != full {
 		t.Fatalf("0-hop advert: got k=%q l=%d src=%q ok=%v", k, l, src, ok)
 	}
-	k, l, src, ok = deriveHeardKey("rx", packetpath.RouteFlood, []string{"aa", "bbccdd"}, "", false)
+	k, l, src, ok = deriveHeardKey("rx", packetpath.RouteFlood, PayloadGRP_TXT, []string{"aa", "bbccdd"}, "", false)
 	if !ok || k != "bbccdd" || l != 3 || src != "rxlog" {
 		t.Fatalf("flood path: got k=%q l=%d src=%q ok=%v", k, l, src, ok)
 	}
 	// DIRECT route: path[last] is the route's far end, not the transmitter — must be rejected.
-	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteDirect, []string{"aa", "bbccdd"}, "", false); ok {
+	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteDirect, PayloadGRP_TXT, []string{"aa", "bbccdd"}, "", false); ok {
 		t.Fatalf("direct-route path must be rejected")
 	}
-	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteTransportDirect, []string{"aa", "bbccdd"}, "", false); ok {
+	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteTransportDirect, PayloadGRP_TXT, []string{"aa", "bbccdd"}, "", false); ok {
 		t.Fatalf("transport-direct-route path must be rejected")
 	}
-	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteFlood, []string{"aa", "bb"}, "", false); ok {
+	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteFlood, PayloadGRP_TXT, []string{"aa", "bb"}, "", false); ok {
 		t.Fatalf("1-byte last hop should be rejected")
 	}
-	if _, _, _, ok = deriveHeardKey("tx", packetpath.RouteFlood, []string{"aabbcc"}, "", false); ok {
+	if _, _, _, ok = deriveHeardKey("tx", packetpath.RouteFlood, PayloadGRP_TXT, []string{"aabbcc"}, "", false); ok {
 		t.Fatalf("tx must be rejected")
 	}
-	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteFlood, nil, "", false); ok {
+	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteFlood, PayloadGRP_TXT, nil, "", false); ok {
 		t.Fatalf("no hops + non-advert must be rejected")
 	}
+	// TRACE repurposes the header path bytes as per-hop SNR values, not node
+	// hashes — a FLOOD-routed TRACE must never be attributable, even though the
+	// route type and hop shape are otherwise identical to the accepted case above.
+	if _, _, _, ok = deriveHeardKey("rx", packetpath.RouteFlood, PayloadTRACE, []string{"aa", "bbccdd"}, "", false); ok {
+		t.Fatalf("FLOOD-routed TRACE must be rejected (path bytes are SNR values, not node hashes)")
+	}
 }

 func TestBuildClientReception(t *testing.T) {
 	acc := 8.0
-	rec, ok := buildClientReception("companionpk", "rx", packetpath.RouteFlood, []string{"aa", "bbccdd"}, "", false,
+	rec, ok := buildClientReception("companionpk", "rx", packetpath.RouteFlood, PayloadGRP_TXT, []string{"aa", "bbccdd"}, "", false,
 		crF(-7.5), crI(-92), 51.05, 3.72, &acc, "2026-06-09T12:00:00Z", "2026-06-09T12:00:01Z")
 	if !ok || rec.HeardKey != "bbccdd" || rec.HeardKeyLen != 3 || rec.Src != "rxlog" {
 		t.Fatalf("bad reception: %+v ok=%v", rec, ok)
 	}
-	if _, ok := buildClientReception("c", "rx", packetpath.RouteDirect, []string{"bbccdd"}, "", false,
+	if _, ok := buildClientReception("c", "rx", packetpath.RouteDirect, PayloadGRP_TXT, []string{"bbccdd"}, "", false,
 		crF(-7.5), crI(-92), 51.05, 3.72, nil, "t", "t"); ok {
 		t.Fatal("direct-route path must be rejected (not the transmitter)")
 	}
-	if _, ok := buildClientReception("c", "rx", packetpath.RouteFlood, []string{"bbccdd"}, "", false, nil, nil, 99.0, 3.72, nil, "t", "t"); ok {
+	if _, ok := buildClientReception("c", "rx", packetpath.RouteFlood, PayloadGRP_TXT, []string{"bbccdd"}, "", false, nil, nil, 99.0, 3.72, nil, "t", "t"); ok {
 		t.Fatal("out-of-range lat must be rejected")
 	}
+	if _, ok := buildClientReception("c", "rx", packetpath.RouteFlood, PayloadTRACE, []string{"aa", "bbccdd"}, "", false,
+		crF(-7.5), crI(-92), 51.05, 3.72, nil, "t", "t"); ok {
+		t.Fatal("FLOOD-routed TRACE must be rejected (path bytes are SNR values, not node hashes)")
+	}
 }

 func TestInsertClientReceptionRoundTripAndIdempotent(t *testing.T) {
@@ -243,7 +325,7 @@ func TestHandleClientPacketRelayedAdvertWritesReception(t *testing.T) {
 		"RSSI":      -92.0,
 		"gps":       map[string]interface{}{"lat": 51.05, "lon": 3.72, "acc_m": 8.0},
 	}
-	handleClientPacket(s, "test", testCompanionPK, msg, nil)
+	handleClientPacket(s, &Config{}, "test", testCompanionPK, msg, nil, nil)

 	var obsName string
 	s.db.QueryRow(`SELECT name FROM client_observers WHERE pubkey=?`, testCompanionPK).Scan(&obsName)
@@ -265,7 +347,7 @@ func TestHandleClientPacketRelayedAdvertWritesReception(t *testing.T) {

 	// No GPS → no row.
 	const companion2 = "b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3"
-	handleClientPacket(s, "test", companion2, map[string]interface{}{"raw": advertHex, "direction": "rx"}, nil)
+	handleClientPacket(s, &Config{}, "test", companion2, map[string]interface{}{"raw": advertHex, "direction": "rx"}, nil, nil)
 	var n2 int
 	s.db.QueryRow(`SELECT COUNT(*) FROM client_receptions WHERE rx_pubkey=?`, companion2).Scan(&n2)
 	if n2 != 0 {
@@ -292,7 +374,7 @@ func TestHandleClientPacketZeroHopAdvertWritesReception(t *testing.T) {
 		"origin": "MyMob", "SNR": -7.0, "RSSI": -92.0,
 		"gps": map[string]interface{}{"lat": 51.05, "lon": 3.72, "acc_m": 8.0},
 	}
-	handleClientPacket(s, "test", testCompanionPK, msg, nil)
+	handleClientPacket(s, &Config{}, "test", testCompanionPK, msg, nil, nil)

 	var heardKey, src string
 	var keylen int
@@ -314,6 +396,331 @@ func TestHandleClientPacketZeroHopAdvertWritesReception(t *testing.T) {
 // pubkey from the topic that isn't lowercase hex (a no-ACL broker could publish
 // meshcore/client/!@#$/packets) writes nothing to either coverage table. Without
 // the clientPubkeyRe guard this fixture would insert a polluting row.
+// cfgWithObservations returns a Config with the observations gate on.
+func cfgWithObservations() *Config {
+	return &Config{ClientRxObservations: &ClientRxObservationsConfig{Enabled: true}}
+}
+
+// TestClientPacketWritesObservationNotCoverage verifies a DIRECT-route packet
+// (not attributable per deriveHeardKey) still produces one diagnostic
+// observation row while writing ZERO coverage rows — the two paths are
+// independent.
+func TestClientPacketWritesObservationNotCoverage(t *testing.T) {
+	s := newTestStore(t)
+	// header 0x16 = route_type 2 (DIRECT), payload_type 5 (GRP_TXT). DIRECT
+	// carries no transport codes. path byte 0x41 = hash_size 2, hop_count 1,
+	// then the 2-byte hop, then the payload.
+	// deriveHeardKey refuses a DIRECT route, so this must produce an
+	// observation row and ZERO coverage rows.
+	raw := "16" + "41" + "1a2b" + strings.Repeat("33", 16)
+	msg := map[string]interface{}{
+		"raw": raw, "direction": "rx", "SNR": 4.5, "RSSI": -101.0,
+		"timestamp": "2026-08-17T10:00:00.123Z",
+		"gps":       map[string]interface{}{"lat": 51.2, "lon": 4.4, "acc_m": 8.0},
+	}
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg, nil, nil)
+
+	var obs, cov int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&obs)
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_receptions`).Scan(&cov)
+	if obs != 1 {
+		t.Errorf("observations = %d, want 1", obs)
+	}
+	if cov != 0 {
+		t.Errorf("coverage rows = %d, want 0 — the invariant must hold", cov)
+	}
+
+	var hash string
+	var hopCount, hashSize int
+	var fwd sql.NullString
+	s.db.QueryRow(`SELECT pkt_hash, hop_count, hash_size, forwarder FROM client_rx_observations`).
+		Scan(&hash, &hopCount, &hashSize, &fwd)
+	if hash != ComputeContentHash(raw) {
+		t.Errorf("pkt_hash = %q, want ComputeContentHash value %q", hash, ComputeContentHash(raw))
+	}
+	if hopCount != 1 || hashSize != 2 {
+		t.Errorf("hop_count/hash_size = %d/%d, want 1/2", hopCount, hashSize)
+	}
+	// path[last] on a DIRECT route is the route's far end, not who transmitted
+	// (deriveHeardKey's "path[last] is flood-only" rule) — a production bug
+	// once got this wrong, so pin it here: forwarder must stay NULL, not the
+	// path's last hop.
+	if fwd.Valid {
+		t.Errorf("forwarder = %q, want NULL for a DIRECT route (path[last] is flood-only)", fwd.String)
+	}
+}
+
+// TestClientPacketFloodWritesBoth verifies a FLOOD-route packet (attributable)
+// writes both an observation row AND a coverage row, and that the observation's
+// forwarder is path[last] — the immediate RF transmitter.
+func TestClientPacketFloodWritesBoth(t *testing.T) {
+	s := newTestStore(t)
+	// header 0x15 = route_type 1 (FLOOD), payload_type 5. 2-byte hash, one hop
+	// → path[last] is the transmitter, so this IS attributable.
+	raw := "15" + "41" + "1a2b" + strings.Repeat("33", 16)
+	msg := map[string]interface{}{
+		"raw": raw, "direction": "rx", "SNR": 4.5, "RSSI": -101.0,
+		"timestamp": "2026-08-17T10:00:00.123Z",
+		"gps":       map[string]interface{}{"lat": 51.2, "lon": 4.4},
+	}
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg, nil, nil)
+
+	var obs, cov int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&obs)
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_receptions`).Scan(&cov)
+	if obs != 1 || cov != 1 {
+		t.Errorf("observations/coverage = %d/%d, want 1/1", obs, cov)
+	}
+	var fwd, pathJSON string
+	s.db.QueryRow(`SELECT forwarder, path_json FROM client_rx_observations`).Scan(&fwd, &pathJSON)
+	if fwd != "1a2b" {
+		t.Errorf("forwarder = %q, want 1a2b", fwd)
+	}
+	// The decoder emits hops uppercase; path_json must be lowercased to match
+	// forwarder (and every other identifier in this table), or a
+	// json_extract(path_json,'$[...]') = forwarder join silently returns zero
+	// rows instead of erroring.
+	if pathJSON != `["1a2b"]` {
+		t.Errorf("path_json = %q, want [\"1a2b\"] (lowercase, matching forwarder's case)", pathJSON)
+	}
+}
+
+// TestClientObservationsGateOff verifies that with clientRxObservations
+// disabled, a client packet writes zero observation rows.
+func TestClientObservationsGateOff(t *testing.T) {
+	s := newTestStore(t)
+	raw := "16" + "41" + "1a2b" + strings.Repeat("33", 16)
+	msg := map[string]interface{}{
+		"raw": raw, "direction": "rx",
+		"timestamp": "2026-08-17T10:00:00.123Z",
+		"gps":       map[string]interface{}{"lat": 51.2, "lon": 4.4},
+	}
+	handleClientPacket(s, &Config{}, "test", "aa11", msg, nil, nil)
+
+	var n int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&n)
+	if n != 0 {
+		t.Errorf("rows = %d, want 0 when the gate is off", n)
+	}
+}
+
+// TestClientObservationsTxDirectionSkipped verifies a companion's own
+// outgoing transmission (direction "tx") writes zero observation rows, even
+// though this FLOOD packet would otherwise be attributable (as
+// TestClientPacketFloodWritesBoth proves for the same shape with "rx"). The
+// observations gate sits above deriveHeardKey's own "rx" check, which runs
+// only on the coverage path — without a matching check here, a companion's
+// own transmissions would be recorded as observed RF and inflate the table's
+// headline per-flood forwarder-multiplicity signal.
+func TestClientObservationsTxDirectionSkipped(t *testing.T) {
+	s := newTestStore(t)
+	raw := "15" + "41" + "1a2b" + strings.Repeat("33", 16)
+	msg := map[string]interface{}{
+		"raw": raw, "direction": "tx",
+		"timestamp": "2026-08-17T10:00:00.123Z",
+		"gps":       map[string]interface{}{"lat": 51.2, "lon": 4.4},
+	}
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg, nil, nil)
+
+	var obs, cov int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&obs)
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_receptions`).Scan(&cov)
+	if obs != 0 {
+		t.Errorf("observations = %d, want 0 for a tx (self-transmission) packet", obs)
+	}
+	if cov != 0 {
+		t.Errorf("coverage rows = %d, want 0 for a tx packet", cov)
+	}
+}
+
+// TestClientObservationsMillisecondPrecisionAllowsDistinctRows is the
+// regression test for the second-resolution rx_at bug: two client packets
+// with the same raw and rx_pubkey, timestamps 40ms apart in the same second,
+// must produce TWO observation rows, not one collapsed by
+// UNIQUE(rx_pubkey, pkt_hash, rx_at) + ON CONFLICT DO NOTHING. resolveRxTime's
+// RFC3339 (second-resolution) formatting would collapse these; the dedicated
+// millisecond-precision rx_at for this table must not.
+func TestClientObservationsMillisecondPrecisionAllowsDistinctRows(t *testing.T) {
+	s := newTestStore(t)
+	raw := "16" + "41" + "1a2b" + strings.Repeat("33", 16)
+	// Derived from time.Now() (not hardcoded) so this test never crosses
+	// resolveRxTimeCore's >30d-stale reject, which would replace rx_at with
+	// ingest time and silently stop exercising the millisecond-distinctness
+	// this test exists to pin.
+	base := time.Now().UTC().Add(-time.Minute)
+	ts1 := base.Format(rxTimeMillisLayout)
+	ts2 := base.Add(40 * time.Millisecond).Format(rxTimeMillisLayout)
+	baseMsg := map[string]interface{}{
+		"raw": raw, "direction": "rx",
+		"gps": map[string]interface{}{"lat": 51.2, "lon": 4.4},
+	}
+	msg1 := map[string]interface{}{"timestamp": ts1}
+	msg2 := map[string]interface{}{"timestamp": ts2}
+	for k, v := range baseMsg {
+		msg1[k] = v
+		msg2[k] = v
+	}
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg1, nil, nil)
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg2, nil, nil)
+
+	var n int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&n)
+	if n != 2 {
+		t.Fatalf("observations = %d, want 2 — sub-second rx_at must keep both forwarder copies distinct", n)
+	}
+	var rxAts []string
+	rows, err := s.db.Query(`SELECT rx_at FROM client_rx_observations ORDER BY rx_at`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var rxAt string
+		if err := rows.Scan(&rxAt); err != nil {
+			t.Fatal(err)
+		}
+		rxAts = append(rxAts, rxAt)
+	}
+	if rxAts[0] != ts1 || rxAts[1] != ts2 {
+		t.Errorf("rx_at values = %v, want millisecond-distinct timestamps %v/%v", rxAts, ts1, ts2)
+	}
+}
+
+// TestClientObservationScopeNameFromTransportCode covers code1/code2/scope_name
+// — the field this whole table exists for ("which repeater forwards which
+// scope"). Uses the same precomputed matchScope vector as main_test.go's
+// TestMatchScope (key = SHA256("#test")[:16], payloadType=5, payload="hello"
+// -> code1 2AB5) rather than inventing a vector, so a wrong HMAC algorithm
+// cannot pass.
+func TestClientObservationScopeNameFromTransportCode(t *testing.T) {
+	s := newTestStore(t)
+	testKey, err := hex.DecodeString("9cd8fcf22a47333b591d96a2b848b73f") // SHA256("#test")[:16]
+	if err != nil {
+		t.Fatal(err)
+	}
+	regionKeys := map[string][]byte{"#test": testKey}
+	helloHex := hex.EncodeToString([]byte("hello"))
+
+	// Derived from time.Now() (not hardcoded) so this test never crosses
+	// resolveRxTimeCore's >30d-stale reject, which would replace rx_at with
+	// ingest time and break the WHERE rx_at = ? lookups below.
+	base := time.Now().UTC().Add(-time.Minute)
+	ts1 := base.Format(rxTimeMillisLayout)
+	ts2 := base.Add(time.Second).Format(rxTimeMillisLayout)
+
+	// header 0x14 = route_type 0 (TRANSPORT_FLOOD), payload_type 5 (GRP_TXT).
+	// Transport routes (0,3) carry code1/code2 right after the header, raw
+	// wire hex uppercased with no byte swap: bytes 2A B5 -> "2AB5", AA BB ->
+	// "AABB". path byte 0x00 = hash_size 1, hop_count 0 (no hops). Payload
+	// bytes are "hello" to match the precomputed matchScope vector.
+	raw := "14" + "2ab5" + "aabb" + "00" + helloHex
+	msg := map[string]interface{}{
+		"raw": raw, "direction": "rx",
+		"timestamp": ts1,
+		"gps":       map[string]interface{}{"lat": 51.2, "lon": 4.4},
+	}
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg, nil, regionKeys)
+
+	// ComputeContentHash deliberately excludes the transport-code bytes (so the
+	// same content dedups across scopes), so raw and raw2 below share one
+	// pkt_hash — distinguish the two inserted rows by rx_at, not pkt_hash.
+	var code1, code2, scopeName sql.NullString
+	if err := s.db.QueryRow(`SELECT code1, code2, scope_name FROM client_rx_observations WHERE rx_at = ?`, ts1).
+		Scan(&code1, &code2, &scopeName); err != nil {
+		t.Fatal(err)
+	}
+	if !code1.Valid || code1.String != "2AB5" {
+		t.Errorf("code1 = %v, want raw wire hex uppercased 2AB5 (bytes 2A B5, no byte swap)", code1)
+	}
+	if !code2.Valid || code2.String != "AABB" {
+		t.Errorf("code2 = %v, want raw wire hex uppercased AABB (bytes AA BB, no byte swap)", code2)
+	}
+	if !scopeName.Valid || scopeName.String != "#test" {
+		t.Errorf("scope_name = %v, want matched region #test", scopeName)
+	}
+
+	// Second packet: code1 "0000" is the real, non-scoped value — stored
+	// literally, and the != "0000" guard must leave scope_name NULL rather
+	// than attempt (and fail) a match.
+	raw2 := "14" + "0000" + "aabb" + "00" + helloHex
+	msg2 := map[string]interface{}{
+		"raw": raw2, "direction": "rx",
+		"timestamp": ts2,
+		"gps":       map[string]interface{}{"lat": 51.2, "lon": 4.4},
+	}
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg2, nil, regionKeys)
+
+	var code1b, scopeNameB sql.NullString
+	if err := s.db.QueryRow(`SELECT code1, scope_name FROM client_rx_observations WHERE rx_at = ?`, ts2).
+		Scan(&code1b, &scopeNameB); err != nil {
+		t.Fatal(err)
+	}
+	if !code1b.Valid || code1b.String != "0000" {
+		t.Errorf("code1 = %v, want literal 0000 (a real value, not NULL)", code1b)
+	}
+	if scopeNameB.Valid {
+		t.Errorf("scope_name = %q, want NULL when code1=0000", scopeNameB.String)
+	}
+}
+
+// TestHandleClientPacketFloodTraceWritesNoCoverage is the CRITICAL-1
+// regression test: a FLOOD-routed TRACE (header 0x25 = payload_type 9 << 2 |
+// route_type 1) repurposes the header path bytes as per-hop SNR values, not
+// node hashes (decoder.go). Before the packetpath.PathBytesAreHops guard was
+// wired into the client path, this shape yielded heardKey=<SNR bytes>,
+// src="rxlog" and a phantom client_receptions row. It must now produce ZERO
+// coverage rows, and the surviving diagnostic observation must carry NULL
+// forwarder/path_json (the route type, codes and signal are still valid
+// diagnostics, so the row itself is kept).
+func TestHandleClientPacketFloodTraceWritesNoCoverage(t *testing.T) {
+	s := newTestStore(t)
+	// header 0x25 = route_type 1 (FLOOD), payload_type 9 (TRACE).
+	// pathByte 0x41 = hash_size 2, hash_count 1 -> path bytes "1a2b" (SNR, not a hop hash).
+	// payload: tag(4) + authCode(4) + flags(1) = 9 bytes, so decodeTrace succeeds cleanly.
+	raw := "25" + "41" + "1a2b" + "aabbccdd1122334400"
+	decoded, err := DecodePacket(raw, nil, false)
+	if err != nil {
+		t.Fatalf("fixture must decode: %v", err)
+	}
+	if decoded.Header.PayloadTypeName != "TRACE" || decoded.Header.RouteType != packetpath.RouteFlood {
+		t.Fatalf("fixture sanity check failed: payloadType=%s routeType=%d", decoded.Header.PayloadTypeName, decoded.Header.RouteType)
+	}
+	if len(decoded.Path.Hops) != 1 || decoded.Path.Hops[0] != "1A2B" {
+		t.Fatalf("fixture sanity check failed: hops=%v, want a 2-byte SNR entry decoded as a hop", decoded.Path.Hops)
+	}
+
+	msg := map[string]interface{}{
+		"raw": raw, "direction": "rx", "SNR": 4.5, "RSSI": -101.0,
+		"timestamp": "2026-08-17T10:00:00.123Z",
+		"gps":       map[string]interface{}{"lat": 51.2, "lon": 4.4, "acc_m": 8.0},
+	}
+	handleClientPacket(s, cfgWithObservations(), "test", "aa11", msg, nil, nil)
+
+	var cov int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_receptions`).Scan(&cov)
+	if cov != 0 {
+		t.Fatalf("coverage rows = %d, want 0 — a FLOOD-routed TRACE must never be attributable", cov)
+	}
+
+	var obs int
+	var fwd, pathJSON sql.NullString
+	if err := s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&obs); err != nil {
+		t.Fatal(err)
+	}
+	if obs != 1 {
+		t.Fatalf("observations = %d, want 1 (the packet was genuinely heard; only forwarder/path_json must be NULL)", obs)
+	}
+	if err := s.db.QueryRow(`SELECT forwarder, path_json FROM client_rx_observations`).Scan(&fwd, &pathJSON); err != nil {
+		t.Fatal(err)
+	}
+	if fwd.Valid {
+		t.Errorf("forwarder = %q, want NULL for a TRACE packet (path bytes are SNR values, not a node hash)", fwd.String)
+	}
+	if pathJSON.Valid {
+		t.Errorf("path_json = %q, want NULL for a TRACE packet (path bytes are SNR values, not a hop chain)", pathJSON.String)
+	}
+}
+
 func TestHandleClientPacketRejectsNonHexPubkey(t *testing.T) {
 	s := newTestStore(t)
 	advertHex := "11451000D818206D3AAC152C8A91F89957E6D30CA51F36E28790228971C473B755F244F718754CF5EE4A2FD58D944466E42CDED140C66D0CC590183E32BAF40F112BE8F3F2BDF6012B4B2793C52F1D36F69EE054D9A05593286F78453E56C0EC4A3EB95DDA2A7543FCCC00B939CACC009278603902FC12BCF84B706120526F6F6620536F6C6172"
@@ -323,7 +730,7 @@ func TestHandleClientPacketRejectsNonHexPubkey(t *testing.T) {
 			"origin": "Spoof", "SNR": -7.0, "RSSI": -92.0,
 			"gps": map[string]interface{}{"lat": 51.05, "lon": 3.72, "acc_m": 8.0},
 		}
-		handleClientPacket(s, "test", bad, msg, nil)
+		handleClientPacket(s, &Config{}, "test", bad, msg, nil, nil)
 	}
 	var nRecept, nObs int
 	s.db.QueryRow(`SELECT COUNT(*) FROM client_receptions`).Scan(&nRecept)
@@ -44,22 +44,22 @@ type MQTTLegacy struct {

 // Config holds the ingestor configuration, compatible with the Node.js config.json format.
 type Config struct {
-	DBPath           string                  `json:"dbPath"`
-	MQTT             *MQTTLegacy             `json:"mqtt,omitempty"`
-	MQTTSources      []MQTTSource            `json:"mqttSources,omitempty"`
-	LogLevel         string                  `json:"logLevel,omitempty"`
-	ChannelKeysPath  string                  `json:"channelKeysPath,omitempty"`
-	ChannelKeys      map[string]string       `json:"channelKeys,omitempty"`
-	HashChannels     []string                `json:"hashChannels,omitempty"`
-	HashRegions      []string                `json:"hashRegions,omitempty"`
-	Retention        *RetentionConfig        `json:"retention,omitempty"`
-	Metrics          *MetricsConfig          `json:"metrics,omitempty"`
-	Runtime          *RuntimeConfig          `json:"runtime,omitempty"`
-	ClientRxCoverage *ClientRxCoverageConfig `json:"clientRxCoverage,omitempty"`
-	GeoFilter        *GeoFilterConfig        `json:"geo_filter,omitempty"`
+	DBPath               string                      `json:"dbPath"`
+	MQTT                 *MQTTLegacy                 `json:"mqtt,omitempty"`
+	MQTTSources          []MQTTSource                `json:"mqttSources,omitempty"`
+	LogLevel             string                      `json:"logLevel,omitempty"`
+	ChannelKeysPath      string                      `json:"channelKeysPath,omitempty"`
+	ChannelKeys          map[string]string           `json:"channelKeys,omitempty"`
+	HashChannels         []string                    `json:"hashChannels,omitempty"`
+	HashRegions          []string                    `json:"hashRegions,omitempty"`
+	Retention            *RetentionConfig            `json:"retention,omitempty"`
+	Metrics              *MetricsConfig              `json:"metrics,omitempty"`
+	Runtime              *RuntimeConfig              `json:"runtime,omitempty"`
+	ClientRxCoverage     *ClientRxCoverageConfig     `json:"clientRxCoverage,omitempty"`
+	ClientRxObservations *ClientRxObservationsConfig `json:"clientRxObservations,omitempty"`
+	GeoFilter            *GeoFilterConfig            `json:"geo_filter,omitempty"`
 	// PathTrust configures the minimum path-hash prefix length trusted as
-	// mapping/topology evidence (issue #1784). Read by the neighbor-edge
-	// builder; see packetpath.MeetsPathTrust.
+	// mapping/topology evidence (issue #1784).
 	PathTrust          *PathTrustConfig     `json:"pathTrust,omitempty"`
 	ForeignAdverts     *ForeignAdvertConfig `json:"foreignAdverts,omitempty"`
 	ValidateSignatures *bool                `json:"validateSignatures,omitempty"`
@@ -76,7 +76,7 @@ type Config struct {
 	obsIATAWhitelistOnce   sync.Once

 	// ObserverBlacklist is a list of observer public keys to drop at ingest.
-	// Messages from blacklisted observers are silently discarded — no DB writes,
+	// Messages from blacklisted observers are silently discarded â€” no DB writes,
 	// no UpsertObserver, no observations, no metrics.
 	ObserverBlacklist []string `json:"observerBlacklist,omitempty"`

@@ -85,7 +85,7 @@ type Config struct {
 	obsBlacklistOnce      sync.Once

 	// NeighborEdgesMaxAgeDays controls neighbor_edges row retention
-	// (#1287 — moved from cmd/server). 0 = default 5.
+	// (#1287 â€” moved from cmd/server). 0 = default 5.
 	NeighborEdgesMaxAgeDays int `json:"neighborEdgesMaxAgeDays,omitempty"`

 	// IngestBufferSize caps the in-memory queue (number of MQTT messages) held
@@ -144,11 +144,24 @@ type ClientRxCoverageConfig struct {
 }

 // ClientRxCoverageEnabled reports whether the opt-in mobile client-RX coverage
-// feature is on. Absent/nil ⇒ off (the safe default).
+// feature is on. Absent/nil â‡’ off (the safe default).
 func (c *Config) ClientRxCoverageEnabled() bool {
 	return c.ClientRxCoverage != nil && c.ClientRxCoverage.Enabled
 }

+// ClientRxObservationsConfig controls the opt-in diagnostic RF observation
+// table. Separate from clientRxCoverage: a deployment may well want coverage
+// without the far higher-volume diagnostic stream.
+type ClientRxObservationsConfig struct {
+	Enabled bool `json:"enabled"`
+}
+
+// ClientRxObservationsEnabled reports whether diagnostic client RF observations
+// are recorded. Default false.
+func (c *Config) ClientRxObservationsEnabled() bool {
+	return c.ClientRxObservations != nil && c.ClientRxObservations.Enabled
+}
+
 // RetentionConfig controls how long stale nodes are kept before being moved to inactive_nodes.
 type RetentionConfig struct {
 	NodeDays     int `json:"nodeDays"`
@@ -167,6 +180,10 @@ type RetentionConfig struct {
 	// coverage rows in client_receptions / client_observers; 0 disables. Bounds
 	// the table the opt-in coverage feature would otherwise grow without limit.
 	ClientRxDays int `json:"clientRxDays"`
+	// ClientRxObsDays is the retention window (by rx_at) for the diagnostic
+	// client_rx_observations table; 0 disables. Shorter than clientRxDays â€”
+	// this table is diagnostic, not archival.
+	ClientRxObsDays int `json:"clientRxObsDays"`
 }

 // PacketDaysOrZero returns the configured retention.packetDays or 0
@@ -196,6 +213,14 @@ func (c *Config) ClientRxDaysOrZero() int {
 	return 0
 }

+// ClientRxObsDaysOrZero returns retention.clientRxObsDays or 0 (disabled).
+func (c *Config) ClientRxObsDaysOrZero() int {
+	if c.Retention != nil && c.Retention.ClientRxObsDays > 0 {
+		return c.Retention.ClientRxObsDays
+	}
+	return 0
+}
+
 // MetricsConfig controls observer metrics collection.
 type MetricsConfig struct {
 	SampleIntervalSec int `json:"sampleIntervalSec"`
@@ -318,7 +343,7 @@ func LoadConfig(path string) (*Config, error) {
 		if !errors.Is(err, os.ErrNotExist) {
 			return nil, fmt.Errorf("reading config %s: %w", path, err)
 		}
-		// Config file doesn't exist — use defaults (zero-config mode)
+		// Config file doesn't exist â€” use defaults (zero-config mode)
 		log.Printf("config file %s not found, using sensible defaults", path)
 	} else {
 		if err := json.Unmarshal(data, &cfg); err != nil {
@@ -331,7 +356,7 @@ func LoadConfig(path string) (*Config, error) {
 		cfg.DBPath = v
 	}
 	if v := os.Getenv("MQTT_BROKER"); v != "" {
-		// Single broker from env — create a source
+		// Single broker from env â€” create a source
 		topic := os.Getenv("MQTT_TOPIC")
 		if topic == "" {
 			topic = "meshcore/#"
@@ -374,10 +399,10 @@ func LoadConfig(path string) (*Config, error) {
 //
 // Scheme mapping:
 //
-//	mqtt://  → tcp://   (paho plain TCP)
-//	mqtts:// → ssl://   (paho TLS over TCP)
-//	ws://               (paho WebSocket — passed through, no mapping needed)
-//	wss://              (paho WebSocket TLS — passed through, no mapping needed)
+//	mqtt://  â†’ tcp://   (paho plain TCP)
+//	mqtts:// â†’ ssl://   (paho TLS over TCP)
+//	ws://               (paho WebSocket â€” passed through, no mapping needed)
+//	wss://              (paho WebSocket TLS â€” passed through, no mapping needed)
 func (c *Config) ResolvedSources() []MQTTSource {
 	for i := range c.MQTTSources {
 		// paho uses tcp:// and ssl:// for plain MQTT; ws:// and wss:// are accepted natively.
@@ -387,7 +412,7 @@ func (c *Config) ResolvedSources() []MQTTSource {
 		} else if strings.HasPrefix(b, "mqtts://") {
 			c.MQTTSources[i].Broker = "ssl://" + b[8:]
 		}
-		// ws:// and wss:// pass through unchanged — paho handles WebSocket
+		// ws:// and wss:// pass through unchanged â€” paho handles WebSocket
 		// connections natively via gorilla/websocket.
 	}
 	return c.MQTTSources
@@ -1,11 +1,10 @@
 package main

 import (
+	"github.com/meshcore-analyzer/packetpath"
 	"os"
 	"path/filepath"
 	"testing"
-
-	"github.com/meshcore-analyzer/packetpath"
 )

 func TestLoadConfigValidJSON(t *testing.T) {
@@ -429,7 +428,7 @@ func TestResolvedSourcesSchemeMapping(t *testing.T) {
 }

 // TestLoadConfigWSSource verifies that a WebSocket MQTT source round-trips through
-// LoadConfig correctly — username/password preserved, scheme unchanged.
+// LoadConfig correctly â€” username/password preserved, scheme unchanged.
 func TestLoadConfigWSSource(t *testing.T) {
 	t.Setenv("DB_PATH", "")
 	t.Setenv("MQTT_BROKER", "")
@@ -499,7 +498,33 @@ func TestIngestBufferSizeOrDefault(t *testing.T) {
 	}
 }

-// ─── #1784: GetPathTrust ────────────────────────────────────────────────────
+func TestClientRxObservationsGate(t *testing.T) {
+	var c Config
+	if c.ClientRxObservationsEnabled() {
+		t.Error("default should be disabled")
+	}
+	c.ClientRxObservations = &ClientRxObservationsConfig{Enabled: true}
+	if !c.ClientRxObservationsEnabled() {
+		t.Error("explicit enable not honoured")
+	}
+	c.ClientRxObservations = &ClientRxObservationsConfig{Enabled: false}
+	if c.ClientRxObservationsEnabled() {
+		t.Error("explicit disable not honoured")
+	}
+	if got := c.ClientRxObsDaysOrZero(); got != 0 {
+		t.Errorf("unset retention = %d, want 0", got)
+	}
+	c.Retention = &RetentionConfig{ClientRxObsDays: 14}
+	if got := c.ClientRxObsDaysOrZero(); got != 14 {
+		t.Errorf("retention = %d, want 14", got)
+	}
+	c.Retention = &RetentionConfig{ClientRxObsDays: 0}
+	if got := c.ClientRxObsDaysOrZero(); got != 0 {
+		t.Errorf("retention=0 = %d, want 0", got)
+	}
+}
+
+// --- #1784: GetPathTrust ---

 func TestGetPathTrustDefaults(t *testing.T) {
 	cfg := &Config{}
@@ -68,6 +68,51 @@ func TestClientRxCoverageGateOn(t *testing.T) {
 	}
 }

+// TestClientRxCoverageGateOffDoesNotFallThroughToObserverPath is the
+// CRITICAL-adjacent regression test: with the feature OFF, a
+// meshcore/client/<pubkey>/packets message must be dropped outright, never
+// fall through to the observer packet path below. Before the fix, the
+// enable-gate lived INSIDE the topic match
+// (cfg.ClientRxCoverageEnabled() && parts[1]=="client" && ...), so a disabled
+// gate made the whole condition false and the message fell through: the
+// observer path takes parts[1] ("client") as a region and parts[2] (the
+// companion pubkey) as an observer id, creating a bogus "client" region and
+// registering the phone as an observer — worse now that fullRfLog multiplies
+// client-topic volume. Asserts zero client_receptions rows (already covered
+// by TestClientRxCoverageGateOff) AND zero observer/region pollution.
+func TestClientRxCoverageGateOffDoesNotFallThroughToObserverPath(t *testing.T) {
+	store := newTestStore(t)
+	source := MQTTSource{Name: "test"}
+	cfg := &Config{} // ClientRxCoverage nil ⇒ disabled
+
+	handleMessage(store, "test", source, clientCoverageMsg(), nil, nil, cfg)
+
+	if n := clientReceptionCount(t, store); n != 0 {
+		t.Fatalf("feature OFF: expected 0 client_receptions rows, got %d", n)
+	}
+	var observerRows int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, testCompanionPK).Scan(&observerRows); err != nil {
+		t.Fatal(err)
+	}
+	if observerRows != 0 {
+		t.Fatalf("feature OFF: the companion pubkey must not be registered as an observer, got %d rows", observerRows)
+	}
+	var clientRegionRows int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE iata = 'client'`).Scan(&clientRegionRows); err != nil {
+		t.Fatal(err)
+	}
+	if clientRegionRows != 0 {
+		t.Fatalf("feature OFF: no observer should be registered under a bogus 'client' region, got %d rows", clientRegionRows)
+	}
+	var txRows int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM transmissions`).Scan(&txRows); err != nil {
+		t.Fatal(err)
+	}
+	if txRows != 0 {
+		t.Fatalf("feature OFF: the client-topic packet must not be ingested as an ordinary observer packet, got %d transmissions rows", txRows)
+	}
+}
+
 // TestClientRxCoverageBlacklistedDropped verifies the #1 fix: a blacklisted
 // operator cannot skirt the observer blacklist via the client topic. With the
 // feature ON but the companion pubkey blacklisted, no row is written. Without
@@ -281,12 +281,20 @@ func applySchema(db *sql.DB) error {
 			decoded_json TEXT,
 			from_pubkey TEXT,
 			last_seen INTEGER NOT NULL DEFAULT 0,
-			created_at TEXT DEFAULT (datetime('now'))
+			created_at TEXT DEFAULT (datetime('now')),
+			code1 TEXT,
+			code2 TEXT
 		);

 		CREATE INDEX IF NOT EXISTS idx_transmissions_hash ON transmissions(hash);
 		CREATE INDEX IF NOT EXISTS idx_transmissions_first_seen ON transmissions(first_seen);
 		CREATE INDEX IF NOT EXISTS idx_transmissions_payload_type ON transmissions(payload_type);
+		-- idx_tx_code1 is created by the transport_codes_v1 migration below,
+		-- after the ALTER runs — same reasoning as idx_transmissions_from_pubkey
+		-- and idx_tx_last_seen_zero above: a legacy DB (table-exists,
+		-- column-missing) would trip on this CREATE INDEX referencing code1
+		-- before the column exists, since CREATE TABLE IF NOT EXISTS is a
+		-- no-op against a pre-existing table.
 		-- idx_transmissions_from_pubkey is created by the from_pubkey_v1
 		-- migration after the column is added on legacy DBs (#1143).
 		-- idx_tx_last_seen_zero (partial, WHERE last_seen=0) is created by
@@ -338,6 +346,37 @@ func applySchema(db *sql.DB) error {
 			name      TEXT,
 			last_seen TEXT
 		);
+
+		-- Diagnostic RF observations from mobile clients. Unlike client_receptions
+		-- this holds EVERY decodable packet, attributable or not, so it must never
+		-- be used for coverage. pkt_hash is ComputeContentHash() — identical to
+		-- transmissions.hash — so dark-traffic queries are a plain equality join.
+		CREATE TABLE IF NOT EXISTS client_rx_observations (
+			id            INTEGER PRIMARY KEY AUTOINCREMENT,
+			rx_pubkey     TEXT NOT NULL,
+			rx_at         TEXT NOT NULL,
+			ingested_at   TEXT NOT NULL,
+			pkt_hash      TEXT NOT NULL,
+			route_type    INTEGER NOT NULL,
+			payload_type  INTEGER NOT NULL,
+			code1         TEXT,
+			code2         TEXT,
+			scope_name    TEXT,
+			hash_size     INTEGER NOT NULL,
+			hop_count     INTEGER NOT NULL,
+			path_json     TEXT,
+			forwarder     TEXT,
+			snr           REAL,
+			rssi          INTEGER,
+			lat           REAL NOT NULL,
+			lon           REAL NOT NULL,
+			pos_acc_m     REAL,
+			UNIQUE(rx_pubkey, pkt_hash, rx_at)
+		);
+		CREATE INDEX IF NOT EXISTS idx_cro_prune     ON client_rx_observations(rx_at);
+		CREATE INDEX IF NOT EXISTS idx_cro_hash      ON client_rx_observations(pkt_hash, rx_at);
+		CREATE INDEX IF NOT EXISTS idx_cro_forwarder ON client_rx_observations(forwarder, rx_at);
+		CREATE INDEX IF NOT EXISTS idx_cro_scope     ON client_rx_observations(scope_name, rx_at);
 	`
 	if _, err := db.Exec(schema); err != nil {
 		return fmt.Errorf("base schema: %w", err)
@@ -768,6 +807,29 @@ func applySchema(db *sql.DB) error {
 		log.Println("[migration] channel_hash casing normalization complete")
 	}

+	// Migration: transmissions.code1/code2 — the transport codes were decoded
+	// and discarded; storing them makes scope forwarding queryable per repeater.
+	row = db.QueryRow("SELECT 1 FROM _migrations WHERE name = 'transport_codes_v1'")
+	if row.Scan(new(int)) != nil {
+		log.Println("[migration] Adding code1/code2 columns to transmissions...")
+		// Each ALTER is independent — ignore "duplicate column" so reruns (and
+		// the fresh-database path, where the base schema already created the
+		// columns) are safe. Any other failure must stop before the guard row
+		// is inserted, or prepareStatements' code1/code2 INSERT fails forever
+		// on every subsequent restart with no way to retry the migration.
+		for _, stmt := range []string{
+			`ALTER TABLE transmissions ADD COLUMN code1 TEXT DEFAULT NULL`,
+			`ALTER TABLE transmissions ADD COLUMN code2 TEXT DEFAULT NULL`,
+		} {
+			if _, err := db.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
+				return fmt.Errorf("transport_codes_v1 migration: %w", err)
+			}
+		}
+		db.Exec(`CREATE INDEX IF NOT EXISTS idx_tx_code1 ON transmissions(code1) WHERE code1 IS NOT NULL`)
+		db.Exec(`INSERT INTO _migrations (name) VALUES ('transport_codes_v1')`)
+		log.Println("[migration] code1/code2 columns added")
+	}
+
 	return nil
 }

@@ -780,8 +842,8 @@ func (s *Store) prepareStatements() error {
 	}

 	s.stmtInsertTransmission, err = s.db.Prepare(`
-		INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, channel_hash, scope_name, from_pubkey, last_seen)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, channel_hash, scope_name, from_pubkey, last_seen, code1, code2)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 	`)
 	if err != nil {
 		return err
@@ -959,6 +1021,8 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) {
 			scopeNameForDB(data),
 			nilIfEmpty(data.FromPubkey),
 			epochSecondsForLastSeen(rxTime),
+			nilIfEmpty(data.Code1),
+			nilIfEmpty(data.Code2),
 		)
 		if err != nil {
 			s.Stats.WriteErrors.Add(1)
@@ -1527,6 +1591,122 @@ func (s *Store) BackfillDefaultScopeAsync(regionKeys map[string][]byte) {
 	}()
 }

+// BackfillTransportCodesAsync populates transmissions.code1/code2 for rows
+// inserted before the transport_codes_v1 migration, by re-parsing raw_hex.
+// Runs in a background goroutine so it does not block MQTT startup.
+//
+// Batched: SQLite holds a single write connection (SetMaxOpenConns(1)), so a
+// multi-million-row UPDATE in one transaction would stall live ingest for the
+// whole run. 5k-row batches keep each transaction short.
+func (s *Store) BackfillTransportCodesAsync() {
+	s.backfillWg.Add(1)
+	go func() {
+		defer s.backfillWg.Done()
+		defer func() {
+			if r := recover(); r != nil {
+				log.Printf("[backfill] transport_codes async panic recovered: %v", r)
+			}
+		}()
+
+		var done int
+		if s.db.QueryRow("SELECT 1 FROM _migrations WHERE name = 'backfill_transport_codes_v1'").Scan(&done) == nil {
+			return // already ran
+		}
+
+		total, err := s.backfillTransportCodes(5000)
+		if err != nil {
+			log.Printf("[backfill] transport_codes: %v", err)
+			return // NOT recording the guard row — retry from scratch on next restart
+		}
+		s.Stats.IncBackfill("transport_codes")
+		log.Printf("[backfill] transport_codes populated for %d transmissions", total)
+		s.db.Exec(`INSERT INTO _migrations (name) VALUES ('backfill_transport_codes_v1')`)
+	}()
+}
+
+// backfillTransportCodes does the actual batched decode-and-update work,
+// returning the number of rows updated. Uses keyset pagination on id rather
+// than counting decoded items per page: a page can hit the LIMIT while
+// containing undecodable rows, so "items returned < batchSize" does NOT mean
+// "no rows remain" — that conflation previously caused the backfill to record
+// its completion migration after silently truncating on the first page with
+// an undecodable row. Termination is driven by how many rows the page
+// actually scanned (seen), not by how many decoded successfully; lastID
+// advances past undecodable rows too, so they are stepped over rather than
+// re-selected forever.
+//
+// Invariant: the caller records the backfill_transport_codes_v1 guard row iff
+// this returns a nil error, i.e. iff the loop ran to genuine exhaustion of all
+// matching rows. Every error path below returns non-nil immediately, without
+// finishing the run, so the guard is withheld and the next startup retries.
+func (s *Store) backfillTransportCodes(batchSize int) (int, error) {
+	var total int
+	var lastID int64
+	for {
+		rows, err := s.db.Query(`
+			SELECT id, raw_hex FROM transmissions
+			WHERE id > ? AND code1 IS NULL AND route_type IN (0, 3)
+			ORDER BY id
+			LIMIT ?`, lastID, batchSize)
+		if err != nil {
+			return total, fmt.Errorf("transport_codes select: %w", err)
+		}
+		type item struct {
+			id           int64
+			code1, code2 string
+		}
+		var items []item
+		var seen int
+		for rows.Next() {
+			var id int64
+			var raw string
+			if err := rows.Scan(&id, &raw); err != nil {
+				rows.Close()
+				return total, fmt.Errorf("transport_codes scan: %w", err)
+			}
+			seen++
+			lastID = id // advance past undecodable rows too, or they are re-selected forever
+			decoded, err := DecodePacket(raw, nil, false)
+			if err != nil || decoded.TransportCodes == nil {
+				continue
+			}
+			items = append(items, item{id, decoded.TransportCodes.Code1, decoded.TransportCodes.Code2})
+		}
+		rows.Close()
+		if seen == 0 {
+			break // no rows left matching the filter
+		}
+
+		if len(items) > 0 {
+			tx, err := s.db.Begin()
+			if err != nil {
+				return total, fmt.Errorf("transport_codes begin: %w", err)
+			}
+			stmt, err := tx.Prepare(`UPDATE transmissions SET code1 = ?, code2 = ? WHERE id = ?`)
+			if err != nil {
+				tx.Rollback()
+				return total, fmt.Errorf("transport_codes prepare: %w", err)
+			}
+			for _, it := range items {
+				if _, err := stmt.Exec(it.code1, it.code2, it.id); err != nil {
+					stmt.Close()
+					tx.Rollback()
+					return total, fmt.Errorf("transport_codes update id=%d: %w", it.id, err)
+				}
+			}
+			stmt.Close()
+			if err := tx.Commit(); err != nil {
+				return total, fmt.Errorf("transport_codes commit: %w", err)
+			}
+			total += len(items)
+		}
+		if seen < batchSize {
+			break // last partial page
+		}
+	}
+	return total, nil
+}
+
 // LogStats logs current operational metrics.
 func (s *Store) LogStats() {
 	log.Printf("[stats] tx_inserted=%d tx_dupes=%d obs_inserted=%d node_upserts=%d observer_upserts=%d write_errors=%d sig_drops=%d",
@@ -1770,6 +1950,8 @@ type PacketData struct {
 	ChannelHash       string // grouping key for channel queries (#762)
 	ScopeName         string // matched region name, or "" for unknown-scoped
 	IsTransportScoped bool   // true when route_type IN (0,3) AND Code1 ≠ "0000"
+	Code1             string // transport code 1 (scope), "" when the route carries none
+	Code2             string // transport code 2 (return-region hint), "" when absent
 	Region            string // observer region: payload > topic > source config (#788)
 	Foreign           bool   // true when ADVERT GPS lies outside configured geofilter (#730)
 	FromPubkey        string // pubkey of the originating node, for exact-match attribution (#1143)
@@ -1929,9 +2111,13 @@ func BuildPacketData(msg *MQTTPacketMessage, decoded *DecodedPacket, observerID,
 		}
 	}

-	if decoded.TransportCodes != nil && decoded.TransportCodes.Code1 != "0000" {
-		pd.IsTransportScoped = true
-		pd.ScopeName = matchScope(regionKeys, byte(decoded.Header.PayloadType), decoded.payloadRaw, decoded.TransportCodes.Code1)
+	if decoded.TransportCodes != nil {
+		pd.Code1 = decoded.TransportCodes.Code1
+		pd.Code2 = decoded.TransportCodes.Code2
+		if decoded.TransportCodes.Code1 != "0000" {
+			pd.IsTransportScoped = true
+			pd.ScopeName = matchScope(regionKeys, byte(decoded.Header.PayloadType), decoded.payloadRaw, decoded.TransportCodes.Code1)
+		}
 	}

 	// Populate from_pubkey at write time (#1143). ADVERTs carry the
@@ -1939,12 +1939,12 @@ func TestExtractObserverMetaNewFields(t *testing.T) {
 	msg := map[string]interface{}{
 		"model": "L1",
 		"stats": map[string]interface{}{
-			"noise_floor":  -112.5,
-			"battery_mv":   3720.0,
-			"uptime_secs":  86400.0,
-			"tx_air_secs":  100.0,
-			"rx_air_secs":  500.0,
-			"recv_errors":  3.0,
+			"noise_floor": -112.5,
+			"battery_mv":  3720.0,
+			"uptime_secs": 86400.0,
+			"tx_air_secs": 100.0,
+			"rx_air_secs": 500.0,
+			"recv_errors": 3.0,
 		},
 	}
 	meta := extractObserverMeta(msg)
@@ -1967,6 +1967,13 @@ func TestExtractObserverMetaNewFields(t *testing.T) {
 // rather than silently discarded. The unique dedup index is
 // (transmission_id, observer_idx, COALESCE(path_json, '')); observer_idx must
 // be non-NULL for the conflict to fire (SQLite treats NULL != NULL).
+//
+// (Blank line above is deliberate: gofmt's Go-1.19+ doc-comment formatter
+// rewrites a doubled straight-quote pair like '' into a single curly “ once
+// this comment is attached as a doc comment to the func below — detaching it
+// keeps the literal '' matching the real index expression in the
+// idx_observations_dedup CREATE UNIQUE INDEX statements elsewhere in this file.)
+
 func TestInsertObservationSNRFillIn(t *testing.T) {
 	s, err := OpenStore(tempDBPath(t))
 	if err != nil {
@@ -2068,9 +2075,9 @@ func TestPerObservationRawHex(t *testing.T) {

 	// First observation from observer A
 	pdA := &PacketData{
-		RawHex:    rawA,
-		Hash:      hash,
-		Timestamp: "2026-04-21T10:00:00Z",
+		RawHex:     rawA,
+		Hash:       hash,
+		Timestamp:  "2026-04-21T10:00:00Z",
 		ObserverID: "obs-A",
 		Direction:  &dir,
 		PathJSON:   "[]",
@@ -2085,9 +2092,9 @@ func TestPerObservationRawHex(t *testing.T) {

 	// Second observation from observer B (same hash, different raw bytes)
 	pdB := &PacketData{
-		RawHex:    rawB,
-		Hash:      hash,
-		Timestamp: "2026-04-21T10:00:01Z",
+		RawHex:     rawB,
+		Hash:       hash,
+		Timestamp:  "2026-04-21T10:00:01Z",
 		ObserverID: "obs-B",
 		Direction:  &dir,
 		PathJSON:   `["aabb"]`,
@@ -2258,6 +2265,262 @@ func TestScopeNameMigration(t *testing.T) {
 	}
 }

+func TestTransportCodesStored(t *testing.T) {
+	s, err := OpenStore(tempDBPath(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	// Byte order matters: Code1 is the RAW wire hex of the two bytes,
+	// uppercased, with NO little-endian swap (decoder.go:954). Bytes AA BB
+	// therefore read back as "AABB". Layout, per firmware packet_format.md:
+	// header + transport_codes(4) + path_len + path + payload.
+	// header 0x14 = route_type 0 (TRANSPORT_FLOOD), payload_type 5 (GRP_TXT).
+	// payload_type 5 keeps the advert validation path out of this test.
+	rawScoped := "14" + "AABB" + "CCDD" + "00" + strings.Repeat("11", 32)
+	// header 0x15 = route_type 1 (FLOOD), payload_type 5 — no transport codes.
+	rawUnscoped := "15" + "00" + strings.Repeat("22", 32)
+	// header 0x14 = route_type 0 (TRANSPORT_FLOOD) with all-zero transport
+	// codes. "0000" is stored literally (NOT NULL) — it is the meaningful
+	// "explicitly unscoped transport packet" case and must not collapse
+	// into the same NULL a non-transport packet gets.
+	rawZeroCodes := "14" + "0000" + "0000" + "00" + strings.Repeat("33", 32)
+
+	for _, raw := range []string{rawScoped, rawUnscoped, rawZeroCodes} {
+		pd := &PacketData{
+			RawHex: raw, Hash: ComputeContentHash(raw), Timestamp: "2026-08-17T10:00:00Z",
+			ObserverID: "obs1", RouteType: 0, PayloadType: 5,
+		}
+		if raw == rawUnscoped {
+			pd.RouteType = 1
+		}
+		if raw == rawScoped {
+			pd.Code1, pd.Code2 = "AABB", "CCDD"
+		}
+		if raw == rawZeroCodes {
+			pd.Code1, pd.Code2 = "0000", "0000"
+		}
+		if _, err := s.InsertTransmission(pd); err != nil {
+			t.Fatalf("insert: %v", err)
+		}
+	}
+
+	var c1, c2 *string
+	if err := s.db.QueryRow(`SELECT code1, code2 FROM transmissions WHERE hash = ?`,
+		ComputeContentHash(rawScoped)).Scan(&c1, &c2); err != nil {
+		t.Fatalf("read scoped: %v", err)
+	}
+	if c1 == nil || *c1 != "AABB" || c2 == nil || *c2 != "CCDD" {
+		t.Errorf("scoped codes = %v/%v, want AABB/CCDD", c1, c2)
+	}
+
+	if err := s.db.QueryRow(`SELECT code1, code2 FROM transmissions WHERE hash = ?`,
+		ComputeContentHash(rawUnscoped)).Scan(&c1, &c2); err != nil {
+		t.Fatalf("read unscoped: %v", err)
+	}
+	if c1 != nil || c2 != nil {
+		t.Errorf("unscoped codes = %v/%v, want NULL/NULL", c1, c2)
+	}
+
+	if err := s.db.QueryRow(`SELECT code1, code2 FROM transmissions WHERE hash = ?`,
+		ComputeContentHash(rawZeroCodes)).Scan(&c1, &c2); err != nil {
+		t.Fatalf("read zero-codes: %v", err)
+	}
+	if c1 == nil || *c1 != "0000" || c2 == nil || *c2 != "0000" {
+		t.Errorf("zero-codes codes = %v/%v, want literal \"0000\"/\"0000\" (not NULL)", c1, c2)
+	}
+}
+
+func TestBackfillTransportCodes(t *testing.T) {
+	s := newTestStore(t)
+
+	// Same layout and byte-order rules as TestTransportCodesStored: raw wire
+	// hex, no swap, so bytes AA BB read back as "AABB".
+	rawScoped := "14" + "AABB" + "CCDD" + "00" + strings.Repeat("11", 32)
+	rawFlood := "15" + "00" + strings.Repeat("22", 32)
+
+	// Simulate pre-migration rows: inserted directly, code1/code2 left NULL.
+	// route_type must match the raw header, or the backfill's route filter
+	// selects the wrong rows: 0 for the transport packet, 1 for the flood.
+	for _, tc := range []struct {
+		raw       string
+		routeType int
+	}{{rawScoped, 0}, {rawFlood, 1}} {
+		if _, err := s.db.Exec(
+			`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type)
+			 VALUES (?,?,?,?,?)`,
+			tc.raw, ComputeContentHash(tc.raw), "2026-08-01T00:00:00Z", tc.routeType, 5); err != nil {
+			t.Fatalf("seed: %v", err)
+		}
+	}
+
+	s.BackfillTransportCodesAsync()
+	s.backfillWg.Wait()
+
+	var c1 *string
+	if err := s.db.QueryRow(`SELECT code1 FROM transmissions WHERE hash = ?`,
+		ComputeContentHash(rawScoped)).Scan(&c1); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if c1 == nil || *c1 != "AABB" {
+		t.Errorf("backfilled code1 = %v, want AABB", c1)
+	}
+
+	if err := s.db.QueryRow(`SELECT code1 FROM transmissions WHERE hash = ?`,
+		ComputeContentHash(rawFlood)).Scan(&c1); err != nil {
+		t.Fatalf("read flood: %v", err)
+	}
+	if c1 != nil {
+		t.Errorf("flood code1 = %v, want NULL", c1)
+	}
+
+	// Idempotent: a second run is a no-op guarded by _migrations.
+	var done int
+	if err := s.db.QueryRow(
+		`SELECT 1 FROM _migrations WHERE name = 'backfill_transport_codes_v1'`).Scan(&done); err != nil {
+		t.Errorf("migration not recorded: %v", err)
+	}
+}
+
+// TestBackfillTransportCodes_MultiBatchWithUndecodableRow is a regression test for
+// a defect in the original loop: it terminated on "items returned < batchSize"
+// rather than "rows scanned < batchSize", so a full page containing even one
+// undecodable row broke the loop early and the exported wrapper then recorded
+// the completion migration — silently truncating the backfill and, because the
+// guard blocks all future runs, leaving every later row permanently NULL.
+// Seeds more rows than one batch holds, with an undecodable row in the first
+// page, and asserts every decodable row across every page got backfilled.
+func TestBackfillTransportCodes_MultiBatchWithUndecodableRow(t *testing.T) {
+	s := newTestStore(t)
+
+	const batchSize = 3
+
+	// route_type 0 (TRANSPORT_FLOOD) header, but only 1 byte follows — too
+	// short for the 4-byte transport codes field, so DecodePacket errors and
+	// this row is skipped. It must not stop rows after it in later pages from
+	// being backfilled.
+	badRaw := "14" + "AA"
+
+	// Payload byte varies per row so ComputeContentHash (which hashes only the
+	// payload, skipping path and transport-code bytes) gives each row a
+	// distinct hash despite the shared header/path/code shape.
+	var goodRaws []string
+	for i := 1; i <= 6; i++ {
+		goodRaws = append(goodRaws, fmt.Sprintf("14%04X%04X00%s", i, i+100, strings.Repeat(fmt.Sprintf("%02X", i), 32)))
+	}
+
+	// Insertion order fixes id order: bad row first, then the 6 good rows,
+	// giving 7 matching rows across 3 pages of 3 — batch 1 = {bad, good1,
+	// good2}, batch 2 = {good3, good4, good5}, batch 3 = {good6}.
+	if _, err := s.db.Exec(
+		`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type)
+		 VALUES (?,?,?,?,?)`,
+		badRaw, ComputeContentHash(badRaw), "2026-08-01T00:00:00Z", 0, 5); err != nil {
+		t.Fatalf("seed bad row: %v", err)
+	}
+	for _, raw := range goodRaws {
+		if _, err := s.db.Exec(
+			`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type)
+			 VALUES (?,?,?,?,?)`,
+			raw, ComputeContentHash(raw), "2026-08-01T00:00:00Z", 0, 5); err != nil {
+			t.Fatalf("seed good row: %v", err)
+		}
+	}
+
+	total, err := s.backfillTransportCodes(batchSize)
+	if err != nil {
+		t.Fatalf("backfillTransportCodes: %v", err)
+	}
+	if total != len(goodRaws) {
+		t.Errorf("backfillTransportCodes returned %d, want %d (one per good row)", total, len(goodRaws))
+	}
+
+	for i, raw := range goodRaws {
+		wantCode1 := fmt.Sprintf("%04X", i+1)
+		var c1 *string
+		if err := s.db.QueryRow(`SELECT code1 FROM transmissions WHERE hash = ?`,
+			ComputeContentHash(raw)).Scan(&c1); err != nil {
+			t.Fatalf("read good row %d: %v", i, err)
+		}
+		if c1 == nil || *c1 != wantCode1 {
+			t.Errorf("good row %d code1 = %v, want %s (row after the undecodable one in an earlier or later page must still be backfilled)", i, c1, wantCode1)
+		}
+	}
+
+	var badC1 *string
+	if err := s.db.QueryRow(`SELECT code1 FROM transmissions WHERE hash = ?`,
+		ComputeContentHash(badRaw)).Scan(&badC1); err != nil {
+		t.Fatalf("read bad row: %v", err)
+	}
+	if badC1 != nil {
+		t.Errorf("undecodable row code1 = %v, want NULL", badC1)
+	}
+}
+
+// TestBackfillTransportCodes_ErrorWithholdsGuardRow is a regression test for the
+// invariant that backfill_transport_codes_v1 is recorded iff and only if the
+// backfill actually ran to exhaustion of all matching rows. Forces the select
+// branch of backfillTransportCodes to fail mid-run by closing the store's DB
+// connection out from under it (simulating a transient DB error), then verifies
+// (a) the guard row is absent afterward and (b) reopening a fresh Store on the
+// same file and running the backfill again still processes the row the failed
+// run never reached.
+func TestBackfillTransportCodes_ErrorWithholdsGuardRow(t *testing.T) {
+	dir := t.TempDir()
+	dbPath := filepath.Join(dir, "test.db")
+
+	store1, err := OpenStore(dbPath)
+	if err != nil {
+		t.Fatalf("OpenStore: %v", err)
+	}
+
+	raw := "14" + "AABB" + "CCDD" + "00" + strings.Repeat("11", 32)
+	if _, err := store1.db.Exec(
+		`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type)
+		 VALUES (?,?,?,?,?)`,
+		raw, ComputeContentHash(raw), "2026-08-01T00:00:00Z", 0, 5); err != nil {
+		t.Fatalf("seed: %v", err)
+	}
+
+	if err := store1.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	if _, err := store1.backfillTransportCodes(5000); err == nil {
+		t.Fatal("backfillTransportCodes on a closed store returned nil error, want non-nil")
+	}
+
+	store2, err := OpenStore(dbPath)
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer store2.Close()
+
+	var done int
+	if err := store2.db.QueryRow(
+		`SELECT 1 FROM _migrations WHERE name = 'backfill_transport_codes_v1'`).Scan(&done); err != sql.ErrNoRows {
+		t.Fatalf("guard row query = (done=%d, err=%v), want sql.ErrNoRows — guard must be withheld after a failed run", done, err)
+	}
+
+	// Unobstructed, a subsequent run must still backfill the row the failed
+	// run never reached.
+	store2.BackfillTransportCodesAsync()
+	store2.backfillWg.Wait()
+
+	var c1 *string
+	if err := store2.db.QueryRow(`SELECT code1 FROM transmissions WHERE hash = ?`,
+		ComputeContentHash(raw)).Scan(&c1); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if c1 == nil || *c1 != "AABB" {
+		t.Errorf("code1 = %v, want AABB — subsequent run must backfill what the failed run skipped", c1)
+	}
+	if err := store2.db.QueryRow(
+		`SELECT 1 FROM _migrations WHERE name = 'backfill_transport_codes_v1'`).Scan(&done); err != nil {
+		t.Errorf("guard row not recorded after a successful run: %v", err)
+	}
+}
+
 // --- Feature 3: default_scope column on nodes (#899) ---

 func TestUpdateNodeDefaultScope(t *testing.T) {
@@ -2533,7 +2796,6 @@ func TestBuildPacketDataRegionFallsBackToTopic(t *testing.T) {
 	}
 }

-
 // TestBackfillPathJSONAsync verifies that the path_json backfill does NOT block
 // OpenStore from returning. MQTT connect happens immediately after OpenStore;
 // if the backfill is synchronous, MQTT would be delayed indefinitely on large DBs.
@@ -2960,3 +3222,66 @@ func TestUpdateNodeDefaultScope_EmptyScopeIsNoop(t *testing.T) {
 		t.Errorf("inactive_nodes.default_scope after empty-scope call = %q, want #belgium (DB-layer guard missing — #1534)", gotInactive)
 	}
 }
+
+func TestInsertClientRxObservation(t *testing.T) {
+	s := newTestStore(t)
+	code1, fwd := "AABB", "1a2b"
+	snr, rssi, posAccM := -7.5, -92, 8.5
+	o := &ClientRxObservation{
+		RxPubkey: "aa11", RxAt: "2026-08-17T10:00:00.123Z", IngestedAt: "2026-08-17T10:00:01Z",
+		PktHash: "0123456789abcdef", RouteType: 0, PayloadType: 4,
+		HashSize: 2, HopCount: 1, Code1: &code1, Forwarder: &fwd,
+		Lat: 51.2, Lon: 4.4, SNR: &snr, RSSI: &rssi, PosAccM: &posAccM,
+	}
+	ins, err := s.InsertClientRxObservation(o)
+	if err != nil || !ins {
+		t.Fatalf("insert: ins=%v err=%v", ins, err)
+	}
+	// Geometry/signal round-trip: lat/lon/snr/rssi/pos_acc_m are never read back
+	// anywhere else in the suite, so a field swap inside the 18-argument Exec
+	// (e.g. lat<->lon) would pass green. lat and lon are deliberately unequal so
+	// a swap is detectable.
+	var gotLat, gotLon, gotSNR, gotPosAccM float64
+	var gotRSSI int
+	if err := s.db.QueryRow(`SELECT lat, lon, snr, rssi, pos_acc_m FROM client_rx_observations WHERE pkt_hash = ?`, o.PktHash).
+		Scan(&gotLat, &gotLon, &gotSNR, &gotRSSI, &gotPosAccM); err != nil {
+		t.Fatalf("read back geometry: %v", err)
+	}
+	if gotLat != 51.2 || gotLon != 4.4 {
+		t.Errorf("lat/lon = %v/%v, want 51.2/4.4 (a lat<->lon swap would pass with symmetric values)", gotLat, gotLon)
+	}
+	if gotSNR != -7.5 {
+		t.Errorf("snr = %v, want -7.5", gotSNR)
+	}
+	if gotRSSI != -92 {
+		t.Errorf("rssi = %v, want -92", gotRSSI)
+	}
+	if gotPosAccM != 8.5 {
+		t.Errorf("pos_acc_m = %v, want 8.5", gotPosAccM)
+	}
+	// Same (rx_pubkey, pkt_hash, rx_at) → idempotent, no second row.
+	ins, err = s.InsertClientRxObservation(o)
+	if err != nil || ins {
+		t.Fatalf("re-insert: ins=%v err=%v (want false, nil)", ins, err)
+	}
+	// A second copy of the same flood from another forwarder, 40 ms later,
+	// MUST create its own row — that multiplicity is the amplification signal.
+	fwd2 := "3c4d"
+	o2 := *o
+	o2.RxAt, o2.Forwarder = "2026-08-17T10:00:00.163Z", &fwd2
+	if ins, err = s.InsertClientRxObservation(&o2); err != nil || !ins {
+		t.Fatalf("second copy: ins=%v err=%v", ins, err)
+	}
+	var n int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations WHERE pkt_hash = ?`, o.PktHash).Scan(&n)
+	if n != 2 {
+		t.Errorf("rows for pkt_hash = %d, want 2", n)
+	}
+}
+
+// NOTE: a TestTransportCodesV1MigrationFailsClosed regression test (forcing a
+// genuine ALTER failure to prove the migration doesn't record its guard row
+// on error) was attempted and removed — see final-fix-report.md for why the
+// failure mode isn't reachable from a black-box test of applySchema without
+// contorting production code. The fix itself (db.go, transport_codes_v1
+// block) still stands, matching the observers_clock_naive_v1 pattern.
@@ -103,6 +103,7 @@ func main() {

 	regionKeys := loadRegionKeys(cfg)
 	store.BackfillDefaultScopeAsync(regionKeys)
+	store.BackfillTransportCodesAsync()

 	// Subscribe-early + buffer (#1608): the MQTT subscription is brought up
 	// before startup maintenance so no packets are missed while the single
@@ -289,6 +290,18 @@ func main() {
 		}
 	}

+	// Diagnostic client_rx_observations retention: separate, typically
+	// shorter window than clientRxDays — this table is diagnostic, not
+	// archival. 0 = disabled.
+	clientRxObsDays := cfg.ClientRxObsDaysOrZero()
+	if clientRxObsDays > 0 {
+		if n, err := store.PruneOldClientRxObservations(clientRxObsDays); err != nil {
+			log.Printf("[prune] client_rx_observations: %v", err)
+		} else if n > 0 {
+			log.Printf("[prune] startup pruned %d client_rx_observations older than %d days", n, clientRxObsDays)
+		}
+	}
+
 	vacuumPages := cfg.IncrementalVacuumPages()
 	store.RunIncrementalVacuum(vacuumPages)

@@ -353,19 +366,36 @@ func main() {
 		log.Printf("[prune] auto-prune enabled: packets older than %d days will be removed daily", packetDays)
 	}

-	// Daily ticker for client-RX coverage retention (#1727).
-	if clientRxDays > 0 {
+	// Daily ticker for client-RX coverage retention (#1727), reused for the
+	// diagnostic client_rx_observations retention (Task 6) rather than
+	// starting a second ticker — the two flags are independent (0 disables
+	// each separately), so the ticker itself must run when either is set.
+	if clientRxDays > 0 || clientRxObsDays > 0 {
 		clientRxRetentionTicker := time.NewTicker(24 * time.Hour)
 		go func() {
 			for range clientRxRetentionTicker.C {
-				if n, err := store.PruneOldClientReceptions(clientRxDays); err != nil {
-					log.Printf("[prune] error: %v", err)
-				} else if n > 0 {
-					store.RunIncrementalVacuum(vacuumPages)
+				if clientRxDays > 0 {
+					if n, err := store.PruneOldClientReceptions(clientRxDays); err != nil {
+						log.Printf("[prune] error: %v", err)
+					} else if n > 0 {
+						store.RunIncrementalVacuum(vacuumPages)
+					}
+				}
+				if clientRxObsDays > 0 {
+					if n, err := store.PruneOldClientRxObservations(clientRxObsDays); err != nil {
+						log.Printf("[prune] client_rx_observations: %v", err)
+					} else if n > 0 {
+						store.RunIncrementalVacuum(vacuumPages)
+					}
 				}
 			}
 		}()
-		log.Printf("[prune] auto-prune enabled: client_receptions older than %d days will be removed daily", clientRxDays)
+		if clientRxDays > 0 {
+			log.Printf("[prune] auto-prune enabled: client_receptions older than %d days will be removed daily", clientRxDays)
+		}
+		if clientRxObsDays > 0 {
+			log.Printf("[prune] auto-prune enabled: client_rx_observations older than %d days will be removed daily", clientRxObsDays)
+		}
 	}

 	// Hourly WAL checkpoint to prevent unbounded WAL growth.
@@ -623,15 +653,27 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message,
 	// Mobile client RX coverage: dedicated topic meshcore/client/{PUBLIC_KEY}/packets.
 	// A roaming companion reports where it directly heard a node; handled in isolation
 	// from the observer/observations path. EMQX ACL binds parts[2] to the client's own key.
-	if cfg.ClientRxCoverageEnabled() && len(parts) >= 4 && parts[1] == "client" && parts[3] == "packets" {
-		// The observer blacklist (checked below) only runs on the observer path,
-		// so a blacklisted operator could otherwise skirt it via the client topic
-		// (#1). Enforce it here before any coverage write.
+	//
+	// The topic match and the enable-gate MUST be separate: matching on
+	// parts[1]=="client" always returns from this branch, whatever the config
+	// says. The gate only decides drop-vs-handle. Previously the gate sat
+	// inside the topic match, so a disabled gate made the whole condition
+	// false and fell through to the observer path below, where parts[1]
+	// ("client") would be taken as a region and the companion pubkey as an
+	// observer id — silently poisoning the region list, and worse now that
+	// fullRfLog multiplies client-topic volume.
+	if len(parts) >= 4 && parts[1] == "client" {
+		if !cfg.ClientRxCoverageEnabled() || parts[3] != "packets" {
+			return
+		}
+		// The observer blacklist (checked below on the observer path) only runs
+		// there, so a blacklisted operator could otherwise skirt it via the
+		// client topic (#1). Enforce it here before any coverage write.
 		if cfg.IsObserverBlacklisted(parts[2]) {
 			log.Printf("MQTT [%s] client %.8s blacklisted, dropping", tag, parts[2])
 			return
 		}
-		handleClientPacket(store, tag, parts[2], msg, channelKeys)
+		handleClientPacket(store, cfg, tag, parts[2], msg, channelKeys, regionKeys)
 		return
 	}

@@ -1323,21 +1365,46 @@ func firstNonEmpty(vals ...string) string {
 // ahead). Caller records this via Store.RecordNaiveSkew so the UI can flag
 // the observer (#1478).
 func resolveRxTime(msg map[string]interface{}, tag string) (string, int64) {
+	t, skew := resolveRxTimeCore(msg, tag)
+	return t.Format(time.RFC3339), skew
+}
+
+// rxTimeMillisLayout is the fixed three-decimal millisecond layout for
+// client_rx_observations.rx_at. UNIQUE(rx_pubkey, pkt_hash, rx_at) on that
+// table relies on sub-second resolution to keep distinct forwarder copies of
+// the same flood as separate rows — RFC3339's second resolution would
+// collapse them and ON CONFLICT DO NOTHING would silently drop all but the
+// first. Deliberately not time.RFC3339Nano, which strips trailing zeros, so
+// this stays a stable uniqueness/sort key. The single definition here is
+// shared by its only production caller (handleClientPacket, which formats a
+// resolveRxTimeCore result with it directly) and by tests that need to
+// derive an rx_at they can query back by.
+const rxTimeMillisLayout = "2006-01-02T15:04:05.000Z07:00"
+
+// resolveRxTimeCore holds the validation/clamping logic shared by
+// resolveRxTime and any caller that needs the resolved time in a different
+// format (handleClientPacket formats it with rxTimeMillisLayout for
+// client_rx_observations). Returns UTC time so callers format it themselves —
+// a caller that needs two different formatted strings for the SAME packet
+// calls this once and formats twice, instead of parsing/validating/logging
+// twice and risking two independent time.Now() reads straddling a second
+// boundary.
+func resolveRxTimeCore(msg map[string]interface{}, tag string) (time.Time, int64) {
 	now := time.Now().UTC()
 	raw, _ := msg["timestamp"].(string)
 	if raw == "" {
-		return now.Format(time.RFC3339), 0
+		return now, 0
 	}
 	t, naive, err := parseEnvelopeTime(raw)
 	if err != nil {
 		log.Printf("MQTT [%s] unparseable timestamp %q, using ingest time", tag, raw)
-		return now.Format(time.RFC3339), 0
+		return now, 0
 	}
 	// Hard reject: > 14h ahead is a genuine clock error (UTC+14 is the maximum
 	// standard offset, so nothing valid should be further ahead than that).
 	if t.After(now.Add(14 * time.Hour)) {
 		log.Printf("MQTT [%s] future timestamp %q, using ingest time", tag, raw)
-		return now.Format(time.RFC3339), 0
+		return now, 0
 	}
 	// Hard reject: > 30 days in the past is an RTC-reset node reporting a
 	// factory date (e.g. 2020-01-01). Such a value would permanently drag
@@ -1345,7 +1412,7 @@ func resolveRxTime(msg map[string]interface{}, tag string) (string, int64) {
 	// InsertTransmission. No legitimate buffered upload is that stale.
 	if t.Before(now.Add(-30 * 24 * time.Hour)) {
 		log.Printf("MQTT [%s] stale timestamp %q (>30d old), using ingest time", tag, raw)
-		return now.Format(time.RFC3339), 0
+		return now, 0
 	}
 	// Symmetric naive-timestamp clamp (issue #1463). Naive (zone-less) ISO
 	// values from observers in non-UTC zones are parsed as-if UTC, leaving a
@@ -1369,16 +1436,16 @@ func resolveRxTime(msg map[string]interface{}, tag string) (string, int64) {
 			// Per-message log was silenced in #1479 — chip + banner in the UI
 			// replace it.
 			deltaSec := int64(signed / time.Second)
-			return now.Format(time.RFC3339), deltaSec
+			return now, deltaSec
 		}
 	}
 	// Legacy soft clamp for zone-aware near-future values: any value ahead of
 	// now is from a slightly skewed observer clock — collapse to now so we
 	// don't render ⚠️ in the UI for live packets from those nodes.
 	if t.After(now) {
-		return now.Format(time.RFC3339), 0
+		return now, 0
 	}
-	return t.UTC().Format(time.RFC3339), 0
+	return t.UTC(), 0
 }

 // parseEnvelopeTime parses the MQTT envelope timestamp. Two on-wire forms
@@ -81,6 +81,29 @@ func (s *Store) PruneOldClientReceptions(days int) (int64, error) {
 	return n, nil
 }

+// PruneOldClientRxObservations deletes diagnostic client_rx_observations rows
+// older than `days` (by rx_at). Unlike client_receptions this table is
+// diagnostic, not archival, so it gets its own (typically shorter) window.
+// 0 disables. Owned by the ingestor writer (#1283). rx_at on this table is
+// stored at millisecond precision via rxTimeMillisLayout, not RFC3339 — the
+// cutoff must be formatted the same way so lexicographic comparison against
+// stored values stays correct.
+func (s *Store) PruneOldClientRxObservations(days int) (int64, error) {
+	if days <= 0 {
+		return 0, nil
+	}
+	cutoff := time.Now().UTC().AddDate(0, 0, -days).Format(rxTimeMillisLayout)
+	res, err := s.db.Exec(`DELETE FROM client_rx_observations WHERE rx_at < ?`, cutoff)
+	if err != nil {
+		return 0, fmt.Errorf("prune client_rx_observations: %w", err)
+	}
+	n, _ := res.RowsAffected()
+	if n > 0 {
+		log.Printf("[prune] deleted %d client_rx_observations older than %d days", n, days)
+	}
+	return n, nil
+}
+
 // SoftDeleteBlacklistedObservers marks observers in the blacklist as
 // inactive=1 so they are hidden from API responses. Owned by ingestor
 // per #1287. Runs once at startup.
@@ -13,7 +13,8 @@
    "observerPurgeDays": 0,
    "packetDays": 30,
    "clientRxDays": 30,
-    "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes."
+    "clientRxObsDays": 30,
+    "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). clientRxObsDays: diagnostic client_rx_observations rows (by rx_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRxObservations table. NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes."
  },
  "db": {
    "vacuumOnStartup": false,
@@ -374,6 +375,8 @@
  "_comment_compression": "Opt-in HTTP gzip middleware + WebSocket permessage-deflate. Both default to false — enable ONLY when your upstream reverse proxy is NOT already compressing. gzip: enables the gzipMiddleware wrapper around the HTTP handler. websocket: sets gorilla websocket Upgrader.EnableCompression. level: gzip compression level 1..9 (1=BestSpeed, 9=BestCompression, default 6). minSizeBytes: advisory minimum response size below which compression would not pay off. contentTypes: MIME allow-list — only responses with these Content-Type values are compressed. Already-compressed types (image/*, video/*, audio/*, application/zip, application/x-gzip, application/pdf, application/octet-stream) are always skipped, as are responses whose handler already set Content-Encoding. Omit contentTypes to use the built-in default allow-list.",
  "clientRxCoverage": { "enabled": false },
  "_comment_clientRxCoverage": "Opt-in mobile client-RX coverage (corescope-rx companions publishing GPS-tagged receptions to meshcore/client/<pubkey>/packets). Default OFF: when disabled the ingestor writes no client_receptions, the /api/rx-coverage|rx-leaderboard|nodes/{pubkey}/rx-coverage endpoints 404, and the UI hides the Coverage dashboard + Reach overlay. Set enabled=true to turn it on. SINGLE FLAG, BOTH PROCESSES: the ingestor and server each parse this same config.json, so this one clientRxCoverage.enabled entry gates both the ingest write path and the read endpoints — set it once, not per-process. TRUST: the feature requires an ACL-capable broker binding meshcore/client/{pubkey}/packets to that publisher; without ACLs the companion GPS is spoofable (see docs/client-rx-coverage.md). Retention: see retention.clientRxDays. Companion app + setup: https://github.com/efiten/corescope-rx.",
+  "clientRxObservations": { "enabled": false },
+  "_comment_clientRxObservations": "Opt-in diagnostic RF observations: every decodable packet a companion heard, whether or not it is attributable to a directly-heard node (client_rx_observations — never a coverage source; see clientRxCoverage for that). Default OFF. TOP-LEVEL FLAG: this is its own top-level Config field, NOT nested inside clientRxCoverage — config loading is plain json.Unmarshal with no DisallowUnknownFields, so nesting it under clientRxCoverage is silently ignored. Gated behind clientRxCoverage being reachable in the ingest control flow, but the JSON key sits alongside it. Requires the companion app's fullRfLog flag to actually upload unattributable packets, or this table stays empty even when enabled — see corescope-rx README. Retention: see retention.clientRxObsDays.",
  "_comment_channelKeys": "Hex keys for decrypting channel messages. Key name = channel display name. public channel key is well-known.",
  "_comment_hashChannels": "Channel names whose keys are derived via SHA256. Key = SHA256(name)[:16]. Listed here so the ingestor can auto-derive keys.",
  "hashRegions": [
@@ -123,6 +123,70 @@ Retention: the table grows on every submission, so set `retention.clientRxDays`
 rows older than N days (and stale `client_observers`); `0` disables it. Without it the table is
 unbounded.

+## Diagnostic observations — `client_rx_observations` (ingestor-owned)
+
+The client topic may also carry packets the companion could not attribute to a directly-heard
+node — a DIRECT-route packet with a path, for instance (see the capture HARD RULE above). Those
+packets are still decodable, and are optionally recorded as a diagnostic RF observation,
+independent of whether they produced a coverage row.
+
+**Not literally every decodable packet, though.** A packet still needs a `gps` fix and
+`direction: "rx"` to reach the decoder/observation write at all — `handleClientPacket` returns
+early (before `DecodePacket` even runs) when `gps` is missing or its `lat`/`lon` don't parse, and
+`direction: "tx"` (a companion's own outgoing transmission) is decoded but explicitly excluded
+from the observation write, the same as it already was from coverage. A diagnostic table silently
+requiring a GPS fix is a bit surprising, so: no `gps` → no observation row either, same constraint
+as coverage.
+
+- Written to `client_rx_observations` only, **never** to `client_receptions` — the coverage
+  invariant (only directly-heard nodes) is unchanged and unaffected by this feature.
+- Gated by its own flag, `"clientRxObservations": { "enabled": true }` — a **top-level** `Config`
+  field, not nested inside `clientRxCoverage` in the JSON. It IS gated behind `clientRxCoverage` in
+  the *control flow*: `handleClientPacket` (where the observation write lives) is only reached when
+  `clientRxCoverage.enabled` is true, so observations require coverage to be enabled even though the
+  two keys are siblings on disk:
+  ```json
+  {
+    "clientRxCoverage": { "enabled": true },
+    "clientRxObservations": { "enabled": true }
+  }
+  ```
+  Config loading is plain `json.Unmarshal` with no `DisallowUnknownFields`, so nesting
+  `clientRxObservations` under `clientRxCoverage` as written above is silently ignored — the key is
+  never read, the feature stays off, and nothing logs or errors. An ingestor without
+  `clientRxObservations.enabled` simply drops these packets (no table writes, no error).
+- Enabling the companion app's `fullRfLog` flag while `clientRxObservations.enabled` is `false` here
+  is pure waste: the phone spends mobile data uploading packets this ingestor decodes and discards,
+  with no row written and no warning anywhere. `fullRfLog` multiplies normal upload volume — see the
+  corescope-rx README.
+- The JSON payload shape from the companion app is **unchanged** either way — this is purely an
+  ingestor-side decision based on what `raw` decodes to, not a new field the app must send.
+- Captures routing detail the coverage path discards: `route_type`, `payload_type`,
+  `code1`/`code2` transport codes (route types 0/3 only), `scope_name` (matched against configured
+  region keys), `hash_size`, `hop_count`, the full forwarder path (`path_json`), and — for FLOOD
+  routes only — the immediate `forwarder`.
+- `rx_at` is stored at millisecond precision (unlike `client_receptions.rx_at`), because
+  `UNIQUE(rx_pubkey, pkt_hash, rx_at)` deliberately allows multiple rows per `pkt_hash`: each row is
+  one forwarder's copy of the same flood, and that multiplicity is the flood-amplification signal
+  this table exists to capture. Retention is `retention.clientRxObsDays` (separate from, and
+  typically shorter than, `retention.clientRxDays` — this table is diagnostic, not archival).
+- `pkt_hash` (`ComputeContentHash`) deliberately excludes both the transport-code bytes and the
+  path bytes, so distinctness inside `UNIQUE(rx_pubkey, pkt_hash, rx_at)` rests entirely on `rx_at`.
+  On the happy path that's fine — real receive times come from the envelope timestamp at
+  millisecond resolution, and same-millisecond collisions aren't physical on a half-duplex LoRa
+  radio. But on any fallback path (missing/unparseable/implausible timestamp — see
+  `resolveRxTimeCore`), every packet in a buffered upload batch is stamped with the same ingest-time
+  `rx_at`, and distinct forwarder copies of one flood inside that batch collapse into a single row
+  via `ON CONFLICT DO NOTHING`. Not a correctness bug — the constraint is doing exactly what it's
+  told — but it means a buffered/late upload with a bad envelope timestamp under-reports flood
+  amplification for that batch. This isn't limited to the server-side fallback path either: the
+  companion app stamps `rx_at` at BLE-frame *processing* time, not true RF receive time (`app.js`),
+  so two forwarder copies processed in the same millisecond collapse just as effectively even when
+  the envelope timestamp itself is fine.
+- A 0-hop advert gets `forwarder = NULL` in `client_rx_observations` even though the transmitter is
+  known — it's the advert's own pubkey, which the coverage path records separately with
+  `src='advert'` (see `client_receptions` above). Don't mistake this `NULL` for "unknown".
+
 ## Read API — coverage GeoJSON

 `GET /api/nodes/{pubkey}/rx-coverage?bbox={minLat,minLon,maxLat,maxLon}&z={zoom}`