// 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));
    });

    test('I continuous live telemetry streaming during finalization completes '
        'durably without throwing StateError', () async {
      final container = buildContainer();
      final notifier = container.read(rideStateMachineProvider.notifier);
      final rideId = await startRide(container);

      // Emit initial 5 samples
      emitSamples(0, 5);
      await pumpEventQueue(times: 20);

      // Start continuous telemetry emission (simulating live 20Hz sensor streaming)
      var streamingIndex = 5;
      final timer = Timer.periodic(const Duration(milliseconds: 15), (_) {
        emitSamples(streamingIndex++, 1);
      });
      addTearDown(timer.cancel);

      // Give timer a moment to emit samples
      await Future<void>.delayed(const Duration(milliseconds: 50));

      // finishRide must complete smoothly without StateError
      await notifier.finishRide();
      timer.cancel();
      await pumpEventQueue(times: 40);

      final session = container.read(rideStateMachineProvider);
      final rows = await persistedRows();
      final counts = countsByIndex(rows);

      expect(session.state, RideState.finished);
      expect(session.healthStatus, RecordingHealthStatus.recordingComplete);
      expect(rows, hasLength(session.totalSamplesPersisted));
      expect(session.totalSamplesReceived, session.totalSamplesPersisted);
      expect(duplicatedIn(counts), isEmpty, reason: 'no duplicates');
      expect(missingOf(rows.length, counts), isEmpty, reason: 'no missing samples');
      expect(finishedTransitions, 1, reason: 'exactly one finished edge');
      expect(rows.map((r) => r.rideId).toSet(), {rideId});
    });
  });
}