offload data loss fixes

lib/ble/ble_engine.dart

@@ -155,6 +155,33 @@ bool burstPacketCountMatches({
 }) =>
     expectedPacketCount == actualBurstPacketCount + droppedThisBurst;
 
+/// Honest burst-completeness signal for TELEMETRY ONLY — this NEVER gates the
+/// commit/ACK decision (see the log-only call site).
+///
+/// [receivedTrafficCount] is every frame we actually received this burst, ALL
+/// types (historical R24 data + interleaved console/event/unknown) — i.e.
+/// [BurstStats.totalTrafficPacketCount], NOT the banked historical subset. The
+/// band's [expectedPacketCount] (num_packets) likewise counts every frame it
+/// transmitted, so comparing the two all-types totals is type-agnostic and
+/// interleaving-immune: benign console/event frames riding along cannot fake a
+/// shortfall the way comparing against the R24-only subset did.
+///
+/// [droppedThisBurst] (RecordGate plausibility rejections this burst) is added
+/// back because the band counted those frames but they never entered
+/// [receivedTrafficCount]. A POSITIVE result is frames the band counted that we
+/// did NOT count as valid received traffic — i.e. missing OR corrupted traffic
+/// (would-flag / potential loss): CRC-failed frames also never enter
+/// [receivedTrafficCount], so a positive shortfall cannot by itself prove a
+/// frame never arrived. Zero is complete; negative just means we tallied more
+/// than expected (retried/duplicate frames), which is not loss.
+@visibleForTesting
+int burstPacketShortfall({
+  required int expectedPacketCount,
+  required int receivedTrafficCount,
+  int droppedThisBurst = 0,
+}) =>
+    expectedPacketCount - (receivedTrafficCount + droppedThisBurst);
+
 /// Fired for every LIVE high-rate frame (0x28/0x2B/0x33). These are EPHEMERAL —
 /// they are NOT persisted to raw_records (that bloated storage ~50x and stalled
 /// derivation). The caller routes them to an in-memory sink for the live UI /
@@ -820,6 +847,20 @@ class BleEngine {
   // Lifetime count of GET_CLOCK `clock_epoch` reads rejected by the same gate
   // (ClockPolicy.acceptsClockRead) — see the clock_epoch handler below.
   int _corruptClockReadCount = 0;
+  // True when the last GET_CLOCK showed a plausible strap RTC reading > 1 day in
+  // the FUTURE relative to the phone — the phone clock is likely wrong (slow), so
+  // history offload is DEFERRED (not drained-and-trimmed) until the clocks agree.
+  // See ClockPolicy.phoneClockSuspect and _startHistoricalRefresh.
+  bool _phoneClockSuspect = false;
+  DateTime? _phoneClockSuspectSince;
+  bool get historyPausedForClock => _deferForClock;
+  /// Defer history only while the disagreement is still young. A slow phone
+  /// re-syncs over NTP in minutes; one that persists past the grace window is a
+  /// strap RTC running fast, and deferring forever would stall sync for good.
+  bool get _deferForClock =>
+      _phoneClockSuspect &&
+      !ClockPolicy.suspectGraceExpired(_phoneClockSuspectSince, DateTime.now());
+  int _clockPausedOffloads = 0; // diagnostics: offloads deferred for this reason
   DateTime? _bondTime; // when the handshake completed (bond confirmed)
   DateTime? _armTime; // when live (R10/R11) streams were last armed
   // Run-state for a chain of auto-continued offload rounds: how many
@@ -1311,10 +1352,20 @@ class BleEngine {
       _clockCorrectTries = 0; // fresh retry budget for this connection
       // Drop the previous session's clock correlation so an alarm armed before
       // THIS session's GET_CLOCK reply lands falls back to the raw wall epoch
-      // (drift 0) instead of the stale strap-RTC frame. setClock()→getClock()
-      // below repopulates it for this connection.
+      // (drift 0) instead of the stale strap-RTC frame. The reads below
+      // repopulate it for this connection.
       _clockRef = null;
-      await setClock();
+      // READ BEFORE WRITE. This used to be an unconditional SET_CLOCK, which is
+      // precisely the write [ClockPolicy.phoneClockSuspect] says we must never
+      // make: on a phone running >1 day slow it stamps that slow time onto a
+      // CORRECT strap RTC — and worse, it destroys the evidence, because the
+      // read-back then "agrees" and every later suspect-clock gate sees a
+      // healthy pair. Read first; skip the write while the PHONE is the suspect
+      // one. Unset/behind/garbage-low RTCs are unaffected (not suspect) and are
+      // still corrected here and by the clock_epoch handler's bounded re-issue.
+      await getClock();
+      await Future.delayed(const Duration(milliseconds: 120));
+      if (!_deferForClock) await setClock();
       _lastClockVerifyAt = DateTime.now();
       // Per-connection policy reset. Marginal-radio + post-bond-loop are NOT reset
       // here — they count consecutive bad cycles across reconnects and self-reset on
@@ -1393,9 +1444,25 @@ class BleEngine {
       );
       _setPhase(BleConnState.listening);
       _log('Connected + subscribed — listening (history + live).');
-      _setOffloadActive(true);
-      _lastBackfillAt = _wallSecs();
-      await sendInit(); // triggers the historical offload flood
+      // INIT seq4 IS SEND_HISTORICAL_DATA, so it needs the SAME data-safety gate
+      // as _startHistoricalRefresh — without it every fresh connection drains
+      // and trims under exactly the untrustworthy phone clock we refuse to drain
+      // under there, which is the common case (a dead-battery reboot lands a bad
+      // clock and a reconnect together).
+      final drainOnInit = !_deferForClock;
+      if (!drainOnInit) {
+        _clockPausedOffloads++;
+        _log(
+          '[SYNC] INIT drain DEFERRED — phone clock appears wrong relative to '
+          'the strap RTC; not draining history until they agree '
+          '(deferred_total=$_clockPausedOffloads).',
+        );
+      }
+      _setOffloadActive(drainOnInit);
+      // Only a real drain spends the backfill floor; a deferred one leaves it
+      // open so a foreground trigger can retry as soon as the phone corrects.
+      if (drainOnInit) _lastBackfillAt = _wallSecs();
+      await sendInit(drain: drainOnInit); // seq4 triggers the offload flood
       return true;
     } catch (e) {
       _log('connect setup failed: $e');
@@ -1565,6 +1632,27 @@ class BleEngine {
       // has time to emit the range response before we request another drain.
       await Future.delayed(const Duration(milliseconds: 120));
     }
+    // Data-safety gate: never drain-and-trim history under an untrustworthy phone
+    // clock. Poll the strap RTC and compare; if the phone clock looks slow (strap
+    // plausible but > 1 day ahead), DEFER — draining now would drop the strap's
+    // real records as "future" and the ACK would trim them off the band forever.
+    // The strap retains everything; we drain on a later refresh once the clocks
+    // agree (the phone's clock almost always self-corrects via NTP). SET_CLOCK is
+    // deliberately NOT issued here — pushing the strap back to the slow phone
+    // would corrupt a correct RTC (see ClockPolicy.phoneClockSuspect).
+    await _send(Cmd.getClock, const <int>[]);
+    await Future.delayed(const Duration(milliseconds: 120));
+    if (_session?.connected != true) return;
+    if (_deferForClock) {
+      _clockPausedOffloads++;
+      _log(
+        '[SYNC] refresh($reason) DEFERRED — phone clock appears wrong relative '
+        'to the strap RTC; not draining history until they agree '
+        '(deferred_total=$_clockPausedOffloads).',
+      );
+      _setOffloadActive(false);
+      return;
+    }
     final wait = HistoricalSyncCommandPolicy.waitSeconds(
       _lastHistoricalSendAt,
       _wallSecs(),
@@ -2209,6 +2297,26 @@ class BleEngine {
     if (f.containsKey('clock_epoch')) {
       final dev = f['clock_epoch'] as int;
       final wall = DateTime.now().millisecondsSinceEpoch ~/ 1000;
+      // Assess phone-clock trust from the RAW read, before the alarm-safety gate
+      // below diverts a future reading. A plausible strap RTC that reads > 1 day
+      // ahead of the phone means the phone clock is likely slow — history offload
+      // then DEFERS (see _startHistoricalRefresh) instead of dropping the strap's
+      // real records as "future" and trimming them off the band. Cleared the
+      // moment a read agrees (the phone almost always self-corrects via NTP).
+      final wasSuspect = _phoneClockSuspect;
+      _phoneClockSuspect = ClockPolicy.phoneClockSuspect(dev, wall);
+      if (_phoneClockSuspect && !wasSuspect) {
+        _phoneClockSuspectSince = DateTime.now();
+      } else if (!_phoneClockSuspect) {
+        _phoneClockSuspectSince = null;
+      }
+      if (_phoneClockSuspect != wasSuspect) {
+        _log(_phoneClockSuspect
+            ? '[SYNC] Phone clock appears wrong: strap RTC=$dev is > 1 day ahead '
+                'of phone wall=$wall — DEFERRING history offload until they agree.'
+            : '[SYNC] Phone/strap clocks agree again (strap=$dev wall=$wall) — '
+                'history offload may resume.');
+      }
       // SANITY GATE, mirroring the one `range_newest` gets below. An
       // implausibly far-future `clock_epoch` yields a large NEGATIVE driftSec,
       // and setAlarm arms at `when - driftSec` — years out, where the alarm
@@ -2596,6 +2704,20 @@ class BleEngine {
             expectedPacketCount: expected,
             droppedThisBurst: droppedThisBurst,
           );
+      // Honest, LOG-ONLY completeness signal (never gates the ACK). Compares
+      // num_packets against the ALL-TYPES received total (currentBurstTrafficCount),
+      // not the banked R24 subset — see burstPacketShortfall. Only a POSITIVE
+      // shortfall means frames the band counted that we did not count as valid
+      // received traffic (missing OR CRC-corrupted — potential loss); this is
+      // the signal we want visible in telemetry BEFORE ever wiring a FAIL gate
+      // (which needs its own design + field validation to avoid re-flood).
+      final shortfall = expected == null
+          ? 0
+          : burstPacketShortfall(
+              expectedPacketCount: expected,
+              receivedTrafficCount: d.currentBurstTrafficCount,
+              droppedThisBurst: droppedThisBurst,
+            );
       // ADVISORY ONLY, never a gate: `expectedPacketCount`'s exact semantics
       // (which transport packet types the band itself counts — command
       // responses interleaved with the burst? retried/duplicate frames?) are
@@ -2635,11 +2757,27 @@ class BleEngine {
             'traffic_burst_packets': d.currentBurstTrafficCount,
             'burst_validation_failures': d.consecutiveValidationFailures,
             'burst_breakdown': d.currentBurstBreakdown,
+            'burst_shortfall': shortfall,
           },
         ));
       } else {
         _burstMismatchStreak = 0;
       }
+      // Would-flag: the correct-signal completeness diagnostic. LOG-ONLY — the
+      // commit + verbatim-token ACK below are unchanged. A positive shortfall
+      // is the honest missing/corrupted-traffic telemetry we want to watch
+      // before a later, field-validated FAIL gate ever acts on it.
+      if (shortfall > 0) {
+        _log(
+          '[SYNC] burst completeness would-flag (LOG-ONLY, commit+ACK '
+          'unchanged): expected=$expected '
+          'received=${d.currentBurstTrafficCount} '
+          'dropped_this_burst=$droppedThisBurst shortfall=$shortfall '
+          '(all-types received total — frames the band counted that we did '
+          'not; missing or CRC-corrupted, potential loss; groundwork for a '
+          'future FAIL gate, NOT gating today)',
+        );
+      }
       final r = d.bufferedRecTsRange;
       final droppedThisBurstForLog = droppedThisBurst;
       final hadDurableRows =
@@ -2939,10 +3077,15 @@ class BleEngine {
       inner.map((b) => b.toRadixString(16).padLeft(2, '0')).join();
 
   // ── high-level flows ─────────────────────────────────────────────────────────────
-  Future<void> sendInit() async {
-    _log('Sending 5-packet INIT…');
+  /// [drain] false sends the first FOUR packets only: seq4 is
+  /// SEND_HISTORICAL_DATA (the flash drain), and it is skipped when the phone
+  /// clock is suspect — see _doConnect and [ClockPolicy.phoneClockSuspect].
+  Future<void> sendInit({bool drain = true}) async {
+    final pkts =
+        drain ? initPackets : initPackets.take(initPackets.length - 1).toList();
+    _log('Sending ${pkts.length}-packet INIT…');
     try {
-      for (final pkt in initPackets) {
+      for (final pkt in pkts) {
         await _write(pkt);
         await Future.delayed(const Duration(milliseconds: 120));
       }

lib/compute/derivation_engine.dart

@@ -1836,13 +1836,16 @@ class DerivationEngine {
             rangePages: rangePages,
             rangeRows: rangeRows,
           );
-          final firstCounter = (decodedRows.first['counter'] as num?)?.toInt();
-          final lastCounter = (decodedRows.last['counter'] as num?)?.toInt();
-          final rrRows = firstCounter == null || lastCounter == null
+          // The page is ordered rec_ts ASC, so first = min second, last = max.
+          // decoded_rr shares the rec_ts key, so this pulls exactly the page's
+          // beats — no counter span (which broke across the strap's reboot reset).
+          final firstRecTs = (decodedRows.first['rec_ts'] as num?)?.toInt();
+          final lastRecTs = (decodedRows.last['rec_ts'] as num?)?.toInt();
+          final rrRows = firstRecTs == null || lastRecTs == null
               ? const <Map<String, dynamic>>[]
-              : await LocalDb.decodedRrByCounterRange(
-                  fromCounter: firstCounter,
-                  toCounter: lastCounter,
+              : await LocalDb.decodedRrByRecTsRange(
+                  fromRecTs: firstRecTs,
+                  toRecTs: lastRecTs,
                 );
           worker.send({'type': 'page', 'frames': decodedRows, 'rr': rrRows});
           final last = decodedRows.last;

lib/compute/derive_prepare.dart

@@ -431,11 +431,14 @@ class _PrepareAccumulator {
     List<Map<String, dynamic>> rrRows,
   ) {
     if (frames.isEmpty) return;
-    final rrByCounter = <int, List<Map<String, dynamic>>>{};
+    // Associate beats to frames by rec_ts (their shared key). The strap's counter
+    // resets on reboot, so grouping by counter mis-joined two seconds that reused
+    // one counter within a page.
+    final rrByRecTs = <int, List<Map<String, dynamic>>>{};
     for (final row in rrRows) {
-      final counter = _num(row['counter'])?.toInt();
-      if (counter == null) continue;
-      rrByCounter.putIfAbsent(counter, () => <Map<String, dynamic>>[]).add(row);
+      final recTs = _num(row['rec_ts'])?.toInt();
+      if (recTs == null) continue;
+      rrByRecTs.putIfAbsent(recTs, () => <Map<String, dynamic>>[]).add(row);
     }
     for (final row in frames) {
       final recTs = _num(row['rec_ts'])?.toInt();
@@ -454,9 +457,7 @@ class _PrepareAccumulator {
       // tsSec is what lets `Substrate.fromJson` tell "absent" (empty ⇒
       // zero-filled) from "present but zero".
       skinContact.add(_num(row['skin_contact'])?.toInt() ?? 0);
-      final counter = _num(row['counter'])?.toInt();
-      if (counter == null) continue;
-      final beats = rrByCounter[counter];
+      final beats = rrByRecTs[recTs];
       if (beats == null) continue;
       for (final beat in beats) {
         final rr = _num(beat['rr_ms'])?.toDouble();