diff --git a/Packages/StrandAnalytics/Sources/StrandAnalytics/AnalyzeRecentDayCache.swift b/Packages/StrandAnalytics/Sources/StrandAnalytics/AnalyzeRecentDayCache.swift index 8654ee8bd1..58f9c45b28 100644 --- a/Packages/StrandAnalytics/Sources/StrandAnalytics/AnalyzeRecentDayCache.swift +++ b/Packages/StrandAnalytics/Sources/StrandAnalytics/AnalyzeRecentDayCache.swift @@ -31,11 +31,21 @@ public enum AnalyzeRecentDayCache { /// both a 4.0 and a 5/MG): when a day's resolved owner flips between straps, keying on the owner makes /// the reuse invalidate **explicitly**, rather than relying on two different devices never producing an /// identical `count`+`maxTs` for the same window. + /// - `streams`: the opaque per-day witness of every OTHER scored stream — PPG-derived HR, R-R, + /// respiration, SpO2, gravity, steps, skin temp and events (`WhoopStore.dayStreamFingerprint`). + /// #29: a history offload does not commit its channels together, and an offloaded HR row duplicating + /// a live one is dropped by `ON CONFLICT DO NOTHING`, so a night can be scored from HR alone and then + /// gain its R-R with `hrCount`/`hrMaxTs` completely unmoved. Keyed on HR alone this said "reuse", and + /// the HRV-less scan was re-served for the rest of the session — a force refresh included, since + /// `force` only bypasses the whole-pass watermark gate. The whole-pass gate had already been widened + /// to every stream (`analysisFingerprint`, v2); this is the same widening at day granularity, which is + /// where the reuse decision is actually made. /// /// Inputs that feed `analyzeDay` but are pass-global rather than per-day (profile, baselines1, sleep /// need / consistency, habitual midsleep, tz, stager toggles) are NOT in this key — the engine drops the /// whole cache when its pass config signature changes, which covers them. public static func cacheKey(owner: String, hrCount: Int, hrMaxTs: Int, skinAnchorRaw: Double?, + streams: String, // #1575: whether this day is the one that emits the PER-WINDOW HRV detail // (`dayStart == nowLocalMidnight`). Now that an active trace no longer // disables reuse, this has to invalidate: the night cached as "today" with @@ -47,6 +57,6 @@ public enum AnalyzeRecentDayCache { // a trace mode is on. hrvWindowDetail: Bool) -> String { let anchor = skinAnchorRaw.map { String($0.bitPattern) } ?? "nil" - return "\(owner)|\(hrCount):\(hrMaxTs):\(anchor):\(hrvWindowDetail ? "d" : "s")" + return "\(owner)|\(hrCount):\(hrMaxTs):\(anchor):\(hrvWindowDetail ? "d" : "s")|\(streams)" } } diff --git a/Packages/StrandAnalytics/Tests/StrandAnalyticsTests/AnalyzeRecentDayCacheTests.swift b/Packages/StrandAnalytics/Tests/StrandAnalyticsTests/AnalyzeRecentDayCacheTests.swift index b95f531504..f9b8910b21 100644 --- a/Packages/StrandAnalytics/Tests/StrandAnalyticsTests/AnalyzeRecentDayCacheTests.swift +++ b/Packages/StrandAnalytics/Tests/StrandAnalyticsTests/AnalyzeRecentDayCacheTests.swift @@ -6,18 +6,22 @@ import XCTest /// (the exact key STRING may differ across platforms — the cache is in-memory and per-platform — but the /// set of changes that must / must not invalidate a reused day is a shared contract). final class AnalyzeRecentDayCacheTests: XCTestCase { + /// A settled night's other-stream witness (`WhoopStore.dayStreamFingerprint`) — opaque to the key, so + /// these tests only need two values that differ. + private let baseStreams = "s1|p0:0|r120000:1700000000|x0:0|o0:0|g86400:1700000000|z0:0|t0:0|e0:0" + // Unchanged inputs → identical key → the day is reused. func testStableInputsReuse() { - let a = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, hrvWindowDetail: false) - let b = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, hrvWindowDetail: false) + let a = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) + let b = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) XCTAssertEqual(a, b) } // A new HR row (count moves) OR a later newest-ts must invalidate — the day changed. func testHrChangeInvalidates() { - let base = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, hrvWindowDetail: false) - let moreRows = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_001, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, hrvWindowDetail: false) - let laterTs = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_060, skinAnchorRaw: 1290, hrvWindowDetail: false) + let base = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) + let moreRows = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_001, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) + let laterTs = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_060, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) XCTAssertNotEqual(base, moreRows) XCTAssertNotEqual(base, laterTs) } @@ -25,17 +29,17 @@ final class AnalyzeRecentDayCacheTests: XCTestCase { // A shifted window-wide skin anchor (another night's skin changed the 4.0 median) must invalidate even // when this night's HR fingerprint is identical — the skin conversion changed. func testAnchorShiftInvalidates() { - let a = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, hrvWindowDetail: false) - let b = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290.5, hrvWindowDetail: false) + let a = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) + let b = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290.5, streams: baseStreams, hrvWindowDetail: false) XCTAssertNotEqual(a, b) } // A 5/MG night (no anchor) is a distinct, self-consistent key: nil reuses nil, and nil ≠ a real anchor // (so a 4.0 and a 5/MG night with the same fingerprint never alias to each other's scan). func testNilAnchorDistinctButStable() { - let nilA = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 5_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: nil, hrvWindowDetail: false) - let nilB = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 5_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: nil, hrvWindowDetail: false) - let real = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 5_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 0, hrvWindowDetail: false) + let nilA = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 5_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: nil, streams: baseStreams, hrvWindowDetail: false) + let nilB = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 5_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: nil, streams: baseStreams, hrvWindowDetail: false) + let real = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 5_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 0, streams: baseStreams, hrvWindowDetail: false) XCTAssertEqual(nilA, nilB) XCTAssertNotEqual(nilA, real) } @@ -45,8 +49,8 @@ final class AnalyzeRecentDayCacheTests: XCTestCase { // count+maxTs for the same window. The owner id is part of the key, so it never falsely reuses one // strap's scan for the other. func testDifferentOwnerInvalidates() { - let whoop4 = AnalyzeRecentDayCache.cacheKey(owner: "whoop4-A", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, hrvWindowDetail: false) - let whoop5 = AnalyzeRecentDayCache.cacheKey(owner: "whoop5-B", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, hrvWindowDetail: false) + let whoop4 = AnalyzeRecentDayCache.cacheKey(owner: "whoop4-A", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) + let whoop5 = AnalyzeRecentDayCache.cacheKey(owner: "whoop5-B", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, streams: baseStreams, hrvWindowDetail: false) XCTAssertNotEqual(whoop4, whoop5) } @@ -62,10 +66,42 @@ final class AnalyzeRecentDayCacheTests: XCTestCase { func testTheHrvDetailDayIsNotReusedAsAnOrdinaryNight() { let detail = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, - hrvWindowDetail: true) + streams: baseStreams, hrvWindowDetail: true) let summary = AnalyzeRecentDayCache.cacheKey(owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, - hrvWindowDetail: false) + streams: baseStreams, hrvWindowDetail: false) XCTAssertNotEqual(detail, summary, "the detail day must invalidate when it stops being today") } + + /// #29: a stream that is not HR moving must invalidate, even with the HR fingerprint frozen. + /// + /// A history offload delivers channels independently and drops offloaded HR rows that duplicate live + /// ones (`ON CONFLICT DO NOTHING`), so a night scored once from HR alone can gain its R-R — or its + /// respiration, SpO2, gravity, steps, skin temp, events, or a PPG-derived HR second — with this night's + /// `hrCount`/`hrMaxTs` completely unmoved. Keyed on HR alone that said "reuse" and re-served the + /// HRV-less scan for the rest of the session, force refreshes included. + /// + /// Byte-parity twin of Kotlin `aStreamArrivingAfterHrInvalidatesTheReusedNight`. + func testAStreamArrivingAfterHrInvalidatesTheReusedNight() { + // Same night, same HR rows, same anchor — only the other-stream witness moved. + let hrOnly = AnalyzeRecentDayCache.cacheKey( + owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, + streams: "s1|p0:0|r0:0|x0:0|o0:0|g0:0|z0:0|t0:0|e0:0", hrvWindowDetail: false) + let rrLanded = AnalyzeRecentDayCache.cacheKey( + owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, + streams: "s1|p0:0|r120000:1700000000|x0:0|o0:0|g0:0|z0:0|t0:0|e0:0", hrvWindowDetail: false) + XCTAssertNotEqual(hrOnly, rrLanded, "the night's R-R arrived — its cached scan has no HRV in it") + } + + /// The other half of the same contract: a witness that did NOT move must still reuse. #29's fix must + /// not turn every pass into a full re-score — that is the drain #1005 closed. + func testUnchangedStreamWitnessStillReuses() { + let a = AnalyzeRecentDayCache.cacheKey( + owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, + streams: baseStreams, hrvWindowDetail: false) + let b = AnalyzeRecentDayCache.cacheKey( + owner: "dev1", hrCount: 178_000, hrMaxTs: 1_700_000_000, skinAnchorRaw: 1290, + streams: baseStreams, hrvWindowDetail: false) + XCTAssertEqual(a, b) + } } diff --git a/Packages/WhoopStore/Sources/WhoopStore/Reads.swift b/Packages/WhoopStore/Sources/WhoopStore/Reads.swift index 73a3971387..25bcc59300 100644 --- a/Packages/WhoopStore/Sources/WhoopStore/Reads.swift +++ b/Packages/WhoopStore/Sources/WhoopStore/Reads.swift @@ -99,6 +99,73 @@ extension WhoopStore { } } + /// Per-day, per-owner change detector for every scored stream that + /// [hrFingerprint(deviceId:from:to:)] does NOT witness. + /// + /// The whole-pass re-score gate answers "did anything change anywhere"; this answers "did THIS + /// night's scored input change", which is the question the `analyzeRecent` per-day reuse cache + /// asks. A history offload + /// does not commit its channels together — HR can land first and R-R, respiration or SpO2 minutes later, + /// and an offloaded HR row duplicating a live one is dropped by `ON CONFLICT DO NOTHING` — so a night + /// scored once from HR alone can gain its R-R with `(count, maxTs)` over `hrSample` completely unmoved + /// (#29). Keyed on HR alone the cache re-served that HRV-less scan for the rest of the session. + /// + /// One statement, one index range walk per stream over the same `(deviceId, ts)` primary keys + /// [hrFingerprint(deviceId:from:to:)] uses, materialising no rows. `COUNT(*)` is the + /// load-bearing half: it moves when a + /// backfill lands rows INSIDE a window already covered, which `MAX(ts)` alone would miss. + /// + /// The streams are exactly the ones the per-day loop reads and hands to `analyzeDay`: + /// - `ppgHrSample`: the day's HR read is measured ∪ PPG-derived ([hrSamples]), so a PPG row for a second + /// with no measured HR changes the scored series while `hrSample` stays put. + /// - `rrInterval`: filtered exactly as [rrIntervals] filters at read (the 0x6E SpO2-IBI duplicate and + /// future-stamped beats are excluded), so the witness counts the beats that are actually scored. + /// - `sleepStateSample`: appended from the SAME v18 record at the same `ts` as HR, so a re-offloaded + /// record whose HR row is dropped on conflict still lands a new band row that `analyzeDay` scores. + /// Its letter is `b` (band), not `s` — `s1|` is the version prefix. + /// - `respSample`, `spo2Sample`, `gravitySample`, `stepSample`, `skinTempSample`, `event`. + /// + /// Returned as an opaque string: it is only ever compared to itself in memory, so no cross-platform or + /// cross-launch byte identity is required. The Kotlin twin is `WhoopDao.dayStreamFingerprint`. + public func dayStreamFingerprint(deviceId: String, from: Int, to: Int) async throws -> String { + try syncRead { db in + // Every sub-select is COUNT/COALESCE(MAX(...), 0), so each column is non-null and the aggregate + // query always returns exactly one row — same idiom as `hrFingerprint` above. + guard let row = try Row.fetchOne(db, sql: """ + SELECT + (SELECT COUNT(*) FROM ppgHrSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS pc, + (SELECT COALESCE(MAX(ts), 0) FROM ppgHrSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS pm, + (SELECT COUNT(*) FROM rrInterval WHERE deviceId = :d AND ts >= :f AND ts <= :t + AND (srcChannel IS NULL OR srcChannel <> :rrx) + AND (tsSuspect IS NULL OR tsSuspect <> 1)) AS rc, + (SELECT COALESCE(MAX(ts), 0) FROM rrInterval WHERE deviceId = :d AND ts >= :f AND ts <= :t + AND (srcChannel IS NULL OR srcChannel <> :rrx) + AND (tsSuspect IS NULL OR tsSuspect <> 1)) AS rm, + (SELECT COUNT(*) FROM respSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS xc, + (SELECT COALESCE(MAX(ts), 0) FROM respSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS xm, + (SELECT COUNT(*) FROM spo2Sample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS oc, + (SELECT COALESCE(MAX(ts), 0) FROM spo2Sample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS om, + (SELECT COUNT(*) FROM gravitySample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS gc, + (SELECT COALESCE(MAX(ts), 0) FROM gravitySample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS gm, + (SELECT COUNT(*) FROM stepSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS zc, + (SELECT COALESCE(MAX(ts), 0) FROM stepSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS zm, + (SELECT COUNT(*) FROM skinTempSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS tc, + (SELECT COALESCE(MAX(ts), 0) FROM skinTempSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS tm, + (SELECT COUNT(*) FROM sleepStateSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS bc, + (SELECT COALESCE(MAX(ts), 0) FROM sleepStateSample WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS bm, + (SELECT COUNT(*) FROM event WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS ec, + (SELECT COALESCE(MAX(ts), 0) FROM event WHERE deviceId = :d AND ts >= :f AND ts <= :t) AS em + """, arguments: ["d": deviceId, "f": from, "t": to, + "rrx": RRSourceChannel.spo2Ibi.rawValue]) else { return "" } + let keys = ["p", "r", "x", "o", "g", "z", "t", "b", "e"] + let parts = keys.map { key -> String in + let count: Int = row[key + "c"], maxTs: Int = row[key + "m"] + return "\(key)\(count):\(maxTs)" + } + return "s1|" + parts.joined(separator: "|") + } + } + /// Aggregate HR over a window: `(n, avg, max)` computed in SQLite over the same measured-∪-PPG rows /// [hrSamples] returns, WITHOUT materialising them and WITHOUT a row limit. /// diff --git a/Packages/WhoopStore/Tests/WhoopStoreTests/ReadTests.swift b/Packages/WhoopStore/Tests/WhoopStoreTests/ReadTests.swift index 5737a694cd..75ede16cbd 100644 --- a/Packages/WhoopStore/Tests/WhoopStoreTests/ReadTests.swift +++ b/Packages/WhoopStore/Tests/WhoopStoreTests/ReadTests.swift @@ -68,6 +68,62 @@ final class ReadTests: XCTestCase { XCTAssertEqual(scoped.count, 3) } + /// #29: the per-DAY witness. The whole-pass re-score gate answers "did anything change + /// anywhere"; the `analyzeRecent` reuse cache asks "did THIS night's scored input change", and keyed on + /// HR alone it could not see a channel that landed after HR — so a night scored from HR alone kept + /// being re-served with no HRV in it. + /// + /// Each stream is committed on its own here, because the failure mode IS independent commits: every one + /// of them must move the witness while `hrFingerprint` stays frozen. + func testDayStreamFingerprintMovesForEveryStreamThatLandsAfterHr() async throws { + let store = try await seeded() + let hrBefore = try await store.hrFingerprint(deviceId: "dev1", from: 0, to: 1000) + var previous = try await store.dayStreamFingerprint(deviceId: "dev1", from: 0, to: 1000) + + // Every arm of the witness gets an entry: an arm that silently reads a NEIGHBOURING table is + // valid SQL and invisible to a test that never commits its stream. + let commits: [(String, Streams)] = [ + ("ppgHr", Streams(ppgHr: [PpgHrSample(ts: 401, bpm: 62, conf: 0.9)])), + ("rr", Streams(rr: [RRInterval(ts: 400, rrMs: 810)])), + ("resp", Streams(resp: [RespSample(ts: 400, raw: 12)])), + ("spo2", Streams(spo2: [SpO2Sample(ts: 400, red: 100, ir: 200)])), + ("gravity", Streams(gravity: [GravitySample(ts: 400, x: 0, y: 0, z: 1)])), + ("steps", Streams(steps: [StepSample(ts: 400, counter: 7)])), + ("skin", Streams(skinTemp: [SkinTempSample(ts: 400, raw: 1290)])), + ("sleepState", Streams(sleepState: [SleepStateSample(ts: 400, state: 2)])), + ("event", Streams(events: [WhoopEvent(ts: 410, kind: "WRIST_OFF", payload: [:])])), + ] + for (label, commit) in commits { + _ = try await store.insert(commit, deviceId: "dev1") + let now = try await store.dayStreamFingerprint(deviceId: "dev1", from: 0, to: 1000) + XCTAssertNotEqual(previous, now, "a \(label) row landing after HR must move the day witness") + previous = now + } + + // The HR witness never moved across ANY of those commits — the reason HR alone was not enough. + let hrAfter = try await store.hrFingerprint(deviceId: "dev1", from: 0, to: 1000) + XCTAssertEqual(hrBefore.count, hrAfter.count) + XCTAssertEqual(hrBefore.maxTs, hrAfter.maxTs) + + // Reading twice with nothing committed in between must be identical, or the reuse cache would + // re-score every night on every pass and give back the drain it exists to close. + let idle = try await store.dayStreamFingerprint(deviceId: "dev1", from: 0, to: 1000) + XCTAssertEqual(previous, idle) + } + + /// The witness is scoped exactly like the reads it stands in for: another device's rows and rows + /// outside the night window must not move it, or every night would invalidate on any strap's traffic. + func testDayStreamFingerprintIsDeviceAndWindowScoped() async throws { + let store = try await seeded() + let base = try await store.dayStreamFingerprint(deviceId: "dev1", from: 0, to: 1000) + _ = try await store.insert(Streams(rr: [RRInterval(ts: 400, rrMs: 810)]), deviceId: "other") + let afterOtherDevice = try await store.dayStreamFingerprint(deviceId: "dev1", from: 0, to: 1000) + XCTAssertEqual(base, afterOtherDevice, "another device's R-R is not this owner's night") + _ = try await store.insert(Streams(rr: [RRInterval(ts: 5000, rrMs: 810)]), deviceId: "dev1") + let afterOutsideWindow = try await store.dayStreamFingerprint(deviceId: "dev1", from: 0, to: 1000) + XCTAssertEqual(base, afterOutsideWindow, "an R-R beat outside the window is not this night's input") + } + func testHrBucketsAveragePerBucketOrderedAndDeviceScoped() async throws { let store = try await seeded() // 200s buckets over dev1's ts 100/200/300 (bpm 60/61/62): diff --git a/Strand/Data/IntelligenceEngine.swift b/Strand/Data/IntelligenceEngine.swift index ecb8e01162..ed65cad573 100644 --- a/Strand/Data/IntelligenceEngine.swift +++ b/Strand/Data/IntelligenceEngine.swift @@ -919,10 +919,19 @@ final class IntelligenceEngine: ObservableObject { } skinAnchorResolvedOwners.insert(owner) } - if let fp = try? await store.hrFingerprint(deviceId: owner, from: from, to: to) { + if let fp = try? await store.hrFingerprint(deviceId: owner, from: from, to: to), + let streamFp = try? await store.dayStreamFingerprint(deviceId: owner, from: from, to: to) { let key = AnalyzeRecentDayCache.cacheKey( owner: owner, hrCount: fp.count, hrMaxTs: fp.maxTs, skinAnchorRaw: skinAnchorByOwner[owner], + // #29: the OTHER scored streams for this night. Without it a night whose R-R + // (or resp/SpO2) landed after its HR keys identically to the HR-only scan it + // was scored from, and the HRV-less result is re-served for the rest of the + // session — force refresh included, since `force` only bypasses the whole-pass + // watermark gate above, never this one. Both reads are index-only aggregates + // over the same `(deviceId, ts)` keys; a miss costs the 7 full stream reads + // this gate exists to skip. + streams: streamFp, // #1575: `hrvTraceActive &&` matters. With the HRV trace OFF no detail // line is ever produced, so the flag describes nothing — but it would still // flip at midnight and invalidate yesterday, charging EVERY user an extra diff --git a/android/app/src/main/java/com/noop/analytics/AnalyzeRecentDayCache.kt b/android/app/src/main/java/com/noop/analytics/AnalyzeRecentDayCache.kt index 6131ec6003..287fee57ba 100644 --- a/android/app/src/main/java/com/noop/analytics/AnalyzeRecentDayCache.kt +++ b/android/app/src/main/java/com/noop/analytics/AnalyzeRecentDayCache.kt @@ -35,6 +35,15 @@ object AnalyzeRecentDayCache { * a 4.0 and a 5/MG): when a day's resolved owner flips between straps, keying on the owner makes the * reuse invalidate **explicitly**, rather than relying on two different devices never producing an * identical `count`+`maxTs` for the same window. + * - [streams]: the opaque per-day witness of every OTHER scored stream — PPG-derived HR, R-R, + * respiration, SpO2, gravity, steps, skin temp and events (`WhoopRepository.dayStreamFingerprint`). + * #29: a history offload does not commit its channels together, and an offloaded HR row duplicating a + * live one is ignored on conflict, so a night can be scored from HR alone and then gain its R-R with + * [hrCount]/[hrMaxTs] completely unmoved. Keyed on HR alone this said "reuse", and the HRV-less scan + * was re-served for the rest of the process, a user-initiated refresh included, since a forced pass + * only bypasses the whole-pass watermark gate. That gate had already been widened to every stream + * (`analysisFingerprint`, v2); this is the same widening at day granularity, which is where the reuse + * decision is actually made. * * Inputs that feed `analyzeDay` but are pass-global rather than per-day (profile, baselines1, sleep need * / consistency, habitual midsleep, tz, stager toggles) are NOT in this key — the engine drops the whole @@ -42,6 +51,7 @@ object AnalyzeRecentDayCache { */ fun cacheKey( owner: String, hrCount: Int, hrMaxTs: Long, skinAnchorRaw: Double?, + streams: String, // #1575: whether this day is the one that emits the PER-WINDOW HRV detail (`dayStart == // nowLocalMidnight`). Now that trace lines are recorded and replayed, this has to invalidate: // the night cached as "today" with its detailed trace becomes an ordinary night after midnight, @@ -52,6 +62,6 @@ object AnalyzeRecentDayCache { hrvWindowDetail: Boolean, ): String { val anchor = skinAnchorRaw?.toRawBits()?.toString() ?: "nil" - return "$owner|$hrCount:$hrMaxTs:$anchor:${if (hrvWindowDetail) "d" else "s"}" + return "$owner|$hrCount:$hrMaxTs:$anchor:${if (hrvWindowDetail) "d" else "s"}|$streams" } } diff --git a/android/app/src/main/java/com/noop/analytics/IntelligenceEngine.kt b/android/app/src/main/java/com/noop/analytics/IntelligenceEngine.kt index 8c6e3e4682..3d95070881 100644 --- a/android/app/src/main/java/com/noop/analytics/IntelligenceEngine.kt +++ b/android/app/src/main/java/com/noop/analytics/IntelligenceEngine.kt @@ -485,6 +485,32 @@ object IntelligenceEngine { flagSet() } + // The per-day reuse key's own two store reads, lifted out of [analyzeRecentOnCpu]'s body so that + // single coroutine state machine stays inside the JVM method-size budget the Jacoco-instrumented + // build asserts. Behaviour is unchanged: the HR fingerprint is read first, the other scored streams + // second, over the same window, and the caller awaits this exactly where the inline assembly stood. + private suspend fun dayReuseCacheKey( + repo: WhoopRepository, + owner: String, + from: Long, + to: Long, + skinAnchorRaw: Double?, + hrvWindowDetail: Boolean, + ): String { + val (fpCount, fpMaxTs) = repo.hrFingerprintWindow(owner, from, to) + return AnalyzeRecentDayCache.cacheKey( + owner, fpCount, fpMaxTs, skinAnchorRaw, + // #29: the OTHER scored streams for this night. Without it a night whose R-R (or + // resp/SpO2) landed after its HR keys identically to the HR-only scan it was scored + // from, and the HRV-less result is re-served for the rest of the process — a + // user-initiated refresh included, since that only bypasses the whole-pass watermark + // gate, never this one. Both reads are index-only aggregates over the same + // (deviceId, ts) keys; a miss costs the full stream reads this gate exists to skip. + streams = repo.dayStreamFingerprint(owner, from, to), + hrvWindowDetail = hrvWindowDetail, + ) + } + private suspend fun analyzeRecentOnCpu( repo: WhoopRepository, profile: UserProfile = UserProfile(), @@ -806,9 +832,8 @@ object IntelligenceEngine { Whoop4SkinTemp.deviceAnchorRaw(windowSkin.map { it.raw })?.let { skinAnchorByOwner[owner] = it } skinAnchorResolvedOwners.add(owner) } - val (fpCount, fpMaxTs) = repo.hrFingerprintWindow(owner, from, to) - val key = AnalyzeRecentDayCache.cacheKey( - owner, fpCount, fpMaxTs, skinAnchorByOwner[owner], + val key = dayReuseCacheKey( + repo, owner, from, to, skinAnchorByOwner[owner], // #1575: `&& hrvTraceSink != null` matters. With the HRV trace OFF no detail line // is ever produced, so the flag describes nothing — but it would still flip at // midnight and invalidate yesterday, charging EVERY user an extra day's re-score to diff --git a/android/app/src/main/java/com/noop/data/WhoopDao.kt b/android/app/src/main/java/com/noop/data/WhoopDao.kt index 818d9f7cfd..5882545bda 100644 --- a/android/app/src/main/java/com/noop/data/WhoopDao.kt +++ b/android/app/src/main/java/com/noop/data/WhoopDao.kt @@ -8,6 +8,55 @@ import androidx.room.Transaction import androidx.room.Upsert import kotlinx.coroutines.flow.Flow +/** Per-day, per-owner witness of every scored stream the HR fingerprint does NOT cover (#29). + * + * The whole-pass re-score gate answers "did anything change anywhere"; the analyzeRecent per-day reuse cache + * asks "did THIS night's scored input change". A history offload does not commit its channels together — + * HR can land first and R-R, respiration or SpO2 minutes later, and an offloaded HR row duplicating a live + * one is ignored on conflict — so a night scored once from HR alone can gain its R-R with COUNT/MAX over + * hrSample completely unmoved. Keyed on HR alone the cache re-served that HRV-less scan for the rest of the + * process, a user-initiated refresh included. + * + * COUNT(*) is the load-bearing half: it moves when a backfill lands rows INSIDE a window already covered, + * which MAX(ts) alone would miss. Every arm is an index range walk over the same (deviceId, ts) key the HR + * fingerprint uses, materializing no rows. + * + * rrInterval is filtered exactly as [WhoopDao.rrIntervals] filters at read (the 0x6E SpO2-IBI duplicate and + * future-stamped beats), so the witness counts the beats that are actually scored. The literal 2 is + * RrSourceChannel.SPO2_IBI.code, pinned to the enum by RrChannelTest, for the same reason it is a literal + * there: a Room @Query is a compile-time constant string. + * + * sleepStateSample is in it because the band state is appended from the SAME v18 record at the same ts as + * HR: a re-offloaded record whose HR row is ignored on conflict still lands a new band row that the day is + * scored from. Its letter is 'b' (band), not 's', which the version prefix 's1|' already spends. + * + * The result is opaque and only ever compared to itself in memory, so it needs no byte identity with the + * Swift twin, `WhoopStore.dayStreamFingerprint`. Kept as one named constant so every caller executes the + * exact same statement; it carries :deviceId/:from/:to binds, so a plain-JVM SQLite harness could not run + * it verbatim. Room's KSP verification is what checks it. */ +internal const val DAY_STREAM_FINGERPRINT_SQL = + "SELECT 's1|' || " + + "'p' || (SELECT COUNT(*) FROM ppgHrSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM ppgHrSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || '|' || " + + "'r' || (SELECT COUNT(*) FROM rrInterval WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to " + + "AND (srcChannel IS NULL OR srcChannel <> 2) AND (tsSuspect IS NULL OR tsSuspect <> 1)) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM rrInterval WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to " + + "AND (srcChannel IS NULL OR srcChannel <> 2) AND (tsSuspect IS NULL OR tsSuspect <> 1)) || '|' || " + + "'x' || (SELECT COUNT(*) FROM respSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM respSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || '|' || " + + "'o' || (SELECT COUNT(*) FROM spo2Sample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM spo2Sample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || '|' || " + + "'g' || (SELECT COUNT(*) FROM gravitySample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM gravitySample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || '|' || " + + "'z' || (SELECT COUNT(*) FROM stepSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM stepSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || '|' || " + + "'t' || (SELECT COUNT(*) FROM skinTempSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM skinTempSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || '|' || " + + "'b' || (SELECT COUNT(*) FROM sleepStateSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM sleepStateSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || '|' || " + + "'e' || (SELECT COUNT(*) FROM event WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to) || " + + "':' || (SELECT COALESCE(MAX(ts), 0) FROM event WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to)" + /** * Data-access for the local store. Mirrors the GRDB reads/writes in WhoopStore * (StreamStore.swift, Reads.swift, MetricsCache.swift). @@ -948,6 +997,11 @@ interface WhoopDao : DeviceRegistryDao { suspend fun countHrInWindow(deviceId: String, from: Long, to: Long): Int @Query("SELECT COALESCE(MAX(ts), 0) FROM hrSample WHERE deviceId = :deviceId AND ts >= :from AND ts <= :to") suspend fun maxHrTsInWindow(deviceId: String, from: Long, to: Long): Long + // #29: the same per-day (device + window) witness for every OTHER scored stream — see + // DAY_STREAM_FINGERPRINT_SQL. Without it a night whose R-R landed after its HR keyed identically to the + // HR-only scan it was scored from, and that HRV-less scan was re-served for the rest of the process. + @Query(DAY_STREAM_FINGERPRINT_SQL) + suspend fun dayStreamFingerprint(deviceId: String, from: Long, to: Long): String @Query("SELECT COUNT(*) FROM rrInterval") suspend fun countRr(): Int @Query("SELECT COUNT(*) FROM event") suspend fun countEvents(): Int @Query("SELECT COUNT(*) FROM battery") suspend fun countBattery(): Int diff --git a/android/app/src/main/java/com/noop/data/WhoopRepository.kt b/android/app/src/main/java/com/noop/data/WhoopRepository.kt index b0fe987961..05069801d9 100644 --- a/android/app/src/main/java/com/noop/data/WhoopRepository.kt +++ b/android/app/src/main/java/com/noop/data/WhoopRepository.kt @@ -609,6 +609,14 @@ class WhoopRepository( suspend fun hrFingerprintWindow(deviceId: String, from: Long, to: Long): Pair = Pair(dao.countHrInWindow(deviceId, from, to), dao.maxHrTsInWindow(deviceId, from, to)) + /** #29 — the same per-day (device + window) witness for every OTHER stream analyzeDay scores: PPG-derived + * HR, R-R, respiration, SpO2, gravity, steps, skin temp and events. [hrFingerprintWindow] cannot see a + * channel that lands after HR, and an offload commits its channels independently, so keyed on HR alone + * the reuse cache re-served an HRV-less scan for the rest of the process. See + * DAY_STREAM_FINGERPRINT_SQL; mirrors Swift WhoopStore.dayStreamFingerprint. */ + suspend fun dayStreamFingerprint(deviceId: String, from: Long, to: Long): String = + dao.dayStreamFingerprint(deviceId, from, to) + // MARK: - Server-derived caches (latest value wins on conflict) suspend fun upsertDailyMetrics(days: List) = dao.upsertDailyMetrics(days) diff --git a/android/app/src/test/java/com/noop/analytics/AnalyzeRecentDayCacheTest.kt b/android/app/src/test/java/com/noop/analytics/AnalyzeRecentDayCacheTest.kt index 503dac36b0..8c2d5f17df 100644 --- a/android/app/src/test/java/com/noop/analytics/AnalyzeRecentDayCacheTest.kt +++ b/android/app/src/test/java/com/noop/analytics/AnalyzeRecentDayCacheTest.kt @@ -11,6 +11,10 @@ import org.junit.Test * must / must not invalidate a reused day is a shared contract). */ class AnalyzeRecentDayCacheTest { + /** A settled night's other-stream witness (`WhoopRepository.dayStreamFingerprint`) — opaque to the key, + * so these tests only need two values that differ. */ + private val baseStreams = "s1|p0:0|r120000:1700000000|x0:0|o0:0|g86400:1700000000|z0:0|t0:0|e0:0" + /** * #1575: the day that emits the per-window HRV DETAIL must not be reused as an ordinary night. @@ -24,23 +28,23 @@ class AnalyzeRecentDayCacheTest { @Test fun theHrvDetailDayIsNotReusedAsAnOrdinaryNight() { val detail = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, - hrvWindowDetail = true) + baseStreams, hrvWindowDetail = true) val summary = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, - hrvWindowDetail = false) + baseStreams, hrvWindowDetail = false) assertNotEquals("the detail day must invalidate when it stops being today", detail, summary) } // Unchanged inputs -> identical key -> the day is reused. @Test fun stableInputsReuse() { - val a = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, hrvWindowDetail = false) - val b = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, hrvWindowDetail = false) + val a = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, baseStreams, hrvWindowDetail = false) + val b = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, baseStreams, hrvWindowDetail = false) assertEquals(a, b) } // A new HR row (count moves) OR a later newest-ts must invalidate — the day changed. @Test fun hrChangeInvalidates() { - val base = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, hrvWindowDetail = false) - val moreRows = AnalyzeRecentDayCache.cacheKey("dev1", 178_001, 1_700_000_000L, 1290.0, hrvWindowDetail = false) - val laterTs = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_060L, 1290.0, hrvWindowDetail = false) + val base = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, baseStreams, hrvWindowDetail = false) + val moreRows = AnalyzeRecentDayCache.cacheKey("dev1", 178_001, 1_700_000_000L, 1290.0, baseStreams, hrvWindowDetail = false) + val laterTs = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_060L, 1290.0, baseStreams, hrvWindowDetail = false) assertNotEquals(base, moreRows) assertNotEquals(base, laterTs) } @@ -48,17 +52,17 @@ class AnalyzeRecentDayCacheTest { // A shifted window-wide skin anchor (another night's skin changed the 4.0 median) must invalidate even // when this night's HR fingerprint is identical — the skin conversion changed. @Test fun anchorShiftInvalidates() { - val a = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, hrvWindowDetail = false) - val b = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.5, hrvWindowDetail = false) + val a = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, baseStreams, hrvWindowDetail = false) + val b = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.5, baseStreams, hrvWindowDetail = false) assertNotEquals(a, b) } // A night with no anchor (null) is a distinct, self-consistent key: null reuses null, and null != a real // anchor (so it never aliases to a real-anchor night's scan). @Test fun nullAnchorDistinctButStable() { - val nilA = AnalyzeRecentDayCache.cacheKey("dev1", 5_000, 1_700_000_000L, null, hrvWindowDetail = false) - val nilB = AnalyzeRecentDayCache.cacheKey("dev1", 5_000, 1_700_000_000L, null, hrvWindowDetail = false) - val real = AnalyzeRecentDayCache.cacheKey("dev1", 5_000, 1_700_000_000L, 0.0, hrvWindowDetail = false) + val nilA = AnalyzeRecentDayCache.cacheKey("dev1", 5_000, 1_700_000_000L, null, baseStreams, hrvWindowDetail = false) + val nilB = AnalyzeRecentDayCache.cacheKey("dev1", 5_000, 1_700_000_000L, null, baseStreams, hrvWindowDetail = false) + val real = AnalyzeRecentDayCache.cacheKey("dev1", 5_000, 1_700_000_000L, 0.0, baseStreams, hrvWindowDetail = false) assertEquals(nilA, nilB) assertNotEquals(nilA, real) } @@ -68,8 +72,42 @@ class AnalyzeRecentDayCacheTest { // count+maxTs for the same window. The owner id is part of the key, so it never falsely reuses one // strap's scan for the other. @Test fun differentOwnerInvalidates() { - val whoop4 = AnalyzeRecentDayCache.cacheKey("whoop4-A", 178_000, 1_700_000_000L, 1290.0, hrvWindowDetail = false) - val whoop5 = AnalyzeRecentDayCache.cacheKey("whoop5-B", 178_000, 1_700_000_000L, 1290.0, hrvWindowDetail = false) + val whoop4 = AnalyzeRecentDayCache.cacheKey("whoop4-A", 178_000, 1_700_000_000L, 1290.0, baseStreams, hrvWindowDetail = false) + val whoop5 = AnalyzeRecentDayCache.cacheKey("whoop5-B", 178_000, 1_700_000_000L, 1290.0, baseStreams, hrvWindowDetail = false) assertNotEquals(whoop4, whoop5) } + + /** + * #29: a stream that is not HR moving must invalidate, even with the HR fingerprint frozen. + * + * A history offload delivers channels independently and ignores offloaded HR rows that duplicate live + * ones, so a night scored once from HR alone can gain its R-R — or its respiration, SpO2, gravity, + * steps, skin temp, events, or a PPG-derived HR second — with this night's `hrCount`/`hrMaxTs` + * completely unmoved. Keyed on HR alone that said "reuse" and re-served the HRV-less scan for the rest + * of the process, user-initiated refreshes included. + * + * Twin of Swift `testAStreamArrivingAfterHrInvalidatesTheReusedNight`. + */ + @Test fun aStreamArrivingAfterHrInvalidatesTheReusedNight() { + // Same night, same HR rows, same anchor — only the other-stream witness moved. + val hrOnly = AnalyzeRecentDayCache.cacheKey( + "dev1", 178_000, 1_700_000_000L, 1290.0, + "s1|p0:0|r0:0|x0:0|o0:0|g0:0|z0:0|t0:0|e0:0", hrvWindowDetail = false) + val rrLanded = AnalyzeRecentDayCache.cacheKey( + "dev1", 178_000, 1_700_000_000L, 1290.0, + "s1|p0:0|r120000:1700000000|x0:0|o0:0|g0:0|z0:0|t0:0|e0:0", hrvWindowDetail = false) + assertNotEquals("the night's R-R arrived — its cached scan has no HRV in it", hrOnly, rrLanded) + } + + /** + * The other half of the same contract: a witness that did NOT move must still reuse. #29's fix must not + * turn every pass into a full re-score — that is the drain #1005 closed. + */ + @Test fun unchangedStreamWitnessStillReuses() { + val a = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, baseStreams, + hrvWindowDetail = false) + val b = AnalyzeRecentDayCache.cacheKey("dev1", 178_000, 1_700_000_000L, 1290.0, baseStreams, + hrvWindowDetail = false) + assertEquals(a, b) + } }