fix: wire replication params through YAML/params path for Cassandra and Qdrant

resolve_cassandra_config did not accept replication_factor as a kwarg,
so cassandra_replication_factor from YAML params was silently ignored
by all 6 callers. Add the kwarg and pass it from every caller.

Same fix for Qdrant: 3 writers now pass qdrant_replication_factor and
qdrant_shard_number from params.

Add tests covering the params path for both helpers.

tests/unit/test_base/test_cassandra_config.py

@@ -409,4 +409,57 @@ class TestEdgeCases:
         
         assert hosts == ['mixed-host']
         assert username is None  # Stays None
-        assert password == 'mixed-pass'
+        assert password == 'mixed-pass'
+
+
+class TestReplicationFactorParamPath:
+
+    def test_explicit_kwarg(self):
+        with patch.dict(os.environ, {}, clear=True):
+            _, _, _, _, rf = resolve_cassandra_config(
+                replication_factor=3,
+            )
+            assert rf == 3
+
+    def test_kwarg_overrides_env(self):
+        with patch.dict(os.environ, {'CASSANDRA_REPLICATION_FACTOR': '5'}, clear=True):
+            _, _, _, _, rf = resolve_cassandra_config(
+                replication_factor=3,
+            )
+            assert rf == 3
+
+    def test_env_fallback_when_kwarg_none(self):
+        with patch.dict(os.environ, {'CASSANDRA_REPLICATION_FACTOR': '5'}, clear=True):
+            _, _, _, _, rf = resolve_cassandra_config(
+                replication_factor=None,
+            )
+            assert rf == 5
+
+    def test_default_when_no_kwarg_no_env(self):
+        with patch.dict(os.environ, {}, clear=True):
+            _, _, _, _, rf = resolve_cassandra_config()
+            assert rf == 1
+
+    def test_params_dict_path(self):
+        with patch.dict(os.environ, {}, clear=True):
+            params = {'cassandra_replication_factor': 3}
+            _, _, _, _, rf = resolve_cassandra_config(
+                replication_factor=params.get('cassandra_replication_factor'),
+            )
+            assert rf == 3
+
+    def test_params_dict_overrides_env(self):
+        with patch.dict(os.environ, {'CASSANDRA_REPLICATION_FACTOR': '5'}, clear=True):
+            params = {'cassandra_replication_factor': 3}
+            _, _, _, _, rf = resolve_cassandra_config(
+                replication_factor=params.get('cassandra_replication_factor'),
+            )
+            assert rf == 3
+
+    def test_params_dict_missing_falls_to_env(self):
+        with patch.dict(os.environ, {'CASSANDRA_REPLICATION_FACTOR': '5'}, clear=True):
+            params = {}
+            _, _, _, _, rf = resolve_cassandra_config(
+                replication_factor=params.get('cassandra_replication_factor'),
+            )
+            assert rf == 5