From 146d5cebc8b7002b7df089e32b9fc48ab7e0ae68 Mon Sep 17 00:00:00 2001
From: Baz <22987674+bazarnov@users.noreply.github.com>
Date: Mon, 5 Oct 2026 23:10:33 +0300
Subject: [PATCH] fix(low-code): evaluate `concurrency_level` with the
 normalized config

`ConcurrencyLevel` was built from the raw config, while every other
component reads `self._config`, the copy that `config_normalization_rules`
migrated and transformed. A migration that rewrote the worker count took
effect one run late, and a transformation never did.
---
 .../concurrent_declarative_source.py          |  2 +-
 .../test_concurrent_declarative_source.py     | 36 +++++++++++++++++++
 2 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
index 0c5e9da08c..055576f45f 100644
--- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
+++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
@@ -247,7 +247,7 @@ def __init__(
             concurrency_level_component = self._constructor.create_component(
                 model_type=ConcurrencyLevelModel,
                 component_definition=concurrency_level_from_manifest,
-                config=config or {},
+                config=self._config,
             )
             if not isinstance(concurrency_level_component, ConcurrencyLevel):
                 raise ValueError(
diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py
index 47b80df66b..c25831b9ab 100644
--- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py
+++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py
@@ -1797,6 +1797,42 @@ def test_concurrency_level_initial_number_partitions_to_generate_is_always_one_o
     assert source._concurrent_source._initial_number_partitions_to_generate == 1
 
 
+_SET_NUM_WORKERS_TO_4 = {
+    "type": "ConfigAddFields",
+    "fields": [{"type": "AddedFieldDefinition", "path": ["num_workers"], "value": "4"}],
+}
+
+
+@pytest.mark.parametrize(
+    "config_normalization_rules",
+    [
+        pytest.param(
+            {
+                "config_migrations": [
+                    {"type": "ConfigMigration", "transformations": [_SET_NUM_WORKERS_TO_4]}
+                ]
+            },
+            id="config_migrations",
+        ),
+        pytest.param({"transformations": [_SET_NUM_WORKERS_TO_4]}, id="transformations"),
+    ],
+)
+def test_concurrency_level_uses_normalized_config(config_normalization_rules):
+    manifest = copy.deepcopy(_MANIFEST)
+    manifest["spec"] = {
+        "type": "Spec",
+        "connection_specification": {},
+        "config_normalization_rules": {
+            "type": "ConfigNormalizationRules",
+            **config_normalization_rules,
+        },
+    }
+
+    source = ConcurrentDeclarativeSource(source_config=manifest, config={"num_workers": 1})
+
+    assert source._concurrent_source._threadpool._threadpool._max_workers == 4
+
+
 def test_async_incremental_stream_uses_concurrent_cursor_with_state():
     state = [
         AirbyteStateMessage(