web: node table keeps latest telemetry of every type (per-field merge) (#847)

ACCEPTANCE.md

@@ -2857,3 +2857,57 @@ differ). Still green unchanged: **HT-A4 / A5 / A6 / A7** (fallback ladder, one
 shared factory on both maps, no attribution, apex/contract untouched), **A1** (no
 broker — the basemap hosts are raster CDNs), **B1** (all suites), and **B4** (exact
 Apache header on the new `basemap-blend.test.js`).
+
+---
+
+## Bugfix: Node-table telemetry hidden by newer packets of another type
+
+Meshtastic telemetry is a protobuf `oneof` — each packet carries exactly one
+metric family (device / environment / power / air-quality;
+`data/mesh_ingestor/handlers/telemetry.py`). The node table's environment
+columns exist only through the client-side per-node telemetry merge
+(`aggregateTelemetrySnapshots` → `mergeTelemetryIntoNodes`), which merged a
+fixed `SNAPSHOT_WINDOW = 7` packet window: seven newer device/power packets
+evicted the last environment packet wholesale, hiding temperature / humidity /
+pressure (and, on the node detail page, IAQ etc.) although the rows were still
+in the accumulator and the DB. Selection and precedence were also array-order
+driven (first-7-encountered, position-0 wins), which is wrong for warm
+IndexedDB cache seeds (key order) and incremental `mergeById` appends — stale
+values could beat fresh ones. Fix: `aggregateTelemetrySnapshots` now performs a
+**per-field latest-non-null merge** — each field takes the value from the
+node's newest packet (by `rx_time`, falling back to `telemetry_time`) that
+carries it non-null, order-independently, bounded by the caller's existing
+7-day accumulator window instead of a packet count. A null/absent field never
+clears an older valid value. Frontend read-side only — no API/DB/ingestor
+change; apex (I) and privacy (II) untouched; protocol-neutral (IV). The raw
+accumulators stay raw (CL-A1/bugfix A1 unchanged).
+
+### TM-A1 — per-field latest-non-null telemetry merge
+```bash
+( cd web && node --test public/assets/js/app/__tests__/snapshot-aggregator.test.js )
+```
+**Expected:** pass. With one environment packet followed by more than
+`SNAPSHOT_WINDOW` newer device/power packets for the same node, the aggregate
+retains the environment metrics (temperature / humidity / pressure) alongside
+the newest device metrics; the newest non-null value per field wins regardless
+of input array order (inputs that differ only in order produce identical
+aggregates whenever timestamps differ; an equal-timestamp conflict resolves
+deterministically to the row later in the input); a null/absent field never
+overwrites an older valid value; the hidden `snapshots` history is
+chronological and `latestSnapshot` is the newest packet by timestamp, not by
+array position.
+
+### TM-R1 — Regression: prior acceptance still holds
+```bash
+( cd web && npm test ) && ( cd web && bundle exec rspec )
+```
+**Expected:** every prior check still passes. At risk and explicitly required
+to remain green: **CL-A1** (the Log's raw-accumulator retention —
+`main-log-snapshot-retention.test.js` — the fix changes only the aggregated
+locals, never the accumulators), the node detail page and chart suites
+(`node-details.test.js`, node-page chart tests — the aggregate keeps its
+`snapshots` / `latestSnapshot` shape), and `data-merge.test.js`
+(`mergeTelemetryIntoNodes` consumes one aggregate per node unchanged). Node /
+position / neighbor aggregation keep their existing `SNAPSHOT_WINDOW`
+semantics — only telemetry aggregation changes. No Ruby/Python surface is
+touched (**C2** and the Python suite unaffected).

web/public/assets/js/app/__tests__/snapshot-aggregator.test.js

@@ -80,7 +80,7 @@ test('aggregateNodeSnapshots reconciles identifiers and fills missing values', (
   assert.equal(node.snapshots.length, 3);
 });
 
-test('aggregateTelemetrySnapshots and aggregatePositionSnapshots mirror node aggregation', () => {
+test('aggregateTelemetrySnapshots and aggregatePositionSnapshots retain older non-null fields', () => {
   const telemetryEntries = [
     { node_id: SAMPLE_NODE_ID, node_num: 5, temperature: null, rx_time: 20 },
     { node_num: 5, temperature: 21.5, humidity: 52, rx_time: 10 },
@@ -89,7 +89,7 @@ test('aggregateTelemetrySnapshots and aggregatePositionSnapshots mirror node agg
     { node_id: SAMPLE_NODE_ID, node_num: 5, longitude: 13.4, rx_time: 25 },
     { node_num: 5, latitude: 52.5, rx_time: 15 },
   ];
-  const telemetryAggregated = aggregateTelemetrySnapshots(telemetryEntries, { limit: 3 });
+  const telemetryAggregated = aggregateTelemetrySnapshots(telemetryEntries);
   const positionAggregated = aggregatePositionSnapshots(positionEntries, { limit: 3 });
   assert.equal(telemetryAggregated.length, 1);
   assert.equal(positionAggregated.length, 1);
@@ -119,3 +119,103 @@ test('aggregateSnapshots returns an empty array when no entries are provided', (
   assert.deepEqual(aggregateSnapshots(null, { keySelector: () => 'noop' }), []);
   assert.deepEqual(aggregateNodeSnapshots([], {}), []);
 });
+
+// Regression: hidden telemetry in the node table. Meshtastic telemetry is a
+// protobuf oneof — each packet carries exactly one metric family — so the
+// aggregate must keep the newest non-null value per field across families
+// instead of merging a fixed packet-count window (TM-A1 in ACCEPTANCE.md).
+
+/**
+ * Build a newest-first packet stream: one environment packet followed by
+ * ``newerCount`` newer device/power packets for the same node.
+ *
+ * @param {number} newerCount Number of non-environment packets newer than the
+ *   environment packet.
+ * @returns {Array<Object>} Telemetry entries ordered newest-first.
+ */
+function environmentThenNewerPackets(newerCount) {
+  const entries = [];
+  for (let i = newerCount; i >= 1; i -= 1) {
+    entries.push({
+      id: 9000 + i,
+      node_id: SAMPLE_NODE_ID,
+      rx_time: 1000 + i * 60,
+      telemetry_type: i % 2 ? 'device' : 'power',
+      // Power packets carry no extracted metric columns at all; device packets
+      // carry the device family only.
+      ...(i % 2 ? { battery_level: 80, voltage: 4.05 } : {}),
+    });
+  }
+  entries.push({
+    id: 9000,
+    node_id: SAMPLE_NODE_ID,
+    rx_time: 1000,
+    telemetry_type: 'environment',
+    temperature: 21.5,
+    relative_humidity: 40.2,
+    barometric_pressure: 1013.2,
+  });
+  return entries;
+}
+
+test('aggregateTelemetrySnapshots keeps environment metrics past SNAPSHOT_WINDOW newer packets of other types', () => {
+  const aggregated = aggregateTelemetrySnapshots(environmentThenNewerPackets(SNAPSHOT_WINDOW + 1));
+  assert.equal(aggregated.length, 1);
+  const record = aggregated[0];
+  // Latest known environment readings survive alongside the newest device values.
+  assert.equal(record.temperature, 21.5);
+  assert.equal(record.relative_humidity, 40.2);
+  assert.equal(record.barometric_pressure, 1013.2);
+  assert.equal(record.battery_level, 80);
+  assert.equal(record.voltage, 4.05);
+});
+
+test('aggregateTelemetrySnapshots prefers the newest non-null value per field regardless of input order', () => {
+  // Append/IndexedDB-seed order: the older packet sits first in the array, the
+  // way a warm cache seed (key order) or an incremental mergeById append holds it.
+  const older = { id: 3, node_id: SAMPLE_NODE_ID, rx_time: 1000, telemetry_type: 'device', battery_level: 20, voltage: 3.2 };
+  const newer = { id: 9, node_id: SAMPLE_NODE_ID, rx_time: 2000, telemetry_type: 'device', battery_level: 90, voltage: 4.1 };
+  for (const entries of [[older, newer], [newer, older]]) {
+    const aggregated = aggregateTelemetrySnapshots(entries);
+    assert.equal(aggregated.length, 1);
+    const record = aggregated[0];
+    assert.equal(record.battery_level, 90);
+    assert.equal(record.voltage, 4.1);
+    // The hidden history stays chronological and the latest snapshot is the
+    // newest packet by timestamp, not by array position.
+    assert.deepEqual(record.snapshots.map(s => s.rx_time), [1000, 2000]);
+    assert.equal(record.latestSnapshot.rx_time, 2000);
+  }
+});
+
+test('aggregateTelemetrySnapshots resolves equal timestamps to the row later in the input', () => {
+  // Same-second packets are equally fresh; the deterministic tie rule is that
+  // the row appearing later in the input (a fresher append) wins the conflict.
+  const tied = aggregateTelemetrySnapshots([
+    { id: 1, node_id: SAMPLE_NODE_ID, rx_time: 1000, battery_level: 20 },
+    { id: 2, node_id: SAMPLE_NODE_ID, rx_time: 1000, battery_level: 90 },
+  ]);
+  assert.equal(tied.length, 1);
+  assert.equal(tied[0].battery_level, 90);
+  assert.equal(tied[0].latestSnapshot.id, 2);
+  // Two untimestamped rows tie at -Infinity: the explicit comparator keeps the
+  // ordering deterministic (no NaN from subtracting infinities) and the later
+  // row still wins.
+  const untimed = aggregateTelemetrySnapshots([
+    { id: 3, node_id: SAMPLE_NODE_ID, voltage: 3.2 },
+    { id: 4, node_id: SAMPLE_NODE_ID, voltage: 4.1 },
+  ]);
+  assert.equal(untimed.length, 1);
+  assert.equal(untimed[0].voltage, 4.1);
+  assert.deepEqual(untimed[0].snapshots.map(s => s.id), [3, 4]);
+});
+
+test('aggregateTelemetrySnapshots never overwrites a valid value with an empty field', () => {
+  const aggregated = aggregateTelemetrySnapshots([
+    { id: 2, node_id: SAMPLE_NODE_ID, rx_time: 2000, telemetry_type: 'environment', temperature: 22.1, relative_humidity: null },
+    { id: 1, node_id: SAMPLE_NODE_ID, rx_time: 1000, telemetry_type: 'environment', temperature: 21.5, relative_humidity: 40.2 },
+  ]);
+  assert.equal(aggregated.length, 1);
+  assert.equal(aggregated[0].temperature, 22.1);
+  assert.equal(aggregated[0].relative_humidity, 40.2);
+});

web/public/assets/js/app/main.js

@@ -3526,7 +3526,9 @@ export function initializeApp(config) {
     // the non-enumerable ``snapshots`` history) and merges oldest-last (pinning to
     // the stalest reading), collapsing each node's history to {stale-first, newest}
     // so a telemetry/position Log entry flashes in and vanishes on the next tick.
-    // Keeping the accumulators raw gives every packet a stable, id-keyed Log entry
+    // (Telemetry now aggregates per-field by timestamp — TM-A1 — but re-storing
+    // its aggregate would still collapse the Log's per-packet history.) Keeping
+    // the accumulators raw gives every packet a stable, id-keyed Log entry
     // (bugfix A1).
   }
 

web/public/assets/js/app/snapshot-aggregator.js

@@ -210,15 +210,89 @@ export function aggregateNodeSnapshots(entries, { limit = SNAPSHOT_WINDOW } = {}
 }
 
 /**
- * Aggregate telemetry packets for each node.
+ * Resolve the timestamp used to order a telemetry packet.
+ *
+ * ``rx_time`` is authoritative (stamped on every stored packet); the
+ * self-reported ``telemetry_time`` is the fallback. Entries carrying neither
+ * are treated as infinitely old so they only ever fill fields that no
+ * timestamped packet provides.
+ *
+ * @param {Object} entry Telemetry payload.
+ * @returns {number} Unix timestamp in seconds, or ``-Infinity`` when absent.
+ */
+function resolveTelemetryTimestamp(entry) {
+  const rxTime = normaliseNum(entry.rx_time ?? entry.rxTime);
+  if (rxTime != null) return rxTime;
+  const telemetryTime = normaliseNum(entry.telemetry_time ?? entry.telemetryTime);
+  if (telemetryTime != null) return telemetryTime;
+  return Number.NEGATIVE_INFINITY;
+}
+
+/**
+ * Aggregate telemetry packets for each node with per-field latest-non-null
+ * semantics.
+ *
+ * Meshtastic telemetry is a protobuf ``oneof``: each packet carries exactly
+ * one metric family (device / environment / power / air-quality), so a
+ * packet-count window can evict one family wholesale when another family
+ * transmits more often. Instead of merging a bounded window, every field of
+ * the aggregate takes the value from the node's newest packet (per
+ * {@link resolveTelemetryTimestamp}) that carries it non-null. A null/absent
+ * field never overwrites an older valid value, and the result is independent
+ * of input order (warm cache seeds arrive in IndexedDB key order and
+ * incremental refreshes append, so the caller's array is not reliably
+ * newest-first). The retained set is bounded by the caller's accumulator
+ * window (7 days), not a packet count.
+ *
+ * Each aggregate exposes the same non-enumerable metadata as
+ * {@link aggregateSnapshots}: ``snapshots`` (the node's packets in
+ * chronological order, ties broken by input order; read-only references — do
+ * not mutate) and ``latestSnapshot`` (the newest packet by timestamp).
  *
  * @param {Array<Object>} entries Telemetry payloads.
- * @param {{ limit?: number }} [options] Aggregation options.
- * @returns {Array<Object>} Aggregated telemetry data.
+ * @returns {Array<Object>} Aggregated telemetry data, one entry per node.
  */
-export function aggregateTelemetrySnapshots(entries, { limit = SNAPSHOT_WINDOW } = {}) {
+export function aggregateTelemetrySnapshots(entries) {
+  if (!Array.isArray(entries) || entries.length === 0) {
+    return [];
+  }
   const resolveKey = createNodeKeyResolver();
-  return aggregateSnapshots(entries, { keySelector: resolveKey, limit });
+  const groups = new Map();
+  for (const entry of entries) {
+    if (!isObject(entry)) continue;
+    const key = resolveKey(entry);
+    if (!key) continue;
+    const group = groups.get(key);
+    if (group) {
+      group.push(entry);
+    } else {
+      groups.set(key, [entry]);
+    }
+  }
+  const aggregates = [];
+  for (const group of groups.values()) {
+    // Sort ascending by timestamp, preserving input order on ties so a row
+    // appended later (a fresher delta with the same rx second) wins.
+    const orderedSnapshots = group
+      .map((snapshot, index) => ({ snapshot, timestamp: resolveTelemetryTimestamp(snapshot), index }))
+      .sort((a, b) => {
+        // Explicit compare: subtraction yields NaN for two -Infinity stamps.
+        if (a.timestamp < b.timestamp) return -1;
+        if (a.timestamp > b.timestamp) return 1;
+        return a.index - b.index;
+      })
+      .map(item => item.snapshot);
+    // Folding the null-skipping merge oldest→newest leaves each field holding
+    // the newest non-null value across all of the node's packets.
+    const target = {};
+    for (const snapshot of orderedSnapshots) {
+      mergeSnapshotFields(target, snapshot);
+    }
+    defineHiddenProperty(target, 'snapshots', orderedSnapshots);
+    defineHiddenProperty(target, 'latestSnapshot', orderedSnapshots[orderedSnapshots.length - 1] ?? null);
+    aggregates.push(target);
+  }
+  return aggregates;
 }
 
 /**