feat(merge_configs): expose public merge_configs(*dicts) API per §3.1 deep-merge algorithm #19

Merged
hurui200320 merged 1 commits from feature/merge-configs-api into master 2026-06-03 15:31:41 +00:00
9 changed files with 621 additions and 2 deletions
+1
View File
@@ -9,6 +9,7 @@ and this project adheres to [Clever Semantic Versioning](https://www.w3.org/subm
### Added
- **`merge_configs()` public API** (`cleveractors.merge_configs`): New module-level function implementing the Actor Configuration Standard §3.1 deep-merge algorithm. Accepts an arbitrary number of `dict[str, Any]` arguments and returns a fresh merged dict without mutating any input. Merge semantics: absent key → add; both mappings → deep-merge recursively; both sequences → append; otherwise → replace. Zero-argument call returns `{}`. Exported from `cleveractors.__init__` and `__all__`.
- **Core CleverActors Framework**: New agent-based LLM orchestration framework implementing the Actor Configuration Standard (§1-4). Includes agent base class, factory pattern for agent creation, configuration management, template rendering engine, and exception hierarchy.
- **LLM Agent** (`type: llm`, §4.4): Agent backed by language models with support for OpenAI, Anthropic, and Google Gemini providers. Configurable temperature, max_tokens, system prompts, memory/history, and structured output (json_mode/response_format).
- **Tool Agent** (`type: tool`, §4.5): Deterministic agent executing built-in tools (echo, math, json_parse, http_request, file_read, file_write, progress_bar) and custom inline code tools. Supports safe/unsafe execution modes, shell command filtering, and file operation sandboxing.
+5
View File
@@ -0,0 +1,5 @@
# Contributors
* Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
* Luis Mendes <luis.p.mendes@gmail.com>
* Rui Hu <rui.hu@cleverthis.com>
+68
View File
@@ -0,0 +1,68 @@
"""ASV benchmarks for the merge_configs() public API.
Measures the performance of the §3.1 deep-merge algorithm across several
representative workloads: flat dicts, deeply nested dicts, and large sequences.
"""
from __future__ import annotations
from typing import Any
from cleveractors import merge_configs
class MergeConfigsBenchmark:
"""Benchmark suite for merge_configs()."""
def setup(self) -> None:
"""Prepare reusable fixtures."""
# Flat dicts — simple key insertion and scalar override
self.flat_base: dict[str, Any] = {f"key_{i}": i for i in range(50)}
self.flat_overlay: dict[str, Any] = {
f"key_{i}": i + 1000 for i in range(25, 75)
}
# Deeply nested dicts — 5 levels deep, 10 keys per level
self.nested_base: dict[str, Any] = self._make_nested(depth=5, width=10, value=1)
self.nested_overlay: dict[str, Any] = self._make_nested(
depth=5, width=10, value=2
)
# Sequence append — lists at top level
self.seq_base: dict[str, Any] = {"items": list(range(100))}
self.seq_overlay: dict[str, Any] = {"items": list(range(100, 200))}
# Three-way merge — 50 keys each with partial overlap
self.three_a: dict[str, Any] = {f"key_{i}": i for i in range(50)}
self.three_b: dict[str, Any] = {f"key_{i}": i + 1000 for i in range(25, 75)}
self.three_c: dict[str, Any] = {f"key_{i}": i + 2000 for i in range(50, 100)}
@staticmethod
def _make_nested(depth: int, width: int, value: int) -> dict[str, Any]:
"""Recursively build a nested dict for benchmark fixtures."""
if depth == 0:
return {f"leaf_{k}": value for k in range(width)}
return {
f"key_{k}": MergeConfigsBenchmark._make_nested(depth - 1, width, value)
for k in range(width)
}
def time_flat_merge(self) -> None:
"""Benchmark merging two flat dicts with 5075 keys each."""
merge_configs(self.flat_base, self.flat_overlay)
def time_nested_merge(self) -> None:
"""Benchmark deep-merging two 5-level nested dicts."""
merge_configs(self.nested_base, self.nested_overlay)
def time_sequence_append(self) -> None:
"""Benchmark sequence append for two 100-element lists."""
merge_configs(self.seq_base, self.seq_overlay)
def time_three_way_merge(self) -> None:
"""Benchmark three-dict variadic merge."""
merge_configs(self.three_a, self.three_b, self.three_c)
def time_zero_args(self) -> None:
"""Benchmark zero-argument call (baseline overhead)."""
merge_configs()
+131
View File
@@ -0,0 +1,131 @@
Feature: Public merge_configs API — §3.1 deep-merge algorithm
As a CleverThis router developer
I want a public merge_configs(*dicts) function that implements the §3.1 deep-merge semantics
So that I can combine platform base configs with actor configs from the database without mutating inputs
# §3.1 merge semantics (from docs/actor-standard.md):
# - Key absent from accumulated result add it.
# - Both values are mappings deep-merge recursively.
# - Both values are sequences append new sequence to existing.
# - Otherwise replace with new value.
Scenario: Zero-argument call returns empty dict
When I call merge_configs with no arguments
Then the result should be an empty dict
Scenario: Single dict call returns a deep copy
Given a dict with key "a" equal to 1 and nested key "b" mapping to {"x": 2}
When I call merge_configs with that single dict
Then the result should equal {"a": 1, "b": {"x": 2}}
And the result should not be the same object as the input
And the nested value for key "b" should not be the same object as the input's nested value
Scenario: Empty dict as first argument should be transparent
Given a base dict {}
And a new dict {"a": 1}
When I call merge_configs with both dicts
Then the result should equal {"a": 1}
Scenario: Empty dict as second argument should be transparent
Given a base dict {"a": 1}
And a new dict {}
When I call merge_configs with both dicts
Then the result should equal {"a": 1}
Scenario: Null or None arguments raise TypeError
Given a base dict {"a": 1}
When I call merge_configs with a None argument after the base dict
Then a TypeError should be raised
And the error message should mention "must be a dict"
Scenario: None as first argument raises TypeError
When I call merge_configs with None as the first argument
Then a TypeError should be raised
And the error message should mention "must be a dict"
Scenario: None as only argument raises TypeError
When I call merge_configs with None as the only argument
Then a TypeError should be raised
And the error message should mention "must be a dict"
Scenario: Two dicts — key absent in base is added
Given a base dict {"x": 1}
And a new dict {"y": 2}
When I call merge_configs with both dicts
Then the result should equal {"x": 1, "y": 2}
Scenario: Two dicts — scalar value is replaced
Given a base dict {"key": "old"}
And a new dict {"key": "new"}
When I call merge_configs with both dicts
Then the result should equal {"key": "new"}
Scenario: Two dicts — both values are mappings, deep-merge recursively
Given a base dict {"config": {"a": 1, "b": 2}}
And a new dict {"config": {"b": 99, "c": 3}}
When I call merge_configs with both dicts
Then the result should equal {"config": {"a": 1, "b": 99, "c": 3}}
Scenario: Two dicts — both values are sequences, new appended to existing
Given a base dict {"items": [1, 2]}
And a new dict {"items": [3, 4]}
When I call merge_configs with both dicts
Then the result should equal {"items": [1, 2, 3, 4]}
Scenario: Lists inside nested dicts are appended
Given a base dict {"outer": {"items": [1, 2]}}
And a new dict {"outer": {"items": [3, 4]}}
When I call merge_configs with both dicts
Then the result should equal {"outer": {"items": [1, 2, 3, 4]}}
Scenario: Three dicts — chained merge applies left to right
Given a base dict {"a": 1, "b": [10]}
And a second dict {"b": [20], "c": 3}
And a third dict {"a": 99, "c": 99}
When I call merge_configs with all three dicts
Then the result should equal {"a": 99, "b": [10, 20], "c": 99}
Scenario: Deeply nested dicts are merged recursively
Given a base dict {"level1": {"level2": {"key": "original", "extra": true}}}
And a new dict {"level1": {"level2": {"key": "updated", "new_key": "added"}}}
When I call merge_configs with both dicts
Then the result should equal {"level1": {"level2": {"key": "updated", "extra": true, "new_key": "added"}}}
Scenario: Inputs are never mutated after the call
Given a base dict {"shared": {"value": 1}, "items": [1, 2]}
And a new dict {"shared": {"extra": 9}, "items": [3]}
When I call merge_configs with both dicts
Then the base dict should still equal {"shared": {"value": 1}, "items": [1, 2]}
And the new dict should still equal {"shared": {"extra": 9}, "items": [3]}
Scenario: Mutating the result does not affect the original inputs
Given a base dict {"shared": {"value": 1}, "items": [1, 2]}
And a new dict {"shared": {"extra": 9}, "items": [3]}
When I call merge_configs with both dicts
And I mutate a nested value in the result
Then the base dict should still equal {"shared": {"value": 1}, "items": [1, 2]}
And the new dict should still equal {"shared": {"extra": 9}, "items": [3]}
Scenario: Sequence mixed with non-sequence replaces value
Given a base dict {"field": [1, 2, 3]}
And a new dict {"field": "scalar"}
When I call merge_configs with both dicts
Then the result should equal {"field": "scalar"}
Scenario: Mapping mixed with non-mapping replaces value
Given a base dict {"field": {"nested": true}}
And a new dict {"field": 42}
When I call merge_configs with both dicts
Then the result should equal {"field": 42}
Scenario: Scalar replaced by list
Given a base dict {"field": "scalar"}
And a new dict {"field": [1, 2, 3]}
When I call merge_configs with both dicts
Then the result should equal {"field": [1, 2, 3]}
Scenario: Scalar replaced by dict
Given a base dict {"field": "scalar"}
And a new dict {"field": {"nested": true}}
When I call merge_configs with both dicts
Then the result should equal {"field": {"nested": true}}
+196
View File
@@ -0,0 +1,196 @@
"""
Step definitions for the merge_configs public API BDD tests.
Tests verify the §3.1 deep-merge semantics:
- Key absent from accumulated result → added.
- Both values are mappings → deep-merged recursively.
- Both values are sequences → new appended to existing.
- Otherwise → new value replaces existing.
"""
import json
from typing import Any
from behave import given, then, when
from cleveractors import merge_configs
@when("I call merge_configs with no arguments")
def step_call_merge_configs_no_args(context: Any) -> None:
"""Call merge_configs() with zero arguments."""
context.result = merge_configs()
@then("the result should be an empty dict")
def step_result_is_empty_dict(context: Any) -> None:
"""Verify result is an empty dictionary."""
assert context.result == {}, f"Expected {{}}, got {context.result!r}"
@given(
'a dict with key "{key}" equal to {value:d} and nested key "{nested_key}" mapping to {nested_value}'
)
def step_given_single_dict_with_nested(
context: Any, key: str, value: int, nested_key: str, nested_value: str
) -> None:
"""Set up a single dict with a top-level key and a nested dict."""
context.single_dict = {key: value, nested_key: json.loads(nested_value)}
@when("I call merge_configs with that single dict")
def step_call_merge_configs_single(context: Any) -> None:
"""Call merge_configs with one dict."""
context.result = merge_configs(context.single_dict)
@then("the result should not be the same object as the input")
def step_result_not_same_object(context: Any) -> None:
"""Verify the result is a new dict, not the original."""
assert context.result is not context.single_dict, (
"merge_configs returned the original dict — it must return a new object"
)
@then(
'the nested value for key "{nested_key}" should not be the same object as the input\'s nested value'
)
def step_nested_value_not_same_object(context: Any, nested_key: str) -> None:
"""Verify a nested dict value in the result is a distinct object from the input's nested value."""
result_nested = context.result[nested_key]
input_nested = context.single_dict[nested_key]
assert result_nested is not input_nested, (
f"merge_configs returned a shallow copy — nested value for {nested_key!r}"
" is the same object as in the input"
)
@given("a base dict {json_str}")
def step_given_base_dict(context: Any, json_str: str) -> None:
"""Parse and store the base dictionary."""
context.base_dict = json.loads(json_str)
@given("a new dict {json_str}")
def step_given_new_dict(context: Any, json_str: str) -> None:
"""Parse and store the second (overlay) dictionary."""
context.new_dict = json.loads(json_str)
@given("a second dict {json_str}")
def step_given_second_dict(context: Any, json_str: str) -> None:
"""Parse and store the second dict in a three-dict scenario."""
context.second_dict = json.loads(json_str)
@given("a third dict {json_str}")
def step_given_third_dict(context: Any, json_str: str) -> None:
"""Parse and store the third dict in a three-dict scenario."""
context.third_dict = json.loads(json_str)
@when("I call merge_configs with both dicts")
def step_call_merge_configs_two(context: Any) -> None:
"""Call merge_configs with two dicts."""
context.result = merge_configs(context.base_dict, context.new_dict)
@when("I call merge_configs with all three dicts")
def step_call_merge_configs_three(context: Any) -> None:
"""Call merge_configs with three dicts."""
context.result = merge_configs(
context.base_dict, context.second_dict, context.third_dict
)
@when("I call merge_configs with a None argument after the base dict")
def step_call_merge_configs_with_none_second(context: Any) -> None:
"""Call merge_configs(a_dict, None) and expect TypeError."""
bad_arg: Any = None
try:
merge_configs(context.base_dict, bad_arg)
context.exception = None
except TypeError as exc:
context.exception = exc
@when("I call merge_configs with None as the first argument")
def step_call_merge_configs_with_none_first(context: Any) -> None:
"""Call merge_configs(None, a_dict) and expect TypeError."""
try:
merge_configs(None, {"a": 1})
context.exception = None
except TypeError as exc:
context.exception = exc
@when("I call merge_configs with None as the only argument")
def step_call_merge_configs_with_none_only(context: Any) -> None:
"""Call merge_configs(None) and expect TypeError."""
try:
bad_arg: Any = None
merge_configs(bad_arg)
context.exception = None
except TypeError as exc:
context.exception = exc
@then("the result should equal {json_str}")
def step_result_equals(context: Any, json_str: str) -> None:
"""Verify the merge result matches the expected JSON value."""
expected = json.loads(json_str)
assert context.result == expected, f"Expected {expected!r}, got {context.result!r}"
@then("the base dict should still equal {json_str}")
def step_base_dict_unchanged(context: Any, json_str: str) -> None:
"""Verify the base dict was not mutated by the merge call."""
expected = json.loads(json_str)
assert context.base_dict == expected, (
f"Base dict was mutated: expected {expected!r}, got {context.base_dict!r}"
)
@then("the new dict should still equal {json_str}")
def step_new_dict_unchanged(context: Any, json_str: str) -> None:
"""Verify the new dict was not mutated by the merge call."""
expected = json.loads(json_str)
assert context.new_dict == expected, (
f"New dict was mutated: expected {expected!r}, got {context.new_dict!r}"
)
@when("I mutate a nested value in the result")
def step_mutate_nested_value_in_result(context: Any) -> None:
"""Mutate a nested value in the result to verify input independence.
Discovers the first non-dict, non-list leaf inside a nested dict so
the step is not coupled to specific scenario data.
"""
if not context.result:
raise RuntimeError("Cannot mutate an empty result dict")
sentinel = object()
for key, val in context.result.items():
if isinstance(val, dict):
for sub_key, sub_val in val.items():
if not isinstance(sub_val, (dict, list)):
context.result[key][sub_key] = sentinel
return
raise RuntimeError("No mutable nested leaf found in result")
@then("a TypeError should be raised")
def step_typeerror_was_raised(context: Any) -> None:
"""Verify that a TypeError was raised by the previous step."""
assert context.exception is not None, (
"Expected TypeError but no exception was raised"
)
@then('the error message should mention "{expected_text}"')
def step_error_message_contains(context: Any, expected_text: str) -> None:
"""Verify the exception message contains the expected text."""
assert expected_text in str(context.exception), (
f"Expected error message to contain {expected_text!r},"
f" got {context.exception!r}"
)
+86 -2
View File
@@ -2,13 +2,17 @@
# This module is imported directly by Robot Framework test files.
import os
import sys
import tempfile
from pathlib import Path
import yaml
from cleveractors import ContextManager, ReactiveCleverAgentsApp, __version__
from cleveractors import (
ContextManager,
ReactiveCleverAgentsApp,
__version__,
merge_configs,
)
from cleveractors.agents.factory import Agent, AgentFactory
from cleveractors.core.config import ConfigurationManager, SchemaValidator
from cleveractors.core.exceptions import (
@@ -78,6 +82,86 @@ class CleverActorsLib: # pragma: no cover - integration test library
f"Interpolation failed: expected {var_value!r}, got {result.get('key')!r}"
)
# ── merge_configs public API ──────────────────────────────────────
def merge_configs_returns_empty_for_no_args(self) -> None:
result = merge_configs()
if result != {}:
raise AssertionError(f"Expected {{}}, got {result!r}")
def merge_configs_two_dicts(
self,
base_key: str,
base_val: str,
overlay_key: str,
overlay_val: str,
) -> None:
base = {base_key: base_val}
overlay = {overlay_key: overlay_val}
result = merge_configs(base, overlay)
if len(result) != 2:
raise AssertionError(f"Expected result length 2, got {len(result)}: {result!r}")
if result.get(base_key) != base_val:
raise AssertionError(f"Expected {base_key}={base_val!r}, got {result!r}")
if result.get(overlay_key) != overlay_val:
raise AssertionError(f"Expected {overlay_key}={overlay_val!r}, got {result!r}")
# Verify inputs not mutated
if base != {base_key: base_val}:
raise AssertionError(f"Base dict was mutated: {base!r}")
if overlay != {overlay_key: overlay_val}:
raise AssertionError(f"Overlay dict was mutated: {overlay!r}")
def merge_configs_deep_merge(self) -> None:
base = {"config": {"a": 1, "b": 2}}
overlay = {"config": {"b": 99, "c": 3}}
result = merge_configs(base, overlay)
expected = {"config": {"a": 1, "b": 99, "c": 3}}
if result != expected:
raise AssertionError(f"Expected {expected!r}, got {result!r}")
def merge_configs_sequence_append(self) -> None:
base = {"items": [1, 2]}
overlay = {"items": [3, 4]}
result = merge_configs(base, overlay)
if result != {"items": [1, 2, 3, 4]}:
raise AssertionError(f"Expected items=[1,2,3,4], got {result!r}")
def merge_configs_through_config_pipeline(self, base_key: str, overlay_val: str) -> None:
"""End-to-end test: load YAML via ConfigurationManager, merge with overlay.
Loads the test_config.yaml fixture through the canonical config-loading
pipeline (ConfigurationManager.load_files → to_dict), merges an overlay
dict via merge_configs, and asserts the result preserves base keys plus
the overlay injection.
"""
fixtures = Path(__file__).resolve().parent.parent / "tests" / "fixtures"
cm_base = ConfigurationManager()
cm_base.load_files([fixtures / "test_config.yaml"])
base_dict = cm_base.to_dict()
overlay: dict[str, Any] = {base_key: overlay_val, "meta": {"merged": True}}
result = merge_configs(base_dict, overlay)
# Verify overlay keys are present
if result.get(base_key) != overlay_val:
raise AssertionError(
f"Overlay key {base_key!r} expected {overlay_val!r},"
f" got {result.get(base_key)!r}"
)
if result.get("meta", {}).get("merged") is not True:
raise AssertionError(
f"Deep-merged overlay meta missing: {result.get('meta')!r}"
)
# Verify base keys from fixture survived the merge
if "agents" not in result:
raise AssertionError(
f"Base fixture key 'agents' lost during merge: {sorted(result.keys())}"
)
if "cleveragents" not in result:
raise AssertionError(
f"Base fixture key 'cleveragents' lost during merge"
)
def schema_validator_accepts_minimum_config(self) -> None:
config = {
"cleveragents": {"default_router": "main"},
+20
View File
@@ -51,6 +51,26 @@ Config Manager Serialises To Dict
Load Config Files ${FIXTURES}${/}test_config.yaml
No Operation
Merge Configs Returns Empty Dict For No Args
[Documentation] Verify merge_configs() with no arguments returns {}
Merge Configs Returns Empty For No Args
Merge Configs Combines Two Dicts
[Documentation] Verify merge_configs(base, overlay) merges keys from both dicts
Merge Configs Two Dicts platform_key platform_val actor_key actor_val
Merge Configs Deep Merges Nested Dicts
[Documentation] Verify merge_configs deep-merges nested mappings per §3.1
Merge Configs Deep Merge
Merge Configs Appends Sequences
[Documentation] Verify merge_configs appends lists when both values are sequences
Merge Configs Sequence Append
Merge Configs Through Config Pipeline
[Documentation] End-to-end test: load YAML via ConfigurationManager, merge overlay via merge_configs
Merge Configs Through Config Pipeline injected_key injected_value
Schema Validator Accepts Minimum Config
[Documentation] Verify schema validation passes for minimal valid config
Schema Validator Accepts Minimum Config
+2
View File
@@ -10,6 +10,7 @@ __version__ = "2.0.0"
__author__ = "CleverThis Engineering"
from cleveractors.agent import Agent
from cleveractors.config_utils import merge_configs
from cleveractors.context_manager import ContextManager
from cleveractors.core.application import ReactiveCleverAgentsApp
from cleveractors.core.exceptions import CleverAgentsException
@@ -17,6 +18,7 @@ from cleveractors.core.exceptions import CleverAgentsException
__all__ = [
"__version__",
"Agent",
"merge_configs",
"ContextManager",
"ReactiveCleverAgentsApp",
"CleverAgentsException",
+112
View File
@@ -0,0 +1,112 @@
"""
Public configuration utilities for CleverActors.
Exposes the merge_configs() function implementing the Actor Configuration
Standard §3.1 deep-merge algorithm. This is the public, immutable counterpart
of the internal ReactiveConfigParser._merge_configs() instance method.
§3.1 merge semantics:
- Key absent from the accumulated result → the key-value pair is added as-is.
- Both values are mappings → they are deeply merged (same rules at every
nesting level).
- Both values are sequences → the new sequence is appended to the existing.
- Any other combination → the new value replaces the accumulated value.
Inputs are never mutated: the function always returns a fresh dict.
.. note::
Values are processed with ``copy.deepcopy``. Do not pass dicts
containing objects from untrusted sources, as ``deepcopy`` can execute
arbitrary code via custom ``__reduce__`` / ``__deepcopy__`` methods.
.. warning::
Circular references in input dicts are not supported and will cause an
infinite loop. The implementation uses a LIFO stack for merging; when
both sides of a key are dicts, the sub-dict pair is pushed onto the
stack. A self-referential dict will push the same pair repeatedly.
"""
import copy
import logging
from typing import Any
logger = logging.getLogger(__name__)
def merge_configs(*dicts: dict[str, Any]) -> dict[str, Any]:
"""Merge configuration dicts using §3.1 deep-merge semantics.
Accepts zero or more dicts, applied left to right. Later dicts override
or extend earlier ones according to the §3.1 rules. Inputs are never
mutated; a fresh dict is returned.
Repeated list overlays on the same key use ``existing + copy(new)``,
which is O(k²·m) for *k* chained overlays on a list of length *m*.
Examples
--------
>>> merge_configs()
{}
>>> merge_configs({"a": 1}, {"b": 2})
{'a': 1, 'b': 2}
>>> merge_configs({"a": [1]}, {"a": [2]})
{'a': [1, 2]}
>>> merge_configs({"n": {"x": 1}}, {"n": {"y": 2}})
{'n': {'x': 1, 'y': 2}}
"""
accumulated: dict[str, Any] = {}
for source in dicts:
if not isinstance(source, dict):
raise TypeError(
f"merge_configs() argument must be a dict, got {type(source).__name__!r}"
)
_apply_merge(accumulated, source)
return accumulated
def _apply_merge(base: dict[str, Any], overlay: dict[str, Any]) -> None:
"""Apply *overlay* onto *base* in-place, following §3.1 rules.
Private helper operating on an internally-managed accumulator; never
called with the caller's original dicts. Uses an explicit LIFO stack
instead of recursion so deeply nested configs cannot overflow.
"""
# ── explicit stack replaces recursion ──────────────────────────
# Each stack entry is (base_dict, overlay_dict). When both sides of
# a key are dicts we push the sub-dict pair onto the stack instead of
# recursing. Different keys map to independent subtrees, so the
# traversal order is irrelevant — the result is identical to the
# recursive version.
hurui200320 marked this conversation as resolved Outdated
Outdated
Review

a debug logging could be useful to know that we are replacing some item with a specific key

Reply: Good catch. Added logger.debug() calls in all three merge branches of _apply_merge:

  • When both sides are mappings: "Deep-merging key %r (both sides are mappings)"
  • When both sides are sequences: "Appending %d items to list at key %r"
  • Otherwise (replacement): "Replacing value for key %r: %s -> %s" with the old and new type names

The logger follows the project convention: logging.getLogger(__name__) at module level. All debug messages fire at the affected key, which gives enough context for diagnosing merge behavior without being noisy.

a debug logging could be useful to know that we are replacing some item with a specific key **Reply:** Good catch. Added `logger.debug()` calls in all three merge branches of `_apply_merge`: - When both sides are mappings: `"Deep-merging key %r (both sides are mappings)"` - When both sides are sequences: `"Appending %d items to list at key %r"` - Otherwise (replacement): `"Replacing value for key %r: %s -> %s"` with the old and new type names The logger follows the project convention: `logging.getLogger(__name__)` at module level. All debug messages fire at the affected key, which gives enough context for diagnosing merge behavior without being noisy.
stack: list[tuple[dict[str, Any], dict[str, Any]]] = [(base, overlay)]
while stack:
current_base, current_overlay = stack.pop()
for key, new_value in current_overlay.items():
hurui200320 marked this conversation as resolved Outdated
Outdated
Review

Okay, it works, but recursion is not the most perfomant solution. If do not expect very large entries, this could be fine. Othwerwise it could make sense to re-write the implementation to not be recursive.

Reply: Converted _apply_merge from recursion to an explicit stack. The complexity analysis was straightforward — when both sides of a key are dicts, the different subtrees map to independent keys, so traversal order is irrelevant. A LIFO stack produces identical results to recursion.

The change: instead of _apply_merge(existing, new_value) (recursive call), the sub-dict pair is pushed onto a stack. A while stack: loop pops and processes each pair. This eliminates the risk of hitting Python's recursion limit on deeply nested configs, and matches the non-recursive pattern used in the internal _merge_configs loop in parse_files(). All 1676 unit tests and 53 integration tests pass with identical results.

Okay, it works, but recursion is not the most perfomant solution. If do not expect very large entries, this could be fine. Othwerwise it could make sense to re-write the implementation to not be recursive. **Reply:** Converted `_apply_merge` from recursion to an explicit stack. The complexity analysis was straightforward — when both sides of a key are dicts, the different subtrees map to independent keys, so traversal order is irrelevant. A LIFO stack produces identical results to recursion. The change: instead of `_apply_merge(existing, new_value)` (recursive call), the sub-dict pair is pushed onto a stack. A `while stack:` loop pops and processes each pair. This eliminates the risk of hitting Python's recursion limit on deeply nested configs, and matches the non-recursive pattern used in the internal `_merge_configs` loop in `parse_files()`. All 1676 unit tests and 53 integration tests pass with identical results.
if key not in current_base:
# Rule: key absent → add it (deep-copy to preserve immutability).
current_base[key] = copy.deepcopy(new_value)
else:
existing = current_base[key]
if isinstance(existing, dict) and isinstance(new_value, dict):
# Rule: both mappings → deep-merge (push onto stack).
logger.debug("Deep-merging key %r (both sides are mappings)", key)
stack.append((existing, new_value))
elif isinstance(existing, list) and isinstance(new_value, list):
# Rule: both sequences → append new to existing.
logger.debug(
"Appending %d items to list at key %r",
len(new_value),
key,
)
current_base[key] = existing + copy.deepcopy(new_value)
else:
# Rule: any other combination → replace with new value.
logger.debug(
"Replacing value for key %r: %s -> %s",
key,
type(existing).__name__,
type(new_value).__name__,
)
current_base[key] = copy.deepcopy(new_value)