Test day-cache invalidation for late-arriving streams

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)"
     }
 }

Packages/StrandAnalytics/Tests/StrandAnalyticsTests/AnalyzeRecentDayCacheTests.swift

@@ -6,36 +6,40 @@ 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)
     }
 
     // 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)
+    }
 }

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.
     ///