mc-store: keep the materialized boundary when a reconnecting host echoes it

A host without a whole-message boundary of its own sends back the boundary the
module reported in its inventory. After compartments are published but not yet
folded, that boundary lags the newest compartment, and the seed was refused with
state_sync_seed_boundary_mismatch. An echo that is still a real compartment end
asks for no change, so the module keeps its boundary; an echo whose compartment
the host rewrote still falls through to validation and is refused.

Co-authored-by: Alfonso <alfonso-magic-context@users.noreply.github.com>

Author: ualtinok

crates/mc-store/src/context_boundaries.rs

@@ -301,6 +301,55 @@ mod tests {
     fn seed(store: &McStore) {
         store.with_context_conn_for_test(|conn| conn.execute_batch("INSERT INTO compartments(session_id, sequence, start_message, end_message, start_message_id, end_message_id, title, content, created_at) VALUES ('raw', 0, 2, 5, 'm1', 'm4', 'summary', 'body', 1)")).unwrap();
     }
+    #[test]
+    fn echoed_materialized_boundary_is_retained_when_a_newer_compartment_exists() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = McStore::open_for_test(&descriptor(dir.path())).unwrap();
+        seed(&store);
+        let first = [boundary()];
+        store
+            .apply_authority_state_sync(request(&first, 0))
+            .unwrap();
+        let (meta, materialized, _) = store.load_state_sync_inventory("raw", true).unwrap();
+        assert!(meta.initialized);
+        assert_eq!(materialized, "m4#0");
+
+        // A newer compartment is published but not folded yet; the host echoes the
+        // module's own boundary, which now lags the newest compartment.
+        store
+            .with_context_conn_for_test(|conn| conn.execute_batch(
+                "INSERT INTO compartments(session_id, sequence, start_message, end_message, start_message_id, end_message_id, title, content, created_at) VALUES ('raw', 1, 6, 9, 'm5', 'm8', 'later', 'body', 2)",
+            ))
+            .unwrap();
+        let mut newer = boundary();
+        newer.sequence = 1;
+        newer.source_start_message = 6;
+        newer.source_end_message = 9;
+        newer.source_start_message_id = "m5".into();
+        newer.source_end_message_id = "m8".into();
+        newer.start_message = 5;
+        newer.end_message = 8;
+        newer.start_message_id = "m5#0".into();
+        newer.end_message_id = "m8#0".into();
+        let rows = [boundary(), newer];
+        let (meta, _, _) = store.load_state_sync_inventory("raw", true).unwrap();
+        store
+            .apply_authority_state_sync(request(&rows, meta.shadow_seq))
+            .unwrap();
+        let (_, retained, _) = store.load_state_sync_inventory("raw", true).unwrap();
+        assert_eq!(retained, "m4#0");
+
+        // A declared boundary that is neither the materialized one nor the newest
+        // compartment is still refused.
+        let (meta, _, _) = store.load_state_sync_inventory("raw", true).unwrap();
+        let mut stale = request(&rows, meta.shadow_seq);
+        stale.seed_boundary_id = Some("m1#0");
+        assert!(matches!(
+            store.apply_authority_state_sync(stale),
+            Err(crate::ModuleStateSyncError::InvalidSeedBoundary { .. })
+        ));
+    }
+
     #[test]
     fn empty_legacy_source_id_accepts_host_resolution_without_rewriting_shared_row() {
         let dir = tempfile::tempdir().unwrap();

crates/mc-store/src/lib.rs

@@ -10854,7 +10854,30 @@ impl McStore {
             if !request.resolved_compartment_boundaries.is_empty() {
                 meta.resolved_compartment_boundaries = resolved_boundaries.clone();
             }
-            if let Some(declared) = request.seed_boundary_id {
+            // A reconnecting host without a whole-message boundary of its own echoes the
+            // boundary this module reported in its inventory. Compartments published after
+            // that fold can already be in context.db, so the echo may lag the newest
+            // compartment; it asks for no change, so the materialized boundary is kept
+            // rather than validated against the newest compartment and refused. It must
+            // still be a real compartment end: if the host rewrote that compartment, the
+            // echo falls through to validation and is refused.
+            let echoes_materialized_boundary = meta.initialized
+                && request.seed_boundary_id.is_some_and(|declared| {
+                    declared == core.boundary_id
+                        && seed_compartments
+                            .iter()
+                            .any(|compartment| compartment.end_message_id == declared)
+                });
+            if echoes_materialized_boundary {
+                tracing::info!(
+                    "mc-store: state-sync seed echoed materialized boundary {:?}; retained for session {}",
+                    core.boundary_id, request.session_id
+                );
+            }
+            if let Some(declared) = request
+                .seed_boundary_id
+                .filter(|_| !echoes_materialized_boundary)
+            {
                 let adoption = match validated_seed_boundary(declared, &seed_compartments) {
                     Ok(adoption) => adoption,
                     Err(detail) => {