fix: enforce telemetry durability before ride finish

lib/features/rides/ride_state_machine.dart

@@ -145,6 +145,29 @@ class RideStateMachine extends StateNotifier<RideSession> {
   // persistence await is outstanding would re-open the shared-buffer race.
   bool _isFlushingTelemetryBatch = false;
 
+  // RID-3B-1: coordination primitive exposing the in-flight telemetry
+  // persistence operation, so an explicit Stop can await the current owner
+  // instead of observing a no-op flush. This is NOT a second buffer and NOT a
+  // serialization framework: it publishes exactly one Future for exactly one
+  // owned snapshot and is cleared when that snapshot's attempt settles.
+  Future<void>? _activeTelemetryFlush;
+
+  // RID-3B-1: late private telemetry-acceptance cutoff (invariant I3). Owned
+  // solely by the in-flight finish operation and reopened in its finally path,
+  // so a failed-but-retryable ride keeps recording and the next ride is
+  // unaffected. Strictly private: no public RideState or RideSession change.
+  bool _telemetryAcceptanceClosed = false;
+
+  // RID-3B-1: bound on the convergent final drain. Telemetry arrives at 20 Hz
+  // (one sample per 50 ms) while each pass awaits a single db.batch round-trip,
+  // so the samples that can accumulate during a pass are normally 0 or 1 and
+  // convergence is reached in 1-2 passes. Four passes tolerate a persistence
+  // latency of roughly 150 ms per pass while still bounding finalization to at
+  // most four batch writes. The bound never fabricates durability: if telemetry
+  // is still outstanding when it is exhausted, finalization FAILS and remains
+  // retryable rather than transitioning to finished.
+  static const int _maxFinalDrainPasses = 4;
+
   RideStateMachine(this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref) : super(RideSession()) {
     _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate);
     
@@ -375,6 +398,13 @@ class RideStateMachine extends StateNotifier<RideSession> {
       await _performFinishRide();
     } finally {
       _isFinishing = false;
+      // RID-3B-1: the telemetry-acceptance cutoff is owned solely by this
+      // in-flight finish operation, so it is reopened here. A failed but
+      // retryable ride therefore keeps recording, and the next ride is
+      // unaffected. Deliberately NOT cleared from reset(): source analysis shows
+      // reset() has no callers, and clearing finish-owned state elsewhere would
+      // weaken the barrier.
+      _telemetryAcceptanceClosed = false;
     }
   }
 
@@ -393,7 +423,11 @@ class RideStateMachine extends StateNotifier<RideSession> {
         );
       } catch (_) {}
 
-      await _flushBatch(); // write remaining points
+      // RID-3B-1: the explicit-Stop durability barrier replaces the single
+      // tolerant flush. It awaits any in-flight periodic flush owner, drains to
+      // convergence, closes telemetry acceptance, and performs a final strict
+      // drain - throwing if durability cannot be established (invariant I2).
+      await _establishTelemetryDurabilityForFinish();
       
       await db.endRide(
         state.dbId!,
@@ -451,14 +485,19 @@ class RideStateMachine extends StateNotifier<RideSession> {
       }
     } catch (_) {}
 
-    state = state.copyWith(
-      state: RideState.finished,
-      healthStatus: RecordingHealthStatus.recordingComplete,
-    );
     try {
       WakelockPlus.disable().catchError((_) {});
     } catch (_) {}
     audioAlertService.speak('Session ended');
+
+    // RID-3B-1: the sync-visible finished edge is the LAST side effect of
+    // finalization. It is emitted only after telemetry durability, endTime,
+    // personal records and the forensic summary are complete, so SyncObserver
+    // can keep observing only this edge (invariant I1).
+    state = state.copyWith(
+      state: RideState.finished,
+      healthStatus: RecordingHealthStatus.recordingComplete,
+    );
   }
 
   void reset() {
@@ -480,7 +519,17 @@ class RideStateMachine extends StateNotifier<RideSession> {
 
     if (state.state != RideState.riding) return;
 
-    if (state.healthStatus != RecordingHealthStatus.recordingActive) {
+    // RID-3B-1 (invariant I3): telemetry accepted after the finalization
+    // boundary is never attributed to the finishing ride. The gate is private
+    // and is reopened by finishRide()'s finally path, so a retryable ride and
+    // every subsequent ride continue to record normally.
+    if (_telemetryAcceptanceClosed) return;
+
+    // RID-3B-1: incoming telemetry must not clobber recordingFinalizing back to
+    // recordingActive while finalization is in progress. Minimum private
+    // condition: the existing RID-2 in-flight finish flag.
+    if (!_isFinishing &&
+        state.healthStatus != RecordingHealthStatus.recordingActive) {
       state = state.copyWith(healthStatus: RecordingHealthStatus.recordingActive);
     }
 
@@ -534,8 +583,143 @@ class RideStateMachine extends StateNotifier<RideSession> {
     state = state.copyWith(totalSamplesReceived: state.totalSamplesReceived + 1);
   }
   
+  /// RID-3A ownership semantics, plus RID-3B-1 active-flush publication.
+  ///
+  /// Takes exclusive ownership of the current buffer contents, removes them from
+  /// the live buffer BEFORE awaiting persistence, and restores them to the FRONT
+  /// on failure so chronological identity order survives. Returns true only when
+  /// the owned snapshot is durable (or there was nothing to own); it never
+  /// reports success for a failed persistence.
+  ///
+  /// Tolerant by design for the periodic path: a persistence failure is
+  /// retained, restored and logged with the existing bounded warning, never
+  /// thrown into the timer callback.
+  /// RID-3B-1: finish-only strict drain.
+  ///
+  /// Awaits any active telemetry flush owner first - looping, because a periodic
+  /// tick may take ownership again while waiting - then takes ownership under
+  /// RID-3A's guard and persists. Returns false whenever durability could not be
+  /// established, so finishRide can distinguish success from failure. It never
+  /// reports success for a failed persistence and never silently swallows a
+  /// durability failure. No automatic retry worker is added.
+  Future<bool> _drainTelemetryBatchStrictly() async {
+    while (_isFlushingTelemetryBatch) {
+      final active = _activeTelemetryFlush;
+      if (active == null) break;
+      try {
+        await active;
+      } catch (_) {
+        // The owner retains, restores and logs its own failure; ownership is
+        // re-checked by the loop condition.
+      }
+    }
+    if (_telemetryBatch.isEmpty) return true;
+    // Another owner is still active and could not be awaited, so durability is
+    // not established and must not be assumed.
+    if (_isFlushingTelemetryBatch) return false;
+    return _persistOwnedTelemetrySnapshot();
+  }
+
+  /// RID-3B-1: the explicit-Stop durability barrier (invariants I1-I3).
+  ///
+  /// Awaits any in-flight periodic flush, drains the buffer to convergence within
+  /// a bounded policy while telemetry capture continues, closes acceptance, then
+  /// performs one final strict drain for anything queued between the last
+  /// convergence check and the cutoff assignment.
+  ///
+  /// Throws if durability cannot be established, so the caller aborts
+  /// finalization without writing endTime, evaluating personal records,
+  /// generating a forensic summary, announcing completion, or emitting the
+  /// sync-visible finished edge.
+  Future<void> _establishTelemetryDurabilityForFinish() async {
+    var durable = true;
+    var passes = 0;
+
+    // Bounded convergent drain. Telemetry keeps being accepted during this phase
+    // so existing capture semantics are preserved; the final few samples are
+    // never intentionally dropped to simplify finalization.
+    while (passes < _maxFinalDrainPasses &&
+        (_telemetryBatch.isNotEmpty || _activeTelemetryFlush != null)) {
+      passes++;
+      if (!await _drainTelemetryBatchStrictly()) {
+        durable = false;
+        break;
+      }
+    }
+
+    // Convergence must actually have been reached. The cap never fabricates
+    // durability: telemetry still outstanding means finalization fails.
+    if (durable &&
+        (_telemetryBatch.isNotEmpty || _activeTelemetryFlush != null)) {
+      durable = false;
+    }
+
+    if (durable) {
+      // I3: close acceptance, then one final strict drain.
+      _telemetryAcceptanceClosed = true;
+      durable = await _drainTelemetryBatchStrictly();
+    }
+
+    if (!durable) {
+      // I2: restore a truthful recoverable recording status and propagate, so
+      // RID-2's existing UI boundary containment handles the rejection and the
+      // ride remains retryable. RID-3A has already restored any failed owned
+      // snapshot to the front of the buffer in correct order.
+      state = state.copyWith(healthStatus: RecordingHealthStatus.recordingActive);
+      throw StateError(
+        'Ride finalization aborted: telemetry durability could not be established',
+      );
+    }
+  }
+
+  Future<bool> _persistOwnedTelemetrySnapshot() async {
+    if (_isFlushingTelemetryBatch || _telemetryBatch.isEmpty) return true;
+
+    _isFlushingTelemetryBatch = true;
+    final batchToInsert = List<TelemetrySamplesCompanion>.from(_telemetryBatch);
+    _telemetryBatch.clear();
+    // Publish this exact in-flight operation so an explicit Stop can await the
+    // current owner instead of observing a no-op flush. One Future for one owned
+    // snapshot - a coordination primitive, not a second buffering system.
+    final completion = Completer<void>();
+    _activeTelemetryFlush = completion.future;
+    try {
+      await db.batch((batch) {
+        batch.insertAll(db.telemetrySamples, batchToInsert);
+      });
+      // Success: account only for the owned snapshot that is now durable.
+      state = state.copyWith(totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length);
+      return true;
+    } catch (e) {
+      // Persistence failed, so the owned snapshot is not durable. Restore it to
+      // the FRONT of the live buffer, preserving both the snapshot's internal
+      // order and the order of samples that arrived during the await, so a later
+      // legitimate flush can retry them. No automatic retry worker is added.
+      _telemetryBatch.insertAll(0, batchToInsert);
+      // Keep in queue for retry on next flush interval
+      try {
+        ref.read(fieldTestLoggerProvider).log(
+          eventType: FieldTestEventType.databaseQueueWarning,
+          message: 'Database batch write retry: $e',
+          severity: EventSeverity.warning,
+        );
+      } catch (_) {}
+      return false;
+    } finally {
+      _isFlushingTelemetryBatch = false;
+      if (!completion.isCompleted) completion.complete();
+      _activeTelemetryFlush = null;
+    }
+  }
+
   Future<void> _flushBatch() async {
-    // RID-3A: telemetry batch persistence with exclusive snapshot ownership.
+    // RID-3A telemetry batch ownership (exclusive snapshot, clear-before-await,
+    // front-restoration on failure, success-only counter) now lives in
+    // _persistOwnedTelemetrySnapshot(), unchanged in behaviour. RID-3B-1 also
+    // publishes the in-flight operation there so an explicit Stop can await the
+    // current owner instead of observing a no-op flush. The periodic path stays
+    // tolerant and never throws into the timer callback; timer Future ownership
+    // and teardown safety are RID-3B-2 scope and are deliberately unchanged.
     //
     // The guard is raised synchronously, before the first `await`, so a
     // competing legitimate flush (the 3 s periodic timer against the flush
@@ -552,35 +736,7 @@ class RideStateMachine extends StateNotifier<RideSession> {
     //
     // The incremental ride-stats persistence below is deliberately left outside
     // this guard so its existing every-tick cadence is unchanged.
-    if (!_isFlushingTelemetryBatch && _telemetryBatch.isNotEmpty) {
+    await _persistOwnedTelemetrySnapshot();
-      _isFlushingTelemetryBatch = true;
-      final batchToInsert = List<TelemetrySamplesCompanion>.from(_telemetryBatch);
-      _telemetryBatch.clear();
-      try {
-        await db.batch((batch) {
-          batch.insertAll(db.telemetrySamples, batchToInsert);
-        });
-        // Success: account only for the owned snapshot that is now durable.
-        state = state.copyWith(totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length);
-      } catch (e) {
-        // Persistence failed, so the owned snapshot is not durable. Restore it
-        // to the FRONT of the live buffer, preserving both the snapshot's
-        // internal order and the order of samples that arrived during the await,
-        // so chronological identity order survives and a later legitimate flush
-        // can retry them. No automatic retry worker is introduced.
-        _telemetryBatch.insertAll(0, batchToInsert);
-        // Keep in queue for retry on next flush interval
-        try {
-          ref.read(fieldTestLoggerProvider).log(
-            eventType: FieldTestEventType.databaseQueueWarning,
-            message: 'Database batch write retry: $e',
-            severity: EventSeverity.warning,
-          );
-        } catch (_) {}
-      } finally {
-        _isFlushingTelemetryBatch = false;
-      }
-    }
 
     // Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats
     if (state.dbId != null && state.state == RideState.riding) {

test/rid3a_telemetry_batch_ownership_test.dart

@@ -321,32 +321,48 @@ void main() {
       await pumpEventQueue(times: 40);
       expect(container.read(rideStateMachineProvider).totalSamplesReceived, 30);
 
-      // Competing legitimate flush: finishRide() awaits _flushBatch(). It must
+      // Competing legitimate flush: finishRide() now awaits the active owner
-      // take no ownership of the samples A already owns, and must not touch the
+      // (RID-3B-1) instead of observing a no-op. While flush A is still parked it
-      // newer ones either.
+      // must take no ownership of A's snapshot and issue no batch of its own; it
+      // drains the newer samples only after A has committed.
       final finishing = notifier.finishRide();
       await pumpEventQueue(times: 30);
       expect(db.batchInvocations, 1,
-          reason: 'a competing flush while the guard is raised must take no '
+          reason: 'while the owner is parked, no second flush may take ownership');
-              'ownership of the live buffer');
 
-      // Release A: it persists exactly its owned snapshot (0..9) and nothing
+      // Release ONLY flush A, leaving the gate armed so the RID-3B-1 durability
-      // else is removed from the live buffer.
+      // barrier's own drain parks as the next gate. That yields a deterministic
-      db.gateBatch = false;
+      // observation point at which flush A's committed write is isolated: A has
+      // persisted and nothing else has been written yet.
       flushA.complete();
-      await finishing;
       await pumpEventQueue(times: 40);
 
-      final rowsAfterA = await persistedRows();
+      final rowsOwnedByA = await persistedRows();
-      final countsAfterA = countsByIndex(rowsAfterA);
+      final countsOwnedByA = countsByIndex(rowsOwnedByA);
-      expect(duplicatedIn(countsAfterA), isEmpty);
+      expect(duplicatedIn(countsOwnedByA), isEmpty,
-      expect(missingOf(10, countsAfterA), isEmpty,
+          reason: 'no duplication from overlapping ownership');
-          reason: 'A persisted its owned snapshot');
+      expect(missingOf(10, countsOwnedByA), isEmpty,
-      expect(countsAfterA.length, 10,
+          reason: 'A persisted its whole owned snapshot');
-          reason: 'A must not have persisted samples it did not own');
+      // The real RID-3A ownership property: flush A's write contains exactly its
+      // own identity set, and none of the newer samples - those remain buffered
+      // for later ownership. This is observed BEFORE finishRide can complete the
+      // durability barrier, not after `await finishing`, because RID-3B-1 now
+      // intentionally drains all accepted pre-boundary telemetry before
+      // finalizing. The previous "only 10 rows after await finishing" expectation
+      // depended on the finish flush being a no-op, which RID-3B-1 removed.
+      expect(countsOwnedByA.keys.toList()..sort(),
+          [for (var i = 0; i < 10; i++) i],
+          reason: 'flush A owns exactly indices 0-9');
+      expect(countsOwnedByA.keys.where((i) => i >= 10).toList(), isEmpty,
+          reason: 'no newer sample was included in flush A write');
+      expect(db.batchGates, hasLength(2),
+          reason: 'the barrier now owns the newer batch and is parked');
 
-      // Legitimate later flush (the periodic timer) persists 10..29.
+      // Release the barrier's drain and let finishRide complete. It persists
-      await Future<void>.delayed(tickWait);
+      // 10..29 itself, so no later periodic flush is required.
+      db.gateBatch = false;
+      db.batchGates.last.complete();
+      await finishing;
       await pumpEventQueue(times: 40);
 
       final rows = await persistedRows();

test/rid3b1_stop_durability_barrier_test.dart

@@ -0,0 +1,731 @@
+// RID-3B-1 regression coverage: explicit-Stop telemetry durability barrier.
+//
+// Invariants under test:
+//   I1 - finished is emitted only after every sample accepted before the
+//        finalization boundary is durably persisted exactly once.
+//   I2 - if durability cannot be established, finalization fails: no endTime,
+//        no records, no summary, no finished edge, retry remains possible.
+//   I3 - telemetry accepted after the boundary is not attributed to the ride.
+//   I4 - these apply to explicit Stop only (dispose/process death are RID-3B-2).
+//
+// Ordering is MEASURED, not inferred: the db.batch / endRide / records /
+// stopSession overrides and the Riverpod finished-edge listener all append to one
+// shared ordered event log. Race windows use Completer gates, never sleeps.
+import 'dart:async';
+
+import 'package:flutter/foundation.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:flutter_riverpod/flutter_riverpod.dart';
+import 'package:geolocator/geolocator.dart';
+import 'package:shared_preferences/shared_preferences.dart';
+import 'package:drift/native.dart';
+import 'package:drift/drift.dart' as drift;
+import 'package:motorcycle_dash/database/database.dart'
+    hide TelemetrySample, FieldTestSummary, FieldTestEvent;
+import 'package:motorcycle_dash/database/database.dart' as dbgen
+    show TelemetrySample;
+import 'package:motorcycle_dash/models/telemetry_sample.dart';
+import 'package:motorcycle_dash/models/field_test_summary.dart';
+import 'package:motorcycle_dash/models/recording_health_status.dart';
+import 'package:motorcycle_dash/services/telemetry_service.dart';
+import 'package:motorcycle_dash/services/location_service.dart';
+import 'package:motorcycle_dash/services/sensor_fusion_service.dart';
+import 'package:motorcycle_dash/services/compass_service.dart';
+import 'package:motorcycle_dash/services/audio_alert_service.dart';
+import 'package:motorcycle_dash/services/field_test/field_test_logger.dart';
+import 'package:motorcycle_dash/services/validation/telemetry_validator.dart';
+import 'package:motorcycle_dash/features/rides/ride_state_machine.dart';
+import 'package:motorcycle_dash/features/records/records_service.dart';
+
+typedef TelemetryRow = dbgen.TelemetrySample;
+
+class _Rid3b1AppDatabase extends AppDatabase {
+  _Rid3b1AppDatabase(this.eventLog) : super.forTesting(NativeDatabase.memory());
+
+  final List<String> eventLog;
+  int batchInvocations = 0;
+  int endRideInvocations = 0;
+  bool gateBatch = false;
+  final List<Completer<void>> batchGates = [];
+  int batchFailuresRemaining = 0;
+
+  @override
+  Future<void> batch(
+      FutureOr<void> Function(drift.Batch batch) runInBatch) async {
+    final index = batchInvocations++;
+    if (gateBatch) {
+      final gate = Completer<void>();
+      batchGates.add(gate);
+      eventLog.add('batch$index:parked');
+      await gate.future;
+    }
+    if (batchFailuresRemaining > 0) {
+      batchFailuresRemaining--;
+      eventLog.add('batch$index:FAILED');
+      throw StateError('RID-3B-1 simulated telemetry persistence failure');
+    }
+    await super.batch(runInBatch);
+    eventLog.add('batch$index:committed');
+  }
+
+  @override
+  Future<void> endRide(int rideId, RidesCompanion data) async {
+    endRideInvocations++;
+    eventLog.add('endRide');
+    return super.endRide(rideId, data);
+  }
+}
+
+/// Neutralises FieldTestLogger's own 3 s periodic flush() -> _db.batch
+/// (field_test_logger.dart:138, :592) so db.batch accounting refers only to
+/// telemetry persistence. FieldTestLogger internals are NOT modified.
+class _Rid3b1FieldTestLogger extends FieldTestLogger {
+  _Rid3b1FieldTestLogger(super.db, this.eventLog);
+
+  final List<String> eventLog;
+  int stopSessionInvocations = 0;
+
+  @override
+  Future<void> flush() async {}
+
+  @override
+  Future<FieldTestSummary> stopSession() async {
+    stopSessionInvocations++;
+    eventLog.add('stopSession');
+    return super.stopSession();
+  }
+}
+
+class _Rid3b1RecordsService extends RecordsService {
+  _Rid3b1RecordsService(super.prefs, this.eventLog);
+
+  final List<String> eventLog;
+  int checkInvocations = 0;
+
+  @override
+  Future<void> checkAndSaveRecords(
+    PersonalRecords newStats,
+    void Function(String, double)? onNewRecord, {
+    double confidence = 1.0,
+  }) async {
+    checkInvocations++;
+    eventLog.add('records');
+    return super.checkAndSaveRecords(newStats, onNewRecord,
+        confidence: confidence);
+  }
+}
+
+class MockAudioAlertService extends Fake implements AudioAlertService {
+  final List<String> spokenMessages = [];
+  @override
+  Future<void> speak(String text, {bool force = false}) async {
+    spokenMessages.add(text);
+  }
+}
+
+class FakeTelemetryService extends TelemetryService {
+  final StreamController<TelemetrySample> _controller =
+      StreamController<TelemetrySample>.broadcast();
+
+  FakeTelemetryService(
+    super.loc,
+    super.fusion,
+    super.compass,
+    super.logger,
+    super.validator,
+  );
+
+  @override
+  Stream<TelemetrySample> get telemetryStream => _controller.stream;
+
+  void emitSample(TelemetrySample sample) => _controller.add(sample);
+
+  @override
+  void dispose() => _controller.close();
+}
+
+class _NullLocationService extends Fake implements LocationService {
+  @override
+  void Function(Position position, bool accepted, String? rejectionReason)?
+      onGpsDiagnostic;
+  @override
+  Stream<Position> getFilteredPositionStream() => const Stream.empty();
+}
+
+class _NullSensorFusionService extends Fake implements SensorFusionService {
+  @override
+  Stream<double> get leanAngleStream => const Stream.empty();
+  @override
+  Stream<double> get pitchAngleStream => const Stream.empty();
+  @override
+  Stream<GForceData> get gForceStream => const Stream.empty();
+  @override
+  double get latestRollRateDegPerSec => 0.0;
+  @override
+  double get latestRawAccelX => 0.0;
+  @override
+  double get latestRawAccelY => 0.0;
+  @override
+  double get latestRawAccelZ => 0.0;
+  @override
+  double get latestRawGyroX => 0.0;
+  @override
+  double get latestRawGyroY => 0.0;
+  @override
+  double get latestRawGyroZ => 0.0;
+}
+
+class _NullCompassService extends Fake implements CompassService {
+  @override
+  Stream<double> get headingStream => const Stream.empty();
+}
+
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  const tickWait = Duration(milliseconds: 3300);
+
+  late _Rid3b1AppDatabase db;
+  late _Rid3b1FieldTestLogger logger;
+  late _Rid3b1RecordsService records;
+  late FakeTelemetryService fakeTelemetry;
+  late MockAudioAlertService mockAudio;
+  late TelemetryValidator validator;
+  late SharedPreferences prefs;
+  late List<String> eventLog;
+  late int finishedTransitions;
+
+  final sampleBase = DateTime(2026, 9, 10, 12);
+
+  setUp(() async {
+    SharedPreferences.setMockInitialValues({});
+    prefs = await SharedPreferences.getInstance();
+    eventLog = <String>[];
+    db = _Rid3b1AppDatabase(eventLog);
+    logger = _Rid3b1FieldTestLogger(db, eventLog);
+    records = _Rid3b1RecordsService(prefs, eventLog);
+    validator = TelemetryValidator();
+    fakeTelemetry = FakeTelemetryService(
+      _NullLocationService(),
+      _NullSensorFusionService(),
+      _NullCompassService(),
+      logger,
+      validator,
+    );
+    mockAudio = MockAudioAlertService();
+    finishedTransitions = 0;
+  });
+
+  tearDown(() async {
+    for (final g in db.batchGates) {
+      if (!g.isCompleted) g.complete();
+    }
+    db.batchGates.clear();
+    db.gateBatch = false;
+    fakeTelemetry.dispose();
+    logger.dispose();
+    await db.close();
+  });
+
+  ProviderContainer buildContainer() {
+    final container = ProviderContainer(
+      overrides: [
+        sharedPreferencesProvider.overrideWithValue(prefs),
+        databaseProvider.overrideWithValue(db),
+        telemetryServiceProvider.overrideWithValue(fakeTelemetry),
+        audioAlertServiceProvider.overrideWithValue(mockAudio),
+        fieldTestLoggerProvider.overrideWithValue(logger),
+        recordsServiceProvider.overrideWith((ref) => records),
+      ],
+    );
+    addTearDown(container.dispose);
+    container.listen<RideSession>(rideStateMachineProvider, (prev, next) {
+      if (prev?.state != RideState.finished &&
+          next.state == RideState.finished) {
+        finishedTransitions++;
+        eventLog.add('FINISHED-EDGE(sync-notify)');
+      }
+    });
+    return container;
+  }
+
+  Future<int> startRide(ProviderContainer container) async {
+    final notifier = container.read(rideStateMachineProvider.notifier);
+    await notifier.startRide();
+    return container.read(rideStateMachineProvider).dbId!;
+  }
+
+  /// Unique identity: sample index N carries speed 1000 + N.
+  void emitSamples(int startIndex, int count) {
+    for (var i = 0; i < count; i++) {
+      final index = startIndex + i;
+      fakeTelemetry.emitSample(
+        TelemetrySample(
+          timestamp: sampleBase.add(Duration(milliseconds: index * 50)),
+          latitude: 34.0 + (index * 0.0001),
+          longitude: -85.0,
+          speed: 1000.0 + index,
+          heading: 90.0,
+          gpsAccuracy: 3.0,
+        ),
+      );
+    }
+  }
+
+  Future<List<TelemetryRow>> persistedRows() => (db.select(db.telemetrySamples)
+        ..orderBy([(t) => drift.OrderingTerm(expression: t.id)]))
+      .get();
+
+  Map<int, int> countsByIndex(List<TelemetryRow> rows) {
+    final counts = <int, int>{};
+    for (final row in rows) {
+      final speed = row.speed;
+      if (speed == null || speed < 1000) continue;
+      final idx = (speed - 1000.0).round();
+      counts[idx] = (counts[idx] ?? 0) + 1;
+    }
+    return counts;
+  }
+
+  List<int> duplicatedIn(Map<int, int> c) =>
+      c.entries.where((e) => e.value > 1).map((e) => e.key).toList()..sort();
+
+  List<int> missingOf(int total, Map<int, int> c) =>
+      [for (var i = 0; i < total; i++) if ((c[i] ?? 0) == 0) i];
+
+  int sessionEndedCount() =>
+      mockAudio.spokenMessages.where((m) => m == 'Session ended').length;
+
+  void releaseAllGates() {
+    db.gateBatch = false;
+    for (final g in db.batchGates) {
+      if (!g.isCompleted) g.complete();
+    }
+  }
+
+  group('RID-3B-1: explicit-Stop durability barrier', () {
+    test('A normal Stop makes telemetry durable before endTime, records, '
+        'summary and the finished edge', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      final rideId = await startRide(container);
+
+      emitSamples(0, 10);
+      await pumpEventQueue(times: 30);
+      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10);
+
+      await notifier.finishRide();
+      await pumpEventQueue(times: 30);
+
+      final rows = await persistedRows();
+      final counts = countsByIndex(rows);
+      final session = container.read(rideStateMachineProvider);
+      final rideRow =
+          (await db.select(db.rides).get()).singleWhere((r) => r.id == rideId);
+
+      debugPrint('==== RID-3B-1 A (normal stop) ====');
+      debugPrint('event log : $eventLog');
+      debugPrint('rows      : ${rows.length}  distinct: ${counts.length}');
+      debugPrint('counters  : ${session.totalSamplesReceived} / '
+          '${session.totalSamplesPersisted}');
+
+      // I1 ordering, measured from the shared event log.
+      final commit = eventLog.indexOf('batch0:committed');
+      final endRideAt = eventLog.indexOf('endRide');
+      final recordsAt = eventLog.indexOf('records');
+      final stopAt = eventLog.indexOf('stopSession');
+      final edge = eventLog.indexOf('FINISHED-EDGE(sync-notify)');
+      expect(commit, greaterThanOrEqualTo(0));
+      expect(commit, lessThan(endRideAt), reason: 'durability before endTime');
+      expect(endRideAt, lessThan(recordsAt), reason: 'endTime before records');
+      expect(recordsAt, lessThan(stopAt), reason: 'records before summary');
+      expect(stopAt, lessThan(edge), reason: 'summary before the finished edge');
+
+      expect(rows, hasLength(10));
+      expect(counts.length, 10);
+      expect(duplicatedIn(counts), isEmpty);
+      expect(missingOf(10, counts), isEmpty);
+      expect(rows.map((r) => r.rideId).toSet(), {rideId});
+      expect(rows.map((r) => r.speed).toList(),
+          [for (var i = 0; i < 10; i++) 1000.0 + i]);
+      expect(rideRow.endTime, isNotNull);
+      expect(finishedTransitions, 1, reason: 'exactly one finished edge');
+      expect(sessionEndedCount(), 1);
+      expect(session.state, RideState.finished);
+      expect(session.healthStatus, RecordingHealthStatus.recordingComplete);
+      expect(session.totalSamplesPersisted, 10);
+      expect(await db.select(db.fieldTestSummaries).get(), hasLength(1));
+      expect(records.checkInvocations, 1);
+      expect(logger.stopSessionInvocations, 1);
+    });
+
+    test('B finishRide waits for an in-flight periodic flush and emits no '
+        'finished edge until every sample is durable', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      final rideId = await startRide(container);
+
+      emitSamples(0, 10);
+      await pumpEventQueue(times: 30);
+
+      // The production periodic flush takes ownership of 0..9 and parks.
+      db.gateBatch = true;
+      await Future<void>.delayed(tickWait);
+      expect(db.batchInvocations, 1, reason: 'periodic flush owns batch A');
+      expect(db.batchGates, hasLength(1));
+
+      // Newer telemetry accumulates while A is in flight.
+      emitSamples(10, 5);
+      await pumpEventQueue(times: 30);
+      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 15);
+
+      final finishing = notifier.finishRide();
+      await pumpEventQueue(times: 40);
+
+      // B.4: finishRide MUST NOT finish while the periodic owner is parked.
+      expect(container.read(rideStateMachineProvider).state,
+          isNot(RideState.finished));
+      expect(finishedTransitions, 0, reason: 'no sync-visible edge yet');
+      expect(eventLog.contains('endRide'), isFalse,
+          reason: 'no endTime before durability');
+      expect(eventLog.contains('records'), isFalse);
+      expect(eventLog.contains('stopSession'), isFalse);
+      expect(await persistedRows(), isEmpty);
+
+      // B.5/B.6: release A; the barrier then drains the remainder itself.
+      releaseAllGates();
+      await finishing;
+      await pumpEventQueue(times: 40);
+
+      final rows = await persistedRows();
+      final counts = countsByIndex(rows);
+      final session = container.read(rideStateMachineProvider);
+      debugPrint('==== RID-3B-1 B (in-flight periodic flush) ====');
+      debugPrint('event log : $eventLog');
+      debugPrint('rows      : ${rows.length}  distinct: ${counts.length}');
+      debugPrint('counters  : ${session.totalSamplesReceived} / '
+          '${session.totalSamplesPersisted}');
+
+      // B.7: the finished edge only after ALL telemetry is durable.
+      final edge = eventLog.indexOf('FINISHED-EDGE(sync-notify)');
+      for (final entry in eventLog) {
+        if (entry.endsWith(':committed')) {
+          expect(eventLog.indexOf(entry), lessThan(edge),
+              reason: 'every commit precedes the finished edge');
+        }
+      }
+      expect(rows, hasLength(15));
+      expect(counts.length, 15);
+      expect(duplicatedIn(counts), isEmpty);
+      expect(missingOf(15, counts), isEmpty);
+      expect(rows.map((r) => r.rideId).toSet(), {rideId});
+      expect(rows.map((r) => r.speed).toList(),
+          [for (var i = 0; i < 15; i++) 1000.0 + i]);
+      expect(finishedTransitions, 1);
+      expect(session.totalSamplesPersisted, 15);
+    });
+
+    test('C telemetry arriving during finalization is retained, persisted '
+        'exactly once, and durable before the finished edge', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      final rideId = await startRide(container);
+
+      emitSamples(0, 5);
+      await pumpEventQueue(times: 30);
+
+      db.gateBatch = true;
+      final finishing = notifier.finishRide();
+      await pumpEventQueue(times: 20);
+      expect(db.batchInvocations, 1, reason: 'first drain pass parked');
+
+      // Arrives before the acceptance cutoff, so it must be retained.
+      emitSamples(5, 5);
+      await pumpEventQueue(times: 30);
+      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10);
+
+      releaseAllGates();
+      await finishing;
+      await pumpEventQueue(times: 40);
+
+      final rows = await persistedRows();
+      final counts = countsByIndex(rows);
+      final session = container.read(rideStateMachineProvider);
+      debugPrint('==== RID-3B-1 C (during-finalization arrival) ====');
+      debugPrint('event log : $eventLog');
+      debugPrint('rows      : ${rows.length}  distinct: ${counts.length}');
+
+      expect(rows, hasLength(10), reason: 'nothing dropped to simplify finish');
+      expect(counts.length, 10);
+      expect(duplicatedIn(counts), isEmpty);
+      expect(missingOf(10, counts), isEmpty);
+      expect(rows.map((r) => r.rideId).toSet(), {rideId});
+      expect(rows.map((r) => r.speed).toList(),
+          [for (var i = 0; i < 10; i++) 1000.0 + i]);
+      expect(eventLog.indexOf('batch1:committed'),
+          lessThan(eventLog.indexOf('FINISHED-EDGE(sync-notify)')),
+          reason: 'the during-finalization batch is durable before the edge');
+      expect(finishedTransitions, 1);
+      expect(session.totalSamplesPersisted, 10);
+    });
+
+    test('E final drain failure aborts finalization: no endTime, records, '
+        'summary, announcement or finished edge', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      final rideId = await startRide(container);
+
+      emitSamples(0, 10);
+      await pumpEventQueue(times: 30);
+
+      db.batchFailuresRemaining = 1;
+      Object? escaped;
+      try {
+        await notifier.finishRide();
+      } catch (e) {
+        escaped = e;
+      }
+      await pumpEventQueue(times: 40);
+
+      final session = container.read(rideStateMachineProvider);
+      final rows = await persistedRows();
+      final rideRow =
+          (await db.select(db.rides).get()).singleWhere((r) => r.id == rideId);
+
+      debugPrint('==== RID-3B-1 E (final drain failure) ====');
+      debugPrint('event log : $eventLog');
+      debugPrint('escaped   : ${escaped.runtimeType}');
+      debugPrint('state     : ${session.state} / ${session.healthStatus}');
+      debugPrint('counters  : ${session.totalSamplesReceived} / '
+          '${session.totalSamplesPersisted}');
+
+      expect(escaped, isA<StateError>(),
+          reason: 'the durability failure must propagate from finishRide');
+      expect(session.state, isNot(RideState.finished), reason: 'I2');
+      expect(session.healthStatus, RecordingHealthStatus.recordingActive,
+          reason: 'truthful recoverable recording status');
+      expect(rideRow.endTime, isNull, reason: 'I2: no endTime written');
+      expect(db.endRideInvocations, 0);
+      expect(records.checkInvocations, 0, reason: 'I2: no records pass');
+      expect(logger.stopSessionInvocations, 0, reason: 'I2: no logger stop');
+      expect(await db.select(db.fieldTestSummaries).get(), isEmpty,
+          reason: 'I2: no forensic summary');
+      expect(finishedTransitions, 0, reason: 'I2: no sync-visible edge');
+      expect(eventLog.contains('FINISHED-EDGE(sync-notify)'), isFalse);
+      expect(sessionEndedCount(), 0, reason: 'I2: no Session ended');
+      expect(rows, isEmpty);
+      expect(session.totalSamplesPersisted, 0);
+      expect(session.totalSamplesReceived, 10,
+          reason: 'RID-3A retained the failed snapshot for retry');
+      expect(session.dbId, rideId, reason: 'ride remains retryable');
+    });
+
+    test('F after a failed finalization an explicit retry finalizes exactly '
+        'once with every accepted sample durable', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      final rideId = await startRide(container);
+      final cloudRideIdBefore = (await db.select(db.rides).get())
+          .singleWhere((r) => r.id == rideId)
+          .cloudRideId;
+
+      emitSamples(0, 10);
+      await pumpEventQueue(times: 30);
+
+      db.batchFailuresRemaining = 1;
+      try {
+        await notifier.finishRide();
+      } catch (_) {}
+      await pumpEventQueue(times: 40);
+      expect(finishedTransitions, 0);
+      expect(await persistedRows(), isEmpty);
+
+      // Explicit retry, with no injected failure remaining.
+      await notifier.finishRide();
+      await pumpEventQueue(times: 40);
+
+      final rows = await persistedRows();
+      final counts = countsByIndex(rows);
+      final session = container.read(rideStateMachineProvider);
+      final rides = await db.select(db.rides).get();
+      final rideRow = rides.singleWhere((r) => r.id == rideId);
+
+      debugPrint('==== RID-3B-1 F (failure then retry) ====');
+      debugPrint('event log : $eventLog');
+      debugPrint('rows      : ${rows.length}  distinct: ${counts.length}');
+      debugPrint('counters  : ${session.totalSamplesReceived} / '
+          '${session.totalSamplesPersisted}');
+
+      expect(rows, hasLength(10));
+      expect(counts.length, 10);
+      expect(duplicatedIn(counts), isEmpty, reason: 'retry causes no duplicates');
+      expect(missingOf(10, counts), isEmpty, reason: 'retry loses nothing');
+      expect(rows.map((r) => r.speed).toList(),
+          [for (var i = 0; i < 10; i++) 1000.0 + i],
+          reason: 'chronological order preserved across the failure');
+      expect(db.endRideInvocations, 1, reason: 'exactly one endTime write');
+      expect(rideRow.endTime, isNotNull);
+      expect(records.checkInvocations, 1);
+      expect(logger.stopSessionInvocations, 1);
+      expect(await db.select(db.fieldTestSummaries).get(), hasLength(1));
+      expect(sessionEndedCount(), 1);
+      expect(finishedTransitions, 1);
+      expect(session.state, RideState.finished);
+      expect(rides, hasLength(1), reason: 'no duplicate ride row');
+      expect(rides.single.cloudRideId, cloudRideIdBefore,
+          reason: 'no new cloudRideId minted');
+      expect(session.totalSamplesPersisted, 10);
+    });
+
+    test('D telemetry after the boundary is not attributed to the finished '
+        'ride, and acceptance reopens for the next ride', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      final rideId = await startRide(container);
+
+      emitSamples(0, 5);
+      await pumpEventQueue(times: 30);
+      await notifier.finishRide();
+      await pumpEventQueue(times: 40);
+
+      final receivedAtFinish =
+          container.read(rideStateMachineProvider).totalSamplesReceived;
+      expect(receivedAtFinish, 5);
+
+      // I3: telemetry after the finalization boundary never enters the finished
+      // ride's buffer. Observed indirectly, since the cutoff instant itself is
+      // private and widening the public API to expose it is not permitted.
+      emitSamples(5, 5);
+      await pumpEventQueue(times: 40);
+      expect(container.read(rideStateMachineProvider).totalSamplesReceived,
+          receivedAtFinish,
+          reason: 'no post-boundary sample accepted into the finishing ride');
+      var rows = await persistedRows();
+      expect(rows, hasLength(5));
+      expect(rows.map((r) => r.rideId).toSet(), {rideId});
+
+      // The cutoff is owned solely by the in-flight finish operation, so it must
+      // have reopened: the NEXT ride records normally and its samples are
+      // attributed only to the new ride (no cross-ride leakage).
+      final rideId2 = await startRide(container);
+      expect(rideId2, isNot(rideId));
+      emitSamples(100, 5);
+      await pumpEventQueue(times: 40);
+      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 5,
+          reason: 'acceptance reopened for the new ride');
+      await notifier.finishRide();
+      await pumpEventQueue(times: 40);
+
+      rows = await persistedRows();
+      final ride1Rows = rows.where((r) => r.rideId == rideId).toList();
+      final ride2Rows = rows.where((r) => r.rideId == rideId2).toList();
+      debugPrint('==== RID-3B-1 D (after-boundary attribution) ====');
+      debugPrint('ride1 rows : ${ride1Rows.length}');
+      debugPrint('ride2 rows : ${ride2Rows.length}');
+      debugPrint('ride2 ids  : ${ride2Rows.map((r) => r.speed).toList()}');
+
+      expect(ride1Rows, hasLength(5),
+          reason: 'the finished ride gained no later samples');
+      expect(ride2Rows, hasLength(5), reason: 'the next ride recorded normally');
+      expect(ride2Rows.map((r) => r.speed).toList(),
+          [for (var i = 0; i < 5; i++) 1100.0 + i]);
+      expect(duplicatedIn(countsByIndex(rows)), isEmpty);
+      expect(finishedTransitions, 2, reason: 'one finished edge per ride');
+    });
+
+    test('H telemetry arriving during finalization does not clobber '
+        'recordingFinalizing', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      await startRide(container);
+
+      emitSamples(0, 5);
+      await pumpEventQueue(times: 30);
+
+      db.gateBatch = true;
+      final finishing = notifier.finishRide();
+      await pumpEventQueue(times: 20);
+
+      final healthWhileFinalizing =
+          container.read(rideStateMachineProvider).healthStatus;
+
+      emitSamples(5, 5);
+      await pumpEventQueue(times: 30);
+      final healthAfterArrival =
+          container.read(rideStateMachineProvider).healthStatus;
+
+      releaseAllGates();
+      await finishing;
+      await pumpEventQueue(times: 40);
+
+      debugPrint('==== RID-3B-1 H (healthStatus) ====');
+      debugPrint('while finalizing : $healthWhileFinalizing');
+      debugPrint('after arrival    : $healthAfterArrival');
+      debugPrint('final            : '
+          '${container.read(rideStateMachineProvider).healthStatus}');
+
+      expect(healthWhileFinalizing, RecordingHealthStatus.recordingFinalizing);
+      expect(healthAfterArrival, RecordingHealthStatus.recordingFinalizing,
+          reason: 'incoming telemetry must not reset recordingFinalizing');
+      expect(container.read(rideStateMachineProvider).healthStatus,
+          RecordingHealthStatus.recordingComplete);
+    });
+
+    test('G RID-3A exactly-once accounting is preserved under the durability '
+        'barrier (30-sample overlap shape)', () async {
+      final container = buildContainer();
+      final notifier = container.read(rideStateMachineProvider.notifier);
+      final rideId = await startRide(container);
+
+      emitSamples(0, 10);
+      await pumpEventQueue(times: 30);
+
+      // Periodic flush owns 0..9 and parks.
+      db.gateBatch = true;
+      await Future<void>.delayed(tickWait);
+      expect(db.batchInvocations, 1);
+
+      // 20 newer samples accumulate while the owner is in flight.
+      emitSamples(10, 20);
+      await pumpEventQueue(times: 40);
+      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 30);
+
+      // finishRide must await the owner rather than observe a no-op flush.
+      final finishing = notifier.finishRide();
+      await pumpEventQueue(times: 40);
+      expect(container.read(rideStateMachineProvider).state,
+          isNot(RideState.finished));
+      expect(finishedTransitions, 0);
+
+      releaseAllGates();
+      await finishing;
+      await pumpEventQueue(times: 40);
+
+      final rows = await persistedRows();
+      final counts = countsByIndex(rows);
+      final session = container.read(rideStateMachineProvider);
+      debugPrint('==== RID-3B-1 G (RID-3A accounting under barrier) ====');
+      debugPrint('event log : $eventLog');
+      debugPrint('accepted  : ${session.totalSamplesReceived}');
+      debugPrint('rows      : ${rows.length}');
+      debugPrint('distinct  : ${counts.length}');
+      debugPrint('duplicated: ${duplicatedIn(counts)}');
+      debugPrint('missing   : ${missingOf(30, counts)}');
+      debugPrint('counters  : ${session.totalSamplesReceived} / '
+          '${session.totalSamplesPersisted}');
+
+      expect(rows, hasLength(30), reason: '30 accepted -> 30 durable rows');
+      expect(counts.length, 30, reason: '30 distinct identities');
+      expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates');
+      expect(missingOf(30, counts), isEmpty, reason: '0 missing');
+      expect(rows.map((r) => r.speed).toList(),
+          [for (var i = 0; i < 30; i++) 1000.0 + i],
+          reason: 'chronological identity order');
+      expect(rows.map((r) => r.rideId).toSet(), {rideId});
+      expect(session.totalSamplesPersisted, 30,
+          reason: 'counter integrity: accepted distinct == durable distinct');
+      expect(finishedTransitions, 1);
+      expect((await db.select(db.rides).get()), hasLength(1));
+    });
+  });
+}