feat(sync): implement data lifecycle staging

android/app/src/main/java/com/labfytools/trainlog/model/SessionModels.kt

@@ -101,6 +101,13 @@ data class SessionDraftForm(
  * mistake unfinished capture for a completed session.
  */
 data class ActiveSessionDraft(
+    /** Stable lifecycle identity, allocated once when first persisted. */
+    val sessionId: String? = null,
+    /** Causal draft revision; local UI-only persistence may retain its tip. */
+    val revisionId: String = "dr_legacy_v1",
+    val parentRevisionId: String? = null,
+    /** Genuine lifecycle start, distinct from local form update time. */
+    val startedAt: String? = null,
     val exercises: List<SessionExerciseDraft> = emptyList(),
     val sessionType: SessionType = SessionType.TRAINING,
     /** Existing max_test session updated atomically on finalization, if any. */

android/app/src/test/java/com/labfytools/trainlog/data/AndroidV10PlanningMigrationTest.kt

@@ -66,7 +66,7 @@ class AndroidV10PlanningMigrationTest {
 
         TrainlogRepository(context, name).useForTest { it.listSessions() }
         SQLiteDatabase.openDatabase(path.path, null, SQLiteDatabase.OPEN_READONLY).use { db ->
-            assertEquals(17, scalar(db, "PRAGMA user_version"))
+            assertEquals(18, scalar(db, "PRAGMA user_version"))
             for (table in listOf("session_exercises", "draft_session_exercises")) {
                 assertEquals(6, scalar(db, "SELECT COUNT(*) FROM pragma_table_info('$table') WHERE name IN ('load_mode','rest_seconds','target_sets','target_reps','target_duration_seconds','target_weight_kg')"))
                 assertEquals(0, scalar(db, "SELECT COUNT(*) FROM $table WHERE load_mode<>'none' OR rest_seconds<>0 OR target_sets IS NOT NULL OR target_reps IS NOT NULL OR target_duration_seconds IS NOT NULL OR target_weight_kg IS NOT NULL"))

android/app/src/test/java/com/labfytools/trainlog/data/AndroidV11ExerciseAliasMigrationTest.kt

@@ -69,7 +69,7 @@ class AndroidV11ExerciseAliasMigrationTest {
             migrated.close()
         }
         SQLiteDatabase.openDatabase(path, null, SQLiteDatabase.OPEN_READONLY).use { db ->
-            assertEquals(17, scalar(db, "PRAGMA user_version"))
+            assertEquals(18, scalar(db, "PRAGMA user_version"))
             /* v13 intentionally adds the frozen machine identities and
              * exercise-owned metadata; all non-exercise facts stay exact. */
             val after = snapshot(db).filterKeys { it !in setOf("exercise_aliases", "exercise_profile_state", "exercise_profile_revisions") }

android/app/src/test/java/com/labfytools/trainlog/data/AndroidV16AiSessionDraftMigrationTest.kt

@@ -78,7 +78,7 @@ class AndroidV16AiSessionDraftMigrationTest {
             migrated.close()
         }
         SQLiteDatabase.openDatabase(path, null, SQLiteDatabase.OPEN_READONLY).use { db ->
-            assertEquals(17, scalar(db, "PRAGMA user_version"))
+            assertEquals(18, scalar(db, "PRAGMA user_version"))
             assertEquals(before, listOf(
                 scalar(db, "SELECT COUNT(*) FROM sessions"),
                 scalar(db, "SELECT COUNT(*) FROM performed_sets"),

android/app/src/test/java/com/labfytools/trainlog/data/BodyZonesTest.kt

@@ -328,7 +328,7 @@ class BodyZonesTest {
         SQLiteDatabase.openDatabase(context.getDatabasePath(name).path, null,
             SQLiteDatabase.OPEN_READONLY).use { db ->
             db.rawQuery("PRAGMA user_version", null).use { cursor ->
-                assertTrue(cursor.moveToFirst()); assertEquals(17, cursor.getInt(0))
+                assertTrue(cursor.moveToFirst()); assertEquals(18, cursor.getInt(0))
             }
             db.rawQuery("SELECT id FROM exercises", null).use { cursor ->
                 assertTrue(cursor.moveToFirst()); assertEquals(42L, cursor.getLong(0))

android/app/src/test/java/com/labfytools/trainlog/data/TrainlogRepositoryDraftTest.kt

@@ -1794,7 +1794,7 @@ class TrainlogRepositoryDraftTest {
         ).use { db ->
             db.rawQuery("PRAGMA user_version;", null).use { cursor ->
                 assertTrue(cursor.moveToFirst())
-                assertEquals(17, cursor.getInt(0))
+                assertEquals(18, cursor.getInt(0))
             }
             db.rawQuery(
                 "SELECT eq.equipment_id, ps.reps, ps.weight_kg FROM session_exercises se " +
@@ -1921,7 +1921,7 @@ class TrainlogRepositoryDraftTest {
         ).use { db ->
             db.rawQuery("PRAGMA user_version;", null).use { cursor ->
                 assertTrue(cursor.moveToFirst())
-                assertEquals(17, cursor.getInt(0))
+                assertEquals(18, cursor.getInt(0))
             }
             db.rawQuery("SELECT weight_kg FROM performed_sets WHERE id = 1;", null).use { cursor ->
                 assertTrue(cursor.moveToFirst())
@@ -2446,6 +2446,60 @@ class TrainlogRepositoryDraftTest {
         assertEquals(3, exported.getJSONArray("sets").length())
     }
 
+    @Test
+    fun stagedExecutionDraftPersistsPendingAndFinalizationRejectsStaleReplay() {
+        val source = openRepository()
+        val exercise = createExercise(source, "Lifecycle press", RecordingMode.SETS, TrackingMode.REPS)
+        assertEquals(ActiveDraftMutationResult.Saved, source.saveActiveSessionDraft(
+            ActiveSessionDraft(exercises = listOf(SessionExerciseDraft(
+                entryId = "sxe_lifecycle_active", exercise = exercise,
+                plan = SessionExercisePlan(2, reps = 8, weightKg = 40.0,
+                    loadMode = SessionLoadMode.EXTERNAL, restSeconds = 90),
+                sets = listOf(SessionSetDraft(reps = 8, weightKg = 42.5)),
+            )), form = SessionDraftForm(weightText = "42.")),
+        ))
+        val artifact = JSONObject(source.buildExecutionDraftExportV1Json())
+        val active = artifact.getJSONArray("drafts").getJSONObject(0)
+        assertFalse(active.has("weight_text"))
+        assertFalse(active.toString().contains("weightText"))
+        val sessionId = active.getString("session_id")
+        val exerciseCatalog = JSONObject(source.buildMobileExportV4Json()).getJSONArray("exercises")
+        source.close(); repository = null
+        context.deleteDatabase(databaseName)
+
+        val destination = openRepository()
+        val catalog = JSONObject().put("format", "trainlog-pc-catalog").put("version", 1)
+            .put("exercises", exerciseCatalog)
+        assertTrue(destination.applyPcCatalogJson(catalog.toString()) is PcCatalogImportResult.Applied)
+        assertEquals(ExecutionDraftImportResult.Applied(1, 0, 0, 0),
+            destination.applyExecutionDraftExportV1Json(artifact.toString()))
+
+        val pending = JSONObject(active.toString())
+            .put("session_id", "se_00000000-0000-4000-8000-000000000042")
+            .put("revision_id", "dr_00000000-0000-4000-8000-000000000042")
+            .put("state", "pending")
+        pending.getJSONArray("entries").getJSONObject(0)
+            .put("entry_id", "sxe_lifecycle_pending")
+        val pendingArtifact = JSONObject(artifact.toString())
+            .put("drafts", JSONArray().put(pending))
+        assertEquals(ExecutionDraftImportResult.Applied(0, 1, 0, 0),
+            destination.applyExecutionDraftExportV1Json(pendingArtifact.toString()))
+        destination.close(); repository = null
+        val reopened = openRepository()
+        assertEquals(2, JSONObject(reopened.buildExecutionDraftExportV1Json())
+            .getJSONArray("drafts").length())
+        assertEquals(ActivateExecutionDraftResult.ActiveDraftOccupied,
+            reopened.activatePendingExecutionDraft(pending.getString("session_id")))
+        assertEquals(FinalizeActiveDraftResult.Saved(sessionId), reopened.finalizeExecutionDraft(sessionId))
+        assertEquals(FinalizeActiveDraftResult.Saved(sessionId), reopened.finalizeExecutionDraft(sessionId))
+        assertEquals(ExecutionDraftImportResult.Applied(0, 0, 0, 1),
+            reopened.applyExecutionDraftExportV1Json(artifact.toString()))
+        assertEquals(ActivateExecutionDraftResult.Activated,
+            reopened.activatePendingExecutionDraft(pending.getString("session_id")))
+        assertEquals(pending.getString("session_id"),
+            (reopened.loadActiveSessionDraft() as ActiveDraftLoadResult.Loaded).draft.sessionId)
+    }
+
     @Test
     fun exerciseAliasCompanionRekeysHistoryAndDraftAndReplaysIdempotently() {
         val repo = openRepository()

tools/execution_draft_exchange.py

@@ -0,0 +1,275 @@
+#!/usr/bin/env python3
+"""Explicit staged execution-draft codec and desktop lifecycle service.
+
+This module is deliberately absent from trainlog_sync_run(): local persistence
+success is not peer delivery and the active MTP protocol remains V3.
+"""
+import argparse
+import json
+import sqlite3
+from datetime import datetime
+from pathlib import Path
+
+from trainlog_sqlite import connect_database
+from validate_json import TrainlogSemanticError, parse_timestamp
+
+FORMAT = "trainlog-execution-drafts"
+VERSION = 1
+MAX_DOCUMENT_BYTES = 4 * 1024 * 1024
+MAX_DRAFTS = 16
+MAX_ENTRIES = 64
+ROOT_KEYS = {"format", "version", "generated_at", "drafts"}
+DRAFT_KEYS = {"session_id", "session_type", "source_session_id", "started_at",
+              "revision_id", "parent_revision_id", "state", "note", "entries"}
+ENTRY_KEYS = {"entry_id", "position", "exercise_id", "equipment_id",
+              "recording_mode", "tracking_mode", "data_fields", "load_mode",
+              "rest_seconds", "target", "sets", "continuous", "max_weight_kg",
+              "note", "feedback"}
+
+
+class LifecycleError(RuntimeError):
+    pass
+
+
+def exact(value, keys, label):
+    if not isinstance(value, dict) or set(value) != keys:
+        raise LifecycleError(f"{label}: exact keys required")
+
+
+def text(value, label, nullable=False, maximum=128):
+    if nullable and value is None:
+        return
+    if not isinstance(value, str) or not value or len(value.encode()) > maximum:
+        raise LifecycleError(f"{label}: invalid text")
+
+
+def timestamp(value, label, nullable=False):
+    if nullable and value is None:
+        return
+    text(value, label)
+    try:
+        parse_timestamp(value, label)
+    except TrainlogSemanticError as error:
+        raise LifecycleError(f"{label}: invalid timestamp") from error
+
+
+def note(value, label):
+    if value is None:
+        return
+    exact(value, {"value", "revision_id", "parent_revision_id"}, label)
+    text(value["revision_id"], label + ".revision_id")
+    text(value["parent_revision_id"], label + ".parent_revision_id", nullable=True)
+    if value["parent_revision_id"] == value["revision_id"]:
+        raise LifecycleError(f"{label}: self parent")
+    if value["value"] is not None and (not isinstance(value["value"], str) or
+                                       len(value["value"].encode()) > 4096):
+        raise LifecycleError(f"{label}: note exceeds 4096 UTF-8 bytes")
+
+
+def validate(document):
+    exact(document, ROOT_KEYS, "root")
+    if document["format"] != FORMAT or document["version"] != VERSION:
+        raise LifecycleError("unsupported execution-draft format/version")
+    timestamp(document["generated_at"], "generated_at")
+    drafts = document["drafts"]
+    if not isinstance(drafts, list) or len(drafts) > MAX_DRAFTS:
+        raise LifecycleError("drafts: invalid bound")
+    identities = set()
+    for di, draft in enumerate(drafts):
+        label = f"drafts[{di}]"
+        exact(draft, DRAFT_KEYS, label)
+        text(draft["session_id"], label + ".session_id")
+        if draft["session_id"] in identities:
+            raise LifecycleError(label + ": duplicate session_id")
+        identities.add(draft["session_id"])
+        if draft["session_type"] not in ("training", "max_test") or draft["state"] not in ("active", "pending"):
+            raise LifecycleError(label + ": invalid lifecycle enum")
+        text(draft["source_session_id"], label + ".source_session_id", nullable=True)
+        timestamp(draft["started_at"], label + ".started_at", nullable=True)
+        text(draft["revision_id"], label + ".revision_id")
+        text(draft["parent_revision_id"], label + ".parent_revision_id", nullable=True)
+        if draft["revision_id"] == draft["parent_revision_id"]:
+            raise LifecycleError(label + ": self parent")
+        note(draft["note"], label + ".note")
+        entries = draft["entries"]
+        if not isinstance(entries, list) or len(entries) > MAX_ENTRIES:
+            raise LifecycleError(label + ": entries bound")
+        entry_ids, positions = set(), set()
+        for ei, entry in enumerate(entries):
+            elabel = f"{label}.entries[{ei}]"
+            exact(entry, ENTRY_KEYS, elabel)
+            text(entry["entry_id"], elabel + ".entry_id")
+            text(entry["exercise_id"], elabel + ".exercise_id")
+            text(entry["equipment_id"], elabel + ".equipment_id", nullable=True)
+            if entry["entry_id"] in entry_ids or not isinstance(entry["position"], int) or entry["position"] < 0 or entry["position"] in positions:
+                raise LifecycleError(elabel + ": duplicate identity/position")
+            entry_ids.add(entry["entry_id"]); positions.add(entry["position"])
+            if entry["recording_mode"] not in ("sets", "continuous") or entry["tracking_mode"] not in ("reps", "duration"):
+                raise LifecycleError(elabel + ": invalid profile")
+            if entry["recording_mode"] == "continuous" and entry["tracking_mode"] != "duration":
+                raise LifecycleError(elabel + ": continuous requires duration")
+            if not isinstance(entry["data_fields"], int) or entry["data_fields"] < 0 or entry["data_fields"] & ~3:
+                raise LifecycleError(elabel + ": invalid data_fields")
+            if entry["load_mode"] not in ("none", "external", "assistance") or not isinstance(entry["rest_seconds"], int) or not 0 <= entry["rest_seconds"] <= 86400:
+                raise LifecycleError(elabel + ": invalid plan")
+            note(entry["note"], elabel + ".note")
+            if not isinstance(entry["feedback"], list) or len(entry["feedback"]) > 256:
+                raise LifecycleError(elabel + ": feedback bound")
+            feedback_ids = set()
+            for fi, feedback in enumerate(entry["feedback"]):
+                flabel = f"{elabel}.feedback[{fi}]"
+                exact(feedback, {"feedback_id", "observed_at", "raw_text", "revisions"}, flabel)
+                text(feedback["feedback_id"], flabel + ".feedback_id")
+                if feedback["feedback_id"] in feedback_ids:
+                    raise LifecycleError(flabel + ": duplicate feedback_id")
+                feedback_ids.add(feedback["feedback_id"])
+                timestamp(feedback["observed_at"], flabel + ".observed_at")
+                text(feedback["raw_text"], flabel + ".raw_text", maximum=8192)
+                if not isinstance(feedback["revisions"], list) or len(feedback["revisions"]) > 32:
+                    raise LifecycleError(flabel + ": revisions bound")
+                revision_ids = set()
+                for ri, revision in enumerate(feedback["revisions"]):
+                    rlabel = f"{flabel}.revisions[{ri}]"
+                    exact(revision, {"revision_id", "created_at", "raw_text"}, rlabel)
+                    text(revision["revision_id"], rlabel + ".revision_id")
+                    timestamp(revision["created_at"], rlabel + ".created_at")
+                    text(revision["raw_text"], rlabel + ".raw_text", maximum=8192)
+                    if revision["revision_id"] in revision_ids:
+                        raise LifecycleError(rlabel + ": duplicate revision_id")
+                    revision_ids.add(revision["revision_id"])
+            # The V4 history validator owns exact performed-data shapes. Draft
+            # import refuses ambiguous simultaneous representations here.
+            shapes = sum(value is not None for value in
+                         (entry["sets"], entry["continuous"], entry["max_weight_kg"]))
+            if shapes > 1:
+                raise LifecycleError(elabel + ": contradictory actual shapes")
+    return document
+
+
+def load(path):
+    raw = path.read_bytes()
+    if len(raw) > MAX_DOCUMENT_BYTES:
+        raise LifecycleError("document exceeds 4 MiB")
+    def unique(pairs):
+        result = {}
+        for key, value in pairs:
+            if key in result:
+                raise LifecycleError("duplicate JSON key: " + key)
+            result[key] = value
+        return result
+    try:
+        return validate(json.loads(raw.decode("utf-8"), object_pairs_hook=unique))
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise LifecycleError("invalid UTF-8/JSON") from error
+
+
+def canonical(value):
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def require_schema(connection):
+    version = connection.execute("PRAGMA user_version").fetchone()[0]
+    if version != 19:
+        raise LifecycleError(f"desktop schema v19 required, found v{version}")
+
+
+def import_document(connection, document):
+    require_schema(connection)
+    outcomes = []
+    connection.execute("BEGIN IMMEDIATE")
+    try:
+        for draft in document["drafts"]:
+            session_id, revision = draft["session_id"], draft["revision_id"]
+            payload = canonical(draft)
+            if connection.execute("SELECT 1 FROM execution_draft_finalizations WHERE session_id=?", (session_id,)).fetchone():
+                outcomes.append((session_id, "stale")); continue
+            current = connection.execute(
+                "SELECT revision_id,payload_json,state FROM execution_drafts WHERE session_id=?", (session_id,)).fetchone()
+            known = connection.execute(
+                "SELECT parent_revision_id,payload_json FROM execution_draft_revisions WHERE session_id=? AND revision_id=?",
+                (session_id, revision)).fetchone()
+            if known:
+                if tuple(known) != (draft["parent_revision_id"], payload):
+                    raise LifecycleError("contradictory revision: " + session_id)
+                outcomes.append((session_id, "unchanged" if current and current[0] == revision else "stale")); continue
+            if current and draft["parent_revision_id"] != current[0]:
+                outcomes.append((session_id, "conflict")); continue
+            if not current:
+                active = connection.execute("SELECT 1 FROM execution_drafts WHERE state='active'").fetchone()
+                pending = connection.execute("SELECT COUNT(*) FROM execution_drafts WHERE state='pending'").fetchone()[0]
+                state = "pending" if active else "active"
+                if state == "pending" and pending >= MAX_DRAFTS:
+                    raise LifecycleError("pending execution-draft capacity exhausted")
+                connection.execute(
+                    "INSERT INTO execution_drafts VALUES(?,?,?,?,?,?,?,?)",
+                    (session_id, draft["session_type"], draft["source_session_id"], draft["started_at"],
+                     revision, draft["parent_revision_id"], state, payload))
+                outcome = state
+            else:
+                connection.execute(
+                    "UPDATE execution_drafts SET revision_id=?,parent_revision_id=?,payload_json=? WHERE session_id=?",
+                    (revision, draft["parent_revision_id"], payload, session_id))
+                outcome = "applied"
+            connection.execute("INSERT INTO execution_draft_revisions VALUES(?,?,?,?)",
+                               (session_id, revision, draft["parent_revision_id"], payload))
+            outcomes.append((session_id, outcome))
+        connection.commit()
+    except Exception:
+        connection.rollback(); raise
+    return outcomes
+
+
+def export_document(connection):
+    require_schema(connection)
+    drafts = [json.loads(row[0]) for row in connection.execute(
+        "SELECT payload_json FROM execution_drafts ORDER BY CASE state WHEN 'active' THEN 0 ELSE 1 END,session_id")]
+    for draft, row in zip(drafts, connection.execute(
+            "SELECT state FROM execution_drafts ORDER BY CASE state WHEN 'active' THEN 0 ELSE 1 END,session_id")):
+        draft["state"] = row[0]
+    return {"format": FORMAT, "version": VERSION,
+            "generated_at": datetime.now().astimezone().isoformat(), "drafts": drafts}
+
+
+def activate(connection, session_id):
+    require_schema(connection)
+    connection.execute("BEGIN IMMEDIATE")
+    try:
+        if connection.execute("SELECT 1 FROM execution_drafts WHERE state='active'").fetchone():
+            result = "occupied"
+        elif connection.execute("SELECT 1 FROM execution_drafts WHERE session_id=? AND state='pending'", (session_id,)).fetchone():
+            connection.execute("UPDATE execution_drafts SET state='active' WHERE session_id=?", (session_id,)); result = "activated"
+        else:
+            result = "not_found"
+        connection.commit(); return result
+    except Exception:
+        connection.rollback(); raise
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("mode", choices=("import", "export", "activate"))
+    parser.add_argument("path", type=Path)
+    parser.add_argument("--database", type=Path, required=True)
+    parser.add_argument("--session-id")
+    args = parser.parse_args()
+    connection = connect_database(args.database)
+    try:
+        if args.mode == "import":
+            for identity, outcome in import_document(connection, load(args.path)):
+                print(f"{identity}={outcome}")
+        elif args.mode == "export":
+            args.path.write_text(canonical(export_document(connection)), encoding="utf-8")
+            print("EXECUTION_DRAFT_EXPORT=PASS")
+        else:
+            if not args.session_id: raise LifecycleError("--session-id required")
+            print("activation=" + activate(connection, args.session_id))
+    finally:
+        connection.close()
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        print("EXECUTION_DRAFT=FAIL " + str(error))
+        raise SystemExit(1)

tools/export_ai_history.py

@@ -12,7 +12,7 @@ from pathlib import Path
 from trainlog_sqlite import configure_connection
 
 
-SUPPORTED_SCHEMA_VERSIONS = (15, 16, 17, 18)
+SUPPORTED_SCHEMA_VERSIONS = (15, 16, 17, 18, 19)
 MEASUREMENT_FIELDS = (
     "body_weight_kg", "neck_cm", "shoulders_cm", "chest_cm", "waist_cm",
     "hips_cm", "left_arm_cm", "right_arm_cm", "left_forearm_cm",

tools/export_ai_session_drafts.py

@@ -20,7 +20,7 @@ ID_PREFIX = "aid_"
 def ensure_publication_state(connection):
     """Install the additive v18 cursor for direct syncd/tool invocations."""
     owned_transaction = not connection.in_transaction
-    if connection.execute("PRAGMA user_version").fetchone()[0] != 18:
+    if connection.execute("PRAGMA user_version").fetchone()[0] not in (18, 19):
         raise ValueError("schema desktop v18 requis")
     columns = {row[1] for row in connection.execute(
         "PRAGMA table_info(ai_session_drafts)")}

tools/export_equipment_associations.py

@@ -28,7 +28,7 @@ def main():
     args = parser.parse_args()
     connection = connect_database(args.database)
     try:
-        if connection.execute("PRAGMA user_version;").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18):
+        if connection.execute("PRAGMA user_version;").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19):
             raise ValueError("schema desktop v8 à v16 requis")
         known_equipment = load_supplied_equipment_ids(args.catalog)
         known_equipment.update(row[0] for row in connection.execute(

tools/export_equipment_definitions.py

@@ -24,7 +24,7 @@ def main():
     connection = connect_database(args.database)
     connection.row_factory = sqlite3.Row
     try:
-        if connection.execute("PRAGMA user_version").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18):
+        if connection.execute("PRAGMA user_version").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19):
             raise ValueError("schema desktop v8 à v16 requis")
         equipment = [dict(row) for row in connection.execute(
             "SELECT equipment_id,display_name,label_name,equipment_type,load_semantics "

tools/export_exercise_aliases.py

@@ -16,7 +16,7 @@ def main():
     args = parser.parse_args()
     con = connect_database(args.database)
     try:
-        if con.execute("PRAGMA user_version").fetchone()[0] not in (12, 13, 14, 15, 16, 17, 18):
+        if con.execute("PRAGMA user_version").fetchone()[0] not in (12, 13, 14, 15, 16, 17, 18, 19):
             raise ValueError("schema desktop v12-v16 requis")
         rows = con.execute(
             "SELECT source_exercise_id,canonical_exercise_id FROM exercise_aliases "

tools/export_exercise_body_zones.py

@@ -41,7 +41,7 @@ def main():
     connection = connect_database(args.database)
     connection.row_factory = sqlite3.Row
     try:
-        if connection.execute("PRAGMA user_version").fetchone()[0] not in (11, 12, 13, 14, 15, 16, 17, 18):
+        if connection.execute("PRAGMA user_version").fetchone()[0] not in (11, 12, 13, 14, 15, 16, 17, 18, 19):
             raise ValueError("schema desktop v11-v16 requis")
         exercises = []
         for exercise in connection.execute("SELECT id,exercise_id FROM exercises ORDER BY exercise_id"):

tools/export_exercise_profile_state.py

@@ -25,7 +25,7 @@ def main():
     con.row_factory = sqlite3.Row
     try:
         version = con.execute("PRAGMA user_version").fetchone()[0]
-        if version not in (17, 18):
+        if version not in (17, 18, 19):
             raise SystemExit(f"EXERCISE_PROFILE_STATE_EXPORT=FAIL schema={version}")
         rows = con.execute("""
             SELECT e.exercise_id,e.recording_mode,e.tracking_mode,e.data_fields,

tools/export_pc_catalog.py

@@ -71,7 +71,7 @@ def main():
         # CONTRACT: v8 adds only desktop-local custom equipment.  The PC
         # catalogue artifact is unchanged, but it must read the current
         # canonical desktop schema rather than accept a stale pre-v8 database.
-        if version not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18):
+        if version not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19):
             raise SystemExit(
                 "PC_CATALOG_EXPORT=FAIL "
                 f"schema={version}"

tools/export_pc_mobile.py

@@ -59,17 +59,33 @@ def validate_v3_timestamp(value, label):
         raise ValueError(f"{label}: date-heure Trainlog persistée invalide") from error
 
 
+def note_payload(connection, owner_kind, owner_id, value):
+    row = connection.execute(
+        "SELECT r.revision_id,r.parent_revision_id,r.note FROM sync_note_state s "
+        "JOIN sync_note_revisions r USING(owner_kind,owner_id,revision_id) "
+        "WHERE s.owner_kind=? AND s.owner_id=?", (owner_kind, owner_id)).fetchone()
+    if row is None:
+        if value is not None:
+            raise ValueError(f"note héritée sans révision causale: {owner_kind}/{owner_id}")
+        return None
+    if row[2] != value:
+        raise ValueError(f"cache de note incohérent: {owner_kind}/{owner_id}")
+    return {"value": row[2], "revision_id": row[0], "parent_revision_id": row[1]}
+
+
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("output", type=Path)
     parser.add_argument("--database", type=Path, default=default_database())
-    parser.add_argument("--version", type=int, choices=(2, 3), default=3)
+    parser.add_argument("--version", type=int, choices=(2, 3, 4), default=3)
     args = parser.parse_args()
     con = connect_database(args.database)
     con.row_factory = sqlite3.Row
     try:
         schema_version = con.execute("PRAGMA user_version").fetchone()[0]
-        if schema_version not in (11, 12, 13, 14, 15, 16, 17, 18) and not (args.version == 2 and schema_version == 10):
+        if args.version == 4 and schema_version != 19:
+            raise ValueError("mobile V4 exige le schéma desktop v19")
+        if schema_version not in (11, 12, 13, 14, 15, 16, 17, 18, 19) and not (args.version == 2 and schema_version == 10):
             raise ValueError("schema desktop v11-v16 requis (v10 accepté pour export V2 explicite)")
         known_equipment = supplied_equipment_ids()
         canonical_names = load_exercise_names()
@@ -82,18 +98,25 @@ def main():
             exported = dict(row)
             exported["name"] = canonical_names.get(row["exercise_id"], row["name"])
             root["exercises"].append(exported)
-        for session in con.execute("SELECT id,session_id,started_at,session_type FROM sessions ORDER BY started_at,id"):
+        for session in con.execute("SELECT id,session_id,started_at,session_type,ended_at,notes FROM sessions ORDER BY started_at,id"):
             # CONTRACT: a V3 producer must not publish history which the exact
             # temporal readers reject. V2 export retains its published behavior.
-            if args.version == 3:
+            if args.version >= 3:
                 validate_v3_timestamp(session["started_at"],
                                       f"session_id={session['session_id']} started_at")
             payload = {key: session[key] for key in ("session_id", "started_at", "session_type")}
+            if args.version == 4:
+                if session["ended_at"] is not None:
+                    validate_v3_timestamp(session["ended_at"], f"session_id={session['session_id']} ended_at")
+                    if parse_timestamp(session["ended_at"], "ended_at") <= parse_timestamp(session["started_at"], "started_at"):
+                        raise ValueError("ended_at doit être postérieur à started_at")
+                payload["ended_at"] = session["ended_at"]
+                payload["note"] = note_payload(con, "session", session["session_id"], session["notes"])
             payload["exercises"] = []
             # INVARIANT: v16 exports the occurrence-owned tracking snapshot;
             # older readable schemas retain their bounded catalogue fallback.
             tracking_sql = "se.tracking_mode" if schema_version >= 16 else "e.tracking_mode"
-            sql = f"SELECT se.id,se.entry_id,se.position,se.recording_mode,{tracking_sql},se.data_fields,se.equipment_id,e.exercise_id,e.name,mr.max_weight_kg,se.load_mode,se.rest_seconds,se.target_sets,se.target_reps,se.target_duration_seconds,se.target_weight_kg FROM session_exercises se JOIN exercises e ON e.id=se.exercise_row_id LEFT JOIN max_results mr ON mr.session_exercise_row_id=se.id WHERE se.session_row_id=? ORDER BY se.position"
+            sql = f"SELECT se.id,se.entry_id,se.position,se.recording_mode,{tracking_sql},se.data_fields,se.equipment_id,e.exercise_id,e.name,mr.max_weight_kg,se.load_mode,se.rest_seconds,se.target_sets,se.target_reps,se.target_duration_seconds,se.target_weight_kg,se.notes FROM session_exercises se JOIN exercises e ON e.id=se.exercise_row_id LEFT JOIN max_results mr ON mr.session_exercise_row_id=se.id WHERE se.session_row_id=? ORDER BY se.position"
             for entry in con.execute(sql, (session["id"],)):
                 validate_plan(entry)
                 if entry["recording_mode"] == "continuous" or entry["max_weight_kg"] is not None:
@@ -118,8 +141,8 @@ def main():
                         "name": canonical_names.get(entry["exercise_id"], entry["name"]),
                         "recording_mode": entry["recording_mode"], "tracking_mode": entry["tracking_mode"],
                         "data_fields": entry["data_fields"],
-                        "load_mode": entry["load_mode"] if args.version == 3 else "none",
-                        "rest_seconds": entry["rest_seconds"] if args.version == 3 else 0,
+                        "load_mode": entry["load_mode"] if args.version >= 3 else "none",
+                        "rest_seconds": entry["rest_seconds"] if args.version >= 3 else 0,
                         "equipment_id": entry["equipment_id"]}
                 if args.version == 3:
                     if not has_target:
@@ -134,6 +157,16 @@ def main():
                         if entry["target_duration_seconds"] is not None: target["duration_seconds"] = entry["target_duration_seconds"]
                         if entry["target_weight_kg"] is not None: target["weight_kg"] = entry["target_weight_kg"]
                         item["target"] = target
+                elif args.version == 4:
+                    if not has_target:
+                        item["target"] = None
+                    else:
+                        target = {"sets": entry["target_sets"]}
+                        if entry["target_reps"] is not None: target["reps"] = entry["target_reps"]
+                        if entry["target_duration_seconds"] is not None: target["duration_seconds"] = entry["target_duration_seconds"]
+                        if entry["target_weight_kg"] is not None: target["weight_kg"] = entry["target_weight_kg"]
+                        item["target"] = target
+                    item["note"] = note_payload(con, "occurrence", entry["entry_id"], entry["notes"])
                 if entry["max_weight_kg"] is not None:
                     # CONTRACT: explicit max is an occurrence result, never a
                     # synthetic one-repetition performed set.
@@ -155,15 +188,18 @@ def main():
                         "waist_cm", "hips_cm", "left_arm_cm", "right_arm_cm",
                         "left_forearm_cm", "right_forearm_cm", "left_thigh_cm",
                         "right_thigh_cm", "left_calf_cm", "right_calf_cm")
-        columns = ",".join(("observation_id", "observed_at") + metric_names)
-        for row in con.execute(f"SELECT {columns} FROM body_observations ORDER BY observed_at,id"):
-            if args.version == 3:
+        columns = ",".join(("bo.observation_id", "bo.observed_at") + tuple("bo." + n for n in metric_names))
+        for row in con.execute(f"SELECT {columns},s.session_id,bo.notes FROM body_observations bo LEFT JOIN sessions s ON s.id=bo.session_row_id ORDER BY bo.observed_at,bo.id"):
+            if args.version >= 3:
                 # INVARIANT: validate every stored instant before touching the
                 # destination artifact, so corruption cannot clobber a prior file.
                 validate_v3_timestamp(row["observed_at"],
                                       f"observation_id={row['observation_id']} observed_at")
             item = {"observation_id": row["observation_id"], "observed_at": row["observed_at"]}
             item.update({name: row[name] for name in metric_names if row[name] is not None})
+            if args.version == 4:
+                item["session_id"] = row["session_id"]
+                item["note"] = note_payload(con, "observation", row["observation_id"], row["notes"])
             root["body_observations"].append(item)
         args.output.write_text(json.dumps(root, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
         print("PC_MOBILE_EXPORT=PASS")

tools/export_training_feedback.py

@@ -10,7 +10,7 @@ def main():
     parser=argparse.ArgumentParser(); parser.add_argument("output",type=Path); parser.add_argument("--database",type=Path,required=True); args=parser.parse_args()
     db=connect_database(args.database)
     try:
-        if db.execute("PRAGMA user_version").fetchone()[0] not in (15, 16, 17, 18): raise ValueError("schema desktop v15-v18 requis")
+        if db.execute("PRAGMA user_version").fetchone()[0] not in (15, 16, 17, 18, 19): raise ValueError("schema desktop v15-v19 requis")
         roots=db.execute("SELECT f.feedback_id,s.session_id,se.entry_id,e.exercise_id,f.observed_at FROM exercise_feedback f JOIN session_exercises se ON se.id=f.session_exercise_row_id JOIN sessions s ON s.id=se.session_row_id JOIN exercises e ON e.id=se.exercise_row_id").fetchall()
         followups=db.execute("SELECT f.followup_id,s.session_id,f.observed_at FROM session_followups f JOIN sessions s ON s.id=f.session_row_id").fetchall()
         if len(roots)>4096 or len(followups)>4096: raise ValueError("trop d'observations")

tools/import_ai_session_draft.py

@@ -193,7 +193,7 @@ def import_payload(connection, payload, imported_at=None):
     INVARIANT: validation and every child insert share one IMMEDIATE transaction;
     any failure leaves no draft, child, or import-identity row behind.
     """
-    if connection.execute("PRAGMA user_version").fetchone()[0] != 18:
+    if connection.execute("PRAGMA user_version").fetchone()[0] not in (18, 19):
         raise ImportFailure("schema desktop v18 requis")
     draft, entries = validate(payload, connection)
     digest = canonical_digest(payload)

tools/import_equipment_associations.py

@@ -173,7 +173,7 @@ def main():
         fail("clés extension équipement invalides")
     connection = connect_database(args.database)
     try:
-        if connection.execute("PRAGMA user_version;").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18):
+        if connection.execute("PRAGMA user_version;").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19):
             fail("schema desktop v8 à v16 requis")
         known = load_catalog(args.catalog)
         known.update(row[0] for row in connection.execute(

tools/import_equipment_definitions.py

@@ -61,7 +61,7 @@ def main():
     definitions = validate(json.loads(args.artifact.read_text(encoding="utf-8")), supplied_ids(args.catalog))
     connection = connect_database(args.database)
     try:
-        if connection.execute("PRAGMA user_version").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18):
+        if connection.execute("PRAGMA user_version").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19):
             fail("schema desktop v8 à v16 requis")
         imported = skipped = 0
         # Validate every same-ID row before inserting any definition.

tools/import_exercise_aliases.py

@@ -63,7 +63,7 @@ def main():
     con = connect_database(args.database)
     try:
         con.execute("PRAGMA foreign_keys=ON")
-        if con.execute("PRAGMA user_version").fetchone()[0] not in (12, 13, 14, 15, 16, 17, 18):
+        if con.execute("PRAGMA user_version").fetchone()[0] not in (12, 13, 14, 15, 16, 17, 18, 19):
             fail("schema desktop v12-v16 requis")
         con.execute("BEGIN IMMEDIATE")
         for source, canonical in aliases:

tools/import_exercise_body_zones.py

@@ -205,7 +205,7 @@ def main():
     connection = connect_database(args.database)
     updated = skipped = kept_local = 0
     try:
-        if connection.execute("PRAGMA user_version").fetchone()[0] not in (11, 12, 13, 14, 15, 16, 17, 18):
+        if connection.execute("PRAGMA user_version").fetchone()[0] not in (11, 12, 13, 14, 15, 16, 17, 18, 19):
             raise ImportFailure("schema desktop v11-v16 requis")
         connection.execute("PRAGMA foreign_keys=ON")
         connection.execute("BEGIN IMMEDIATE")

tools/import_exercise_profile_state.py

@@ -53,7 +53,7 @@ def main():
         if not isinstance(root,dict) or set(root)!={"format","version","generated_at","exercises"} or root["format"]!="trainlog-exercise-profile-state" or type(root["version"]) is not int or root["version"]!=1 or not isinstance(root["exercises"],list): fail("enveloppe profile-state invalide")
         parse_timestamp(root["generated_at"], "generated_at")
         con=connect_database(a.database);con.row_factory=sqlite3.Row
-        if con.execute("PRAGMA user_version").fetchone()[0] not in (17,18): fail("schema desktop v17/v18 requis")
+        if con.execute("PRAGMA user_version").fetchone()[0] not in (17,18,19): fail("schema desktop v17/v18/v19 requis")
         applied=equal=ancestor=pending=0;seen=set();con.execute("BEGIN IMMEDIATE")
         for item in root["exercises"]:
             validate(item);eid=item["exercise_id"]

tools/import_mobile_export.py

@@ -48,6 +48,7 @@ SESSION_KEYS = {
     "session_type",
     "exercises",
 }
+V4_SESSION_KEYS = SESSION_KEYS | {"ended_at", "note"}
 
 SESSION_EXERCISE_KEYS = {
     "exercise_id",
@@ -65,11 +66,13 @@ V2_SESSION_EXERCISE_KEYS = SESSION_EXERCISE_KEYS | {
     "entry_id", "position", "equipment_id", "max_weight_kg"
 }
 V3_SESSION_EXERCISE_KEYS = V2_SESSION_EXERCISE_KEYS | {"target"}
+V4_SESSION_EXERCISE_KEYS = V3_SESSION_EXERCISE_KEYS | {"note"}
 
 BODY_BASE_KEYS = {
     "observation_id",
     "observed_at",
 }
+V4_BODY_KEYS = BODY_BASE_KEYS | {"session_id", "note"}
 
 BODY_METRIC_KEYS = {
     "body_weight_kg",
@@ -255,14 +258,14 @@ def load_payload(path):
             "format mobile export invalide"
         )
 
-    if payload["version"] not in (1, 2, 3):
+    if payload["version"] not in (1, 2, 3, 4):
         raise ImportFailure(
             "version mobile export non supportée"
         )
 
     # CONTRACT: V1/V2 retain their published nonempty-string admission. V3 is
     # the current analysis-bearing artifact and uses the exact Trainlog parser.
-    if payload["version"] == 3:
+    if payload["version"] >= 3:
         require_v3_timestamp(payload["generated_at"], "generated_at")
     else:
         require_nonempty_string(payload["generated_at"], "generated_at")
@@ -447,6 +450,24 @@ def validate_set_item(
     return ("duration", duration)
 
 
+def validate_note(value, label):
+    if value is None:
+        return
+    require_exact_keys(value, {"value", "revision_id", "parent_revision_id"},
+                       {"value", "revision_id", "parent_revision_id"}, label)
+    revision = require_nonempty_string(value["revision_id"], f"{label}.revision_id")
+    if len(revision.encode("utf-8")) > 128:
+        raise ImportFailure(f"{label}.revision_id: hors bornes")
+    parent = value["parent_revision_id"]
+    if parent is not None:
+        require_nonempty_string(parent, f"{label}.parent_revision_id")
+        if len(parent.encode("utf-8")) > 128 or parent == revision:
+            raise ImportFailure(f"{label}.parent_revision_id: invalide")
+    note = value["value"]
+    if note is not None and (not isinstance(note, str) or len(note.encode("utf-8")) > 4096):
+        raise ImportFailure(f"{label}.value: texte UTF-8 hors bornes")
+
+
 def validate_session_exercise(
     item,
     label,
@@ -455,7 +476,7 @@ def validate_session_exercise(
     version,
 ):
     is_v2 = version >= 2
-    keys = V3_SESSION_EXERCISE_KEYS if version == 3 else (
+    keys = V4_SESSION_EXERCISE_KEYS if version == 4 else V3_SESSION_EXERCISE_KEYS if version == 3 else (
         V2_SESSION_EXERCISE_KEYS if is_v2 else SESSION_EXERCISE_KEYS)
     require_exact_keys(
         item,
@@ -464,6 +485,8 @@ def validate_session_exercise(
         - {"sets", "continuous", "max_weight_kg"},
         label,
     )
+    if version == 4:
+        validate_note(item["note"], f"{label}.note")
 
     if is_v2:
         require_nonempty_string(item["entry_id"], f"{label}.entry_id")
@@ -650,12 +673,8 @@ def validate_sessions(
     ):
         label = f"sessions[{index}]"
 
-        require_exact_keys(
-            session,
-            SESSION_KEYS,
-            SESSION_KEYS,
-            label,
-        )
+        session_keys = V4_SESSION_KEYS if payload["version"] == 4 else SESSION_KEYS
+        require_exact_keys(session, session_keys, session_keys, label)
 
         session_id = require_nonempty_string(
             session["session_id"],
@@ -669,13 +688,21 @@ def validate_sessions(
 
         seen_ids.add(session_id)
 
-        if payload["version"] == 3:
+        if payload["version"] >= 3:
             # INVARIANT: accepted V3 history must remain readable by temporal
             # analysis; validation completes before run_import mutates SQLite.
             require_v3_timestamp(session["started_at"], f"{label}.started_at")
         else:
             require_nonempty_string(session["started_at"], f"{label}.started_at")
 
+        if payload["version"] == 4:
+            validate_note(session["note"], f"{label}.note")
+            if session["ended_at"] is not None:
+                require_v3_timestamp(session["ended_at"], f"{label}.ended_at")
+                if parse_timestamp(session["ended_at"], f"{label}.ended_at") <= parse_timestamp(
+                    session["started_at"], f"{label}.started_at"):
+                    raise ImportFailure(f"{label}.ended_at: doit être postérieur au début")
+
         if session["session_type"] not in (
             "training",
             "max_test",
@@ -748,18 +775,16 @@ def validate_sessions(
 def validate_body(payload):
     seen_ids = set()
     allowed = BODY_BASE_KEYS | BODY_METRIC_KEYS
+    if payload["version"] == 4:
+        allowed |= V4_BODY_KEYS
 
     for index, observation in enumerate(
         payload["body_observations"]
     ):
         label = f"body_observations[{index}]"
 
-        require_exact_keys(
-            observation,
-            allowed,
-            BODY_BASE_KEYS,
-            label,
-        )
+        required = BODY_BASE_KEYS | ({"session_id", "note"} if payload["version"] == 4 else set())
+        require_exact_keys(observation, allowed, required, label)
 
         observation_id = require_nonempty_string(
             observation["observation_id"],
@@ -773,10 +798,14 @@ def validate_body(payload):
 
         seen_ids.add(observation_id)
 
-        if payload["version"] == 3:
+        if payload["version"] >= 3:
             require_v3_timestamp(observation["observed_at"], f"{label}.observed_at")
         else:
             require_nonempty_string(observation["observed_at"], f"{label}.observed_at")
+        if payload["version"] == 4:
+            validate_note(observation["note"], f"{label}.note")
+            if observation["session_id"] is not None:
+                require_nonempty_string(observation["session_id"], f"{label}.session_id")
 
         present_metrics = (
             set(observation.keys())
@@ -816,9 +845,9 @@ def require_supported_schema(connection):
 
     # CONTRACT: v9 owns explicit max_results; earlier supported schemas remain
     # readable for legacy artifacts and are never made to fake that table.
-    if version not in (5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18):
+    if version not in (5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19):
         raise ImportFailure(
-            f"base desktop schema v5 à v18 attendue, version trouvée: {version}"
+            f"base desktop schema v5 à v19 attendue, version trouvée: {version}"
         )
 
 
@@ -1268,6 +1297,68 @@ def occurrence_plan_values(item):
     )
 
 
+def apply_note(connection, owner_kind, owner_id, payload, table, where, arguments):
+    """Apply one direct causal child without using wall-clock ordering."""
+    if payload is None:
+        retained = connection.execute(
+            "SELECT r.note FROM sync_note_state s JOIN sync_note_revisions r "
+            "USING(owner_kind,owner_id,revision_id) WHERE s.owner_kind=? AND s.owner_id=?",
+            (owner_kind, owner_id)).fetchone()
+        if retained is not None:
+            connection.execute(f"UPDATE {table} SET notes=? WHERE {where}",
+                               (retained[0], *arguments))
+        return "unchanged"
+    current = connection.execute(
+        "SELECT revision_id FROM sync_note_state WHERE owner_kind=? AND owner_id=?",
+        (owner_kind, owner_id)).fetchone()
+    existing = connection.execute(
+        "SELECT parent_revision_id,note FROM sync_note_revisions "
+        "WHERE owner_kind=? AND owner_id=? AND revision_id=?",
+        (owner_kind, owner_id, payload["revision_id"])).fetchone()
+    if existing is not None:
+        if tuple(existing) != (payload["parent_revision_id"], payload["value"]):
+            raise ImportFailure(f"révision de note contradictoire: {owner_kind}/{owner_id}")
+        if current is not None and current[0] == payload["revision_id"]:
+            return "unchanged"
+        raise ImportFailure(f"révision de note obsolète: {owner_kind}/{owner_id}")
+    if current is not None and payload["parent_revision_id"] != current[0]:
+        raise ImportFailure(f"conflit causal de note: {owner_kind}/{owner_id}")
+    if current is None and payload["parent_revision_id"] is not None:
+        raise ImportFailure(f"parent de note inconnu: {owner_kind}/{owner_id}")
+    connection.execute(
+        "INSERT INTO sync_note_revisions(owner_kind,owner_id,revision_id,parent_revision_id,note) "
+        "VALUES(?,?,?,?,?)", (owner_kind, owner_id, payload["revision_id"],
+                              payload["parent_revision_id"], payload["value"]))
+    connection.execute(
+        "INSERT INTO sync_note_state(owner_kind,owner_id,revision_id) VALUES(?,?,?) "
+        "ON CONFLICT(owner_kind,owner_id) DO UPDATE SET revision_id=excluded.revision_id",
+        (owner_kind, owner_id, payload["revision_id"]))
+    connection.execute(f"UPDATE {table} SET notes=? WHERE {where}",
+                       (payload["value"], *arguments))
+    return "applied"
+
+
+def note_matches(connection, owner_kind, owner_id, payload):
+    row = connection.execute(
+        "SELECT r.revision_id,r.parent_revision_id,r.note FROM sync_note_state s "
+        "JOIN sync_note_revisions r USING(owner_kind,owner_id,revision_id) "
+        "WHERE s.owner_kind=? AND s.owner_id=?", (owner_kind, owner_id)).fetchone()
+    if payload is None:
+        return row is None
+    return row is not None and tuple(row) == (
+        payload["revision_id"], payload["parent_revision_id"], payload["value"])
+
+
+def v4_session_metadata_matches(connection, session_row_id, session):
+    header = connection.execute(
+        "SELECT ended_at FROM sessions WHERE id=?", (session_row_id,)).fetchone()
+    if header is None or header[0] != session["ended_at"] or not note_matches(
+            connection, "session", session["session_id"], session["note"]):
+        return False
+    return all(note_matches(connection, "occurrence", item["entry_id"], item["note"])
+               for item in session["exercises"])
+
+
 def import_set_session_exercise(
     connection,
     session_row_id,
@@ -1348,6 +1439,7 @@ def import_set_session_exercise(
                 duration, set_item.get("weight_kg"),
             ),
         )
+    return session_exercise_row_id
 
 def import_continuous_session_exercise(
     connection,
@@ -1413,6 +1505,7 @@ def import_continuous_session_exercise(
             continuous.get("distance_km"),
         ),
     )
+    return cursor.lastrowid
 
 
 def import_max_session_exercise(
@@ -1452,6 +1545,7 @@ def import_max_session_exercise(
         "VALUES(?,?);",
         (cursor.lastrowid, item["max_weight_kg"]),
     )
+    return cursor.lastrowid
 
 
 def import_sessions(
@@ -1491,12 +1585,12 @@ def import_sessions(
             current = [(row[0], row[1]) for row in rows]
             legacy = all(value[0].startswith("sxe_legacy_") or value[0].startswith("sxe_v1_") for value in current)
             header = connection.execute(
-                "SELECT started_at,session_type FROM sessions WHERE id=?;",
+                "SELECT started_at,session_type,ended_at FROM sessions WHERE id=?;",
                 (session_row_id,),
             ).fetchone()
             resumable_max = (
                 header is not None
-                and tuple(header) == (session["started_at"], "max_test")
+                and tuple(header[:2]) == (session["started_at"], "max_test")
                 and session["session_type"] == "max_test"
                 and len(incoming) >= len(current)
                 and all(current[index] == incoming[index]
@@ -1504,9 +1598,9 @@ def import_sessions(
             )
             incoming_by_id = dict(incoming)
             correction = (
-                payload["version"] == 3
+                payload["version"] >= 3
                 and header is not None
-                and tuple(header) == (session["started_at"], session["session_type"])
+                and tuple(header[:2]) == (session["started_at"], session["session_type"])
                 and len(incoming_by_id) == len(incoming)
                 and all(entry_id not in dict(current)
                         or dict(current)[entry_id] == exercise_id
@@ -1524,7 +1618,8 @@ def import_sessions(
                 session_row_id,
                 session,
                 exercise_mapping,
-            ):
+            ) and (payload["version"] != 4 or v4_session_metadata_matches(
+                    connection, session_row_id, session)):
                 report["sessions_skipped"] += 1
                 continue
             if not legacy and not (resumable_max or correction):
@@ -1563,17 +1658,30 @@ def import_sessions(
                     ended_at,
                     session_type,
                     notes
-                ) VALUES(?, ?, NULL, ?, NULL);
+                ) VALUES(?, ?, ?, ?, NULL);
                 """,
                 (
                     session["session_id"],
                     session["started_at"],
+                    session.get("ended_at"),
                     session["session_type"],
                 ),
             )
 
             session_row_id = cursor.lastrowid
 
+        if payload["version"] == 4:
+            current_end = connection.execute(
+                "SELECT ended_at FROM sessions WHERE id=?", (session_row_id,)).fetchone()[0]
+            incoming_end = session["ended_at"]
+            if current_end is None and incoming_end is not None:
+                connection.execute("UPDATE sessions SET ended_at=? WHERE id=?",
+                                   (incoming_end, session_row_id))
+            elif current_end is not None and incoming_end is not None and current_end != incoming_end:
+                raise ImportFailure("conflit ended_at pour " + session["session_id"])
+            apply_note(connection, "session", session["session_id"], session["note"],
+                       "sessions", "id=?", (session_row_id,))
+
         for position, item in enumerate(
             session["exercises"]
         ):
@@ -1616,7 +1724,7 @@ def import_sessions(
             )
 
             if "max_weight_kg" in item:
-                import_max_session_exercise(
+                occurrence_row_id = import_max_session_exercise(
                     connection,
                     session_row_id,
                     position,
@@ -1624,7 +1732,7 @@ def import_sessions(
                     row_id,
                 )
             elif item["recording_mode"] == "continuous":
-                import_continuous_session_exercise(
+                occurrence_row_id = import_continuous_session_exercise(
                     connection,
                     session_row_id,
                     position,
@@ -1632,13 +1740,16 @@ def import_sessions(
                     row_id,
                 )
             else:
-                import_set_session_exercise(
+                occurrence_row_id = import_set_session_exercise(
                     connection,
                     session_row_id,
                     position,
                     item,
                     row_id,
                 )
+            if payload["version"] == 4:
+                apply_note(connection, "occurrence", item["entry_id"], item["note"],
+                           "session_exercises", "id=?", (occurrence_row_id,))
 
         retained_ids = {item["entry_id"] for item in session["exercises"]}
         for entry_id, feedback_id, observed_at, raw_text in retained_feedback:
@@ -1784,13 +1895,29 @@ def import_body(
     for observation in payload[
         "body_observations"
     ]:
+        parent_row_id = None
+        if payload["version"] == 4 and observation["session_id"] is not None:
+            parent = connection.execute(
+                "SELECT id FROM sessions WHERE session_id=?", (observation["session_id"],)).fetchone()
+            if parent is None:
+                raise ImportFailure("parent de l'observation introuvable: " + observation["session_id"])
+            parent_row_id = parent[0]
         existing = connection.execute(
-            f"SELECT observed_at,{','.join(metric_order)} FROM body_observations WHERE observation_id=?",
+            f"SELECT id,observed_at,{','.join(metric_order)},session_row_id FROM body_observations WHERE observation_id=?",
             (observation["observation_id"],)).fetchone()
         if existing is not None:
             expected = [observation["observed_at"]] + [observation.get(metric) for metric in metric_order]
-            if list(existing) != expected:
+            if list(existing[1:1 + len(expected)]) != expected:
                 raise ImportFailure("conflit observation corporelle: " + observation["observation_id"])
+            if payload["version"] == 4:
+                current_parent = existing[-1]
+                if current_parent is None and parent_row_id is not None:
+                    connection.execute("UPDATE body_observations SET session_row_id=? WHERE id=?",
+                                       (parent_row_id, existing[0]))
+                elif current_parent is not None and parent_row_id is not None and current_parent != parent_row_id:
+                    raise ImportFailure("conflit parent observation: " + observation["observation_id"])
+                apply_note(connection, "observation", observation["observation_id"], observation["note"],
+                           "body_observations", "id=?", (existing[0],))
             report["body_skipped"] += 1
             continue
 
@@ -1809,6 +1936,16 @@ def import_body(
             values,
         )
 
+        if payload["version"] == 4:
+            row_id = connection.execute(
+                "SELECT id FROM body_observations WHERE observation_id=?",
+                (observation["observation_id"],)).fetchone()[0]
+            if parent_row_id is not None:
+                connection.execute("UPDATE body_observations SET session_row_id=? WHERE id=?",
+                                   (parent_row_id, row_id))
+            apply_note(connection, "observation", observation["observation_id"], observation["note"],
+                       "body_observations", "id=?", (row_id,))
+
         report["body_imported"] += 1
 
 
@@ -1850,6 +1987,8 @@ def run_import(
         )
 
         schema_version = connection.execute("PRAGMA user_version;").fetchone()[0]
+        if payload["version"] == 4 and schema_version != 19:
+            raise ImportFailure("mobile V4 exige le schéma desktop v19")
         has_explicit_max = any(
             "max_weight_kg" in entry
             for session in payload["sessions"]

tools/import_training_feedback.py

@@ -51,7 +51,7 @@ def main():
     roots_added=roots_skipped=revisions_added=revisions_skipped=0
     try:
       db.execute("PRAGMA foreign_keys=ON")
-      if db.execute("PRAGMA user_version").fetchone()[0] not in (15,16,17,18):fail("schema desktop v15-v18 requis")
+      if db.execute("PRAGMA user_version").fetchone()[0] not in (15,16,17,18,19):fail("schema desktop v15-v19 requis")
       db.execute("BEGIN IMMEDIATE");seen=set()
       for index,item in enumerate(root["exercise_feedback"]):
         keys={"feedback_id","session_id","entry_id","exercise_id","observed_at","raw_text"} if root["version"]==1 else {"feedback_id","session_id","entry_id","exercise_id","observed_at","revisions"}

tui/include/trainlog/database.h

@@ -16,7 +16,7 @@
 #include "trainlog/model.h"
 #include "trainlog/status.h"
 
-#define TRAINLOG_DATABASE_SCHEMA_VERSION 18
+#define TRAINLOG_DATABASE_SCHEMA_VERSION 19
 
 typedef struct TrainlogDatabase TrainlogDatabase;
 

tui/src/database.c

@@ -804,6 +804,34 @@ static const char *const MIGRATE_V17_TO_V18_SQL =
     "BEGIN SELECT RAISE(ABORT,'AI draft import identity is permanent');END;"
     "PRAGMA user_version=18;COMMIT;";
 
+/* WHY: V3 cannot carry lifecycle-owned notes, observation parents, or
+ * resumable execution drafts. CONTRACT: v19 is additive and provides only the
+ * durable state required by the explicitly selected V4/lifecycle codecs.
+ * INVARIANT: completed history, AI proposals, feedback, aliases, and every
+ * existing stable identity remain untouched; NULL legacy notes keep unknown
+ * ancestry and no timestamp or causal parent is fabricated. */
+static const char *const MIGRATE_V18_TO_V19_SQL =
+    "BEGIN IMMEDIATE;"
+    "CREATE TABLE sync_note_revisions("
+    "owner_kind TEXT NOT NULL CHECK(owner_kind IN('session','occurrence','observation')),"
+    "owner_id TEXT NOT NULL,revision_id TEXT NOT NULL,parent_revision_id TEXT,note TEXT,"
+    "CHECK(note IS NULL OR length(CAST(note AS BLOB))<=4096),"
+    "PRIMARY KEY(owner_kind,owner_id,revision_id));"
+    "CREATE TABLE sync_note_state("
+    "owner_kind TEXT NOT NULL,owner_id TEXT NOT NULL,revision_id TEXT NOT NULL,"
+    "PRIMARY KEY(owner_kind,owner_id),"
+    "FOREIGN KEY(owner_kind,owner_id,revision_id) REFERENCES sync_note_revisions(owner_kind,owner_id,revision_id));"
+    "CREATE TABLE execution_drafts("
+    "session_id TEXT PRIMARY KEY,session_type TEXT NOT NULL CHECK(session_type IN('training','max_test')),"
+    "source_session_id TEXT,started_at TEXT,revision_id TEXT NOT NULL,parent_revision_id TEXT,"
+    "state TEXT NOT NULL CHECK(state IN('active','pending')),payload_json TEXT NOT NULL);"
+    "CREATE TABLE execution_draft_revisions("
+    "session_id TEXT NOT NULL,revision_id TEXT NOT NULL,parent_revision_id TEXT,payload_json TEXT NOT NULL,"
+    "PRIMARY KEY(session_id,revision_id));"
+    "CREATE TABLE execution_draft_finalizations("
+    "session_id TEXT PRIMARY KEY,final_revision_id TEXT NOT NULL,finalized_at TEXT NOT NULL);"
+    "PRAGMA user_version=19;COMMIT;";
+
 /* CONTRACT: v18 was already deployed before bounded draft publication was
  * added. Keep its version number and add only the nullable lifecycle cursor;
  * NULL deliberately means retryable/not yet published. */
@@ -1496,7 +1524,8 @@ static TrainlogStatus initialize_or_validate_schema(
     } else if (version == 10) {
         status = TRAINLOG_STATUS_OK;
     } else if (version == 11 || version == 12 || version == 13 || version == 14 ||
-               version == 15 || version == 16 || version == 17 || version == 18) {
+               version == 15 || version == 16 || version == 17 || version == 18 ||
+               version == 19) {
         status = TRAINLOG_STATUS_OK;
     } else {
         if (version == 1) {
@@ -1660,6 +1689,9 @@ static TrainlogStatus initialize_or_validate_schema(
     if (status == TRAINLOG_STATUS_OK && version < 18) {
         status = execute_sql(database, MIGRATE_V17_TO_V18_SQL);
     }
+    if (status == TRAINLOG_STATUS_OK && version < 19) {
+        status = execute_sql(database, MIGRATE_V18_TO_V19_SQL);
+    }
     if (status == TRAINLOG_STATUS_OK) {
         status = ensure_v18_ai_draft_publication_state(database);
     }
@@ -1675,7 +1707,7 @@ static TrainlogStatus initialize_or_validate_schema(
         set_open_diagnostic(
             output_diagnostic,
             output_diagnostic_capacity,
-            version == 0 ? "create schema v18" : "migrate database to schema v18",
+            version == 0 ? "create schema v19" : "migrate database to schema v19",
             database->connection,
             SQLITE_ERROR
         );

tui/tests/test_body_zones.c

@@ -204,7 +204,8 @@ static int migration_preserves_identity_and_history(void)
     CHECK(sqlite3_close(raw) == SQLITE_OK);
     raw = NULL;
     CHECK(trainlog_database_open(path, &database) == TRAINLOG_STATUS_OK);
-    CHECK(trainlog_database_schema_version(database, &version) == TRAINLOG_STATUS_OK && version == 18);
+    CHECK(trainlog_database_schema_version(database, &version) == TRAINLOG_STATUS_OK &&
+          version == TRAINLOG_DATABASE_SCHEMA_VERSION);
     CHECK(trainlog_database_list_exercise_body_zones(database,
         "ex_b432623f-bfe9-4daf-a653-60ec7fdffbde", relations, 4U, &count) == TRAINLOG_STATUS_OK);
     CHECK(count == 2U);

tui/tests/test_schema_v16_migration.c

@@ -84,9 +84,14 @@ static bool run_test(void)
         (void)fprintf(stderr, "open diagnostic: %s\n", diagnostic);
         CHECK(false);
     }
-    CHECK(trainlog_database_schema_version(db, &version) == TRAINLOG_STATUS_OK && version == 18);
+    CHECK(trainlog_database_schema_version(db, &version) == TRAINLOG_STATUS_OK &&
+          version == TRAINLOG_DATABASE_SCHEMA_VERSION);
     trainlog_database_close(db); db = NULL;
     CHECK(sqlite3_open_v2(path, &raw, SQLITE_OPEN_READWRITE, NULL) == SQLITE_OK);
+    CHECK(scalar(raw, "SELECT user_version FROM pragma_user_version", TRAINLOG_DATABASE_SCHEMA_VERSION));
+    CHECK(scalar(raw, "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN('sync_note_revisions','sync_note_state','execution_drafts','execution_draft_revisions','execution_draft_finalizations')", 5));
+    CHECK(scalar(raw, "SELECT COUNT(*) FROM execution_drafts", 0));
+    CHECK(scalar(raw, "SELECT COUNT(*) FROM execution_draft_finalizations", 0));
     CHECK(scalar(raw, "SELECT COUNT(*) FROM exercises e LEFT JOIN exercise_profile_state s ON s.exercise_row_id=e.id WHERE s.revision_id IS NULL OR s.revision_id<>'pr_legacy_v1' OR s.legacy_seed<>1", 0));
     CHECK(scalar(raw, "SELECT COUNT(*) FROM exercises e LEFT JOIN exercise_profile_revisions r ON r.exercise_row_id=e.id AND r.revision_id='pr_legacy_v1' AND r.legacy_seed=1 WHERE r.revision_id IS NULL", 0));
     CHECK(scalar(raw, "SELECT COUNT(*) FROM pragma_table_info('session_exercises') WHERE name='tracking_mode' AND \"notnull\"=1", 1));

tui/tests/test_schema_v7_migration.c

@@ -134,8 +134,8 @@ static bool test_v7_migration_sqlite_failure_has_diagnostic(void)
     CHECK(trainlog_database_open_with_diagnostic(path, &database, diagnostic,
         sizeof(diagnostic)) == TRAINLOG_STATUS_DATABASE_ERROR);
     CHECK(database == NULL);
-    CHECK(strncmp(diagnostic, "migrate database to schema v18: SQLite rc=",
-        strlen("migrate database to schema v18: SQLite rc=")) == 0);
+    CHECK(strncmp(diagnostic, "migrate database to schema v19: SQLite rc=",
+        strlen("migrate database to schema v19: SQLite rc=")) == 0);
     CHECK(strstr(diagnostic, "extended_rc=") != NULL);
     CHECK(strstr(diagnostic, "custom_equipment") != NULL);
     CHECK(strstr(diagnostic, "already exists") != NULL);
@@ -154,12 +154,12 @@ static bool test_newer_schema_has_application_diagnostic(void)
     CHECK(fd >= 0);
     CHECK(close(fd) == 0);
     CHECK(sqlite3_open(path, &raw) == SQLITE_OK);
-    CHECK(sqlite3_exec(raw, "PRAGMA user_version=19;", NULL, NULL, NULL) == SQLITE_OK);
+    CHECK(sqlite3_exec(raw, "PRAGMA user_version=20;", NULL, NULL, NULL) == SQLITE_OK);
     CHECK(sqlite3_close(raw) == SQLITE_OK);
     CHECK(trainlog_database_open_with_diagnostic(path, &database, diagnostic,
         sizeof(diagnostic)) == TRAINLOG_STATUS_SCHEMA_UNSUPPORTED);
     CHECK(database == NULL);
-    CHECK(strstr(diagnostic, "schema version 19 is newer") != NULL);
+    CHECK(strstr(diagnostic, "schema version 20 is newer") != NULL);
     CHECK(strstr(diagnostic, "SQLite") == NULL);
     CHECK(unlink(path) == 0);
     return true;

tui/tests/test_schema_v9_migration.c

@@ -151,7 +151,7 @@ static bool test_v9_to_v10_failure_rolls_back(void)
     CHECK(trainlog_database_open_with_diagnostic(path, &database, diagnostic,
         sizeof(diagnostic)) == TRAINLOG_STATUS_DATABASE_ERROR);
     CHECK(database == NULL);
-    CHECK(strstr(diagnostic, "migrate database to schema v18") != NULL);
+    CHECK(strstr(diagnostic, "migrate database to schema v19") != NULL);
     CHECK(strstr(diagnostic, "performed_sets_v9") != NULL);
 
     CHECK(sqlite3_open(path, &raw) == SQLITE_OK);