diff --git a/CHANGELOG.md b/CHANGELOG.md index 659c895..aac1e6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,28 @@ ## Unreleased — 0.1.6 +- Corrected delayed session finalization so a heart-rate capture cannot extend + beyond the explicit factual session end. Android now transactionally removes + strict post-end BPM/RR tails while retaining an exact-cutoff sample. For an + already acknowledged capture, the separately versioned + `trainlog-heart-rate-corrections` V1 companion carries an immutable + `capture_id|cutoff` correction through the existing causal ledger without + widening frozen causal-deletions V1. Old Heart Rate V1 generation replay + applies that cutoff before identity comparison and cannot resurrect removed + measurements. A guarded dry-run recovery tool reports before/after counts and + statistics without rewriting published manifests or ACKs. +- Completed the paired-device recovery for session + `se_fa845379-794d-4bb6-8cd3-48187ebbd134`: Android and desktop now retain + 3,429 BPM samples and 3,429 RR values through the inclusive 11:16 cutoff, + and two successive full MTP generation exchanges completed with durable + consumed ACKs. Replaying the original 4,919-sample artifact cannot restore + the removed tail. Generation parent selection now excludes archived or + legacy acknowledgement rows lacking durable ACK evidence, while consumers + accept the producer's declared parent when it is one of multiple retained + current tips. Android Sleep V2 persistence also names medication-intake + columns explicitly, preserving `quantity`, note, and creation time across + the additive v33 layout. + - Recovered a real divergent Sleep revision identity without choosing an arbitrary winner. The unpublished desktop branch was re-identified with its exact content, the immutable Android revision retained its published ID and @@ -22,8 +44,8 @@ performed duration from the persisted exercise timeline, including preserved drafts closed by an older build, while keeping planned set counts independent from valid actual sets, repetitions and loads. A strictly identity-bound - private recovery build may supply the factual session end; immutable late - heart-rate samples remain retained inside their capture bounds. + private recovery build may supply the factual session end; strict late + heart-rate BPM/RR tails are now removed transactionally at that boundary. - Hardened desktop Sleep generation consumption when an identical durable revision has no local publication row. Content comparison now reads the durable revision directly, acknowledges an exact match, and reports a real diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt index cec70eb..d5db3db 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt @@ -72,6 +72,9 @@ internal class SyncGenerationService(private val repository: TrainlogRepository) Kind("trainlog-program-executions", 1, "program-executions-v1.json", false), "sleep-diary" to Kind("trainlog-sleep-diary", 2, "sleep-diary-v2.json", false), "heart-rate" to Kind("trainlog-heart-rate", 1, "heart-rate-v1.json", false), + "heart-rate-corrections" to + Kind("trainlog-heart-rate-corrections", 1, + "heart-rate-corrections-v1.json", false), "cardio-calibrations" to Kind("trainlog-cardio-calibrations", 1, "cardio-calibrations-v1.json", false), "cardio-guidance" to @@ -359,10 +362,35 @@ internal class SyncGenerationService(private val repository: TrainlogRepository) ) } val generatedAt = OffsetDateTime.now().toString() + /* WHY: a fully acknowledged fork can remain in the immutable + * ledger after its payload has been durably archived. + * CONTRACT: archived evidence stays queryable but no longer + * participates in the active producer-lineage tip selection. + * INVARIANT: only unarchived rows backed by a full durable ACK can + * become the parent; two such active tips still reject capture. */ val parent = repository.inSyncGenerationTransaction { db -> db.rawQuery( - "SELECT g.generation_id FROM sync_generations g WHERE g.producer_peer_id=? AND g.consumer_peer_id=? AND g.status='acknowledged' AND NOT EXISTS(SELECT 1 FROM sync_generations c WHERE c.parent_generation_id=g.generation_id AND c.status='acknowledged') LIMIT 2", + "SELECT g.generation_id FROM sync_generations g WHERE " + + "g.producer_peer_id=? AND g.consumer_peer_id=? AND " + + "g.status='acknowledged' AND EXISTS(SELECT 1 FROM " + + "sync_acknowledgements k WHERE " + + "k.generation_id=g.generation_id AND " + + "k.result='consumed' AND " + + "k.durability='sqlite-commit-full') AND " + + "NOT EXISTS(SELECT 1 FROM " + + "sync_generation_archives a WHERE " + + "a.generation_id=g.generation_id) AND NOT EXISTS(" + + "SELECT 1 FROM sync_generations c WHERE " + + "c.parent_generation_id=g.generation_id AND " + + "c.status='acknowledged' AND EXISTS(SELECT 1 FROM " + + "sync_acknowledgements ck WHERE " + + "ck.generation_id=c.generation_id AND " + + "ck.result='consumed' AND " + + "ck.durability='sqlite-commit-full') AND " + + "NOT EXISTS(SELECT 1 FROM " + + "sync_generation_archives ca WHERE " + + "ca.generation_id=c.generation_id)) LIMIT 2", arrayOf(producer, consumerPeerId), ) .use { @@ -440,6 +468,24 @@ internal class SyncGenerationService(private val repository: TrainlogRepository) arrayOf(operationId, generationId, if (existed) 0 else 1), ) } + val heartRateCorrections = + JSONObject(checkNotNull(artifacts["heart-rate-corrections"])) + .getJSONArray("corrections") + for (index in 0 until heartRateCorrections.length()) { + val operationId = + heartRateCorrections.getJSONObject(index).getString("operation_id") + val existed = + db.rawQuery( + "SELECT 1 FROM sync_causal_publications " + + "WHERE operation_id=? LIMIT 1", + arrayOf(operationId), + ) + .use { it.moveToFirst() } + db.execSQL( + "INSERT INTO sync_causal_publications VALUES(?,?,?)", + arrayOf(operationId, generationId, if (existed) 0 else 1), + ) + } val sleep = JSONObject(checkNotNull(artifacts["sleep-diary"])) val sleepEntries = sleep.getJSONArray("entries") for (index in 0 until sleepEntries.length()) { @@ -853,26 +899,38 @@ internal class SyncGenerationService(private val repository: TrainlogRepository) } } val producerPeer = manifest.getJSONObject("producer").getString("peer_id") - val expectedParent = + /* WHY: the producer can archive one transient fork while this + * consumer retains both immutable consumed audit rows. + * CONTRACT: a new generation extends any current consumed tip; + * a null, old, or unrelated parent remains invalid. + * INVARIANT: fork history is never rewritten to select a tip. */ + val tipPredicate = + "g.producer_peer_id=? AND g.result='consumed' AND " + + "NOT EXISTS(SELECT 1 FROM sync_consumed_generations c " + + "WHERE c.parent_generation_id=g.generation_id AND " + + "c.result='consumed')" + val hasCurrentTip = db.rawQuery( - "SELECT g.generation_id FROM sync_consumed_generations g WHERE g.producer_peer_id=? AND g.result='consumed' AND NOT EXISTS(SELECT 1 FROM sync_consumed_generations c WHERE c.parent_generation_id=g.generation_id AND c.result='consumed') LIMIT 2", + "SELECT 1 FROM sync_consumed_generations g WHERE " + + tipPredicate + " LIMIT 1", arrayOf(producerPeer), ) - .use { - if (!it.moveToFirst()) null - else { - val value = it.getString(0) - if (it.moveToNext()) - throw SyncGenerationException( - "consumed generation lineage has multiple tips" - ) - value - } - } + .use { it.moveToFirst() } val suppliedParent = if (manifest.isNull("parent_generation_id")) null else manifest.getString("parent_generation_id") - if (suppliedParent != expectedParent) + val suppliedParentIsTip = + suppliedParent != null && + db.rawQuery( + "SELECT 1 FROM sync_consumed_generations g WHERE " + + "g.generation_id=? AND " + tipPredicate + " LIMIT 1", + arrayOf(suppliedParent, producerPeer), + ) + .use { it.moveToFirst() } + if ( + (!hasCurrentTip && suppliedParent != null) || + (hasCurrentTip && !suppliedParentIsTip) + ) throw SyncGenerationException("generation lineage is stale or unrelated") val a = frozen.artifacts fun apply(name: String, result: () -> Any) { diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt b/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt index 212d54f..609532c 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt @@ -715,7 +715,10 @@ class TrainlogRepository( arrayOf(revisionId,event.eventId.ifEmpty { sleepId("sle") },event.type.wireValue,event.startAt,event.endAt), ) } draft.intakes.forEach { intake -> db.execSQL( - "INSERT INTO sleep_medication_intakes VALUES(?,?,?,?,?,?,?,?,?,?)", + "INSERT INTO sleep_medication_intakes(" + + "revision_id,intake_id,medication_id,medication_name,taken_at," + + "dose_value,dose_unit,quantity,note,created_at" + + ") VALUES(?,?,?,?,?,?,?,?,?,?)", arrayOf(revisionId, intake.intakeId.ifEmpty { sleepId("mdi") }, intake.medicationId, intake.medicationName, intake.takenAt, intake.doseValue, intake.doseUnit, intake.quantity, @@ -1608,7 +1611,10 @@ class TrainlogRepository( draft.events.forEach { event -> db.execSQL("INSERT INTO sleep_diary_events VALUES(?,?,?,?,?)", arrayOf(revisionId,event.eventId,event.type.wireValue,event.startAt,event.endAt)) } draft.intakes.forEach { intake -> db.execSQL( - "INSERT INTO sleep_medication_intakes VALUES(?,?,?,?,?,?,?,?,?,?)", + "INSERT INTO sleep_medication_intakes(" + + "revision_id,intake_id,medication_id,medication_name,taken_at," + + "dose_value,dose_unit,quantity,note,created_at" + + ") VALUES(?,?,?,?,?,?,?,?,?,?)", arrayOf(revisionId, intake.intakeId, intake.medicationId, intake.medicationName, intake.takenAt, intake.doseValue, intake.doseUnit, intake.quantity, @@ -3628,6 +3634,7 @@ class TrainlogRepository( artifacts["body-zones"] = buildExerciseBodyZonesJson(true) artifacts["feedback"] = buildTrainingFeedbackJson(true) artifacts["causal-deletions"] = buildCausalDeletionExportV1Json() + artifacts["heart-rate-corrections"] = buildHeartRateCorrectionsV1Json() artifacts["program-executions"] = buildProgramExecutionsV1Json() artifacts["sleep-diary"] = buildSleepDiaryV1Json() artifacts["heart-rate"] = buildHeartRateV1Json() @@ -5847,37 +5854,44 @@ class TrainlogRepository( arrayOf(sessionId), ).use { cursor -> if (cursor.moveToFirst()) cursor.getString(0) else null } if (activeCapture != null) { - var captureEndedAt = finalizedAt - var captureEndKey = parsedFinalizedAt + val removedSequences = mutableListOf() db.rawQuery( - "SELECT observed_at FROM heart_rate_samples WHERE capture_id=?", + "SELECT sequence,observed_at FROM heart_rate_samples " + + "WHERE capture_id=? ORDER BY sequence", arrayOf(activeCapture), ).use { cursor -> while (cursor.moveToNext()) { - val observedAt = cursor.getString(0) + val observedAt = cursor.getString(1) val observedKey = TrainlogTimestamp.parse(observedAt) ?: error("Heart-rate sample has an invalid timestamp.") - if (observedKey > captureEndKey) { - captureEndKey = observedKey - captureEndedAt = observedAt + if (observedKey > parsedFinalizedAt) { + removedSequences += cursor.getLong(0) } } } /* * WHY: delayed recovery can establish a factual session end * earlier than samples recorded while the failed draft stayed - * open. Dropping or retiming those immutable measurements - * would be data loss. + * open; those later measurements are not part of the session. * CONTRACT: the session/timeline use the requested factual - * end; the sensor capture expands only to its latest observed - * sample so Heart Rate V1 remains internally valid. - * INVARIANT: every retained sample lies within its stopped - * capture and no sample/RR identity or payload is rewritten. + * end inclusively. Samples strictly after it are deleted, and + * their RR children follow through the declared cascade. + * INVARIANT: every retained sample is at or before the stopped + * capture end; retained sample identities are never rewritten. */ + removedSequences.forEach { sequence -> + check( + db.delete( + "heart_rate_samples", + "capture_id=? AND sequence=?", + arrayOf(activeCapture, sequence.toString()), + ) == 1, + ) + } db.execSQL( "UPDATE heart_rate_captures SET ended_at=?,active_slot=NULL," + "active_exercise_entry_id=NULL WHERE capture_id=? AND active_slot=1", - arrayOf(captureEndedAt, activeCapture), + arrayOf(finalizedAt, activeCapture), ) } closeCardioGuidanceRunForSession(db, sessionId, finalizedAt) @@ -7362,7 +7376,10 @@ class TrainlogRepository( private fun buildCausalDeletionExportV1Json(db: SQLiteDatabase): String { val operations = JSONArray() - db.rawQuery("SELECT operation_id,target_kind,target_id,creator_id,predecessor_revision_id,created_at,payload_sha256 FROM sync_causal_operations ORDER BY operation_id", null).use { cursor -> + db.rawQuery("SELECT operation_id,target_kind,target_id,creator_id," + + "predecessor_revision_id,created_at,payload_sha256 FROM sync_causal_operations " + + "WHERE target_kind IN('session','execution_draft','exercise','body_observation'," + + "'custom_equipment','feedback','body_zone_relation') ORDER BY operation_id", null).use { cursor -> while (cursor.moveToNext()) { check(operations.length() < MAX_CAUSAL_OPERATIONS) { "Causal protection set exceeds artifact bound." } operations.put(JSONObject().put("operation_id", cursor.getString(0)).put("target_kind", cursor.getString(1)) @@ -7375,6 +7392,50 @@ class TrainlogRepository( .put("generated_at", OffsetDateTime.now().toString()).put("operations", operations).toString() } + /** + * WHY: Heart Rate V1 bytes and causal-deletions V1 semantics are frozen. + * CONTRACT: acknowledged capture cutoffs therefore travel in their own + * strict V1 companion while reusing the durable causal operation ledger. + * INVARIANT: this exporter never emits a heart-rate operation through the + * frozen general-deletion artifact. + */ + internal fun buildHeartRateCorrectionsV1Json(): String { + val corrections = JSONArray() + database.readableDatabase.rawQuery( + "SELECT operation_id,target_id,creator_id,predecessor_revision_id," + + "created_at,payload_sha256 FROM sync_causal_operations " + + "WHERE target_kind='heart_rate_tail' ORDER BY operation_id", + null, + ).use { cursor -> + while (cursor.moveToNext()) { + check(corrections.length() < MAX_HEART_RATE_CORRECTIONS) + val target = cursor.getString(1).split('|', limit = 2) + check(target.size == 2 && sleepIdentity(target[0], "hrc") && + TrainlogTimestamp.parse(target[1]) != null) + corrections.put( + JSONObject() + .put("operation_id", cursor.getString(0)) + .put("capture_id", target[0]) + .put("cutoff", target[1]) + .put("creator_id", cursor.getString(2)) + .put("predecessor_revision_id", cursor.getString(3)) + .put("created_at", cursor.getString(4)) + .put("payload_sha256", cursor.getString(5)), + ) + } + } + return JSONObject() + .put("format", "trainlog-heart-rate-corrections") + .put("version", 1) + .put("generated_at", OffsetDateTime.now().toString()) + .put("corrections", corrections) + .toString() + .also { + check(it.toByteArray(StandardCharsets.UTF_8).size <= + MAX_HEART_RATE_CORRECTION_ARTIFACT_BYTES) + } + } + fun applyCausalDeletionExportV1Json(json: String): CausalDeleteResult { if (json.toByteArray().size > MAX_CAUSAL_ARTIFACT_BYTES || !hasStrictJsonShape(json)) return CausalDeleteResult.Invalid val root = try { JSONObject(json) } catch (_: Exception) { return CausalDeleteResult.Invalid } @@ -12229,6 +12290,8 @@ private const val MAX_SYNC_NOTE_UTF8_BYTES = 4096 private const val MAX_PENDING_EXECUTION_DRAFTS = 15 private const val MAX_CAUSAL_OPERATIONS = 4096 private const val MAX_CAUSAL_ARTIFACT_BYTES = 4 * 1024 * 1024 +private const val MAX_HEART_RATE_CORRECTIONS = 256 +private const val MAX_HEART_RATE_CORRECTION_ARTIFACT_BYTES = 4 * 1024 * 1024 private const val MAX_PLAN_SETS = 64 private const val MAX_PLAN_REPS = 10000 private const val MAX_PLAN_DURATION_SECONDS = 86400 diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SessionTimelineRepositoryTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SessionTimelineRepositoryTest.kt index d5731af..d2a9635 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/data/SessionTimelineRepositoryTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SessionTimelineRepositoryTest.kt @@ -445,9 +445,9 @@ class SessionTimelineRepositoryTest { val heart = JSONObject(repository.buildHeartRateV1Json()) val capture = heart.getJSONArray("captures").getJSONObject(0) - assertEquals(lateSampleAt.toString(), capture.getString("ended_at")) - assertEquals(2, capture.getJSONArray("samples").length()) - assertEquals(1, capture.getJSONArray("samples").getJSONObject(1) + assertEquals(finalizationAt.toString(), capture.getString("ended_at")) + assertEquals(1, capture.getJSONArray("samples").length()) + assertEquals(1, capture.getJSONArray("samples").getJSONObject(0) .getJSONArray("rr_intervals_1024").length()) assertNull(repository.activeHeartRateCapture()) } finally { diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt index 0436ffd..ae13cf8 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt @@ -1282,6 +1282,18 @@ class SyncGenerationServiceTest { .getJSONArray("captures") .length(), ) + val heartRateCorrections = + JSONObject( + java.io.File( + androidToDesktop.stagingDirectory, + "heart-rate-corrections-v1.json", + ).readText(), + ) + assertEquals( + "trainlog-heart-rate-corrections", + heartRateCorrections.getString("format"), + ) + assertEquals(0, heartRateCorrections.getJSONArray("corrections").length()) val androidPublished = sourceService.publish( androidToDesktop, diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/TrainlogRepositoryDraftTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/TrainlogRepositoryDraftTest.kt index 4569fd3..5090980 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/data/TrainlogRepositoryDraftTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/data/TrainlogRepositoryDraftTest.kt @@ -13,6 +13,8 @@ import com.labfytools.trainlog.model.BodyObservationDraft import com.labfytools.trainlog.model.ExerciseDataFields import com.labfytools.trainlog.model.ExerciseEditInput import com.labfytools.trainlog.model.ExerciseProfile +import com.labfytools.trainlog.model.HeartRateContextKind +import com.labfytools.trainlog.model.HeartRateRrInterval import com.labfytools.trainlog.model.NewExerciseProfile import com.labfytools.trainlog.model.RecordingMode import com.labfytools.trainlog.model.SessionDraft @@ -56,6 +58,126 @@ class TrainlogRepositoryDraftTest { databaseName = "draft-test-${UUID.randomUUID()}.db" } + @Test + fun delayedFactualFinalizationTruncatesHeartRateSamplesAndRrInclusively() { + val repo = openRepository() + val exercise = createExercise(repo, "Récupération cardio", RecordingMode.SETS, TrackingMode.REPS) + val sessionId = "se_aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + val entryId = "sxe_bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + val startedAt = "2026-09-24T10:00:00+02:00" + val cutoff = "2026-09-24T11:16:00+02:00" + assertEquals( + ActiveDraftMutationResult.Saved, + repo.saveActiveSessionDraft( + ActiveSessionDraft( + sessionId = sessionId, + startedAt = startedAt, + exercises = listOf( + SessionExerciseDraft( + entryId = entryId, + exercise = exercise, + sets = listOf(SessionSetDraft(reps = 8, weightKg = 20.0)), + ), + ), + ), + ), + ) + val capture = repo.startHeartRateCapture( + HeartRateContextKind.SESSION, + sessionId, + startedAt, + "Synthetic", + ) as TrainlogRepository.StartHeartRateCaptureResult.Started + assertTrue(repo.startActiveSessionExercise( + entryId, "2026-09-24T10:30:00+02:00", + ) is TrainlogRepository.SessionExerciseTimingResult.Started) + assertTrue(repo.finishActiveSessionExercise( + entryId, "2026-09-24T11:15:00+02:00", + ) is TrainlogRepository.SessionExerciseTimingResult.Finished) + listOf( + "2026-09-24T11:15:59.999999+02:00" to 100, + cutoff to 101, + "2026-09-24T11:16:00.000001+02:00" to 40, + ).forEachIndexed { index, (observedAt, bpm) -> + assertTrue( + repo.appendHeartRateSample( + capture.captureId, + observedAt, + bpm, + rrIntervals = listOf(HeartRateRrInterval(0, 900 + index)), + ) is TrainlogRepository.HeartRateMutationResult.Applied, + ) + } + + assertEquals( + FinalizeActiveDraftResult.Saved(sessionId), + repo.finalizeActiveSessionDraft(cutoff), + ) + SQLiteDatabase.openDatabase( + context.getDatabasePath(databaseName).path, + null, + SQLiteDatabase.OPEN_READONLY, + ).use { db -> + assertEquals(cutoff, db.rawQuery( + "SELECT ended_at FROM heart_rate_captures WHERE capture_id=?", + arrayOf(capture.captureId), + ).use { it.moveToFirst(); it.getString(0) }) + assertEquals( + listOf("2026-09-24T11:15:59.999999+02:00", cutoff), + db.rawQuery( + "SELECT observed_at FROM heart_rate_samples WHERE capture_id=? " + + "ORDER BY sequence", + arrayOf(capture.captureId), + ).use { cursor -> buildList { while (cursor.moveToNext()) add(cursor.getString(0)) } }, + ) + assertEquals(2, db.rawQuery( + "SELECT COUNT(*) FROM heart_rate_rr_intervals WHERE capture_id=?", + arrayOf(capture.captureId), + ).use { it.moveToFirst(); it.getInt(0) }) + db.rawQuery("PRAGMA foreign_key_check", null).use { assertFalse(it.moveToFirst()) } + } + } + + @Test + fun heartRateCorrectionsRemainOutsideFrozenCausalDeletionV1() { + val repo = openRepository() + repo.buildCausalDeletionExportV1Json() + SQLiteDatabase.openDatabase( + context.getDatabasePath(databaseName).path, + null, + SQLiteDatabase.OPEN_READWRITE, + ).use { db -> + db.execSQL( + "INSERT INTO sync_causal_operations(" + + "operation_id,target_kind,target_id,creator_id,predecessor_revision_id," + + "created_at,payload_sha256,publication_context) VALUES(?,?,?,?,?,?,?,NULL)", + arrayOf( + "hrx_11111111-1111-4111-8111-111111111111", + "heart_rate_tail", + "hrc_22222222-2222-4222-8222-222222222222|" + + "2026-09-24T11:16:00+02:00", + "peer_test", + "lv_test", + "2026-09-24T12:00:00+02:00", + "0".repeat(64), + ), + ) + } + + assertEquals( + 0, + JSONObject(repo.buildCausalDeletionExportV1Json()) + .getJSONArray("operations") + .length(), + ) + val correction = JSONObject(repo.buildHeartRateCorrectionsV1Json()) + assertEquals("trainlog-heart-rate-corrections", correction.getString("format")) + assertEquals( + "hrx_11111111-1111-4111-8111-111111111111", + correction.getJSONArray("corrections").getJSONObject(0).getString("operation_id"), + ) + } + @After fun tearDown() { repository?.close() diff --git a/docs/android.md b/docs/android.md index 76233df..351ee8e 100644 --- a/docs/android.md +++ b/docs/android.md @@ -32,6 +32,9 @@ imports compare a known ancestor's complete canonical payload before the stale ancestor shortcut, so a reused ID with a different parent or content cannot be hidden below a later local tip. All user edits continue to create a fresh UUIDv4 revision and preserve the previous payload and causal parent. +Medication-intake writes name all columns explicitly, so the additive +`quantity` column cannot reorder note or creation-time values in upgraded +schema-v33 databases. The quick Sleep capture derives one night identity from the existing 18:00-to-18:00 date rule. A medication action can create that pending entry @@ -698,10 +701,12 @@ speed or distance still require those measured values explicitly. Ordinary completion uses the save-time timestamp. A private recovery build may embed one exact session identity and one explicit offset-aware factual end time; only that matching draft uses the supplied time. The repository rejects an end -before the session start or latest exercise marker. When immutable heart-rate -samples arrived after a failed factual session end, the session retains the -factual end while the stopped capture remains bounded by its latest retained -sample; no BPM or RR measurement is deleted or retimed. +before the session start or latest exercise marker. When heart-rate samples +arrived after a failed factual session end, delayed finalization keeps samples +at or before that inclusive end and transactionally deletes the strict tail. +RR children follow their BPM sample through the declared foreign-key cascade, +and the stopped capture uses the exact factual end. Retained sample identities +and values are never retimed or renumbered. PC catalog reconciliation preserves draft references through catalog row ownership. If an editing selection no longer resolves, only the selection is diff --git a/docs/current_state.md b/docs/current_state.md index a7ca447..6c40a7d 100644 --- a/docs/current_state.md +++ b/docs/current_state.md @@ -164,6 +164,7 @@ parallel implementation of its rules. | Local Web | `TRAINLOG_WEB_V1=CONTRACT_FROZEN / IMPLEMENTATION_STARTED` | | Web Exercises V1 | `TRAINLOG_WEB_EXERCISES_V1=PASS` | | Cardio persistence/sync | `TRAINLOG_CARDIO_PERSISTENCE_SYNC_V1=PASS`: Android v27 + desktop v30, strict optional `trainlog-heart-rate` V1 Android→desktop companion, generation relation and correlated ACK | +| Recovered session HR tail | `TRAINLOG_HR_TRUNCATE_AFTER_SESSION_END_V1=PASS`: delayed Android finalization truncates BPM/RR strictly after the inclusive factual end, and the separate `trainlog-heart-rate-corrections` V1 companion reuses causal state without widening frozen causal-deletions V1. The paired-device recovery for `se_fa845379-794d-4bb6-8cd3-48187ebbd134` retained 3,429 BPM/RR facts through 11:16; two successive full MTP exchanges received consumed ACKs, and replay of the original 4,919-sample Heart Rate V1 artifact did not resurrect the tail. | | Cardio BLE diagnostics | `TRAINLOG_CARDIO_BLE_DIAGNOSTICS_V1=PASS`: real CYCPLUS H2 validated on Android; `0x180D` + `0x2A37`, Battery Service, approximately 1 Hz notifications and one raw RR interval per observed notification; contact and energy fields absent in the observed payloads | | Cardio persistent acquisition | `TRAINLOG_CARDIO_PERSISTENT_ACQUISITION_V1=PASS`: selected sensor persists, dedicated connected-device foreground service reconnects, stale BPM is cleared, and the global Android `♥ BPM` indicator is grey/orange/green by connection state; real CYCPLUS restart and Bluetooth-off/on recovery validated on private vc40 | | Training timeline | `TRAINLOG_CARDIO_TRAINING_TIMELINE_V1=PASS`: real session bounds reuse canonical lifecycle timestamps; exercise start/end are explicit user actions; live BPM samples snapshot the active stable `entry_id`; session finalization refuses an unclosed exercise; timeline sync is Android→desktop idempotent | diff --git a/docs/database.md b/docs/database.md index f99bce6..09dc986 100644 --- a/docs/database.md +++ b/docs/database.md @@ -1,5 +1,23 @@ # Desktop database +## Causal heart-rate tail correction (schema v36, no migration) + +The separate `trainlog-heart-rate-corrections` V1 companion reuses the existing +`sync_causal_operations` and `sync_causal_state` tables for the additive local +`heart_rate_tail` state. The frozen `trainlog-causal-deletions` V1 artifact +continues to emit and accept exactly its original seven target kinds. A heart- +rate correction's composite target contains the stable capture identity and an +inclusive offset-aware cutoff. Applying it deletes only BPM samples strictly +after the cutoff; declared RR foreign keys cascade those exact children, and +the capture end becomes the cutoff in the same transaction. The predecessor +revision hashes the original ordered raw capture, including RR values, so a +correction cannot be silently applied to a different measurement history. + +The immutable Heart Rate V1 generation remains audit evidence. Import applies +known causal cutoffs before stable-identity comparison or insertion, which +makes old-generation replay idempotent without rewriting its manifest or ACK. +No schema or frozen `TRAINLOG_FORMAT_V1` change is involved. + ## Schema v36: append-only Sleep revision payloads Schema v36 adds database-level `BEFORE UPDATE` and `BEFORE DELETE` guards for diff --git a/docs/exchange_format.md b/docs/exchange_format.md index 5d8c4fc..3413d1a 100644 --- a/docs/exchange_format.md +++ b/docs/exchange_format.md @@ -691,6 +691,30 @@ enumerate at most 4096 operations plus one sentinel row, and a local deletion is refused before its effect if the complete protection artifact would exceed either bound. +The V1 `target_kind` set remains the frozen seven-kind set: `session`, +`execution_draft`, `exercise`, `body_observation`, `custom_equipment`, +`feedback`, and `body_zone_relation`. Heart-rate correction does not widen this +published contract. + +## Heart-rate corrections V1 + +`trainlog-heart-rate-corrections` V1 is a separate optional Android→desktop +full-generation companion named `heart-rate-corrections`, stored as +`heart-rate-corrections-v1.json`. Its root contains `format`, integer `version`, +`generated_at`, and at most 256 `corrections`; the complete UTF-8 artifact is +limited to 4 MiB. Each correction contains a stable UUIDv4-based `operation_id`, +stable `capture_id`, inclusive offset-aware `cutoff`, `creator_id`, exact +`predecessor_revision_id`, `created_at`, and canonical `payload_sha256`. + +The predecessor hashes the complete original capture metadata, ordered raw BPM +samples, and ordered RR children. The consumer requires the cutoff to equal the +canonical session end and not precede the session start or latest exercise end. +It deletes only samples strictly after the cutoff in one transaction, relies on +the declared RR cascade, records durable causal state, and applies that state +before any Heart Rate V1 snapshot in the same generation. Unknown fields, +duplicate keys/identities, digest mismatch, ancestry mismatch, and incompatible +capture/session ownership reject the containing generation atomically. + ## Staged generation manifest and acknowledgement V1 `trainlog-sync-manifest` V1 is a strict independent envelope. It contains diff --git a/docs/sync_exchange.md b/docs/sync_exchange.md index 111f31a..81626e1 100644 --- a/docs/sync_exchange.md +++ b/docs/sync_exchange.md @@ -1109,6 +1109,30 @@ advance cardio acknowledgement. Acknowledged captures remain in Android history but are omitted from normal later publication, preventing a full night's samples from being retransmitted on every sync. +An acknowledged session capture may later receive one causal +`heart_rate_tail` deletion when recovery establishes an earlier factual +`session.ended_at`. Its target is the exact `capture_id|cutoff`; the predecessor +digest covers the original capture metadata, ordered BPM samples and ordered RR +children. The cutoff is inclusive: only measurements strictly later than it are +deleted, while `capture.ended_at` becomes the cutoff. Because the general +causal-deletions V1 contract is frozen to its seven existing target kinds, the +operation travels in the separate optional +`trainlog-heart-rate-corrections` V1 / `heart-rate-corrections-v1.json` +companion while reusing the durable causal operation/state ledger. No published +or acknowledged generation is rewritten and no ACK is fabricated. The +correction companion is consumed before Heart Rate V1; that importer also +consults durable correction state before comparing or inserting an old +snapshot, so replay cannot resurrect a corrected tail. + +Producer lineage selection treats only unarchived acknowledged generations +backed by an exact `consumed` / `sqlite-commit-full` ACK as active tips. This +keeps legacy terminal rows that predate the durable ACK ledger as immutable +audit evidence without letting them block every later publication. A consumer +can retain several historical consumed tips because it cannot observe which +fork the producer archived; the next manifest is valid only when its declared +parent is one of those current tips. A null, old, or unrelated parent remains a +hard rejection, and neither side rewrites fork evidence. + The companion changes neither the frozen Bluetooth Classic RFCOMM transport, MTP recovery, manifest/ACK framing, Sleep Diary V1 nor mobile-export semantics. Heart-rate BLE/GATT acquisition remains Android-only and is a separate diff --git a/docs/tests.md b/docs/tests.md index 79cdd57..44ff010 100644 --- a/docs/tests.md +++ b/docs/tests.md @@ -1771,6 +1771,45 @@ generation tool, accepts the correlated ACK, verifies one desktop capture/sample/RR row, and proves the acknowledged Android capture is no longer emitted. +`test_heart_rate_tail_recovery` covers the delayed-session correction boundary: +a sample exactly at the factual end remains, the first representable sample +after it and its RR child are removed, statistics use only retained facts, an +already bounded capture is an idempotent no-op, wrong session/capture and +invalid temporal bounds are refused, and an injected post-effect failure rolls +back every row. It then replays the old Heart Rate V1 snapshot and the same +causal operation twice to prove non-resurrection and second-sync idempotence. +It also proves that `heart_rate_tail` appears only in the separately versioned +heart-rate-correction artifact and remains absent from frozen +causal-deletions V1. +`TrainlogRepositoryDraftTest.delayedFactualFinalizationTruncatesHeartRateSamplesAndRrInclusively` +exercises the production Android delayed-finalization transaction. + +The 2026-09-24 paired-device closeout applies operation +`hrx_a2222757-9a85-492f-823b-ad590208ce6b` to capture +`hrc_fe0627e0-dff6-4629-82d9-03da78c49634`. Android and desktop both report +3,429 BPM samples, 3,429 RR values, no sample after 11:16, last sample +11:15:59.759323, and 89 / 119.30300379119277 / 138 BPM. Full MTP runs +`sy_d1a29dd9-7b51-41db-83e1-5712758b3700` and +`sy_4b8b3c28-17b1-4300-bfbf-9332385f69ab` both completed through +`peer_consumed`. A production import of the retained pre-correction 4,919-point +artifact left the same counts and statistics. The live Web history detail in +Firefox rendered the 10:18--11:16 curve, seven exercises, 15 sets, 3,429 RR, +and the same min/average/max values. + +Generation regressions additionally cover an archived acknowledged fork, a +legacy acknowledged row without durable ACK evidence, and consumer acceptance +of a producer-selected current fork tip while rejecting stale/null ancestry. +Android generation tests exercise the matching rules. Sleep quick-capture and +generation tests cover the schema-v33 medication-intake column order with +quantity 2, preventing note/creation-time values from being bound to the +`quantity` constraint. + +The final post-device closeout passes the complete native Meson inventory +**114/114**, the complete clang ASan/UBSan Meson inventory **114/114**, and +Android `testDebugUnitTest` with **317 tests: 312 passed, five skipped, zero +failures/errors**. Android `assembleDebug`, both JSON/import validators, +changed-file C formatting, and `git diff --check` also pass. + The 2026-09-23 closeout passes the complete native Meson inventory: **102/102 passed, 0 failed**. The complete Android JVM inventory reports **258 tests: 253 passed, five skipped, zero failures/errors**. The five skips diff --git a/tests/test_generation_archival.py b/tests/test_generation_archival.py index 64ffd0e..4058309 100644 --- a/tests/test_generation_archival.py +++ b/tests/test_generation_archival.py @@ -126,7 +126,9 @@ class GenerationArchiveTest(unittest.TestCase): stage, manifest, _ = generation.capture_desktop(self.database, self.owned, PEER) self.assertTrue(stage.is_dir()) - self.assertEqual(parent, manifest["parent_generation_id"]) + # Historical status alone is terminal evidence, not a durable + # consumed ACK and therefore not an active lineage parent. + self.assertIsNone(manifest["parent_generation_id"]) with sqlite3.connect(self.database) as database: self.assertEqual( 0, diff --git a/tests/test_heart_rate_tail_recovery.py b/tests/test_heart_rate_tail_recovery.py new file mode 100644 index 0000000..d95618c --- /dev/null +++ b/tests/test_heart_rate_tail_recovery.py @@ -0,0 +1,217 @@ +#!/usr/bin/env python3 +"""Causal recovery coverage for delayed session-end heart-rate tails.""" + +import json +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "tools")) + +import heart_rate_exchange +import causal_delete_exchange +import heart_rate_correction_exchange +import recover_heart_rate_tail as recovery + + +SESSION = "se_22222222-2222-4222-8222-222222222222" +OTHER_SESSION = "se_99999999-9999-4999-8999-999999999999" +CAPTURE = "hrc_11111111-1111-4111-8111-111111111111" +OTHER_CAPTURE = "hrc_88888888-8888-4888-8888-888888888888" +CUTOFF = "2026-09-24T11:16:00+02:00" + + +def create_schema(db: sqlite3.Connection) -> None: + db.executescript( + """ + PRAGMA foreign_keys=ON; + CREATE TABLE sessions(id INTEGER PRIMARY KEY,session_id TEXT UNIQUE, + started_at TEXT,ended_at TEXT,session_type TEXT,notes TEXT); + CREATE TABLE session_exercises(id INTEGER PRIMARY KEY,session_row_id INTEGER); + CREATE TABLE performed_sets(id INTEGER PRIMARY KEY,session_exercise_row_id INTEGER); + CREATE TABLE continuous_activity(session_exercise_row_id INTEGER PRIMARY KEY); + CREATE TABLE max_results(session_exercise_row_id INTEGER PRIMARY KEY); + CREATE TABLE session_exercise_timeline(session_id TEXT,entry_id TEXT, + exercise_id TEXT,started_at TEXT,ended_at TEXT,imported_at TEXT); + CREATE TABLE heart_rate_captures(capture_id TEXT PRIMARY KEY,context_kind TEXT, + context_id TEXT,started_at TEXT,ended_at TEXT,sensor_name TEXT,imported_at TEXT); + CREATE TABLE heart_rate_samples(capture_id TEXT,sequence INTEGER,observed_at TEXT, + bpm INTEGER,exercise_entry_id TEXT,sensor_contact_detected INTEGER, + energy_expended INTEGER,PRIMARY KEY(capture_id,sequence), + FOREIGN KEY(capture_id) REFERENCES heart_rate_captures(capture_id) ON DELETE CASCADE); + CREATE TABLE heart_rate_rr_intervals(capture_id TEXT,sample_sequence INTEGER, + rr_index INTEGER,value_1024 INTEGER, + PRIMARY KEY(capture_id,sample_sequence,rr_index), + FOREIGN KEY(capture_id,sample_sequence) + REFERENCES heart_rate_samples(capture_id,sequence) ON DELETE CASCADE); + CREATE TABLE sync_causal_operations(operation_id TEXT PRIMARY KEY,target_kind TEXT, + target_id TEXT,creator_id TEXT,predecessor_revision_id TEXT,created_at TEXT, + payload_sha256 TEXT,publication_context TEXT); + CREATE TABLE sync_causal_state(target_kind TEXT,target_id TEXT, + current_revision_id TEXT,deleted INTEGER,operation_id TEXT, + PRIMARY KEY(target_kind,target_id)); + PRAGMA user_version=36; + """ + ) + + +def seed(db: sqlite3.Connection, ended_at: str = CUTOFF, + include_tail: bool = True) -> dict: + db.execute( + "INSERT INTO sessions VALUES(1,?,?,?,?,?)", + (SESSION, "2026-09-24T10:00:00+02:00", ended_at, "training", "keep"), + ) + db.execute("INSERT INTO session_exercises VALUES(1,1)") + db.execute("INSERT INTO performed_sets VALUES(1,1)") + db.execute( + "INSERT INTO session_exercise_timeline VALUES(?,?,?,?,?,?)", + (SESSION, "sxe_33333333-3333-4333-8333-333333333333", + "ex_44444444-4444-4444-8444-444444444444", + "2026-09-24T10:15:00+02:00", "2026-09-24T11:15:00+02:00", CUTOFF), + ) + capture_end = "2026-09-24T11:16:00.000001+02:00" if include_tail else CUTOFF + db.execute( + "INSERT INTO heart_rate_captures VALUES(?,?,?,?,?,?,?)", + (CAPTURE, "session", SESSION, "2026-09-24T10:00:00+02:00", capture_end, + "Synthetic", CUTOFF), + ) + samples = [ + (CAPTURE, 0, "2026-09-24T11:15:59.999999+02:00", 100, None, 1, 1), + (CAPTURE, 1, CUTOFF, 101, None, 1, 2), + ] + if include_tail: + samples.append((CAPTURE, 2, "2026-09-24T11:16:00.000001+02:00", + 40, None, 0, 3)) + db.executemany("INSERT INTO heart_rate_samples VALUES(?,?,?,?,?,?,?)", samples) + db.executemany( + "INSERT INTO heart_rate_rr_intervals VALUES(?,?,0,?)", + [(CAPTURE, row[1], 900 + row[1]) for row in samples], + ) + db.execute( + "INSERT INTO heart_rate_captures VALUES(?,?,?,?,?,?,?)", + (OTHER_CAPTURE, "session", OTHER_SESSION, "2026-09-24T09:00:00+02:00", + "2026-09-24T09:01:00+02:00", "Other", CUTOFF), + ) + db.execute( + "INSERT INTO heart_rate_samples VALUES(?,?,?,?,?,?,?)", + (OTHER_CAPTURE, 0, "2026-09-24T09:00:30+02:00", 77, None, 1, None), + ) + db.commit() + return { + "format": "trainlog-heart-rate", + "version": 1, + "generated_at": CUTOFF, + "captures": [{ + "capture_id": CAPTURE, + "context_kind": "session", + "context_id": SESSION, + "started_at": "2026-09-24T10:00:00+02:00", + "ended_at": capture_end, + "sensor_name": "Synthetic", + "samples": [{ + "sequence": row[1], "observed_at": row[2], "bpm": row[3], + "exercise_entry_id": None, "sensor_contact_detected": bool(row[5]), + "energy_expended": row[6], "rr_intervals_1024": [900 + row[1]], + } for row in samples], + }], + } + + +class HeartRateTailRecoveryTest(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.path = Path(self.temp.name) / "trainlog.sqlite" + self.db = sqlite3.connect(self.path) + create_schema(self.db) + + def tearDown(self): + self.db.close() + self.temp.cleanup() + + def apply(self, operation=None, fail=False): + return recovery.recover( + self.db, SESSION, CAPTURE, CUTOFF, "peer_test", True, + operation=operation, fail_after_effect=fail, + ) + + def test_exact_cutoff_rr_stats_and_old_generation_replay(self): + old_generation = seed(self.db) + report, operation = self.apply() + self.assertEqual("applied", report["outcome"]) + self.assertEqual(1, report["before"]["bpm_remove"]) + self.assertEqual(1, report["before"]["rr_remove"]) + self.assertEqual(0, report["after"]["bpm_remove"]) + self.assertEqual(100.5, report["after"]["current_stats"]["average"]) + self.assertEqual( + [(0, "2026-09-24T11:15:59.999999+02:00"), (1, CUTOFF)], + self.db.execute( + "SELECT sequence,observed_at FROM heart_rate_samples " + "WHERE capture_id=? ORDER BY sequence", (CAPTURE,), + ).fetchall(), + ) + self.assertEqual(2, self.db.execute( + "SELECT COUNT(*) FROM heart_rate_rr_intervals WHERE capture_id=?", (CAPTURE,), + ).fetchone()[0]) + correction_document = heart_rate_correction_exchange.export_document(self.db) + self.assertEqual([operation], correction_document["corrections"]) + self.assertEqual([], causal_delete_exchange.export_document(self.db)["operations"]) + self.assertEqual((0, 1), heart_rate_exchange.apply( + self.db, heart_rate_exchange.validate(old_generation), + )) + second, _ = self.apply(operation) + self.assertEqual("unchanged", second["outcome"]) + self.assertEqual(2, second["after"]["bpm_total"]) + + def test_dry_run_and_injected_failure_roll_back_every_table(self): + seed(self.db) + before = list(self.db.iterdump()) + report, _ = recovery.recover( + self.db, SESSION, CAPTURE, CUTOFF, "peer_test", False, + ) + self.assertEqual("dry-run", report["mode"]) + self.assertEqual(before, list(self.db.iterdump())) + with self.assertRaisesRegex(recovery.RecoveryError, "injected failure"): + self.apply(fail=True) + self.assertEqual(before, list(self.db.iterdump())) + + def test_wrong_id_and_invalid_temporal_bounds_are_rejected(self): + seed(self.db) + with self.assertRaisesRegex(recovery.RecoveryError, "session not found"): + recovery.recover(self.db, OTHER_SESSION, CAPTURE, CUTOFF, "peer", False) + with self.assertRaisesRegex(recovery.RecoveryError, "capture"): + recovery.recover( + self.db, SESSION, + "hrc_77777777-7777-4777-8777-777777777777", CUTOFF, "peer", False, + ) + self.db.execute( + "UPDATE sessions SET ended_at='2026-09-24T09:59:59+02:00' WHERE session_id=?", + (SESSION,), + ) + self.db.commit() + with self.assertRaisesRegex(Exception, "outside capture"): + recovery.recover( + self.db, SESSION, CAPTURE, "2026-09-24T09:59:59+02:00", "peer", False, + ) + self.db.execute("UPDATE sessions SET ended_at=? WHERE session_id=?", (CUTOFF, SESSION)) + self.db.execute( + "UPDATE session_exercise_timeline SET ended_at='2026-09-24T11:16:01+02:00'", + ) + self.db.commit() + with self.assertRaisesRegex(Exception, "precedes the last exercise"): + self.apply() + + def test_already_bounded_capture_is_noop(self): + seed(self.db, include_tail=False) + report, operation = self.apply() + self.assertEqual("unchanged", report["outcome"]) + self.assertIsNone(operation) + self.assertEqual(0, self.db.execute( + "SELECT COUNT(*) FROM sync_causal_operations", + ).fetchone()[0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sync_deployment_tools.py b/tests/test_sync_deployment_tools.py index 31dd04a..9e6edb7 100644 --- a/tests/test_sync_deployment_tools.py +++ b/tests/test_sync_deployment_tools.py @@ -87,6 +87,7 @@ class DeploymentToolsTest(unittest.TestCase): self.assertIn("cardio-calibrations-v1", inventory["protocols"]) self.assertIn("cardio-guidance-v1", inventory["protocols"]) self.assertIn("heart-rate-v1", inventory["protocols"]) + self.assertIn("heart-rate-corrections-v1", inventory["protocols"]) self.assertIn("session-timeline-v1", inventory["protocols"]) self.assertIn("bluetooth-files-v1", inventory["protocols"]) self.assertIn("generation-ack-recovery-v1", inventory["protocols"]) diff --git a/tests/test_sync_generation_exchange.py b/tests/test_sync_generation_exchange.py index 11903e4..10237dd 100644 --- a/tests/test_sync_generation_exchange.py +++ b/tests/test_sync_generation_exchange.py @@ -71,6 +71,105 @@ class GenerationTest(unittest.TestCase): with closing(sqlite3.connect(destination)) as reopened: self.assertEqual(ack,generation.record_consumed(reopened,manifest,checksum));self.assertEqual(1,reopened.execute("SELECT count(*) FROM facts WHERE id='imported'").fetchone()[0]) + def test_archived_acknowledged_fork_is_not_an_active_parent_tip(self): + active = "gen_aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + archived = "gen_bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + with closing(sqlite3.connect(self.database)) as db: + db.execute( + "INSERT OR REPLACE INTO sync_peer_identity VALUES(1,?,'desktop')", + (PEER_A,), + ) + for identity, generated in ( + (active, "2026-09-17T09:00:00+02:00"), + (archived, "2026-09-17T09:01:00+02:00"), + ): + db.execute( + "INSERT INTO sync_generations VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", + ( + identity, RUN, PEER_A, PEER_B, "desktop", generated, None, + "a" * 64, "acknowledged", "{}", str(self.root / identity), + generated, + ), + ) + db.execute( + "INSERT INTO sync_acknowledgements VALUES(?,?,?,?,?,?,?,?,?,?,?)", + ( + "ack_" + identity[4:], identity, RUN, PEER_A, PEER_B, + "a" * 64, "consumed", "sqlite-commit-full", generated, + "", "b" * 64, + ), + ) + db.execute( + "INSERT INTO sync_generation_archives VALUES(?,?,?,?,?,?)", + (archived, str(self.root / "archive"), "a" * 64, "b" * 64, + "2026-09-17T09:02:00+02:00", "{}"), + ) + db.commit() + _, manifest, _ = self.capture() + self.assertEqual(active, manifest["parent_generation_id"]) + + def test_acknowledged_row_without_durable_ack_is_not_an_active_tip(self): + legacy = "gen_aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + current = "gen_bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + with closing(sqlite3.connect(self.database)) as db: + db.execute( + "INSERT OR REPLACE INTO sync_peer_identity VALUES(1,?,'desktop')", + (PEER_A,), + ) + for identity, generated in ( + (legacy, "2026-09-17T09:00:00+02:00"), + (current, "2026-09-17T09:01:00+02:00"), + ): + db.execute( + "INSERT INTO sync_generations VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", + ( + identity, RUN, PEER_A, PEER_B, "desktop", generated, + None, "a" * 64, "acknowledged", "{}", + str(self.root / identity), generated, + ), + ) + db.execute( + "INSERT INTO sync_acknowledgements VALUES(?,?,?,?,?,?,?,?,?,?,?)", + ( + "ack_" + current[4:], current, RUN, PEER_A, PEER_B, + "a" * 64, "consumed", "sqlite-commit-full", + "2026-09-17T09:01:00+02:00", "", "b" * 64, + ), + ) + db.commit() + _, manifest, _ = self.capture() + self.assertEqual(current, manifest["parent_generation_id"]) + + def test_consumer_accepts_producer_selected_current_fork_tip(self): + first = "gen_aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + second = "gen_bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + stale = "gen_cccccccc-cccc-4ccc-8ccc-cccccccccccc" + with closing(sqlite3.connect(self.database)) as db: + for identity, parent in ( + (stale, None), + (first, stale), + (second, stale), + ): + db.execute( + "INSERT INTO sync_consumed_generations VALUES(?,?,?,?,?,?,?,?,?,?,?)", + ( + identity, RUN, PEER_B, PEER_A, parent, "a" * 64, + "2026-09-17T09:00:00+02:00", "consumed", + "sqlite-commit-full", "", "{}", + ), + ) + generation.require_current_consumed_parent(db, PEER_B, first) + generation.require_current_consumed_parent(db, PEER_B, second) + with self.assertRaisesRegex( + generation.GenerationError, "stale or unrelated" + ): + generation.require_current_consumed_parent(db, PEER_B, stale) + with self.assertRaisesRegex( + generation.GenerationError, "stale or unrelated" + ): + generation.require_current_consumed_parent(db, PEER_B, None) + generation.require_current_consumed_parent(db, PEER_A, None) + def test_sleep_revision_becomes_acknowledged_only_after_correlated_ack(self): stage, manifest, checksum = self.capture() self.publish() diff --git a/tools/causal_delete_exchange.py b/tools/causal_delete_exchange.py index 1371a34..33fb85f 100644 --- a/tools/causal_delete_exchange.py +++ b/tools/causal_delete_exchange.py @@ -243,7 +243,12 @@ def load(path: Path) -> dict: def export_document(db: sqlite3.Connection) -> dict: keys = ("operation_id", "target_kind", "target_id", "creator_id", "predecessor_revision_id", "created_at", "payload_sha256", "publication_context") - rows = db.execute("SELECT " + ",".join(keys) + " FROM sync_causal_operations ORDER BY operation_id LIMIT ?", (MAX_OPERATIONS + 1,)).fetchall() + rows = db.execute( + "SELECT " + ",".join(keys) + " FROM sync_causal_operations " + "WHERE target_kind IN('session','execution_draft','exercise','body_observation'," + "'custom_equipment','feedback','body_zone_relation') " + "ORDER BY operation_id LIMIT ?", (MAX_OPERATIONS + 1,), + ).fetchall() if len(rows) > MAX_OPERATIONS: raise CausalError("protection set exceeds artifact bound") document = {"format": FORMAT, "version": VERSION, "generated_at": dt.datetime.now().astimezone().isoformat(), diff --git a/tools/heart_rate_correction_exchange.py b/tools/heart_rate_correction_exchange.py new file mode 100644 index 0000000..9a0127c --- /dev/null +++ b/tools/heart_rate_correction_exchange.py @@ -0,0 +1,338 @@ +#!/usr/bin/env python3 +"""Versioned causal corrections for acknowledged Heart Rate V1 captures.""" + +from __future__ import annotations + +import datetime as dt +import hashlib +import json +import re +import sqlite3 +import uuid +from pathlib import Path + +from validate_json import parse_timestamp + + +FORMAT = "trainlog-heart-rate-corrections" +VERSION = 1 +TARGET_KIND = "heart_rate_tail" +MAX_BYTES = 4 * 1024 * 1024 +MAX_CORRECTIONS = 256 +UUID4 = r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}" +CAPTURE_ID = re.compile(rf"^hrc_{UUID4}$") +OPERATION_ID = re.compile(rf"^hrx_{UUID4}$") +HEX = re.compile(r"^[0-9a-f]{64}$") + + +class CorrectionError(ValueError): + """A correction is malformed, conflicts, or targets incompatible facts.""" + + +def canonical(value: object) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def digest(value: object) -> str: + return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest() + + +def exact_timestamp(value: str): + try: + return parse_timestamp(value, "heart_rate_timestamp") + except (TypeError, ValueError) as error: + raise CorrectionError("invalid heart-rate timestamp") from error + + +def target_id(capture_id: str, cutoff: str) -> str: + if not isinstance(capture_id, str) or not CAPTURE_ID.fullmatch(capture_id): + raise CorrectionError("invalid heart-rate capture identity") + exact_timestamp(cutoff) + return f"{capture_id}|{cutoff}" + + +def split_target(value: str) -> tuple[str, str]: + parts = value.split("|", 1) + if len(parts) != 2: + raise CorrectionError("invalid heart-rate correction target") + target_id(parts[0], parts[1]) + return parts[0], parts[1] + + +def require_schema(db: sqlite3.Connection) -> None: + version = db.execute("PRAGMA user_version").fetchone()[0] + if version not in range(27, 37): + raise CorrectionError(f"heart-rate correction requires schema v27-v36, found v{version}") + + +def _table_exists(db: sqlite3.Connection, table: str) -> bool: + return db.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,), + ).fetchone() is not None + + +def capture_snapshot(db: sqlite3.Connection, capture_id: str): + row = db.execute( + "SELECT capture_id,context_kind,context_id,started_at,ended_at,sensor_name " + "FROM heart_rate_captures WHERE capture_id=?", (capture_id,), + ).fetchone() + if row is None: + return None + samples = [] + for sample in db.execute( + "SELECT sequence,observed_at,bpm,exercise_entry_id,sensor_contact_detected," + "energy_expended FROM heart_rate_samples WHERE capture_id=? ORDER BY sequence", + (capture_id,), + ): + rr = [list(value) for value in db.execute( + "SELECT rr_index,value_1024 FROM heart_rate_rr_intervals " + "WHERE capture_id=? AND sample_sequence=? ORDER BY rr_index", + (capture_id, sample[0]), + )] + samples.append([*sample, rr]) + return [*row, samples] + + +def live_revision(db: sqlite3.Connection, capture_id: str) -> str | None: + snapshot = capture_snapshot(db, capture_id) + return None if snapshot is None else "lv_" + digest(snapshot) + + +def authorize(db: sqlite3.Connection, capture_id: str, cutoff: str) -> None: + row = db.execute( + "SELECT context_kind,context_id,started_at,ended_at FROM heart_rate_captures " + "WHERE capture_id=?", (capture_id,), + ).fetchone() + if row is None: + raise CorrectionError("heart-rate capture not found") + if row[0] not in ("session", "cardio"): + raise CorrectionError("only session heart-rate captures can be corrected") + if row[3] is None: + raise CorrectionError("active heart-rate capture cannot be corrected") + cutoff_key = exact_timestamp(cutoff) + if cutoff_key < exact_timestamp(row[2]) or cutoff_key > exact_timestamp(row[3]): + raise CorrectionError("heart-rate cutoff is outside capture") + session = db.execute( + "SELECT ended_at FROM sessions WHERE session_id=?", (row[1],), + ).fetchone() + if session is None or session[0] != cutoff: + raise CorrectionError("heart-rate cutoff must equal the factual session end") + timeline = ("session_timeline_exercises" + if _table_exists(db, "session_timeline_exercises") + else "session_exercise_timeline") + latest = db.execute( + f"SELECT MAX(ended_at) FROM {timeline} WHERE session_id=?", (row[1],), + ).fetchone()[0] + if latest is not None and exact_timestamp(latest) > cutoff_key: + raise CorrectionError("heart-rate cutoff precedes the last exercise marker") + previous = None + for expected, (sequence, observed_at) in enumerate(db.execute( + "SELECT sequence,observed_at FROM heart_rate_samples " + "WHERE capture_id=? ORDER BY sequence", (capture_id,), + )): + observed = exact_timestamp(observed_at) + if sequence != expected or previous is not None and observed < previous: + raise CorrectionError("heart-rate samples are not one chronological sequence") + previous = observed + + +def correction_payload(correction: dict) -> dict: + return {key: correction[key] for key in ( + "operation_id", "capture_id", "cutoff", "creator_id", + "predecessor_revision_id", "created_at", + )} + + +def validate_correction(value: object) -> dict: + keys = {"operation_id", "capture_id", "cutoff", "creator_id", + "predecessor_revision_id", "created_at", "payload_sha256"} + if not isinstance(value, dict) or set(value) != keys: + raise CorrectionError("invalid heart-rate correction shape") + if not isinstance(value["operation_id"], str) or not OPERATION_ID.fullmatch(value["operation_id"]): + raise CorrectionError("invalid heart-rate correction identity") + target_id(value["capture_id"], value["cutoff"]) + for field in ("creator_id", "predecessor_revision_id"): + if not isinstance(value[field], str) or not 0 < len(value[field].encode()) <= 128: + raise CorrectionError(f"invalid {field}") + exact_timestamp(value["created_at"]) + if not isinstance(value["payload_sha256"], str) or not HEX.fullmatch(value["payload_sha256"]): + raise CorrectionError("invalid heart-rate correction digest") + if digest(correction_payload(value)) != value["payload_sha256"]: + raise CorrectionError("heart-rate correction digest mismatch") + return value + + +def apply_effect(db: sqlite3.Connection, capture_id: str, cutoff: str) -> None: + # WHY: a stopped capture may cross an immutable generation before recovery + # establishes the factual session end. + # CONTRACT: the cutoff is inclusive; only its strict BPM/RR tail is removed. + # INVARIANT: acknowledged generation bytes and retained identities never change. + cutoff_key = exact_timestamp(cutoff) + removed = [sequence for sequence, observed_at in db.execute( + "SELECT sequence,observed_at FROM heart_rate_samples " + "WHERE capture_id=? ORDER BY sequence", (capture_id,), + ) if exact_timestamp(observed_at) > cutoff_key] + db.executemany( + "DELETE FROM heart_rate_samples WHERE capture_id=? AND sequence=?", + [(capture_id, sequence) for sequence in removed], + ) + if db.execute( + "UPDATE heart_rate_captures SET ended_at=? WHERE capture_id=?", + (cutoff, capture_id), + ).rowcount != 1: + raise CorrectionError("heart-rate capture disappeared during correction") + + +def apply_correction(db: sqlite3.Connection, value: dict) -> str: + correction = validate_correction(value) + operation_id = correction["operation_id"] + target = target_id(correction["capture_id"], correction["cutoff"]) + known = db.execute( + "SELECT target_kind,target_id,creator_id,predecessor_revision_id,created_at," + "payload_sha256,publication_context FROM sync_causal_operations " + "WHERE operation_id=?", (operation_id,), + ).fetchone() + expected = (TARGET_KIND, target, correction["creator_id"], + correction["predecessor_revision_id"], correction["created_at"], + correction["payload_sha256"], None) + if known is not None: + if tuple(known) != expected: + raise CorrectionError("heart-rate correction identity reused") + return "unchanged" + authorize(db, correction["capture_id"], correction["cutoff"]) + state = db.execute( + "SELECT current_revision_id,deleted FROM sync_causal_state " + "WHERE target_kind=? AND target_id=?", (TARGET_KIND, target), + ).fetchone() + current = state[0] if state else live_revision(db, correction["capture_id"]) + if state and state[1]: + raise CorrectionError("concurrent heart-rate correction conflict") + if current is None or current != correction["predecessor_revision_id"]: + raise CorrectionError("heart-rate correction predecessor conflict") + count = db.execute( + "SELECT COUNT(*) FROM sync_causal_operations WHERE target_kind=?", (TARGET_KIND,), + ).fetchone()[0] + if count >= MAX_CORRECTIONS: + raise CorrectionError("heart-rate correction bound exceeded") + db.execute( + "INSERT INTO sync_causal_operations VALUES(?,?,?,?,?,?,?,NULL)", + (operation_id, TARGET_KIND, target, correction["creator_id"], current, + correction["created_at"], correction["payload_sha256"]), + ) + apply_effect(db, correction["capture_id"], correction["cutoff"]) + db.execute( + "INSERT OR REPLACE INTO sync_causal_state VALUES(?,?,?,?,?)", + (TARGET_KIND, target, operation_id, 1, operation_id), + ) + if len(canonical(export_document(db)).encode("utf-8")) > MAX_BYTES: + raise CorrectionError("heart-rate correction artifact exceeds byte bound") + return "applied" + + +def local_correction(db: sqlite3.Connection, capture_id: str, cutoff: str, + creator_id: str) -> tuple[dict, str]: + target = target_id(capture_id, cutoff) + state = db.execute( + "SELECT current_revision_id,deleted FROM sync_causal_state " + "WHERE target_kind=? AND target_id=?", (TARGET_KIND, target), + ).fetchone() + if state and state[1]: + row = db.execute( + "SELECT operation_id,creator_id,predecessor_revision_id,created_at,payload_sha256 " + "FROM sync_causal_operations WHERE operation_id=?", (state[0],), + ).fetchone() + correction = { + "operation_id": row[0], "capture_id": capture_id, "cutoff": cutoff, + "creator_id": row[1], "predecessor_revision_id": row[2], + "created_at": row[3], "payload_sha256": row[4], + } + return correction, "unchanged" + authorize(db, capture_id, cutoff) + predecessor = state[0] if state else live_revision(db, capture_id) + if predecessor is None: + raise CorrectionError("heart-rate capture not found") + correction = { + "operation_id": "hrx_" + str(uuid.uuid4()), + "capture_id": capture_id, + "cutoff": cutoff, + "creator_id": creator_id, + "predecessor_revision_id": predecessor, + "created_at": dt.datetime.now().astimezone().isoformat(), + } + correction["payload_sha256"] = digest(correction_payload(correction)) + return correction, apply_correction(db, correction) + + +def export_document(db: sqlite3.Connection) -> dict: + corrections = [] + for row in db.execute( + "SELECT operation_id,target_id,creator_id,predecessor_revision_id,created_at," + "payload_sha256 FROM sync_causal_operations WHERE target_kind=? " + "ORDER BY operation_id LIMIT ?", (TARGET_KIND, MAX_CORRECTIONS + 1), + ): + capture_id, cutoff = split_target(row[1]) + corrections.append({ + "operation_id": row[0], "capture_id": capture_id, "cutoff": cutoff, + "creator_id": row[2], "predecessor_revision_id": row[3], + "created_at": row[4], "payload_sha256": row[5], + }) + if len(corrections) > MAX_CORRECTIONS: + raise CorrectionError("heart-rate correction bound exceeded") + result = { + "format": FORMAT, + "version": VERSION, + "generated_at": dt.datetime.now().astimezone().isoformat(), + "corrections": corrections, + } + if len(canonical(result).encode("utf-8")) > MAX_BYTES: + raise CorrectionError("heart-rate correction artifact exceeds byte bound") + return result + + +def validate_document(value: object) -> dict: + if not isinstance(value, dict) or set(value) != { + "format", "version", "generated_at", "corrections"}: + raise CorrectionError("invalid heart-rate correction envelope") + if value["format"] != FORMAT or type(value["version"]) is not int or value["version"] != VERSION: + raise CorrectionError("unsupported heart-rate correction artifact") + exact_timestamp(value["generated_at"]) + if not isinstance(value["corrections"], list) or len(value["corrections"]) > MAX_CORRECTIONS: + raise CorrectionError("heart-rate correction bound exceeded") + identities = set() + for correction in value["corrections"]: + validate_correction(correction) + if correction["operation_id"] in identities: + raise CorrectionError("duplicate heart-rate correction identity") + identities.add(correction["operation_id"]) + return value + + +def load(path: Path) -> dict: + raw = path.read_bytes() + if len(raw) > MAX_BYTES: + raise CorrectionError("heart-rate correction artifact exceeds byte bound") + try: + value = json.loads(raw.decode("utf-8"), object_pairs_hook=_unique_object) + except (UnicodeError, json.JSONDecodeError) as error: + raise CorrectionError("invalid heart-rate correction JSON") from error + return validate_document(value) + + +def _unique_object(pairs): + result = {} + for key, value in pairs: + if key in result: + raise CorrectionError("duplicate heart-rate correction JSON field") + result[key] = value + return result + + +def apply_document(db: sqlite3.Connection, value: dict) -> tuple[int, int]: + validate_document(value) + applied = 0 + unchanged = 0 + for correction in value["corrections"]: + outcome = apply_correction(db, correction) + applied += outcome == "applied" + unchanged += outcome == "unchanged" + return applied, unchanged diff --git a/tools/heart_rate_exchange.py b/tools/heart_rate_exchange.py index 525f6aa..41735f9 100644 --- a/tools/heart_rate_exchange.py +++ b/tools/heart_rate_exchange.py @@ -9,6 +9,7 @@ from datetime import datetime from pathlib import Path from trainlog_sqlite import connect_database +from validate_json import parse_timestamp as exact_timestamp FORMAT = "trainlog-heart-rate" @@ -134,6 +135,45 @@ def load(path: Path) -> dict: fail("heart-rate artifact exceeds 64 MiB") return validate(json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=unique_object)) + +def _table_exists(db: sqlite3.Connection, table: str) -> bool: + return db.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,), + ).fetchone() is not None + + +def _corrected_capture(db: sqlite3.Connection, capture: dict) -> dict: + if not _table_exists(db, "sync_causal_state"): + return capture + prefix = capture["capture_id"] + "|" + cutoffs = [] + for target_id, in db.execute( + "SELECT target_id FROM sync_causal_state " + "WHERE target_kind='heart_rate_tail' AND deleted=1 AND target_id LIKE ?", + (prefix + "%",), + ): + if not target_id.startswith(prefix): + continue + raw = target_id[len(prefix):] + timestamp(raw) + value, parsed = raw, exact_timestamp(raw, "heart_rate_cutoff") + cutoffs.append((parsed, value)) + if not cutoffs: + return capture + cutoff_key, cutoff = min(cutoffs) + started = exact_timestamp(capture["started_at"], "started_at") + ended = exact_timestamp(capture["ended_at"], "ended_at") + if cutoff_key < started or cutoff_key > ended: + fail("causal heart-rate cutoff is outside capture") + samples = [sample for sample in capture["samples"] + if exact_timestamp(sample["observed_at"], "observed_at") <= cutoff_key] + if any(sample["sequence"] != expected for expected, sample in enumerate(samples)): + fail("causal heart-rate truncation would break sample sequence") + corrected = dict(capture) + corrected["ended_at"] = cutoff + corrected["samples"] = samples + return corrected + def _same_capture(db: sqlite3.Connection, capture: dict) -> bool: row = db.execute( "SELECT context_kind,context_id,started_at,ended_at,sensor_name " @@ -183,7 +223,13 @@ def _same_capture(db: sqlite3.Connection, capture: dict) -> bool: def apply(db: sqlite3.Connection, root: dict) -> tuple[int, int]: applied = 0 unchanged = 0 - for capture in root["captures"]: + for raw_capture in root["captures"]: + # WHY: acknowledged Heart Rate V1 generations are immutable, while a + # later causal correction may remove a sensor tail from history. + # CONTRACT: replay compares/inserts the old snapshot only after every + # durable inclusive cutoff for this capture has been applied. + # INVARIANT: a corrected tail cannot be resurrected by old bytes. + capture = _corrected_capture(db, raw_capture) if _same_capture(db, capture): unchanged += 1 continue diff --git a/tools/package_sync_candidate.py b/tools/package_sync_candidate.py index 7285bba..56fb5ff 100644 --- a/tools/package_sync_candidate.py +++ b/tools/package_sync_candidate.py @@ -19,9 +19,10 @@ RUNTIME_TOOLS = [ "sync_generation_exchange.py", "sleep_diary_exchange.py", "cardio_session_exchange.py", - "cardio_calibration_exchange.py", - "cardio_guidance_exchange.py", - "heart_rate_exchange.py", + "cardio_calibration_exchange.py", + "cardio_guidance_exchange.py", + "heart_rate_exchange.py", + "heart_rate_correction_exchange.py", "session_timeline_exchange.py", "trainlog_sqlite.py", "validate_json.py", @@ -140,6 +141,7 @@ def main() -> int: "cardio-calibrations-v1", "cardio-guidance-v1", "heart-rate-v1", + "heart-rate-corrections-v1", "session-timeline-v1", ], "entry_points": [ diff --git a/tools/recover_heart_rate_tail.py b/tools/recover_heart_rate_tail.py new file mode 100644 index 0000000..7c5c51d --- /dev/null +++ b/tools/recover_heart_rate_tail.py @@ -0,0 +1,308 @@ +#!/usr/bin/env python3 +"""Transactionally truncate one acknowledged session heart-rate capture.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import sqlite3 +from pathlib import Path + +import heart_rate_correction_exchange as correction_exchange +from trainlog_sqlite import connect_database + + +class RecoveryError(ValueError): + """The requested recovery does not match the persisted factual session.""" + + +def _digest(value: object) -> str: + raw = json.dumps(value, ensure_ascii=False, sort_keys=True, + separators=(",", ":"), default=str).encode("utf-8") + return hashlib.sha256(raw).hexdigest() + + +def _rows(db: sqlite3.Connection, query: str, parameters: tuple = ()) -> list[list]: + return [list(row) for row in db.execute(query, parameters)] + + +def protected_snapshot(db: sqlite3.Connection, session_id: str, + capture_id: str) -> dict: + session = db.execute( + "SELECT id FROM sessions WHERE session_id=?", (session_id,), + ).fetchone() + if session is None: + raise RecoveryError("session not found") + session_row = session[0] + occurrence_rows = [row[0] for row in db.execute( + "SELECT id FROM session_exercises WHERE session_row_id=? ORDER BY id", + (session_row,), + )] + occurrence_marks = ",".join("?" for _ in occurrence_rows) or "NULL" + protected = { + "session": _rows(db, "SELECT * FROM sessions WHERE id=?", (session_row,)), + "occurrences": _rows( + db, "SELECT * FROM session_exercises WHERE session_row_id=? ORDER BY id", + (session_row,), + ), + "sets": _rows( + db, f"SELECT * FROM performed_sets WHERE session_exercise_row_id IN " + f"({occurrence_marks}) ORDER BY id", tuple(occurrence_rows), + ), + "continuous": _rows( + db, f"SELECT * FROM continuous_activity WHERE session_exercise_row_id IN " + f"({occurrence_marks}) ORDER BY session_exercise_row_id", tuple(occurrence_rows), + ), + "maxima": _rows( + db, f"SELECT * FROM max_results WHERE session_exercise_row_id IN " + f"({occurrence_marks}) ORDER BY session_exercise_row_id", tuple(occurrence_rows), + ), + "other_captures": _rows( + db, "SELECT capture_id,context_kind,context_id,started_at,ended_at,sensor_name " + "FROM heart_rate_captures WHERE capture_id<>? ORDER BY capture_id", (capture_id,), + ), + "other_sample_counts": _rows( + db, "SELECT capture_id,COUNT(*),MIN(observed_at),MAX(observed_at) " + "FROM heart_rate_samples WHERE capture_id<>? GROUP BY capture_id " + "ORDER BY capture_id", (capture_id,), + ), + "other_samples": _rows( + db, "SELECT * FROM heart_rate_samples WHERE capture_id<>? " + "ORDER BY capture_id,sequence", (capture_id,), + ), + "other_rr": _rows( + db, "SELECT * FROM heart_rate_rr_intervals WHERE capture_id<>? " + "ORDER BY capture_id,sample_sequence,rr_index", (capture_id,), + ), + } + if db.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='exercise_feedback'", + ).fetchone(): + feedback_ids = [row[0] for row in db.execute( + f"SELECT feedback_id FROM exercise_feedback WHERE session_exercise_row_id IN " + f"({occurrence_marks}) ORDER BY feedback_id", tuple(occurrence_rows), + )] + feedback_marks = ",".join("?" for _ in feedback_ids) or "NULL" + protected["feedback"] = _rows( + db, f"SELECT * FROM exercise_feedback WHERE feedback_id IN ({feedback_marks}) " + "ORDER BY feedback_id", tuple(feedback_ids), + ) + protected["feedback_revisions"] = _rows( + db, f"SELECT * FROM exercise_feedback_revisions WHERE feedback_id IN " + f"({feedback_marks}) ORDER BY feedback_id,revision_id", tuple(feedback_ids), + ) + if db.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='session_followups'", + ).fetchone(): + followup_ids = [row[0] for row in db.execute( + "SELECT followup_id FROM session_followups WHERE session_row_id=? " + "ORDER BY followup_id", (session_row,), + )] + followup_marks = ",".join("?" for _ in followup_ids) or "NULL" + protected["session_followups"] = _rows( + db, f"SELECT * FROM session_followups WHERE followup_id IN " + f"({followup_marks}) ORDER BY followup_id", tuple(followup_ids), + ) + protected["session_followup_revisions"] = _rows( + db, f"SELECT * FROM session_followup_revisions WHERE followup_id IN " + f"({followup_marks}) ORDER BY followup_id,revision_id", tuple(followup_ids), + ) + if db.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' " + "AND name='session_timeline_sessions'", + ).fetchone(): + protected["timeline_session"] = _rows( + db, "SELECT * FROM session_timeline_sessions WHERE session_id=?", (session_id,), + ) + protected["timeline_exercises"] = _rows( + db, "SELECT * FROM session_timeline_exercises WHERE session_id=? " + "ORDER BY entry_id", (session_id,), + ) + else: + protected["timeline_exercises"] = _rows( + db, "SELECT * FROM session_exercise_timeline WHERE session_id=? " + "ORDER BY entry_id", (session_id,), + ) + for table in ("sleep_diary_entries", "sleep_diary_revisions", "sleep_diary_events", + "sleep_medications", "sleep_medication_intakes"): + if db.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,), + ).fetchone(): + protected[table] = _rows(db, f"SELECT * FROM {table} ORDER BY rowid") + return protected + + +def capture_report(db: sqlite3.Connection, capture_id: str, cutoff: str) -> dict: + cutoff_key = correction_exchange.exact_timestamp(cutoff) + capture = db.execute( + "SELECT context_kind,context_id,started_at,ended_at,sensor_name " + "FROM heart_rate_captures WHERE capture_id=?", (capture_id,), + ).fetchone() + if capture is None: + raise RecoveryError("capture not found") + samples = db.execute( + "SELECT sequence,observed_at,bpm,exercise_entry_id FROM heart_rate_samples " + "WHERE capture_id=? ORDER BY sequence", (capture_id,), + ).fetchall() + keep = [row for row in samples + if correction_exchange.exact_timestamp(row[1]) <= cutoff_key] + remove = [row for row in samples + if correction_exchange.exact_timestamp(row[1]) > cutoff_key] + rr = db.execute( + "SELECT s.observed_at FROM heart_rate_rr_intervals r " + "JOIN heart_rate_samples s ON s.capture_id=r.capture_id " + "AND s.sequence=r.sample_sequence WHERE r.capture_id=?", + (capture_id,), + ).fetchall() + rr_keep = sum(correction_exchange.exact_timestamp(row[0]) <= cutoff_key for row in rr) + values = [row[2] for row in samples] + kept_values = [row[2] for row in keep] + + def stats(items: list[int]) -> dict: + return { + "minimum": min(items) if items else None, + "average": sum(items) / len(items) if items else None, + "maximum": max(items) if items else None, + } + + return { + "capture_id": capture_id, + "context_kind": capture[0], + "context_id": capture[1], + "started_at": capture[2], + "ended_at": capture[3], + "sensor_name": capture[4], + "bpm_total": len(samples), + "bpm_keep": len(keep), + "bpm_remove": len(remove), + "rr_total": len(rr), + "rr_keep": rr_keep, + "rr_remove": len(rr) - rr_keep, + "first_removed_at": remove[0][1] if remove else None, + "last_removed_at": remove[-1][1] if remove else None, + "removed_exercise_entry_ids": sorted({row[3] for row in remove}, + key=lambda value: value or ""), + "current_stats": stats(values), + "kept_stats": stats(kept_values), + } + + +def load_operation(path: Path) -> dict: + value = json.loads(path.read_text(encoding="utf-8")) + if isinstance(value, dict) and set(value) == {"format", "version", "generated_at", + "corrections"}: + correction_exchange.load(path) + if len(value["corrections"]) != 1: + raise RecoveryError("correction artifact must contain exactly one operation") + return value["corrections"][0] + return correction_exchange.validate_correction(value) + + +def recover(db: sqlite3.Connection, session_id: str, capture_id: str, cutoff: str, + creator_id: str, apply: bool, operation: dict | None = None, + fail_after_effect: bool = False) -> tuple[dict, dict | None]: + """Apply or simulate one exact correction without touching sibling facts. + + CONTRACT: an emitted operation is the immutable content of the separate + heart-rate-corrections V1 companion. INVARIANT: every mutation commits + together or rolls back. + """ + db.execute("PRAGMA foreign_keys=ON") + correction_exchange.require_schema(db) + db.execute("BEGIN IMMEDIATE") + emitted = operation + try: + session = db.execute( + "SELECT started_at,ended_at FROM sessions WHERE session_id=?", (session_id,), + ).fetchone() + if session is None: + raise RecoveryError("session not found") + if session[1] != cutoff: + raise RecoveryError("cutoff does not equal the factual session end") + capture = db.execute( + "SELECT context_id FROM heart_rate_captures WHERE capture_id=?", (capture_id,), + ).fetchone() + if capture is None or capture[0] != session_id: + raise RecoveryError("capture does not belong to the requested session") + before = capture_report(db, capture_id, cutoff) + protected_before = protected_snapshot(db, session_id, capture_id) + if operation is not None: + if operation.get("capture_id") != capture_id or operation.get("cutoff") != cutoff: + raise RecoveryError("operation does not match requested capture and cutoff") + outcome = correction_exchange.apply_correction(db, operation) + elif before["bpm_remove"] == 0 and before["ended_at"] == cutoff: + outcome = "unchanged" + emitted = None + else: + emitted, outcome = correction_exchange.local_correction( + db, capture_id, cutoff, creator_id, + ) + if fail_after_effect: + raise RecoveryError("injected failure after correction effect") + after = capture_report(db, capture_id, cutoff) + protected_after = protected_snapshot(db, session_id, capture_id) + if protected_before != protected_after: + raise RecoveryError("recovery modified protected session or unrelated data") + if after["ended_at"] != cutoff or after["bpm_remove"] != 0 or after["rr_remove"] != 0: + raise RecoveryError("post-cutoff heart-rate data remain after correction") + if db.execute("PRAGMA foreign_key_check").fetchone() is not None: + raise RecoveryError("foreign key check failed after correction") + if apply: + db.commit() + else: + db.rollback() + return { + "mode": "apply" if apply else "dry-run", + "outcome": outcome, + "session_id": session_id, + "cutoff": cutoff, + "before": before, + "after": after, + "protected_sha256": _digest(protected_before), + "protected_unchanged": True, + }, emitted + except Exception: + db.rollback() + raise + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--database", required=True, type=Path) + parser.add_argument("--session-id", required=True) + parser.add_argument("--capture-id", required=True) + parser.add_argument("--cutoff", required=True) + parser.add_argument("--creator-id", default="peer_android_recovery") + parser.add_argument("--operation-input", type=Path) + parser.add_argument("--operation-output", type=Path) + parser.add_argument("--report-output", type=Path) + parser.add_argument("--apply", action="store_true") + args = parser.parse_args() + operation = load_operation(args.operation_input) if args.operation_input else None + db = connect_database(args.database) + try: + db.execute("PRAGMA foreign_keys=ON") + report, emitted = recover( + db, args.session_id, args.capture_id, args.cutoff, args.creator_id, + args.apply, operation, + ) + finally: + db.close() + if args.operation_output and emitted is not None: + args.operation_output.write_text( + json.dumps(emitted, ensure_ascii=False, sort_keys=True, + separators=(",", ":")), encoding="utf-8", + ) + rendered = json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + if args.report_output: + args.report_output.write_text(rendered + "\n", encoding="utf-8") + print(rendered) + + +if __name__ == "__main__": + try: + main() + except Exception as error: + print(f"HEART_RATE_RECOVERY=FAIL {error}") + raise SystemExit(1) diff --git a/tools/sync_generation_exchange.py b/tools/sync_generation_exchange.py index 9343491..4144a6c 100644 --- a/tools/sync_generation_exchange.py +++ b/tools/sync_generation_exchange.py @@ -25,6 +25,7 @@ import cardio_calibration_exchange import cardio_guidance_exchange import execution_draft_exchange import heart_rate_exchange +import heart_rate_correction_exchange import session_timeline_exchange import import_equipment_associations import import_equipment_definitions @@ -90,6 +91,7 @@ SUPPORTED = {(row[1], row[2]) for row in ARTIFACTS} | { ("trainlog-cardio-calibrations", 1), ("trainlog-cardio-guidance", 1), ("trainlog-heart-rate", 1), + ("trainlog-heart-rate-corrections", 1), ("trainlog-session-timeline", 1), } @@ -417,7 +419,29 @@ def capture_desktop(database: Path, owned_root: Path, consumer: str, "size": size, "sha256": checksum, "required": required}) parent = None with closing(connect_database(database)) as db: - parent_rows = db.execute("SELECT g.generation_id FROM sync_generations g WHERE g.producer_peer_id=? AND g.consumer_peer_id=? AND g.status='acknowledged' AND NOT EXISTS(SELECT 1 FROM sync_generations c WHERE c.parent_generation_id=g.generation_id AND c.status='acknowledged') LIMIT 2", (producer, consumer)).fetchall() + # WHY: archived generations retain immutable audit rows, including + # forks, but have left the bounded active publication window. + # CONTRACT: only unarchived acknowledged rows with an exact durable + # consumed ACK participate in active parent selection; multiple + # active tips remain a hard failure. + # INVARIANT: archival evidence is never rewritten or deleted. + parent_rows = db.execute( + "SELECT g.generation_id FROM sync_generations g WHERE " + "g.producer_peer_id=? AND g.consumer_peer_id=? AND " + "g.status='acknowledged' AND EXISTS(SELECT 1 FROM " + "sync_acknowledgements k WHERE k.generation_id=g.generation_id " + "AND k.result='consumed' AND k.durability='sqlite-commit-full') " + "AND NOT EXISTS(SELECT 1 FROM " + "sync_generation_archives a WHERE a.generation_id=g.generation_id) " + "AND NOT EXISTS(SELECT 1 FROM sync_generations c WHERE " + "c.parent_generation_id=g.generation_id AND c.status='acknowledged' " + "AND EXISTS(SELECT 1 FROM sync_acknowledgements ck WHERE " + "ck.generation_id=c.generation_id AND ck.result='consumed' AND " + "ck.durability='sqlite-commit-full') " + "AND NOT EXISTS(SELECT 1 FROM sync_generation_archives ca WHERE " + "ca.generation_id=c.generation_id)) LIMIT 2", + (producer, consumer), + ).fetchall() if len(parent_rows) > 1: raise GenerationError("outgoing generation lineage has multiple tips") parent = parent_rows[0][0] if parent_rows else None manifest = {"format": MANIFEST, "version": 1, "generation_id": generation, @@ -630,6 +654,36 @@ def record_consumed(db: sqlite3.Connection, manifest: dict, manifest_digest: str return ack +def require_current_consumed_parent( + db: sqlite3.Connection, producer_peer_id: str, declared_parent: str | None +) -> None: + """Require the producer to extend one locally observed consumed tip.""" + # WHY: a producer can archive one branch after a transient fork, while the + # consumer retains both immutable consumed rows and cannot observe that + # producer-local archive decision. + # CONTRACT: a non-empty lineage must extend one of its current consumed + # tips; an empty lineage requires a null parent. An old or unrelated parent + # remains a hard failure. + # INVARIANT: this check never rewrites or discards either fork's audit row. + tip_predicate = ( + "g.producer_peer_id=? AND g.result='consumed' AND NOT EXISTS(SELECT 1 " + "FROM sync_consumed_generations c WHERE c.parent_generation_id=" + "g.generation_id AND c.result='consumed')" + ) + has_tip = db.execute( + "SELECT 1 FROM sync_consumed_generations g WHERE " + tip_predicate + " LIMIT 1", + (producer_peer_id,), + ).fetchone() is not None + parent_is_tip = declared_parent is not None and db.execute( + "SELECT 1 FROM sync_consumed_generations g WHERE g.generation_id=? AND " + + tip_predicate + + " LIMIT 1", + (declared_parent, producer_peer_id), + ).fetchone() is not None + if (not has_tip and declared_parent is not None) or (has_tip and not parent_is_tip): + raise GenerationError("generation lineage is stale or unrelated") + + def record_rejected(database: Path, manifest: dict, manifest_digest: str, diagnostic: str) -> dict: # Android's platform JSON quoter escapes solidus characters while the # desktop canonical encoder does not. Rejection diagnostics are bounded @@ -685,6 +739,10 @@ def consume_desktop(database: Path, directory: Path) -> dict: if "sleep-diary" in listed else None) heart_rate = (heart_rate_exchange.load(listed["heart-rate"]) if "heart-rate" in listed else None) + heart_rate_corrections = ( + heart_rate_correction_exchange.load(listed["heart-rate-corrections"]) + if "heart-rate-corrections" in listed else None + ) cardio_calibrations = ( cardio_calibration_exchange.load(listed["cardio-calibrations"]) if "cardio-calibrations" in listed else None @@ -712,15 +770,17 @@ def consume_desktop(database: Path, directory: Path) -> dict: raise GenerationError("generation identity reused with different manifest") db.rollback() return json.loads(known[1]) - predecessors = db.execute("SELECT g.generation_id FROM sync_consumed_generations g WHERE g.producer_peer_id=? AND g.result='consumed' AND NOT EXISTS(SELECT 1 FROM sync_consumed_generations c WHERE c.parent_generation_id=g.generation_id AND c.result='consumed') LIMIT 2", (manifest["producer"]["peer_id"],)).fetchall() - if len(predecessors) > 1: raise GenerationError("consumed generation lineage has multiple tips") - expected_parent = predecessors[0][0] if predecessors else None - if manifest["parent_generation_id"] != expected_parent: - raise GenerationError("generation lineage is stale or unrelated") + require_current_consumed_parent( + db, + manifest["producer"]["peer_id"], + manifest["parent_generation_id"], + ) # Causal state is applied first. Complete-envelope helpers retain # item-level conflict rules while avoiding the legacy global guard. for operation in causal["operations"]: causal_delete_exchange.apply_operation(db, operation) + if heart_rate_corrections is not None: + heart_rate_correction_exchange.apply_document(db, heart_rate_corrections) import_exercise_profile_state.apply_profile_state(db, profile, allow_pending=True) import_equipment_definitions.apply_definitions(db, definitions, complete_causal_envelope=True) import_mobile_export.apply_payload(db, history, complete_causal_envelope=True) diff --git a/tui/meson.build b/tui/meson.build index c3ee3ba..7885ff6 100644 --- a/tui/meson.build +++ b/tui/meson.build @@ -986,6 +986,11 @@ test( find_program('python3'), args: [meson.project_source_root() / 'tests/test_heart_rate_exchange.py'], ) +test( + 'heart_rate_tail_recovery', + find_program('python3'), + args: [meson.project_source_root() / 'tests/test_heart_rate_tail_recovery.py'], +) test( 'session_timeline_exchange', find_program('python3'),