diff --git a/lib/compute/derivation_engine.dart b/lib/compute/derivation_engine.dart index 4a27646..0653cc6 100644 --- a/lib/compute/derivation_engine.dart +++ b/lib/compute/derivation_engine.dart @@ -391,7 +391,20 @@ import 'substrate.dart'; // hypnogram, stage minutes and sleep-derived scalars change, so every day must // re-derive. Also picks up the protocol realtimeRr bound (live HRV no longer // sees implausible sub-100ms "beats" from a misaligned 0x28 frame). -const int kAlgoVersion = 53; +// v54: NOOP CSV imports now bank the export's `step_counter` as REAL steps. +// NOOP's schema gained a `steps` stream (and a `band_sleep_state` column that +// shifted event_kind/event_payload) — the importer read columns by name so it +// never misparsed, but it dropped `steps` into its default branch and every +// imported day reported steps = 0 while the band had actually counted them +// (2,572 over the 3.5 h in the OpenStrap/edge#160 export). The counter is now +// differenced into contiguous runs and written to `live_coverage`, the same +// table the live 100 Hz pedometer uses, so imported and live days count steps +// identically and the 1 Hz estimate still cannot double-count those minutes. +// Only the `steps`/`active_min` block of IMPORTED days changes; no live-sync +// output moves. NOTE this bump does not retro-fix an existing import — imported +// days are force-finalized snapshots with no stored raw to recompute from, so +// an already-imported day needs a re-import to pick its steps up. +const int kAlgoVersion = 54; // Fold idempotency, the minimum-nights warm-up, and legacy-payload handling // all live in SleepProfilePolicy (pure, unit-tested) — see diff --git a/lib/import/noop_import.dart b/lib/import/noop_import.dart index 245b9bd..de6703b 100644 --- a/lib/import/noop_import.dart +++ b/lib/import/noop_import.dart @@ -1,10 +1,31 @@ // noop_import.dart — import a NOOP raw-sensor CSV export into the local store. // // The NOOP export is LONG-FORMAT raw 1 Hz: one row per decoded sample, with a -// `stream` discriminator (hr / rr / gravity / spo2 / skintemp / resp / event) and -// only that stream's columns filled. Header (line 4): +// `stream` discriminator and only that stream's columns filled. Columns are read +// by NAME from the header, never by fixed position — NOOP has already shipped one +// schema change (see below) and name-keyed reads absorbed it without misparsing. +// +// SCHEMA AS SHIPPED (observed 2026-08, NOOP 9.1/9.2 — OpenStrap/edge#160): // unix_s,iso_utc,stream,hr_bpm,rr_ms,grav_x,grav_y,grav_z,step_counter, -// ppg_bpm,ppg_conf,spo2_red,spo2_ir,skintemp_raw,resp_raw,event_kind,event_payload +// ppg_bpm,ppg_conf,spo2_red,spo2_ir,skintemp_raw,resp_raw,band_sleep_state, +// event_kind,event_payload +// `band_sleep_state` was INSERTED at index 15, shifting event_kind/event_payload +// to 16/17 (the older documented layout, still in [_defaultCols], ends at +// event_payload=16). Streams now seen: hr, rr, gravity, skintemp, steps, +// band_sleep_state, ppghr, event. Note `spo2` and `resp` rows are no longer +// emitted at all even though their columns survive in the header. +// +// WHAT WE CONSUME, AND WHY NOT THE REST: +// • hr / rr / gravity / skintemp / spo2 → the Substrate (full 1 Hz analytics). +// • steps → `live_coverage` as a REAL step count (see [_flushStepCoverage]). +// • band_sleep_state → deliberately ignored. `segmentSleep` is the single sleep +// source (a second one would contradict it), and in the real #160 export this +// column was constant 0 across all 12,663 rows — no signal to gain. +// • ppghr → deliberately ignored. The `hr` stream already covers every second +// the export carries, and PPG-derived HR is not trusted as a substitute. +// • resp_raw → ignored; respiration rate is derived from RR downstream. +// An unrecognised future `stream` also lands in the default branch and is skipped +// safely, so a further NOOP schema change degrades rather than throws. // // Because this is RAW 1 Hz — the SAME signal family as our own Substrate — we run // it through the FULL local pipeline (full-fidelity analytics, identical to a live @@ -22,17 +43,45 @@ import 'dart:io'; import '../compute/derivation_engine.dart'; import '../compute/profile.dart'; import '../compute/substrate.dart'; +import '../data/db.dart'; class NoopImportResult { final int days; final int rows; + /// Real steps recovered from the export's `step_counter` and banked as + /// `live_coverage` (0 when the export carries no `steps` stream). + final int steps; + /// Rows whose local date had ALREADY been derived and pruned by the time /// they appeared in the file (a genuinely out-of-order export). They cannot /// be folded back in, so they are counted here rather than silently dropped /// — a non-zero value means the export was not fully time-ordered. final int lateRows; - NoopImportResult(this.days, this.rows, [this.lateRows = 0]); + NoopImportResult(this.days, this.rows, + [this.lateRows = 0, this.steps = 0]); +} + +/// One contiguous run of the band's cumulative `step_counter`, ready to bank as +/// a `live_coverage` row: real steps over a known [startSec]..[endSec] window. +class StepRun { + final int startSec; + final int endSec; + final int steps; + const StepRun(this.startSec, this.endSec, this.steps); + + @override + String toString() => 'StepRun($startSec..$endSec, $steps)'; + + @override + bool operator ==(Object other) => + other is StepRun && + other.startSec == startSec && + other.endSec == endSec && + other.steps == steps; + + @override + int get hashCode => Object.hash(startSec, endSec, steps); } /// What to do with an incoming row given the import's high-water date. @@ -76,9 +125,18 @@ class NoopImporter { final rrMs = []; String? curDate; final derived = {}; // dates already derived + pruned out of `secs` - var totalRows = 0, daysDone = 0, lateRows = 0; + var totalRows = 0, daysDone = 0, lateRows = 0, stepsBanked = 0; + + // Band step counter, ts(sec) → cumulative value, keyed by local date so a + // run is never attributed across midnight. Flushed to `live_coverage` + // BEFORE the date derives, since the derivation reads real steps from there. + final stepsByDate = >{}; Future deriveAndPrune(String date) async { + // Real steps must be banked BEFORE deriving: _deriveDay reads them via + // LocalDb.liveStepsForDay/coverageWindowsOverlapping at derive time. + final st = stepsByDate.remove(date); + if (st != null) stepsBanked += await _flushStepCoverage(st, date); // Build a Substrate from everything buffered (prev + current date) and // derive ONLY [date]; calendarDays gives [date] its prior-evening context. final sub = _buildSubstrate(secs, rrTs, rrMs); @@ -182,22 +240,37 @@ class NoopImporter { case 'skintemp': (secs[ts] ??= _Sec()).skinTemp = int.tryParse(at(f, 'skintemp_raw')); break; - // resp / event / step / ppg: not part of the Substrate — resp rate is - // derived from RR downstream, so resp_raw is intentionally ignored. An - // unknown future `stream` value also lands here and is skipped safely. + case 'steps': + // CUMULATIVE band counter, not a per-second increment — differenced + // into real step windows at flush time (see [stepRuns]). Kept out of + // the Substrate: it is a real count, not a 1 Hz signal to analyse. + final v = int.tryParse(at(f, 'step_counter')); + if (v != null && v >= 0) (stepsByDate[date] ??= {})[ts] = v; + break; + // band_sleep_state / ppghr / resp / event: intentionally not consumed — + // see the stream inventory in the file header for why each is skipped. + // An unknown future `stream` value also lands here and is skipped safely. default: break; } } // EOF — derive the final buffered date. if (curDate != null && secs.isNotEmpty) { + final st = stepsByDate.remove(curDate); + if (st != null) stepsBanked += await _flushStepCoverage(st, curDate); final sub = _buildSubstrate(secs, rrTs, rrMs); daysDone += await engine.deriveImportedDays(sub, profile, {curDate}); onProgress?.call(daysDone); } + // Any date whose steps were buffered but which never derived (e.g. a date + // that carried ONLY a `steps` stream) still banks its real count — dropping + // it would silently lose steps the band actually measured. + for (final e in stepsByDate.entries) { + stepsBanked += await _flushStepCoverage(e.value, e.key); + } await engine.finalizeImport(profile); - return NoopImportResult(daysDone, totalRows, lateRows); + return NoopImportResult(daysDone, totalRows, lateRows, stepsBanked); } /// Build a Substrate from the buffered seconds + RR beats. Gravity / SpO₂ / @@ -258,6 +331,115 @@ class NoopImporter { ); } + /// Maximum gap (seconds) between consecutive `step_counter` samples that is + /// still treated as one continuous run. The #160 export samples steps every + /// second, but drops out whenever the band is off-wrist or unsynced; a gap + /// wider than this is a hole we know nothing about, so we refuse to span it. + static const int stepRunMaxGapSec = 60; + + /// Turn the band's CUMULATIVE `step_counter` samples into discrete + /// [StepRun]s that can be banked as real (non-estimated) step counts. + /// + /// [samples] is ts(sec) → counter value, any order. Runs are split on a gap + /// wider than [stepRunMaxGapSec], so a 20 h hole in the export (exactly what + /// the #160 file has) never becomes one window claiming to cover the day. + /// + /// Only POSITIVE deltas within a run are summed: the counter resets to 0 on a + /// band reboot, and a negative delta is that reset, not −24,000 steps. Deltas + /// ACROSS a run boundary are deliberately NOT counted — we cannot attribute + /// steps to a window we have no samples for, and inventing that attribution is + /// exactly the kind of fabrication the honesty contract forbids. + /// + /// A run with zero steps still yields no window: `live_coverage` exists to + /// suppress the 1 Hz estimate over minutes the real counter already covered, + /// and a 0-step run would suppress a real estimate while contributing nothing. + /// + /// [covered] lists time spans ALREADY banked in `live_coverage` (device-time + /// seconds, inclusive). A per-second delta whose interval intersects one of + /// them is skipped and BREAKS the run, so those steps are never banked twice. + /// This is what makes a re-import safe in general rather than only for a + /// byte-identical file: an exact-window check alone is defeated the moment the + /// user exports again over a longer span (09:00-09:20 then 09:00-09:40), where + /// the run boundary shifts, no exact window matches, and the overlap is banked + /// a second time — measured at 3,598 steps against a true 2,399 before this. + /// Clipping is exact, not pro-rated: we hold the counter value at every + /// second, so the uncovered sub-intervals are summed from real deltas. + /// It also means an imported span cannot double-count against a LIVE 100 Hz + /// pedometer window, which shares this table. + static List stepRuns( + Map samples, { + List> covered = const [], + }) { + if (samples.isEmpty) return const []; + final ts = samples.keys.toList()..sort(); + + // Does the half-open delta interval (a, b] touch anything already banked? + bool isCovered(int a, int b) { + for (final w in covered) { + if (b > w[0] && a < w[1]) return true; + } + return false; + } + + final out = []; + int? runStart, runLast; + var runSteps = 0; + void closeRun() { + if (runStart != null && + runLast != null && + runSteps > 0 && + runLast! > runStart!) { + out.add(StepRun(runStart!, runLast!, runSteps)); + } + runStart = null; + runLast = null; + runSteps = 0; + } + + for (var i = 1; i <= ts.length; i++) { + final bankable = i < ts.length && + ts[i] - ts[i - 1] <= stepRunMaxGapSec && + !isCovered(ts[i - 1], ts[i]); + if (!bankable) { + closeRun(); + continue; + } + runStart ??= ts[i - 1]; + final d = samples[ts[i]]! - samples[ts[i - 1]]!; + if (d > 0) runSteps += d; + runLast = ts[i]; + } + return out; + } + + /// Bank [date]'s step runs into `live_coverage` so the derivation picks them up + /// as REAL steps (`liveStepsForDay`) and excludes those minutes from the 1 Hz + /// estimate (`coverageWindowsOverlapping`) — the same contract the live 100 Hz + /// pedometer uses, so imported and live days are counted identically. + /// + /// IDEMPOTENT BY TIME SPAN, not by exact window: `live_coverage` is an + /// append-only SUM with no uniqueness constraint, so anything already banked + /// over a second must not be banked again. The spans covering this batch are + /// read back and passed to [stepRuns], which clips them out. An exact-window + /// check is deliberately NOT used — it only catches a byte-identical + /// re-import and silently double-counts the overlap when the user exports + /// again over a longer span (see the note on [stepRuns]). + /// + /// Returns the steps actually banked (0 when the span was fully covered). + static Future _flushStepCoverage( + Map stepSamples, String date) async { + if (stepSamples.isEmpty) return 0; + final ts = stepSamples.keys.toList()..sort(); + final existing = + await LocalDb.coverageWindowsOverlapping(ts.first, ts.last + 1); + var banked = 0; + for (final r in stepRuns(stepSamples, covered: existing)) { + await LocalDb.addLiveCoverage(r.startSec, r.endSec, r.steps, date); + banked += r.steps; + } + return banked; + } + /// String date compare 'YYYY-MM-DD' — true when [a] is strictly after [b]. static bool _after(String a, String b) => a.compareTo(b) > 0; diff --git a/test/noop_schema_drift_test.dart b/test/noop_schema_drift_test.dart new file mode 100644 index 0000000..476bbfd --- /dev/null +++ b/test/noop_schema_drift_test.dart @@ -0,0 +1,316 @@ +// Regression tests for NOOP CSV schema drift (OpenStrap/edge#160). +// +// NOOP shipped a schema change that these tests pin: +// • `band_sleep_state` INSERTED at index 15, shifting event_kind/event_payload +// to 16/17 — the importer must keep reading by NAME, not position. +// • new streams `steps` / `band_sleep_state` / `ppghr`. +// • `spo2` and `resp` rows no longer emitted at all. +// +// Before this, `steps` fell into the importer's default branch: every imported +// day reported 0 steps while the band's own counter had measured thousands. +// +// There was no test that parsed a real NOOP CSV at all — only the pure +// `decideRow` ordering contract — which is why the drift shipped unnoticed. + +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:sqflite_common_ffi/sqflite_ffi.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:openstrap_edge/import/noop_import.dart'; + +/// The CURRENT NOOP header — `band_sleep_state` at 15, event_* shifted to 16/17. +const _header = 'unix_s,iso_utc,stream,hr_bpm,rr_ms,grav_x,grav_y,grav_z,' + 'step_counter,ppg_bpm,ppg_conf,spo2_red,spo2_ir,skintemp_raw,resp_raw,' + 'band_sleep_state,event_kind,event_payload'; + +/// Build one long-format row with only [stream]'s columns filled. +String _row(int ts, String stream, + {String hr = '', + String rr = '', + String gx = '', + String gy = '', + String gz = '', + String stepCounter = '', + String skinTemp = '', + String sleepState = '', + String eventKind = '', + String eventPayload = ''}) { + final iso = DateTime.fromMillisecondsSinceEpoch(ts * 1000, isUtc: true) + .toIso8601String(); + return [ + '$ts', iso, stream, hr, rr, gx, gy, gz, stepCounter, '', '', '', '', + skinTemp, '', sleepState, eventKind, eventPayload, + ].join(','); +} + +void main() { + group('stepRuns — cumulative counter → real step windows (pure)', () { + test('sums positive deltas across one contiguous run', () { + // 100 → 130 over 4 s = 30 steps, one window. + final runs = NoopImporter.stepRuns({0: 100, 1: 110, 2: 120, 3: 130}); + expect(runs, [const StepRun(0, 3, 30)]); + }); + + test('splits on a gap wider than stepRunMaxGapSec', () { + // The #160 export has a 20.5 h hole; a single window spanning it would + // claim to cover — and therefore suppress the 1 Hz estimate over — a day + // we have no samples for. + final runs = NoopImporter.stepRuns({ + 0: 100, 1: 110, // run A: +10 + 5000: 500, 5001: 520, // run B: +20 + }); + expect(runs, [const StepRun(0, 1, 10), const StepRun(5000, 5001, 20)]); + }); + + test('a counter RESET is not a negative step count', () { + // Band reboot: 900 → 0. Must not subtract 900, and must not fabricate. + final runs = NoopImporter.stepRuns({0: 890, 1: 900, 2: 0, 3: 15}); + expect(runs, [const StepRun(0, 3, 25)]); // +10 then +15, reset ignored + }); + + test('does not attribute steps across a run boundary', () { + // 100 → 900 happened DURING a 5000 s hole. We cannot say when, so the + // delta is dropped rather than pinned to either window. + final runs = NoopImporter.stepRuns({0: 100, 5000: 900, 5001: 905}); + expect(runs, [const StepRun(5000, 5001, 5)]); + }); + + test('emits nothing for a zero-step run', () { + // A 0-step window would suppress a real 1 Hz estimate for no gain. + expect(NoopImporter.stepRuns({0: 100, 1: 100, 2: 100}), isEmpty); + }); + + test('handles empty / single-sample input', () { + expect(NoopImporter.stepRuns({}), isEmpty); + expect(NoopImporter.stepRuns({5: 100}), isEmpty); + }); + + test('is order-independent', () { + final a = NoopImporter.stepRuns({3: 130, 0: 100, 2: 120, 1: 110}); + expect(a, [const StepRun(0, 3, 30)]); + }); + + test('skips deltas already covered, and breaks the run there', () { + // 0..5 covered → only the 5→8 tail is bankable. + final runs = NoopImporter.stepRuns( + {0: 100, 1: 110, 2: 120, 5: 150, 6: 160, 7: 170, 8: 180}, + covered: const [ + [0, 5] + ], + ); + expect(runs, [const StepRun(5, 8, 30)]); + }); + + test('a fully covered span banks nothing', () { + expect( + NoopImporter.stepRuns({0: 100, 1: 110, 2: 120}, + covered: const [ + [0, 2] + ]), + isEmpty, + ); + }); + + test('a covered span in the MIDDLE splits into two runs', () { + final runs = NoopImporter.stepRuns( + {0: 100, 1: 110, 2: 120, 3: 130, 4: 140, 5: 150}, + covered: const [ + [2, 3] + ], + ); + expect(runs, [const StepRun(0, 2, 20), const StepRun(3, 5, 20)]); + }); + }); + + group('end-to-end import of the CURRENT NOOP schema', () { + late Directory tmp; + + setUpAll(() async { + sqfliteFfiInit(); + databaseFactory = databaseFactoryFfi; + LocalDb.dbName = 'openstrap_noop_drift_test.db'; + final dir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); + tmp = await Directory.systemTemp.createTemp('noop_drift'); + }); + + tearDownAll(() async { + await LocalDb.close(); + final dir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); + if (tmp.existsSync()) tmp.deleteSync(recursive: true); + }); + + /// ~40 min of 1 Hz data in the new schema, walking the step counter by 1/s. + File writeCsv(String name, {required int t0, required int seconds}) { + final b = StringBuffer()..writeln(_header); + for (var i = 0; i < seconds; i++) { + final ts = t0 + i; + b.writeln(_row(ts, 'hr', hr: '${60 + (i % 20)}')); + b.writeln(_row(ts, 'gravity', gx: '0.1', gy: '0.2', gz: '0.97')); + b.writeln(_row(ts, 'skintemp', skinTemp: '3240')); + b.writeln(_row(ts, 'steps', stepCounter: '${24302 + i}')); + b.writeln(_row(ts, 'band_sleep_state', sleepState: '0')); + if (i % 5 == 0) b.writeln(_row(ts, 'ppghr')); + } + // An event row whose QUOTED payload contains commas — the naive + // `split(',')` must not let this corrupt unix_s / stream (indexes 0/2). + b.writeln('${t0 + 1},x,event,,,,,,,,,,,,,,BATTERY_LEVEL(3),' + '"{""battery_mV"":4155,""battery_pct"":78.4}"'); + final f = File(p.join(tmp.path, name))..writeAsStringSync(b.toString()); + return f; + } + + test('banks the band step counter as REAL steps, and is idempotent', + () async { + // 2026-07-31T09:00:00Z, 2400 s of data. + const t0 = 1785488400; + const secs = 2400; + final csv = writeCsv('a.csv', t0: t0, seconds: secs); + + final res = await NoopImporter.importFile( + csv.path, const Profile(), DerivationEngine()); + + // The whole point: steps are no longer silently dropped. + expect(res.steps, secs - 1, reason: 'counter advanced 1/s for $secs s'); + expect(res.days, greaterThan(0)); + expect(res.lateRows, 0); + + // Banked where the derivation actually reads real steps from. + final db = await LocalDb.instance; + final cov = await db.query('live_coverage'); + expect(cov, isNotEmpty); + final total = cov.fold(0, (a, r) => a + (r['steps'] as int)); + expect(total, secs - 1); + + // RE-IMPORT must not double-count: live_coverage is an append-only SUM + // with no uniqueness on the window. + final res2 = await NoopImporter.importFile( + csv.path, const Profile(), DerivationEngine()); + expect(res2.steps, 0, reason: 're-import banks nothing new'); + final cov2 = await db.query('live_coverage'); + expect(cov2.length, cov.length, reason: 'no duplicate windows'); + final total2 = cov2.fold(0, (a, r) => a + (r['steps'] as int)); + expect(total2, total, reason: 'step total unchanged after re-import'); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('an OVERLAPPING re-export does not double-count', () async { + // The realistic re-import: the user exports again later over a LONGER + // span covering the same session. An exact-window guard misses this + // entirely (the run boundary moved), so the overlap gets banked twice — + // measured at 3,598 against a true 2,399 before the covered-clipping fix. + const t0 = 1785834000; + final short = writeCsv('ov_short.csv', t0: t0, seconds: 1200); + final long = writeCsv('ov_long.csv', t0: t0, seconds: 2400); + + final r1 = await NoopImporter.importFile( + short.path, const Profile(), DerivationEngine()); + expect(r1.steps, 1199); + + final r2 = await NoopImporter.importFile( + long.path, const Profile(), DerivationEngine()); + // Only the NEW tail is banked, not the whole longer span. + expect(r2.steps, 2399 - 1199, + reason: 'second import banks only the previously uncovered tail'); + + final db = await LocalDb.instance; + final cov = await db.query('live_coverage', + where: 'start_ts >= ? AND start_ts < ?', whereArgs: [t0, t0 + 2400]); + final total = cov.fold(0, (a, r) => a + (r['steps'] as int)); + expect(total, 2399, reason: 'total equals the truth, not 3598'); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('a PARTIALLY flushed import self-heals on re-import', () async { + // Covered-clipping is keyed by TIME SPAN, not by an exact window row, so + // a flush interrupted between two runs is recoverable: the run that never + // landed is not covered, and the next import banks it. This is why the + // flush does not need to be transactional. + const t0 = 1785920400; + final b = StringBuffer()..writeln(_header); + void block(int start, int n, int base) { + for (var i = 0; i < n; i++) { + b.writeln(_row(start + i, 'hr', hr: '65')); + b.writeln( + _row(start + i, 'gravity', gx: '0.1', gy: '0.2', gz: '0.97')); + b.writeln(_row(start + i, 'steps', stepCounter: '${base + i}')); + } + } + + block(t0, 600, 1000); // run A: 599 steps + block(t0 + 1200, 600, 2000); // run B: 599 steps, past the 60 s split + final f = File(p.join(tmp.path, 'partial.csv')) + ..writeAsStringSync(b.toString()); + + final r1 = await NoopImporter.importFile( + f.path, const Profile(), DerivationEngine()); + expect(r1.steps, 1198); + + final db = await LocalDb.instance; + final cov = await db.query('live_coverage', + where: 'start_ts >= ? AND start_ts < ?', + whereArgs: [t0, t0 + 2000], + orderBy: 'start_ts'); + expect(cov.length, 2); + + // Simulate a crash after run A was written but before run B. + await db.delete('live_coverage', + where: 'start_ts = ?', whereArgs: [cov.last['start_ts']]); + + final r2 = await NoopImporter.importFile( + f.path, const Profile(), DerivationEngine()); + expect(r2.steps, 599, reason: 'the lost run is re-banked, and only it'); + + final after = await db.query('live_coverage', + where: 'start_ts >= ? AND start_ts < ?', whereArgs: [t0, t0 + 2000]); + final total = after.fold(0, (a, r) => a + (r['steps'] as int)); + expect(total, 1198, reason: 'back to truth, with no double-count'); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('shifted event_kind/event_payload columns do not misparse', () async { + // `band_sleep_state` at 15 pushes event_* to 16/17. Reading by NAME means + // hr/gravity/steps still land correctly; a positional reader would not. + const t0 = 1785574800; // a different day, so it derives independently + final csv = writeCsv('b.csv', t0: t0, seconds: 600); + final res = await NoopImporter.importFile( + csv.path, const Profile(), DerivationEngine()); + expect(res.steps, 599); + expect(res.rows, greaterThan(0)); + expect(res.lateRows, 0); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('an export with NO steps stream still imports (steps = 0)', () async { + // `spo2`/`resp` already vanished from the schema; `steps` could too. + const t0 = 1785661200; + final b = StringBuffer()..writeln(_header); + for (var i = 0; i < 600; i++) { + b.writeln(_row(t0 + i, 'hr', hr: '65')); + b.writeln(_row(t0 + i, 'gravity', gx: '0.1', gy: '0.2', gz: '0.97')); + } + final f = File(p.join(tmp.path, 'c.csv')) + ..writeAsStringSync(b.toString()); + final res = await NoopImporter.importFile( + f.path, const Profile(), DerivationEngine()); + expect(res.steps, 0); + expect(res.days, greaterThan(0)); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('an UNKNOWN future stream is skipped, not fatal', () async { + const t0 = 1785747600; + final b = StringBuffer()..writeln(_header); + for (var i = 0; i < 300; i++) { + b.writeln(_row(t0 + i, 'hr', hr: '65')); + b.writeln(_row(t0 + i, 'gravity', gx: '0.1', gy: '0.2', gz: '0.97')); + b.writeln(_row(t0 + i, 'some_future_stream_we_have_never_seen')); + } + final f = File(p.join(tmp.path, 'd.csv')) + ..writeAsStringSync(b.toString()); + final res = await NoopImporter.importFile( + f.path, const Profile(), DerivationEngine()); + expect(res.days, greaterThan(0)); + }, timeout: const Timeout(Duration(minutes: 5))); + }); +}