feat(plans): implement configurable merge strategy (prefer-parent, prefer-subplan, manual)

Implemented a configurable merge strategy framework for three-way merges.

- New module: src/cleveragents/domain/models/core/merge_strategy.py
  - MergeStrategy StrEnum with options: prefer-parent, prefer-subplan, manual
  - Helper methods: is_auto_resolve(), is_manual(), from_string()

- New module: src/cleveragents/domain/models/core/merge_strategy_service.py
  - MergeConflict class with __eq__ for value equality comparison
  - MergeStrategyService to apply strategies and resolve conflicts
  - Proper type annotations (dict, Any) with no unused imports

- BDD test suite: features/plan_merge_strategy.feature (8 scenarios)
- Step definitions: features/steps/plan_merge_strategy_steps.py

- Robot Framework integration tests: robot/merge_strategy.robot
- Helper script: robot/helper_merge_strategy.py

- Updated src/cleveragents/domain/models/core/__init__.py exports

ISSUES CLOSED: #9559
This commit is contained in:
2026-04-15 00:38:39 +00:00
committed by drew
parent e2915ed18f
commit eb36363b1e
9 changed files with 606 additions and 0 deletions
+12
View File
@@ -195,6 +195,18 @@ ensuring data is stored with proper parameter values.
10 MB total per project), and fragment structure (`TieredFragment` with HOT tier placement
and metadata keys `path`, `detail_depth`, `relevance_score`). Closes #6175.
### Added
- **Configurable merge strategy for plan three-way merges** (#9559): Introduced
three configurable merge strategies — `prefer-parent`, `prefer-subplan`, and
`manual` — allowing teams to choose their preferred conflict resolution
behavior. MergeStrategy enum (StrEnum) with helper methods
(`is_auto_resolve()`, `is_manual()`, `from_string()`), MergeStrategyService
for applying strategies to resolve conflicts, comprehensive BDD test suite
(8 scenarios across all three strategies), and Robot Framework integration tests
verifying runtime behavior against live Python modules.
### Fixed
- **fileConfig error handling in alembic env.py** (#7874): Wrapped the `fileConfig()`
+6
View File
@@ -75,3 +75,9 @@ Below are some of the specific details of various contributions.
* HAL 9000 has contributed the `--clone-into` CLI argument for `container-instance`, the `CloneIntoHandler` module, the `devcontainer-instance` snapshot sandbox strategy, and the `ContainerLifecycleState.DISCOVERED` terminology alignment (PR #8304, issue #7555).
* HAL 9000 has contributed the ACMS Context Tier Hydration documentation (PR #9208 / issue #6175): documented the `context_tier_hydrator` module in the ACMS Architecture section of the specification, covering its public interface, file listing strategy, budget limits, and fragment structure.
* HAL 9000 has contributed the agent task memory leak fix (#9044): replaced `list.remove` with `set.discard` as the done_callback for asyncio tasks in `Agent._tasks`, preventing unbounded memory growth in long-lived agents and ensuring safe concurrent task removal.
# Details (PR Contributions)
Below are some specific details of individual PR contributions.
* HAL 9000 has contributed the configurable merge strategy implementation (PR #9610 / issue #9559): three configurable merge strategies (prefer-parent, prefer-subplan, manual) for plan three-way merges, MergeStrategy StrEnum with helper methods, MergeStrategyService for conflict resolution, BDD test suite with 8 scenarios, and Robot Framework integration tests.
+53
View File
@@ -0,0 +1,53 @@
Feature: Plan Merge Strategy Configuration
As a team lead
I want to configure how plan conflicts are resolved
So that my team can choose the appropriate conflict resolution behavior
Background:
Given a merge strategy service is initialized
Scenario: Prefer-parent strategy auto-resolves conflicts in favor of parent
Given the merge strategy is set to "prefer-parent"
When a conflict occurs with parent value "parent_data" and subplan value "subplan_data"
Then the conflict should be auto-resolved with value "parent_data"
And the strategy should support auto-resolution
Scenario: Prefer-subplan strategy auto-resolves conflicts in favor of subplan
Given the merge strategy is set to "prefer-subplan"
When a conflict occurs with parent value "parent_data" and subplan value "subplan_data"
Then the conflict should be auto-resolved with value "subplan_data"
And the strategy should support auto-resolution
Scenario: Manual strategy surfaces conflicts for user resolution
Given the merge strategy is set to "manual"
When a conflict occurs with parent value "parent_data" and subplan value "subplan_data"
Then the conflict should require manual resolution
And the strategy should not support auto-resolution
Scenario: Multiple conflicts are resolved according to strategy
Given the merge strategy is set to "prefer-parent"
When multiple conflicts occur:
| path | parent_value | subplan_value |
| field1 | value1 | value2 |
| field2 | value3 | value4 |
Then all conflicts should be resolved with parent values
Scenario: Manual strategy raises error when attempting auto-resolution
Given the merge strategy is set to "manual"
When attempting to auto-resolve conflicts
Then an error should be raised indicating manual resolution is required
Scenario: Merge strategy can be created from string value
When a merge strategy is created from string "prefer-parent"
Then the strategy should be "prefer-parent"
And the strategy string representation should be "prefer-parent"
Scenario: Invalid merge strategy string raises error
When attempting to create a merge strategy from invalid string "invalid-strategy"
Then an error should be raised with message containing "Invalid merge strategy"
Scenario: No conflicts means no resolution needed
Given the merge strategy is set to "prefer-parent"
When there are no conflicts
Then the resolution should return an empty result
And no conflicts should be detected
+165
View File
@@ -0,0 +1,165 @@
"""Step definitions for plan merge strategy feature tests."""
from behave import given, then, when
from cleveragents.domain.models.core.merge_strategy import MergeStrategy
from cleveragents.domain.models.core.merge_strategy_service import (
MergeConflict,
MergeStrategyService,
)
@given("a merge strategy service is initialized")
def step_initialize_service(context):
"""Initialize a merge strategy service."""
context.service = MergeStrategyService()
context.conflicts = []
context.resolved = {}
context.error = None
@given('the merge strategy is set to "{strategy}"')
def step_set_merge_strategy(context, strategy):
"""Set the merge strategy."""
context.service = MergeStrategyService(strategy=MergeStrategy.from_string(strategy))
@when(
'a conflict occurs with parent value "{parent_value}" and subplan value "{subplan_value}"'
)
def step_create_conflict(context, parent_value, subplan_value):
"""Create a conflict."""
conflict = MergeConflict(
path="test_field", parent_value=parent_value, subplan_value=subplan_value
)
context.conflicts = [conflict]
@then('the conflict should be auto-resolved with value "{expected_value}"')
def step_verify_auto_resolution(context, expected_value):
"""Verify that the conflict was auto-resolved with the expected value."""
try:
context.resolved = context.service.resolve_conflicts(context.conflicts)
assert "test_field" in context.resolved
assert context.resolved["test_field"] == expected_value
except ValueError as e:
context.error = e
@then("the strategy should support auto-resolution")
def step_verify_auto_resolve_support(context):
"""Verify that the strategy supports auto-resolution."""
assert context.service.can_auto_resolve()
@then("the conflict should require manual resolution")
def step_verify_manual_resolution(context):
"""Verify that the conflict requires manual resolution."""
try:
context.service.resolve_conflicts(context.conflicts)
except ValueError as e:
context.error = e
assert "manual" in str(e).lower()
@then("the strategy should not support auto-resolution")
def step_verify_no_auto_resolve_support(context):
"""Verify that the strategy does not support auto-resolution."""
assert not context.service.can_auto_resolve()
@when("multiple conflicts occur:")
def step_create_multiple_conflicts(context):
"""Create multiple conflicts from table."""
context.conflicts = []
for row in context.table:
conflict = MergeConflict(
path=row["path"],
parent_value=row["parent_value"],
subplan_value=row["subplan_value"],
)
context.conflicts.append(conflict)
@then("all conflicts should be resolved with parent values")
def step_verify_all_parent_resolution(context):
"""Verify that all conflicts are resolved with parent values."""
context.resolved = context.service.resolve_conflicts(context.conflicts)
for conflict in context.conflicts:
assert conflict.path in context.resolved
assert context.resolved[conflict.path] == conflict.parent_value
@when("attempting to auto-resolve conflicts")
def step_attempt_auto_resolve(context):
"""Attempt to auto-resolve conflicts."""
try:
context.service.resolve_conflicts(context.conflicts)
context.error = None
except ValueError as e:
context.error = e
@then("an error should be raised indicating manual resolution is required")
def step_verify_manual_resolution_error(context):
"""Verify that an error was raised for manual resolution."""
assert context.error is not None
assert "manual" in str(context.error).lower()
@when('a merge strategy is created from string "{strategy_str}"')
def step_create_strategy_from_string(context, strategy_str):
"""Create a merge strategy from a string."""
try:
context.strategy = MergeStrategy.from_string(strategy_str)
context.error = None
except ValueError as e:
context.error = e
@then('the strategy should be "{expected_strategy}"')
def step_verify_strategy_value(context, expected_strategy):
"""Verify the strategy value."""
assert context.strategy.value == expected_strategy
@then('the strategy string representation should be "{expected_str}"')
def step_verify_strategy_string(context, expected_str):
"""Verify the strategy string representation."""
assert str(context.strategy) == expected_str
@when('attempting to create a merge strategy from invalid string "{invalid_str}"')
def step_create_invalid_strategy(context, invalid_str):
"""Attempt to create a strategy from an invalid string."""
try:
MergeStrategy.from_string(invalid_str)
context.error = None
except ValueError as e:
context.error = e
@then('an error should be raised with message containing "{message_part}"')
def step_verify_error_message(context, message_part):
"""Verify that an error was raised with the expected message."""
assert context.error is not None
assert message_part in str(context.error)
@when("there are no conflicts")
def step_no_conflicts(context):
"""Set up a scenario with no conflicts."""
context.conflicts = []
@then("the resolution should return an empty result")
def step_verify_empty_resolution(context):
"""Verify that resolution returns an empty result."""
context.resolved = context.service.resolve_conflicts(context.conflicts)
assert context.resolved == {}
@then("no conflicts should be detected")
def step_verify_no_conflicts_detected(context):
"""Verify that no conflicts are detected."""
assert not context.service.has_conflicts(context.conflicts)
+124
View File
@@ -0,0 +1,124 @@
"""Helper utilities for merge strategy Robot tests."""
from __future__ import annotations
import sys
from cleveragents.domain.models.core.merge_strategy import MergeStrategy
from cleveragents.domain.models.core.merge_strategy_service import (
MergeConflict,
MergeStrategyService,
)
def _prefer_parent_test() -> None:
"""Test prefer-parent strategy auto-resolves conflicts."""
service = MergeStrategyService(strategy=MergeStrategy.PREFER_PARENT)
conflicts = [
MergeConflict(path="field1", parent_value="p1", subplan_value="s1"),
MergeConflict(path="field2", parent_value="p2", subplan_value="s2"),
]
resolved = service.resolve_conflicts(conflicts)
assert resolved["field1"] == "p1"
assert resolved["field2"] == "p2"
assert service.can_auto_resolve() is True
print("merge-strategy-prefer-parent-ok")
def _prefer_subplan_test() -> None:
"""Test prefer-subplan strategy auto-resolves conflicts."""
service = MergeStrategyService(strategy=MergeStrategy.PREFER_SUBPLAN)
conflicts = [
MergeConflict(path="field1", parent_value="p1", subplan_value="s1"),
MergeConflict(path="field2", parent_value="p2", subplan_value="s2"),
]
resolved = service.resolve_conflicts(conflicts)
assert resolved["field1"] == "s1"
assert resolved["field2"] == "s2"
assert service.can_auto_resolve() is True
print("merge-strategy-prefer-subplan-ok")
def _manual_test() -> None:
"""Test manual strategy raises error on auto-resolution."""
service = MergeStrategyService(strategy=MergeStrategy.MANUAL)
conflicts = [
MergeConflict(path="field1", parent_value="p1", subplan_value="s1"),
]
try:
service.resolve_conflicts(conflicts)
raise AssertionError("Expected ValueError for manual strategy")
except ValueError as e:
assert "manual" in str(e).lower()
assert service.can_auto_resolve() is False
print("merge-strategy-manual-ok")
def _no_conflicts_test() -> None:
"""Test no conflicts returns empty result."""
service = MergeStrategyService(strategy=MergeStrategy.PREFER_PARENT)
resolved = service.resolve_conflicts([])
assert resolved == {}
assert not service.has_conflicts([])
print("merge-strategy-no-conflicts-ok")
def _from_string_test() -> None:
"""Test MergeStrategy.from_string creates correct enum."""
strategy = MergeStrategy.from_string("prefer-parent")
assert strategy.value == "prefer-parent"
assert str(strategy) == "prefer-parent"
strategy_subplan = MergeStrategy.from_string("prefer-subplan")
assert strategy_subplan.value == "prefer-subplan"
strategy_manual = MergeStrategy.from_string("manual")
assert strategy_manual.is_manual() is True
assert strategy_manual.is_auto_resolve() is False
print("merge-strategy-from-string-ok")
def _invalid_string_test() -> None:
"""Test invalid string raises ValueError."""
try:
MergeStrategy.from_string("invalid-strategy")
raise AssertionError("Expected ValueError for invalid strategy string")
except ValueError as e:
assert "Invalid merge strategy" in str(e)
print("merge-strategy-invalid-string-ok")
def _enums_test() -> None:
"""Test MergeStrategy enum values and methods."""
assert MergeStrategy.PREFER_PARENT == "prefer-parent"
assert MergeStrategy.PREFER_SUBPLAN == "prefer-subplan"
assert MergeStrategy.MANUAL == "manual"
assert MergeStrategy.PREFER_PARENT.is_auto_resolve() is True
assert MergeStrategy.PREFER_SUBPLAN.is_auto_resolve() is True
assert MergeStrategy.MANUAL.is_auto_resolve() is False
assert MergeStrategy.PREFER_PARENT.is_manual() is False
assert MergeStrategy.PREFER_SUBPLAN.is_manual() is False
assert MergeStrategy.MANUAL.is_manual() is True
print("merge-strategy-enums-ok")
def main() -> None:
if len(sys.argv) < 2:
raise SystemExit("Expected command argument")
command = sys.argv[1]
tests = {
"prefer_parent": _prefer_parent_test,
"prefer_subplan": _prefer_subplan_test,
"manual": _manual_test,
"no_conflicts": _no_conflicts_test,
"from_string": _from_string_test,
"invalid_string": _invalid_string_test,
"enums": _enums_test,
}
if command not in tests:
raise SystemExit(f"Unknown command: {command}")
tests[command]()
+49
View File
@@ -0,0 +1,49 @@
*** Settings ***
Documentation Integration tests for merge strategy configuration (prefer-parent, prefer-subplan, manual)
Resource ${CURDIR}/common.resource
*** Variables ***
${HELPER_SCRIPT} robot/helper_merge_strategy.py
*** Test Cases ***
Prefer-Parent Auto-Resolves Conflicts
[Documentation] Verify prefer-parent strategy resolves all conflicts in favor of parent values
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} prefer_parent cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} merge-strategy-prefer-parent-ok
Prefer-Subplan Auto-Resolves Conflicts
[Documentation] Verify prefer-subplan strategy resolves all conflicts in favor of subplan values
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} prefer_subplan cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} merge-strategy-prefer-subplan-ok
Manual Strategy Requires Resolution
[Documentation] Verify manual strategy raises ValueError on attempted auto-resolution
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} manual cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} merge-strategy-manual-ok
No Conflicts Returns Empty Result
[Documentation] Verify empty conflict list returns empty resolution
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} no_conflicts cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} merge-strategy-no-conflicts-ok
Strategy From String Creates Correct Enum
[Documentation] Verify MergeStrategy.from_string creates proper enum values
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} from_string cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} merge-strategy-from-string-ok
Invalid String Raises Error
[Documentation] Verify invalid string raises ValueError with descriptive message
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} invalid_string cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} merge-strategy-invalid-string-ok
Enum Values And Methods Validated
[Documentation] Verify all enum member values and helper methods are correct
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} enums cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} merge-strategy-enums-ok
@@ -157,6 +157,13 @@ from cleveragents.domain.models.core.invariant import (
InvariantViolation,
merge_invariants,
)
# Merge strategy domain models (v3.3.0 M4 subplan merge strategies)
from cleveragents.domain.models.core.merge_strategy import MergeStrategy
from cleveragents.domain.models.core.merge_strategy_service import (
MergeConflict,
MergeStrategyService,
)
from cleveragents.domain.models.core.multi_project import (
ChangeSetSummary,
CrossProjectDependency,
@@ -453,6 +460,9 @@ __all__ = [
"LinkedPlan",
"LinkedResource",
"MaxContextCount",
"MergeConflict",
"MergeStrategy",
"MergeStrategyService",
"MessageRole",
"MultiProjectMetadata",
"NamespacedName",
@@ -0,0 +1,69 @@
"""Merge strategy configuration for plan three-way merges.
This module provides configurable merge strategies for resolving conflicts
when merging parent plans with subplans.
"""
from enum import StrEnum
class MergeStrategy(StrEnum):
"""Enumeration of available merge strategies for plan merging.
Attributes:
PREFER_PARENT: Auto-resolve conflicts by choosing parent values
PREFER_SUBPLAN: Auto-resolve conflicts by choosing subplan values
MANUAL: Surface conflicts for user resolution
"""
PREFER_PARENT = "prefer-parent"
"""Auto-resolve conflicts in favor of parent plan."""
PREFER_SUBPLAN = "prefer-subplan"
"""Auto-resolve conflicts in favor of subplan."""
MANUAL = "manual"
"""Surface conflicts for manual user resolution."""
def __str__(self) -> str:
"""Return the string representation of the merge strategy."""
return self.value
@classmethod
def from_string(cls, value: str) -> "MergeStrategy":
"""Create a MergeStrategy from a string value.
Args:
value: The string representation of the merge strategy
Returns:
The corresponding MergeStrategy enum value
Raises:
ValueError: If the value is not a valid merge strategy
"""
try:
return cls(value)
except ValueError:
valid_values = ", ".join(str(s) for s in cls)
raise ValueError(
f"Invalid merge strategy '{value}'. "
f"Valid options are: {valid_values}"
) from None
def is_auto_resolve(self) -> bool:
"""Check if this strategy auto-resolves conflicts.
Returns:
True if the strategy automatically resolves conflicts,
False if conflicts need manual resolution
"""
return self in (MergeStrategy.PREFER_PARENT, MergeStrategy.PREFER_SUBPLAN)
def is_manual(self) -> bool:
"""Check if this strategy requires manual conflict resolution.
Returns:
True if conflicts need manual resolution, False otherwise
"""
return self == MergeStrategy.MANUAL
@@ -0,0 +1,118 @@
"""Service for handling plan merge strategy operations.
This module provides the MergeStrategyService for applying merge strategies
to resolve conflicts during plan merging.
"""
from typing import Any
from .merge_strategy import MergeStrategy
class MergeConflict:
"""Represents a conflict during plan merge.
Attributes:
path: The path to the conflicting field
parent_value: The value from the parent plan
subplan_value: The value from the subplan
"""
def __init__(
self, path: str, parent_value: Any, subplan_value: Any
) -> None:
"""Initialize a merge conflict.
Args:
path: The path to the conflicting field
parent_value: The value from the parent plan
subplan_value: The value from the subplan
"""
self.path = path
self.parent_value = parent_value
self.subplan_value = subplan_value
def __eq__(self, other: object) -> bool:
"""Check equality based on all fields."""
if not isinstance(other, MergeConflict):
return NotImplemented
return (
self.path == other.path
and self.parent_value == other.parent_value
and self.subplan_value == other.subplan_value
)
def __repr__(self) -> str:
"""Return string representation of the conflict."""
return (
f"MergeConflict(path={self.path!r}, "
f"parent_value={self.parent_value!r}, "
f"subplan_value={self.subplan_value!r})"
)
class MergeStrategyService:
"""Service for applying merge strategies to resolve conflicts.
This service handles the resolution of conflicts that occur during
three-way merges of parent plans with subplans.
"""
def __init__(self, strategy: MergeStrategy = MergeStrategy.MANUAL) -> None:
"""Initialize the merge strategy service.
Args:
strategy: The merge strategy to use (default: MANUAL)
"""
self.strategy = strategy
def resolve_conflicts(
self, conflicts: list[MergeConflict]
) -> dict[str, Any]:
"""Resolve conflicts using the configured merge strategy.
Args:
conflicts: List of conflicts to resolve
Returns:
Dictionary mapping conflict paths to resolved values
Raises:
ValueError: If strategy is MANUAL and conflicts exist
"""
if not conflicts:
return {}
if self.strategy.is_manual():
raise ValueError(
f"Cannot auto-resolve {len(conflicts)} conflicts with "
f"MANUAL merge strategy. Conflicts must be resolved manually."
)
resolved: dict[str, Any] = {}
for conflict in conflicts:
if self.strategy == MergeStrategy.PREFER_PARENT:
resolved[conflict.path] = conflict.parent_value
elif self.strategy == MergeStrategy.PREFER_SUBPLAN:
resolved[conflict.path] = conflict.subplan_value
return resolved
def has_conflicts(self, conflicts: list[MergeConflict]) -> bool:
"""Check if there are any conflicts.
Args:
conflicts: List of conflicts to check
Returns:
True if there are conflicts, False otherwise
"""
return len(conflicts) > 0
def can_auto_resolve(self) -> bool:
"""Check if the current strategy can auto-resolve conflicts.
Returns:
True if the strategy can auto-resolve, False otherwise
"""
return self.strategy.is_auto_resolve()