Document history deletion private rollout

CHANGELOG.md

@@ -10,7 +10,11 @@
   revision-guarded, atomic and replay-safe. Retained Program-execution
   generations cannot resurrect a causally deleted session, and Web errors now
   distinguish stale, absent, invalid and internal failures without exposing
-  database details.
+  database details. The private rollout additionally fixed complete-generation
+  dominance for occurrence equipment companions and made bounded ACK recovery
+  prioritize recent terminal evidence. Two correlated direct-MTP exchanges,
+  component restart and exact replay of a retained pre-deletion generation
+  preserved the single tombstone without duplication or resurrection.
 
 - Added the Program execution flow. Android schema v25 can start or resume one
   Program session through the existing durable singleton, preserves stable

docs/current_state.md

@@ -83,7 +83,7 @@ parallel implementation of its rules.
 | Web Sessions deletion and Programs V1 | `TRAINLOG_WEB_SESSIONS_DELETE_AND_PROGRAMS_V1=PASS` (private grouped rollout validated) |
 | Programs presentation, Android projection, and deletion | `TRAINLOG_PROGRAMS_PRESENTATION_ANDROID_DELETE_V1=PASS` (private coordinated deployment and restart/replay validated) |
 | Program execution flow | `TRAINLOG_PROGRAM_EXECUTION_FLOW_V1=PASS` (private coordinated deployment, real execution, correlated ACK, restart, replay, and deletion validated) |
-| Web history deletion incident | `TRAINLOG_HISTORY_DELETE_HTTP500_FIX_V1=IMPLEMENTED` (Program-origin completed sessions use causal deletion and terminal execution provenance) |
+| Web history deletion incident | `TRAINLOG_HISTORY_DELETE_HTTP500_FIX_V1=PASS` (private rollout, correlated full-generation exchanges, restart, retained-generation replay, and non-resurrection validated) |
 | Sync orchestrator/report V1 | `TRAINLOG_SYNC_ORCHESTRATOR_REPORT_V1=PASS/FROZEN` |
 | Web sync API V1 | `TRAINLOG_WEB_SYNC_API_V1=PASS/FROZEN` |
 | Web sync button V1 | `TRAINLOG_WEB_SYNC_BUTTON_V1=PASS/FROZEN` |

docs/reviews/history_delete_http500_fix_v1_evidence.md

@@ -44,3 +44,90 @@ exposing SQL or paths.
 The rollout-specific backup hashes, deployed commit, causal operation,
 generation/ACK identifiers and restart/non-resurrection results are appended
 after the controlled private rollout. No public tag or release is created.
+
+During the first post-delete hardware exchange, the desktop correctly rejected
+Android's still-live snapshot with `causal protection refuses legacy/live
+replay`. This exposed a missing full-envelope dominance rule: Android publishes
+its half before consuming the desktop tombstone in the same conversation. The
+production full-generation consumer now filters only live facts already
+dominated by causal state applied from that complete envelope; standalone
+snapshot imports retain their strict rejection. A regression exercises this
+ordering and proves the tombstone and absent session remain unchanged.
+The recovery envelope also republishes both consumed and rejected correlated
+ACKs. This lets Android close an interrupted generation with the real durable
+desktop rejection evidence and reclaim normal bounded capacity; no generation,
+ACK, tombstone or archive is manually purged or synthesized.
+
+## Controlled private rollout
+
+The rollout source was advanced through bounded corrective commits to
+`a8c50a9512d8754477a2491b73c0f9e0f92263bf`. The coordinated desktop archive is
+`trainlog-0.1.2-a8c50a9-desktop-v28.tar.gz`, SHA-256
+`e9e08b317fb97807987b8e027a072cc8235bd9ee6dff7d393df4acb8f04c8a46`.
+Android code and schema were unchanged; the installed private application
+remained `com.labfytools.trainlog`, versionCode 17, versionName 0.1.2, schema
+25.
+
+Before deployment, the real desktop v27 database was backed up as
+`trainlog-desktop-v27.db`, SHA-256
+`ffa4cd5c5fc832b4054ee6543ff89b74be7148b19ffb9af718aa0dfe787ab20a`.
+The real Android product backup was
+`trainlog-android-v25.tlbackup.zip`, SHA-256
+`8d198dc5fb6ca5c410029342f0e3074d288dac844ce686e3a70782bdad9e8bce`.
+A populated copy migration changed only `PRAGMA user_version` from 27 to 28:
+all 47 business-table counts and canonical typed hashes were identical,
+`integrity_check` was `ok`, and `foreign_key_check` was empty. The production
+migration reached the same integrity state.
+
+The real deletion used the Web HTTP command with the session's current causal
+revision. It removed only
+`se_6bb66f59-a6fb-4759-a75d-c1a79ea107e8`, created operation
+`del_af8a82de-2825-45af-9f62-ec94cc61a556`, and retained exactly one deleted
+causal state. Its Program execution provenance moved from `completed` to
+terminal `deleted` while preserving the original observation timestamp. The
+session detail then returned the bounded absent response and reload did not
+restore it.
+
+The first hardware exchange exposed two real recovery defects before a final
+PASS. Generation `gen_42f8d681-373e-4f07-b08d-792a1e8e516d` was rejected
+because the live Android snapshot reached the importer after its tombstone;
+the subsequent generation `gen_c9d1fda8-8375-4d89-8c29-a5ae120c6c7d` proved
+that the equipment-association companion required the same dominance rule.
+Later retries reported `peer_capacity_exhausted`: a product backup showed four
+older immutable Android generations still waiting for ACK. Their retained
+manifests were processed by the production consumer and received genuine
+correlated stale-lineage rejection ACKs. The recovery envelope then accepted
+that exact terminal evidence. No ACK, tombstone, generation, archive or ledger
+row was fabricated, deleted or edited manually.
+
+The final post-correction exchange completed as run
+`sy_b343619f-a1d2-4bea-a0af-c57aa5792aca`: Android generation
+`gen_b2290cad-0554-46f9-bc30-0fbd0c8190e8` was consumed with
+`sqlite-commit-full`, desktop generation
+`gen_21d7efa0-bd97-4843-a0d6-111e94d669b4` was acknowledged, no capability was
+missing, and `sessions_reconciled` was zero. The second exchange completed as
+run `sy_58b38f82-a70c-48c3-834c-81436c687c8c`, with generations
+`gen_65e4833c-5383-4374-b37f-d8add8a106e8` and
+`gen_76e15f8a-1258-49f1-ac0e-b04fe9e6a8a5`, again with no missing capability
+and zero reconciled sessions.
+
+After both stores and the Web component were reopened, the Android product
+backup `trainlog-android-postrestart-20260919.tlbackup.zip` had SHA-256
+`7ae8f88c60bd44355c19c89e7996d8c71ec178c75aabff0979065a5a70936ed5`,
+schema 25, `integrity_check=ok`, an empty foreign-key check, no disposable
+session, and exactly one matching causal state and operation. Desktop remained
+schema 28 with the same integrity results and terminal provenance. Canonical
+hashes for every other session, occurrence, performed set, continuous result
+and body observation were identical to both pre-rollout backups.
+
+Finally, the retained pre-deletion generation
+`gen_965fdf80-4412-41fb-9a46-28744a3e115b` was replayed through the production
+consumer. It returned its byte-identical stored ACK; the before/after tuple was
+unchanged (`target session=0`, `target operation=1`, `deleted provenance=1`,
+`all sessions=9`). Real Firefox against the deployed C server passed at
+1440x1000 and exactly 390x844 with no horizontal overflow. Evidence screenshots
+are retained privately under
+`~/.local/state/trainlog/history-delete-http500-fix-v1-web-evidence/`; their
+SHA-256 values are respectively
+`9c9fd42aa58b0818488ef9f8b3412abfb56e96ea3f9d30c8da2fcfe3da8c10f1` and
+`70838305fe6d17485f3d5d80cf60e2639baf4ed87d148f4b55e4b6e8c25fdbca`.

docs/sync_exchange.md

@@ -830,13 +830,21 @@ than fabricating an ACK or discarding evidence.
 Before Android captures for a negotiated `generation-archive-v1` conversation,
 the desktop consumer publishes
 `desktop-archive-acknowledgements-v1.json`. Its strictly correlated envelope
-contains at most 32 exact consumed ACK documents already retained in the
-desktop consumption ledger. This repairs the evidence gap left by older
+contains at most 32 exact terminal consumed or rejected ACK documents already
+retained in the desktop consumption ledger. The bounded window selects the
+newest terminal evidence first, because older acknowledged generations are
+normally already archived and must not starve interrupted recent
+conversations. This repairs the evidence gap left by older
 Android producers that advanced generation status without retaining the
 received ACK. Android revalidates every document against its immutable local
-generation and manifest before storing it; missing, mismatched, rejected, or
-ambiguous evidence cannot make a generation archivable. The object is a
-bounded coordination artifact, not a new ACK source.
+generation and manifest before storing it. A consumed ACK makes its exact
+generation archive-eligible; a rejected ACK closes the exact producer row but
+does not make it an acknowledged archive candidate. Missing, mismatched, or
+ambiguous evidence cannot advance any generation. Rejected ACKs created before
+the cross-platform diagnostic canonicalization fix remain durable desktop
+evidence but are omitted from the recovery envelope when Android cannot verify
+their digest. The object is a bounded coordination artifact, not a new ACK
+source.
 
 If Android still cannot admit a generation, it durably publishes the
 run-correlated `android-generation-error-v1.json` with
@@ -852,6 +860,10 @@ helpers and the Android repository's outer transaction apply every business
 row, causal state and the consumption/ACK record together. A late semantic
 failure rolls back earlier writes and commits a separate bounded `rejected`
 ACK; missing/incomplete transport input remains retryable and produces no ACK.
+Within a complete causal envelope, a history occurrence and its equipment
+association companion are both dominated when their owning session already has
+a durable deletion state. Standalone history and association imports remain
+strict and still reject unknown sessions or occurrences.
 
 Lineage is scoped to the producer/consumer pair. Exact replay returns the
 stored result; an accepted successor must name the last consumed generation as

docs/tests.md

@@ -170,6 +170,15 @@ npm run build
 Release-oriented Meson builds must use `-Dweb=enabled`; Meson never provisions
 or downloads npm dependencies.
 
+The history-deletion regression suite also exercises a real Program-origin
+completed session through the Core and HTTP command paths, including stale
+revision, cancellation, injected transaction failure, exact request replay and
+terminal execution provenance. Full-generation tests apply causal deletions
+before older history and equipment-association companions, prove that
+standalone imports remain strict, recover recent consumed and rejected ACKs
+without letting an older bounded window starve active producer capacity, and
+replay an already consumed generation without mutation.
+
 `WEB_DASHBOARD_DATA_CONTRACT_V1` adds a standalone bounded Core test for empty
 history, explicit unavailable domains, two observable sessions on one local
 day, an unfinished session, continuous activity, MAX-only sessions, primary

tests/test_equipment_associations_exchange.py

@@ -1,4 +1,5 @@
 """Exercise the real desktop companion exporter/importer on temporary DBs."""
+import importlib.util
 import json
 import sqlite3
 import subprocess
@@ -7,6 +8,14 @@ import tempfile
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "tools"))
+
+SPEC = importlib.util.spec_from_file_location(
+    "import_equipment_associations",
+    ROOT / "tools/import_equipment_associations.py",
+)
+ASSOCIATIONS = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(ASSOCIATIONS)
 SOURCE_ID = "ex_43c7375f-934c-4650-930c-45807d2f2929"
 CANONICAL_ID = "ex_2d488c08-194c-4051-a3c9-34471646c1d3"
 
@@ -236,6 +245,67 @@ def main():
         )
         assert exported.returncode == 0, exported.stdout + exported.stderr
         assert json.loads(custom_artifact.read_text())['associations'][0]['equipment_id'] == 'custom_rack'
+
+        # A complete causal envelope may contain an older association snapshot
+        # for a session whose tombstone was applied earlier in the transaction.
+        # The same companion remains invalid as a standalone import.
+        deleted_session = 'se_deleted'
+        deleted_entry = 'sxe_deleted'
+        dominated_payload = {
+            'format': 'trainlog-equipment-associations',
+            'version': 2,
+            'generated_at': '2026-01-01T00:00:00+00:00',
+            'associations': [{
+                'session_id': deleted_session,
+                'entry_id': deleted_entry,
+                'exercise_id': 'ex_fixture',
+                'state': 'set',
+                'equipment_id': 'leg_press',
+            }],
+        }
+        causal_db = sqlite3.connect(':memory:')
+        causal_db.executescript("""
+            CREATE TABLE exercises(id INTEGER PRIMARY KEY, exercise_id TEXT UNIQUE);
+            CREATE TABLE sessions(id INTEGER PRIMARY KEY, session_id TEXT UNIQUE);
+            CREATE TABLE session_exercises(
+                id INTEGER PRIMARY KEY,
+                entry_id TEXT UNIQUE,
+                session_row_id INTEGER,
+                exercise_row_id INTEGER,
+                equipment_id TEXT
+            );
+            CREATE TABLE custom_equipment(equipment_id TEXT PRIMARY KEY);
+            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)
+            );
+        """)
+        causal_db.execute(
+            "INSERT INTO sync_causal_state VALUES('session',?,'del_fixture',1,'del_fixture')",
+            (deleted_session,),
+        )
+        try:
+            ASSOCIATIONS.apply_associations(
+                causal_db,
+                dominated_payload,
+                {'leg_press'},
+                {(deleted_session, deleted_entry): 'ex_fixture'},
+            )
+            raise AssertionError('standalone missing association was accepted')
+        except ValueError as error:
+            assert 'entrée séance inconnue' in str(error)
+        assert ASSOCIATIONS.apply_associations(
+            causal_db,
+            dominated_payload,
+            {'leg_press'},
+            {(deleted_session, deleted_entry): 'ex_fixture'},
+            complete_causal_envelope=True,
+        ) == 1
+        causal_db.close()
     print('PASS equipment association exchange')
 
 

tests/test_session_exchange_v3.py

@@ -11,6 +11,8 @@ from pathlib import Path
 ROOT = Path(__file__).resolve().parents[1]
 IMPORTER = ROOT / "tools/import_mobile_export.py"
 EXPORTER = ROOT / "tools/export_pc_mobile.py"
+sys.path.insert(0, str(ROOT / "tools"))
+import import_mobile_export
 from test_mobile_import_variable_sets import SCHEMA
 
 V11_SCHEMA = SCHEMA.replace("PRAGMA user_version=10;", "") + """
@@ -163,6 +165,29 @@ def main():
                 "SELECT target_id,current_revision_id,deleted FROM sync_causal_state "
                 "WHERE target_kind='session' ORDER BY target_id"
             ).fetchall() == revisions_before
+
+            # A full generation applies causal operations before its mobile
+            # snapshot. The older live fact is dominated rather than rejected
+            # or reinserted; standalone import keeps the stricter collision.
+            con.execute("DELETE FROM sessions WHERE session_id='se_v3'")
+            con.execute(
+                "UPDATE sync_causal_state SET current_revision_id='del_fixture',"
+                "deleted=1,operation_id='del_fixture' WHERE target_kind='session' "
+                "AND target_id='se_v3'"
+            )
+            report = import_mobile_export.apply_payload(
+                con,
+                import_mobile_export.load_payload(artifact),
+                complete_causal_envelope=True,
+            )
+            assert report["sessions_imported"] == 0
+            assert con.execute(
+                "SELECT COUNT(*) FROM sessions WHERE session_id='se_v3'"
+            ).fetchone()[0] == 0
+            assert tuple(con.execute(
+                "SELECT current_revision_id,deleted FROM sync_causal_state "
+                "WHERE target_kind='session' AND target_id='se_v3'"
+            ).fetchone()) == ("del_fixture", 1)
         exported = root / "pc-v3.json"
         run([sys.executable, str(EXPORTER), str(exported), "--database", str(db)])
         roundtrip = json.loads(exported.read_text())

tests/test_sync_error_classification.py

@@ -1,4 +1,5 @@
 import json
+import sqlite3
 import subprocess
 import sys
 import tempfile
@@ -15,6 +16,69 @@ import sync_peer_worker  # noqa: E402
 
 
 class SyncErrorClassificationTest(unittest.TestCase):
+    def test_recovery_envelope_includes_consumed_and_rejected_acknowledgements(self):
+        with sqlite3.connect(":memory:") as database:
+            database.execute(
+                "CREATE TABLE sync_consumed_generations("
+                "generation_id TEXT,producer_peer_id TEXT,consumer_peer_id TEXT,"
+                "consumed_at TEXT,result TEXT,ack_json TEXT)"
+            )
+            for generation, result, diagnostic in (
+                ("gen_consumed", "consumed", ""),
+                ("gen_rejected", "rejected", "bounded rejection"),
+                ("gen_legacy", "rejected", "legacy/path"),
+                ("gen_other", "consumed", ""),
+            ):
+                database.execute(
+                    "INSERT INTO sync_consumed_generations VALUES(?,?,?,?,?,?)",
+                    (generation, "android" if generation != "gen_other" else "other",
+                     "desktop", generation, result,
+                     json.dumps({"generation_id": generation, "result": result,
+                                 "diagnostic": diagnostic})),
+                )
+
+            acknowledgements = sync_peer_worker.recovery_acknowledgements(
+                database, "android", "desktop"
+            )
+
+        self.assertEqual(
+            [(value["generation_id"], value["result"]) for value in acknowledgements],
+            [("gen_rejected", "rejected"), ("gen_consumed", "consumed")],
+        )
+
+    def test_recovery_envelope_prioritizes_recent_terminal_evidence(self):
+        with sqlite3.connect(":memory:") as database:
+            database.execute(
+                "CREATE TABLE sync_consumed_generations("
+                "generation_id TEXT,producer_peer_id TEXT,consumer_peer_id TEXT,"
+                "consumed_at TEXT,result TEXT,ack_json TEXT)"
+            )
+            for index in range(40):
+                generation = f"gen_{index:02d}"
+                database.execute(
+                    "INSERT INTO sync_consumed_generations VALUES(?,?,?,?,?,?)",
+                    (
+                        generation,
+                        "android",
+                        "desktop",
+                        f"2026-09-18T12:{index:02d}:00Z",
+                        "consumed",
+                        json.dumps({
+                            "generation_id": generation,
+                            "result": "consumed",
+                            "diagnostic": "",
+                        }),
+                    ),
+                )
+
+            acknowledgements = sync_peer_worker.recovery_acknowledgements(
+                database, "android", "desktop"
+            )
+
+        self.assertEqual(32, len(acknowledgements))
+        self.assertEqual("gen_39", acknowledgements[0]["generation_id"])
+        self.assertEqual("gen_08", acknowledgements[-1]["generation_id"])
+
     def test_mtp_refreshes_a_retained_peer_advertisement_before_validation(self):
         expected_peer = "peer_00000000-0000-4000-8000-000000000001"
         with tempfile.TemporaryDirectory() as directory:

tests/test_sync_generation_exchange.py

@@ -210,7 +210,10 @@ class GenerationTest(unittest.TestCase):
             filename=name+".json";(directory/filename).write_text("{}")
             artifacts.append({"logical_name":name,"format":"test","version":1,"filename":f"generations/{GEN}/{filename}","size":2,"sha256":"0"*64,"required":True})
         manifest={"format":generation.MANIFEST,"version":1,"generation_id":GEN,"run_id":RUN,"producer":{"peer_id":PEER_A,"kind":"android"},"consumer_peer_id":PEER_B,"generated_at":"2026-09-17T10:00:00+02:00","parent_generation_id":None,"artifacts":artifacts}
-        def write_fact(db,_payload,_trace=False):db.execute("INSERT INTO facts VALUES('partial','must roll back')");return {}
+        def write_fact(db, _payload, _trace=False, complete_causal_envelope=False):
+            self.assertTrue(complete_causal_envelope)
+            db.execute("INSERT INTO facts VALUES('partial','must roll back')")
+            return {}
         with mock.patch.object(generation,"validate_published",return_value=(manifest,"a"*64)), \
              mock.patch.object(generation.causal_delete_exchange,"load",return_value={"operations":[]}), \
              mock.patch.object(generation.execution_draft_exchange,"load",return_value={}), \
@@ -227,11 +230,12 @@ class GenerationTest(unittest.TestCase):
              mock.patch.object(generation.import_equipment_definitions,"apply_definitions"), \
              mock.patch.object(generation.import_mobile_export,"apply_payload",side_effect=write_fact), \
              mock.patch.object(generation.import_exercise_aliases,"apply_aliases"), \
-             mock.patch.object(generation.import_equipment_associations,"apply_associations"), \
+             mock.patch.object(generation.import_equipment_associations,"apply_associations") as apply_associations, \
              mock.patch.object(generation.import_exercise_body_zones,"apply_body_zones"), \
              mock.patch.object(generation.import_training_feedback,"apply_feedback",side_effect=ValueError("late semantic failure")):
             rejected=generation.consume_desktop(self.database,directory)
             self.assertEqual("rejected",rejected["result"]);self.assertIn("late semantic",rejected["diagnostic"])
+            self.assertTrue(apply_associations.call_args.kwargs["complete_causal_envelope"])
         with closing(sqlite3.connect(self.database)) as db:
             self.assertEqual(0,db.execute("SELECT COUNT(*) FROM facts WHERE id='partial'").fetchone()[0])
             self.assertEqual(1,db.execute("SELECT COUNT(*) FROM sync_consumed_generations WHERE result='rejected'").fetchone()[0])

tools/import_equipment_associations.py

@@ -117,14 +117,33 @@ def load_mobile_occurrences(path):
     return occurrences
 
 
-def apply_associations(connection, payload, known, mobile_occurrences):
+def session_is_causally_deleted(connection, session_id):
+    """Return whether durable causal state dominates one historical session."""
+    row = connection.execute(
+        "SELECT deleted FROM sync_causal_state "
+        "WHERE target_kind='session' AND target_id=?",
+        (session_id,),
+    ).fetchone()
+    return row is not None and row[0] == 1
+
+
+def apply_associations(connection, payload, known, mobile_occurrences,
+                       complete_causal_envelope=False):
     if payload.get("format") != FORMAT or payload.get("version") != VERSION or set(payload) != {"format", "version", "generated_at", "associations"}:
         fail("extension équipement non supportée")
     known = set(known); known.update(row[0] for row in connection.execute("SELECT equipment_id FROM custom_equipment"))
     associations = validate_associations(payload["associations"], known)
     for session_id, entry_id, exercise_id, equipment_id in associations:
         exists = connection.execute("SELECT e.exercise_id,se.equipment_id FROM session_exercises se JOIN sessions s ON s.id=se.session_row_id JOIN exercises e ON e.id=se.exercise_row_id WHERE s.session_id=? AND se.entry_id=?", (session_id, entry_id)).fetchone()
-        if exists is None: fail(f"entrée séance inconnue: {session_id}/{entry_id}")
+        if exists is None:
+            # INVARIANT: a complete generation applies its causal operations
+            # before snapshot companions. An association owned by a deleted
+            # session is therefore an older live fact and cannot resurrect or
+            # invalidate that tombstone. Standalone imports remain strict.
+            if complete_causal_envelope and session_is_causally_deleted(
+                    connection, session_id):
+                continue
+            fail(f"entrée séance inconnue: {session_id}/{entry_id}")
         stored_canonical = canonical_exercise_id(connection, exists[0]); incoming_canonical = canonical_exercise_id(connection, exercise_id)
         if stored_canonical != incoming_canonical:
             proof = mobile_occurrences.get((session_id, entry_id))

tools/import_mobile_export.py

@@ -1960,7 +1960,30 @@ def import_body(
         report["body_imported"] += 1
 
 
-def apply_payload(connection, payload, trace_exercises=False):
+def causal_snapshot_view(payload, protected):
+    """Return the live subset of a snapshot after its causal envelope."""
+    filtered = dict(payload)
+    filtered["exercises"] = [
+        value for value in payload["exercises"]
+        if ("exercise", value["exercise_id"]) not in protected
+    ]
+    filtered["sessions"] = [
+        value for value in payload["sessions"]
+        if ("session", value["session_id"]) not in protected
+    ]
+    filtered["body_observations"] = []
+    for value in payload["body_observations"]:
+        if ("body_observation", value["observation_id"]) in protected:
+            continue
+        if payload["version"] == 4 and value["session_id"] is not None and \
+                ("session", value["session_id"]) in protected:
+            value = dict(value)
+            value["session_id"] = None
+        filtered["body_observations"].append(value)
+    return filtered
+
+
+def apply_payload(connection, payload, trace_exercises=False, complete_causal_envelope=False):
     """Apply a validated artifact without owning commit or rollback."""
     connection.row_factory = sqlite3.Row
     report = {
@@ -1988,9 +2011,17 @@ def apply_payload(connection, payload, trace_exercises=False):
                     {("body_observation", value["observation_id"]) for value in payload["body_observations"]} |
                     {("exercise", value["exercise_id"]) for value in payload["exercises"]})
         collision = protected & incoming
-        if collision:
+        if collision and not complete_causal_envelope:
             kind, identity = sorted(collision)[0]
             raise ImportFailure(f"causal protection refuses legacy/live replay: {kind}/{identity}")
+        if complete_causal_envelope:
+            # WHY: Android publishes its snapshot before consuming this run's
+            # desktop half, so that snapshot may still contain objects deleted
+            # by the causal artifact already applied above. CONTRACT: only the
+            # complete generation consumer may discard those dominated live
+            # facts. INVARIANT: the durable tombstone remains authoritative and
+            # ordinary standalone imports continue to reject the collision.
+            payload = causal_snapshot_view(payload, protected)
     has_explicit_max = any("max_weight_kg" in entry for session in payload["sessions"] for entry in session["exercises"])
     if has_explicit_max and schema_version < 9:
         raise ImportFailure("max_weight_kg exige le schéma desktop v9")

tools/sync_generation_exchange.py

@@ -566,6 +566,11 @@ def record_consumed(db: sqlite3.Connection, manifest: dict, manifest_digest: str
 
 
 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
+    # presentation text, so normalize that separator before hashing the ACK;
+    # identities and protocol fields remain untouched.
+    diagnostic = diagnostic.replace("/", ":")
     diagnostic = diagnostic.encode("utf-8")[:MAX_DIAGNOSTIC].decode("utf-8", "ignore")
     with closing(connect_database(database)) as db:
         require_schema(db); db.execute("BEGIN IMMEDIATE")
@@ -633,10 +638,16 @@ def consume_desktop(database: Path, directory: Path) -> dict:
                 causal_delete_exchange.apply_operation(db, operation)
             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)
+            import_mobile_export.apply_payload(db, history, complete_causal_envelope=True)
             import_exercise_profile_state.apply_profile_state(db, profile)
             import_exercise_aliases.apply_aliases(db, aliases)
-            import_equipment_associations.apply_associations(db, association_payload, reserved, mobile_occurrences)
+            import_equipment_associations.apply_associations(
+                db,
+                association_payload,
+                reserved,
+                mobile_occurrences,
+                complete_causal_envelope=True,
+            )
             import_exercise_body_zones.apply_body_zones(db, parsed_zones, {}, complete_causal_envelope=True)
             import_training_feedback.apply_feedback(db, feedback, complete_causal_envelope=True)
             execution_draft_exchange.import_document(db, drafts, own_transaction=False)

tools/sync_peer_worker.py

@@ -67,6 +67,29 @@ def publish_json(path: Path, value: dict) -> None:
     os.replace(temporary, path)
 
 
+def recovery_acknowledgements(db, producer_peer_id: str, consumer_peer_id: str) -> list[dict]:
+    """Load bounded terminal ACK evidence for interrupted peer conversations."""
+    rows = db.execute(
+        "SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? "
+        "AND consumer_peer_id=? AND result IN('consumed','rejected') "
+        "ORDER BY consumed_at DESC,generation_id DESC LIMIT 32",
+        (producer_peer_id, consumer_peer_id),
+    ).fetchall()
+    acknowledgements = [json.loads(row[0]) for row in rows]
+    # WHY: the bounded envelope must prioritize the newest interrupted
+    # conversations. Older acknowledged generations are normally already
+    # archived; selecting them first can starve the exact ACKs needed to free
+    # active producer capacity.
+    # ACKs created before rejection diagnostics were normalized may contain a
+    # solidus that Android's platform JSON canonicalizer hashes differently.
+    # They remain durable evidence but cannot close a peer row, so do not let
+    # one legacy object block recovery of later compatible conversations.
+    return [
+        value for value in acknowledgements
+        if value.get("result") != "rejected" or "/" not in value.get("diagnostic", "")
+    ]
+
+
 def run_adapter(
     adapter: Path,
     operation: str,
@@ -309,18 +332,14 @@ def main() -> int:
     request_path = args.transport_root / "request-v1.json"
     publish_json(request_path, request)
     with closing(generation.connect_database(args.database)) as db:
-        acknowledgement_rows = db.execute(
-            "SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? "
-            "AND consumer_peer_id=? AND result='consumed' ORDER BY consumed_at,generation_id LIMIT 32",
-            (peer["peer_id"], desktop_peer),
-        ).fetchall()
+        acknowledgements = recovery_acknowledgements(db, peer["peer_id"], desktop_peer)
     archive_acknowledgements = {
         "format": "trainlog-sync-archive-acknowledgements",
         "version": 1,
         "run_id": args.run_id,
         "android_peer_id": peer["peer_id"],
         "desktop_peer_id": desktop_peer,
-        "acknowledgements": [json.loads(row[0]) for row in acknowledgement_rows],
+        "acknowledgements": acknowledgements,
     }
     archive_ack_path = args.transport_root / "desktop-archive-acknowledgements-v1.json"
     publish_json(archive_ack_path, archive_acknowledgements)