fix: handle missing asyncConfig in merge_with_existing

When updating a collection that was created without async replication
config, the existing schema has no asyncConfig key. merge_with_existing
would raise KeyError trying to access schema["asyncConfig"].

Use schema.get(cls_field, {}) to gracefully handle missing nested
_ConfigUpdateModel fields.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jfrancoa

integration/test_collection_config.py

@@ -1634,13 +1634,49 @@ def test_replication_config_remove_async_config(collection_factory: CollectionFa
     collection.config.update(
         replication_config=Reconfigure.replication(
             async_enabled=False,
+            async_config=None,
         ),
     )
     config = collection.config.get()
     assert config.replication_config.async_enabled is False
     assert config.replication_config.async_config is None
 
 
+def test_replication_config_add_async_config_to_existing_collection(
+    collection_factory: CollectionFactory,
+) -> None:
+    """Test updating a collection that was created without async_config to add one.
+
+    This covers the case where the existing schema has no asyncConfig key
+    and merge_with_existing must handle the missing field gracefully.
+    """
+    collection_dummy = collection_factory("dummy")
+    if collection_dummy._connection._weaviate_version.is_lower_than(1, 36, 0):
+        pytest.skip("async replication config requires Weaviate >= 1.36.0")
+
+    # Create without async_config
+    collection = collection_factory(
+        replication_config=Configure.replication(factor=1, async_enabled=True),
+    )
+    config = collection.config.get()
+    assert config.replication_config.async_config is None
+
+    # Update to add async_config
+    collection.config.update(
+        replication_config=Reconfigure.replication(
+            async_config=Reconfigure.Replication.async_config(
+                max_workers=12,
+                propagation_concurrency=4,
+            ),
+        ),
+    )
+    config = collection.config.get()
+    assert config.replication_config.async_config is not None
+    ac = config.replication_config.async_config
+    assert ac.max_workers == 12
+    assert ac.propagation_concurrency == 4
+
+
 def test_update_property_descriptions(collection_factory: CollectionFactory) -> None:
     collection = collection_factory(
         vectorizer_config=Configure.Vectorizer.none(),

test/collection/test_config.py

@@ -2986,6 +2986,31 @@ def test_replication_config_to_dict_without_async_config() -> None:
     assert "asyncConfig" not in d
 
 
+def test_replication_config_update_merge_with_missing_async_config() -> None:
+    """Test that merge_with_existing handles a schema without asyncConfig.
+
+    When a collection was created without async replication config and we
+    update it to add one, the existing schema won't have the asyncConfig key.
+    merge_with_existing must not raise KeyError in this case.
+    """
+    update = Reconfigure.replication(
+        async_config=Reconfigure.Replication.async_config(
+            max_workers=12,
+            propagation_concurrency=4,
+        ),
+    )
+    # Simulate an existing schema that has no asyncConfig key
+    existing_schema = {
+        "factor": 3,
+        "asyncEnabled": True,
+        "deletionStrategy": "NoAutomatedResolution",
+    }
+    result = update.merge_with_existing(existing_schema)
+    assert result["asyncConfig"]["maxWorkers"] == 12
+    assert result["asyncConfig"]["propagationConcurrency"] == 4
+    assert result["factor"] == 3
+
+
 def test_nested_property_with_id_name_is_allowed() -> None:
     """A nested property named 'id' must not raise — only top-level 'id' is reserved."""
     prop = Property(

weaviate/collections/classes/config_base.py

@@ -42,7 +42,7 @@ def merge_with_existing(self, schema: Dict[str, Any]) -> Dict[str, Any]:
                     )
                     schema[quantizer]["enabled"] = False
             elif isinstance(val, _ConfigUpdateModel):
-                schema[cls_field] = val.merge_with_existing(schema[cls_field])
+                schema[cls_field] = val.merge_with_existing(schema.get(cls_field, {}))
             else:
                 pass  # ignore unknown types so that individual classes can be extended
         return schema