diff --git a/android/app/src/main/AndroidManifest.xml b/android/app/src/main/AndroidManifest.xml
index c0dd590c..22b93dbd 100644
--- a/android/app/src/main/AndroidManifest.xml
+++ b/android/app/src/main/AndroidManifest.xml
@@ -74,6 +74,37 @@
android:label="Edge"
android:name=".EdgeApplication"
android:icon="@mipmap/launcher_icon">
+
+
+
+
+
+
+ 0, so an unknown need leaves it empty
+ // instead of filling against a fabricated 8h denominator.
+ val needMin = w.readInt(prefs, "sleep_need_min", -1)
val hrv = w.readInt(prefs, "hrv", -1)
val hrvBaseline = w.readInt(prefs, "hrv_baseline", -1)
diff --git a/ios/OpenStrapWatch Watch App/WatchMetrics.swift b/ios/OpenStrapWatch Watch App/WatchMetrics.swift
index 93f46414..60cccd63 100644
--- a/ios/OpenStrapWatch Watch App/WatchMetrics.swift
+++ b/ios/OpenStrapWatch Watch App/WatchMetrics.swift
@@ -19,7 +19,7 @@ struct WatchMetrics {
var readiness: Int // 0–100, -1 = none
var strain: Double // 0–21, -1 = none
var sleepMin: Int // minutes asleep, -1 = none
- var needMin: Int // sleep need (min)
+ var needMin: Int // sleep need (min); -1 = none — never fabricate 8h
var hrv: Int // RMSSD ms, -1 = none
var hrvBaseline: Int // baseline RMSSD ms, -1 = none
var rhr: Int // bpm, -1 = none
@@ -29,7 +29,7 @@ struct WatchMetrics {
var themeDark: Bool // mirror the app's Ember-on-Paper (false) / Char (true)
static let empty = WatchMetrics(
- hasData: false, readiness: -1, strain: -1, sleepMin: -1, needMin: 480,
+ hasData: false, readiness: -1, strain: -1, sleepMin: -1, needMin: -1,
hrv: -1, hrvBaseline: -1, rhr: -1, coachLine: "", battPct: -1, updatedAt: 0,
themeDark: true)
@@ -40,7 +40,7 @@ struct WatchMetrics {
readiness: d?.object(forKey: "readiness") as? Int ?? -1,
strain: d?.object(forKey: "strain") as? Double ?? -1,
sleepMin: d?.object(forKey: "sleep_min") as? Int ?? -1,
- needMin: d?.object(forKey: "sleep_need_min") as? Int ?? 480,
+ needMin: d?.object(forKey: "sleep_need_min") as? Int ?? -1,
hrv: d?.object(forKey: "hrv") as? Int ?? -1,
hrvBaseline: d?.object(forKey: "hrv_baseline") as? Int ?? -1,
rhr: d?.object(forKey: "rhr") as? Int ?? -1,
diff --git a/ios/OpenStrapWidget/OpenStrapWidget.swift b/ios/OpenStrapWidget/OpenStrapWidget.swift
index b24bfa33..99b57963 100644
--- a/ios/OpenStrapWidget/OpenStrapWidget.swift
+++ b/ios/OpenStrapWidget/OpenStrapWidget.swift
@@ -58,7 +58,7 @@ struct OpenStrapEntry: TimelineEntry {
let readiness: Int // -1 = none (composite 0..100) — the headline
let strain: Double // -1 = none
let sleepMin: Int // -1 = none
- let needMin: Int
+ let needMin: Int // -1 = none (sleep need, min) — never fabricate 8h
let hrv: Int // -1 = none (RMSSD, ms)
let hrvBaseline: Int // -1 = none (personal RMSSD baseline, ms)
let rhr: Int // -1 = none
@@ -106,7 +106,7 @@ private enum Store {
readiness: d?.object(forKey: "readiness") as? Int ?? -1,
strain: d?.object(forKey: "strain") as? Double ?? -1,
sleepMin: d?.object(forKey: "sleep_min") as? Int ?? -1,
- needMin: (d?.object(forKey: "sleep_need_min") as? Int) ?? 480,
+ needMin: (d?.object(forKey: "sleep_need_min") as? Int) ?? -1,
hrv: d?.object(forKey: "hrv") as? Int ?? -1,
hrvBaseline: d?.object(forKey: "hrv_baseline") as? Int ?? -1,
rhr: d?.object(forKey: "rhr") as? Int ?? -1,
@@ -171,7 +171,7 @@ private enum TodayAPI {
let strain = val(daily, "strain") ?? -1
let rhr = val(daily, "resting_hr").map { Int($0.rounded()) } ?? -1
let sleepMin = val(sleep, "duration_min").map { Int($0.rounded()) } ?? -1
- let needMin = val(sleep, "need_min").map { Int($0.rounded()) } ?? 480
+ let needMin = val(sleep, "need_min").map { Int($0.rounded()) } ?? -1
let hrv = (hrvObj?["rmssd"] as? NSNumber).map { Int($0.doubleValue.rounded()) } ?? -1
let hrvBase = (hrvObj?["baseline"] as? NSNumber).map { Int($0.doubleValue.rounded()) } ?? -1
diff --git a/ios/Runner/Info.plist b/ios/Runner/Info.plist
index e6ad6b44..3cfe2647 100644
--- a/ios/Runner/Info.plist
+++ b/ios/Runner/Info.plist
@@ -4,6 +4,31 @@
CADisableMinimumFrameDurationOnPhone
+
+ FIREBASE_ANALYTICS_COLLECTION_ENABLED
+
+ FirebaseCrashlyticsCollectionEnabled
+
+ firebase_performance_collection_enabled
+
+
+ GOOGLE_ANALYTICS_DEFAULT_ALLOW_AD_PERSONALIZATION_SIGNALS
+
+ GOOGLE_ANALYTICS_IDFV_COLLECTION_ENABLED
+ CFBundleDevelopmentRegion$(DEVELOPMENT_LANGUAGE)CFBundleDisplayName
diff --git a/lib/app.dart b/lib/app.dart
index f0b35985..45bba860 100644
--- a/lib/app.dart
+++ b/lib/app.dart
@@ -6,6 +6,7 @@ import 'package:provider/provider.dart';
import 'ai/briefing.dart';
import 'coach/coach_config.dart';
+import 'notify/notification_service.dart';
import 'notify/tap_router.dart';
import 'state/app_state.dart';
import 'state/prefs.dart';
@@ -67,6 +68,12 @@ class _OpenStrapAppState extends State with WidgetsBindingObserver
void didChangeAppLifecycleState(AppLifecycleState state) {
final app = context.read();
if (state == AppLifecycleState.resumed) {
+ // The user may have flipped our notification switch either way in OS
+ // Settings while we were backgrounded. Drop the cached authorization
+ // decision so the next present/schedule re-reads reality — a denial used
+ // to latch for the whole process, silencing every notification and
+ // scheduled reminder until a full app restart.
+ NotificationService.instance.invalidatePermissionCache();
app.maybeFinishFromLiveActivity();
unawaited(app.maybeStopBreathingFromLiveActivity());
app.refreshAppStatus(); // re-check OTA + admin banner on every foreground
diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart
index 9659e7d2..ed063a64 100644
--- a/lib/ble/ble_engine.dart
+++ b/lib/ble/ble_engine.dart
@@ -129,17 +129,6 @@ int countBurstTrafficPackets({
unknownCount;
}
-@visibleForTesting
-int nextBurstStablePollStreak({
- required bool queueEmpty,
- required int currentCount,
- required int previousCount,
- required int stableStreak,
-}) {
- if (!queueEmpty) return 0;
- return currentCount == previousCount ? (stableStreak + 1) : 0;
-}
-
@visibleForTesting
bool shouldPauseMaintenanceTraffic({required bool offloadActive}) =>
offloadActive;
@@ -427,30 +416,64 @@ class BleEngine {
/// harmless once we own the band).
Future _claimBand() async {
final other = _bandOwner;
- if (other != null && !identical(other, this)) {
- if (isBackgroundDrainer) {
+ final incumbentPresent = other != null && !identical(other, this);
+ final decision = BandClaimPolicy.decide(
+ incumbentPresent: incumbentPresent,
+ // LIVENESS, not just non-nullness: a claim held by an engine that has no
+ // session (its connect threw, or its link went down) is a STALE claim,
+ // and honouring it wedged every later background drain for the whole
+ // process lifetime ("strap not reachable this cycle", forever).
+ incumbentLive: incumbentPresent && other.holdsBandLink,
+ isBackgroundDrainer: isBackgroundDrainer,
+ );
+ switch (decision) {
+ case BandClaimDecision.yieldToOwner:
_log(
- 'band already owned by the foreground session — background drain '
+ 'band already owned by a live foreground session — background drain '
'yielding (avoids duplicate ACKs on the same offload).',
);
return false;
- }
- _log('preempting a background drain to take the foreground session.');
- try {
- await other
- .disconnect()
- .timeout(const Duration(seconds: 10));
- } on TimeoutException {
- _log('preempted engine teardown timed out after 10s — proceeding '
- 'with the foreground connect anyway.');
- } catch (e) {
- _log('preempted engine teardown failed ($e) — proceeding.');
- }
+ case BandClaimDecision.preemptThenClaim:
+ _log('preempting a background drain to take the foreground session.');
+ try {
+ await other!.disconnect().timeout(const Duration(seconds: 10));
+ } on TimeoutException {
+ _log('preempted engine teardown timed out after 10s — proceeding '
+ 'with the foreground connect anyway.');
+ } catch (e) {
+ _log('preempted engine teardown failed ($e) — proceeding.');
+ }
+ break;
+ case BandClaimDecision.claim:
+ if (incumbentPresent) {
+ _log('taking over a STALE band claim (the previous owner has no '
+ 'live link).');
+ }
+ break;
}
_bandOwner = this;
return true;
}
+ /// Whether this engine actually holds (or is actively bringing up) a BLE
+ /// link — the liveness test [BandClaimPolicy] uses on the incumbent owner.
+ /// A session object exists from the moment `_doConnect` starts, so a connect
+ /// still in flight correctly counts as live; every failure path nulls the
+ /// session and drops the phase to idle/error before returning.
+ bool get holdsBandLink =>
+ _session != null &&
+ _phase != BleConnState.idle &&
+ _phase != BleConnState.error;
+
+ /// Test-only view of the process-wide single-owner claim.
+ @visibleForTesting
+ static bool get bandClaimed => _bandOwner != null;
+
+ /// Test-only reset of the process-wide claim (static state otherwise leaks
+ /// across test cases).
+ @visibleForTesting
+ static void resetBandClaimForTest() => _bandOwner = null;
+
void _releaseBand() {
if (identical(_bandOwner, this)) _bandOwner = null;
}
@@ -567,7 +590,7 @@ class BleEngine {
// Historical-offload bookkeeping. A controller is live for the whole connection
// (we keep ACKing HISTORY_END markers as they arrive, even after the first
// HISTORY_COMPLETE — a later strap-triggered offload reuses it).
- _DrainController? _drain;
+ DrainController? _drain;
bool _liveEnabled = false;
// Background live downgrade: only the compact realtime-HR stream is armed
// (no high-rate R10/R11 + IMU + optical flood). Set by [enableHrOnlyLive].
@@ -639,6 +662,9 @@ class BleEngine {
// Lifetime count of GET_DATA_RANGE reads rejected by isCorruptFutureRtc —
// see the range_oldest/range_newest handler below.
int _corruptDataRangeCount = 0;
+ // Lifetime count of GET_CLOCK `clock_epoch` reads rejected by the same gate
+ // (ClockPolicy.acceptsClockRead) — see the clock_epoch handler below.
+ int _corruptClockReadCount = 0;
DateTime? _bondTime; // when the handshake completed (bond confirmed)
DateTime? _armTime; // when live (R10/R11) streams were last armed
int _autoContinueCount = 0; // consecutive auto-continues this connection
@@ -814,6 +840,10 @@ class BleEngine {
// recovery already happens automatically at the DB layer.
'counter_regressions_total': _counterRegression.regressions,
'corrupt_data_ranges_total': _corruptDataRangeCount,
+ 'corrupt_clock_reads_total': _corruptClockReadCount,
+ // Bursts whose open chunk was discarded un-committed and whose HISTORY_END
+ // token was therefore refused (see BurstTrimGuard).
+ 'poisoned_bursts_total': _drain?.poisonedBursts ?? 0,
'history_requests': _historyRequests,
'history_completions': _historyCompletions,
'successful_bursts': _successfulBursts,
@@ -922,9 +952,35 @@ class BleEngine {
if (!await _claimBand()) return false;
// Any prior session is dead to us now — tear it down before a new one.
await _teardownSession(intentional: true);
- return _doConnect(device);
+ try {
+ return await _doConnect(device);
+ } catch (e) {
+ // _doConnect guards its own known failure modes, but anything thrown
+ // OUTSIDE those guards (e.g. the connectionState subscription setup, which
+ // runs before the first try block) used to escape with the band claim
+ // still held and a half-built session left in `connecting` — which
+ // [BandClaimPolicy] would then read as a LIVE incumbent, permanently
+ // starving every later background drain. Route every escape through the
+ // one failure exit.
+ _log('connect failed (unhandled): $e');
+ await _failConnect();
+ return false;
+ }
});
+ /// Common exit for every failed-connect path: tear the half-built session
+ /// down, drop to `idle`, AND RELEASE THE BAND CLAIM.
+ ///
+ /// [_claimBand] runs BEFORE the link is up, so a connect that threw used to
+ /// leave `_bandOwner` pointing at an engine with no link — and only
+ /// `disconnect()` ever released it, which nothing calls on this path. Every
+ /// later background drain then saw a non-null owner and yielded forever.
+ Future _failConnect() async {
+ await _teardownSession(intentional: true);
+ _releaseBand();
+ _setPhase(BleConnState.idle);
+ }
+
Future _doConnect(BluetoothDevice device) async {
state.address = device.remoteId.str;
_setPhase(BleConnState.connecting);
@@ -958,8 +1014,7 @@ class BleEngine {
);
} catch (e) {
_log('connect failed: $e');
- await _teardownSession(intentional: true);
- _setPhase(BleConnState.idle);
+ await _failConnect();
return false;
}
@@ -1026,6 +1081,7 @@ class BleEngine {
if (!session.connected) {
_log('connect: link dropped during setup.');
+ await _failConnect();
return false;
}
@@ -1039,8 +1095,7 @@ class BleEngine {
}
if (svc == null) {
_log('Harvard service not found on device.');
- await _teardownSession(intentional: true);
- _setPhase(BleConnState.idle);
+ await _failConnect();
return false;
}
BluetoothCharacteristic? find(String prefix) {
@@ -1059,8 +1114,7 @@ class BleEngine {
events == null ||
data == null) {
_log('Missing one or more Harvard characteristics.');
- await _teardownSession(intentional: true);
- _setPhase(BleConnState.idle);
+ await _failConnect();
return false;
}
@@ -1144,7 +1198,7 @@ class BleEngine {
// fire INIT — which triggers the historical flood. Historical + live records
// then arrive on the same subscription; HISTORY_END markers are committed
// (raw+samples+cursor, atomically) BEFORE we ACK, so the offload is resumable.
- _drain = _DrainController(
+ _drain = DrainController(
onRecord: _storeRecord,
onRecordsBatch: onRecordsBatch == null ? null : _storeRecordsBatch,
onCommit: onCommitBatch == null ? null : _commitBatch,
@@ -1159,8 +1213,7 @@ class BleEngine {
return true;
} catch (e) {
_log('connect setup failed: $e');
- await _teardownSession(intentional: true);
- _setPhase(BleConnState.idle);
+ await _failConnect();
return false;
}
}
@@ -1313,7 +1366,7 @@ class BleEngine {
_lastRx = DateTime.now();
for (final frame in session.asm[role]!.feed(chunk)) {
if (frame.valid) {
- _onFrame(role, frame);
+ _onFrame(role, frame, session);
} else {
// Previously silent: a degrading radio corrupting frames looked
// identical to a healthy one everywhere. Now counted (surfaced in
@@ -1339,7 +1392,10 @@ class BleEngine {
// ── link-down handling (drives reconnect via the caller's contract) ─────────────
void _onLinkDown(_Session session) {
- if (_session != session) return; // a stale session's stream
+ if (LinkDownPolicy.evaluate(sessionIsCurrent: _session == session) ==
+ LinkDownAction.ignoreStaleSession) {
+ return; // a stale session's stream
+ }
final wasIntentional = session.intentionalClose;
session.connected = false;
// A drain in flight must complete (with linkDown) immediately, not run out
@@ -1359,6 +1415,26 @@ class BleEngine {
// through the same single-flight connect, so there's still exactly one path.
_setOffloadActive(false);
_setPhase(BleConnState.idle);
+ // TEAR THE SESSION DOWN NOW. This used to happen ONLY on the next
+ // connect()/disconnect() — which never comes when BondRefusalGiveUp pauses
+ // auto-reconnect (`state.autoReconnectPaused`), so the dead session's five
+ // timers (heartbeat 10 s, keep-alive 30 s, periodic backfill 900 s, idle
+ // watchdog, historical retry) kept firing into a dead characteristic
+ // forever and its four onValueReceived subscriptions stayed registered —
+ // one more full set leaked on every drop. Deferred off this notification
+ // callback (we are inside one of the very subscriptions being cancelled)
+ // and non-intentional, so no redundant device.disconnect() is issued.
+ unawaited(
+ Future(() async {
+ if (_session != session) return; // a connect already replaced us
+ await _teardownSession(intentional: false);
+ // A claim held with no link is a stale claim (see [BandClaimPolicy]);
+ // releasing it here is what stops a failed foreground link from
+ // wedging every later background drain. The caller's reconnect loop
+ // re-claims through connect(), where foreground still preempts.
+ _releaseBand();
+ }),
+ );
}
/// Feed an UNINTENTIONAL disconnect to the cross-reconnect detectors. A timeout
@@ -1416,7 +1492,12 @@ class BleEngine {
static const Duration _serviceDiscoveryTimeout = Duration(seconds: 15);
static const Duration _notifySetupTimeout = Duration(seconds: 15);
- Future _write(Uint8List raw) {
+ /// [owner] pins the write to ONE session. Without it a write queued by a
+ /// long-parked drain (a big commit, then up to ~25 s of ACK retries) lands on
+ /// whatever session happens to be current when the write chain reaches it —
+ /// i.e. an OLD connection's batch-ACK, with a re-used sync seq, written onto
+ /// a BRAND NEW link. Every offload write passes its owning session.
+ Future _write(Uint8List raw, {_Session? owner}) {
final session = _session;
final completer = Completer();
_writeChain = _writeChain.then((_) async {
@@ -1427,6 +1508,10 @@ class BleEngine {
_log('write skipped: link not ready.');
return;
}
+ if (owner != null && !identical(owner, session)) {
+ _log('write skipped: it belongs to a session that is no longer live.');
+ return;
+ }
// allowLongWrite: the rich SET_ALARM_TIME frame is 32B — the only write
// that exceeds the 20B ATT limit of a default (23B) MTU. Without a long
// write, flutter_blue_plus throws "value > mtu-3" if the negotiated MTU
@@ -1454,16 +1539,17 @@ class BleEngine {
/// false only after every attempt failed — the caller must then bounce the
/// link (the chunk is already durably committed; the band re-delivers it next
/// session and the decoded store dedups by REPLACE).
- Future _writeAckVerified(Uint8List ack) async {
+ Future _writeAckVerified(Uint8List ack, _Session session) async {
var failures = 0;
while (true) {
- if (await _write(ack)) return true;
+ if (_sessionIsStale(session)) return false;
+ if (await _write(ack, owner: session)) return true;
failures++;
if (!ackRetryPolicy.shouldRetry(failures)) return false;
_log('[SYNC] batch-ACK write failed (attempt $failures/'
'${ackRetryPolicy.maxAttempts}) — retrying.');
await Future.delayed(ackRetryPolicy.delayFor(failures));
- if (_session?.connected != true) return false;
+ if (_sessionIsStale(session)) return false;
}
}
@@ -1563,11 +1649,22 @@ class BleEngine {
}
// ── frame handling ─────────────────────────────────────────────────────────────
- void _onFrame(String role, Frame frame) {
+ void _onFrame(String role, Frame frame, _Session session) {
final pt = frame.packetType;
- if (role == 'data' &&
- (pt == PacketType.metadata || pt == PacketType.historicalData)) {
- _enqueueOffloadFrame(frame);
+ // Metadata ALWAYS takes the serialized queue, whatever characteristic it
+ // was reassembled on. It used to take the queue only on the `data` role;
+ // metadata off `cmd_from`/`events` was fired unawaited on the immediate
+ // path — the one route that could run a HISTORY_END handler CONCURRENTLY
+ // with the queued drain, i.e. two handlers on the same DrainController,
+ // where one snapshots an empty buffer and writes its ACK before the
+ // other's commit is durable. See [FrameRoutePolicy].
+ final route = FrameRoutePolicy.route(
+ isMetadata: pt == PacketType.metadata,
+ isHistorical: pt == PacketType.historicalData,
+ isDataRole: role == 'data',
+ );
+ if (route == FrameRoute.serializedQueue) {
+ _enqueueOffloadFrame(frame, session);
return;
}
_processImmediateFrame(frame);
@@ -1575,10 +1672,8 @@ class BleEngine {
void _processImmediateFrame(Frame frame) {
final pt = frame.packetType;
- if (pt == PacketType.metadata) {
- unawaited(_handleSyncMarker(frame));
- return;
- }
+ // NOTE: metadata never reaches here — [FrameRoutePolicy] routes every
+ // metadata frame to the serialized offload queue regardless of role.
// LIVE streams: realtime HR/RR (0x28), realtime R10 (0x2B), IMU (0x33).
// EPHEMERAL — these are the high-rate flood (~655 MB/day) and the daily
// metrics need ONLY the 1 Hz historical substrate (0x2F / R24). We do NOT
@@ -1639,17 +1734,28 @@ class BleEngine {
_absorbState(decoded);
}
- void _enqueueOffloadFrame(Frame frame) {
+ void _enqueueOffloadFrame(Frame frame, _Session session) {
+ if (_session != session || !session.connected) return; // stale session
_offloadFrames.add(frame);
if (_offloadActive || frame.packetType == PacketType.historicalData) {
_setOffloadActive(true);
}
if (_drainingOffloadFrames) return;
_drainingOffloadFrames = true;
- unawaited(_drainOffloadFrames());
- }
-
- Future _drainOffloadFrames() async {
+ unawaited(_drainOffloadFrames(session));
+ }
+
+ /// The ONE serialized offload-frame processor. [session] is the connection
+ /// that started this loop; it is re-checked around every await because a
+ /// drain can be parked for a long time (a multi-second large-batch commit,
+ /// then up to ~25 s of ACK retries) and `_teardownSession` clears
+ /// `_drainingOffloadFrames` underneath it. Without the guard a SECOND
+ /// drainer starts on the new session while this one is still alive — two
+ /// loops on the same DrainController, so one commit() snapshots an empty
+ /// buffer and its ACK can be written before the other's commit is durable —
+ /// and this stale loop would go on to write the OLD connection's token onto
+ /// the NEW link, tearing down a healthy session when that write failed.
+ Future _drainOffloadFrames(_Session session) async {
// this used to have no try/finally, so if anything inside the loop threw
// (a couple of the ledger writes in _handleSyncMarker weren't guarded),
// _drainingOffloadFrames never got reset back to false and every future
@@ -1657,6 +1763,7 @@ class BleEngine {
// sync progress until a full disconnect/reconnect.
try {
while (_offloadFrames.isNotEmpty) {
+ if (_sessionIsStale(session)) return;
final count = _offloadFrames.length > 64 ? 64 : _offloadFrames.length;
final batch = _offloadFrames.sublist(0, count);
_offloadFrames.removeRange(0, count);
@@ -1665,8 +1772,9 @@ class BleEngine {
// no Timer churn at flood rates. Markers re-arm it in _handleSyncMarker.
_armIdleWatchdog();
for (final frame in batch) {
+ if (_sessionIsStale(session)) return;
if (frame.packetType == PacketType.metadata) {
- await _handleSyncMarker(frame);
+ await _handleSyncMarker(frame, session);
} else {
_ingestHistoricalFrame(frame);
}
@@ -1676,10 +1784,18 @@ class BleEngine {
}
}
} finally {
- _drainingOffloadFrames = false;
+ // Only the CURRENT session's loop may clear the flag. A stale loop
+ // unwinding after the new session's drainer already started would
+ // otherwise re-open the door to a second concurrent drainer.
+ if (_session == session) _drainingOffloadFrames = false;
}
}
+ /// True once [session] is no longer the engine's live session — the guard
+ /// every long-parked offload callback shares.
+ bool _sessionIsStale(_Session session) =>
+ _session != session || !session.connected;
+
/// THE single historical-record processing path — used by BOTH the queued
/// offload drain (real traffic) and the immediate fallback. Decode → gate
/// (plausibility + frontier via [RecordGate]) → storage enqueue. Keeping one
@@ -1846,29 +1962,48 @@ class BleEngine {
if (f.containsKey('clock_epoch')) {
final dev = f['clock_epoch'] as int;
final wall = DateTime.now().millisecondsSinceEpoch ~/ 1000;
- _clockRef = ClockRef(device: dev, wall: wall);
- _log('Clock correlated: device=$dev wall=$wall (drift=${wall - dev}s).');
- // Re-issue SET_CLOCK if the strap RTC has drifted > 1 day or is unset —
- // but BOUND the retries: setClock() reads the clock back, so an unbounded
- // re-issue on a firmware that never latches either payload form would spin
- // SET_CLOCK/GET_CLOCK forever. Historical records carry their own embedded
- // unix time regardless, so giving up after a few tries is safe.
- if (ClockPolicy.shouldSetClock(dev, wall)) {
- if (_clockCorrectTries < 3) {
- _clockCorrectTries++;
- _log(
- 'Clock drift over policy — re-issuing SET_CLOCK '
- '(attempt $_clockCorrectTries/3).',
- );
- unawaited(setClock());
+ // SANITY GATE, mirroring the one `range_newest` gets below. An
+ // implausibly far-future `clock_epoch` yields a large NEGATIVE driftSec,
+ // and setAlarm arms at `when - driftSec` — years out, where the alarm
+ // silently never fires — while the bounded SET_CLOCK retry budget is
+ // spent chasing a value that was never real. Reject the read: with no
+ // correlation the alarm falls back to the raw wall epoch. connect()
+ // already issues an unconditional SET_CLOCK, and the periodic re-verify
+ // re-reads, so a genuinely-wrong RTC still gets corrected.
+ if (!ClockPolicy.acceptsClockRead(dev, wall)) {
+ _corruptClockReadCount++;
+ _log(
+ '[SYNC] GET_CLOCK clock_epoch=$dev is implausibly far in the future '
+ '— treating as a corrupt strap RTC read; NOT correlating the strap '
+ 'clock (alarms fall back to the raw wall epoch) '
+ '(corrupt_clock_reads_total=$_corruptClockReadCount).',
+ );
+ } else {
+ _clockRef = ClockRef(device: dev, wall: wall);
+ _log('Clock correlated: device=$dev wall=$wall (drift=${wall - dev}s).');
+ // Re-issue SET_CLOCK if the strap RTC has drifted > 1 day or is unset —
+ // but BOUND the retries: setClock() reads the clock back, so an
+ // unbounded re-issue on a firmware that never latches either payload
+ // form would spin SET_CLOCK/GET_CLOCK forever. Historical records carry
+ // their own embedded unix time regardless, so giving up after a few
+ // tries is safe.
+ if (ClockPolicy.shouldSetClock(dev, wall)) {
+ if (_clockCorrectTries < 3) {
+ _clockCorrectTries++;
+ _log(
+ 'Clock drift over policy — re-issuing SET_CLOCK '
+ '(attempt $_clockCorrectTries/3).',
+ );
+ unawaited(setClock());
+ } else {
+ _log(
+ 'Clock still off after 3 SET_CLOCK attempts — giving up; '
+ 'firmware may not accept our payload length.',
+ );
+ }
} else {
- _log(
- 'Clock still off after 3 SET_CLOCK attempts — giving up; '
- 'firmware may not accept our payload length.',
- );
+ _clockCorrectTries = 0; // latched — reset for the next drift episode
}
- } else {
- _clockCorrectTries = 0; // latched — reset for the next drift episode
}
}
if (f.containsKey('range_oldest') && f.containsKey('range_newest')) {
@@ -2007,7 +2142,84 @@ class BleEngine {
}
}
- Future _handleSyncMarker(Frame frame) async {
+ /// Refuse to echo a HISTORY_END token, for one of the [TrimAckVerdict]
+ /// blocked reasons. The band keeps the chunk and re-delivers it on the next
+ /// offload; re-delivery is dedup-safe (decoded rows REPLACE by rec_ts, raw
+ /// rows key on the record hex).
+ Future _refuseHistoryEndTrim(
+ TrimAckVerdict verdict, {
+ required DrainController d,
+ required _Session session,
+ required String tokenHex,
+ required int? batchId,
+ }) async {
+ switch (verdict) {
+ case TrimAckVerdict.send:
+ return;
+ case TrimAckVerdict.blockedStaleSession:
+ // Deliberately no ledger write and no teardown: we may be mid-teardown
+ // already, and every side effect here would land on a session that is
+ // not ours.
+ _log(
+ '[SYNC] HISTORY_END token=$tokenHex belongs to a session that is no '
+ 'longer live — NOT ACKing. Writing it would put an old connection\'s '
+ 'token (with a re-used sync seq) onto the new link. The band '
+ 're-delivers this chunk next offload.',
+ );
+ return;
+ case TrimAckVerdict.blockedDiscardedBurst:
+ // The idle watchdog abandoned this burst's buffered records. Persist
+ // anything that arrived since (dedup-safe) but WITHOUT the token, so
+ // the trim cursor never claims a chunk we threw away.
+ await d.commit(null);
+ _log(
+ '[SYNC] HISTORY_END token=$tokenHex terminates a DISCARDED burst '
+ '(its open chunk was abandoned un-committed) — NOT ACKing, so the '
+ 'band cannot trim the records we dropped. It re-delivers them next '
+ 'offload.',
+ );
+ await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry(
+ chunkId: 'batch:$tokenHex',
+ kind: 'historical_batch',
+ status: 'trim_refused',
+ lastError: 'discarded_burst',
+ metaPatch: {'batch_id': batchId, 'records': d.records},
+ ));
+ return;
+ case TrimAckVerdict.blockedCommitFailed:
+ // THE safe-trim invariant. The transaction rolled back, so the cursor
+ // did not advance and the rows are not durable — commit() re-buffered
+ // them rather than losing them. Never ACK here: that is exactly the
+ // path where records existed nowhere, permanently and silently.
+ _log(
+ '[SYNC] DURABLE COMMIT FAILED for token=$tokenHex — NOT ACKing (the '
+ 'band must keep this chunk). Records were re-buffered; bouncing the '
+ 'link so the next session retries the commit from a clean batch.',
+ );
+ await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry(
+ chunkId: 'batch:$tokenHex',
+ kind: 'historical_batch',
+ status: 'commit_failed',
+ lastError: 'durable_commit_failed',
+ metaPatch: {'batch_id': batchId, 'records': d.records},
+ ));
+ // Bounce rather than retry in place: a commit that failed on a large
+ // batch (the observed production OOM inside commitSyncBatch) only gets
+ // bigger if we keep appending to the same buffer. A reconnect drops
+ // the buffer, and the band re-delivers from its un-advanced cursor.
+ if (!_sessionIsStale(session)) {
+ unawaited(
+ _teardownSession(intentional: false).then((_) {
+ _setPhase(BleConnState.idle); // caller's reconnect loop takes over
+ }),
+ );
+ }
+ return;
+ }
+ }
+
+ Future _handleSyncMarker(Frame frame, _Session session) async {
+ if (_sessionIsStale(session)) return;
final m = parseMetadata(frame.inner);
if (m == null) return;
_armIdleWatchdog();
@@ -2033,6 +2245,9 @@ class BleEngine {
if (m.sub == SyncMeta.historyEnd && m.token != null) {
final d = _drain;
if (d == null) return;
+ final tokenHex = m.token!
+ .map((b) => b.toRadixString(16).padLeft(2, '0'))
+ .join();
if (!_offloadActive) {
_setHpsTerminal(
_HpsTerminalKind.metadataWhileNotSyncing,
@@ -2040,7 +2255,25 @@ class BleEngine {
drain: d,
);
}
- await _awaitBurstTrafficSettle(d);
+ // PRE-COMMIT GATE. Refuse the trim before anything else touches the link
+ // or the durable cursor: a stale session must not be written to at all,
+ // and a poisoned burst's records are already gone — there is nothing
+ // this token may legitimately trim.
+ final preVerdict = TrimAckPolicy.evaluate(
+ sessionCurrent: !_sessionIsStale(session),
+ burstDiscarded: d.burstDiscarded,
+ commitDurable: true,
+ );
+ if (preVerdict != TrimAckVerdict.send) {
+ await _refuseHistoryEndTrim(
+ preVerdict,
+ d: d,
+ session: session,
+ tokenHex: tokenHex,
+ batchId: m.batchId,
+ );
+ return;
+ }
final expected = m.expectedPacketCount;
// Records the plausibility gate silently rejected THIS burst (stale/
// wandering-clock block — by design, "neither stored nor counted",
@@ -2098,9 +2331,6 @@ class BleEngine {
}
_successfulBursts++;
_mergeValidatedBurst(d);
- final tokenHex = m.token!
- .map((b) => b.toRadixString(16).padLeft(2, '0'))
- .join();
final r = d.bufferedRecTsRange;
_log(
'[SYNC] HistoryEnd batch=${m.batchId} records=${d.records} '
@@ -2114,18 +2344,40 @@ class BleEngine {
// once the ACK is link-layer confirmed, so a crash before the ACK
// re-delivers the chunk. Echo the 8-byte slice the band acks verbatim —
// a mangled echo is the "Groundhog Day" re-flood bug.
- await d.commit(m.token); // raw + samples + strap_trim cursor, atomic
+ final durable = await d.commit(m.token); // raw+samples+cursor, atomic
+ // RE-GATE after the await. commit() can take seconds on a large batch —
+ // long enough for the link to die under us — and it now reports whether
+ // the transaction actually became durable instead of swallowing the
+ // exception. Either way the ACK is refused, which is the whole point:
+ // the ACK is what makes the band trim its flash.
+ final verdict = TrimAckPolicy.evaluate(
+ sessionCurrent: !_sessionIsStale(session),
+ burstDiscarded: d.burstDiscarded,
+ commitDurable: durable,
+ );
+ if (verdict != TrimAckVerdict.send) {
+ await _refuseHistoryEndTrim(
+ verdict,
+ d: d,
+ session: session,
+ tokenHex: tokenHex,
+ batchId: m.batchId,
+ );
+ return;
+ }
final ack = buildHistoryResultOk(_seq.nextSync(), m.token!);
_log(
'[SYNC] ACK frame='
'${ack.map((b) => b.toRadixString(16).padLeft(2, '0')).join()}',
);
- // VERIFIED ACK (retried): the cursor above is already durably committed,
- // so a silently-failed ACK write would leave the band never trimming and
- // re-flooding the same chunk forever. On persistent failure bounce the
- // link — the committed data is safe, and the next session's re-delivery
- // is dedup-safe (decoded rows REPLACE by rec_ts).
- if (!await _writeAckVerified(ack)) {
+ // VERIFIED ACK (retried). We only reach here when [TrimAckPolicy]
+ // returned `send` — i.e. the commit above REPORTED durable (it no longer
+ // swallows its exception) and the session is still ours — so the cursor
+ // genuinely is committed. A silently-failed ACK write would leave the
+ // band never trimming and re-flooding the same chunk forever. On
+ // persistent failure bounce the link — the committed data is safe, and
+ // the next session's re-delivery is dedup-safe (rows REPLACE by rec_ts).
+ if (!await _writeAckVerified(ack, session)) {
// Real per-chunk ledger row, keyed by the token itself — previously
// every ledger write here collapsed onto one shared 'capture' row,
// so a token that kept failing ACROSS reconnects (the "Groundhog
@@ -2168,11 +2420,16 @@ class BleEngine {
'${ackRetryPolicy.maxAttempts} attempts (token=$tokenHex, '
'failures_for_this_token=$failCount) — bouncing the link; data '
'is committed and the band will re-send.');
- unawaited(
- _teardownSession(intentional: false).then((_) {
- _setPhase(BleConnState.idle); // caller's reconnect loop takes over
- }),
- );
+ // ONLY bounce a session that is still OURS. _writeAckVerified also
+ // returns false when the session died under it, and tearing down then
+ // would kill the healthy session that replaced it.
+ if (!_sessionIsStale(session)) {
+ unawaited(
+ _teardownSession(intentional: false).then((_) {
+ _setPhase(BleConnState.idle); // caller's reconnect loop takes over
+ }),
+ );
+ }
return;
}
_chunkFailures.recordSuccess(tokenHex);
@@ -2215,7 +2472,24 @@ class BleEngine {
// Backlog fully handed over (cursor is now at the live edge). Commit the tail
// and KEEP LISTENING — live records continue on the same subscription. We do
// NOT ACK a HISTORY_COMPLETE and we do NOT switch modes.
- await d.commit(null); // tail (no new token) — persist anything buffered
+ final tailDurable = await d.commit(null); // tail — no new trim token
+ if (_sessionIsStale(session)) return; // link died under the tail commit
+ if (!tailDurable) {
+ // No ACK is written for a HISTORY_COMPLETE, so nothing was trimmed and
+ // no data is at risk — but the tail is NOT durable yet. commit() left
+ // the records buffered, so the next commit (the next burst's
+ // HISTORY_END, or the flush on teardown) re-attempts them.
+ _log(
+ '[SYNC] HistoryComplete tail commit FAILED — ${d.bufferedRecords} '
+ 'records stay buffered for the next commit. Nothing was trimmed '
+ '(HISTORY_COMPLETE is never ACKed), so no data is at risk.',
+ );
+ await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry(
+ status: 'tail_commit_failed',
+ lastError: 'durable_commit_failed',
+ metaPatch: {'records_seen': d.records},
+ ));
+ }
d.onComplete();
_historyCompletions++;
_session?.idleWatchdog?.cancel();
@@ -2241,51 +2515,6 @@ class BleEngine {
}
}
- Future _awaitBurstTrafficSettle(_DrainController d) async {
- const poll = Duration(milliseconds: 60);
- const budget = Duration(milliseconds: 720);
- const requiredStablePolls = 3;
- final deadline = DateTime.now().add(budget);
- var previousCount = d.currentBurstPacketCount;
- var waitedMs = 0;
- var stablePolls = 0;
- while (DateTime.now().isBefore(deadline)) {
- if (_offloadFrames.isNotEmpty) {
- await Future.delayed(poll);
- waitedMs += poll.inMilliseconds;
- previousCount = d.currentBurstPacketCount;
- stablePolls = 0;
- continue;
- }
- await Future.delayed(poll);
- waitedMs += poll.inMilliseconds;
- final currentCount = d.currentBurstPacketCount;
- stablePolls = nextBurstStablePollStreak(
- queueEmpty: _offloadFrames.isEmpty,
- currentCount: currentCount,
- previousCount: previousCount,
- stableStreak: stablePolls,
- );
- if (_offloadFrames.isEmpty && stablePolls >= requiredStablePolls) {
- if (waitedMs > 0) {
- _log(
- '[SYNC] history-end settle: waited=${waitedMs}ms '
- 'traffic=$currentCount historical=${d.currentBurstHistoricalPacketCount} '
- 'stable_polls=$stablePolls',
- );
- }
- return;
- }
- previousCount = currentCount;
- }
- _log(
- '[SYNC] history-end settle timed out at ${waitedMs}ms '
- 'traffic=${d.currentBurstPacketCount} '
- 'historical=${d.currentBurstHistoricalPacketCount} '
- 'stable_polls=$stablePolls',
- );
- }
-
// ── post-offload policy: empty-sync, stuck-strap, auto-continue ──────────────
Future _onOffloadFinished({required bool complete}) async {
final d = _drain;
@@ -2735,7 +2964,7 @@ class BleEngine {
void _setHpsTerminal(
_HpsTerminalKind kind, {
String? reason,
- _DrainController? drain,
+ DrainController? drain,
}) {
final d = drain ?? _drain;
_lastHpsTerminal = _HpsTerminal(
@@ -2748,7 +2977,7 @@ class BleEngine {
);
}
- void _mergeValidatedBurst(_DrainController d) {
+ void _mergeValidatedBurst(DrainController d) {
final burstCounts = d.burstStats.dataPacketCountsByRevision;
final mergedCounts = {
..._sessionPacketCounts.dataPacketCountsByRevision,
@@ -2846,14 +3075,19 @@ class BleEngine {
/// [awaitComplete] future that resolves when the band signals HISTORY_COMPLETE (or
/// the link drops / a safety timeout elapses), so a caller can block until the
/// backlog is fully handed over without disturbing the continuous listen.
-class _DrainController {
+///
+/// ENGINE-INTERNAL. Public only so the safe-trim invariant it enforces (a
+/// commit that fails must re-buffer and must NOT let the caller ACK) can be
+/// regression-tested directly without a real band.
+@visibleForTesting
+class DrainController {
final SampleSink onRecord;
final BatchSink? onRecordsBatch;
final CommitSyncBatchSink? onCommit;
final ArchiveSink? onArchive;
final void Function(String) log;
- _DrainController({
+ DrainController({
required this.onRecord,
required this.onRecordsBatch,
required this.onCommit,
@@ -2869,7 +3103,7 @@ class _DrainController {
final List _archives = [];
// Per-burst packet accounting (per-revision counts + sequence gap detection),
// merged into the session totals when a burst validates.
- final _BurstStats burstStats = _BurstStats();
+ final BurstStats burstStats = BurstStats();
int records = 0; // total this connection
int recordsThisOffload = 0; // since the last HISTORY_COMPLETE / rearm
@@ -2908,6 +3142,17 @@ class _DrainController {
bool lastTrimAdvanced = false;
int consecutiveValidationFailures = 0;
+ // Poison latch for the burst currently open (see BurstTrimGuard): once its
+ // records have been discarded un-committed, its straggler HISTORY_END must
+ // never be echoed.
+ final BurstTrimGuard _trimGuard = BurstTrimGuard();
+
+ /// True while the open burst's terminal may NOT be ACKed.
+ bool get burstDiscarded => _trimGuard.discarded;
+
+ /// Bursts poisoned this connection (diagnostics).
+ int get poisonedBursts => _trimGuard.poisonedBursts;
+
bool get _buffering => onCommit != null || onRecordsBatch != null;
int get currentBurstPacketCount => burstStats.totalTrafficPacketCount;
int get currentBurstTrafficCount => burstStats.totalTrafficPacketCount;
@@ -2987,6 +3232,9 @@ class _DrainController {
_linkDown = false;
_lastProgressAt = DateTime.now();
burstStats.reset();
+ // A fresh burst starts un-poisoned: whatever was discarded belonged to the
+ // burst that just ended, and the band will re-deliver it.
+ _trimGuard.beginBurst();
}
void onLinkDown() => _linkDown = true;
@@ -2997,10 +3245,19 @@ class _DrainController {
/// Abandon the buffered-but-not-yet-committed chunk WITHOUT persisting (idle
/// watchdog). These records were never ACKed, so the band re-delivers them on the
/// next offload — dropping them here just avoids ACKing a partial.
+ ///
+ /// POISONS THE OPEN BURST. Discarding alone was not enough: the band had
+ /// already put that burst's HISTORY_END on the wire, and the terminal handler
+ /// went on to commit the (now empty) buffer and echo the token verbatim —
+ /// trimming exactly the records that were just dropped. The poison is
+ /// unconditional (even with an empty buffer, this burst was abandoned) and
+ /// is cleared only by [rearm] / a fresh HISTORY_START.
void discardOpenChunk() {
+ _trimGuard.discardOpenChunk();
if (_raws.isEmpty && _archives.isEmpty) return;
log('discarding ${_raws.length} un-ACKed buffered records + '
- '${_archives.length} archived (idle).');
+ '${_archives.length} archived (idle). This burst\'s HISTORY_END token '
+ 'is now un-ACKable — the band keeps the chunk.');
_raws.clear();
_samples.clear();
_archives.clear();
@@ -3010,10 +3267,24 @@ class _DrainController {
/// ATOMICALLY (via onCommit) and return only once durable — the caller writes
/// the ACK afterwards. Snapshots the buffer so records arriving during the await
/// land in the next commit. Updates [lastTrimAdvanced].
- Future commit(List? token) async {
+ ///
+ /// RETURNS WHETHER THE CHUNK IS ACTUALLY DURABLE. This used to be
+ /// `Future` with the exception swallowed into a log line, while the
+ /// buffer had ALREADY been snapshotted and cleared — so a failed transaction
+ /// left the rows nowhere (buffer cleared, transaction rolled back, cursor
+ /// unadvanced) and the caller went straight on to echo the HISTORY_END trim
+ /// token, telling the band to delete them from its flash. Those records
+ /// existed nowhere, permanently and silently. On failure the buffer is now
+ /// RESTORED (at the front, so arrival order is preserved) and the trim
+ /// bookkeeping rolled back, and the caller MUST NOT ACK — the band keeps the
+ /// chunk and re-delivers it, which is dedup-safe (decoded rows REPLACE by
+ /// rec_ts).
+ Future commit(List? token) async {
final tokenHex = token
?.map((b) => b.toRadixString(16).padLeft(2, '0'))
.join();
+ final previousAckedToken = _lastAckedToken;
+ final previousTrimAdvanced = lastTrimAdvanced;
lastTrimAdvanced = tokenHex != null && tokenHex != _lastAckedToken;
if (tokenHex != null) _lastAckedToken = tokenHex;
final raws = List.from(_raws);
@@ -3028,12 +3299,24 @@ class _DrainController {
} else if (onRecordsBatch != null && raws.isNotEmpty) {
await onRecordsBatch!(raws, samples);
}
+ return true;
} catch (e) {
- log('offload commit error: $e');
+ // Put the snapshot back at the FRONT: records that arrived during the
+ // await are already appended behind it, so this preserves arrival order.
+ _raws.insertAll(0, raws);
+ _samples.insertAll(0, samples);
+ _archives.insertAll(0, archives);
+ // Roll back the trim bookkeeping too — nothing advanced.
+ _lastAckedToken = previousAckedToken;
+ lastTrimAdvanced = previousTrimAdvanced;
+ log('offload commit FAILED ($e) — ${raws.length} records + '
+ '${archives.length} archived re-buffered; the caller MUST NOT ACK '
+ 'this chunk (the band still holds it).');
+ return false;
}
}
- Future flush() => commit(null);
+ Future flush() => commit(null);
/// Resolve once the current offload reaches HISTORY_COMPLETE, the link drops, or
/// [timeout] elapses. Pure waiting — NO abort is ever sent (cutting the offload
@@ -3077,7 +3360,8 @@ class _DrainController {
}
}
-class _BurstStats {
+@visibleForTesting
+class BurstStats {
static const Set _ordinaryHistoricalRevisions = {
7,
9,
@@ -3091,7 +3375,7 @@ class _BurstStats {
};
final Map _dataPacketCountsByRevision = {};
- final Map _sequenceByRevision = {};
+ final Map _sequenceByRevision = {};
int _eventCount = 0;
int _consoleCount = 0;
int _unknownCount = 0;
@@ -3103,8 +3387,8 @@ class _BurstStats {
Map get dataPacketCountsByRevision =>
Map.unmodifiable(_dataPacketCountsByRevision);
- Map get sequenceByRevision =>
- Map.unmodifiable(_sequenceByRevision);
+ Map get sequenceByRevision =>
+ Map.unmodifiable(_sequenceByRevision);
int get eventCount => _eventCount;
int get consoleCount => _consoleCount;
int get unknownCount => _unknownCount;
@@ -3194,7 +3478,7 @@ class _BurstStats {
(_dataPacketCountsByRevision[revision] ?? 0) + 1;
final seq = _sequenceByRevision.putIfAbsent(
revision,
- () => _SequenceState(firstSequence: counter),
+ () => SequenceState(firstSequence: counter),
);
seq.observe(counter);
return;
@@ -3241,8 +3525,9 @@ class _BurstStats {
}
}
-class _SequenceState {
- _SequenceState({required this.firstSequence});
+@visibleForTesting
+class SequenceState {
+ SequenceState({required this.firstSequence});
final int firstSequence;
int? lastSequence;
diff --git a/lib/ble/ble_state.dart b/lib/ble/ble_state.dart
index b55660bf..039dc37c 100644
--- a/lib/ble/ble_state.dart
+++ b/lib/ble/ble_state.dart
@@ -297,6 +297,199 @@ class AckRetryPolicy {
}
}
+/// Why an in-flight HISTORY_END token may (or may not) be echoed back to the
+/// band. See [TrimAckPolicy].
+enum TrimAckVerdict {
+ /// Every precondition holds — echo the verbatim 8-byte token.
+ send,
+
+ /// The session that received this HISTORY_END is no longer the engine's
+ /// current session (the link dropped / was replaced while the handler was
+ /// parked mid-await). Writing now would put an OLD connection's token, with
+ /// a re-used sync seq, onto a BRAND NEW link — and a failed write would tear
+ /// down the healthy new session.
+ blockedStaleSession,
+
+ /// The burst this token terminates had its buffered records DISCARDED
+ /// without ever being committed (idle watchdog / abort-and-retry). Its
+ /// straggler terminal is still in flight; echoing it would trim flash we
+ /// deliberately threw away.
+ blockedDiscardedBurst,
+
+ /// The durable commit did NOT complete. The records are not in the database,
+ /// so the band must keep them.
+ blockedCommitFailed,
+}
+
+/// THE gate on the one irreversible act in the whole offload protocol: echoing
+/// a HISTORY_END continuation token, which is what tells the band it may trim
+/// that chunk out of its flash.
+///
+/// The safe-trim invariant is "every row is durable BEFORE the caller echoes
+/// the HISTORY_END trim token, or none is". Historically only the HAPPY path
+/// honoured it: the commit's exception was swallowed and the caller ACKed
+/// regardless, so a failed transaction (a real production OOM inside
+/// `commitSyncBatch` — see `lib/data/db.dart`) meant the buffer had already
+/// been cleared, the transaction had rolled back, the cursor had not advanced,
+/// and the band was then told to trim — those records existed nowhere,
+/// permanently and silently.
+///
+/// Pure and total: the engine supplies three observations, this decides. The
+/// engine must call it AGAIN after the commit await (the commit can take
+/// seconds on a large batch — long enough for the session to die under it).
+class TrimAckPolicy {
+ const TrimAckPolicy._();
+
+ /// [sessionCurrent] — the receiving session is still the engine's session
+ /// AND still connected.
+ /// [burstDiscarded] — this burst's open chunk was discarded un-committed.
+ /// [commitDurable] — the atomic commit completed (pass `true` when asking
+ /// the pre-commit question "should I even commit this
+ /// token?").
+ static TrimAckVerdict evaluate({
+ required bool sessionCurrent,
+ required bool burstDiscarded,
+ required bool commitDurable,
+ }) {
+ // Order is deliberate: a stale session must be refused before anything
+ // else touches the (new) link, and a poisoned burst must be refused before
+ // the commit result is even considered — its records are already gone.
+ if (!sessionCurrent) return TrimAckVerdict.blockedStaleSession;
+ if (burstDiscarded) return TrimAckVerdict.blockedDiscardedBurst;
+ if (!commitDurable) return TrimAckVerdict.blockedCommitFailed;
+ return TrimAckVerdict.send;
+ }
+}
+
+/// Per-burst poison latch for [TrimAckVerdict.blockedDiscardedBurst].
+///
+/// The idle watchdog abandons the open chunk (N buffered, never-committed
+/// records) on the contract "the band re-delivers them next offload", and the
+/// abort path then sends ABORT_HISTORICAL. But that burst's HISTORY_END is
+/// ALREADY in flight and arrives anyway — and nothing stopped the handler from
+/// committing an empty buffer and echoing the token verbatim, which trims
+/// exactly the records that were just thrown away. The token is unknown at
+/// discard time (it only arrives with the terminal), so the poison is keyed to
+/// the BURST, not the token: a discard poisons the open burst, and only a
+/// fresh HISTORY_START / re-arm clears it.
+class BurstTrimGuard {
+ bool _discarded = false;
+
+ /// Bursts poisoned since construction (diagnostics — a rising count means
+ /// the drain keeps stalling mid-burst).
+ int poisonedBursts = 0;
+
+ /// True while the open burst may NOT be trimmed.
+ bool get discarded => _discarded;
+
+ /// A fresh burst begins (HISTORY_START / re-arm) — nothing lost yet.
+ void beginBurst() => _discarded = false;
+
+ /// The open chunk was abandoned without a durable commit.
+ void discardOpenChunk() {
+ if (_discarded) return;
+ _discarded = true;
+ poisonedBursts++;
+ }
+}
+
+/// What a link-down must do to the session that just died.
+enum LinkDownAction {
+ /// Cancel every timer + subscription the session owns, then surface `idle`.
+ tearDownSession,
+
+ /// The event belongs to a session we already replaced — ignore it entirely.
+ ignoreStaleSession,
+}
+
+/// A dropped link used to only flip a flag and surface the phase; the actual
+/// teardown happened solely on the NEXT connect()/disconnect(). When
+/// [BondRefusalGiveUp] trips, the app pauses auto-reconnect and never calls
+/// disconnect() — so the dead session's five timers (heartbeat, keep-alive,
+/// periodic backfill, idle watchdog, historical retry) kept firing forever and
+/// its four `onValueReceived` subscriptions stayed registered, one more full
+/// set leaked per drop.
+class LinkDownPolicy {
+ const LinkDownPolicy._();
+
+ static LinkDownAction evaluate({required bool sessionIsCurrent}) =>
+ sessionIsCurrent
+ ? LinkDownAction.tearDownSession
+ : LinkDownAction.ignoreStaleSession;
+}
+
+/// Outcome of a process-wide single-owner band claim.
+enum BandClaimDecision {
+ /// Take the claim (nobody holds it, or the incumbent's claim is stale).
+ claim,
+
+ /// A live foreground owner exists and we are the background drainer —
+ /// don't touch the band this cycle.
+ yieldToOwner,
+
+ /// A live background owner exists and we are foreground — drop its link
+ /// (awaited) and then take the claim.
+ preemptThenClaim,
+}
+
+/// Pure arbitration for the process-wide single-owner guard.
+///
+/// The claim used to be tested for NON-NULLNESS only, and was taken BEFORE the
+/// link was up and released only by an explicit `disconnect()`. So a connect
+/// that threw left the claim pointing at an engine with no link, forever: a
+/// later iOS restore wake spun up a background drainer, saw a non-null owner,
+/// yielded, and reported "strap not reachable this cycle" for the rest of the
+/// process lifetime. [incumbentLive] is the fix — a claim held by an engine
+/// with no session is not a claim.
+class BandClaimPolicy {
+ const BandClaimPolicy._();
+
+ static BandClaimDecision decide({
+ required bool incumbentPresent,
+ required bool incumbentLive,
+ required bool isBackgroundDrainer,
+ }) {
+ if (!incumbentPresent) return BandClaimDecision.claim;
+ if (!incumbentLive) return BandClaimDecision.claim;
+ if (isBackgroundDrainer) return BandClaimDecision.yieldToOwner;
+ return BandClaimDecision.preemptThenClaim;
+ }
+}
+
+/// Where an inbound reassembled frame is processed.
+enum FrameRoute {
+ /// The single serialized offload queue (`_offloadFrames`), so history
+ /// records and their terminals are handled strictly in arrival order by ONE
+ /// loop.
+ serializedQueue,
+
+ /// Handled inline (command responses, events, live high-rate frames).
+ immediate,
+}
+
+/// Pure routing decision for [FrameRoute].
+///
+/// Metadata (HISTORY_START / HISTORY_END / HISTORY_COMPLETE) used to reach the
+/// serialized queue only when it arrived on the `data` characteristic;
+/// metadata reassembled on `cmd_from`/`events` was fired unawaited on the
+/// immediate path instead — the one route that could run a HISTORY_END handler
+/// CONCURRENTLY with the queued drain, i.e. two handlers racing on the same
+/// drain controller (one snapshots an empty buffer and can ACK before the
+/// other's commit is durable). Metadata now always takes the queue.
+class FrameRoutePolicy {
+ const FrameRoutePolicy._();
+
+ static FrameRoute route({
+ required bool isMetadata,
+ required bool isHistorical,
+ required bool isDataRole,
+ }) {
+ if (isMetadata) return FrameRoute.serializedQueue;
+ if (isHistorical && isDataRole) return FrameRoute.serializedQueue;
+ return FrameRoute.immediate;
+ }
+}
+
/// Tracks ACK-write failures per historical-batch token ACROSS RECONNECTS —
/// a chunk whose ACK keeps failing for the SAME token (the "Groundhog Day"
/// re-flood signature: the band never trims, so it re-sends the identical
diff --git a/lib/cloud/cloud_import.dart b/lib/cloud/cloud_import.dart
index e0aa1eef..c6fd7ac6 100644
--- a/lib/cloud/cloud_import.dart
+++ b/lib/cloud/cloud_import.dart
@@ -17,6 +17,8 @@
import 'dart:convert';
+import 'package:flutter/foundation.dart' show visibleForTesting;
+
import '../compute/derivation_engine.dart' show kAlgoVersion;
import '../data/db.dart';
import 'backend_client.dart';
@@ -79,8 +81,7 @@ class CloudImporter {
var sessCount = 0;
for (final w in sessions) {
if (w is! Map) continue;
- await _writeSession(w.cast());
- sessCount++;
+ if (await _writeSession(w.cast())) sessCount++;
}
return CloudImportResult(dayCount, sessCount, _mapProfile(profileRaw));
@@ -250,14 +251,26 @@ class CloudImporter {
);
}
- static Future _writeSession(Map w) async {
+ /// Test seam for [_writeSession] — the malformed-row skip is a data-integrity
+ /// invariant, and `run()` needs a whole authenticated BackendClient to reach.
+ @visibleForTesting
+ static Future debugWriteSession(Map w) =>
+ _writeSession(w);
+
+ /// Returns true when a session row was actually written.
+ static Future _writeSession(Map w) async {
num? n(Object? v) => v is num ? v : null;
final start = n(w['start_ts'])?.toInt();
final end = n(w['end_ts'])?.toInt();
+ // A session with no usable start is not a workout — writing it as epoch 0
+ // filed a phantom workout on 1970-01-01 that then showed up in every query
+ // keyed on start_ts. Skip the row (same contract as WhoopImporter's
+ // _writeWorkout).
+ if (start == null) return false;
final zones = w['zones'];
await LocalDb.putSession({
'id': (w['id'] ?? 'cloud_$start').toString(),
- 'start_ts': start ?? 0,
+ 'start_ts': start,
'end_ts': end,
'type': (w['type'] ?? w['detected_type'] ?? 'other').toString(),
'status': (w['status'] ?? 'done').toString(),
@@ -266,10 +279,11 @@ class CloudImporter {
'strain': n(w['strain'])?.toDouble(),
'max_hr': n(w['max_hr'])?.toInt(),
'duration_min':
- (start != null && end != null) ? ((end - start) / 60).round() : null,
+ end != null ? ((end - start) / 60).round() : null,
'zone_min_json': zones == null ? null : jsonEncode(zones),
'created_at': DateTime.now().millisecondsSinceEpoch,
});
+ return true;
}
static String _ymd(DateTime d) => '${d.year.toString().padLeft(4, '0')}-'
diff --git a/lib/coach/coach_db.dart b/lib/coach/coach_db.dart
index 941f96d1..1d999bfb 100644
--- a/lib/coach/coach_db.dart
+++ b/lib/coach/coach_db.dart
@@ -4,15 +4,38 @@
// handle, over DERIVED-only views (v_metric/v_daily/v_series/v_hypnogram/
// v_sessions/v_baselines/v_insights — created by LocalDb._ensureCoachViews).
//
-// Two layers of safety, both fail-closed:
-// 1. guardAndPrepare() — a static validator: SELECT/WITH only, single
-// statement, no comments, no DML/DDL/PRAGMA, every FROM/JOIN target must be
-// an allow-listed view, no base/raw table name anywhere, auto-LIMIT.
-// 2. A read-only Database handle (openDatabase readOnly:true) — even if the
-// guard were bypassed, SQLite physically refuses writes/DDL, and raw byte
-// tables are blocked by the guard's identifier allow-list.
+// THREE layers of safety, all fail-closed:
+//
+// 1. guardAndPrepare() — a static validator built on a real (small) SQL
+// lexer + table-reference parser, NOT a single regex. It is an
+// ALLOW-LIST: SELECT/WITH only, single statement, no comments, no
+// DML/DDL/PRAGMA, and EVERY table position (including the extra members
+// of a comma-separated implicit cross-join, which the old FROM/JOIN
+// regex never even looked at) must resolve to an allow-listed `v_*` view
+// or a CTE declared in the same statement. Anything else — a base table,
+// a schema-qualified name, sqlite_master/sqlite_schema, a table-valued
+// function, a subquery in FROM — is rejected because it is not on the
+// list, so a table added to the schema tomorrow is closed by default
+// instead of open by default.
+//
+// 2. _assertAllowedBtrees() — a STRUCTURAL check that does not trust the
+// parser at all. sqflite exposes no sqlite3_set_authorizer() hook and a
+// read-only handle still sees every table in the file, so we ask SQLite
+// itself which btrees a statement would open: `EXPLAIN ` lists the
+// OpenRead/ReopenIdx opcodes with the ROOT PAGE of each btree, and
+// sqlite_master maps root pages back to the owning table. The allowed
+// root-page set is derived (once, at runtime) by EXPLAINing the allowed
+// views themselves, so it is exactly "the base tables the coach views
+// are made of" — every other btree in the database, including
+// workout_route (raw GPS) and sqlite_master, is unreachable no matter
+// what the text-level parser thinks. This is the authorizer we can't
+// install, implemented with the compiler's own answer.
+//
+// 3. A read-only Database handle (openDatabase readOnly:true) — even if
+// both layers above were bypassed, SQLite physically refuses writes/DDL.
import 'dart:convert';
+import 'package:flutter/foundation.dart' show visibleForTesting;
import 'package:path/path.dart' as p;
import 'package:sqflite/sqflite.dart';
@@ -30,7 +53,9 @@ class SqlGuardError implements Exception {
class CoachDb {
CoachDb._();
- /// The only tables the coach may read — derived, never raw.
+ /// The ONLY relations the coach may name in a table position — derived,
+ /// never raw. This is the whole allow-list: a CTE declared inside the same
+ /// statement is the only other thing that may appear after FROM/JOIN.
static const Set allowedViews = {
'v_metric',
'v_daily',
@@ -47,20 +72,42 @@ class CoachDb {
'insert', 'update', 'delete', 'drop', 'alter', 'create', 'replace',
'truncate', 'attach', 'detach', 'pragma', 'vacuum', 'reindex', 'analyze',
'begin', 'commit', 'rollback', 'savepoint', 'grant', 'revoke', 'trigger',
- 'into', 'load_extension',
+ 'into', 'load_extension', 'explain',
+ };
+
+ /// Real on-disk relations. This is NOT the security boundary any more (the
+ /// allow-list above is) — it is a redundant net whose only jobs are to give
+ /// the model a clearer rejection reason and to stop a CTE from SHADOWING a
+ /// real table name. Anything missing from it is still rejected by the
+ /// allow-list; nothing here is ever reachable.
+ static const Set reservedTableNames = {
+ // derived / user tables
+ 'day_result', 'metric_series', 'baselines', 'derived_day', 'sessions',
+ 'notifications', 'journal', 'cycle_log', 'cycle_symptom', 'notif_fired',
+ 'sleep_override', 'sleep_session_candidates', 'wake_day_features',
+ 'workout_suggestions', 'workout_route', 'live_coverage',
+ // raw / decoded substrate
+ 'raw_records', 'raw_archive', 'decoded_onehz', 'decoded_rr', 'samples',
+ 'events', 'band_events', 'band_battery',
+ // sync / compute bookkeeping
+ 'sync_ledger', 'sync_quarantine', 'sync_cursor', 'sync_ledger_legacy',
+ 'sync_quarantine_legacy', 'sync_cursor_legacy', 'compute_jobs',
+ 'compute_freshness',
+ // sqlite internals (the sqlite_* prefix is rejected wholesale as well)
+ 'sqlite_master', 'sqlite_schema', 'sqlite_temp_master',
+ 'sqlite_temp_schema', 'sqlite_sequence', 'sqlite_stat1', 'dbstat',
};
- // Base/raw table names that must NOT appear anywhere (force the views).
- static const Set _denyTables = {
- 'raw_records', 'decoded_onehz', 'decoded_rr', 'samples', 'sqlite_master',
- 'day_result', 'metric_series', 'baselines', 'sessions', 'notifications',
- 'derived_day', 'events', 'band_events', 'band_battery', 'live_coverage',
- 'sync_ledger', 'sync_quarantine', 'sync_cursor', 'compute_jobs',
- 'compute_freshness', 'journal', 'cycle_log', 'cycle_symptom',
- 'primitive_artifacts', 'wake_day_features', 'workout_suggestions',
+ /// Tokens that can follow a table reference without being an alias.
+ static const Set _clauseKeywords = {
+ 'where', 'group', 'order', 'limit', 'offset', 'having', 'window', 'union',
+ 'intersect', 'except', 'on', 'using', 'join', 'inner', 'left', 'right',
+ 'full', 'cross', 'natural', 'and', 'or', 'as', 'values', 'returning',
+ 'select', 'from', 'with',
};
static Database? _ro;
+ static Set? _allowedRoots;
/// Open (and cache) a read-only handle. Ensures the RW handle first so the
/// views exist (a read-only handle can't CREATE VIEW).
@@ -75,69 +122,344 @@ class CoachDb {
static Future close() async {
await _ro?.close();
_ro = null;
+ _allowedRoots = null;
}
+ // ── layer 1: lexer + table-reference parser ────────────────────────────────
+
/// Validate + normalize an LLM-supplied SELECT. Throws [SqlGuardError] on any
/// rejection. Returns the (possibly LIMIT-appended) query to execute.
static String guardAndPrepare(String raw, {int rowCap = 200}) {
var s = raw.trim();
if (s.isEmpty) throw SqlGuardError('Empty query.');
+ if (s.codeUnits.contains(0)) throw SqlGuardError('Illegal character.');
+
+ // (a) Strip one optional trailing ';' before anything else, so the
+ // single-statement check below sees only genuine inner separators.
+ var body = s.endsWith(';') ? s.substring(0, s.length - 1).trim() : s;
+ if (body.isEmpty) throw SqlGuardError('Empty query.');
+
+ // (b) Mask string literals to a neutral numeric token. Quoted IDENTIFIERS
+ // ("x", `x`, [x]) are rejected outright — the coach never needs them,
+ // and SQLite's double-quote fallback (a bare "foo" silently becomes a
+ // string when it doesn't resolve to a column) is a parser trap.
+ final masked = _maskStrings(body);
- // (a) No comments (they can hide payloads).
- if (s.contains('--') || s.contains('/*') || s.contains('*/')) {
+ // (c) No comments (they can hide payloads) — checked on the MASKED text so
+ // an inline '--' inside a legitimate string value isn't a false hit.
+ if (masked.contains('--') ||
+ masked.contains('/*') ||
+ masked.contains('*/')) {
throw SqlGuardError('Comments are not allowed.');
}
- // (b) Single statement — strip one optional trailing ';', reject any inner.
- var body = s.endsWith(';') ? s.substring(0, s.length - 1).trim() : s;
- if (body.contains(';')) {
+ // (d) Single statement.
+ if (masked.contains(';')) {
throw SqlGuardError('Only one statement is allowed.');
}
- // (c) Must start with SELECT or WITH.
- final lower = body.toLowerCase();
+ // (e) Must start with SELECT or WITH.
+ final lower = masked.toLowerCase();
if (!(lower.startsWith('select') || lower.startsWith('with'))) {
throw SqlGuardError('Query must start with SELECT or WITH.');
}
- // (d) Tokenize; reject any banned keyword + collect identifiers.
- final tokens = RegExp(r'[A-Za-z_][A-Za-z0-9_]*')
- .allMatches(lower)
- .map((m) => m.group(0)!)
- .toList();
- for (final t in tokens) {
- if (_banned.contains(t)) throw SqlGuardError('Disallowed keyword: $t');
- if (_denyTables.contains(t)) throw SqlGuardError('Table not allowed: $t');
- }
- // (e) CTE names declared via WITH … AS ( become valid FROM targets.
- final cte = {};
- for (final m in RegExp(r'(?:with|,)\s+([A-Za-z_][A-Za-z0-9_]*)\s+as\s*\(',
- caseSensitive: false)
- .allMatches(lower)) {
- cte.add(m.group(1)!);
- }
- // (f) Every FROM/JOIN target must be an allowed view (or a CTE name).
- final refRe = RegExp(r'\b(?:from|join)\s+("?)([A-Za-z_][A-Za-z0-9_.]*)\1',
- caseSensitive: false);
- final refs = [];
- for (final m in refRe.allMatches(lower)) {
- final id = m.group(2)!;
- if (id.contains('.')) {
- throw SqlGuardError('Schema-qualified names are not allowed: $id');
+
+ final toks = _lex(lower);
+
+ // (f) Banned keywords + reserved relation names, anywhere.
+ for (final t in toks) {
+ if (!t.ident) continue;
+ if (_banned.contains(t.text)) {
+ throw SqlGuardError('Disallowed keyword: ${t.text}');
+ }
+ if (t.text.startsWith('sqlite_') || t.text.startsWith('pragma_')) {
+ throw SqlGuardError('SQLite internals are not queryable: ${t.text}');
+ }
+ if (reservedTableNames.contains(t.text)) {
+ throw SqlGuardError('Table not allowed: ${t.text}. '
+ 'Allowed views: ${allowedViews.join(', ')}.');
}
- refs.add(id);
}
- if (refs.isEmpty) throw SqlGuardError('No table reference found.');
- for (final r in refs) {
- if (!allowedViews.contains(r) && !cte.contains(r)) {
- throw SqlGuardError(
- 'Table "$r" is not queryable. Allowed views: ${allowedViews.join(', ')}.');
+
+ // (g) CTE names declared via ` AS (` become valid table targets.
+ // A CTE may not shadow an allowed view (that would silently redefine
+ // the coach's data surface) — shadowing a REAL table is already
+ // impossible because (f) rejects every reserved name outright.
+ final cte = _cteNames(toks);
+ for (final c in cte) {
+ if (allowedViews.contains(c)) {
+ throw SqlGuardError('A CTE may not shadow the view "$c".');
}
}
- // (g) Auto-append a LIMIT to protect context.
+
+ // (h) THE allow-list gate: parse every table position and require it to be
+ // an allowed view or a declared CTE. Comma-separated cross-joins,
+ // subqueries in FROM, table-valued functions and schema-qualified
+ // names are all handled here (the old single-regex version only ever
+ // saw the identifier directly after FROM/JOIN).
+ final refs = _tableRefs(toks, allowed: {...allowedViews, ...cte});
+ if (refs.isEmpty) throw SqlGuardError('No table reference found.');
+
+ // (i) Auto-append a LIMIT to protect context.
if (!RegExp(r'\blimit\b', caseSensitive: false).hasMatch(lower)) {
body = '$body LIMIT $rowCap';
}
return body;
}
+ /// Replace every single-quoted string literal with the neutral token `0`.
+ /// Rejects unterminated literals and every quoted-identifier form.
+ static String _maskStrings(String src) {
+ final out = StringBuffer();
+ var i = 0;
+ while (i < src.length) {
+ final c = src[i];
+ if (c == "'") {
+ i++;
+ var closed = false;
+ while (i < src.length) {
+ if (src[i] == "'") {
+ if (i + 1 < src.length && src[i + 1] == "'") {
+ i += 2;
+ continue;
+ }
+ i++;
+ closed = true;
+ break;
+ }
+ i++;
+ }
+ if (!closed) throw SqlGuardError('Unterminated string literal.');
+ out.write('0');
+ continue;
+ }
+ if (c == '"' || c == '`' || c == '[' || c == ']') {
+ throw SqlGuardError(
+ 'Quoted identifiers are not allowed — use bare view names.');
+ }
+ out.write(c);
+ i++;
+ }
+ return out.toString();
+ }
+
+ static List<_Tok> _lex(String s) {
+ final out = <_Tok>[];
+ var i = 0;
+ bool identStart(String c) =>
+ (c.codeUnitAt(0) >= 97 && c.codeUnitAt(0) <= 122) || c == '_';
+ bool identPart(String c) =>
+ identStart(c) || (c.codeUnitAt(0) >= 48 && c.codeUnitAt(0) <= 57);
+ bool digit(String c) => c.codeUnitAt(0) >= 48 && c.codeUnitAt(0) <= 57;
+ while (i < s.length) {
+ final c = s[i];
+ if (c.trim().isEmpty) {
+ i++;
+ continue;
+ }
+ if (identStart(c)) {
+ final st = i;
+ while (i < s.length && identPart(s[i])) {
+ i++;
+ }
+ out.add(_Tok(s.substring(st, i), true));
+ continue;
+ }
+ if (digit(c)) {
+ final st = i;
+ while (i < s.length && (digit(s[i]) || s[i] == '.')) {
+ i++;
+ }
+ out.add(_Tok(s.substring(st, i), false));
+ continue;
+ }
+ out.add(_Tok(c, false));
+ i++;
+ }
+ return out;
+ }
+
+ /// CTE names: an identifier preceded by `with`/`recursive`/`,` and followed
+ /// (past an optional parenthesised column list) by `as (`.
+ static Set _cteNames(List<_Tok> toks) {
+ final names = {};
+ for (var i = 1; i < toks.length; i++) {
+ final prev = toks[i - 1];
+ final t = toks[i];
+ if (!t.ident) continue;
+ final opensCte = (prev.ident && (prev.text == 'with' || prev.text == 'recursive')) ||
+ (!prev.ident && prev.text == ',');
+ if (!opensCte) continue;
+ var j = i + 1;
+ if (j < toks.length && !toks[j].ident && toks[j].text == '(') {
+ j = _skipParens(toks, j);
+ }
+ if (j < toks.length && toks[j].ident && toks[j].text == 'as') {
+ var k = j + 1;
+ // AS [NOT] MATERIALIZED (
+ while (k < toks.length &&
+ toks[k].ident &&
+ (toks[k].text == 'not' || toks[k].text == 'materialized')) {
+ k++;
+ }
+ if (k < toks.length && !toks[k].ident && toks[k].text == '(') {
+ names.add(t.text);
+ }
+ }
+ }
+ return names;
+ }
+
+ /// Index just past the balanced ')' that matches the '(' at [open].
+ static int _skipParens(List<_Tok> toks, int open) {
+ var depth = 0;
+ for (var i = open; i < toks.length; i++) {
+ if (toks[i].ident) continue;
+ if (toks[i].text == '(') depth++;
+ if (toks[i].text == ')') {
+ depth--;
+ if (depth == 0) return i + 1;
+ }
+ }
+ return toks.length;
+ }
+
+ /// Every table position in the statement, validated against [allowed].
+ /// Walks the FULL table-reference list after each FROM/JOIN — including the
+ /// comma-separated members of an implicit cross-join.
+ static List _tableRefs(List<_Tok> toks, {required Set allowed}) {
+ final refs = [];
+ for (var i = 0; i < toks.length; i++) {
+ final t = toks[i];
+ if (!t.ident || (t.text != 'from' && t.text != 'join')) continue;
+ var j = i + 1;
+ while (true) {
+ if (j >= toks.length) {
+ throw SqlGuardError('Malformed FROM clause.');
+ }
+ final n = toks[j];
+ if (!n.ident) {
+ if (n.text == '(') {
+ throw SqlGuardError(
+ 'Subqueries in FROM are not allowed — query a view directly.');
+ }
+ throw SqlGuardError('Expected a view name after FROM/JOIN.');
+ }
+ final name = n.text;
+ j++;
+ // Schema-qualified (main.x / temp.x) or a table-valued function.
+ if (j < toks.length && !toks[j].ident && toks[j].text == '.') {
+ throw SqlGuardError('Schema-qualified names are not allowed: $name');
+ }
+ if (j < toks.length && !toks[j].ident && toks[j].text == '(') {
+ throw SqlGuardError('Functions are not allowed in FROM: $name');
+ }
+ if (!allowed.contains(name)) {
+ throw SqlGuardError(
+ 'Table "$name" is not queryable. Allowed views: '
+ '${allowedViews.join(', ')}.');
+ }
+ refs.add(name);
+ // Optional alias: `AS x` or a bare `x`.
+ if (j < toks.length && toks[j].ident && toks[j].text == 'as') {
+ j++;
+ if (j >= toks.length || !toks[j].ident) {
+ throw SqlGuardError('Malformed alias after "$name".');
+ }
+ j++;
+ } else if (j < toks.length &&
+ toks[j].ident &&
+ !_clauseKeywords.contains(toks[j].text)) {
+ j++;
+ }
+ // Another member of the same (comma-separated) table list?
+ if (j < toks.length && !toks[j].ident && toks[j].text == ',') {
+ j++;
+ continue;
+ }
+ break;
+ }
+ i = j - 1;
+ }
+ return refs;
+ }
+
+ // ── layer 2: structural btree gate (SQLite's own answer) ───────────────────
+
+ /// Root pages of every btree the ALLOWED VIEWS themselves read. Computed
+ /// once per read-only handle by EXPLAINing `SELECT * FROM `; SQLite
+ /// expands the view and reports exactly which base tables/indexes it opens.
+ static Future> _allowedRootPages(Database db) async {
+ final cached = _allowedRoots;
+ if (cached != null) return cached;
+ final roots = {};
+ for (final v in allowedViews) {
+ try {
+ roots.addAll(await _btreeRoots(db, 'SELECT * FROM $v'));
+ } catch (_) {/* a view that can't compile contributes nothing */}
+ }
+ // Indexes on an allowed base table are allowed too (the planner may pick
+ // one for a WHERE the plain scan above never used).
+ final schema = await db.rawQuery(
+ 'SELECT type, name, tbl_name, rootpage FROM sqlite_master');
+ final tableOfRoot = {};
+ for (final r in schema) {
+ final rp = (r['rootpage'] as num?)?.toInt();
+ final tbl = r['tbl_name']?.toString();
+ if (rp != null && rp > 0 && tbl != null) tableOfRoot[rp] = tbl;
+ }
+ final allowedTables = {
+ for (final rp in roots)
+ if (tableOfRoot[rp] != null) tableOfRoot[rp]!,
+ };
+ for (final e in tableOfRoot.entries) {
+ if (allowedTables.contains(e.value)) roots.add(e.key);
+ }
+ _allowedRoots = roots;
+ return roots;
+ }
+
+ /// Root pages of every btree a prepared [sql] opens, per SQLite's own
+ /// bytecode. Only cursor-opening opcodes carry a root page in P2 —
+ /// OpenEphemeral/SorterOpen/OpenPseudo reuse P2 for a column count and are
+ /// deliberately ignored.
+ static Future> _btreeRoots(Database db, String sql) async {
+ const opening = {'OpenRead', 'OpenWrite', 'ReopenIdx'};
+ final rows = await db.rawQuery('EXPLAIN $sql');
+ final out = {};
+ for (final r in rows) {
+ final op = r['opcode']?.toString();
+ if (op == null || !opening.contains(op)) continue;
+ final p2 = (r['p2'] as num?)?.toInt();
+ if (p2 != null && p2 > 0) out.add(p2);
+ }
+ return out;
+ }
+
+ /// Fail-closed structural gate: reject unless every btree the statement
+ /// opens belongs to a base table one of the allowed views is built from.
+ static Future _assertAllowedBtrees(Database db, String sql) async {
+ final allowed = await _allowedRootPages(db);
+ if (allowed.isEmpty) {
+ throw SqlGuardError('Query surface unavailable.');
+ }
+ final Set used;
+ try {
+ used = await _btreeRoots(db, sql);
+ } catch (e) {
+ throw SqlGuardError('Query could not be compiled: $e');
+ }
+ for (final r in used) {
+ if (!allowed.contains(r)) {
+ throw SqlGuardError('Query reaches storage outside the coach views. '
+ 'Allowed views: ${allowedViews.join(', ')}.');
+ }
+ }
+ }
+
+ /// Test seam for layer 2 on its own (layer 1 normally rejects these first).
+ @visibleForTesting
+ static Future debugAssertAllowedBtrees(String sql) async =>
+ _assertAllowedBtrees(await _readonly(), sql);
+
/// Run an LLM SELECT and return compact JSON for the tool result. On a guard
/// rejection, returns the reason (so the model fixes its query) — never throws.
static Future runCoachSql(String llmSql, {int rowCap = 200}) async {
@@ -149,6 +471,7 @@ class CoachDb {
}
try {
final db = await _readonly();
+ await _assertAllowedBtrees(db, sql);
final rows = await db.rawQuery(sql);
final shown = rows.take(rowCap).toList();
final out = {
@@ -164,8 +487,16 @@ class CoachDb {
var s = jsonEncode(out);
if (s.length > 16000) s = '${s.substring(0, 16000)}…(truncated)';
return s;
+ } on SqlGuardError catch (e) {
+ return jsonEncode({'error': e.reason});
} catch (e) {
return jsonEncode({'error': 'Query failed: $e'});
}
}
}
+
+class _Tok {
+ final String text;
+ final bool ident;
+ const _Tok(this.text, this.ident);
+}
diff --git a/lib/coach/coach_engine.dart b/lib/coach/coach_engine.dart
index 3cfade04..40a2d529 100644
--- a/lib/coach/coach_engine.dart
+++ b/lib/coach/coach_engine.dart
@@ -14,6 +14,7 @@
import 'dart:convert';
import 'dart:io';
import 'dart:math';
+import 'package:flutter/foundation.dart';
import 'package:http/http.dart' as http;
import 'package:path_provider/path_provider.dart';
@@ -170,6 +171,15 @@ class CoachEngine {
// Serializable display transcript (text bubbles + charts) shown in the UI.
final List transcript = [];
+ /// Test seams for the history-trimming invariant. `_history` is private and
+ /// the trim only runs deep inside the tool-calling loop, so there is no other
+ /// way to pin the orphaned-`tool` edge case without a live provider.
+ @visibleForTesting
+ List