Honor session tombstones in association companions

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_sync_generation_exchange.py

@@ -230,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/sync_generation_exchange.py

@@ -641,7 +641,13 @@ def consume_desktop(database: Path, directory: Path) -> dict:
             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)