diff --git a/lib/compute/derivation_engine.dart b/lib/compute/derivation_engine.dart index 1afb400..ed07e56 100644 --- a/lib/compute/derivation_engine.dart +++ b/lib/compute/derivation_engine.dart @@ -38,6 +38,7 @@ import '../notify/tap_router.dart' show kRouteWorkoutSuggestion; import '../telemetry/telemetry_service.dart'; import 'crossday_pipeline.dart'; import 'derive_pacing.dart'; +import 'sleep_profile_policy.dart'; import 'derive_prepare.dart'; import 'onehz_pipeline.dart'; import 'profile.dart'; @@ -362,7 +363,24 @@ import 'substrate.dart'; // (fever/heat/anxiety) that have no discernible onset — so this changes which // suggestions autoDetectWorkouts emits without loosening the false-positive // gate it exists to protect. -const int kAlgoVersion = 51; +// v52: the rolling per-user sleep profile (`sleep_user_profile`) was folded on +// EVERY staging pass for a day, not once per day — a real 12-day export carried +// `nights: 1348`. Two consequences, both bad: `personalWeight` pinned at its +// 0.5 cap from the first sweep, and an EWMA collapsed onto whichever day was +// re-derived last. Replaying that profile against the same 11 nights moved wake +// 4.3% -> 36.4% and deep 1.9% -> 0.0% on the worst night, i.e. the +// personalization layer was re-creating the wake over-call cardioStager exists +// to avoid. Fixed by (1) folding at most once per day_id (tracked in the +// profile payload), (2) withholding the profile from staging until +// kMinNightsForSleepProfile nights (van der Aar 2025: gains need >=3 nights and +// ~17.5% of subjects get WORSE from personalization), and (3) discarding +// pre-tracking profiles, which cannot be repaired, so they rebuild honestly. +// Bump so every day re-stages without the corrupt blend. +const int kAlgoVersion = 52; + +// Fold idempotency, the minimum-nights warm-up, and legacy-payload handling +// all live in SleepProfilePolicy (pure, unit-tested) — see +// lib/compute/sleep_profile_policy.dart for the evidence behind each rule. /// Raw is kept this many days past derivation, then pruned (derived stays). const int rawRetentionDays = 3; @@ -1155,17 +1173,32 @@ class DerivationEngine { // The worker isolate dies after `Isolate.run`, so the recording flag can't // leak into the next day's derivation — no try/finally reset needed. final profileJson = await _loadSleepUserProfileJson(); + // Which day_ids have ALREADY been folded into that profile. See + // [_kFoldedDaysKey] for why this exists and why a legacy profile that + // lacks it is discarded rather than trusted. + final foldedDays = SleepProfilePolicy.foldedDays(profileJson); + final mayFold = SleepProfilePolicy.shouldFold( + alreadyFolded: foldedDays, + dayId: dayId, + hasOverride: override != null, + ); // Cancellable + TIMED OUT. This site previously used a bare `Isolate.run` // with no timeout at all, so a hung staging pass never completed its future // — `_running` stayed true and `DeriveScheduler._drain` never returned, i.e. // all derivation was dead until app restart. - final (candidateJson, updatedProfileJson) = + final (candidateJson, observationJson) = await _runIsolateCancellable(() { try { - ana.cardioUserProfile = profileJson == null + final p = profileJson == null ? null : ana.SleepUserProfile.fromJson( (jsonDecode(profileJson) as Map).cast()); + // Warm-up gate — see SleepProfilePolicy.shouldBlend. Note the profile + // is only WITHHELD FROM STAGING here; accumulation into it happens on + // the main isolate in _foldObservationIntoProfile, which re-reads the + // current profile, so a withheld night still counts toward `nights`. + ana.cardioUserProfile = + SleepProfilePolicy.shouldBlend(p?.nights) ? p : null; } catch (_) { // Defense in depth: an incompatible/outdated persisted profile must // fall back to a cold start, never throw inside the worker (an uncaught @@ -1183,20 +1216,44 @@ class DerivationEngine { // Fold the MAIN sleep (most epochs) of a freshly-staged night into the // rolling profile — done here in the worker because the observations live // in THIS isolate's globals. Skipped for overrides. EWMA self-seeds. - String? foldedJson; - if (override == null) { + String? observationJson; + // IDEMPOTENT PER DAY. `fold()` is an EWMA step that also increments + // `nights`, and this path runs on EVERY staging pass for a day — an + // algo-version bump, a BLE-drain re-derive, a backfill sweep. Without a + // guard the same handful of real nights fold hundreds of times: a real + // user export showed `nights: 1348` against 12 days of data, which pins + // `personalWeight` at its 0.5 cap from day one and collapses the EWMA + // onto whichever day was re-derived last. Measured effect of that + // corrupt profile on the same nights: wake 4.3% -> 36.4%, deep 1.9% -> + // 0.0%. One fold per day_id, ever. + if (mayFold) { final obs = ana.takeCardioObservations(); if (obs.isNotEmpty) { obs.sort((a, b) => b.epochs.compareTo(a.epochs)); final main = obs.first; if (main.epochs >= 120) { - // require ≥60 min — not a nap - final base = ana.cardioUserProfile ?? const ana.SleepUserProfile(); - foldedJson = jsonEncode(base.fold(main).toJson()); + // require ≥60 min — not a nap. + // Return the raw OBSERVATION, not a folded profile. Folding here + // would bake in the profile this worker read before staging began, + // and a concurrent day may have written a newer one since. The + // fold happens on the main isolate under the profile lock. + observationJson = jsonEncode({ + 'epochs': main.epochs, + 'hr_floor_p5': main.hrFloorP5, + 'hr_floor_p25': main.hrFloorP25, + 'hr_sleep_median': main.hrSleepMedian, + 'hr_arousal': main.hrArousal, + 'rmssd_med': main.rmssdMed, + 'rmssd_mad': main.rmssdMad, + 'enmo_still_cut': main.enmoStillCut, + 'enmo_move_cut': main.enmoMoveCut, + 'lfhf_med': main.lfhfMed, + 'rk_med': main.rkMed, + }); } } } - return (jsonEncode(candidate.toJson()), foldedJson); + return (jsonEncode(candidate.toJson()), observationJson); }, _perDayTimeout, label: 'sleep-staging $dayId'); final candidate = SleepSessionCandidate.fromJson( (jsonDecode(candidateJson) as Map).cast()); @@ -1206,17 +1263,117 @@ class DerivationEngine { algoVersion: kAlgoVersion, payloadJson: candidateJson, ); - if (updatedProfileJson != null) { - await LocalDb.putBaseline('sleep_user_profile', updatedProfileJson); + if (observationJson != null) { + // BEST-EFFORT, and deliberately isolated from the day's success path. + // The fold is bookkeeping; the day's real result is already persisted + // above. `updateBaseline` takes an exclusive SQLite write lock, and the + // whole point of this change is that two derivation isolates contend + // for it — so SQLITE_BUSY here is an EXPECTED outcome, not an + // exceptional one. Letting it escape would hit processDay's broad + // catch, which calls `_markDaySkipped` and increments `failures`, + // throwing away a fully computed day (and holding the timezone) over a + // bookkeeping write. + // + // KNOWN LIMITATION — a swallowed failure here is PERMANENT for this + // day, not retried. Once the day finalizes, the cached-candidate + // short-circuit at the top of this method returns before staging runs, + // so `observationJson` is never regenerated and the fold never happens. + // Same for a day whose override is later removed if it already has a + // cached candidate from before the override. + // + // Accepted deliberately rather than fixed: the profile is an EWMA with + // a ~14-night horizon and a hard 0.5 blend cap, so one missing night is + // a small perturbation, whereas a retry path needs durable pending + // state and a way to distinguish "failed, retry" from "declined + // permanently" (a <120-epoch nap never folds, and would otherwise + // bypass the candidate cache and re-stage on every sweep forever). + // If the fold ever stops being best-effort, that state machine is the + // thing to build — do not simply bypass the cache. + try { + await _foldObservationIntoProfile(dayId, observationJson); + } catch (e) { + _log('sleep profile fold skipped for $dayId (day result kept, ' + 'this night will not contribute to the profile): $e'); + } } } return candidate; } + /// Fold one night's observation into the shared profile, serialised against + /// every other day in the sweep. + /// + /// The profile is RE-READ inside the lock and [SleepProfilePolicy.shouldFold] + /// re-checked, because the value this day read before staging is stale by + /// definition — a concurrently-derived day may have folded since. Skipping + /// that re-check is what turns a read-modify-write race into a lost fold plus + /// a lost day_id, and the day then re-folds forever. + Future _foldObservationIntoProfile( + String dayId, String observationJson) async { + final Map o; + try { + o = (jsonDecode(observationJson) as Map).cast(); + } catch (_) { + return; + } + double? d(String k) => (o[k] as num?)?.toDouble(); + final observed = ana.SleepNightObservation( + epochs: (o['epochs'] as num?)?.toInt() ?? 0, + hrFloorP5: d('hr_floor_p5'), + hrFloorP25: d('hr_floor_p25'), + hrSleepMedian: d('hr_sleep_median'), + hrArousal: d('hr_arousal'), + rmssdMed: d('rmssd_med'), + rmssdMad: d('rmssd_mad'), + enmoStillCut: d('enmo_still_cut'), + enmoMoveCut: d('enmo_move_cut'), + lfhfMed: d('lfhf_med'), + rkMed: d('rk_med'), + ); + // The whole read-modify-write happens inside ONE exclusive DB transaction. + // A Dart mutex cannot do this job: `derivationDispatcher` is a + // vm:entry-point WorkManager entry that builds its own DerivationEngine in + // a SEPARATE background isolate, and a `static` lock has one copy per + // isolate — so a background heavy pass and a foreground sweep would each + // read the same profile, fold, and clobber the other, losing both the fold + // and its day_id from folded_days. SQLite's write lock is cross-connection + // and therefore cross-isolate. + await LocalDb.updateBaseline('sleep_user_profile', (current) { + // Re-derive freshness INSIDE the transaction: the value this day read + // before staging is stale by definition, another lane may have folded + // since. Returning null leaves the row untouched. + final usable = SleepProfilePolicy.usableProfileJson(current); + final freshDays = SleepProfilePolicy.foldedDays(usable); + if (!SleepProfilePolicy.shouldFold( + alreadyFolded: freshDays, dayId: dayId, hasOverride: false)) { + return null; + } + final ana.SleepUserProfile base; + try { + base = usable == null + ? const ana.SleepUserProfile() + : ana.SleepUserProfile.fromJson( + (jsonDecode(usable) as Map).cast()); + } catch (_) { + return null; // unreadable — leave it for the cold-start path + } + return jsonEncode(SleepProfilePolicy.withFoldedDays( + base.fold(observed).toJson(), freshDays, dayId)); + }); + } + /// Read the persisted per-user sleep profile (`baselines` key /// `sleep_user_profile`) as raw JSON, for passing into the staging worker /// isolate. Absent/corrupt ⇒ null (cold start). DB read stays on the main /// isolate (the DB owner); the worker reconstructs the profile from this JSON. + /// + /// A profile written before per-day fold tracking existed carries no + /// [_kFoldedDaysKey] and therefore an untrustworthy `nights` count and an + /// EWMA skewed by repeated re-folds of the same nights. We cannot repair it + /// (there is no record of which days went in), so we DISCARD it and rebuild. + /// That degrades to pure per-night-local baselines — the cold-start path + /// cardio_stager.dart was validated on — and the profile re-earns its weight + /// over the next few nights under the corrected accounting. Future _loadSleepUserProfileJson() async { final row = await LocalDb.baseline('sleep_user_profile'); final raw = row?['payload_json']; @@ -1224,12 +1381,7 @@ class DerivationEngine { // Validate here (mirrors the cached-candidate guard above) so a corrupt // payload becomes a cold start, per this method's contract — rather than // throwing later inside the staging worker's `jsonDecode(...) as Map`. - try { - if (jsonDecode(raw) is Map) return raw; - } catch (_) { - // corrupt payload → null (cold start) - } - return null; + return SleepProfilePolicy.usableProfileJson(raw); } Future _loadSubstrateRange( diff --git a/lib/compute/sleep_profile_policy.dart b/lib/compute/sleep_profile_policy.dart new file mode 100644 index 0000000..1e938bd --- /dev/null +++ b/lib/compute/sleep_profile_policy.dart @@ -0,0 +1,165 @@ +// PURE POLICY — when the rolling per-user sleep profile may be folded, and when +// it is allowed to influence staging. +// +// Background. `cardioStager` (analytics) can blend a persisted per-user profile +// (`baselines` key `sleep_user_profile`) into tonight's per-night-local +// baselines, bounded by `SleepUserProfile.personalWeight` (0 → 0.5 as nights +// accumulate). The profile is produced by EWMA-folding one observation per +// finalized night. +// +// Two defects made that layer harmful in the field, and this policy exists to +// prevent both: +// +// 1. NOT IDEMPOTENT. The fold ran on every staging pass for a day — algo +// bumps, BLE-drain re-derives, backfill sweeps — not once per day. A real +// user export carried `nights: 1348` against 12 days of data. Because +// `personalWeight` saturates at 28 nights, that pins the blend at its 0.5 +// cap immediately, and the EWMA (alpha ~2/15) collapses onto whichever day +// was re-derived last instead of a representative fortnight. Replaying the +// corrupt profile over the same 11 nights moved wake 4.3% → 36.4% and deep +// 1.9% → 0.0% on the worst night: the personalization layer was +// re-manufacturing the very wake over-call cardio_stager.dart was written +// to eliminate. Fold at most ONCE per day_id. +// +// 2. NO WARM-UP. Personalization was applied from the first night. The one +// direct trial of per-subject personalization for wrist-PPG staging (van +// der Aar et al. 2025, Physiol Meas, n=59, fine-tuned per subject and +// scored against PSG) found performance improved in 82.5% of subjects and +// that significant gains required >= 3 nights — which also means ~17.5% +// got WORSE. A one- or two-night profile is therefore pure downside. Below +// the floor we stay on per-night-local baselines, the cold-start path +// cardio_stager.dart was actually validated on. +// +// A profile written before fold tracking existed carries no [foldedDaysKey]. +// Its `nights` is untrustworthy and there is no record of which days went in, +// so it CANNOT be repaired — [usableProfileJson] discards it and lets the +// profile rebuild honestly. + +import 'dart:convert'; + +class SleepProfilePolicy { + /// Minimum folded nights before the profile may influence staging. + static const int minNightsForBlend = 3; + + /// Key inside the profile payload holding the day_ids already folded in. + /// Its ABSENCE marks a pre-tracking (legacy) payload. + static const String foldedDaysKey = 'folded_days'; + + /// Cap on tracked day_ids so the `baselines` row cannot grow unbounded. + static const int maxFoldedDays = 400; + + const SleepProfilePolicy._(); + + /// Day_ids already folded into [payloadJson]. Empty for absent, corrupt, or + /// legacy payloads — a legacy payload is being discarded anyway, so treating + /// its (unknown) history as empty is the consistent answer. + static Set foldedDays(String? payloadJson) { + final m = _decode(payloadJson); + final days = m?[foldedDaysKey]; + if (days is! List) return const {}; + return {for (final d in days) if (d is String) d}; + } + + /// True when [payloadJson] predates fold tracking and must be discarded. + /// A null/corrupt payload is NOT "legacy" — there is simply nothing to + /// discard, and callers already treat it as a cold start. + static bool isLegacy(String? payloadJson) { + final m = _decode(payloadJson); + if (m == null) return false; + return m[foldedDaysKey] is! List; + } + + /// The payload to hand the staging worker, or null for a cold start. + /// Corrupt and legacy payloads both collapse to null. + static String? usableProfileJson(String? payloadJson) { + final m = _decode(payloadJson); + if (m == null) return null; + if (m[foldedDaysKey] is! List) return null; // legacy ⇒ rebuild + return payloadJson; + } + + /// Whether tonight's observation may be folded into the profile. + /// Overrides (manual / user-confirmed windows) never fold: the window is + /// asserted by the human, so its baselines are not evidence about the + /// sleeper's typical autonomic signature. + static bool shouldFold({ + required Set alreadyFolded, + required String dayId, + required bool hasOverride, + }) => + !hasOverride && !alreadyFolded.contains(dayId); + + /// Whether a profile with [nights] folded nights may influence staging. + static bool shouldBlend(int? nights) => + nights != null && nights >= minNightsForBlend; + + /// [alreadyFolded] + [dayId], sorted and capped. The cap evicts the OLDEST + /// entries: re-folding a day that has aged out is far less harmful than an + /// unbounded row, and at a ~14-night EWMA horizon an ancient re-fold is + /// nearly a no-op. + /// + /// PRECONDITION: [dayId] is a `YYYY-MM-DD` local day label (what + /// `day_label.dart` produces, and the only thing derivation passes). The cap + /// leans on that: eviction is by LEXICOGRAPHIC order, which equals + /// chronological order only for zero-padded ISO dates. Feed this an epoch + /// string or a UUID and the sort no longer means "age", so a RECENT day could + /// be evicted while an older one is kept — and an evicted day passes + /// [shouldFold] again, i.e. the double-fold this class exists to prevent. + /// Asserted rather than silently sorted differently, because the failure is + /// invisible in the payload. + static List appendFoldedDay( + Set alreadyFolded, String dayId) { + assert(_isDayLabel(dayId), + 'folded day_ids must be YYYY-MM-DD (eviction sorts on them); got "$dayId"'); + final out = ({...alreadyFolded, dayId}).toList()..sort(); + if (out.length > maxFoldedDays) { + out.removeRange(0, out.length - maxFoldedDays); + } + return out; + } + + static final RegExp _dayLabel = RegExp(r'^\d{4}-\d{2}-\d{2}$'); + + static bool _isDayLabel(String s) => _dayLabel.hasMatch(s); + + /// Stamp the folded-day set into a profile map produced by + /// `SleepUserProfile.toJson()` (which does not know about this key). + static Map withFoldedDays( + Map profileMap, + Set alreadyFolded, + String dayId, + ) => + { + ...profileMap, + foldedDaysKey: appendFoldedDay(alreadyFolded, dayId), + }; + + // NO DART-LEVEL LOCK HERE, DELIBERATELY. + // + // An earlier revision serialised the fold with a `static Future` mutex. That + // is not sufficient and is worse than nothing, because it looks sufficient: + // derivation runs in more than one isolate (`derivationDispatcher` is a + // vm:entry-point WorkManager entry that constructs its own DerivationEngine + // in a background isolate), and a Dart static has one copy PER ISOLATE. A + // background heavy pass and a foreground sweep would each hold "the" lock and + // still clobber each other. + // + // The read-modify-write is instead done inside one exclusive SQLite + // transaction — see `LocalDb.updateBaseline` and + // `DerivationEngine._foldObservationIntoProfile`. SQLite's write lock is + // cross-connection, so it holds across isolates AND across processes. + // + // Callers must re-read the payload and re-check [shouldFold] INSIDE that + // transaction; any value read before the staging isolate ran is stale by + // definition. + + static Map? _decode(String? payloadJson) { + if (payloadJson == null || payloadJson.isEmpty) return null; + try { + final d = jsonDecode(payloadJson); + return d is Map ? d.cast() : null; + } catch (_) { + return null; + } + } +} diff --git a/lib/data/db.dart b/lib/data/db.dart index ea66fbf..d76a842 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -3891,6 +3891,51 @@ class LocalDb { }, conflictAlgorithm: ConflictAlgorithm.replace); } + /// Atomically read-modify-write one `baselines` row. + /// + /// [transform] receives the current `payload_json` (null when the row does + /// not exist) and returns the replacement, or null to leave the row alone. + /// + /// Needed because a Dart-level lock CANNOT serialize this. Derivation runs in + /// more than one isolate — `derivationDispatcher` is a `vm:entry-point` + /// WorkManager entry that constructs its own `DerivationEngine` in a separate + /// background isolate — and a `static` mutex has one copy per isolate. Two + /// isolates would each read the same payload, merge into it, and write back, + /// dropping the other's changes. That matters most for accumulator payloads + /// like `sleep_user_profile`, where a lost write also loses the record of + /// which days were already folded. + /// + /// `exclusive: true` issues BEGIN IMMEDIATE, taking SQLite's write lock up + /// front rather than on first write. Without it a deferred transaction that + /// reads and then writes can fail to upgrade under WAL when another + /// connection holds the write lock. The lock is cross-connection and + /// therefore cross-isolate, which is exactly the guarantee a Dart static + /// cannot give. + static Future updateBaseline( + String key, + String? Function(String? current) transform, + ) async { + final db = await instance; + await db.transaction((txn) async { + final rows = await txn.query( + 'baselines', + columns: ['payload_json'], + where: 'key = ?', + whereArgs: [key], + limit: 1, + ); + final current = + rows.isEmpty ? null : rows.first['payload_json'] as String?; + final next = transform(current); + if (next == null) return; + await txn.insert('baselines', { + 'key': key, + 'payload_json': next, + 'updated_at': DateTime.now().millisecondsSinceEpoch, + }, conflictAlgorithm: ConflictAlgorithm.replace); + }, exclusive: true); + } + static Future?> computeFreshness(String key) async { final db = await instance; final rows = await db.query( diff --git a/test/db_update_baseline_test.dart b/test/db_update_baseline_test.dart new file mode 100644 index 0000000..8466e55 --- /dev/null +++ b/test/db_update_baseline_test.dart @@ -0,0 +1,235 @@ +// LocalDb.updateBaseline — the cross-isolate synchronization primitive behind +// the rolling sleep-profile fold, exercised against the REAL LocalDb over +// sqflite_ffi. +// +// Why this has its own suite: a Dart `static` mutex cannot serialize the fold, +// because derivation also runs in a background isolate (`derivationDispatcher` +// is a vm:entry-point WorkManager entry that builds its own DerivationEngine), +// and a static has one copy per isolate. The read-modify-write therefore has to +// be atomic in the DATABASE. `sleep_profile_policy_test.dart` covers the pure +// decision contract; these cover the storage contract it depends on. +// +// WHAT THESE DO AND DO NOT PROVE. Every test here runs in ONE isolate against +// ONE sqflite_ffi connection, so the "concurrent" cases exercise interleaved +// async access to a single connection — real, and they do fail against a naive +// read-then-write (19 of 20 increments lost), but not the same thing as two +// OS-level connections contending. The cross-ISOLATE guarantee rests on +// SQLite's documented locking (BEGIN IMMEDIATE takes the write lock up front +// and it is cross-connection), which these tests assume rather than verify. +// Verifying it properly needs a spawned isolate opening the same file, and +// LocalDb's singleton/static setup does not currently have an entry point for +// that. Worth building if this primitive picks up more callers. + +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:path_provider_platform_interface/path_provider_platform_interface.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; +import 'package:openstrap_edge/data/db.dart'; + +class _FakePathProvider extends PathProviderPlatform { + _FakePathProvider(this.root); + final String root; + @override + Future getTemporaryPath() async => root; + @override + Future getApplicationSupportPath() async => root; + @override + Future getApplicationDocumentsPath() async => root; + @override + Future getApplicationCachePath() async => root; + @override + Future getLibraryPath() async => root; + @override + Future getDownloadsPath() async => root; +} + +void main() { + late Directory tmp; + + setUpAll(() async { + sqfliteFfiInit(); + databaseFactory = databaseFactoryFfi; + tmp = await Directory.systemTemp.createTemp('openstrap_ub_'); + PathProviderPlatform.instance = _FakePathProvider(tmp.path); + LocalDb.dbName = 'openstrap_update_baseline_test.db'; + await databaseFactory.deleteDatabase( + p.join(await databaseFactory.getDatabasesPath(), LocalDb.dbName), + ); + }); + + tearDownAll(() async { + await LocalDb.close(); + final dir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); + if (await tmp.exists()) await tmp.delete(recursive: true); + }); + + test('transform receives null when the key is absent, and can create it', + () async { + String? seen = 'not-called'; + var called = false; + await LocalDb.updateBaseline('ub_absent', (current) { + called = true; + seen = current; + return '{"v":1}'; + }); + expect(called, isTrue); + expect(seen, isNull, reason: 'no row yet ⇒ transform sees null'); + final row = await LocalDb.baseline('ub_absent'); + expect(row?['payload_json'], '{"v":1}'); + }); + + test('a null return leaves the existing row byte-identical', () async { + await LocalDb.putBaseline('ub_untouched', '{"v":"original"}'); + final before = await LocalDb.baseline('ub_untouched'); + + var sawCurrent = ''; + await LocalDb.updateBaseline('ub_untouched', (current) { + sawCurrent = current ?? ''; + return null; // decline + }); + + expect(sawCurrent, '{"v":"original"}'); + final after = await LocalDb.baseline('ub_untouched'); + expect(after?['payload_json'], before?['payload_json']); + expect(after?['updated_at'], before?['updated_at'], + reason: 'a declined update must not even bump updated_at'); + }); + + test('a non-null return replaces the payload and advances updated_at', + () async { + await LocalDb.putBaseline('ub_replace', '{"n":1}'); + final before = await LocalDb.baseline('ub_replace'); + final beforeAt = before!['updated_at'] as int; + + // updated_at is millisecond-resolution wall clock; without a gap the + // rewrite can land in the same millisecond and the assertion below would + // be testing the clock, not the write. + await Future.delayed(const Duration(milliseconds: 5)); + + await LocalDb.updateBaseline('ub_replace', (current) { + final n = (jsonDecode(current!) as Map)['n'] as int; + return jsonEncode({'n': n + 1}); + }); + + final after = await LocalDb.baseline('ub_replace'); + expect(jsonDecode(after!['payload_json'] as String), {'n': 2}); + expect(after['updated_at'] as int, greaterThan(beforeAt)); + }); + + test('sequential accumulate: every update observes the previous commit', + () async { + await LocalDb.updateBaseline('ub_accum', (_) => jsonEncode({'n': 0})); + for (var i = 0; i < 25; i++) { + await LocalDb.updateBaseline('ub_accum', (current) { + final n = (jsonDecode(current!) as Map)['n'] as int; + return jsonEncode({'n': n + 1}); + }); + } + final row = await LocalDb.baseline('ub_accum'); + expect((jsonDecode(row!['payload_json'] as String) as Map)['n'], 25); + }); + + test('concurrent accumulate: no increment is lost to a read-modify-write race', + () async { + // The whole reason this method exists. Fired without awaiting between + // them, these interleave; a plain read + putBaseline pair loses writes. + await LocalDb.updateBaseline('ub_race', (_) => jsonEncode({'n': 0})); + const lanes = 20; + await Future.wait([ + for (var i = 0; i < lanes; i++) + LocalDb.updateBaseline('ub_race', (current) { + final n = (jsonDecode(current!) as Map)['n'] as int; + return jsonEncode({'n': n + 1}); + }) + ]); + final row = await LocalDb.baseline('ub_race'); + expect((jsonDecode(row!['payload_json'] as String) as Map)['n'], lanes, + reason: 'each lane must observe every earlier commit'); + }); + + test('concurrent set-union: no member is dropped', () async { + // Closer to the real payload shape — folded_days is a set that must only + // ever grow, and a lost write drops a day_id as well as a count. + await LocalDb.updateBaseline( + 'ub_set', (_) => jsonEncode({'days': []})); + final days = [for (var d = 10; d < 30; d++) '2026-07-$d']; + await Future.wait([ + for (final day in days) + LocalDb.updateBaseline('ub_set', (current) { + final cur = ((jsonDecode(current!) as Map)['days'] as List) + .cast() + .toSet(); + if (cur.contains(day)) return null; + return jsonEncode({ + 'days': (cur..add(day)).toList()..sort(), + }); + }) + ]); + final row = await LocalDb.baseline('ub_set'); + final stored = + ((jsonDecode(row!['payload_json'] as String) as Map)['days'] as List) + .cast(); + expect(stored, days..sort()); + }); + + test('concurrent LEGACY discard: the rebuild loses nothing either', () async { + // Specifically the legacy-transition case. The worry is that several lanes + // all observe the pre-tracking row at once, each treat it as a cold start, + // and each write a fresh profile containing only its own day — so the + // rebuild silently drops folds. + // + // It cannot happen through this method: BEGIN IMMEDIATE serialises the + // transactions, so only the FIRST lane sees the legacy row. By the time + // the second runs, the row already carries folded_days and is no longer + // legacy. Asserting it rather than reasoning about it. + await LocalDb.putBaseline('ub_legacy', jsonEncode({'nights': 1348})); + final days = [for (var d = 10; d < 25; d++) '2026-07-$d']; + + await Future.wait([ + for (final day in days) + LocalDb.updateBaseline('ub_legacy', (current) { + // Mirrors _foldObservationIntoProfile: legacy ⇒ treat as cold start. + final map = current == null + ? null + : (jsonDecode(current) as Map).cast(); + final isLegacy = map != null && map['folded_days'] is! List; + final usable = (map == null || isLegacy) ? null : map; + final known = + ((usable?['folded_days'] as List?) ?? const []).cast(); + if (known.contains(day)) return null; + final nights = (usable?['nights'] as int?) ?? 0; + return jsonEncode({ + 'nights': nights + 1, + 'folded_days': [...known, day]..sort(), + }); + }) + ]); + + final row = await LocalDb.baseline('ub_legacy'); + final decoded = jsonDecode(row!['payload_json'] as String) as Map; + expect(decoded['nights'], days.length, + reason: 'rebuilt from 0, and every lane counted exactly once'); + expect((decoded['folded_days'] as List).cast(), days..sort(), + reason: 'no day_id lost to the legacy-discard transition'); + expect(decoded['nights'], isNot(1348), reason: 'legacy count discarded'); + }); + + test('a throwing transform rolls back and leaves the row intact', () async { + await LocalDb.putBaseline('ub_throw', '{"v":"keep"}'); + await expectLater( + LocalDb.updateBaseline('ub_throw', (_) => throw StateError('boom')), + throwsStateError, + ); + final row = await LocalDb.baseline('ub_throw'); + expect(row?['payload_json'], '{"v":"keep"}'); + + // and the connection is still usable — a failed fold must not wedge the DB + await LocalDb.updateBaseline('ub_throw', (_) => '{"v":"next"}'); + expect((await LocalDb.baseline('ub_throw'))?['payload_json'], + '{"v":"next"}'); + }); +} diff --git a/test/sleep_profile_policy_test.dart b/test/sleep_profile_policy_test.dart new file mode 100644 index 0000000..d26dc53 --- /dev/null +++ b/test/sleep_profile_policy_test.dart @@ -0,0 +1,338 @@ +// Regression coverage for the rolling sleep-profile fold rules. +// +// The motivating defect, from a real user export: `sleep_user_profile` held +// `"nights": 1348` against 12 days of data, because the EWMA fold ran on every +// staging pass rather than once per day. That saturated `personalWeight` at its +// 0.5 cap immediately and collapsed the EWMA onto the most recently re-derived +// day. Replaying that profile over the same 11 nights moved wake 4.3% → 36.4% +// and deep 1.9% → 0.0% on the worst night. + +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/compute/sleep_profile_policy.dart'; + +String _payload({List? foldedDays, int nights = 0}) => jsonEncode({ + 'nights': nights, + 'hr_sleep_median': 52.5, + SleepProfilePolicy.foldedDaysKey: ?foldedDays, + }); + +void main() { + group('fold idempotency (the nights:1348 bug)', () { + test('a day already folded is never folded again', () { + final folded = SleepProfilePolicy.foldedDays( + _payload(foldedDays: ['2026-07-30'], nights: 1)); + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: folded, dayId: '2026-07-30', hasOverride: false), + isFalse, + ); + }); + + test('a day not yet folded is folded once', () { + final folded = SleepProfilePolicy.foldedDays( + _payload(foldedDays: ['2026-07-30'], nights: 1)); + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: folded, dayId: '2026-07-31', hasOverride: false), + isTrue, + ); + }); + + test('repeated staging passes over the same days cannot inflate nights', + () { + // Simulate what actually happened: 12 real days, re-derived 100x each. + var folded = {}; + var foldCount = 0; + final days = [for (var d = 20; d < 32; d++) '2026-07-$d']; + for (var pass = 0; pass < 100; pass++) { + for (final day in days) { + if (SleepProfilePolicy.shouldFold( + alreadyFolded: folded, dayId: day, hasOverride: false)) { + foldCount++; + folded = {...SleepProfilePolicy.appendFoldedDay(folded, day)}; + } + } + } + expect(foldCount, days.length, reason: 'one fold per distinct day'); + expect(folded.length, days.length); + }); + + test('an override night never folds — the window is asserted, not measured', + () { + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: const {}, dayId: '2026-07-31', hasOverride: true), + isFalse, + ); + }); + + test('skipping an override records nothing, so the POLICY stays eligible', + () { + // Scope note, because the obvious reading of this test is wrong: + // it asserts the POLICY only. Declining to fold an override records + // nothing in folded_days, so `shouldFold` keeps saying yes afterwards. + // Worth pinning because the tempting alternative — marking it folded to + // "remember we skipped it" — would exclude that night permanently. + // + // It does NOT assert that the engine actually re-folds after an override + // is removed. It often will not: `_sleepCandidateForDay` short-circuits + // on a cached finalized candidate before staging runs, so a day that had + // a candidate cached before the override was applied never regenerates an + // observation. See the KNOWN LIMITATION comment at the fold call site. + const day = '2026-07-31'; + var folded = {}; + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: folded, dayId: day, hasOverride: true), + isFalse, + ); + expect(folded, isEmpty, reason: 'a skipped override records nothing'); + // user deletes the override, day is re-derived + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: folded, dayId: day, hasOverride: false), + isTrue, + ); + folded = {...SleepProfilePolicy.appendFoldedDay(folded, day)}; + // ...and still only once thereafter + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: folded, dayId: day, hasOverride: false), + isFalse, + ); + }); + }); + + group('minimum-nights warm-up gate', () { + test('withholds the profile below the floor', () { + expect(SleepProfilePolicy.shouldBlend(0), isFalse); + expect(SleepProfilePolicy.shouldBlend(1), isFalse); + expect(SleepProfilePolicy.shouldBlend(2), isFalse); + }); + + test('applies the profile at and above the floor', () { + expect(SleepProfilePolicy.shouldBlend(3), isTrue); + expect(SleepProfilePolicy.shouldBlend(30), isTrue); + }); + + test('a null nights count never blends', () { + expect(SleepProfilePolicy.shouldBlend(null), isFalse); + }); + }); + + group('legacy payloads are discarded, not trusted', () { + test('a pre-tracking profile is legacy and yields a cold start', () { + final legacy = _payload(nights: 1348); // no folded_days key + expect(SleepProfilePolicy.isLegacy(legacy), isTrue); + expect(SleepProfilePolicy.usableProfileJson(legacy), isNull); + }); + + test('a tracked profile survives unchanged', () { + final tracked = _payload(foldedDays: ['2026-07-30'], nights: 1); + expect(SleepProfilePolicy.isLegacy(tracked), isFalse); + expect(SleepProfilePolicy.usableProfileJson(tracked), tracked); + }); + + test('null and corrupt payloads are cold starts but not "legacy"', () { + for (final bad in [null, '', 'not json', '[1,2,3]']) { + expect(SleepProfilePolicy.usableProfileJson(bad), isNull); + expect(SleepProfilePolicy.isLegacy(bad), isFalse); + expect(SleepProfilePolicy.foldedDays(bad), isEmpty); + } + }); + + test('a tracked-but-empty profile is usable (mid-rebuild, not legacy)', () { + final rebuilding = _payload(foldedDays: const [], nights: 0); + expect(SleepProfilePolicy.isLegacy(rebuilding), isFalse); + expect(SleepProfilePolicy.usableProfileJson(rebuilding), rebuilding); + }); + }); + + group('concurrent-fold semantics (DB transaction contract)', () { + // The real serialization is an exclusive SQLite transaction in + // LocalDb.updateBaseline — a Dart lock cannot span isolates. What is + // testable here without a DB is the PURE contract the transaction body + // relies on: given the payload as it exists at commit time, decide once. + // + // These model the transaction body running serially (which is what the + // exclusive write lock guarantees) and assert the outcome is correct for + // any interleaving. + + String? foldInto(String? current, String dayId) { + final usable = SleepProfilePolicy.usableProfileJson(current); + final days = SleepProfilePolicy.foldedDays(usable); + if (!SleepProfilePolicy.shouldFold( + alreadyFolded: days, dayId: dayId, hasOverride: false)) { + return null; // leave the row untouched + } + final nights = usable == null + ? 0 + : ((jsonDecode(usable) as Map)['nights'] as num?)?.toInt() ?? 0; + return jsonEncode(SleepProfilePolicy.withFoldedDays( + {'nights': nights + 1}, days, dayId)); + } + + test('two lanes folding the SAME day commit exactly one fold', () { + // The case the old test failed to cover: BOTH lanes see an empty + // folded_days when they start. Serialized at commit time, the second + // must observe the first's write and decline. + var row = jsonEncode({'nights': 0, 'folded_days': []}); + var writes = 0; + for (var lane = 0; lane < 2; lane++) { + final next = foldInto(row, '2026-07-30'); + if (next != null) { + row = next; + writes++; + } + } + expect(writes, 1, reason: 'the second lane must see the first write'); + final decoded = jsonDecode(row) as Map; + expect(decoded['nights'], 1); + expect(decoded[SleepProfilePolicy.foldedDaysKey], ['2026-07-30']); + }); + + test('distinct days each commit once and none is lost', () { + var row = jsonEncode({'nights': 0, 'folded_days': []}); + final days = ['2026-07-28', '2026-07-29', '2026-07-30', '2026-07-31']; + for (final d in days) { + final next = foldInto(row, d); + if (next != null) row = next; + } + final decoded = jsonDecode(row) as Map; + expect(decoded['nights'], days.length); + expect((decoded[SleepProfilePolicy.foldedDaysKey] as List).cast(), + days); + }); + + test('a stale pre-staging read cannot resurrect an already-folded day', () { + // Lane A read an empty profile, went off to stage for 90s, and comes back + // to find lane B folded the same day. Re-checking against the CURRENT row + // (what the transaction body does) is what prevents the double count. + const staleView = '{"nights":0,"folded_days":[]}'; + final committed = jsonEncode({ + 'nights': 1, + 'folded_days': const ['2026-07-30'], + }); + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: SleepProfilePolicy.foldedDays(staleView), + dayId: '2026-07-30', + hasOverride: false, + ), + isTrue, + reason: 'the stale view alone would wrongly permit a second fold', + ); + expect(foldInto(committed, '2026-07-30'), isNull, + reason: 'deciding against the committed row declines correctly'); + }); + + test('a legacy row is discarded, not merged into', () { + final legacy = jsonEncode({'nights': 1348}); // no folded_days + final next = foldInto(legacy, '2026-07-30'); + expect(next, isNotNull); + final decoded = jsonDecode(next!) as Map; + expect(decoded['nights'], 1, + reason: 'rebuild from cold start, not from 1348'); + expect(decoded[SleepProfilePolicy.foldedDaysKey], ['2026-07-30']); + }); + }); + + group('folded-day bookkeeping', () { + test('append is sorted and de-duplicated', () { + final out = SleepProfilePolicy.appendFoldedDay( + {'2026-07-31', '2026-07-29'}, '2026-07-30'); + expect(out, ['2026-07-29', '2026-07-30', '2026-07-31']); + expect(SleepProfilePolicy.appendFoldedDay(out.toSet(), '2026-07-30'), + hasLength(3)); + }); + + test('eviction is chronological for real day labels', () { + // The cap sorts lexicographically, which only means "age" for zero-padded + // ISO dates. Pin it across a month and year boundary, where a naive + // non-padded format would misorder. + var days = {}; + for (final d in [ + '2025-12-30', + '2025-12-31', + '2026-01-01', + '2026-01-02', + '2026-01-09', + '2026-01-10', + ]) { + days = {...SleepProfilePolicy.appendFoldedDay(days, d)}; + } + expect(days.toList(), [ + '2025-12-30', + '2025-12-31', + '2026-01-01', + '2026-01-02', + '2026-01-09', + '2026-01-10', + ]); + }); + + test('a non-date day_id trips the precondition', () { + // A UUID or epoch string would break the sort, so a RECENT day could be + // evicted and then re-folded — the exact bug this class prevents. + expect( + () => SleepProfilePolicy.appendFoldedDay(const {}, '1785522024'), + throwsA(isA()), + ); + expect( + () => SleepProfilePolicy.appendFoldedDay(const {}, '2026-7-4'), + throwsA(isA()), + reason: 'unpadded dates sort wrong too', + ); + }); + + test('the set is capped, evicting the oldest', () { + String label(int i) { + final d = DateTime.utc(2020, 1, 1).add(Duration(days: i)); + return '${d.year.toString().padLeft(4, '0')}-' + '${d.month.toString().padLeft(2, '0')}-' + '${d.day.toString().padLeft(2, '0')}'; + } + + const overflow = 50; + const total = SleepProfilePolicy.maxFoldedDays + overflow; + var days = {}; + for (var i = 0; i < total; i++) { + days = {...SleepProfilePolicy.appendFoldedDay(days, label(i))}; + } + expect(days, hasLength(SleepProfilePolicy.maxFoldedDays)); + expect(days.contains(label(0)), isFalse, reason: 'oldest evicted'); + expect(days.contains(label(overflow - 1)), isFalse, + reason: 'everything past the cap is evicted, oldest first'); + expect(days.contains(label(overflow)), isTrue, + reason: 'the first surviving day'); + expect(days.contains(label(total - 1)), isTrue, reason: 'newest kept'); + }); + + test('withFoldedDays stamps the key without disturbing profile fields', () { + final stamped = SleepProfilePolicy.withFoldedDays( + {'nights': 4, 'hr_sleep_median': 52.5}, + {'2026-07-30'}, + '2026-07-31', + ); + expect(stamped['nights'], 4); + expect(stamped['hr_sleep_median'], 52.5); + expect(stamped[SleepProfilePolicy.foldedDaysKey], + ['2026-07-30', '2026-07-31']); + }); + + test('round-trips through JSON so the next pass reads what we wrote', () { + final stamped = SleepProfilePolicy.withFoldedDays( + {'nights': 1}, const {}, '2026-07-31'); + final reread = SleepProfilePolicy.foldedDays(jsonEncode(stamped)); + expect(reread, {'2026-07-31'}); + expect( + SleepProfilePolicy.shouldFold( + alreadyFolded: reread, dayId: '2026-07-31', hasOverride: false), + isFalse, + reason: 'the day we just folded must not fold again next pass', + ); + }); + }); +}