fix: preserve HRV with partial WHOOP 4 history

Packages/WhoopProtocol/Sources/WhoopProtocol/Streams.swift

@@ -37,7 +37,8 @@ public enum StandardHRContact: String, Equatable, Codable, Sendable {
 /// The sensor channel or transport that produced an R-R interval.
 ///
 /// WHOOP 5 exposes one beat train over multiple transports; codes 5–7 distinguish those observations.
-/// WHOOP 4 standard-BLE rows and legacy data keep nil; its type-47 historical rows use code 8. An Oura ring has more than one optical channel:
+/// WHOOP 4 type-40 realtime and standard-BLE rows use codes 9 and 10; legacy data keeps nil.
+/// Its type-47 historical rows use code 8. An Oura ring has more than one optical channel:
 /// the green-quality tag (0x80) and the SpO2 tag (0x6E) both
 /// decode to R-R and both were stored, so the table held roughly TWO complete copies of every night —
 /// not duplicate rows to de-duplicate, but the SAME heartbeats measured twice. Labelling the channel is
@@ -73,6 +74,10 @@ public enum RRSourceChannel: Int, Equatable, Codable, Sendable, CaseIterable {
     /// WHOOP 4.0 type-47 historical R-R intervals. The 4.0 standard BLE feed has no source label,
     /// so this distinguishes the clock-anchored offload from the overlapping live feed at read time.
     case whoop4Historical = 8
+    /// WHOOP 4.0 type-40 live R-R intervals, timestamped from the strap's realtime record clock.
+    case whoop4Realtime = 9
+    /// WHOOP 4.0 standard BLE 0x2A37 R-R intervals, timestamped on notification receipt.
+    case whoop4Standard = 10
 
     public var isWhoop5Transport: Bool { (5...7).contains(rawValue) }
 
@@ -826,7 +831,8 @@ private func toWall(_ deviceTs: Int?, _ deviceClockRef: Int, _ wallClockRef: Int
 /// would double-count HR for the same instants. Frames whose integrity verdict is negative are
 /// skipped — no row is derived from a frame that is not intact.
 public func extractStreams(_ parsed: [ParsedFrame],
-                           deviceClockRef: Int, wallClockRef: Int) -> Streams {
+                           deviceClockRef: Int, wallClockRef: Int,
+                           family: DeviceFamily? = nil) -> Streams {
     var out = Streams()
     for r in parsed {
         // `ok` is now the FULL verdict — header checksum, payload CRC32 and structural length — so it
@@ -844,6 +850,7 @@ public func extractStreams(_ parsed: [ParsedFrame],
             // Unlike Python, drop RR rows when timestamp is absent (a ts-less RR row is unstorable).
             if let ts = ts, let rrs = p["rr_intervals"]?.intArrayValue {
                 let source = p["rr_source_channel"]?.intValue.flatMap(RRSourceChannel.init(rawValue:))
+                    ?? (family == .whoop4 ? .whoop4Realtime : nil)
                 for rr in rrs { out.rr.append(RRInterval(ts: ts, rrMs: rr, srcChannel: source)) }
             }
         case "EVENT":

Packages/WhoopProtocol/Tests/WhoopProtocolTests/StreamsTests.swift

@@ -41,6 +41,18 @@ final class StreamsTests: XCTestCase {
         XCTAssertTrue(s.rr.isEmpty)
     }
 
+    func testRealtimeRRSourceIsTaggedOnlyWhenFamilyIsKnown() {
+        let frame = ParsedFrame(ok: true, typeName: "REALTIME_DATA", seq: nil, cmdName: nil,
+                                crcOK: true, lenBytes: 0, rawHex: "", fields: [],
+                                parsed: ["timestamp": .int(deviceClockRef),
+                                         "rr_intervals": .intArray([800])])
+        let tagged = extractStreams([frame], deviceClockRef: deviceClockRef,
+                                    wallClockRef: wallClockRef, family: .whoop4)
+        XCTAssertEqual(tagged.rr.map(\.srcChannel), [.whoop4Realtime])
+        let legacy = extractStreams([frame], deviceClockRef: deviceClockRef, wallClockRef: wallClockRef)
+        XCTAssertNil(legacy.rr.first?.srcChannel)
+    }
+
     func testEventTimestampIsNotOffset() {
         let s = extractStreams(parsedFrames([ev]),
                                deviceClockRef: deviceClockRef, wallClockRef: wallClockRef)

Packages/WhoopProtocol/Tests/WhoopProtocolTests/Whoop5RRTests.swift

@@ -50,7 +50,7 @@ final class Whoop5RRTests: XCTestCase {
     }
 
     func testTransportCodes() {
-        XCTAssertEqual(RRSourceChannel.allCases.map(\.rawValue), Array(1...8))
+        XCTAssertEqual(RRSourceChannel.allCases.map(\.rawValue), Array(1...10))
     }
 
     /// The table below is the parity contract for the "this night cannot be scored" explanation. The

Packages/WhoopStore/Sources/WhoopStore/Reads.swift

@@ -231,7 +231,7 @@ extension WhoopStore {
             }
             let historyCount: Int = row["w5"]
             let registry: String = row["registry"]
-            return "v4|h\(hc):\(hm)|" + tails.joined(separator: "|")
+            return "v5|h\(hc):\(hm)|" + tails.joined(separator: "|")
                 + "|w5\(historyCount)|w7\(row["w7"] as Int)|tagged\(row["w5tagged"] as Int)"
                 + "|w4history\(row["w4history"] as Int)|registry\(registry)"
         }
@@ -311,7 +311,7 @@ extension WhoopStore {
             let historyCount: Int = row["w5"]
             let registry: String = row["registry"]
             let strictRR = try Self.isWhoop5RRSource(db: db, deviceId: deviceId)
-            return "s3|" + parts.joined(separator: "|") + "|w5\(historyCount)|w7\(row["w7"] as Int)"
+            return "s4|" + parts.joined(separator: "|") + "|w5\(historyCount)|w7\(row["w7"] as Int)"
                 + "|w4h\(row["w4h"] as Int)|ownerTagged\(row["w5owner"] as Int)|registry\(registry)|rr5=\(strictRR)"
         }
     }
@@ -466,8 +466,9 @@ extension WhoopStore {
             // One transport for the complete requested interval. Legacy WHOOP 5 rows mix units and
             // origins, so they remain stored but cannot be converted or spliced into a scored beat train.
             // This subquery uses the SAME time/suspect predicates as the outer read, before LIMIT.
-            // A WHOOP 4 takes the history branch below: its type-47 offload is labelled, its standard-BLE
-            // feed is not, and the two overlap, so one or the other is scored for the whole interval.
+            // A WHOOP 4 has labelled type-47 history, type-40 realtime, standard-BLE, and legacy
+            // unlabelled rows. Choose one source per UTC hour in provenance order, so overlapping live
+            // transports are never merged and partial history takes precedence only where it exists.
             // Every other source takes the Oura branch: one beat channel for the interval, the fuller one
             // (see the doc comment on `rrIntervals`), with NULL and non-Oura codes passing untouched.
             let strictWhoop4History: Bool
@@ -484,29 +485,61 @@ extension WhoopStore {
                 srcChannel = (SELECT MIN(srcChannel) FROM rrInterval
                     WHERE deviceId = :d AND ts >= :f AND ts <= :t AND srcChannel IN \(Self.scorableWhoop5Channels)
                     AND (tsSuspect IS NULL OR tsSuspect <> 1))
-                """ : (strictWhoop4History ? """
-                ((EXISTS(SELECT 1 FROM rrInterval WHERE deviceId = :d AND ts >= :f AND ts <= :t
-                    AND srcChannel = :whoop4Historical AND (tsSuspect IS NULL OR tsSuspect <> 1))
-                    AND srcChannel = :whoop4Historical)
-                 OR (NOT EXISTS(SELECT 1 FROM rrInterval WHERE deviceId = :d AND ts >= :f AND ts <= :t
-                    AND srcChannel = :whoop4Historical AND (tsSuspect IS NULL OR tsSuspect <> 1))
-                    AND srcChannel IS NULL))
                 """ : """
                 (srcChannel IS NULL OR srcChannel NOT IN \(Self.scorableOuraChannels) OR (srcChannel = 1) = (
                     SELECT SUM(srcChannel = 1) > SUM(srcChannel <> 1) FROM rrInterval
                     WHERE deviceId = :d AND ts >= :f AND ts <= :t AND srcChannel IN \(Self.scorableOuraChannels)
                     AND (tsSuspect IS NULL OR tsSuspect <> 1)))
-                """)
-            let rows = try Row.fetchAll(db, sql: """
-                SELECT ts, rrMs, srcChannel, ord, seq FROM rrInterval
-                WHERE deviceId = :d AND ts >= :f AND ts <= :t
-                AND (srcChannel IS NULL OR srcChannel <> :rrx)
-                AND \(sourcePredicate)
-                AND (tsSuspect IS NULL OR tsSuspect <> 1)   -- #1073: exclude future-stamped beats
-                ORDER BY ts ASC, ord ASC, rrMs ASC, seq ASC LIMIT :lim
-                """, arguments: ["d": deviceId, "f": from, "t": to,
-                                 "rrx": RRSourceChannel.spo2Ibi.rawValue,
-                                 "whoop4Historical": RRSourceChannel.whoop4Historical.rawValue, "lim": limit])
+                """
+            // The WHOOP 4 source decision is per UTC hour. Materialize one choice per hour, then join
+            // that small result to the requested beats. Counting the same hour from every beat made
+            // wide HRV reads quadratic in the number of rows per hour and could stall metric screens.
+            let sql: String
+            if strictWhoop4History {
+                sql = """
+                    WITH whoop4HourChoice AS MATERIALIZED (
+                        SELECT ts / 3600 AS hour,
+                               MIN(CASE WHEN srcChannel = :whoop4Historical THEN 1
+                                        WHEN srcChannel = :whoop4Realtime THEN 2
+                                        WHEN srcChannel = :whoop4Standard THEN 3
+                                        WHEN srcChannel IS NULL THEN 4 END) AS sourceChoice
+                        FROM rrInterval
+                        WHERE deviceId = :d
+                          AND ts >= (:f / 3600) * 3600
+                          AND ts < ((:t / 3600) + 1) * 3600
+                          AND (tsSuspect IS NULL OR tsSuspect <> 1)
+                          AND (srcChannel IS NULL OR srcChannel IN
+                               (:whoop4Historical, :whoop4Realtime, :whoop4Standard))
+                        GROUP BY ts / 3600
+                    )
+                    SELECT r.ts, r.rrMs, r.srcChannel, r.ord, r.seq FROM rrInterval r
+                    JOIN whoop4HourChoice h ON h.hour = r.ts / 3600
+                    WHERE r.deviceId = :d AND r.ts >= :f AND r.ts <= :t
+                      AND (r.srcChannel IS NULL OR r.srcChannel <> :rrx)
+                      AND ((r.srcChannel = :whoop4Historical AND h.sourceChoice = 1)
+                           OR (r.srcChannel = :whoop4Realtime AND h.sourceChoice = 2)
+                           OR (r.srcChannel = :whoop4Standard AND h.sourceChoice = 3)
+                           OR (r.srcChannel IS NULL AND h.sourceChoice = 4))
+                      AND (r.tsSuspect IS NULL OR r.tsSuspect <> 1)
+                    ORDER BY r.ts ASC, r.ord ASC, r.rrMs ASC, r.seq ASC LIMIT :lim
+                    """
+            } else {
+                sql = """
+                    SELECT ts, rrMs, srcChannel, ord, seq FROM rrInterval
+                    WHERE deviceId = :d AND ts >= :f AND ts <= :t
+                    AND (srcChannel IS NULL OR srcChannel <> :rrx)
+                    AND \(sourcePredicate)
+                    AND (tsSuspect IS NULL OR tsSuspect <> 1)   -- #1073: exclude future-stamped beats
+                    ORDER BY ts ASC, ord ASC, rrMs ASC, seq ASC LIMIT :lim
+                    """
+            }
+            let rows = try Row.fetchAll(db, sql: sql,
+                                        arguments: ["d": deviceId, "f": from, "t": to,
+                                                    "rrx": RRSourceChannel.spo2Ibi.rawValue,
+                                                    "whoop4Historical": RRSourceChannel.whoop4Historical.rawValue,
+                                                    "whoop4Realtime": RRSourceChannel.whoop4Realtime.rawValue,
+                                                    "whoop4Standard": RRSourceChannel.whoop4Standard.rawValue,
+                                                    "lim": limit])
                 .map { row in
                     RRInterval(ts: row["ts"], rrMs: row["rrMs"],
                                srcChannel: (row["srcChannel"] as Int?).flatMap(RRSourceChannel.init(rawValue:)),

Packages/WhoopStore/Tests/WhoopStoreTests/RrSourceChannelTests.swift

@@ -18,12 +18,16 @@ import OuraProtocol
 final class RrSourceChannelTests: XCTestCase {
     private let ts = 1_750_000_000
 
-    func testWhoop4HistoricalSourceWinsAndPromotesLegacyLiveRows() async throws {
-        let store = try await WhoopStore.inMemory()
-        try await store.upsertDevice(id: "strap", mac: nil, name: nil)
+    private func setRegistry(_ store: WhoopStore) throws {
         try DeviceRegistryStore(dbQueue: store.registryWriter).add(PairedDevice(
             id: "strap", brand: "WHOOP", model: "4.0", sourceKind: .liveBLE,
             capabilities: [.hr, .hrv], status: .active, addedAt: ts, lastSeenAt: ts))
+    }
+
+    func testWhoop4HistoricalSourceWinsAndPromotesLegacyLiveRows() async throws {
+        let store = try await WhoopStore.inMemory()
+        try await store.upsertDevice(id: "strap", mac: nil, name: nil)
+        try setRegistry(store)
 
         _ = try await store.insert(Streams(rr: [RRInterval(ts: ts, rrMs: 800),
                                                   RRInterval(ts: ts + 1, rrMs: 810)]), deviceId: "strap")
@@ -39,22 +43,18 @@ final class RrSourceChannelTests: XCTestCase {
     func testWhoop4FallsBackToUnlabelledRowsWhenNoHistoricalDataExists() async throws {
         let store = try await WhoopStore.inMemory()
         try await store.upsertDevice(id: "strap", mac: nil, name: nil)
-        try DeviceRegistryStore(dbQueue: store.registryWriter).add(PairedDevice(
-            id: "strap", brand: "WHOOP", model: "4.0", sourceKind: .liveBLE,
-            capabilities: [.hr, .hrv], status: .active, addedAt: ts, lastSeenAt: ts))
+        try setRegistry(store)
         _ = try await store.insert(Streams(rr: [RRInterval(ts: ts, rrMs: 800)]), deviceId: "strap")
 
         let selected = try await store.rrIntervals(deviceId: "strap", from: ts, to: ts, limit: 100)
         XCTAssertEqual(selected.map(\.rrMs), [800])
         XCTAssertNil(selected.first?.srcChannel)
     }
 
-    func testWhoop4UsesOneHistoricalTransportForTheWholeWindow() async throws {
+    func testWhoop4HistoricalSourceHasPriorityWithinItsHour() async throws {
         let store = try await WhoopStore.inMemory()
         try await store.upsertDevice(id: "strap", mac: nil, name: nil)
-        try DeviceRegistryStore(dbQueue: store.registryWriter).add(PairedDevice(
-            id: "strap", brand: "WHOOP", model: "4.0", sourceKind: .liveBLE,
-            capabilities: [.hr, .hrv], status: .active, addedAt: ts, lastSeenAt: ts))
+        try setRegistry(store)
         _ = try await store.insert(Streams(rr: [RRInterval(ts: ts, rrMs: 800),
                                                   RRInterval(ts: ts + 1, rrMs: 810)]), deviceId: "strap")
         _ = try await store.insert(Streams(rr: [RRInterval(ts: ts, rrMs: 805, srcChannel: .whoop4Historical)]),
@@ -65,6 +65,46 @@ final class RrSourceChannelTests: XCTestCase {
         XCTAssertEqual(selected.map(\.srcChannel), [.whoop4Historical])
     }
 
+    func testWhoop4RealtimeSourceWinsOverStandardAndLegacyRows() async throws {
+        let store = try await WhoopStore.inMemory()
+        try await store.upsertDevice(id: "strap", mac: nil, name: nil)
+        try setRegistry(store)
+        _ = try await store.insert(Streams(rr: [
+            RRInterval(ts: ts, rrMs: 800),
+            RRInterval(ts: ts, rrMs: 810, srcChannel: .whoop4Standard),
+            RRInterval(ts: ts, rrMs: 820, srcChannel: .whoop4Realtime),
+        ]), deviceId: "strap")
+
+        let selected = try await store.rrIntervals(deviceId: "strap", from: ts, to: ts, limit: 100)
+        XCTAssertEqual(selected.map(\.rrMs), [820])
+        XCTAssertEqual(selected.map(\.srcChannel), [.whoop4Realtime])
+    }
+
+    func testWhoop4PartialHistoryKeepsUnlabelledRowsFromOtherHours() async throws {
+        let store = try await WhoopStore.inMemory()
+        try await store.upsertDevice(id: "strap", mac: nil, name: nil)
+        try setRegistry(store)
+
+        let base = (ts / 3600) * 3600
+        _ = try await store.insert(Streams(rr: [
+            RRInterval(ts: base + 10, rrMs: 800),
+            RRInterval(ts: base + 3600 + 10, rrMs: 820),
+            RRInterval(ts: base + 3600 + 11, rrMs: 830),
+            RRInterval(ts: base + 3600 + 12, rrMs: 840),
+        ]), deviceId: "strap")
+        _ = try await store.insert(Streams(rr: [
+            RRInterval(ts: base + 10, rrMs: 805, srcChannel: .whoop4Historical),
+            RRInterval(ts: base + 11, rrMs: 815, srcChannel: .whoop4Historical),
+            RRInterval(ts: base + 3600 + 10, rrMs: 825, srcChannel: .whoop4Historical),
+        ]), deviceId: "strap")
+
+        let selected = try await store.rrIntervals(deviceId: "strap", from: base,
+                                                    to: base + 7200, limit: 100)
+        XCTAssertEqual(selected.map(\.rrMs), [805, 815, 825])
+        XCTAssertEqual(selected.map(\.srcChannel),
+                       [.whoop4Historical, .whoop4Historical, .whoop4Historical])
+    }
+
     // MARK: - The label survives the mapping
 
     /// A 0x6E record and a 0x80 record covering the same interval must produce rows with DISTINCT

Strand/Collect/Collector.swift

@@ -231,7 +231,8 @@ final class Collector {
 
         let frames = batch.map(\.frame)         // still needed for the raw-capture outbox
         let parsed = batch.map(\.parsed)        // #47: the seam already decoded these — don't re-parse
-        let streams = extractStreams(parsed, deviceClockRef: ref.device, wallClockRef: ref.wall)
+        let streams = extractStreams(parsed, deviceClockRef: ref.device, wallClockRef: ref.wall,
+                                     family: family)
         // #1118: the SECOND live transport. `flushStandardHR` stamps a beat at the second it arrived over
         // 0x2A37; this one stamps it from the strap's own record clock. The same beat reaching both lands
         // on two different seconds, which no same-second de-dup can collapse — the signature every
@@ -291,7 +292,8 @@ final class Collector {
         let acceptedHR = (30...220).contains(hr) ? 1 : 0
         let acceptedRR = rr.filter { (250...3000).contains($0) }
         if acceptedHR == 1 { stdHR.append(HRSample(ts: ts, bpm: hr)) }
-        let source: RRSourceChannel? = family == .whoop5 ? .whoop5Standard : nil
+        let source: RRSourceChannel? = family == .whoop5 ? .whoop5Standard
+            : (family == .whoop4 ? .whoop4Standard : nil)
         stdRR.append(contentsOf: acceptedRR.map { RRInterval(ts: ts, rrMs: $0, srcChannel: source) })
         // Only the CHANGES. Advanced here rather than at flush because the event travels in the buffer
         // until it persists: a failed insert re-inserts it at the front, so nothing has to be unwound.

android/app/src/main/java/com/noop/ble/WhoopBleClient.kt

@@ -10266,7 +10266,8 @@ class WhoopBleClient(
             .filter { it.ok && it.typeName == "REALTIME_DATA" }
             .mapNotNull { (it.parsed["timestamp"] as? Number)?.toInt() }
             .maxOrNull() ?: now
-        val streams: Streams = extractStreams(parsed, deviceClockRef = newestRealtimeTs, wallClockRef = now)
+        val streams: Streams = extractStreams(parsed, deviceClockRef = newestRealtimeTs,
+            wallClockRef = now, family = connectedFamily)
         val batch = StreamPersistence.toBatch(streams)
         // #1118: the SECOND live transport. The standard 0x2A37 path above stamps a beat at the second
         // it arrived; this one stamps it from the strap's own record clock. The same beat reaching both
@@ -10334,8 +10335,11 @@ class WhoopBleClient(
                                  family: DeviceFamily) {
         val shouldFlush = synchronized(collectorLock) {
             if (hr in 30..220) stdHr.add(HrRow(ts, hr))
-            val source = if (family == DeviceFamily.WHOOP5)
-                com.noop.protocol.RrSourceChannel.WHOOP5_STANDARD else null
+            val source = when (family) {
+                DeviceFamily.WHOOP5 -> com.noop.protocol.RrSourceChannel.WHOOP5_STANDARD
+                DeviceFamily.WHOOP4 -> com.noop.protocol.RrSourceChannel.WHOOP4_STANDARD
+                else -> null
+            }
             for (r in rr) if (r in 250..3000) stdRr.add(RrRow(ts, r, source))
             stdContact.add(StandardHrMapping.contactEvent(ts, contact))
             standardHrBufferReachedFlushThreshold(stdHr.size, stdRr.size, stdContact.size)