Skip to content
64 changes: 54 additions & 10 deletions lib/compute/derivation_engine.dart
Original file line number Diff line number Diff line change
Expand Up @@ -3681,7 +3681,7 @@ class DerivationEngine {
onDayDone?.call(dayId, completed, orderedDays.length);
}

await runWithConcurrency(orderedDays, _deriveConcurrency, processDay);
await _runDayPool(orderedDays, processDay);
// BEFORE the prune below: a day still owed a strain score is re-scored
// while its raw exists (and the prune holds it if this fails).
_diag['stage'] = 'quiet_repass';
Expand Down Expand Up @@ -3928,7 +3928,7 @@ class DerivationEngine {
onDayDone?.call(dayId, completed, orderedDays.length);
}

await runWithConcurrency(orderedDays, _deriveConcurrency, processDay);
await _runDayPool(orderedDays, processDay);
done += await _quietLevelRepass(profile, dataNowSec);
// A SELECTED re-analyze that happens to cover the whole raw history, with
// every day resolved, is a full restage by any other name — it re-derived
Expand Down Expand Up @@ -4060,9 +4060,15 @@ class DerivationEngine {
}
// Tonight's sleep is the NEXT day's main sleep; it is only known once
// that day has derived, and this day keeps re-deriving until it
// finalizes, so it converges on the following pass.
// finalizes, so it converges on the following pass. A day that would
// finalize in the same pass its next day is still deriving is held back
// ([PreparedDerivationDay.tonightPending]), or which of the two finished
// first would decide its stored result for good.
final nextDayId = localDateLabel(dayEnd);
final nextSpan =
storedWindowSpan(await LocalDb.sleepWindowJsonFor(nextDayId));
final tonight = tonightSleep(
storedWindowSpan(await LocalDb.sleepWindowJsonFor(localDateLabel(dayEnd))),
nextSpan,
dayStartSec: dayStart,
dayEndSec: dayEnd,
sleepOffsetSec: candidate.present ? candidate.sleepOffsetSec : 0,
Expand All @@ -4073,6 +4079,11 @@ class DerivationEngine {
sleepSub: sleepSub,
tonightSleepOnsetSec: tonight?.startSec ?? 0,
tonightSleepOffsetSec: tonight?.endSec ?? 0,
// No span, not just no row: a stored '{}' is no window either. Not
// `tonight == null`: a next night that starts after midnight has a
// window but no tonight, and holding on it would hold every such day.
tonightPending: (nextSpan == null && _passUnfinished.contains(nextDayId)) ||
debugHoldTonight.contains(dayId),
ownership: ownership,
priority: priority,
);
Expand Down Expand Up @@ -5247,7 +5258,7 @@ class DerivationEngine {
)..sort((a, b) => b.compareTo(a)); // newest first, like the sweep
if (days.isEmpty) return 0;
_log('derive: re-scoring ${days.length} day(s) owed a quiet level');
await runWithConcurrency(days, _deriveConcurrency, (dayId) async {
await _runDayPool(days, (dayId) async {
try {
await debugQuietRepassHook?.call(dayId);
final prepared = await _prepareTargetDay(dayId);
Expand Down Expand Up @@ -5394,7 +5405,7 @@ class DerivationEngine {
onDayDone?.call(dayId, completed, orderedDays.length);
}

await runWithConcurrency(orderedDays, _deriveConcurrency, processDay);
await _runDayPool(orderedDays, processDay);

if (orderedDays.isNotEmpty) {
await _refreshBaselines();
Expand Down Expand Up @@ -5603,6 +5614,33 @@ class DerivationEngine {
/// real wall-clock speedup from the device's other cores — in the FOREGROUND.
/// A headless background slot has no spare cores to soak up, so it takes one
/// lane; [DerivePacing] owns that decision and explains it.
/// The days this pass's pool has not finished yet ([_runDayPool]).
final Set<String> _passUnfinished = {};

/// Test seam: days prepared as [PreparedDerivationDay.tonightPending]
/// whatever the next day's state, so a hold can be exercised without
/// racing two workers. Empty in production.
@visibleForTesting
static Set<String> debugHoldTonight = {};

/// [runWithConcurrency] over [days], tracking which are still unfinished so
/// a day can tell its next day is mid-derive ([_prepareTargetDay]).
Future<void> _runDayPool(
List<String> days, Future<void> Function(String dayId) body) async {
_passUnfinished.addAll(days);
try {
await runWithConcurrency(days, _deriveConcurrency, (dayId) async {
try {
await body(dayId);
} finally {
_passUnfinished.remove(dayId);
}
});
} finally {
_passUnfinished.removeAll(days);
}
}

int get _deriveConcurrency {
try {
return _pacing.concurrency(Platform.numberOfProcessors);
Expand Down Expand Up @@ -6424,8 +6462,9 @@ class DerivationEngine {
// forceFinalize. Locking an all-absent row is what made the destructive
// re-analyze permanent; leaving it unlocked means a later pass (or a
// restored/backfilled substrate) can still fill the day in.
final finalized =
!producedNothing && (forceFinalize || (ageFinalized && secondHalfOk));
final finalized = !producedNothing &&
(forceFinalize ||
(ageFinalized && secondHalfOk && !day.tonightPending));

// A failed/timed-out second half yields a headline-only bundle, and
// putDayResult replaces the row wholesale — so re-deriving an already
Expand Down Expand Up @@ -6454,7 +6493,10 @@ class DerivationEngine {
finalizedByAge: finalized,
);
effectivePartial = outcome.partial;
effectiveFinalized = outcome.finalized;
// The tonight hold applies here too: carrying a finalized row
// forward would otherwise lock a day the hold kept open.
effectiveFinalized =
outcome.finalized && (forceFinalize || !day.tonightPending);
_log('derive ${day.date}: second half failed — carried the previous '
"result's detail blocks forward (v$prevVersion -> v$kAlgoVersion, "
'partial=$effectivePartial)');
Expand Down Expand Up @@ -6693,7 +6735,9 @@ class DerivationEngine {
],
);
}
return secondHalfOk;
// A held day is not settled either: it must derive again next pass to
// read tonight's window, so it may not be recorded as unchanged.
return secondHalfOk && !day.tonightPending;
}

/// Bank one day's (phone steps, counter ticks) pair for the (family,
Expand Down
10 changes: 10 additions & 0 deletions lib/compute/derive_prepare.dart
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,11 @@ class PreparedDerivationDay {
final int tonightSleepOnsetSec;
final int tonightSleepOffsetSec;

/// The next day is being derived in this same pass and has no stored
/// window yet, so tonight's sleep is unknown only for now. Such a day is not
/// finalized this pass: the next pass reads the window and locks it then.
final bool tonightPending;

/// M5: the resolved exclusive-owner spans, one entry per anchor signal
/// (`hr1Hz`, `rrIntervals`), over the union window `_prepareTargetDay`
/// resolved once. Empty (never populated) on the import path — a bulk
Expand Down Expand Up @@ -73,6 +78,7 @@ class PreparedDerivationDay {
this.sleepSource = 'auto',
this.tonightSleepOnsetSec = 0,
this.tonightSleepOffsetSec = 0,
this.tonightPending = false,
this.ownership = const {},
this.priority = const {},
}) : napSub = napSub ?? daySub;
Expand All @@ -92,6 +98,7 @@ class PreparedDerivationDay {
'nap_sub': napSub.toJson(),
'tonight_sleep_onset_sec': tonightSleepOnsetSec,
'tonight_sleep_offset_sec': tonightSleepOffsetSec,
'tonight_pending': tonightPending,
// ponytail: `ownership` is NOT round-tripped here. Grepped for real
// callers of PreparedDerivationDay.toJson()/fromJson() outside this file
// — none exist (only `candidate.toPreparedDay(...)` at
Expand Down Expand Up @@ -120,6 +127,7 @@ class PreparedDerivationDay {
(m['tonight_sleep_onset_sec'] as num?)?.toInt() ?? 0,
tonightSleepOffsetSec:
(m['tonight_sleep_offset_sec'] as num?)?.toInt() ?? 0,
tonightPending: m['tonight_pending'] == true,
daySub: daySub,
sleepSub: Substrate.fromJson(
((m['sleep_sub'] as Map?) ?? const {}).cast<String, dynamic>(),
Expand Down Expand Up @@ -260,6 +268,7 @@ class SleepSessionCandidate {
Substrate? napSub,
int tonightSleepOnsetSec = 0,
int tonightSleepOffsetSec = 0,
bool tonightPending = false,
Map<InputSignal, List<OwnedSpan>> ownership = const {},
Map<InputSignal, List<String>> priority = const {},
}) => PreparedDerivationDay(
Expand All @@ -286,6 +295,7 @@ class SleepSessionCandidate {
sleepSub: sleepSub,
tonightSleepOnsetSec: tonightSleepOnsetSec,
tonightSleepOffsetSec: tonightSleepOffsetSec,
tonightPending: tonightPending,
ownership: ownership,
priority: priority,
);
Expand Down
118 changes: 118 additions & 0 deletions test/derive_tonight_pending_test.dart
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
// A day old enough to finalize, whose next day has not stored its window yet
// in the same pass, is held: not finalized and not recorded as unchanged, so
// the next pass reads tonight's window and locks the day then. The hold is
// forced through the test seam rather than by racing two workers.

import 'package:flutter_test/flutter_test.dart';
import 'package:openstrap_edge/compute/derivation_engine.dart';
import 'package:openstrap_edge/compute/profile.dart';
import 'package:openstrap_edge/data/db.dart';
import 'package:path/path.dart' as p;
import 'package:shared_preferences/shared_preferences.dart';
import 'package:sqflite_common_ffi/sqflite_ffi.dart';

import 'support/synthetic_day.dart';

const _profile = Profile(ageYears: 32, weightKg: 72, heightCm: 176, sex: 'm');

// Three days of data: the day before the first is old enough to finalize on
// the first pass.
final _days = [
for (var i = 0; i < 3; i++) SyntheticDay(DateTime(2026, 7, 13 + i)),
];
const _held = '2026-07-12';

Future<void> _seed(String name) async {
SharedPreferences.setMockInitialValues({});
await LocalDb.close();
LocalDb.dbName = name;
await databaseFactory.deleteDatabase(
p.join(await databaseFactory.getDatabasesPath(), name),
);
var committed = 0;
for (final d in _days) {
for (final (raws, samples) in d.primaryBatches()) {
final keep = [
for (var j = 0; j < raws.length; j++)
if (raws[j].recTs! > committed) j,
];
if (keep.isEmpty) continue;
await LocalDb.commitSyncBatch(
[for (final j in keep) raws[j]],
[for (final j in keep) samples[j]],
deviceFamily: 'gen4',
);
committed = raws[keep.last].recTs!;
}
}
}

void main() {
setUpAll(() {
sqfliteFfiInit();
databaseFactory = databaseFactoryFfi;
});

tearDown(() async {
DerivationEngine.debugHoldTonight = {};
DerivationEngine.debugSecondHalfHook = null;
await LocalDb.close();
});

test(
'a held day is neither finalized nor fingerprinted, then settles',
() async {
await _seed('derive_tonight_pending.db');

// Two held passes: the second derives from unchanged inputs, which is
// where a fingerprint would be recorded and the day skipped for good.
DerivationEngine.debugHoldTonight = {_held};
for (var i = 0; i < 2; i++) {
await DerivationEngine().run(_profile, heavy: true);
expect(await LocalDb.dayResult(_held), isNotNull);
expect(
await LocalDb.finalizedDayIds(kAlgoVersion),
isNot(contains(_held)),
reason: 'held pass $i',
);
expect(
await LocalDb.deriveFingerprint(_held),
isNull,
reason: 'held pass $i',
);
}

DerivationEngine.debugHoldTonight = {};
await DerivationEngine().run(_profile, heavy: true);
expect(await LocalDb.finalizedDayIds(kAlgoVersion), contains(_held));
},
timeout: const Timeout(Duration(minutes: 5)),
);

test(
'a failed second half does not carry a finalized row past the hold',
() async {
await _seed('derive_tonight_pending_recovery.db');
// A cold first pass may legitimately hold the day (the next day's
// window can land after it in the same pass), so a second pass is what
// guarantees the finalized row this test starts from.
await DerivationEngine().run(_profile, heavy: true);
await DerivationEngine().run(_profile, heavy: true);
expect(await LocalDb.finalizedDayIds(kAlgoVersion), contains(_held));

// Re-derived (forced) while held, with its second half failing: the
// recovery path carries the complete finalized row forward, and must not
// carry the lock with it.
DerivationEngine.debugHoldTonight = {_held};
DerivationEngine.debugSecondHalfHook = (day) async {
if (day == _held) throw StateError('second half down');
};
await DerivationEngine().runDays(_profile, {_held}, force: true);
expect(
await LocalDb.finalizedDayIds(kAlgoVersion),
isNot(contains(_held)),
);
},
timeout: const Timeout(Duration(minutes: 5)),
);
}
63 changes: 61 additions & 2 deletions test/derive_unchanged_skip_test.dart
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,10 @@ Future<String> _snapshot() async {
if (!e.key.endsWith('_at') && e.key != 'updated_ts')
e.key: e.value is String
? (e.value as String).replaceAll(
RegExp(r'"(updated|built)[A-Za-z_]*":\s*\d+'), '')
// Numbers and dates: built_for_day is the wall-clock
// day, so two halves either side of midnight differ.
RegExp(r'"(updated|built)[A-Za-z_]*":\s*(\d+|"[^"]*")'),
'')
: e.value,
},
];
Expand All @@ -88,6 +91,59 @@ Future<String> _snapshot() async {
});
}

/// Where two [_snapshot]s first part: the table, the row, and each column
/// that differs, values cut short.
String _firstDifference(String a, String b) {
String cut(Object? v) {
final t = '$v';
return t.length <= 400 ? t : '${t.substring(0, 400)}… (${t.length} chars)';
}

final x = jsonDecode(a) as Map<String, dynamic>;
final y = jsonDecode(b) as Map<String, dynamic>;
for (final table in x.keys) {
final rx = x[table] as List, ry = y[table] as List;
if (rx.length != ry.length) {
return '$table: ${rx.length} rows with the skip, ${ry.length} without';
}
for (var i = 0; i < rx.length; i++) {
final m = rx[i] as Map, n = ry[i] as Map;
final cols = {...m.keys, ...n.keys}
.where((k) => jsonEncode(m[k]) != jsonEncode(n[k]))
.toList();
if (cols.isEmpty) continue;
final key = m['day_id'] ?? m['date'] ?? m['key'] ?? m['id'] ?? i;
return [
'$table row $i ($key) differs in ${cols.join(', ')}:',
for (final c in cols)
if (m[c] is String && n[c] is String)
_stringDifference(c as String, m[c] as String, n[c] as String)
else ...[
' $c with the skip: ${cut(m[c])}',
' $c without: ${cut(n[c])}',
],
].join('\n');
}
}
return 'snapshots differ outside the compared tables';
}

String _stringDifference(String col, String a, String b) {
var i = 0;
while (i < a.length && i < b.length && a.codeUnitAt(i) == b.codeUnitAt(i)) {
i++;
}
String around(String t) {
final from = i < 200 ? 0 : i - 200;
final to = i + 200 > t.length ? t.length : i + 200;
return t.substring(from, to);
}

return ' $col first differs at char $i\n'
' with the skip: …${around(a)}…\n'
' without: …${around(b)}…';
}

void main() {
setUpAll(() {
sqfliteFfiInit();
Expand Down Expand Up @@ -121,7 +177,10 @@ void main() {
// The skip actually happened: by the last pass nothing was derived.
expect(log.last.derived, 0, reason: '$log');
expect(log.last.unchanged, greaterThan(0), reason: '$log');
expect(withSkip, everyDay);
// Not expect(withSkip, everyDay): on a mismatch that prints both
// multi-megabyte snapshots, and the CI log was cut off right there, so a
// failure read as a silent crash. Name the first difference instead.
if (withSkip != everyDay) fail(_firstDifference(withSkip, everyDay));
}, timeout: const Timeout(Duration(minutes: 10)));

test('an override, new data, a profile change and a version bump re-derive',
Expand Down
Loading