Drain bills to the back office automatically, over MQTT or HTTP

Turns the orders table into a queue that empties itself. Bills were only
uploaded when a cashier pressed Sync at end of day; a till that was never
pressed held a day's takings indefinitely.

Drain engine (lib/data/sync/sync_engine.dart)
- Triggers on sale committed, network regained, 5-minute poll, head-office
  request, and the manual button.
- Single flight: a busy till firing a trigger per sale would otherwise have
  several passes reading the same pending rows and send every bill twice.
  A trigger arriving mid-drain is queued and replayed, so nothing is dropped.
- Exponential backoff with +/-20% jitter to a 5-minute ceiling. The jitter
  matters: a store's terminals all fail at the same instant when the line
  drops, and would retry in lockstep without it.
- Halts rather than loops on a failure retrying cannot fix (bad credential,
  refused batch). Pressing Sync clears the halt.

Transports (lib/data/remote/)
- OrderTransport interface; MQTT, HTTP and simulated implementations. The
  repository does not know which is in use.
- MQTT: QoS 1 uplink, application-level ACK correlated by batch_id on a return
  topic, retained Last Will for terminal-offline detection, downlink for
  catalogue pushes and remote sync requests.
- A broker PUBACK is never treated as acceptance. It means the broker holds
  the bytes, not that the ledger took the sale. Only ids the back office names
  are marked synced; silence leaves a bill pending.
- HTTP carries a stable idempotency key across retries of the same bills.

Retention
- Accepted bills are kept 7 days instead of deleted, so a batch the back
  office later loses can be re-sent in full. Purged after that; archived
  totals stay forever.
- forBusinessDate now reads pending rows only. A retained bill exists in both
  the orders table and day_archive, and summing both would overstate the day.

Fixes found while building this
- SyncEngine._refreshPending wrote state.copyWith(pending: await ...). Dart
  evaluates the receiver before the awaited argument, so a connectivity drop
  during the wait was silently overwritten by the stale snapshot. Caught by
  the first run of the new engine tests.
- PrinterSettingsController wrote state after four awaits with no mounted
  check, throwing "used after dispose" when Settings was left mid-load. This
  was pre-existing and reached the cashier as a red screen.

Also
- Header pill now reports real sync state: LIVE / n QUEUED / SYNCING /
  SYNC HALTED, with an explanation of where the bills are.
- Settings shows the route, last upload, next retry and retention window.
- docs/sync-contract.md states what the back office must implement, including
  the idempotency requirement that at-least-once delivery makes mandatory.

Tests: 90 -> 129 passing. New coverage for backoff shape and jitter band,
single flight, halting, ACK correlation and partial acceptance, at-least-once
duplicate handling, retention and purge, and no double-counting after a sync.
Suite run six times clean.

Not addressed: bills already synced by an older build went up overstated and
still need server-side reconciliation. Broker credentials have no Settings
editor yet.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

test/unit/persistence_test.dart

@@ -2,6 +2,7 @@ import 'package:flutter_test/flutter_test.dart';
 import 'package:nearle_pos/data/datasources/local_store.dart';
 import 'package:nearle_pos/data/datasources/remote_catalogue_source.dart';
 import 'package:nearle_pos/data/datasources/seed_data.dart';
+import 'package:nearle_pos/data/remote/simulated_order_transport.dart';
 import 'package:nearle_pos/data/repositories/customer_repository_impl.dart';
 import 'package:nearle_pos/data/repositories/product_repository_impl.dart';
 import 'package:nearle_pos/data/repositories/sync_repository_impl.dart';
@@ -270,7 +271,7 @@ void main() {
       final sync = SyncRepositoryImpl(
         store,
         RemoteCatalogueSource(isOffline: () => false),
-        RemoteOrderSink(isOffline: () => false),
+        SimulatedOrderTransport(isOffline: () => false),
       );
 
       final milk = (await products.findByBarcode('8901234500011'))!;
@@ -314,7 +315,7 @@ void main() {
       final sync = SyncRepositoryImpl(
         store,
         RemoteCatalogueSource(isOffline: () => false),
-        RemoteOrderSink(isOffline: () => false),
+        SimulatedOrderTransport(isOffline: () => false),
       );
 
       final milk = (await products.findByBarcode('8901234500011'))!;
@@ -343,7 +344,7 @@ void main() {
       final sync = SyncRepositoryImpl(
         store,
         RemoteCatalogueSource(isOffline: () => true),
-        RemoteOrderSink(isOffline: () => true),
+        SimulatedOrderTransport(isOffline: () => true),
       );
 
       await sync.importCatalogue();
@@ -371,7 +372,7 @@ void main() {
       final sync = SyncRepositoryImpl(
         store,
         RemoteCatalogueSource(isOffline: () => false),
-        RemoteOrderSink(isOffline: () => false),
+        SimulatedOrderTransport(isOffline: () => false),
       );
 
       await sell('Divya', 2); // 124
@@ -397,7 +398,7 @@ void main() {
       final sync = SyncRepositoryImpl(
         store,
         RemoteCatalogueSource(isOffline: () => false),
-        RemoteOrderSink(isOffline: () => false),
+        SimulatedOrderTransport(isOffline: () => false),
       );
 
       await sell('Divya', 2);

test/unit/retention_test.dart

@@ -0,0 +1,356 @@
+import 'package:flutter_test/flutter_test.dart';
+import 'package:nearle_pos/data/datasources/local_store.dart';
+import 'package:nearle_pos/data/datasources/remote_catalogue_source.dart';
+import 'package:nearle_pos/data/datasources/seed_data.dart';
+import 'package:nearle_pos/data/local/app_database.dart';
+import 'package:nearle_pos/data/local/order_dao.dart';
+import 'package:nearle_pos/data/remote/order_transport.dart';
+import 'package:nearle_pos/data/remote/simulated_order_transport.dart';
+import 'package:nearle_pos/data/repositories/customer_repository_impl.dart';
+import 'package:nearle_pos/data/repositories/product_repository_impl.dart';
+import 'package:nearle_pos/data/repositories/sync_repository_impl.dart';
+import 'package:nearle_pos/data/repositories/transaction_repository_impl.dart';
+import 'package:nearle_pos/domain/entities/cart.dart';
+import 'package:nearle_pos/domain/entities/transaction.dart';
+import 'package:nearle_pos/domain/usecases/checkout_sale.dart';
+
+/// A transport whose answer each call is dictated by the test.
+class _ScriptedTransport implements OrderTransport {
+  _ScriptedTransport(this.answer);
+
+  /// Given the ids in a batch, returns what the back office says about them.
+  PushReceipt Function(List<String> ids) answer;
+
+  int batches = 0;
+
+  @override
+  String get label => 'Scripted';
+
+  @override
+  bool get isConnected => true;
+
+  @override
+  Stream<DownlinkMessage> get downlink => const Stream.empty();
+
+  @override
+  Stream<bool> get connectionState => const Stream.empty();
+
+  @override
+  Future<void> connect() async {}
+
+  @override
+  Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders) async {
+    batches++;
+    return answer(orders.map((o) => o['id']! as String).toList());
+  }
+
+  @override
+  Future<void> dispose() async {}
+}
+
+/// Bills the back office has taken delivery of stay on the terminal for a
+/// week, so a batch the server later loses can still be re-sent in full.
+///
+/// The risk this buys is double counting: an accepted bill is now in two
+/// places at once — its own row, and the archived day totals. Most of what
+/// follows is about that.
+void main() {
+  late LocalStore store;
+  late ProductRepositoryImpl products;
+  late CustomerRepositoryImpl customers;
+  late TransactionRepositoryImpl transactions;
+  late CheckoutSale checkout;
+
+  setUpAll(() {
+    LocalStore.registerSeed(
+      products: SeedData.products,
+      customers: SeedData.customers,
+    );
+  });
+
+  setUp(() async {
+    store = LocalStore.instance;
+    await store.reset(withCatalogue: true);
+
+    products = ProductRepositoryImpl(store);
+    customers = CustomerRepositoryImpl(store);
+    transactions = TransactionRepositoryImpl(store);
+    checkout = CheckoutSale(
+      productRepository: products,
+      customerRepository: customers,
+      transactionRepository: transactions,
+    );
+  });
+
+  SyncRepositoryImpl syncWith(OrderTransport transport) => SyncRepositoryImpl(
+        store,
+        RemoteCatalogueSource(isOffline: () => false),
+        transport,
+      );
+
+  /// Rings one bill for [quantity] litres of milk at 62.00 each.
+  Future<double> ringSale({
+    double quantity = 2,
+    String cashier = 'Divya',
+  }) async {
+    final milk = (await products.findByBarcode('8901234500011'))!;
+    final cart = Cart(lines: [CartLine(product: milk, quantity: quantity)]);
+    final due = cart.grandTotal;
+
+    await checkout(
+      cart: cart,
+      payments: [
+        PaymentSplit(
+          method: PaymentMethod.cash,
+          amount: due,
+          tendered: due,
+        ),
+      ],
+      cashierName: cashier,
+    );
+    return due;
+  }
+
+  group('accepted bills are kept, not deleted', () {
+    test('a synced bill is still on the terminal and still re-sendable',
+        () async {
+      await ringSale();
+      final sync = syncWith(SimulatedOrderTransport(isOffline: () => false));
+
+      expect(await sync.unsyncedCount(), 1);
+      await sync.syncOrders();
+      expect(await sync.unsyncedCount(), 0);
+
+      // The row survives, carrying its line items, so the full bill can go up
+      // again if the back office loses it.
+      final rows = await sync.orderSyncRows();
+      expect(rows, hasLength(1));
+      expect(rows.single.isSynced, isTrue);
+      expect(rows.single.syncedAt, isNotNull);
+
+      final stored = await store.orders.recent();
+      expect(stored, hasLength(1));
+      expect(stored.single.cart.lines, isNotEmpty,
+          reason: 'a kept bill with no lines could not be re-sent',);
+    });
+
+    test("today's takings are not counted twice while the bill is retained",
+        () async {
+      // The bug this exists to catch: an accepted bill is in the archive *and*
+      // still in the orders table. Summing both would inflate the day.
+      final due = await ringSale(quantity: 3);
+      final sync = syncWith(SimulatedOrderTransport(isOffline: () => false));
+
+      final before = await sync.todayReport(
+        terminalId: 'TERM-01',
+        cashierName: 'Divya',
+      );
+      expect(before.grossSales, closeTo(due, 0.01));
+
+      await sync.syncOrders();
+
+      final after = await sync.todayReport(
+        terminalId: 'TERM-01',
+        cashierName: 'Divya',
+      );
+      expect(after.grossSales, closeTo(due, 0.01),
+          reason: 'syncing must not change what the shop took',);
+      expect(after.billCount, 1);
+    });
+
+    test('a second sync does not send an already-accepted bill again',
+        () async {
+      await ringSale();
+      final transport = _ScriptedTransport((ids) => PushReceipt(accepted: ids));
+      final sync = syncWith(transport);
+
+      await sync.syncOrders();
+      expect(transport.batches, 1);
+
+      final outcome = await sync.syncOrders();
+      expect(outcome.hadNothingToDo, isTrue);
+      expect(transport.batches, 1,
+          reason: 'a retained bill must not be re-uploaded',);
+    });
+  });
+
+  group('purging', () {
+    test('a bill past its window goes, and its archived totals stay', () async {
+      final due = await ringSale(quantity: 4);
+      final sync = syncWith(SimulatedOrderTransport(isOffline: () => false));
+      await sync.syncOrders();
+
+      // Backdate the acceptance past the retention window.
+      await AppDatabase.instance.db.rawUpdate(
+        'UPDATE orders SET synced_at = ?',
+        [
+          DateTime.now()
+              .subtract(OrderDao.retentionWindow + const Duration(days: 1))
+              .millisecondsSinceEpoch,
+        ],
+      );
+
+      expect(await sync.purgeExpired(), 1);
+      expect(await store.orders.recent(), isEmpty);
+
+      // What the shop was paid is unchanged — only the re-sendable copy went.
+      final report = await sync.todayReport(
+        terminalId: 'TERM-01',
+        cashierName: 'Divya',
+      );
+      expect(report.grossSales, closeTo(due, 0.01));
+      expect(report.billCount, 1);
+    });
+
+    test('a bill inside its window is left alone', () async {
+      await ringSale();
+      final sync = syncWith(SimulatedOrderTransport(isOffline: () => false));
+      await sync.syncOrders();
+
+      expect(await sync.purgeExpired(), 0);
+      expect(await store.orders.recent(), hasLength(1));
+    });
+
+    test('an unsynced bill is never purged, however old', () async {
+      // The one thing that must never happen: a bill the back office has not
+      // taken delivery of being deleted from the only place it exists.
+      await ringSale();
+      await AppDatabase.instance.db.rawUpdate(
+        'UPDATE orders SET created_at = ?, synced_at = ?',
+        [0, 0],
+      );
+
+      final sync = syncWith(SimulatedOrderTransport(isOffline: () => false));
+      expect(await sync.purgeExpired(), 0);
+      expect(await sync.unsyncedCount(), 1);
+    });
+  });
+
+  group('partial acceptance', () {
+    test('a bill the back office stayed silent about stays pending', () async {
+      // Silence is not acceptance.
+      await ringSale(quantity: 1);
+      await ringSale(quantity: 2);
+
+      final transport = _ScriptedTransport(
+        (ids) => PushReceipt(accepted: [ids.first]),
+      );
+      final sync = syncWith(transport);
+
+      final outcome = await sync.syncOrders();
+      expect(outcome.uploaded, 1);
+      expect(outcome.rejected, 1);
+      expect(await sync.unsyncedCount(), 1,
+          reason: 'the unconfirmed bill must still be owed',);
+    });
+
+    test('a refusal stops the drain instead of looping on the same rows',
+        () async {
+      await ringSale();
+      final transport = _ScriptedTransport(
+        (ids) => PushReceipt(
+          accepted: const [],
+          rejected: {for (final id in ids) id: 'duplicate invoice'},
+        ),
+      );
+      final sync = syncWith(transport);
+
+      final outcome = await sync.syncOrders();
+
+      expect(outcome.uploaded, 0);
+      expect(outcome.isSuccess, isFalse);
+      expect(outcome.isRetryable, isFalse,
+          reason: 'the same bytes will be refused again',);
+      expect(transport.batches, 1,
+          reason: 'the refused page must not be fetched and sent forever',);
+      expect(outcome.error, contains('duplicate invoice'));
+    });
+
+    test('the refusal reason is recorded against the bill for a person to read',
+        () async {
+      await ringSale();
+      final sync = syncWith(_ScriptedTransport(
+        (ids) => PushReceipt(
+          accepted: const [],
+          rejected: {for (final id in ids) id: 'unknown product code'},
+        ),
+      ),);
+
+      await sync.syncOrders();
+
+      final row = (await sync.orderSyncRows()).single;
+      expect(row.isSynced, isFalse);
+      expect(row.error, 'unknown product code');
+      expect(row.attempts, 1);
+    });
+  });
+
+  group('at-least-once delivery', () {
+    test('a batch accepted twice is banked once', () async {
+      // MQTT will re-deliver, and a lost ack means the terminal sends again.
+      // The second acceptance must not double the archived takings.
+      final due = await ringSale(quantity: 5);
+      final transport = _ScriptedTransport((ids) => PushReceipt(accepted: ids));
+      final sync = syncWith(transport);
+
+      await sync.syncOrders();
+      // A duplicate ack for bills already marked synced — the drain finds
+      // nothing pending and does nothing.
+      await sync.syncOrders();
+
+      final report = await sync.todayReport(
+        terminalId: 'TERM-01',
+        cashierName: 'Divya',
+      );
+      expect(report.grossSales, closeTo(due, 0.01));
+      expect(report.billCount, 1);
+    });
+  });
+
+  group('transport failure', () {
+    test('an unreachable back office leaves every bill exactly where it was',
+        () async {
+      final due = await ringSale(quantity: 6);
+      final sync = syncWith(SimulatedOrderTransport(isOffline: () => true));
+
+      final outcome = await sync.syncOrders();
+
+      expect(outcome.isSuccess, isFalse);
+      expect(outcome.uploaded, 0);
+      expect(await sync.unsyncedCount(), 1);
+
+      final report = await sync.todayReport(
+        terminalId: 'TERM-01',
+        cashierName: 'Divya',
+      );
+      expect(report.grossSales, closeTo(due, 0.01),
+          reason: 'a failed upload must not change the shift total',);
+    });
+
+    test('a batch is bounded so a backlog cannot exceed a broker message',
+        () async {
+      for (var i = 0; i < 5; i++) {
+        await ringSale(quantity: 1);
+      }
+
+      final sizes = <int>[];
+      final transport = _ScriptedTransport((ids) {
+        sizes.add(ids.length);
+        return PushReceipt(accepted: ids);
+      });
+
+      final sync = SyncRepositoryImpl(
+        store,
+        RemoteCatalogueSource(isOffline: () => false),
+        transport,
+        batchSize: 2,
+      );
+
+      final outcome = await sync.syncOrders();
+
+      expect(outcome.uploaded, 5);
+      expect(sizes, [2, 2, 1], reason: 'the backlog must drain in pages');
+      expect(await sync.unsyncedCount(), 0);
+    });
+  });
+}

test/unit/sync_engine_test.dart

@@ -0,0 +1,436 @@
+import 'dart:async';
+import 'dart:math';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:nearle_pos/data/remote/order_transport.dart';
+import 'package:nearle_pos/data/sync/sync_engine.dart';
+import 'package:nearle_pos/domain/entities/shift_report.dart';
+import 'package:nearle_pos/domain/entities/sync_event.dart';
+import 'package:nearle_pos/domain/entities/transaction.dart';
+import 'package:nearle_pos/domain/repositories/sync_repository.dart';
+
+/// A repository whose every answer the test dictates.
+///
+/// The engine is a scheduler; what it schedules is irrelevant here. Driving it
+/// with a stub is what makes "did it retry, and when" answerable without a
+/// database or a network.
+class _StubRepository implements SyncRepository {
+  _StubRepository();
+
+  /// Answers handed out in order; the last one repeats once exhausted.
+  final List<SyncOutcome> scripted = [];
+  int calls = 0;
+  int pending = 0;
+
+  /// Completed by the test to hold a drain open, so overlapping triggers can
+  /// be observed.
+  Completer<void>? gate;
+
+  @override
+  Future<SyncOutcome> syncOrders({
+    void Function(double progress, String stage)? onProgress,
+  }) async {
+    calls++;
+    if (gate != null) await gate!.future;
+    if (scripted.isEmpty) return const SyncOutcome(attempted: 0, uploaded: 0);
+    return scripted[(calls - 1).clamp(0, scripted.length - 1)];
+  }
+
+  @override
+  Future<int> unsyncedCount() async => pending;
+
+  @override
+  Future<int> purgeExpired() async => 0;
+
+  @override
+  bool get hasCatalogue => true;
+
+  @override
+  DateTime? get lastImportAt => null;
+
+  @override
+  String? get catalogueRevision => null;
+
+  @override
+  List<SyncEvent> get events => const [];
+
+  @override
+  Future<SyncEvent> importCatalogue({
+    void Function(double progress, String stage)? onProgress,
+  }) =>
+      throw UnimplementedError();
+
+  @override
+  Future<List<SaleTransaction>> unsyncedOrders() async => const [];
+
+  @override
+  Future<List<OrderSyncRow>> orderSyncRows({int limit = 200}) async => const [];
+
+  @override
+  Future<ShiftReport> todayReport({
+    required String terminalId,
+    required String cashierName,
+    bool scopeToCashier = false,
+  }) =>
+      throw UnimplementedError();
+}
+
+/// Captures what the engine asked to be scheduled instead of really waiting.
+class _FakeScheduler {
+  final List<Duration> delays = [];
+  final List<void Function()> callbacks = [];
+
+  Timer schedule(Duration d, void Function() cb) {
+    delays.add(d);
+    callbacks.add(cb);
+    return Timer(Duration.zero, () {})..cancel();
+  }
+
+  /// Runs the most recently scheduled callback — the backoff retry.
+  void fireLast() => callbacks.last();
+}
+
+void main() {
+  late _StubRepository repo;
+  late _FakeScheduler scheduler;
+
+  SyncEngine build({
+    Stream<bool>? connectivity,
+    Stream<DownlinkMessage>? downlink,
+    Random? random,
+  }) =>
+      SyncEngine(
+        repository: repo,
+        connectivity: connectivity,
+        downlink: downlink,
+        // Fixed seed so the jitter band is assertable rather than flaky.
+        random: random ?? Random(7),
+        scheduleTimer: scheduler.schedule,
+      );
+
+  setUp(() {
+    repo = _StubRepository();
+    scheduler = _FakeScheduler();
+  });
+
+  group('single flight', () {
+    test('overlapping triggers do not start a second pass over the same bills',
+        () async {
+      // Without this guarantee a busy till firing a trigger per sale would have
+      // several drains reading the same pending rows at once, and every bill
+      // would go up two or three times.
+      repo.gate = Completer<void>();
+      final engine = build();
+
+      engine
+        ..nudge(SyncTrigger.saleCommitted)
+        ..nudge(SyncTrigger.saleCommitted)
+        ..nudge(SyncTrigger.saleCommitted);
+
+      await Future<void>.delayed(Duration.zero);
+      expect(repo.calls, 1, reason: 'three triggers must yield one drain');
+
+      repo.gate!.complete();
+      repo.gate = null;
+      await Future<void>.delayed(Duration.zero);
+
+      await engine.dispose();
+    });
+
+    test('a trigger arriving mid-drain is honoured once that drain finishes',
+        () async {
+      // Bills committed during a pass were not in the set it read. Dropping
+      // the trigger would leave them waiting for the next poll.
+      repo
+        ..gate = Completer<void>()
+        ..pending = 3;
+      final engine = build();
+
+      engine.nudge(SyncTrigger.startup);
+      await Future<void>.delayed(Duration.zero);
+      expect(repo.calls, 1);
+
+      engine.nudge(SyncTrigger.saleCommitted);
+      repo.gate!.complete();
+      repo.gate = null;
+
+      await Future<void>.delayed(Duration.zero);
+      await Future<void>.delayed(Duration.zero);
+
+      expect(repo.calls, 2, reason: 'the queued trigger must be replayed');
+      await engine.dispose();
+    });
+
+    test('a queued trigger with nothing left owing does not run a second pass',
+        () async {
+      // The counterpart to the test above. The drain that just finished emptied
+      // the queue, so replaying the trigger would send nothing and only churn
+      // the connection.
+      repo
+        ..gate = Completer<void>()
+        ..pending = 0;
+      final engine = build();
+
+      engine.nudge(SyncTrigger.startup);
+      await Future<void>.delayed(Duration.zero);
+
+      engine.nudge(SyncTrigger.saleCommitted);
+      repo.gate!.complete();
+      repo.gate = null;
+      await _settle();
+
+      expect(repo.calls, 1);
+      await engine.dispose();
+    });
+  });
+
+  group('backoff', () {
+    test('doubles per failure and stops at the ceiling', () {
+      final engine = SyncEngine(
+        repository: repo,
+        baseBackoff: const Duration(seconds: 2),
+        maxBackoff: const Duration(minutes: 5),
+        // No jitter, so the shape of the curve is what is being asserted.
+        random: _ZeroJitter(),
+        scheduleTimer: scheduler.schedule,
+      );
+
+      // 0.8 is the bottom of the jitter band, which _ZeroJitter pins.
+      expect(engine.backoffFor(1).inMilliseconds, 1600); // 2s
+      expect(engine.backoffFor(2).inMilliseconds, 3200); // 4s
+      expect(engine.backoffFor(3).inMilliseconds, 6400); // 8s
+      expect(engine.backoffFor(9).inSeconds, 240); // 512s → capped
+      expect(engine.backoffFor(30).inSeconds, 240); // still capped
+    });
+
+    test('jitter keeps every delay inside ±20% of the nominal wait', () {
+      // A shop's terminals all fail at the same instant when the line drops.
+      // Without jitter they would retry in lockstep and keep colliding.
+      final engine = build(random: Random(1));
+
+      final seen = <int>{};
+      for (var i = 0; i < 200; i++) {
+        final ms = engine.backoffFor(4).inMilliseconds;
+        expect(ms, greaterThanOrEqualTo((16000 * 0.8).round()));
+        expect(ms, lessThanOrEqualTo((16000 * 1.2).round()));
+        seen.add(ms);
+      }
+      expect(seen.length, greaterThan(50), reason: 'delays must actually vary');
+    });
+
+    test('a failed drain schedules a retry and a success clears it', () async {
+      repo.scripted.addAll([
+        const SyncOutcome(attempted: 2, uploaded: 0, error: 'line dropped'),
+        const SyncOutcome(attempted: 2, uploaded: 2),
+      ]);
+
+      final engine = build();
+      engine.nudge(SyncTrigger.saleCommitted);
+      await _settle();
+
+      expect(engine.state.consecutiveFailures, 1);
+      expect(engine.state.nextAttemptAt, isNotNull);
+      expect(scheduler.delays, hasLength(1));
+
+      scheduler.fireLast();
+      await _settle();
+
+      expect(engine.state.consecutiveFailures, 0);
+      expect(engine.state.lastError, isNull);
+      expect(engine.state.nextAttemptAt, isNull);
+      expect(engine.state.lastSuccessAt, isNotNull);
+
+      await engine.dispose();
+    });
+  });
+
+  group('halting', () {
+    test('a refused batch halts instead of retrying the same bytes forever',
+        () async {
+      // A rejection is a decision, not a fault. Re-sending gets the same
+      // answer, and a loop would bury the one message a person needs to see.
+      repo.scripted.add(const SyncOutcome(
+        attempted: 1,
+        uploaded: 0,
+        rejected: 1,
+        error: 'duplicate invoice number',
+        isRetryable: false,
+      ),);
+
+      final engine = build();
+      engine.nudge(SyncTrigger.saleCommitted);
+      await _settle();
+
+      expect(engine.state.isHalted, isTrue);
+      expect(scheduler.delays, isEmpty, reason: 'no retry may be scheduled');
+
+      // Further background triggers are ignored while halted.
+      engine
+        ..nudge(SyncTrigger.periodic)
+        ..nudge(SyncTrigger.saleCommitted);
+      await _settle();
+      expect(repo.calls, 1);
+
+      await engine.dispose();
+    });
+
+    test('pressing sync clears a halt and tries again', () async {
+      // The button is how a cashier retries once the back office is fixed.
+      repo.scripted.add(const SyncOutcome(
+        attempted: 1,
+        uploaded: 0,
+        error: 'bad credential',
+        isRetryable: false,
+      ),);
+
+      final engine = build();
+      engine.nudge(SyncTrigger.saleCommitted);
+      await _settle();
+      expect(engine.state.isHalted, isTrue);
+
+      repo.scripted
+        ..clear()
+        ..add(const SyncOutcome(attempted: 1, uploaded: 1));
+      repo.calls = 0;
+
+      await engine.syncNow();
+      expect(repo.calls, 1);
+      expect(engine.state.isHalted, isFalse);
+
+      await engine.dispose();
+    });
+  });
+
+  group('triggers', () {
+    test('the network coming back starts a drain; going away does not',
+        () async {
+      final connectivity = StreamController<bool>();
+      final engine = build(connectivity: connectivity.stream);
+      await engine.start();
+
+      final atStart = repo.calls;
+
+      connectivity.add(false);
+      await _settle();
+      expect(repo.calls, atStart,
+          reason: 'an attempt certain to fail is not worth making',);
+      expect(engine.state.online, isFalse);
+
+      connectivity.add(true);
+      await _settle();
+      expect(repo.calls, atStart + 1);
+
+      await connectivity.close();
+      await engine.dispose();
+    });
+
+    test('background triggers are ignored while offline, manual is not',
+        () async {
+      final connectivity = StreamController<bool>();
+      final engine = build(connectivity: connectivity.stream);
+      await engine.start();
+
+      connectivity.add(false);
+      await _settle();
+      final atStart = repo.calls;
+
+      engine.nudge(SyncTrigger.saleCommitted);
+      await _settle();
+      expect(repo.calls, atStart);
+
+      // The cashier may know something the engine does not, and being told why
+      // it failed beats a button that does nothing.
+      await engine.syncNow();
+      expect(repo.calls, atStart + 1);
+
+      await connectivity.close();
+      await engine.dispose();
+    });
+
+    test('head office can pull a shift up over the downlink', () async {
+      final downlink = StreamController<DownlinkMessage>();
+      // Bills owed, otherwise a request to sync correctly does nothing.
+      repo.pending = 2;
+      final engine = build(downlink: downlink.stream);
+      await engine.start();
+      // Let the startup drain finish, so this measures the downlink and not a
+      // race with it.
+      await _settle();
+
+      final atStart = repo.calls;
+      downlink.add(const DownlinkMessage(kind: DownlinkKind.syncRequested));
+      await _settle();
+
+      expect(repo.calls, atStart + 1);
+
+      await downlink.close();
+      await engine.dispose();
+    });
+
+    test('an unrecognised downlink message is ignored, not acted on', () async {
+      final downlink = StreamController<DownlinkMessage>();
+      repo.pending = 2;
+      final engine = build(downlink: downlink.stream);
+      await engine.start();
+      await _settle();
+
+      final atStart = repo.calls;
+      downlink.add(const DownlinkMessage(kind: DownlinkKind.unknown));
+      await _settle();
+
+      expect(repo.calls, atStart);
+
+      await downlink.close();
+      await engine.dispose();
+    });
+  });
+
+  test('a repository that throws is treated as a retryable failure, not a crash',
+      () async {
+    // Anything escaping the repository is a defect. The engine must still empty
+    // its queue once the defect is fixed, rather than dying on the first sale.
+    final engine = SyncEngine(
+      repository: _ThrowingRepository(),
+      random: Random(3),
+      scheduleTimer: scheduler.schedule,
+    );
+
+    engine.nudge(SyncTrigger.saleCommitted);
+    await _settle();
+
+    expect(engine.state.consecutiveFailures, 1);
+    expect(engine.state.isHalted, isFalse);
+    expect(scheduler.delays, hasLength(1));
+
+    await engine.dispose();
+  });
+}
+
+/// Pins the jitter multiplier at its lower bound so the backoff curve itself
+/// can be asserted.
+class _ZeroJitter implements Random {
+  @override
+  double nextDouble() => 0;
+
+  @override
+  bool nextBool() => false;
+
+  @override
+  int nextInt(int max) => 0;
+}
+
+class _ThrowingRepository extends _StubRepository {
+  @override
+  Future<SyncOutcome> syncOrders({
+    void Function(double progress, String stage)? onProgress,
+  }) async =>
+      throw StateError('boom');
+}
+
+/// Lets queued microtasks run. The engine never really waits, so a handful of
+/// turns is enough for everything it schedules to settle.
+Future<void> _settle() async {
+  for (var i = 0; i < 6; i++) {
+    await Future<void>.delayed(Duration.zero);
+  }
+}

test/unit/transport_test.dart

@@ -0,0 +1,276 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:http/http.dart' as http;
+import 'package:nearle_pos/core/config/sync_config.dart';
+import 'package:nearle_pos/data/remote/http_order_transport.dart';
+import 'package:nearle_pos/data/remote/mqtt_order_transport.dart';
+import 'package:nearle_pos/data/remote/order_transport.dart';
+
+/// Two bills, enough to tell "all accepted" from "some accepted".
+final _orders = [
+  {'id': 'order-a', 'invoice_number': 'INV-1', 'total': 100.0},
+  {'id': 'order-b', 'invoice_number': 'INV-2', 'total': 250.0},
+];
+
+void main() {
+  group('MQTT ack correlation', () {
+    late MqttOrderTransport transport;
+    const config = SyncConfig(
+      transport: TransportKind.mqtt,
+      storeId: 'store-9',
+      terminalId: 'TERM-04',
+      brokerHost: 'broker.invalid',
+    );
+
+    setUp(() => transport = MqttOrderTransport(config: config));
+    tearDown(() => transport.dispose());
+
+    test('topics are namespaced per store and per terminal', () {
+      // Two stores sharing one broker must never see each other's bills.
+      expect(config.orderTopic, 'pos/store-9/TERM-04/order');
+      expect(config.ackTopic, 'pos/store-9/TERM-04/ack');
+      expect(config.statusTopic, 'pos/store-9/TERM-04/status');
+      expect(config.catalogueTopic, 'pos/store-9/catalogue');
+    });
+
+    test('an ack naming only some ids accepts only those', () async {
+      // The heart of it. A partial ack must not be read as "the batch went up":
+      // order-b stays pending and is sent again.
+      final receipt = _receiptFor(transport, {
+        'accepted': ['order-a'],
+        'rejected': {'order-b': 'unknown product'},
+      });
+
+      final result = await receipt;
+      expect(result.accepted, ['order-a']);
+      expect(result.rejected, {'order-b': 'unknown product'});
+    });
+
+    test('an ack for a different batch does not release this one', () async {
+      final pending = transport.pushOrders(_orders).timeout(
+            const Duration(milliseconds: 300),
+            onTimeout: () => throw TimeoutException('not released'),
+          );
+
+      // Give the publish a turn to register its correlation id, then answer
+      // with someone else's.
+      await Future<void>.delayed(Duration.zero);
+      transport.handleInbound(
+        config.ackTopic,
+        jsonEncode({'batch_id': 'a-different-batch', 'accepted': ['order-a']}),
+      );
+
+      await expectLater(pending, throwsA(isA<Exception>()));
+    });
+
+    test('a malformed ack is discarded rather than taken down the connection',
+        () {
+      // The next message may be a perfectly good ack releasing a day's bills.
+      expect(
+        () => transport.handleInbound(config.ackTopic, 'not json at all'),
+        returnsNormally,
+      );
+      expect(
+        () => transport.handleInbound(config.ackTopic, '{"no":"batch id"}'),
+        returnsNormally,
+      );
+    });
+
+    test('an ack with no accepted list releases nothing', () async {
+      // Silence is not acceptance. A back office that answers `{}` must not
+      // cause a single bill to be marked synced.
+      final result = await _receiptFor(transport, {'accepted': <String>[]});
+      expect(result.accepted, isEmpty);
+    });
+
+    test('a bare list of rejected ids is tolerated', () async {
+      final result = await _receiptFor(transport, {
+        'accepted': ['order-a'],
+        'rejected': ['order-b'],
+      });
+      expect(result.rejected.keys, ['order-b']);
+    });
+
+    test('catalogue pushes arrive on the downlink', () async {
+      final received = <DownlinkMessage>[];
+      final sub = transport.downlink.listen(received.add);
+
+      transport
+        ..handleInbound(config.catalogueTopic, jsonEncode({'revision': 'r9'}))
+        ..handleInbound(config.commandTopic, jsonEncode({'command': 'sync'}))
+        ..handleInbound(
+          config.commandTopic,
+          jsonEncode({'command': 'self-destruct'}),
+        );
+
+      await Future<void>.delayed(Duration.zero);
+      await sub.cancel();
+
+      expect(
+        received.map((m) => m.kind),
+        [
+          DownlinkKind.catalogueChanged,
+          DownlinkKind.syncRequested,
+          // A newer server talking to an older terminal stays visible instead
+          // of being silently dropped.
+          DownlinkKind.unknown,
+        ],
+      );
+      expect(received.first.payload['revision'], 'r9');
+    });
+  });
+
+  group('HTTP transport', () {
+    const SyncConfig config = SyncConfig(
+      transport: TransportKind.http,
+      httpBaseUrl: 'https://back.office.test',
+      apiKey: 'k',
+    );
+
+    test('accepts only the ids the endpoint names', () async {
+      final transport = HttpOrderTransport(
+        config: config,
+        client: _FakeClient((_) => http.Response(
+              jsonEncode({
+                'accepted': ['order-a'],
+                'rejected': {'order-b': 'stale price list'},
+              }),
+              200,
+            ),),
+      );
+
+      final receipt = await transport.pushOrders(_orders);
+      expect(receipt.accepted, ['order-a']);
+      expect(receipt.rejected['order-b'], 'stale price list');
+      await transport.dispose();
+    });
+
+    test('a bare 200 with no body marks nothing synced', () async {
+      // Guessing here would retire a day's takings on an empty response.
+      final transport = HttpOrderTransport(
+        config: config,
+        client: _FakeClient((_) => http.Response('', 200)),
+      );
+
+      await expectLater(
+        transport.pushOrders(_orders),
+        throwsA(isA<TransportException>()),
+      );
+      await transport.dispose();
+    });
+
+    test('a 200 that names nothing accepts nothing, without throwing',
+        () async {
+      final transport = HttpOrderTransport(
+        config: config,
+        client: _FakeClient((_) => http.Response('{"accepted":[]}', 200)),
+      );
+
+      final receipt = await transport.pushOrders(_orders);
+      expect(receipt.accepted, isEmpty);
+      await transport.dispose();
+    });
+
+    test('a bad credential is not retryable, so the engine can halt', () async {
+      // Hammering the endpoint would only bury the one message a person needs.
+      final transport = HttpOrderTransport(
+        config: config,
+        client: _FakeClient((_) => http.Response('nope', 401)),
+      );
+
+      await expectLater(
+        transport.pushOrders(_orders),
+        throwsA(isA<TransportException>()
+            .having((e) => e.retryable, 'retryable', isFalse),),
+      );
+      await transport.dispose();
+    });
+
+    test('a server error is retryable', () async {
+      final transport = HttpOrderTransport(
+        config: config,
+        client: _FakeClient((_) => http.Response('boom', 503)),
+      );
+
+      await expectLater(
+        transport.pushOrders(_orders),
+        throwsA(isA<TransportException>()
+            .having((e) => e.retryable, 'retryable', isTrue),),
+      );
+      await transport.dispose();
+    });
+
+    test('a retry of the same bills carries the same idempotency key',
+        () async {
+      // So the endpoint can collapse a duplicate batch server-side rather than
+      // relying on every order id being checked one at a time.
+      final keys = <String>[];
+      final transport = HttpOrderTransport(
+        config: config,
+        client: _FakeClient((request) {
+          keys.add(request.headers['idempotency-key'] ?? '');
+          return http.Response('{"accepted":["order-a","order-b"]}', 200);
+        }),
+      );
+
+      await transport.pushOrders(_orders);
+      await transport.pushOrders(_orders);
+      expect(keys.first, keys.last);
+      expect(keys.first, isNotEmpty);
+
+      await transport.pushOrders([_orders.first]);
+      expect(keys.last, isNot(keys.first));
+
+      await transport.dispose();
+    });
+
+    test('an unconfigured endpoint fails fast rather than retrying', () async {
+      final transport = HttpOrderTransport(
+        config: const SyncConfig(transport: TransportKind.http),
+        client: _FakeClient((_) => http.Response('', 200)),
+      );
+
+      await expectLater(
+        transport.pushOrders(_orders),
+        throwsA(isA<TransportException>()
+            .having((e) => e.retryable, 'retryable', isFalse),),
+      );
+      await transport.dispose();
+    });
+  });
+}
+
+/// Waits on a known batch id, then feeds it [body] as if the back office had
+/// answered — exercising the real parse and correlation path with no broker.
+Future<PushReceipt> _receiptFor(
+  MqttOrderTransport transport,
+  Map<String, Object?> body,
+) {
+  const batchId = 'test-batch';
+  final receipt = transport.awaitAck(batchId);
+
+  transport.handleInbound(
+    transport.config.ackTopic,
+    jsonEncode({...body, 'batch_id': batchId}),
+  );
+
+  return receipt;
+}
+
+class _FakeClient extends http.BaseClient {
+  _FakeClient(this.respond);
+
+  final http.Response Function(http.BaseRequest) respond;
+
+  @override
+  Future<http.StreamedResponse> send(http.BaseRequest request) async {
+    final response = respond(request);
+    return http.StreamedResponse(
+      Stream.value(utf8.encode(response.body)),
+      response.statusCode,
+      request: request,
+    );
+  }
+}

test/widget/app_smoke_test.dart

@@ -40,6 +40,11 @@ void main() {
     await tester.pumpWidget(
       ProviderScope(
         overrides: [
+          // The background drain would open a broker connection and hit the
+          // disk on a clock this test controls. Neither is what these tests
+          // measure, and a half-driven timer would leak into the next one.
+          syncBootstrapProvider.overrideWith((ref) async {}),
+
           // Catalogue reads come from the in-memory cache and resolve on the
           // spot, but these four go to SQLite. Real disk I/O cannot be driven
           // by the fake clock a widget test runs on: sqflite's own lock-warning