diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart
index a7daacfee..58afc6e30 100644
--- a/lib/ble/ble_engine.dart
+++ b/lib/ble/ble_engine.dart
@@ -7502,7 +7502,10 @@ class BleEngine {
       // band never trimming and re-flooding the same chunk forever. On
       // persistent failure bounce the link — the committed data is safe, and
       // the next session's re-delivery is dedup-safe (rows REPLACE by rec_ts).
-      if (!await _writeAckVerified(ack, session, taskGen: taskGen)) {
+      final ackWatch = clock.stopwatch()..start();
+      final ackWritten = await _writeAckVerified(ack, session, taskGen: taskGen);
+      final ackMs = ackWatch.elapsedMilliseconds;
+      if (!ackWritten) {
         // False for one of three reasons: the writes genuinely exhausted, the
         // session died, or the TASK ended under the retries. Only the first is
         // this continuation's to handle — a stale one must not touch the
@@ -7592,43 +7595,54 @@ class BleEngine {
       // over — records were banked and the band may advance.
       _noDurableProgress.trimAcked();
       d.noteBatchAcked(); // ACKed and KEEP listening
-      await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry(
-        status: 'acknowledged',
-        ackedAt: DateTime.now().millisecondsSinceEpoch,
-        metaPatch: {
-          'last_batch_token': tokenHex,
-          'last_batch_id': m.batchId,
-          'last_batch_records': d.records,
-          'last_ack_batches': d.batches,
-          'strap_history_oldest_ts': _strapHistoryOldestTs,
-          'strap_history_newest_ts': _strapHistoryNewestTs,
-          // Which WHOOP generation this batch came from — records/sessions
-          // vary hugely in richness by generation (and, for gen5, by whether
-          // the R22 deep-buffer opt-in was sent), so downstream diagnostics
-          // need this without reaching into the transport layer.
-          'band_generation': state.generation,
-        },
-      ));
-      // Same event, but a REAL per-chunk row keyed by the token — closes out
-      // whatever ack_failed history this token accumulated above.
-      await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry(
-        chunkId: 'batch:$tokenHex',
-        kind: 'historical_batch',
-        status: 'acked',
-        ackedAt: DateTime.now().millisecondsSinceEpoch,
-        metaPatch: {
-          'batch_id': m.batchId,
-          // Cumulative for the connection, not this chunk (kept as it was);
-          // the chunk's own counts and record-time range follow.
-          'records': d.records,
-          if (d.lastChunkStats case final c?) ...{
-            'chunk_rows': c.raws,
-            'chunk_archived': c.archives,
-            'chunk_ts_min': ?c.tsMin,
-            'chunk_ts_max': ?c.tsMax,
+      // Both post-ACK diagnostic rows share one transaction and one commit.
+      final postWatch = clock.stopwatch()..start();
+      await _bestEffortLedgerWrite(
+          () => LocalDb.syncLedgerTransaction((txn) async {
+        await LocalDb.upsertSyncLedgerEntry(
+          txn: txn,
+          status: 'acknowledged',
+          ackedAt: DateTime.now().millisecondsSinceEpoch,
+          metaPatch: {
+            'last_batch_token': tokenHex,
+            'last_batch_id': m.batchId,
+            'last_batch_records': d.records,
+            'last_ack_batches': d.batches,
+            'strap_history_oldest_ts': _strapHistoryOldestTs,
+            'strap_history_newest_ts': _strapHistoryNewestTs,
+            // Which WHOOP generation this batch came from — records/sessions
+            // vary hugely in richness by generation (and, for gen5, by whether
+            // the R22 deep-buffer opt-in was sent), so downstream diagnostics
+            // need this without reaching into the transport layer.
+            'band_generation': state.generation,
           },
-        },
-      ));
+        );
+        // Same event, but a REAL per-chunk row keyed by the token — closes out
+        // whatever ack_failed history this token accumulated above.
+        await LocalDb.upsertSyncLedgerEntry(
+          txn: txn,
+          chunkId: 'batch:$tokenHex',
+          kind: 'historical_batch',
+          status: 'acked',
+          ackedAt: DateTime.now().millisecondsSinceEpoch,
+          metaPatch: {
+            'batch_id': m.batchId,
+            // Cumulative for the connection, not this chunk (kept as it was);
+            // the chunk's own counts and record-time range follow.
+            'records': d.records,
+            if (d.lastChunkStats case final c?) ...{
+              'chunk_rows': c.raws,
+              'chunk_archived': c.archives,
+              'chunk_ts_min': ?c.tsMin,
+              'chunk_ts_max': ?c.tsMax,
+            },
+          },
+        );
+      }));
+      final stats = d.lastChunkStats;
+      _log('[perf] sync chunk records=${stats?.raws ?? 0} '
+          'beats=${stats?.beats ?? 0} commit_ms=${stats?.commitMs ?? 0} '
+          'ack_ms=$ackMs post_ms=${postWatch.elapsedMilliseconds}');
       _noteStored(); // a banked batch → schedule a (debounced) derive
     } else if (m.sub == SyncMeta.historyComplete) {
       final d = _drain;
@@ -9431,7 +9445,8 @@ class DrainController {
 
   /// What the last token commit banked: rows, archived frames and the range of
   /// record times. Written to that chunk's ledger row when it is ACKed.
-  ({int raws, int archives, int? tsMin, int? tsMax})? lastChunkStats;
+  ({int raws, int archives, int? tsMin, int? tsMax, int beats, int commitMs})?
+      lastChunkStats;
   int recordsThisOffload = 0; // since the last HISTORY_COMPLETE / rearm
   int batches = 0; // total this connection
   // Since the last [startFreshTask] — what [awaitComplete]'s report carries.
@@ -9935,6 +9950,7 @@ class DrainController {
     _samples.clear();
     _archives.clear();
     _ecgRaw.clear();
+    var commitMs = 0;
     try {
       // Defense in depth (constructor already rejects onRecordsBatch-only):
       // never report durable success for buffered content without onCommit.
@@ -9950,15 +9966,19 @@ class DrainController {
             'archives=${archives.length}, token=${tokenHex != null})',
           );
         }
+        final commitWatch = clock.stopwatch()..start();
         await commit(raws, samples, tokenHex,
             archives: archives, ecgRawPackets: ecgRaw);
+        commitMs = commitWatch.elapsedMilliseconds;
       }
       if (tokenHex != null) {
         // What this chunk actually banked, and the record times it carried
         // (#450: a chunk whose times sit far behind the stored edge is a band
         // clock that moved, which the per-chunk ledger row then shows).
         int? lo, hi;
+        var beats = 0;
         for (final x in samples) {
+          beats += x?.rrIntervalsMs.length ?? 0;
           final t = x?.tsEpoch;
           if (t == null || t <= 0) continue;
           if (lo == null || t < lo) lo = t;
@@ -9969,6 +9989,8 @@ class DrainController {
           archives: archives.length,
           tsMin: lo,
           tsMax: hi,
+          beats: beats,
+          commitMs: commitMs,
         );
         _burstFlushedDurable = 0;
         _burstFlushedGateDrops = 0;
diff --git a/lib/data/db.dart b/lib/data/db.dart
index 51cf3d383..623e02588 100644
--- a/lib/data/db.dart
+++ b/lib/data/db.dart
@@ -4877,6 +4877,8 @@ class LocalDb {
           'cursor_advanced counter_hw=$maxCounter rec_ts_hw=$maxRecTs '
           'trim=${trimToken != null}',
         );
+        // Freshness must roll back with the rows and cursor on failure.
+        await _writeCaptureFreshness(txn, raws);
       });
     } finally {
       // ALWAYS restore NORMAL — even if the commit threw — so a leaked FULL does
@@ -4887,7 +4889,6 @@ class LocalDb {
         /* non-fatal — see open-time PRAGMA discipline */
       }
     }
-    await _writeCaptureFreshness(raws);
   }
 
   // ── DERIVED STORE (permanent, rich) ────────────────────────────────────────
@@ -6897,7 +6898,10 @@ class LocalDb {
     return fallbackSec;
   }
 
-  static Future<void> _writeCaptureFreshness(List<RawRecord> raws) async {
+  static Future<void> _writeCaptureFreshness(
+    DatabaseExecutor txn,
+    List<RawRecord> raws,
+  ) async {
     if (raws.isEmpty) return;
     var latest = 0;
     for (final raw in raws) {
@@ -6905,9 +6909,14 @@ class LocalDb {
       if (recTs > latest) latest = recTs;
     }
     if (latest <= 0) return;
-    final prev = await computeFreshness('capture');
+    final prevRows = await txn.query(
+      'compute_freshness',
+      where: 'key = ?',
+      whereArgs: ['capture'],
+      limit: 1,
+    );
     Map<String, dynamic> payload = const {};
-    final rawJson = prev?['payload_json'];
+    final rawJson = prevRows.isEmpty ? null : prevRows.first['payload_json'];
     if (rawJson is String && rawJson.isNotEmpty) {
       try {
         final d = jsonDecode(rawJson);
@@ -6921,7 +6930,7 @@ class LocalDb {
       'latest_raw_rec_ts': latest,
       'latest_raw_day': _localDayLabelFromEpoch(latest),
     };
-    await putComputeFreshness('capture', jsonEncode(payload));
+    await putComputeFreshness('capture', jsonEncode(payload), txn: txn);
   }
 
   // Canonical LOCAL day label — shared with the UI/coach via day_label.dart so
@@ -8966,8 +8975,9 @@ class LocalDb {
 
   static Future<Map<String, dynamic>?> syncLedgerEntry([
     String chunkId = 'capture',
+    DatabaseExecutor? txn,
   ]) async {
-    final db = await instance;
+    final db = txn ?? await instance;
     final rows = await db.query(
       'sync_ledger',
       where: 'chunk_id = ?',
@@ -8986,9 +8996,10 @@ class LocalDb {
     int? ackedAt,
     String? lastError,
     Map<String, dynamic>? metaPatch,
+    DatabaseExecutor? txn,
   }) async {
-    final db = await instance;
-    final existing = await syncLedgerEntry(chunkId);
+    final db = txn ?? await instance;
+    final existing = await syncLedgerEntry(chunkId, txn);
     final now = DateTime.now().millisecondsSinceEpoch;
     final meta = <String, dynamic>{};
     if (existing != null) {
@@ -9017,6 +9028,14 @@ class LocalDb {
     }, conflictAlgorithm: ConflictAlgorithm.replace);
   }
 
+  /// Commit related diagnostic ledger rows together after the trim ACK.
+  static Future<void> syncLedgerTransaction(
+    Future<void> Function(Transaction txn) body,
+  ) async {
+    final db = await instance;
+    await db.transaction(body);
+  }
+
   static Future<List<Map<String, dynamic>>> syncLedger() async {
     final db = await instance;
     return db.query('sync_ledger', orderBy: 'created_at ASC');
@@ -11556,14 +11575,44 @@ class LocalDb {
   @visibleForTesting
   static List<String> get salvageTablesForTest => _salvageTables;
 
+  /// Forces the existing per-row merge for the bulk-restore equivalence tests.
+  @visibleForTesting
+  static bool debugRowByRowMerge = false;
+
+  @visibleForTesting
+  static Future<Map<String, int>> debugMergeRowByRow(String path) async {
+    final previous = debugRowByRowMerge;
+    debugRowByRowMerge = true;
+    try {
+      return await _mergeFromDbFile(path);
+    } finally {
+      debugRowByRowMerge = previous;
+    }
+  }
+
   static Future<Map<String, int>> _mergeFromDbFile(
     String path, {
     List<String>? only,
     bool tolerant = false,
   }) async {
-    final src = await openDatabase(path, readOnly: true);
     final db = await instance;
+    final src = await openDatabase(path, readOnly: true);
     const tables = _restoreTables;
+    const bulkTables = {
+      'decoded_onehz',
+      'decoded_rr',
+      'samples',
+      // Not raw_archive: it has its own merge (_mergeRawArchive), which keeps
+      // the row already on this device instead of replacing it.
+      'events',
+      'band_events',
+      'band_battery',
+      'day_result',
+      'metric_series',
+      'metric_series_version',
+      'metric_method',
+    };
+
     // Columns this app's schema actually has, per table — so a row from a NEWER
     // export carrying extra columns this build doesn't know about is filtered
     // down (dropped) instead of throwing "no such column". A column the source
@@ -11593,18 +11642,321 @@ class LocalDb {
     // version stamp and this device's baselines are the same day's history and
     // get the same protection, or trends and the rolling baselines would be
     // rebuilt from the other export's numbers while day detail serves ours.
-    final finalizedDays = {
-      for (final r in await db.query(
-        'day_result',
-        columns: ['day_id'],
-        where: 'finalized = 1',
-        distinct: true,
-      ))
-        '${r['day_id']}',
-    };
+    final finalizedDays = <String>{};
     var archiveSkipped = 0;
     int? archiveEdgeSec;
+    var impAttached = false;
+    var failed = true;
+
+    // Attempt every cleanup operation. Preserve the original restore error,
+    // but surface a cleanup failure when the copy otherwise succeeded.
+    Future<void> cleanupRestore({
+      List<String> tempTables = const [],
+      bool closeSource = false,
+      required bool preserveError,
+    }) async {
+      Object? firstError;
+      StackTrace? firstStack;
+      Future<void> attempt(Future<void> Function() action) async {
+        try {
+          await action();
+        } catch (error, stack) {
+          firstError ??= error;
+          firstStack ??= stack;
+        }
+      }
+
+      for (final name in tempTables) {
+        await attempt(() => db.execute('DROP TABLE IF EXISTS temp.$name'));
+      }
+      if (closeSource) {
+        if (impAttached) {
+          await attempt(() => db.execute('DETACH DATABASE imp'));
+        }
+        await attempt(src.close);
+      }
+      final error = firstError;
+      if (!preserveError && error != null) {
+        Error.throwWithStackTrace(error, firstStack!);
+      }
+    }
+
+    String quoteSql(String name) => '"${name.replaceAll('"', '""')}"';
+    String sourceColumn(String name) => 's.${quoteSql(name)}';
+    String numericSql(String value) =>
+        "typeof($value) IN ('integer', 'real')";
+
+    Future<bool> mergeBulkTable(String t) async {
+      if (!impAttached || !bulkTables.contains(t)) return false;
+      final exists = await db.rawQuery(
+        "SELECT 1 FROM imp.sqlite_master WHERE type = 'table' AND name = ?",
+        [t],
+      );
+      if (exists.isEmpty) return true;
+
+      if (t == 'day_result') importedDays = <String>{};
+      final destInfo = await db.rawQuery('PRAGMA main.table_info(${quoteSql(t)})');
+      if (destInfo.isEmpty) return false;
+      final sourceInfo = await db.rawQuery('PRAGMA imp.table_info(${quoteSql(t)})');
+      final destColumns = [
+        for (final column in destInfo) column['name'] as String,
+      ];
+      final sourceColumns = {
+        for (final column in sourceInfo) column['name'] as String,
+      };
+      final maxRowid = Sqflite.firstIntValue(
+        await db.rawQuery('SELECT MAX(rowid) FROM imp.${quoteSql(t)}'),
+      );
+      if (maxRowid == null || maxRowid <= 0) {
+        counts[t] = 0;
+        return true;
+      }
+
+      if (!sourceColumns.any(destColumns.contains)) return false;
+
+      const deviceKeyed = {'samples', 'decoded_onehz', 'decoded_rr'};
+      final isDeviceKeyed = deviceKeyed.contains(t);
+      final sourceTsMs = sourceColumns.contains('ts_ms')
+          ? sourceColumn('ts_ms')
+          : 'NULL';
+      final sourceDeviceId = sourceColumns.contains('device_id')
+          ? sourceColumn('device_id')
+          : 'NULL';
+      final sourceRecTs = sourceColumns.contains('rec_ts')
+          ? sourceColumn('rec_ts')
+          : 'NULL';
+      final sourceRrTsMs = sourceColumns.contains('rr_ts_ms')
+          ? sourceColumn('rr_ts_ms')
+          : 'NULL';
+      final recoveredRecTs = t == 'decoded_rr'
+          ? 'CASE WHEN $sourceRecTs IS NULL THEN '
+              'CASE WHEN ${numericSql(sourceRrTsMs)} '
+              'THEN CAST($sourceRrTsMs AS INTEGER) / 1000 ELSE NULL END '
+              'ELSE $sourceRecTs END'
+          : sourceRecTs;
+      final keyTime = t == 'samples'
+          ? (sourceColumns.contains('ts') ? sourceColumn('ts') : 'NULL')
+          : t == 'decoded_rr'
+              ? recoveredRecTs
+              : sourceRecTs;
+      final derivedTsMs =
+          'CASE WHEN $sourceTsMs IS NULL THEN CAST($keyTime AS INTEGER) * 1000 '
+          'ELSE $sourceTsMs END';
+      final derivedDeviceId =
+          'CASE WHEN $sourceTsMs IS NULL THEN COALESCE($sourceDeviceId, '
+          "'$kPrimaryDeviceId') ELSE $sourceDeviceId END";
+
+      final expressions = <String, String>{};
+      for (final column in destColumns) {
+        if (sourceColumns.contains(column)) {
+          expressions[column] = sourceColumn(column);
+        }
+      }
+      if (t == 'decoded_onehz') {
+        if (destColumns.contains('on_wrist')) expressions['on_wrist'] = 'NULL';
+        if (destColumns.contains('hr_valid')) expressions['hr_valid'] = 'NULL';
+        if (destColumns.contains('skin_temp_c') &&
+            sourceColumns.contains('skin_temp_c')) {
+          final skinTemp = sourceColumn('skin_temp_c');
+          expressions['skin_temp_c'] =
+              'CASE WHEN ${numericSql(skinTemp)} AND $skinTemp <= -49.995 '
+              'THEN NULL ELSE $skinTemp END';
+        }
+      }
+      if (t == 'decoded_rr' && destColumns.contains('rec_ts')) {
+        expressions['rec_ts'] = recoveredRecTs;
+      }
+      if (isDeviceKeyed && destColumns.contains('device_id')) {
+        expressions['device_id'] = derivedDeviceId;
+      }
+      if (isDeviceKeyed && destColumns.contains('ts_ms')) {
+        expressions['ts_ms'] = derivedTsMs;
+      }
+      final columns = [
+        for (final column in destColumns)
+          if (expressions.containsKey(column)) column,
+      ];
+      if (columns.isEmpty) return false;
+
+      final filters = <String>[];
+      if (t == 'decoded_rr') {
+        // The oracle drops a beat with no rec_ts and no usable rr_ts_ms even
+        // when ts_ms is present; letting it through fails NOT NULL for the
+        // whole chunk.
+        filters.add('NOT ($sourceRecTs IS NULL AND NOT (${numericSql(sourceRrTsMs)}))');
+      }
+      if (isDeviceKeyed) {
+        filters.add('NOT ($sourceTsMs IS NULL AND NOT (${numericSql(keyTime)}))');
+      }
+      if (t == 'day_result') {
+        // The oracle compares interpolated strings, before destination
+        // affinity can coerce a foreign TEXT version such as '0100' to 100.
+        filters.add('NOT EXISTS (SELECT 1 FROM temp._restore_protected_day_results p '
+            'WHERE CAST(p.day_id AS TEXT) = CAST(${sourceColumn('day_id')} AS TEXT) AND '
+            'CAST(p.algo_version AS TEXT) = CAST(${sourceColumn('algo_version')} AS TEXT))');
+      }
+      if (t == 'metric_series' ||
+          t == 'metric_series_version' ||
+          t == 'metric_method') {
+        filters.add('NOT EXISTS (SELECT 1 FROM temp._restore_finalized_days p '
+            'WHERE CAST(p.day_id AS TEXT) = CAST(${sourceColumn('date')} AS TEXT))');
+      }
+      final unbounded = filters.isEmpty ? '1' : filters.join(' AND ');
+      final where = ['s.rowid > ?', 's.rowid <= ?', ...filters].join(' AND ');
+      final selectSql = 'SELECT ${[
+        for (final column in columns)
+          '${expressions[column]} AS ${quoteSql(column)}',
+      ].join(', ')} FROM imp.${quoteSql(t)} s WHERE $where';
+      // Rowid order: when derived keys collide, the later source row wins, as
+      // on the per-row path.
+      final insertSql = 'INSERT OR REPLACE INTO main.${quoteSql(t)} '
+          '(${columns.map(quoteSql).join(', ')}) $selectSql ORDER BY s.rowid';
+
+      // Small enough that a band-sync commit (which must land before the ACK)
+      // never waits long behind one, and a salvage loses little to one
+      // unreadable chunk.
+      const chunkRows = 25000;
+      var failed = true;
+      var copied = 0;
+      try {
+        // Chunk ends by row count, not fixed rowid ranges: a legacy
+        // `rec_ts INTEGER PRIMARY KEY` table has rowids near 1.8e9, which
+        // fixed 25k ranges walked as ~70k mostly empty transactions.
+        final ends = <int>[];
+        for (var lo = 0;;) {
+          final hi = Sqflite.firstIntValue(await db.rawQuery(
+            'SELECT MAX(r) FROM (SELECT rowid AS r FROM imp.${quoteSql(t)} '
+            'WHERE rowid > ? ORDER BY rowid LIMIT $chunkRows)',
+            [lo],
+          ));
+          if (hi == null) break;
+          ends.add(hi);
+          lo = hi;
+        }
+        if (t == 'decoded_rr') {
+          // The highest beat per second over the WHOLE source, not one chunk:
+          // REPLACE renumbers rowids, so one second's beats can straddle
+          // chunks out of order, and a per-chunk max would clear beats an
+          // earlier chunk just restored.
+          // Unlike the oracle's per-2k-page sweep, this also retains an earlier
+          // high beat when a later oracle page carries only lower indices.
+          await db.execute('DROP TABLE IF EXISTS temp._restore_rr_src_max');
+          await db.execute('''CREATE TEMP TABLE _restore_rr_src_max
+            (d TEXT, k INTEGER, m INTEGER, PRIMARY KEY (d, k))''');
+          // Paged like the copy below, so a band-sync commit can land between
+          // pages instead of waiting out a scan of the whole source.
+          for (var i = 0; i < ends.length; i++) {
+            // REPLACE with a LEFT JOIN rather than an UPSERT: Android 8-10
+            // ship a system SQLite older than ON CONFLICT DO UPDATE.
+            // GROUP BY ordinals, not d/k: a source column of the same name
+            // would win over the alias.
+            await db.execute('''INSERT OR REPLACE INTO temp._restore_rr_src_max
+                (d, k, m)
+              SELECT c.d, c.k, MAX(c.m, COALESCE(x.m, c.m))
+              FROM (SELECT $derivedDeviceId AS d, $derivedTsMs AS k,
+                      MAX(CAST(${sourceColumn('beat_index')} AS INTEGER)) AS m
+                    FROM imp.${quoteSql(t)} s
+                    WHERE s.rowid > ? AND s.rowid <= ? AND $unbounded
+                      AND ${numericSql(sourceColumn('beat_index'))}
+                    GROUP BY 1, 2) c
+              LEFT JOIN temp._restore_rr_src_max x ON x.d = c.d AND x.k = c.k''',
+                [i == 0 ? 0 : ends[i - 1], ends[i]]);
+          }
+        }
+        var lower = 0;
+        for (final upper in ends) {
+          final first = lower == 0;
+          await db.transaction((txn) async {
+            if (t == 'day_result' && first) {
+              // Inside the first chunk's transaction, so a day the engine
+              // finalizes while the restore runs is still protected.
+              await txn.execute(
+                  'DROP TABLE IF EXISTS temp._restore_protected_day_results');
+              await txn.execute(
+                  'CREATE TEMP TABLE _restore_protected_day_results AS '
+                  'SELECT day_id, algo_version FROM main.day_result '
+                  'WHERE finalized = 1');
+            }
+            if (t == 'day_result') {
+              final days = await txn.rawQuery(
+                'SELECT DISTINCT ${sourceColumn('day_id')} AS day_id '
+                'FROM imp.${quoteSql(t)} s WHERE $where',
+                [lower, upper],
+              );
+              importedDays?.addAll(
+                days.map((row) => '${row['day_id']}'),
+              );
+            }
+            await txn.execute(insertSql, [lower, upper]);
+            copied += Sqflite.firstIntValue(
+                  await txn.rawQuery('SELECT changes()'),
+                ) ??
+                0;
+            if (t == 'decoded_rr') {
+              // Materialised and driven from the chunk's seconds, so the
+              // delete walks decoded_rr's (device_id, ts_ms, beat_index) key
+              // once per carried second. A correlated EXISTS over the CTE
+              // scanned every local beat against every carried second.
+              await txn.execute('DROP TABLE IF EXISTS temp._restore_rr_carried');
+              await txn.execute('''CREATE TEMP TABLE _restore_rr_carried AS
+                SELECT DISTINCT x.d AS d, x.k AS k, x.m AS m
+                FROM (SELECT $derivedDeviceId AS d, $derivedTsMs AS k
+                      FROM imp.${quoteSql(t)} s WHERE $where
+                        AND ${numericSql(sourceColumn('beat_index'))}) c
+                JOIN temp._restore_rr_src_max x ON x.d = c.d AND x.k = c.k''',
+                  [lower, upper]);
+              await txn.execute('''DELETE FROM main.decoded_rr WHERE rowid IN (
+                SELECT r.rowid FROM temp._restore_rr_carried c
+                JOIN main.decoded_rr r
+                  ON r.device_id = c.d AND r.ts_ms = c.k AND r.beat_index > c.m)''');
+              await txn.execute('DROP TABLE temp._restore_rr_carried');
+            }
+          });
+          lower = upper;
+        }
+        failed = false;
+      } finally {
+        await cleanupRestore(
+          tempTables: [
+            if (t == 'day_result') '_restore_protected_day_results',
+            if (t == 'decoded_rr') ...[
+              '_restore_rr_carried', '_restore_rr_src_max',
+            ],
+          ],
+          preserveError: failed,
+        );
+      }
+      counts[t] = copied;
+      return true;
+    }
+
     try {
+      if (!debugRowByRowMerge) {
+        try {
+          await db.execute('ATTACH DATABASE ? AS imp', [path]);
+          impAttached = true;
+        } on DatabaseException {
+          // A damaged-file salvage can still try the per-row reader. A public
+          // restore must propagate its failure instead of claiming success.
+          if (!tolerant) rethrow;
+        }
+      }
+      await db.transaction((txn) async {
+        finalizedDays.addAll({
+          for (final r in await txn.query(
+            'day_result',
+            columns: ['day_id'],
+            where: 'finalized = 1',
+            distinct: true,
+          ))
+            '${r['day_id']}',
+        });
+        if (impAttached) {
+          await txn.execute('DROP TABLE IF EXISTS temp._restore_finalized_days');
+          await txn.execute('CREATE TEMP TABLE _restore_finalized_days AS '
+              'SELECT day_id FROM main.day_result WHERE finalized = 1');
+        }
+      });
       if ((only ?? tables).contains(SubstrateArchive.table)) {
         try {
           await SubstrateArchive.adoptPolicy(src, db);
@@ -11640,6 +11992,7 @@ class LocalDb {
           continue;
         }
         try {
+          if (await mergeBulkTable(t)) continue;
           // PAGED SOURCE READ — never `SELECT *` a whole table.
           //
           // This used to be a single `src.query(t)`. sqflite serialises an entire
@@ -12159,8 +12512,16 @@ class LocalDb {
           counts[t] = 0;
         }
       }
+      failed = false;
     } finally {
-      await src.close();
+      await cleanupRestore(
+        tempTables: impAttached ? [
+          '_restore_rr_carried', '_restore_rr_src_max',
+          '_restore_protected_day_results', '_restore_finalized_days',
+        ] : const [],
+        closeSource: true,
+        preserveError: failed,
+      );
     }
     if (counts.values.any((count) => count > 0)) {
       await ActivityStore.suppressImportedAlerts(db);
@@ -12211,7 +12572,8 @@ class LocalDb {
     }
     // Last, so it can never be mistaken for a table row count by anything that
     // walks this map in order.
-    if (importedDays != null) counts['_days'] = importedDays.length;
+    final days = importedDays;
+    if (days != null) counts['_days'] = days.length;
     // Archive buckets that could not be merged (a codec this build does not
     // know) — kept out of the row counts, never silently.
     if (archiveSkipped > 0) counts['_substrate_archive_skipped'] = archiveSkipped;
@@ -13040,9 +13402,10 @@ class LocalDb {
 
   static Future<void> putComputeFreshness(
     String key,
-    String payloadJson,
-  ) async {
-    final db = await instance;
+    String payloadJson, {
+    DatabaseExecutor? txn,
+  }) async {
+    final db = txn ?? await instance;
     await db.insert('compute_freshness', {
       'key': key,
       'payload_json': payloadJson,
diff --git a/test/import_bulk_restore_test.dart b/test/import_bulk_restore_test.dart
new file mode 100644
index 000000000..989f7d478
--- /dev/null
+++ b/test/import_bulk_restore_test.dart
@@ -0,0 +1,995 @@
+// ED-1 E2 RED: ported from cb81734f onto upstream's LocalDb restore owner.
+// Upstream still reads 2,000 rows into Dart maps and batches the writes.
+// The proposed bulk path copies inside SQLite with ATTACH + INSERT SELECT,
+// in 25,000-row chunks. These tests hold that path to the existing merge:
+//   1. the same final database, rule by rule (LocalDb.debugMergeRowByRow is
+//      the oracle);
+//   2. a bounded number of sqflite calls per big table, not one per page;
+//   3. chunks commit separately and a rerun converges;
+//   4. the source is detached on success and on failure.
+
+import 'dart:io';
+import 'dart:typed_data';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:path/path.dart' as p;
+import 'package:shared_preferences/shared_preferences.dart';
+import 'package:sqflite_common/sqflite_logger.dart';
+import 'package:sqflite_common_ffi/sqflite_ffi.dart';
+import 'package:openstrap_edge/data/db.dart';
+import 'package:openstrap_edge/data/substrate_archive_codec.dart';
+
+// Every sqflite call LocalDb makes, while [_counting] is on.
+final _invokes = <SqfliteLoggerInvokeEvent>[];
+var _counting = false;
+
+/// The SQL strings one sqflite call carries (a batch carries one per op).
+Iterable<String> _sqlOf(Object? args) sync* {
+  if (args is Map) {
+    final sql = args['sql'];
+    if (sql is String) yield sql;
+    for (final v in args.values) {
+      if (v is List || v is Map) yield* _sqlOf(v);
+    }
+  } else if (args is List) {
+    for (final v in args) {
+      yield* _sqlOf(v);
+    }
+  }
+}
+
+Iterable<Map> _operations(Object? args) sync* {
+  if (args is Map) {
+    if (args['sql'] is String) yield args;
+    for (final v in args.values) {
+      if (v is List || v is Map) yield* _operations(v);
+    }
+  } else if (args is List) {
+    for (final v in args) {
+      yield* _operations(v);
+    }
+  }
+}
+
+/// sqflite calls whose SQL names [table].
+int _callsTouching(String table) {
+  final re = RegExp('\\b$table\\b');
+  return _invokes
+      .where((e) => _sqlOf(e.arguments).any(re.hasMatch))
+      .length;
+}
+
+const _t0 = 1790000000; // 2026-09-21, an arbitrary fixed second
+
+/// Stamped with DateTime.now(): differs between any two
+/// runs, so the equivalence dump leaves it out.
+const _wallClockColumns = {
+  'compute_freshness': {'updated_at'},
+  'activity_review_meta': {'activated_at', 'alert_after_ts'},
+  'substrate_archive': {'updated_at'},
+};
+
+Map<String, Object?> archiveSecond(int ts, int hr) => {
+  'device_id': '', 'ts_ms': ts * 1000, 'rec_ts': ts, 'counter': ts - _t0,
+  'hr': hr,
+};
+
+Map<String, Object?> archiveRow(List<Map<String, Object?>> rows) {
+  final table = (ArchiveTableBuilder()..addRows(rows)).build();
+  final w = SubstrateArchiveCodec.encodeMergeVerify(
+      ArchiveBucket(table, ArchiveTable.empty));
+  return {
+    'device_id': '', 'utc_day': _t0 ~/ 86400,
+    'codec': SubstrateArchiveCodec.codecId,
+    'from_ts': w.fromTs, 'to_ts': w.toTs,
+    'onehz_rows': w.onehzRows, 'rr_rows': w.rrRows,
+    'raw_bytes': w.rawBytes, 'fingerprint': w.fingerprint,
+    'blob': w.blob, 'created_at': 1, 'updated_at': 1,
+  };
+}
+
+void main() {
+  late Directory tmp;
+  late String dbDir;
+
+  setUpAll(() async {
+    sqfliteFfiInit();
+    // The logger is the only public hook that sees every sqflite call.
+    // ignore: experimental_member_use
+    databaseFactory = SqfliteDatabaseFactoryLogger(
+      databaseFactoryFfi,
+      options: SqfliteLoggerOptions(
+        type: SqfliteDatabaseFactoryLoggerType.invoke,
+        log: (e) {
+          if (_counting && e is SqfliteLoggerInvokeEvent) _invokes.add(e);
+        },
+      ),
+    );
+    SharedPreferences.setMockInitialValues({});
+    LocalDb.dbName = 'openstrap_bulk_restore_test.db';
+    dbDir = await databaseFactory.getDatabasesPath();
+    tmp = await Directory.systemTemp.createTemp('openstrap_bulk_');
+  });
+
+  tearDownAll(() async {
+    await LocalDb.close();
+    await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName));
+    if (await tmp.exists()) await tmp.delete(recursive: true);
+  });
+
+  tearDown(() {
+    _counting = false;
+    _invokes.clear();
+    expect(LocalDb.debugRowByRowMerge, isFalse);
+  });
+
+  /// A brand-new, empty local database.
+  Future<Database> freshLocal() async {
+    await LocalDb.close();
+    await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName));
+    SharedPreferences.setMockInitialValues({});
+    return LocalDb.instance;
+  }
+
+  /// Every table, every row, as SQL literals (quote() keeps the storage
+  /// class: 1, 1.0 and '1' differ), sorted so row order does not matter.
+  Future<Map<String, List<String>>> dump(
+      {Set<String>? only, Set<String> except = const {}}) async {
+    final db = await LocalDb.instance;
+    final tables = [
+      for (final r in await db.rawQuery(
+          "SELECT name FROM sqlite_master WHERE type = 'table' "
+          "AND name NOT LIKE 'sqlite_stat%' ORDER BY name"))
+        r['name'] as String,
+    ].where((t) => (only == null || only.contains(t)) && !except.contains(t));
+    final out = <String, List<String>>{};
+    for (final t in tables) {
+      final cols = [
+        for (final c in await db.rawQuery('PRAGMA table_info("$t")'))
+          if (!(_wallClockColumns[t]?.contains(c['name']) ?? false))
+            c['name'] as String,
+      ];
+      final rows = await db.rawQuery(
+          'SELECT ${cols.map((c) => 'quote("$c")').join(" || '|' || ")} '
+          'AS r FROM "$t"');
+      out[t] = [for (final r in rows) r['r'] as String]..sort();
+    }
+    return out;
+  }
+
+  // ── fixture: one source row per per-row rule ───────────────────────────────
+
+  /// The local state the import lands on: a finalized day, a non-finalized
+  /// day, and beats from two devices on one second.
+  Future<void> seedLocal() async {
+    final db = await LocalDb.instance;
+    // Collision must REPLACE the 1 Hz value, including replacing it with NULL.
+    await db.insert('decoded_onehz', {
+      'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': _t0,
+      'counter': 99, 'hr': 99, 'step_count': 123,
+    });
+    await db.insert('day_result', {
+      'day_id': '2026-09-01', 'algo_version': 100,
+      'payload_json': '{"who":"local"}', 'computed_at': 1, 'finalized': 1,
+    });
+    await db.insert('day_result', {
+      'day_id': '2026-09-03', 'algo_version': 100,
+      'payload_json': '{"who":"local3"}', 'computed_at': 1, 'finalized': 0,
+    });
+    await db.insert('metric_series',
+        {'date': '2026-09-01', 'key': 'rhr', 'value': 50.0});
+    await db.insert('metric_series_version',
+        {'date': '2026-09-01', 'algo_version': 100, 'source': 'band'});
+    await db.insert('metric_method', {
+      'date': '2026-09-01', 'key': 'rhr',
+      'method': 'local', 'family': 'f', 'class': 'c',
+    });
+    await db.insert('baselines', {
+      'key': 'movement_floor', 'payload_json': '{"who":"local"}',
+      'updated_at': 1,
+    });
+    for (final dev in ['', 'band2']) {
+      for (var b = 0; b < 4; b++) {
+        await db.insert('decoded_rr', {
+          'device_id': dev, 'ts_ms': _t0 * 1000, 'rec_ts': _t0,
+          'beat_index': b, 'rr_ts_ms': _t0 * 1000 + b * 800, 'rr_ms': 700 + b,
+        });
+      }
+    }
+    // A second the source never mentions: untouched.
+    for (var b = 0; b < 2; b++) {
+      await db.insert('decoded_rr', {
+        'device_id': '', 'ts_ms': (_t0 + 20) * 1000, 'rec_ts': _t0 + 20,
+        'beat_index': b, 'rr_ts_ms': (_t0 + 20) * 1000, 'rr_ms': 900 + b,
+      });
+    }
+    // Upstream's documented restore gap: a 1 Hz second without source RR
+    // does NOT clear local beats. Keep that behavior in this performance PR.
+    await db.insert('decoded_rr', {
+      'device_id': '', 'ts_ms': (_t0 + 1) * 1000, 'rec_ts': _t0 + 1,
+      'beat_index': 0, 'rr_ts_ms': (_t0 + 1) * 1000, 'rr_ms': 888,
+    });
+    await db.insert(LocalDb.rawArchiveView, {
+      'device_id': '', 'hex': 'local-only', 'counter': 1,
+      'packet_type': 32, 'rec_ts': _t0, 'captured_at': 1,
+      'reason': 'undecodable_rec_v20',
+    });
+    await db.insert('substrate_archive', archiveRow([archiveSecond(_t0, 65)]));
+  }
+
+  /// A foreign export exercising every per-row rule. No PRIMARY KEYs and no
+  /// NOT NULLs, so a value of the wrong storage class can be planted.
+  Future<String> buildRulesSource() async {
+    final path = p.join(tmp.path, 'rules_src.db');
+    await databaseFactoryFfi.deleteDatabase(path);
+    final src = await databaseFactoryFfi.openDatabase(path);
+    // decoded_onehz: a column this build does not know (future_col), columns
+    // this build has that the source lacks (step_count etc.), legacy
+    // on_wrist/hr_valid values and the -50 °C sentinel.
+    await src.execute('''CREATE TABLE decoded_onehz (
+      device_id TEXT, ts_ms INTEGER, rec_ts, counter INTEGER, hr INTEGER,
+      ax REAL, skin_temp_c REAL, on_wrist INTEGER, hr_valid INTEGER,
+      future_col TEXT)''');
+    Future<void> onehz(Object? dev, Object? tsMs, Object? recTs, int counter,
+            double? skin) =>
+        src.insert('decoded_onehz', {
+          'device_id': dev, 'ts_ms': tsMs, 'rec_ts': recTs,
+          'counter': counter, 'hr': 60 + counter, 'ax': 0.5,
+          'skin_temp_c': skin, 'on_wrist': 1, 'hr_valid': 1,
+          'future_col': 'x$counter',
+        });
+    await onehz(null, null, _t0, 1, -50.0); // sentinel -> NULL, key derived
+    await onehz(null, null, _t0 + 1, 2, -49.99); // just above: kept
+    await onehz('band2', (_t0 + 2) * 1000, _t0 + 2, 3, 33.5); // keyed: as is
+    await onehz(null, null, 'abc', 4, 33.0); // non-numeric rec_ts: dropped
+    await onehz(null, null, '${_t0 + 4}', 5, 33.0); // numeric TEXT: dropped
+    await onehz('', null, (_t0 + 6).toDouble(), 6, 33.0); // REAL rec_ts
+    await onehz(null, null, _t0 + 7, 7, -49.995); // boundary: NULL
+
+    // decoded_rr, legacy: no rec_ts, no device_id, no ts_ms.
+    await src.execute('''CREATE TABLE decoded_rr (
+      counter INTEGER, beat_index INTEGER, rr_ts_ms, rr_ms INTEGER)''');
+    Future<void> rr(int counter, int beat, Object? rrTsMs, int ms) =>
+        src.insert('decoded_rr', {
+          'counter': counter, 'beat_index': beat,
+          'rr_ts_ms': rrTsMs, 'rr_ms': ms,
+        });
+    await rr(1, 0, _t0 * 1000, 500); // over local primary beats 0..3
+    await rr(1, 1, _t0 * 1000 + 600, 510);
+    await rr(9, 0, 'bad', 520); // non-numeric: dropped
+    await rr(11, 0, (_t0 + 10) * 1000, 530);
+    await rr(11, 1, (_t0 + 10) * 1000 + 700, 540);
+    await rr(11, 2, (_t0 + 10) * 1000 + 900, 550);
+
+    // samples, legacy: no device_id / ts_ms.
+    await src.execute('CREATE TABLE samples (counter INTEGER, ts, hr INTEGER)');
+    await src.insert('samples', {'counter': 1, 'ts': _t0, 'hr': 61});
+    await src.insert('samples', {'counter': 2, 'ts': _t0 + 1, 'hr': 62});
+    await src.insert('samples', {'counter': 3, 'ts': 'nope', 'hr': 63});
+
+    // day_result without window_json / skipped / partial (dest defaults).
+    await src.execute('''CREATE TABLE day_result (
+      day_id TEXT, algo_version INTEGER, payload_json TEXT,
+      computed_at INTEGER, finalized INTEGER, rhr REAL, rmssd REAL,
+      readiness REAL)''');
+    Future<void> day(String d, int v, String who, int fin) =>
+        src.insert('day_result', {
+          'day_id': d, 'algo_version': v, 'payload_json': '{"who":"$who"}',
+          'computed_at': 2, 'finalized': fin, 'rhr': 55.0,
+        });
+    await day('2026-09-01', 100, 'src', 1); // locally finalized: protected
+    await day('2026-09-01', 101, 'src101', 1); // new version: written
+    await day('2026-09-02', 100, 'staleSrc2', 1); // duplicate: later row wins
+    await day('2026-09-02', 100, 'src2', 1);
+    await day('2026-09-03', 100, 'src3', 0); // local not finalized: replaced
+
+    await src.execute(
+        'CREATE TABLE metric_series (date TEXT, key TEXT, value REAL)');
+    for (final r in [
+      ['2026-09-01', 'rhr', 60.0],
+      ['2026-09-01', 'hrv', 40.0],
+      ['2026-09-02', 'rhr', 55.0],
+      ['2026-09-03', 'rhr', 57.0],
+    ]) {
+      await src.insert(
+          'metric_series', {'date': r[0], 'key': r[1], 'value': r[2]});
+    }
+    await src.execute('CREATE TABLE metric_series_version '
+        '(date TEXT, algo_version INTEGER, source TEXT)');
+    for (final d in ['2026-09-01', '2026-09-02']) {
+      await src.insert('metric_series_version',
+          {'date': d, 'algo_version': 101, 'source': 'src'});
+    }
+    await src.execute('CREATE TABLE metric_method (date TEXT, key TEXT, '
+        'method TEXT, family TEXT, class TEXT)');
+    for (final d in ['2026-09-01', '2026-09-02']) {
+      await src.insert('metric_method', {
+        'date': d, 'key': 'rhr', 'method': 'src', 'family': 'f', 'class': 'c',
+      });
+    }
+    await src.execute('CREATE TABLE baselines '
+        '(key TEXT, payload_json TEXT, updated_at INTEGER)');
+    for (final key in ['movement_floor', 'new_baseline']) {
+      await src.insert('baselines', {
+        'key': key, 'payload_json': '{"who":"source"}', 'updated_at': 2,
+      });
+    }
+
+    // events: upstream's shape plus an unknown column.
+    await src.execute('''CREATE TABLE events (device_id TEXT, hex TEXT,
+      event_id INTEGER, ts INTEGER, captured_at INTEGER, extra TEXT)''');
+    await src.insert('events', {
+      'device_id': '', 'hex': 'aa01', 'event_id': 3, 'ts': _t0,
+      'captured_at': _t0 * 1000, 'extra': 'zz',
+    });
+    await src.execute('CREATE TABLE band_battery (device_id TEXT, ts INTEGER, '
+        'battery_pct REAL, source TEXT)');
+    await src.insert('band_battery', {
+      'device_id': '', 'ts': _t0, 'battery_pct': 70.0, 'source': 'history',
+    });
+    await src.execute('CREATE TABLE band_events (device_id TEXT, hex TEXT, '
+        'event_id INTEGER, name TEXT, ts INTEGER, captured_at INTEGER)');
+    await src.insert('band_events', {
+      'device_id': '', 'hex': 'bb01', 'event_id': 3, 'name': 'test',
+      'ts': _t0, 'captured_at': 2,
+    });
+    await src.execute('CREATE TABLE raw_archive (device_id TEXT, hex TEXT, '
+        'counter INTEGER, packet_type INTEGER, rec_ts INTEGER, '
+        'captured_at INTEGER, reason TEXT)');
+    for (final hex in ['src-a', 'src-b']) {
+      await src.insert('raw_archive', {
+        'device_id': '', 'hex': hex, 'counter': 1, 'packet_type': 32,
+        'rec_ts': _t0, 'captured_at': 1, 'reason': 'undecodable_rec_v20',
+      });
+    }
+    // Use upstream's lossless archive, including a collision and a new second.
+    // A straight REPLACE of this bucket would lose the local value.
+    final db = await LocalDb.instance;
+    final ddl = (await db.rawQuery("SELECT sql FROM sqlite_master "
+        "WHERE type = 'table' AND name = 'substrate_archive'")).single;
+    await src.execute(ddl['sql'] as String);
+    await src.insert('substrate_archive', archiveRow([
+      archiveSecond(_t0, 1), archiveSecond(_t0 + 1, 66),
+    ]));
+    await src.close();
+    return path;
+  }
+
+  group('equivalence law: bulk == per-row', () {
+    late String rulesSrc;
+    setUpAll(() async => rulesSrc = await buildRulesSource());
+
+    Future<(Map<String, int>, Map<String, List<String>>)> run(
+        Future<Map<String, int>> Function(String) merge) async {
+      await freshLocal();
+      await seedLocal();
+      final counts = await merge(rulesSrc);
+      return (counts, await dump());
+    }
+
+    test('every table ends byte-identical, and the counts agree', () async {
+      final (oracleCounts, oracle) = await run(LocalDb.debugMergeRowByRow);
+      final (bulkCounts, bulk) = await run(LocalDb.importFromDbFile);
+      for (final t in oracle.keys) {
+        expect(bulk[t], oracle[t], reason: 'table $t');
+      }
+      expect(bulk.keys.toSet(), oracle.keys.toSet());
+      expect(bulkCounts, oracleCounts);
+    });
+
+    group('each rule, by hand', () {
+      late Map<String, int> counts;
+      late Database db;
+      setUpAll(() async {
+        db = await freshLocal();
+        await seedLocal();
+        counts = await LocalDb.importFromDbFile(rulesSrc);
+      });
+
+      test('decoded_onehz: retired columns NULL, sentinel NULL, keys derived, '
+          'non-numeric rec_ts dropped, unknown column dropped', () async {
+        final rows = await db.rawQuery(
+            'SELECT device_id, ts_ms, rec_ts, counter, skin_temp_c, on_wrist, '
+            'hr_valid, step_count FROM decoded_onehz ORDER BY device_id, ts_ms');
+        expect([for (final r in rows) r['counter']], [1, 2, 6, 7, 3]);
+        final by = {for (final r in rows) r['counter']: r};
+        expect(by[1]!['device_id'], '');
+        expect(by[1]!['ts_ms'], _t0 * 1000);
+        expect(by[1]!['skin_temp_c'], isNull);
+        expect(by[2]!['skin_temp_c'], -49.99);
+        expect(by[7]!['skin_temp_c'], isNull);
+        expect(by[3]!['device_id'], 'band2');
+        expect(by[3]!['ts_ms'], (_t0 + 2) * 1000);
+        expect(by[6]!['ts_ms'], (_t0 + 6) * 1000);
+        for (final r in rows) {
+          expect(r['on_wrist'], isNull);
+          expect(r['hr_valid'], isNull);
+          expect(r['step_count'], isNull, reason: 'missing column: default');
+        }
+        expect(counts['decoded_onehz'], 5);
+      });
+
+      test('decoded_rr: rec_ts from rr_ts_ms, bad row dropped, stale local '
+          'tail cleared for the writing device only', () async {
+        final rows = await db.rawQuery(
+            'SELECT device_id, ts_ms, rec_ts, beat_index, rr_ms FROM decoded_rr '
+            'ORDER BY device_id, ts_ms, beat_index');
+        expect([
+          for (final r in rows)
+            '${r['device_id']}/${r['ts_ms']}/${r['beat_index']}/${r['rr_ms']}'
+        ], [
+          '/${_t0 * 1000}/0/500',
+          '/${_t0 * 1000}/1/510',
+          '/${(_t0 + 1) * 1000}/0/888',
+          '/${(_t0 + 10) * 1000}/0/530',
+          '/${(_t0 + 10) * 1000}/1/540',
+          '/${(_t0 + 10) * 1000}/2/550',
+          '/${(_t0 + 20) * 1000}/0/900',
+          '/${(_t0 + 20) * 1000}/1/901',
+          'band2/${_t0 * 1000}/0/700',
+          'band2/${_t0 * 1000}/1/701',
+          'band2/${_t0 * 1000}/2/702',
+          'band2/${_t0 * 1000}/3/703',
+        ]);
+        expect(rows.firstWhere((r) => r['ts_ms'] == (_t0 + 10) * 1000)['rec_ts'],
+            _t0 + 10);
+        expect(counts['decoded_rr'], 5);
+      });
+
+      test('samples: legacy rows keyed to the primary band at ts*1000', () async {
+        final rows = await db.rawQuery(
+            'SELECT device_id, ts_ms, ts FROM samples ORDER BY ts_ms');
+        expect([for (final r in rows) '${r['device_id']}/${r['ts_ms']}'],
+            ['/${_t0 * 1000}', '/${(_t0 + 1) * 1000}']);
+      });
+
+      test('1 Hz without source beats retains local RR, the upstream restore '
+          'gap; the replacement 1 Hz value still wins', () async {
+        final rows = await db.query('decoded_onehz',
+            where: 'device_id = ? AND ts_ms = ?',
+            whereArgs: ['', _t0 * 1000]);
+        expect(rows.single['hr'], 61);
+        expect(rows.single['step_count'], isNull);
+        final beats = await db.query('decoded_rr',
+            where: 'device_id = ? AND ts_ms = ?',
+            whereArgs: ['', (_t0 + 1) * 1000]);
+        expect(beats.single['rr_ms'], 888);
+      });
+
+      test('day_result: a finalized local (day, version) is never overwritten; '
+          '_days counts distinct days written', () async {
+        final rows = await db.rawQuery(
+            'SELECT day_id, algo_version, payload_json, window_json, skipped, '
+            'partial FROM day_result ORDER BY day_id, algo_version');
+        expect([
+          for (final r in rows)
+            '${r['day_id']}/${r['algo_version']}/${r['payload_json']}'
+        ], [
+          '2026-09-01/100/{"who":"local"}',
+          '2026-09-01/101/{"who":"src101"}',
+          '2026-09-02/100/{"who":"src2"}',
+          '2026-09-03/100/{"who":"src3"}',
+        ]);
+        final imported = rows.where((r) => r['payload_json'] != '{"who":"local"}');
+        for (final r in imported) {
+          expect(r['window_json'], '{}', reason: 'dest default');
+          expect(r['skipped'], 0);
+          expect(r['partial'], 0);
+        }
+        expect(counts['_days'], 3);
+        expect(counts['day_result'], 4);
+      });
+
+      test('metric_series / _version / metric_method skip a locally '
+          'finalized day', () async {
+        final series = await db.rawQuery(
+            'SELECT date, key, value FROM metric_series ORDER BY date, key');
+        expect([for (final r in series) '${r['date']}/${r['key']}/${r['value']}'],
+            [
+              '2026-09-01/rhr/50.0',
+              '2026-09-02/rhr/55.0',
+              '2026-09-03/rhr/57.0',
+            ]);
+        final ver = await db.rawQuery(
+            'SELECT date, algo_version FROM metric_series_version ORDER BY date');
+        expect([for (final r in ver) '${r['date']}/${r['algo_version']}'],
+            ['2026-09-01/100', '2026-09-02/101']);
+        final method = await db.rawQuery(
+            'SELECT date, method FROM metric_method ORDER BY date');
+        expect([for (final r in method) '${r['date']}/${r['method']}'],
+            ['2026-09-01/local', '2026-09-02/src']);
+      });
+
+      test('a table the export lacks is skipped; an unknown column is dropped',
+          () async {
+        expect(counts.containsKey('external_hr'), isFalse);
+        final ev = await db.rawQuery('SELECT * FROM events');
+        expect(ev, hasLength(1));
+        expect(ev.single.containsKey('extra'), isFalse);
+      });
+
+      test('finalized history keeps local baselines and admits missing keys',
+          () async {
+        final rows = await db.query('baselines', orderBy: 'key');
+        expect([for (final r in rows) '${r['key']}/${r['payload_json']}'], [
+          'movement_floor/{"who":"local"}', 'new_baseline/{"who":"source"}',
+        ]);
+      });
+
+      test('raw archive keeps every frame; substrate archive merges losslessly '
+          'with local winning collisions', () async {
+        final raw = await db.query(LocalDb.rawArchiveView, orderBy: 'hex');
+        expect([for (final r in raw) r['hex']], ['local-only', 'src-a', 'src-b']);
+        expect(counts['raw_archive'], 2);
+        final row = (await db.query('substrate_archive')).single;
+        final bucket = SubstrateArchiveCodec.decode(row['blob'] as Uint8List)!;
+        expect(bucket.onehz.rowCount, 2);
+        expect([for (var i = 0; i < 2; i++) bucket.onehz.valueAt('hr', i)],
+            [65, 66]);
+        expect(counts['substrate_archive'], 1);
+      });
+    });
+
+    test('a corrupt source rethrows (a restore is not tolerant)', () async {
+      final path = await buildCorruptSource(tmp.path);
+      await freshLocal();
+      await expectLater(
+          LocalDb.importFromDbFile(path), throwsA(isA<DatabaseException>()));
+    });
+  });
+
+  group('25k boundary equivalence', () {
+    late String boundarySrc;
+    const n = 25001;
+    setUpAll(() async {
+      boundarySrc = p.join(tmp.path, 'boundary_src.db');
+      final src = await databaseFactoryFfi.openDatabase(boundarySrc);
+      // No PK: rows 1 and 25001 derive the same 1 Hz key. Give the last
+      // row an explicit, sparse rowid so range paging must cross a gap too.
+      await src.execute('CREATE TABLE decoded_onehz '
+          '(rec_ts INTEGER, counter INTEGER, hr INTEGER)');
+      await src.execute('''
+        WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n
+                                WHERE i < $n)
+        INSERT INTO decoded_onehz(rowid, rec_ts, counter, hr)
+        SELECT CASE WHEN i = $n THEN 50001 ELSE i END,
+               CASE WHEN i IN (1, $n) THEN $_t0 ELSE $_t0 + i END,
+               i, CASE WHEN i = $n THEN 87 ELSE 60 END FROM n''');
+      await src.execute('''CREATE TABLE decoded_rr (
+        device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER,
+        rr_ts_ms INTEGER, rr_ms INTEGER)''');
+      // Higher beat first at rowid 24999, lower beat last at rowid 25001.
+      // Both fall in the SAME upstream 2k page, but DIFFERENT bulk chunks.
+      // Thus the oracle keeps both, while a per-25k-chunk tail max loses 1.
+      await src.execute('''
+        WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n
+                                WHERE i < $n)
+        INSERT INTO decoded_rr
+        SELECT '',
+               (CASE WHEN i IN (24999, 25001) THEN $_t0 ELSE $_t0 + i END) * 1000,
+               CASE WHEN i IN (24999, 25001) THEN $_t0 ELSE $_t0 + i END,
+               CASE WHEN i = 24999 THEN 1 ELSE 0 END,
+               ($_t0 + i) * 1000, 500 + i % 100 FROM n''');
+      await src.close();
+    });
+
+    test('later source row wins across chunks and rowid gaps; whole-source '
+        'beat maximum keeps an earlier higher index', () async {
+      Future<(Map<String, int>, Map<String, List<String>>)> run(
+          Future<Map<String, int>> Function(String) merge) async {
+        await freshLocal();
+        await seedLocal();
+        final counts = await merge(boundarySrc);
+        return (counts, await dump());
+      }
+      final (refCounts, ref) = await run(LocalDb.debugMergeRowByRow);
+      final (counts, actual) = await run(LocalDb.importFromDbFile);
+      expect(counts, refCounts);
+      expect(actual, ref);
+      final db = await LocalDb.instance;
+      final onehz = await db.query('decoded_onehz',
+          where: 'device_id = ? AND ts_ms = ?', whereArgs: ['', _t0 * 1000]);
+      expect(onehz.single['hr'], 87, reason: 'last source row wins');
+      final rr = await db.query('decoded_rr',
+          where: 'device_id = ? AND ts_ms = ?', whereArgs: ['', _t0 * 1000],
+          orderBy: 'beat_index');
+      expect([for (final r in rr) r['beat_index']], [0, 1]);
+      expect([for (final r in rr) r['rr_ms']], [501, 599]);
+      expect(await db.query('decoded_rr', where: 'device_id = ?',
+          whereArgs: ['band2']), hasLength(4));
+    }, timeout: const Timeout(Duration(minutes: 5)));
+
+    test('whole-source RR max preparation is paged at 25k rows', () async {
+      await freshLocal();
+      _invokes.clear();
+      _counting = true;
+      try {
+        await LocalDb.importFromDbFile(boundarySrc);
+      } finally {
+        _counting = false;
+      }
+      final prep = [
+        for (final e in _invokes)
+          for (final op in _operations(e.arguments))
+            if (RegExp(r'INSERT[\s\S]*\btemp\.[\s\S]*\bMAX\s*\('
+                r'[\s\S]*\bdecoded_rr\b', caseSensitive: false)
+                .hasMatch(op['sql'] as String)) op,
+      ];
+      expect(prep, hasLength(2),
+          reason: 'whole-source beat maxima need two bounded scans for 25001 rows');
+      for (var i = 0; i < prep.length; i++) {
+        final sql = prep[i]['sql'] as String;
+        expect(sql, matches(RegExp(r'rowid\s*>\s*\?', caseSensitive: false)));
+        expect(sql, matches(RegExp(r'rowid\s*<=\s*\?', caseSensitive: false)));
+        final args = prep[i]['arguments'] as List;
+        expect(args, hasLength(2));
+        expect(args[0], i * 25000);
+        expect(args[1], inInclusiveRange(i == 0 ? 25000 : n, (i + 1) * 25000));
+      }
+    }, timeout: const Timeout(Duration(minutes: 5)));
+  });
+
+  test("bulk keeps a second's higher beats that a later source page omits "
+      '(row-by-row drops them)', () async {
+    final path = p.join(tmp.path, 'rr_page_tail_src.db');
+    final src = await databaseFactoryFfi.openDatabase(path);
+    // No PK: repeated lower indices must get later rowids while the
+    // earlier higher beats remain in the source's union.
+    await src.execute('''CREATE TABLE decoded_rr (
+      device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER,
+      rr_ts_ms INTEGER, rr_ms INTEGER)''');
+    Future<void> beat(int b) => src.insert('decoded_rr', {
+      'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': _t0,
+      'beat_index': b, 'rr_ts_ms': _t0 * 1000 + b * 700, 'rr_ms': 700 + b,
+    });
+    for (var b = 0; b < 6; b++) {
+      await beat(b);
+    }
+    await src.execute('''
+      WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n
+                              WHERE i < 2001)
+      INSERT INTO decoded_rr
+      SELECT '', ($_t0 + i) * 1000, $_t0 + i, 0,
+             ($_t0 + i) * 1000, 800 FROM n''');
+    // Rowids 2008 and 2009 land on the oracle's second 2k page.
+    for (var b = 0; b < 2; b++) {
+      await beat(b);
+    }
+    final sourceBeats = await src.rawQuery(
+        'SELECT DISTINCT beat_index, rr_ts_ms, rr_ms FROM decoded_rr '
+        'WHERE device_id = ? AND ts_ms = ? ORDER BY beat_index',
+        ['', _t0 * 1000]);
+    await src.close();
+    expect([for (final r in sourceBeats) r['beat_index']], [0, 1, 2, 3, 4, 5]);
+
+    Future<List<Map<String, Object?>>> restoredBeats(Database db) =>
+        db.query('decoded_rr', columns: ['beat_index', 'rr_ts_ms', 'rr_ms'],
+            where: 'device_id = ? AND ts_ms = ?',
+            whereArgs: ['', _t0 * 1000], orderBy: 'beat_index');
+
+    final bulkDb = await freshLocal();
+    await LocalDb.importFromDbFile(path);
+    final bulk = await restoredBeats(bulkDb);
+    expect([for (final r in bulk) r['beat_index']], [0, 1, 2, 3, 4, 5]);
+    expect(bulk, sourceBeats, reason: 'bulk retains the whole-source union');
+
+    final oracleDb = await freshLocal();
+    await LocalDb.debugMergeRowByRow(path);
+    final oracle = await restoredBeats(oracleDb);
+    expect([for (final r in oracle) r['beat_index']], [0, 1],
+        reason: 'the later page maximum of 1 deletes earlier beats 2..5');
+    expect(oracle, sourceBeats.take(2).toList());
+  });
+
+  test('RR with a supplied key but no usable record time is dropped', () async {
+    final path = p.join(tmp.path, 'invalid_rr_src.db');
+    final src = await databaseFactoryFfi.openDatabase(path);
+    await src.execute('CREATE TABLE decoded_rr (device_id TEXT, ts_ms INTEGER, '
+        'rec_ts, beat_index INTEGER, rr_ts_ms, rr_ms INTEGER)');
+    await src.insert('decoded_rr', {
+      'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': null,
+      'beat_index': 0, 'rr_ts_ms': 'bad', 'rr_ms': 700,
+    });
+    await src.close();
+    await freshLocal();
+    final refCounts = await LocalDb.debugMergeRowByRow(path);
+    final ref = await dump();
+    await freshLocal();
+    final counts = await LocalDb.importFromDbFile(path);
+    expect(counts, refCounts);
+    expect(await dump(), ref);
+    expect(counts['decoded_rr'], 0);
+    expect(await (await LocalDb.instance).query('decoded_rr'), isEmpty);
+  });
+
+  test('a source RR column named d or k does not change the beat maxima',
+      () async {
+    final path = p.join(tmp.path, 'rr_alias_column_src.db');
+    await databaseFactoryFfi.deleteDatabase(path);
+    final src = await databaseFactoryFfi.openDatabase(path);
+    // GROUP BY prefers a source column over a result alias of the same name.
+    await src.execute('''CREATE TABLE decoded_rr (
+      device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER,
+      rr_ts_ms INTEGER, rr_ms INTEGER, d TEXT, k INTEGER)''');
+    for (var b = 0; b < 4; b++) {
+      await src.insert('decoded_rr', {
+        'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': _t0,
+        'beat_index': b, 'rr_ts_ms': _t0 * 1000 + b * 700, 'rr_ms': 700 + b,
+        'd': b == 3 ? 'a' : 'b', 'k': 3 - b,
+      });
+    }
+    await src.close();
+    await freshLocal();
+    final refCounts = await LocalDb.debugMergeRowByRow(path);
+    final ref = await dump();
+    final db = await freshLocal();
+    final counts = await LocalDb.importFromDbFile(path);
+    expect(counts, refCounts);
+    expect(await dump(), ref);
+    expect([for (final r in await db.query('decoded_rr', orderBy: 'beat_index'))
+      r['beat_index']], [0, 1, 2, 3]);
+  });
+
+  test('sparse source rowids are paged by row, not by rowid range', () async {
+    final path = p.join(tmp.path, 'sparse_rowid_src.db');
+    await databaseFactoryFfi.deleteDatabase(path);
+    final src = await databaseFactoryFfi.openDatabase(path);
+    // A legacy `rec_ts INTEGER PRIMARY KEY` table has rowids near 1.79e9.
+    // A smaller gap keeps the red run short: 40 empty 25k ranges.
+    await src.execute('CREATE TABLE decoded_onehz '
+        '(rec_ts INTEGER PRIMARY KEY, counter INTEGER, hr INTEGER)');
+    await src.execute('''CREATE TABLE decoded_rr (
+      device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER,
+      rr_ts_ms INTEGER, rr_ms INTEGER)''');
+    for (final rowid in [1, 1000001]) {
+      await src.insert('decoded_onehz',
+          {'rec_ts': rowid, 'counter': rowid, 'hr': 60});
+      await src.insert('decoded_rr', {
+        'rowid': rowid, 'device_id': '', 'ts_ms': (_t0 + rowid) * 1000,
+        'rec_ts': _t0 + rowid, 'beat_index': 0,
+        'rr_ts_ms': (_t0 + rowid) * 1000, 'rr_ms': 700,
+      });
+    }
+    await src.close();
+    await freshLocal();
+    final refCounts = await LocalDb.debugMergeRowByRow(path);
+    final ref = await dump();
+    await freshLocal();
+    _invokes.clear();
+    _counting = true;
+    final Map<String, int> counts;
+    try {
+      counts = await LocalDb.importFromDbFile(path);
+    } finally {
+      _counting = false;
+    }
+    expect(counts, refCounts);
+    expect(await dump(), ref);
+    for (final t in ['decoded_onehz', 'decoded_rr']) {
+      final inserts = [
+        for (final e in _invokes)
+          for (final sql in _sqlOf(e.arguments))
+            if (sql.startsWith('INSERT OR REPLACE INTO main."$t"')) sql,
+      ];
+      expect(inserts, hasLength(1), reason: '$t: 2 rows are one chunk');
+    }
+    expect(_callsTouching('decoded_rr'), lessThanOrEqualTo(15));
+  });
+
+  test('plain tables use ATTACH and INSERT SELECT on the public restore path',
+      () async {
+    final src = await buildRulesSource();
+    await freshLocal();
+    _invokes.clear();
+    _counting = true;
+    try {
+      await LocalDb.importFromDbFile(src);
+    } finally {
+      _counting = false;
+    }
+    final sql = [for (final e in _invokes) ..._sqlOf(e.arguments)];
+    expect(sql.any((s) => RegExp(r'^ATTACH\b', caseSensitive: false)
+        .hasMatch(s.trimLeft())), isTrue,
+        reason: 'bulk restore must attach the source on the destination handle');
+    for (final t in [
+      'decoded_onehz', 'decoded_rr', 'samples', 'events',
+      'band_events', 'band_battery', 'day_result', 'metric_series',
+      'metric_series_version', 'metric_method',
+    ]) {
+      expect(sql.any((s) => RegExp(
+          'INSERT\\s+OR\\s+REPLACE\\s+INTO\\s+(?:main\\.)?["`]?'
+          '$t\\b[\\s\\S]*\\bSELECT\\b', caseSensitive: false).hasMatch(s)),
+          isTrue, reason: '$t must copy inside SQLite');
+    }
+  });
+
+  // ── volume: sqflite calls per table, chunk commits ────────────────────────
+
+  group('200k decoded_onehz + 100k decoded_rr', () {
+    const nOneHz = 200000, nRr = 100000;
+    late String bigSrc;
+
+    setUpAll(() async {
+      bigSrc = p.join(tmp.path, 'big_src.db');
+      await databaseFactoryFfi.deleteDatabase(bigSrc);
+      final src = await databaseFactoryFfi.openDatabase(bigSrc);
+      await src.execute('''CREATE TABLE decoded_onehz (
+        device_id TEXT NOT NULL DEFAULT '', ts_ms INTEGER NOT NULL DEFAULT 0,
+        rec_ts INTEGER, counter INTEGER NOT NULL, hr INTEGER, ax REAL, ay REAL,
+        az REAL, PRIMARY KEY (device_id, ts_ms))''');
+      await src.execute('''CREATE TABLE decoded_rr (
+        device_id TEXT NOT NULL DEFAULT '', ts_ms INTEGER NOT NULL DEFAULT 0,
+        rec_ts INTEGER NOT NULL, beat_index INTEGER NOT NULL,
+        rr_ts_ms INTEGER NOT NULL, rr_ms INTEGER NOT NULL,
+        PRIMARY KEY (device_id, ts_ms, beat_index))''');
+      await src.execute('''
+        WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n
+                                WHERE i < $nOneHz)
+        INSERT INTO decoded_onehz
+        SELECT '', ($_t0 + i) * 1000, $_t0 + i, i, 40 + i % 100, 0.0, 0.0, 1.0
+        FROM n''');
+      await src.execute('''
+        WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n
+                                WHERE i < ${nRr ~/ 2}),
+             b(k) AS (VALUES (0), (1))
+        INSERT INTO decoded_rr
+        SELECT '', ($_t0 + i) * 1000, $_t0 + i, k,
+               ($_t0 + i) * 1000 + k * 700, 700 + i % 50
+        FROM n, b ORDER BY i, k''');
+      await src.close();
+    });
+
+    test('a bounded number of sqflite calls per table, not one per page',
+        () async {
+      await freshLocal();
+      _invokes.clear();
+      _counting = true;
+      final Map<String, int> counts;
+      try {
+        counts = await LocalDb.importFromDbFile(bigSrc);
+      } finally {
+        _counting = false;
+      }
+      expect(counts['decoded_onehz'], nOneHz);
+      expect(counts['decoded_rr'], nRr);
+      final onehz = _callsTouching('decoded_onehz');
+      final rr = _callsTouching('decoded_rr');
+      // Allow two calls per 25k 1 Hz chunk, five per RR chunk (including
+      // the paged whole-source max build and tail sweep), plus preparation.
+      // Upstream uses about two calls per 2k rows and exceeds both bounds.
+      expect(onehz, lessThanOrEqualTo(2 * (nOneHz ~/ 25000) + 10),
+          reason: 'decoded_onehz: $onehz sqflite calls');
+      expect(rr, lessThanOrEqualTo(5 * (nRr ~/ 25000) + 15),
+          reason: 'decoded_rr: $rr sqflite calls');
+    }, timeout: const Timeout(Duration(minutes: 10)));
+
+    test('a failure in a later chunk leaves earlier chunks committed, and a '
+        'rerun converges to the clean result', () async {
+      const data = {'decoded_onehz', 'decoded_rr'};
+      await freshLocal();
+      await LocalDb.debugMergeRowByRow(bigSrc);
+      final clean = await dump(only: data);
+
+      final db = await freshLocal();
+      // The LAST source row fails, whatever the chunk size.
+      await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON '
+          'decoded_onehz WHEN NEW.counter = $nOneHz '
+          "BEGIN SELECT RAISE(ABORT, 'injected'); END");
+      await expectLater(LocalDb.importFromDbFile(bigSrc), throwsA(anything));
+      final span = (await db.rawQuery(
+              'SELECT COUNT(*) AS n, MAX(counter) AS hi FROM decoded_onehz'))
+          .single;
+      final landed = span['n'] as int;
+      final maxCounter = span['hi'] as int?;
+      expect(landed, greaterThan(0), reason: 'chunk 1 committed');
+      expect(landed, lessThan(nOneHz), reason: 'the failing chunk rolled back');
+      expect(maxCounter, landed, reason: 'a contiguous rowid prefix');
+
+      await db.execute('DROP TRIGGER inject_fail');
+      await LocalDb.importFromDbFile(bigSrc);
+      expect(await dump(only: data), clean);
+    }, timeout: const Timeout(Duration(minutes: 15)));
+
+    test('bulk commits 25000 rows before a failure in the next chunk', () async {
+      final db = await freshLocal();
+      await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON '
+          'decoded_onehz WHEN NEW.counter = 25001 '
+          "BEGIN SELECT RAISE(ABORT, 'injected-25001'); END");
+      try {
+        await expectLater(LocalDb.importFromDbFile(bigSrc), throwsA(
+            isA<DatabaseException>().having((e) => e.toString(), 'message',
+                contains('injected-25001'))));
+        final span = (await db.rawQuery('SELECT COUNT(*) AS n, '
+            'MAX(counter) AS hi FROM decoded_onehz')).single;
+        expect(span, {'n': 25000, 'hi': 25000},
+            reason: 'exactly one 25k chunk must remain committed');
+      } finally {
+        await db.execute('DROP TRIGGER inject_fail');
+      }
+    }, timeout: const Timeout(Duration(minutes: 5)));
+  });
+
+  // ── ATTACH hygiene ─────────────────────────────────────────────────────────
+
+  group('the source is detached', () {
+    Future<void> expectClean() async {
+      final db = await LocalDb.instance;
+      expect([
+        for (final r in await db.rawQuery('PRAGMA database_list'))
+          if (r['name'] != 'temp') r['name'] as String,
+      ], ['main']);
+      expect(await db.rawQuery("SELECT name FROM sqlite_temp_master "
+          "WHERE type = 'table'"), isEmpty,
+          reason: 'restore preparation and carried-beat tables must be dropped');
+      // Exercise the same handle after cleanup. A leaked transaction or
+      // attachment must not prevent the next band-sync write.
+      await db.transaction((txn) => txn.rawQuery('SELECT 1'));
+    }
+
+    test('after a successful restore', () async {
+      final src = await buildRulesSource();
+      await freshLocal();
+      await LocalDb.importFromDbFile(src);
+      await expectClean();
+    });
+
+    test('after a restore that threw mid-copy', () async {
+      final src = await buildRulesSource();
+      final db = await freshLocal();
+      await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON '
+          "decoded_onehz BEGIN SELECT RAISE(ABORT, 'injected'); END");
+      await expectLater(LocalDb.importFromDbFile(src), throwsA(
+          isA<DatabaseException>().having((e) => e.toString(), 'message',
+              contains('injected'))));
+      await db.execute('DROP TRIGGER inject_fail');
+      await expectClean();
+      await LocalDb.importFromDbFile(src);
+      await expectClean();
+    });
+
+    for (final table in ['decoded_rr', 'day_result']) {
+      test('after $table preparation and a failed insert', () async {
+        final src = await buildRulesSource();
+        final db = await freshLocal();
+        await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON '
+            "$table BEGIN SELECT RAISE(ABORT, 'injected-$table'); END");
+        await expectLater(LocalDb.importFromDbFile(src), throwsA(
+            isA<DatabaseException>().having((e) => e.toString(), 'message',
+                contains('injected-$table'))));
+        await db.execute('DROP TRIGGER inject_fail');
+        await expectClean();
+        await LocalDb.importFromDbFile(src);
+        await expectClean();
+      });
+    }
+
+    test('after a corrupt source', () async {
+      final path = await buildCorruptSource(tmp.path);
+      await freshLocal();
+      await expectLater(LocalDb.importFromDbFile(path), throwsA(anything));
+      await expectClean();
+    });
+  });
+}
+
+/// An export whose header and schema read fine but whose decoded_onehz pages
+/// are garbage: opening works, reading the table fails SQLITE_CORRUPT.
+Future<String> buildCorruptSource(String dir) async {
+  final path = p.join(dir, 'corrupt_src.db');
+  await databaseFactoryFfi.deleteDatabase(path);
+  final src = await databaseFactoryFfi.openDatabase(path);
+  await src.execute('''CREATE TABLE decoded_onehz (
+    device_id TEXT NOT NULL DEFAULT '', ts_ms INTEGER NOT NULL DEFAULT 0,
+    rec_ts INTEGER, counter INTEGER NOT NULL, hr INTEGER, pad TEXT,
+    PRIMARY KEY (device_id, ts_ms))''');
+  await src.execute('''
+    WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n
+                            WHERE i < 4000)
+    INSERT INTO decoded_onehz
+    SELECT '', ($_t0 + i) * 1000, $_t0 + i, i, 60, hex(randomblob(40)) FROM n''');
+  await src.close();
+  final bytes = File(path).readAsBytesSync();
+  bytes.fillRange(4096 * 3, bytes.length, 0xA5);
+  File(path).writeAsBytesSync(bytes, flush: true);
+  return path;
+}
diff --git a/test/import_restore_perf_probe_test.dart b/test/import_restore_perf_probe_test.dart
new file mode 100644
index 000000000..e130806e8
--- /dev/null
+++ b/test/import_restore_perf_probe_test.dart
@@ -0,0 +1,113 @@
+// Optional timing probe: restores a supplied export into an empty FFI DB, prints
+// per-table wall ms, sqflite calls and rows. Desktop ffi has no platform
+// channel, so it understates Android.
+//
+//   IMPORT_PERF_PROBE=1 IMPORT_PERF_SRC=/path/export.db \
+//     flutter test test/import_restore_perf_probe_test.dart
+// The supplied file is only read; restore runs against a temporary copy.
+
+import 'dart:io';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:path/path.dart' as p;
+import 'package:shared_preferences/shared_preferences.dart';
+import 'package:sqflite_common/sqflite_logger.dart';
+import 'package:sqflite_common_ffi/sqflite_ffi.dart';
+import 'package:openstrap_edge/data/db.dart';
+
+Iterable<String> _sqlOf(Object? args) sync* {
+  if (args is Map) {
+    final sql = args['sql'];
+    if (sql is String) yield sql;
+    for (final v in args.values) {
+      if (v is List || v is Map) yield* _sqlOf(v);
+    }
+  } else if (args is List) {
+    for (final v in args) {
+      yield* _sqlOf(v);
+    }
+  }
+}
+
+void main() {
+  final srcPath = Platform.environment['IMPORT_PERF_SRC'] ?? '';
+
+  test('restore timing per table', () async {
+    final tables = LocalDb.restoreTablesForTest;
+    final patterns = {for (final t in tables) t: RegExp('\\b$t\\b')};
+    final wallUs = <String, int>{};
+    final calls = <String, int>{};
+    final clock = Stopwatch();
+    var lastUs = 0;
+    var on = false;
+    var current = '(other)';
+
+    sqfliteFfiInit();
+    // The logger is the only public hook that sees every sqflite call.
+    // ignore: experimental_member_use
+    databaseFactory = SqfliteDatabaseFactoryLogger(
+      databaseFactoryFfi,
+      options: SqfliteLoggerOptions(
+        type: SqfliteDatabaseFactoryLoggerType.invoke,
+        log: (e) {
+          if (!on || e is! SqfliteLoggerInvokeEvent) return;
+          final now = clock.elapsedMicroseconds;
+          final sql = _sqlOf(e.arguments).join('\n');
+          String? owner;
+          for (final t in tables) {
+            if (patterns[t]!.hasMatch(sql)) {
+              owner = t;
+              break;
+            }
+          }
+          // BEGIN/COMMIT and other unnamed calls belong to the table being
+          // copied: the Dart time before them is that table's row work.
+          owner ??= current;
+          current = owner;
+          wallUs[owner] = (wallUs[owner] ?? 0) + (now - lastUs);
+          calls[owner] = (calls[owner] ?? 0) + 1;
+          lastUs = now;
+        },
+      ),
+    );
+    SharedPreferences.setMockInitialValues({});
+    final tmp = await Directory.systemTemp.createTemp('openstrap_perf_');
+    LocalDb.dbName = 'openstrap_perf_probe.db';
+    final dbDir = await databaseFactory.getDatabasesPath();
+    await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName));
+    try {
+      final copy = p.join(tmp.path, 'export.db');
+      await File(srcPath).copy(copy);
+      await LocalDb.instance;
+
+      on = true;
+      clock.start();
+      final counts = await LocalDb.importFromDbFile(copy);
+      clock.stop();
+      on = false;
+
+      final rows = <String>[];
+      var totalRows = 0;
+      for (final t in [...tables, '(other)']) {
+        final n = counts[t] ?? 0;
+        if (n > 0) totalRows += n;
+        final ms = (wallUs[t] ?? 0) / 1000;
+        if (n == 0 && ms < 5) continue;
+        rows.add('${t.padRight(26)} ${ms.toStringAsFixed(0).padLeft(8)} ms '
+            '${'${calls[t] ?? 0}'.padLeft(7)} calls ${'$n'.padLeft(9)} rows');
+      }
+      // ignore: avoid_print
+      print('[import-perf] total ${clock.elapsedMilliseconds} ms, '
+          '$totalRows rows, _days=${counts['_days']}\n${rows.join('\n')}');
+    } finally {
+      on = false;
+      await LocalDb.close();
+      await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName));
+      await tmp.delete(recursive: true);
+    }
+  },
+      skip: Platform.environment['IMPORT_PERF_PROBE'] != '1' || srcPath.isEmpty
+          ? 'set IMPORT_PERF_PROBE=1 and IMPORT_PERF_SRC to measure a restore'
+          : false,
+      timeout: const Timeout(Duration(minutes: 30)));
+}
diff --git a/test/sync_chunk_overhead_test.dart b/test/sync_chunk_overhead_test.dart
new file mode 100644
index 000000000..668cd9bef
--- /dev/null
+++ b/test/sync_chunk_overhead_test.dart
@@ -0,0 +1,604 @@
+// Band-sync per-chunk overhead (design: edge.research/design-sync-throughput.md).
+//
+// Each ACK-gating commit sets synchronous=FULL, reads it back, and restores
+// NORMAL in finally. An active offload leaves the connection at NORMAL between
+// chunks. Capture freshness shares the rows + cursor transaction; the two
+// post-ACK ledger upserts share one separate transaction. Each completed chunk
+// logs its counts and commit, ACK and post-ACK timings.
+//
+// AGENTS.md invariant 1: rows + cursor commit BEFORE the verbatim token is
+// echoed; a failed commit or incomplete burst produces no ACK.
+//
+// Layout:
+//   A. LocalDb.commitSyncBatch durability and transactional freshness (2-3).
+//   B. The engine's HISTORY_END path over a real database (2g, 4-6), including
+//      a FULL/read-back/commit/NORMAL bracket for every chunk during an offload.
+//
+// The sqflite logger sees every call LocalDb makes, extending the SQL spy in
+// ack_commit_sync_full_test.dart. Calls made outside a transaction carry
+// no transactionId; that is how "one transaction" is told from "N autocommits".
+
+import 'dart:async';
+import 'dart:typed_data';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:openstrap_edge/ble/ble_engine.dart';
+import 'package:openstrap_edge/data/db.dart';
+import 'package:openstrap_edge/data/models.dart';
+import 'package:openstrap_protocol/openstrap_protocol.dart';
+import 'package:path/path.dart' as p;
+import 'package:shared_preferences/shared_preferences.dart';
+import 'package:sqflite_common/sqflite_logger.dart';
+import 'package:sqflite_common_ffi/sqflite_ffi.dart';
+
+// ── call log ────────────────────────────────────────────────────────────────
+
+/// One thing that happened, in order: either a sqflite call or a marker the
+/// test dropped in ('ack', 'commit-start', 'commit-end').
+class _Ev {
+  final String? method; // null for a marker
+  final List<String> sql;
+  final Object? txId;
+  final String mark;
+  _Ev.call(this.method, this.sql, this.txId) : mark = '';
+  _Ev.mark(this.mark) : method = null, sql = const [], txId = null;
+
+  bool touches(String table) {
+    final re = RegExp('\\b$table\\b');
+    return sql.any(re.hasMatch);
+  }
+
+  bool get isWrite => sql.any(
+        (s) => RegExp(r'^\s*(INSERT|UPDATE|DELETE|REPLACE)\b', caseSensitive: false)
+            .hasMatch(s),
+      );
+
+  bool get isSynchronousPragma =>
+      sql.any((s) => RegExp(r'pragma\s+synchronous', caseSensitive: false).hasMatch(s));
+
+  bool sqlIs(String re) =>
+      sql.any((s) => RegExp(re, caseSensitive: false).hasMatch(s));
+}
+
+final _log = <_Ev>[];
+var _counting = false;
+
+void _mark(String m) {
+  if (_counting) _log.add(_Ev.mark(m));
+}
+
+/// Every SQL string one sqflite call carries (a batch carries one per op).
+Iterable<String> _sqlOf(Object? args) sync* {
+  if (args is Map) {
+    final sql = args['sql'];
+    if (sql is String) yield sql;
+    for (final v in args.values) {
+      if (v is List || v is Map) yield* _sqlOf(v);
+    }
+  } else if (args is List) {
+    for (final v in args) {
+      yield* _sqlOf(v);
+    }
+  }
+}
+
+/// How many separate commits the writes to [table] in [events] cost: every
+/// autocommit write is one, and every transaction holding a write is one.
+/// Reads do not count; a batch outside a transaction is one call.
+int _commitsTouching(Iterable<_Ev> events, String table) {
+  var autocommit = 0;
+  final txns = <Object>{};
+  for (final e in events) {
+    if (e.method == null || !e.touches(table) || !e.isWrite) continue;
+    if (e.txId == null) {
+      autocommit++;
+    } else {
+      txns.add(e.txId!);
+    }
+  }
+  return autocommit + txns.length;
+}
+
+// ── fixtures ────────────────────────────────────────────────────────────────
+
+const _t0 = 1780000000;
+
+String _hex(int counter) => '2f18${counter.toRadixString(16).padLeft(8, '0')}';
+
+RawRecord _raw(int counter) => RawRecord(
+      counter: counter,
+      packetType: 0x2f,
+      hex: _hex(counter),
+      capturedAt: (_t0 + counter) * 1000,
+      recTs: _t0 + counter,
+    );
+
+Sample _sample(int counter) =>
+    Sample(tsEpoch: _t0 + counter, counter: counter, hr: 60);
+
+/// `n` consecutive records starting at counter `from`.
+(List<RawRecord>, List<Sample?>) _records(int from, int n) => (
+      [for (var i = 0; i < n; i++) _raw(from + i)],
+      <Sample?>[for (var i = 0; i < n; i++) _sample(from + i)],
+    );
+
+Future<void> _commit(int from, int n, {String? token}) {
+  final (raws, samples) = _records(from, n);
+  return LocalDb.commitSyncBatch(raws, samples, trimToken: token);
+}
+
+late String _dbDir;
+
+Future<Database> _freshLocal() async {
+  await LocalDb.close();
+  await databaseFactory.deleteDatabase(p.join(_dbDir, LocalDb.dbName));
+  SharedPreferences.setMockInitialValues({});
+  return LocalDb.instance;
+}
+
+/// `PRAGMA synchronous` as the connection reports it now: 2 FULL, 1 NORMAL.
+Future<int?> _syncLevel() async {
+  final db = await LocalDb.instance;
+  final rows = await db.rawQuery('PRAGMA synchronous');
+  final v = rows.isEmpty ? null : rows.first.values.firstOrNull;
+  return v is num ? v.toInt() : null;
+}
+
+Future<int> _count(String table, [String where = '1=1']) async {
+  final db = await LocalDb.instance;
+  final r = await db.rawQuery('SELECT COUNT(*) AS c FROM $table WHERE $where');
+  return (r.first['c'] as num).toInt();
+}
+
+// ── engine rig (B) ──────────────────────────────────────────────────────────
+
+int _wallNow() => DateTime.now().millisecondsSinceEpoch ~/ 1000;
+
+Uint8List _historyStart() =>
+    Uint8List.fromList(<int>[PacketType.metadata, 0x01, SyncMeta.historyStart]);
+
+Uint8List _historyEnd({required int expected, required int token}) {
+  final inner = Uint8List(24);
+  inner[0] = PacketType.metadata;
+  inner[1] = 0x02;
+  inner[2] = SyncMeta.historyEnd;
+  final v = ByteData.sublistView(inner);
+  v.setUint32(3, 1786000000, Endian.little);
+  v.setUint32(9, expected, Endian.little);
+  v.setUint32(13, token, Endian.little);
+  v.setUint32(17, 0x18, Endian.little);
+  return inner;
+}
+
+Uint8List _gen5V18Inner({required int ts, required int counter}) {
+  final inner = Uint8List(kGen5V18InnerLen);
+  final v = ByteData.sublistView(inner);
+  inner[0] = PacketType.historicalData;
+  inner[1] = 18;
+  inner[2] = 0x80;
+  v.setUint32(3, counter, Endian.little);
+  v.setUint32(7, ts, Endian.little);
+  inner[14] = 64;
+  v.setFloat32(33, 0.5, Endian.little);
+  v.setFloat32(45, 1.0, Endian.little);
+  return inner;
+}
+
+/// The same fake gen5 link history_task_safety_test.dart uses, but its commit
+/// sink is the REAL `LocalDb.commitSyncBatch`, so the call log sees the
+/// commit, the ACK write and the post-ACK ledger writes in one timeline.
+class _Rig {
+  final logs = <String>[];
+  final acks = <List<int>>[]; // body of each success result, in order
+  late final BleEngine engine;
+
+  _Rig() {
+    _rigs.add(this);
+    engine = BleEngine(onRecord: (_, _) async {}, onState: (_) {}, log: logs.add);
+    engine.debugInstallFakeLink(
+      band: BandProfile.gen5,
+      onCommit: (raws, samples, token,
+          {archives, ecgRawPackets, deviceFamily}) async {
+        _mark('commit-start');
+        await LocalDb.commitSyncBatch(
+          raws,
+          samples,
+          trimToken: token,
+          archives: archives,
+          ecgRawPackets: ecgRawPackets,
+          deviceFamily: deviceFamily,
+        );
+        _mark('commit-end');
+      },
+      onWrite: (f) async {
+        final parsed = parseFrame(f, profile: BandProfile.gen5);
+        if (parsed == null || !parsed.valid) return true;
+        final opcode = parsed.inner[2];
+        final body = parsed.inner.sublist(3);
+        if (opcode == Cmd.getClock) {
+          final seq = parsed.inner[1];
+          scheduleMicrotask(() => engine.debugAbsorbDecoded(Decoded(
+                'cmd_response',
+                {
+                  'opcode': Cmd.getClock,
+                  'req_seq': seq,
+                  'cmd_status': 1,
+                  'clock_epoch': _wallNow(),
+                },
+              )));
+        }
+        if (opcode == Cmd.historicalDataResult &&
+            body.isNotEmpty &&
+            body[0] == 0x01) {
+          acks.add(body);
+          _mark('ack');
+        }
+        return true;
+      },
+    );
+  }
+
+  void rx(Uint8List inner) =>
+      engine.debugReceiveFrame(Frame(inner, true, true), role: 'data');
+
+  /// One full burst: [n] records, then the marker for [token].
+  void chunk(int n, int token, {required int ts, required int counter}) {
+    for (var i = 0; i < n; i++) {
+      rx(_gen5V18Inner(ts: ts + i, counter: counter + i));
+    }
+    rx(_historyEnd(expected: n, token: token));
+  }
+
+  /// Wait for the real receive queue, then for the logs to stop growing.
+  /// A quiet log alone could mean an ffi transaction is still in flight.
+  Future<void> settle() async {
+    var last = -1;
+    var stable = 0;
+    for (var i = 0; i < 200 && stable < 4; i++) {
+      await pumpEventQueue();
+      await Future<void>.delayed(const Duration(milliseconds: 25));
+      final now = _log.length + logs.length;
+      final snapshot = engine.offloadSnapshot;
+      final idle = snapshot['queue_draining'] == false &&
+          snapshot['queued_frames'] == 0;
+      stable = idle && now == last ? stable + 1 : 0;
+      last = now;
+    }
+    expect(stable, 4, reason: 'the real receive queue must finish before assertions');
+  }
+}
+
+final _rigs = <_Rig>[];
+
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  setUpAll(() async {
+    sqfliteFfiInit();
+    // The logger is the only public hook that sees every sqflite call.
+    // ignore: experimental_member_use
+    databaseFactory = SqfliteDatabaseFactoryLogger(
+      databaseFactoryFfi,
+      options: SqfliteLoggerOptions(
+        type: SqfliteDatabaseFactoryLoggerType.invoke,
+        log: (e) {
+          if (_counting && e is SqfliteLoggerInvokeEvent) {
+            final args = e.arguments;
+            _log.add(_Ev.call(
+              e.method,
+              _sqlOf(args).toList(),
+              args is Map ? args['transactionId'] : null,
+            ));
+          }
+        },
+      ),
+    );
+    SharedPreferences.setMockInitialValues({});
+    LocalDb.dbName = 'openstrap_sync_chunk_overhead_test.db';
+    _dbDir = await databaseFactory.getDatabasesPath();
+  });
+
+  tearDownAll(() async {
+    _counting = false;
+    await LocalDb.close();
+    await databaseFactory.deleteDatabase(p.join(_dbDir, LocalDb.dbName));
+  });
+
+  setUp(() async {
+    _counting = false;
+    _log.clear();
+    await _freshLocal();
+  });
+
+  tearDown(() async {
+    _counting = false;
+    for (final r in _rigs) {
+      await r.engine.disconnect();
+      await r.settle();
+    }
+    _rigs.clear();
+  });
+
+  group('A. commitSyncBatch round trips and durability', () {
+    // 2. Durability is bracketed per commit, including failures.
+    test('2b. a throwing commit restores NORMAL (1)', () async {
+      final db = await LocalDb.instance;
+      await db.execute('CREATE TRIGGER t_cursor_boom BEFORE INSERT ON sync_cursor '
+          "WHEN NEW.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'boom'); END");
+      _counting = true;
+      await expectLater(_commit(1, 50, token: 't'), throwsA(anything));
+      _counting = false;
+
+      final full = _log.indexWhere(
+          (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*FULL'));
+      final verify = _log.indexWhere(
+          (e) => e.sqlIs(r'^\s*pragma\s+synchronous\s*;?\s*$'));
+      final begin = _log.indexWhere((e) => e.sqlIs(r'^\s*BEGIN'));
+      final rollback = _log.indexWhere((e) => e.sqlIs(r'^\s*ROLLBACK'));
+      final normal = _log.indexWhere(
+          (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*NORMAL'));
+      expect(full, isNonNegative);
+      expect(verify, greaterThan(full));
+      expect(begin, greaterThan(verify));
+      expect(rollback, greaterThan(begin));
+      expect(normal, greaterThan(rollback),
+          reason: 'finally restores NORMAL after a failed transaction');
+      expect(await _syncLevel(), 1);
+      expect(await _count('samples'), 0);
+    });
+
+    test('2d. a commit runs at FULL and restores NORMAL after COMMIT', () async {
+      _counting = true;
+      await _commit(1, 50, token: 't');
+      _counting = false;
+
+      int idx(bool Function(_Ev) f) => _log.indexWhere(f);
+      final full = idx((e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*FULL'));
+      final verify = idx(
+          (e) => e.sqlIs(r'^\s*pragma\s+synchronous\s*;?\s*$'));
+      final begin = idx((e) => e.sqlIs(r'^\s*BEGIN'));
+      final commit = idx((e) => e.sqlIs(r'^\s*COMMIT'));
+      final normal = idx((e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*NORMAL'));
+      expect(full, isNonNegative);
+      expect(verify, greaterThan(full));
+      expect(begin, greaterThan(verify),
+          reason: 'FULL is set and read back BEFORE the txn opens');
+      expect(commit, greaterThan(begin));
+      expect(normal, greaterThan(commit), reason: 'NORMAL restored after COMMIT');
+      expect(await _syncLevel(), 1);
+    });
+
+    // 3. Freshness rides in the batch transaction.
+    test('3a. a failing freshness write rolls the whole batch back '
+        '(freshness is inside the transaction)', () async {
+      final db = await LocalDb.instance;
+      await db.execute(
+          'CREATE TRIGGER t_fresh_boom BEFORE INSERT ON compute_freshness '
+          "WHEN NEW.\"key\" = 'capture' BEGIN SELECT RAISE(ABORT, 'boom'); END");
+
+      await expectLater(_commit(1, 50, token: 't'), throwsA(anything));
+
+      expect(await _count('samples'), 0,
+          reason: 'the rows must roll back with the freshness row; today the '
+              'batch is already committed when freshness fails');
+      expect(await _count('sync_cursor', "name = 'counter_hw'"), 0,
+          reason: 'and so must the cursor');
+      expect(await _count('compute_freshness', "\"key\" = 'capture'"), 0);
+    });
+
+    test('3b. a failing cursor write leaves neither rows nor freshness '
+        '(guard: holds today, must stay)', () async {
+      final db = await LocalDb.instance;
+      await db.execute('CREATE TRIGGER t_cursor_boom BEFORE INSERT ON sync_cursor '
+          "WHEN NEW.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'boom'); END");
+
+      await expectLater(_commit(1, 50, token: 't'), throwsA(anything));
+
+      expect(await _count('samples'), 0);
+      expect(await _count('sync_cursor', "name = 'counter_hw'"), 0);
+      expect(await _count('compute_freshness', "\"key\" = 'capture'"), 0);
+    });
+
+    test('3c. the capture-freshness select and write happen inside the batch '
+        'transaction, not after it', () async {
+      _counting = true;
+      await _commit(1, 50, token: 't');
+      _counting = false;
+
+      final fresh = _log.where((e) => e.touches('compute_freshness')).toList();
+      expect(fresh, isNotEmpty, reason: 'harness check: freshness is written');
+      final outside = fresh.where((e) => e.txId == null).length;
+      expect(outside, 0,
+          reason: '$outside freshness call(s) ran outside the commit '
+              'transaction (today: a select and an insert, 2 extra round trips)');
+      final rows = _log.singleWhere((e) => e.touches('samples') && e.isWrite);
+      final cursor = _log.where((e) => e.touches('sync_cursor') && e.isWrite);
+      expect(rows.txId, isNotNull);
+      expect(cursor, isNotEmpty);
+      expect([...fresh, ...cursor].every((e) => e.txId == rows.txId), isTrue,
+          reason: 'freshness, decoded rows and trim cursor share the SAME txn');
+    });
+
+    test('3d. the freshness row still carries the batch max record time '
+        '(guard: holds today, must stay)', () async {
+      await _commit(1, 50, token: 't');
+      final row = await LocalDb.computeFreshness('capture');
+      expect(row, isNotNull);
+      expect(row!['payload_json'] as String,
+          contains('"latest_raw_rec_ts":${_t0 + 50}'));
+    });
+  });
+
+  group('B. HISTORY_END over a real database', () {
+    final ts = _wallNow() - 3600;
+
+    test('2g. EVERY commit during an offload has its own FULL bracket and '
+        'leaves NORMAL between chunks', () async {
+      final r = _Rig();
+      r.engine.debugClaimHistoryTask();
+      expect(await _syncLevel(), 1,
+          reason: 'starting an offload must leave the connection at NORMAL');
+      for (var i = 0; i < 3; i++) {
+        _log.clear();
+        _counting = true;
+        r.rx(_historyStart());
+        r.chunk(3, 0xB200 + i, ts: ts + i * 100, counter: 2000 + i * 100);
+        await r.settle();
+        _counting = false;
+
+        expect(r.acks, hasLength(i + 1));
+        expect(r.engine.offloadActive, isTrue,
+            reason: 'the same offload remains active between chunks');
+        expect(_log.where((e) => e.isSynchronousPragma), hasLength(3),
+            reason: 'chunk $i sets FULL, reads it back, then restores NORMAL');
+        final full = _log.indexWhere(
+            (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*FULL'));
+        final verify = _log.indexWhere(
+            (e) => e.sqlIs(r'^\s*pragma\s+synchronous\s*;?\s*$'));
+        final begin = _log.indexWhere((e) => e.sqlIs(r'^\s*BEGIN'));
+        final commit = _log.indexWhere((e) => e.sqlIs(r'^\s*COMMIT'));
+        final normal = _log.indexWhere(
+            (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*NORMAL'));
+        final ack = _log.indexWhere((e) => e.mark == 'ack');
+        expect(full, isNonNegative);
+        expect(verify, greaterThan(full));
+        expect(begin, greaterThan(verify));
+        expect(commit, greaterThan(begin));
+        expect(normal, greaterThan(commit));
+        expect(ack, greaterThan(normal));
+        expect(await _syncLevel(), 1,
+            reason: 'chunk $i must leave NORMAL for post-ACK writes and idle time');
+      }
+    });
+
+    // 4. Ledger: one transaction per marker.
+    test('4. the two post-ACK ledger upserts cost one commit, and both rows '
+        'still land', () async {
+      final r = _Rig();
+      r.engine.debugClaimHistoryTask();
+      await LocalDb.instance; // opened before counting: opening is not the cost
+      _counting = true;
+      r.rx(_historyStart());
+      r.chunk(3, 0xB400, ts: ts, counter: 4000);
+      await r.settle();
+      _counting = false;
+
+      expect(r.acks, hasLength(1), reason: 'harness check: the chunk was ACKed');
+      final ackAt = _log.indexWhere((e) => e.mark == 'ack');
+      expect(ackAt, isNonNegative);
+      final after = _log.sublist(ackAt + 1);
+      final commits = _commitsTouching(after, 'sync_ledger');
+      expect(commits, 1,
+          reason: 'one marker costs one ledger commit after the ACK; '
+              'today it is $commits (two separate upserts)');
+
+      // Same content as before: the 'capture' row and the per-token row.
+      final ledger = await LocalDb.syncLedger();
+      final capture = ledger.where((e) => e['chunk_id'] == 'capture').toList();
+      expect(capture, hasLength(1));
+      expect(capture.single['status'], 'acknowledged');
+      expect(capture.single['meta_json'] as String, contains('last_batch_token'));
+      final tok = ledger.where((e) => (e['chunk_id'] as String).startsWith('batch:'));
+      expect(tok, hasLength(1));
+      expect(tok.single['status'], 'acked');
+      expect(tok.single['meta_json'] as String, contains('"chunk_rows":3'));
+    });
+
+    // 5. Perf line.
+    test('5. each HISTORY_END commit logs exactly one [perf] sync chunk line '
+        'with per-chunk counts and timings', () async {
+      final r = _Rig();
+      r.engine.debugClaimHistoryTask();
+      r.rx(_historyStart());
+      r.chunk(3, 0xB500, ts: ts, counter: 5000);
+      await r.settle();
+      r.rx(_historyStart());
+      r.chunk(2, 0xB501, ts: ts + 100, counter: 5100);
+      await r.settle();
+
+      expect(r.acks, hasLength(2), reason: 'harness check: both chunks ACKed');
+      final lines = r.logs.where((l) => l.contains('[perf] sync chunk')).toList();
+      expect(lines, hasLength(2),
+          reason: 'one line per HISTORY_END, no more, no fewer; saw $lines');
+
+      final re = RegExp(r'\brecords=(\d+)\b.*\bbeats=(\d+)\b.*\bcommit_ms=(\d+)\b'
+          r'.*\back_ms=(\d+)\b.*\bpost_ms=(\d+)\b');
+      for (final l in lines) {
+        expect(re.hasMatch(l), isTrue,
+            reason: 'line must carry records= beats= commit_ms= ack_ms= '
+                'post_ms= in that order: $l');
+      }
+      // Per chunk, not cumulative for the connection.
+      expect(re.firstMatch(lines[0])!.group(1), '3');
+      expect(re.firstMatch(lines[1])!.group(1), '2');
+    });
+
+    // 6. Invariant guards (must pass today and stay green).
+    test('6a. the ACK is written only after the commit completed, and echoes '
+        'the verbatim token (guard)', () async {
+      final r = _Rig();
+      r.engine.debugClaimHistoryTask();
+      _counting = true;
+      r.rx(_historyStart());
+      r.chunk(3, 0xB600, ts: ts, counter: 6000);
+      await r.settle();
+      _counting = false;
+
+      final commitEnd = _log.indexWhere((e) => e.mark == 'commit-end');
+      final txnCommit = _log.indexWhere((e) => e.sqlIs(r'^\s*COMMIT'));
+      final ack = _log.indexWhere((e) => e.mark == 'ack');
+      expect(commitEnd, isNonNegative);
+      expect(txnCommit, isNonNegative);
+      expect(ack, greaterThan(commitEnd),
+          reason: 'ACK must follow commitSyncBatch returning');
+      expect(ack, greaterThan(txnCommit));
+      expect(r.acks.single.take(9).toList(), [
+        0x01,
+        0x00, 0xB6, 0x00, 0x00, // token u32 LE
+        0x18, 0x00, 0x00, 0x00, // batch id u32 LE
+      ]);
+    });
+
+    test('6b. a commit that throws produces no ACK and no acked ledger row '
+        '(guard)', () async {
+      final db = await LocalDb.instance;
+      await db.execute('CREATE TRIGGER t_cursor_boom BEFORE INSERT ON sync_cursor '
+          "WHEN NEW.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'boom'); END");
+      final r = _Rig();
+      r.engine.debugClaimHistoryTask();
+      r.rx(_historyStart());
+      r.chunk(3, 0xB601, ts: ts, counter: 6100);
+      await r.settle();
+
+      expect(r.acks, isEmpty, reason: 'no ACK for a chunk that rolled back');
+      expect(await _count('samples'), 0, reason: 'harness check: it rolled back');
+      final ledger = await LocalDb.syncLedger();
+      expect(
+          ledger.where((e) => e['status'] == 'acked' || e['status'] == 'acknowledged'),
+          isEmpty);
+      expect(r.logs.where((l) => l.contains('[perf] sync chunk')), isEmpty,
+          reason: 'a failed commit is not a completed chunk');
+    });
+
+    test('6c. an incomplete burst is neither committed nor ACKed (guard)',
+        () async {
+      final r = _Rig();
+      r.engine.debugClaimHistoryTask();
+      _counting = true;
+      r.rx(_historyStart());
+      r.rx(_gen5V18Inner(ts: ts, counter: 6200));
+      r.rx(_historyEnd(expected: 3, token: 0xB602));
+      await r.settle();
+      _counting = false;
+
+      expect(r.acks, isEmpty, reason: 'a partial chunk must never trim flash');
+      // Upstream's count gate stores what arrived without the trim token
+      // (_refuseHistoryEndOnShortCount), so only the cursor and ACK are pinned.
+      expect(await LocalDb.getCursor('strap_trim'), isNull);
+      final ledger = await LocalDb.syncLedger();
+      expect(ledger.where(
+          (e) => e['status'] == 'acked' || e['status'] == 'acknowledged'), isEmpty);
+    });
+
+  });
+}