fix(cardio): truncate recovered session capture at factual end

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
+  private recovery build may supply the factual session end; strict late
-  heart-rate samples remain retained inside their capture bounds.
+  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

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 {
+                        .use { it.moveToFirst() }
-                            if (!it.moveToFirst()) null
-                            else {
-                                val value = it.getString(0)
-                                if (it.moveToNext())
-                                    throw SyncGenerationException(
-                                        "consumed generation lineage has multiple tips"
-                                    )
-                                value
-                            }
-                        }
                 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) {

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<Any?>(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<Any?>(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
+                val removedSequences = mutableListOf<Long>()
-                var captureEndKey = parsedFinalizedAt
                 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) {
+                        if (observedKey > parsedFinalizedAt) {
-                            captureEndKey = observedKey
+                            removedSequences += cursor.getLong(0)
-                            captureEndedAt = observedAt
                         }
                     }
                 }
                 /*
                  * 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
+                 * open; those later measurements are not part of the session.
-                 * would be data loss.
                  * CONTRACT: the session/timeline use the requested factual
-                 * end; the sensor capture expands only to its latest observed
+                 * end inclusively. Samples strictly after it are deleted, and
-                 * sample so Heart Rate V1 remains internally valid.
+                 * their RR children follow through the declared cascade.
-                 * INVARIANT: every retained sample lies within its stopped
+                 * INVARIANT: every retained sample is at or before the stopped
-                 * capture and no sample/RR identity or payload is rewritten.
+                 * 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

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(finalizationAt.toString(), capture.getString("ended_at"))
-            assertEquals(2, capture.getJSONArray("samples").length())
+            assertEquals(1, capture.getJSONArray("samples").length())
-            assertEquals(1, capture.getJSONArray("samples").getJSONObject(1)
+            assertEquals(1, capture.getJSONArray("samples").getJSONObject(0)
                 .getJSONArray("rr_intervals_1024").length())
             assertNull(repository.activeHeartRateCapture())
         } finally {

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,

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()

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
+before the session start or latest exercise marker. When heart-rate samples
-samples arrived after a failed factual session end, the session retains the
+arrived after a failed factual session end, delayed finalization keeps samples
-factual end while the stopped capture remains bounded by its latest retained
+at or before that inclusive end and transactionally deletes the strict tail.
-sample; no BPM or RR measurement is deleted or retimed.
+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

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 |

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

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

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

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

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,

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()

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

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()

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(),

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

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

tools/package_sync_candidate.py

@@ -22,6 +22,7 @@ RUNTIME_TOOLS = [
     "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": [

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)

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()
+            require_current_consumed_parent(
-            if len(predecessors) > 1: raise GenerationError("consumed generation lineage has multiple tips")
+                db,
-            expected_parent = predecessors[0][0] if predecessors else None
+                manifest["producer"]["peer_id"],
-            if manifest["parent_generation_id"] != expected_parent:
+                manifest["parent_generation_id"],
-                raise GenerationError("generation lineage is stale or unrelated")
+            )
             # 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)

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'),