feat(plans): implement configurable merge strategy for three-way merge
CI / lint (pull_request) Failing after 18s
CI / push-validation (pull_request) Successful in 24s
CI / helm (pull_request) Successful in 24s
CI / quality (pull_request) Successful in 33s
CI / build (pull_request) Successful in 42s
CI / typecheck (pull_request) Failing after 51s
CI / security (pull_request) Failing after 1m16s
CI / coverage (pull_request) Has been skipped
CI / e2e_tests (pull_request) Successful in 4m55s
CI / integration_tests (pull_request) Successful in 6m52s
CI / unit_tests (pull_request) Failing after 8m10s
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 1s

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

- New module: src/cleveragents/domain/models/core/merge_strategy.py
  - MergeStrategy enum with options: prefer-parent, prefer-subplan, manual
  - Validation utilities and helper to determine auto-resolution capability for a given strategy

- New module: src/cleveragents/domain/models/core/merge_strategy_service.py
  - MergeConflict class to represent conflicts
  - MergeStrategyService to apply merge strategies
  - Support for resolving conflicts based on the configured strategy

- New BDD feature: features/plan_merge_strategy.feature
  - Tests for all three merge strategies
  - Tests for conflict resolution
  - Tests for error handling

- New step definitions: features/steps/plan_merge_strategy_steps.py
  - Step implementations for BDD tests

ISSUES CLOSED: #9559
This commit is contained in:
2026-04-15 00:38:39 +00:00
parent b752dd485f
commit 48704f8bd3
4 changed files with 396 additions and 0 deletions
+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, when, then
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)
assert False, "Expected ValueError for manual strategy"
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)
@@ -0,0 +1,70 @@
"""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 Enum
from typing import Literal
class MergeStrategy(str, Enum):
"""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: Literal["prefer-parent"] = "prefer-parent"
"""Auto-resolve conflicts in favor of parent plan."""
PREFER_SUBPLAN: Literal["prefer-subplan"] = "prefer-subplan"
"""Auto-resolve conflicts in favor of subplan."""
MANUAL: Literal["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,108 @@
"""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, Dict, Optional
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 __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()