Fix RwFit initialization, history sync, and manual readings

PulseLoop/Diagnostics/DiagnosticsSubscriber.swift

@@ -47,6 +47,16 @@ final class DiagnosticsSubscriber {
             log(.battery, .info, "Battery \(percent)%")
         case let .syncProgress(stage):
             log(.sync, .info, "Sync: \(stage)")
+        case let .rwfitDiagnostic(message, metadata):
+            log(.sync, .info, message, metadata: metadata)
+        case let .rwfitInitialization(state):
+            log(.connection, .info, "RwFit initialization: \(state)")
+        case let .rwfitMeasurement(type, status):
+            log(.sync, .info, "RwFit measurement status", metadata: ["type": String(type), "status": String(status)])
+        case let .rwfitMeasurementOutcome(outcome):
+            log(.sync, .info, "RwFit measurement outcome: \(outcome)")
+        case let .rwfitSyncOutcome(outcome):
+            recordRWfitOutcome(outcome)
         case .heartRateComplete:
             log(.sync, .info, "Heart-rate measurement complete")
         case .spo2Complete:
@@ -56,6 +66,19 @@ final class DiagnosticsSubscriber {
         }
     }
 
+    private func recordRWfitOutcome(_ outcome: RWfitSyncOutcome) {
+        switch outcome {
+        case let .success(records):
+            log(.sync, .info, "RwFit sync complete", metadata: ["importedRecords": String(records)])
+        case let .partial(records, reason):
+            log(.sync, .warn, "RwFit sync incomplete", metadata: ["importedRecords": String(records), "reason": reason])
+        case let .failed(reason):
+            log(.error, .error, "RwFit sync failed", metadata: ["reason": reason])
+        case .cancelled:
+            log(.sync, .info, "RwFit sync cancelled")
+        }
+    }
+
     private func log(_ category: WearableLogCategory, _ level: WearableLogLevel, _ message: String, metadata: [String: String]? = nil) {
         let json = metadata.flatMap { dict -> String? in
             guard let data = try? JSONSerialization.data(withJSONObject: dict) else { return nil }

PulseLoop/Events/PulseEventBus.swift

@@ -13,6 +13,12 @@ enum PulseEvent: Sendable {
     )
     case deviceForgotten
     case batteryLevel(percent: Int)
+    case decodedPacket(RingDecodedEvent)
+    case rwfitSyncOutcome(RWfitSyncOutcome)
+    case rwfitInitialization(RWfitInitializationState)
+    case rwfitDiagnostic(message: String, metadata: [String: String])
+    case rwfitMeasurementOutcome(RWfitMeasurementOutcome)
+    case rwfitMeasurement(type: UInt8, status: UInt8)
     case rawPacket(direction: PacketDirection, data: Data, decoded: RingDecodedEvent)
     case derivedUpdate(kind: String, entityType: String, entityId: String, payloadJSON: String?)
     case activityUpdate(timestamp: Date, steps: Int, distanceMeters: Double, calories: Double)
@@ -105,6 +111,7 @@ final class EventPersistenceSubscriber {
     /// woke every `@Query` hundreds of times (the re-render storm). Instead we insert/mutate without
     /// saving, then flush (one `save()` + one "data changed" signal) after the stream briefly idles
     /// or a hard cap of pending writes is reached.
+    private var mergingRWfitSleep = false
     private var pendingWrites = 0
     private var flushTask: Task<Void, Never>?
     /// Idle window after the last event before we flush a batch.
@@ -145,6 +152,10 @@ final class EventPersistenceSubscriber {
 
     func start() {
         guard task == nil else { return }
+        RWfitHistoryPersistence.save = { [weak self] events in
+            guard let self else { throw RWfitHistoryPersistence.PersistenceError.unavailable }
+            try self.saveRWfitHistory(events)
+        }
         task = Task {
             let stream = await PulseEventBus.shared.stream()
             for await event in stream {
@@ -156,6 +167,7 @@ final class EventPersistenceSubscriber {
     }
 
     func stop() {
+        RWfitHistoryPersistence.save = nil
         flushTask?.cancel()
         flushNow()
         task?.cancel()
@@ -171,6 +183,68 @@ final class EventPersistenceSubscriber {
         flushNow()
     }
 
+    /// Destructive history consumption is allowed only after this synchronous durable commit.
+    func saveRWfitHistory(_ events: [RingDecodedEvent], commit: (() throws -> Void)? = nil) throws {
+        let typed = try events.flatMap { decoded -> [PulseEvent] in
+            let mapped = RingEventBridge.events(for: decoded)
+            guard !mapped.isEmpty else { throw RWfitHistoryPersistence.PersistenceError.rejectedRecord }
+            return mapped
+        }
+        // Existing import helpers intentionally tolerate read errors for live data. Preflight those
+        // tables here so the destructive history path fails closed instead of treating a failed fetch
+        // as an empty database.
+        var measurements = try context.fetch(FetchDescriptor<Measurement>())
+        let activity = try RWfitActivityPersistence(context: context)
+        _ = try context.fetch(FetchDescriptor<SleepSession>())
+        _ = try context.fetch(FetchDescriptor<SleepStageBlock>())
+        // Commit unrelated live-event writes first. A failed history transaction can then roll
+        // back safely without discarding another stream's pending measurements.
+        try context.save()
+        pendingWrites = 0
+        flushTask?.cancel()
+        flushTask = nil
+        mergingRWfitSleep = true
+        defer { mergingRWfitSleep = false }
+        do {
+            for event in typed {
+                if activity.apply(event) { continue }
+                if case let .historyMeasurement(kind, value, timestamp) = event {
+                    if let row = measurements.first(where: {
+                        $0.kindRaw == kind.rawValue && $0.timestamp == timestamp && $0.sourceRaw == MeasurementSource.history.rawValue
+                    }) {
+                        row.value = value
+                    } else {
+                        let row = Measurement(kind: kind, value: value, unit: kind.unit, timestamp: timestamp, source: .history)
+                        context.insert(row)
+                        measurements.append(row)
+                        context.insert(DerivedUpdateRow(kind: "history_measurement", entityType: "measurement",
+                                                       entityId: row.id.uuidString))
+                        _ = ActivityRecorderService.linkSample(kind: kind, value: value, timestamp: timestamp,
+                            measurementId: row.id, source: .history, confidence: .known, context: context)
+                    }
+                    continue
+                }
+                if case let .sleepTimeline(timestamp, stages) = event {
+                    // Inject throwing reads so the shared best-effort sleep helper never performs
+                    // a swallowed fetch on the destructive-import path.
+                    let sessions = try context.fetch(FetchDescriptor<SleepSession>())
+                    let blocks = try context.fetch(FetchDescriptor<SleepStageBlock>())
+                    persistSleepTimeline(start: timestamp, stages: stages, sessions: sessions, blocks: blocks)
+                } else {
+                    applyPersist(event)
+                }
+            }
+            if let commit { try commit() } else { try context.save() }
+            for day in activity.touchedDays { DailyCalorieEstimator.markDirty(day) }
+            pendingWrites = 0
+            PulseDataChange.shared.notify()
+        } catch {
+            context.rollback()
+            seenHistoryKeys.removeAll()
+            throw error
+        }
+    }
+
     func persist(_ event: PulseEvent) {
         applyPersist(event)
         scheduleFlush()
@@ -214,7 +288,7 @@ final class EventPersistenceSubscriber {
             device.bleAddressHint = address ?? device.bleAddressHint
             if state == .connected {
                 device.lastConnectedAt = Date()
-                device.lastSyncAt = Date()
+                if device.deviceType != .rwfit { device.lastSyncAt = Date() }
             }
             context.insert(device)
         case let .deviceIdentified(deviceType, wearableModelID, advertisedName, capabilities):
@@ -352,7 +426,7 @@ final class EventPersistenceSubscriber {
             // Stamp the *completion* of a full history sync so the coach freshness gate can tell a
             // finished sync from a bare CONNECT (`lastSyncAt`, re-stamped every connect).
             if stage == "done" {
-                if let device = DeviceRepository.current(context: context) {
+                if let device = DeviceRepository.current(context: context), device.deviceType != .rwfit {
                     device.lastFullSyncAt = Date()
                 }
                 // The rows are committed by now; the next sync re-checks against the database.
@@ -361,6 +435,16 @@ final class EventPersistenceSubscriber {
                 // batched flush below saves the writes and fires the coalesced change signal.
                 DailyCalorieEstimator.flushDirty(context: context)
             }
+        case let .rwfitSyncOutcome(outcome):
+            if case .success = outcome, let device = DeviceRepository.current(context: context) {
+                let now = Date()
+                device.lastSyncAt = now
+                device.lastFullSyncAt = now
+            }
+            seenHistoryKeys.removeAll(keepingCapacity: true)
+            DailyCalorieEstimator.flushDirty(context: context)
+        case .decodedPacket, .rwfitInitialization, .rwfitDiagnostic, .rwfitMeasurement, .rwfitMeasurementOutcome:
+            break
         // `.wearState` is a live condition the measurement flow reacts to, not data — nothing to store.
         case .heartRateComplete, .spo2Progress, .spo2Complete, .workoutStarted, .workoutPaused,
              .workoutResumed, .workoutFinished, .coachTrace, .wearState:
@@ -466,7 +550,8 @@ final class EventPersistenceSubscriber {
     /// recomputing session bounds. The ring streams ~20 timeline packets (15 samples each) per
     /// night, so blocks must accumulate into one session rather than spawning a session per
     /// packet. Mirrors `persistence._on_sleep_timeline`.
-    private func persistSleepTimeline(start: Date, stages: [SleepStage]) {
+    private func persistSleepTimeline(start: Date, stages: [SleepStage],
+                                      sessions: [SleepSession]? = nil, blocks: [SleepStageBlock]? = nil) {
         let calendar = Calendar.current
 
         // Group packets by the waking-day boundary (sleep from 7 PM rolls to the next morning) so a
@@ -478,7 +563,7 @@ final class EventPersistenceSubscriber {
         // day is empty), deduping by block start across every session on the day. Then hand off to
         // `SleepService.reconcileWakingDay`, which re-splits the day's blocks into distinct sessions
         // (main night vs. naps separated by a >= 60 min gap) and recomputes each session's bounds.
-        let allSessions = (try? context.fetch(FetchDescriptor<SleepSession>())) ?? []
+        let allSessions = sessions ?? ((try? context.fetch(FetchDescriptor<SleepSession>())) ?? [])
         let daySessions = allSessions.filter { calendar.isDate($0.date, inSameDayAs: dateKey) }
         let container = daySessions.min { $0.startAt < $1.startAt }
             ?? {
@@ -491,8 +576,39 @@ final class EventPersistenceSubscriber {
         let sessionsForDay = daySessions.contains(where: { $0.id == container.id }) ? daySessions : daySessions + [container]
 
         let daySessionIds = Set(sessionsForDay.map { $0.id })
-        let existingDayBlocks = ((try? context.fetch(FetchDescriptor<SleepStageBlock>())) ?? [])
+        let existingDayBlocks = (blocks ?? ((try? context.fetch(FetchDescriptor<SleepStageBlock>())) ?? []))
             .filter { daySessionIds.contains($0.sessionId) }
+        if mergingRWfitSleep {
+            // A recovered journal may extend or overlap an already-imported session. Normalize by
+            // minute so a shorter previously saved block cannot suppress the rest of a replay.
+            var minuteStages: [Date: SleepStage] = [:]
+            for block in existingDayBlocks {
+                for minute in 0..<max(0, block.durationMinutes) {
+                    minuteStages[block.startAt.addingTimeInterval(Double(minute) * 60)] = block.stage
+                }
+            }
+            for (minute, stage) in stages.enumerated() {
+                minuteStages[start.addingTimeInterval(Double(minute) * 60)] = stage
+            }
+            for block in existingDayBlocks { context.delete(block) }
+            let dates = minuteStages.keys.sorted()
+            var merged: [SleepStageBlock] = []
+            for date in dates {
+                guard let stage = minuteStages[date] else { continue }
+                if let last = merged.last, last.stage == stage,
+                   last.startAt.addingTimeInterval(Double(last.durationMinutes) * 60) == date {
+                    last.durationMinutes += 1
+                } else {
+                    let block = SleepStageBlock(sessionId: container.id, startAt: date,
+                                                startMinute: 0, durationMinutes: 1, stage: stage)
+                    context.insert(block)
+                    merged.append(block)
+                }
+            }
+            SleepService.reconcileWakingDay(dateKey: dateKey, context: context,
+                                            daySessions: sessionsForDay, dayBlocks: merged)
+            return
+        }
         var existingStarts = Set(existingDayBlocks.map { $0.startAt })
         var newBlocks: [SleepStageBlock] = []
 

PulseLoop/Persistence/SeedData.swift

@@ -328,6 +328,7 @@ enum SeedData {
 
     @MainActor
     static func clearAll(_ context: ModelContext) {
+        try? RWfitHistoryPersistence.clearAllSleepPages()
         deleteAll(Device.self, context)
         deleteAll(ActivityDaily.self, context)
         deleteAll(Measurement.self, context)

PulseLoop/RingProtocol/RWfitActivityPersistence.swift

@@ -0,0 +1,75 @@
+import Foundation
+import SwiftData
+
+/// A history transaction's activity snapshot. Reads throw; mutations never save internally.
+/// The caller commits all buckets and derived daily totals together before consuming ring history.
+@MainActor
+final class RWfitActivityPersistence {
+    nonisolated deinit {}
+    private let context: ModelContext
+    private var days: [ActivityDaily]
+    private var buckets: [ActivityBucketSample]
+    private(set) var touchedDays: Set<Date> = []
+
+    init(context: ModelContext) throws {
+        self.context = context
+        days = try context.fetch(FetchDescriptor<ActivityDaily>())
+        buckets = try context.fetch(FetchDescriptor<ActivityBucketSample>())
+    }
+
+    func apply(_ event: PulseEvent) -> Bool {
+        switch event {
+        case let .activityUpdate(timestamp, steps, distanceMeters, calories):
+            let row = day(timestamp)
+            row.steps = max(row.steps, steps)
+            row.distanceMeters = max(row.distanceMeters, distanceMeters)
+            row.calories = max(row.calories, calories)
+            row.source = "live"
+            recordChange(row)
+            return true
+        case let .activityBucket(timestamp, steps, distanceMeters):
+            let epoch = Int(timestamp.timeIntervalSince1970)
+            if let bucket = buckets.first(where: { $0.startEpoch == epoch }) {
+                bucket.steps = steps
+                bucket.distanceMeters = distanceMeters
+                bucket.updatedAt = Date()
+            } else {
+                let bucket = ActivityBucketSample(timestamp: timestamp, steps: steps, distanceMeters: distanceMeters)
+                context.insert(bucket)
+                buckets.append(bucket)
+            }
+            let row = day(timestamp)
+            let dayBuckets = buckets.filter { $0.date == row.date }
+            let totalSteps = dayBuckets.reduce(0) { $0 + $1.steps }
+            let totalDistance = dayBuckets.reduce(0.0) { $0 + $1.distanceMeters }
+            // Today's cumulative reading can lead the ring's most recent logged bucket.
+            if Calendar.current.isDateInToday(row.date), row.steps > totalSteps || row.distanceMeters > totalDistance {
+                row.steps = max(row.steps, totalSteps)
+                row.distanceMeters = max(row.distanceMeters, totalDistance)
+            } else {
+                row.steps = totalSteps
+                row.distanceMeters = totalDistance
+                row.source = ActivityService.ringHistorySource
+            }
+            recordChange(row)
+            return true
+        default:
+            return false
+        }
+    }
+
+    private func day(_ timestamp: Date) -> ActivityDaily {
+        let date = Calendar.current.startOfDay(for: timestamp)
+        if let row = days.first(where: { $0.date == date }) { return row }
+        let row = ActivityDaily(date: date, source: ActivityService.ringHistorySource)
+        context.insert(row)
+        days.append(row)
+        return row
+    }
+
+    private func recordChange(_ row: ActivityDaily) {
+        row.syncedAt = Date()
+        row.updatedAt = Date()
+        touchedDays.insert(row.date)
+    }
+}