fix(session): tolerate catalog churn during memory admission

tests/test_session.py

@@ -8041,6 +8041,54 @@ class TestMemoryIndexSnapshotLifecycle:
 
         assert get_storage().get_memory_index_snapshot(session.ws_id) is not None
 
+    def test_first_capture_waits_through_catalog_prefix_churn(self, tmp_db):
+        """Cold MCP catalog callbacks may invalidate several capture plans."""
+        session = _make_registered_session(ws_id="catalog-churn-index", user_id="owner")
+        generation = session._claim_generation(principal_id="owner")
+        session._memory_index_admission_generation = generation
+        candidate = {
+            "content": '<memory-index format="1" entries="0" project_id=""></memory-index>',
+            "principal_id": "owner",
+            "project_id": "",
+            "project_name": "",
+            "entry_count": 0,
+            "char_count": 66,
+            "invalid_description_count": 0,
+        }
+        attempts = 0
+
+        def acquire(
+            _ws_id: str,
+            _principal_id: str,
+            *,
+            commit_context: Any,
+        ) -> dict[str, Any]:
+            nonlocal attempts
+            attempts += 1
+            if attempts <= 4:
+                session._invalidate_system_prefix()
+            with commit_context(candidate):
+                pass
+            return candidate
+
+        with (
+            patch(
+                "turnstone.core.session.acquire_memory_index_snapshot",
+                side_effect=acquire,
+            ),
+            patch("turnstone.core.session.time.sleep") as sleep,
+        ):
+            session._admit_memory_index_request(
+                session._primary_lane(),
+                my_generation=generation,
+                principal_id="owner",
+            )
+
+        assert attempts == 5
+        assert [item.args[0] for item in sleep.call_args_list] == [0.025, 0.05, 0.1, 0.2]
+        assert session._memory_index_snapshot == candidate
+        assert "<memory-index" in self._index_text(session)
+
     def test_raced_existing_snapshot_refusal_reports_truthful_remediation(self, tmp_db):
         from turnstone.core.personas import PersonaSnapshot
         from turnstone.core.storage import get_storage

turnstone/core/session.py

@@ -1996,6 +1996,19 @@ class _MemoryIndexPlanStaleError(RuntimeError):
         self.snapshot = snapshot
 
 
+# First-turn memory-index capture can overlap asynchronous MCP catalog
+# priming. Resource/prompt callbacks deliberately invalidate the cached
+# system prefix, which makes the in-flight candidate stale and rolls its
+# transaction back. A cold enterprise catalog can publish several of those
+# callbacks over a few seconds, so three immediate retries are not enough.
+# Back off outside both the storage transaction and the model-capacity lease;
+# this gives catalog publication a bounded window to settle without weakening
+# the prefix epoch fence.
+_MEMORY_INDEX_ADMISSION_RETRIES = 12
+_MEMORY_INDEX_ADMISSION_RETRY_BASE_DELAY = 0.025
+_MEMORY_INDEX_ADMISSION_RETRY_MAX_DELAY = 0.5
+
+
 # ``list_nodes`` reserves four top-level kwargs for control parameters
 # (filters / paging / output verbosity / liveness toggle).  Anything
 # else the model passes at the top level is treated as a flat filter
@@ -6916,7 +6929,7 @@ class ChatSession:
             return
         if snapshot is None:
             project_witness: dict[str, Any] | None = None
-            for _attempt in range(3):
+            for attempt in range(_MEMORY_INDEX_ADMISSION_RETRIES):
                 admission = self._plan_first_memory_index_admission(
                     lane=lane,
                     principal_id=principal_id,
@@ -6935,6 +6948,13 @@ class ChatSession:
                     break
                 except _MemoryIndexPlanStaleError as stale:
                     project_witness = stale.snapshot
+                    if attempt + 1 < _MEMORY_INDEX_ADMISSION_RETRIES:
+                        time.sleep(
+                            min(
+                                _MEMORY_INDEX_ADMISSION_RETRY_BASE_DELAY * (2**attempt),
+                                _MEMORY_INDEX_ADMISSION_RETRY_MAX_DELAY,
+                            )
+                        )
             else:
                 raise RuntimeError("memory context could not be refreshed safely")
             if snapshot is None: