diff --git a/CHANGELOG.next.md b/CHANGELOG.next.md
index bce0a0ec6a..66a1d004e4 100644
--- a/CHANGELOG.next.md
+++ b/CHANGELOG.next.md
@@ -35,6 +35,7 @@ Thanks, you're awesome :-) -->
 #### Improvements
 
 * Streamline RFC process from four stages (Strawperson, Draft, Candidate, Finished) to a single Proposal stage with target maturity. #2600
+* Detect type conflicts when adding intermediate fields. #2671
 
 #### Deprecated
 
diff --git a/scripts/schema/loader.py b/scripts/schema/loader.py
index 7773652e9c..5e4e8cf0c3 100644
--- a/scripts/schema/loader.py
+++ b/scripts/schema/loader.py
@@ -189,10 +189,13 @@ def nest_fields(field_array: List[Field]) -> Dict[str, Dict[str, FieldEntry]]:
             # Respect explicitly defined object fields
             if 'type' in field_details and field_details['type'] in ['object', 'nested']:
                 field_details.setdefault('intermediate', False)
-            else:
+            elif 'type' not in field_details:
                 field_details.setdefault('type', 'object')
                 field_details.setdefault('name', '.'.join(parent_fields[:idx + 1]))
                 field_details.setdefault('intermediate', True)
+            else:
+                msg = f"Type conflict detected when adding intermediate field '{level}' (adding: object, existing: {field_details['type']})"
+                raise ValueError(msg)
 
             # moving the nested_schema cursor deeper
             current_path.extend([level])
@@ -200,6 +203,9 @@ def nest_fields(field_array: List[Field]) -> Dict[str, Dict[str, FieldEntry]]:
         nested_schema.setdefault(leaf_field, {})
         # Overwrite 'name' with the leaf field's name. The flat_name is already computed.
         field['node_name'] = leaf_field
+        if 'field_details' in nested_schema[leaf_field] and nested_schema[leaf_field]['field_details']['type'] != field['type']:
+            msg = f"Type conflict detected when adding leaf field '{leaf_field}' (adding: {field['type']}, existing: {nested_schema[leaf_field]['field_details']['type']})"
+            raise ValueError(msg)
         nested_schema[leaf_field]['field_details'] = field
     return schema_root
 
diff --git a/scripts/tests/unit/test_schema_loader.py b/scripts/tests/unit/test_schema_loader.py
index 3ac7859e8e..cd49899147 100644
--- a/scripts/tests/unit/test_schema_loader.py
+++ b/scripts/tests/unit/test_schema_loader.py
@@ -255,6 +255,19 @@ def test_nest_fields(self):
         nested_fields = loader.nest_fields(process_fields)
         self.assertEqual(nested_fields, expected_nested_fields)
 
+    def test_nest_fields_incompatible_types(self):
+        test_fields = [
+            {'name': 'foo', 'type': 'keyword'},
+            {'name': 'foo.bar', 'type': 'keyword'},
+        ]
+        self.assertRaises(ValueError, loader.nest_fields, test_fields)
+
+        test_fields = [
+            {'name': 'foo.bar', 'type': 'keyword'},
+            {'name': 'foo', 'type': 'keyword'},
+        ]
+        self.assertRaises(ValueError, loader.nest_fields, test_fields)
+
     def test_nest_fields_recognizes_explicitly_defined_object_fields(self):
         dns_fields = [
             {'name': 'question.name', 'type': 'keyword'},