cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(plans): implement conflict detection and structured conflict report for three-way merge - Closes #9558 #9613

Open
HAL9000 wants to merge 7 commits from feat/v3.3.0-merge-conflict-detection into master
pull from: feat/v3.3.0-merge-conflict-detection
merge into: cleveragents:master
Conversation 21 Commits 7 Files Changed 6
+1262 -96
6 changed files with 1262 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/plan_merge_conflict_detection.feature
+184
View File
@@ -0,0 +1,184 @@
Feature: Three-way merge conflict detection for plans
As a plan executor
I want to detect conflicts when both parent plan and subplan modify the same field
So that I can surface them to the user with full context and allow resolution
Background:
Given I have a conflict detector initialized
And I have ancestor plan data with fields:
| field_name | value |
| name | base_plan |
| timeout | 300 |
| retries | 3 |
| tags | [base] |
Scenario: No conflict when only parent modifies a field
Given parent plan modifies field "name" to "parent_plan"
And subplan does not modify field "name"
When I detect conflicts between ancestor, parent, and subplan
Then no plan merge conflicts should be detected
Scenario: No conflict when only subplan modifies a field
Given parent plan does not modify field "timeout"
And subplan modifies field "timeout" to 600
When I detect conflicts between ancestor, parent, and subplan
Then no plan merge conflicts should be detected
Scenario: No conflict when both modify to the same value
Given parent plan modifies field "retries" to 5
And subplan modifies field "retries" to 5
When I detect conflicts between ancestor, parent, and subplan
Then no plan merge conflicts should be detected
Scenario: Detect MODIFY_MODIFY conflict
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
Then 1 plan merge conflict should be detected
And the conflict should have type "modify_modify"
And the conflict should have ancestor value 300
And the conflict should have parent value 500
And the conflict should have subplan value 700
Scenario: Detect MODIFY_DELETE conflict
Given parent plan modifies field "retries" to 10
And subplan deletes field "retries"
When I detect conflicts between ancestor, parent, and subplan
Then 1 plan merge conflict should be detected
And the conflict should have type "modify_delete"
And the conflict should have parent value 10
And the conflict should have subplan value None
Scenario: Detect DELETE_MODIFY conflict
Given parent plan deletes field "tags"
And subplan modifies field "tags" to "[subplan]"
When I detect conflicts between ancestor, parent, and subplan
Then 1 plan merge conflict should be detected
And the conflict should have type "delete_modify"
And the conflict should have parent value None
And the conflict should have subplan value "[subplan]"
Scenario: Multiple conflicts in single merge
Given parent plan modifies field "name" to "parent_plan"
And subplan modifies field "name" to "subplan_plan"
And parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
And parent plan modifies field "retries" to 10
And subplan does not modify field "retries"
When I detect conflicts between ancestor, parent, and subplan
Then 2 plan merge conflicts should be detected
And conflict for field "name" should have type "modify_modify"
And conflict for field "timeout" should have type "modify_modify"
Scenario: Conflict report contains full context
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
Then the conflict report should have plan_id set
And the conflict report should have subplan_id set
And the conflict report should have has_conflicts as true
And the conflict report should have conflict_count as 1
Scenario: Resolve conflict using parent version
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
And I resolve conflict for field "timeout" using "parent"
Then the resolved value for field "timeout" should be 500
Scenario: Resolve conflict using subplan version
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
And I resolve conflict for field "timeout" using "subplan"
Then the resolved value for field "timeout" should be 700
Scenario: Auto-resolve MODIFY_DELETE conflict
Given parent plan modifies field "retries" to 10
And subplan deletes field "retries"
When I detect conflicts between ancestor, parent, and subplan
And I auto-resolve conflicts
Then conflict for field "retries" should be resolved with "parent"
Scenario: Auto-resolve DELETE_MODIFY conflict
Given parent plan deletes field "tags"
And subplan modifies field "tags" to "[subplan]"
When I detect conflicts between ancestor, parent, and subplan
And I auto-resolve conflicts
Then conflict for field "tags" should be resolved with "subplan"
Scenario: MODIFY_MODIFY conflict requires manual resolution
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
And I auto-resolve conflicts
Then conflict for field "timeout" should not be resolved
Scenario: Cannot add duplicate conflict
Given I have a conflict report
When I try to add a conflict for field "name" twice
Then a merge conflict error should be raised with message containing "already exists"
Scenario: Cannot resolve non-existent conflict
Given I have a conflict report with no conflicts
When I try to resolve conflict for field "name"
Then a merge conflict error should be raised with message containing "No conflict found"
Scenario: Cannot get resolved value for unresolved conflict
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
And I try to get resolved value for field "timeout"
Then a merge conflict error should be raised with message containing "not resolved"
Scenario: Unresolved conflicts list
Given parent plan modifies field "name" to "parent_plan"
And subplan modifies field "name" to "subplan_plan"
And parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
And I resolve conflict for field "name" using "parent"
Then unresolved conflicts count should be 1
And resolved conflicts count should be 1
Scenario: Empty ancestor version
Given ancestor plan data is empty
And parent plan has field "name" with value "parent_plan"
And subplan has field "name" with value "subplan_plan"
When I detect conflicts between ancestor, parent, and subplan
Then 1 plan merge conflict should be detected
And the conflict should have type "add_add"
Scenario: Conflict with complex nested values
Given ancestor plan has field "config" with value "{}"
And parent plan modifies field "config" to "{'timeout': 500}"
And subplan modifies field "config" to "{'timeout': 700}"
When I detect conflicts between ancestor, parent, and subplan
Then 1 plan merge conflict should be detected
And the conflict should have type "modify_modify"
Scenario: No conflict when values are identical in all three versions
Given parent plan does not modify field "timeout"
And subplan does not modify field "timeout"
When I detect conflicts between ancestor, parent, and subplan
Then no plan merge conflicts should be detected
Scenario: Cannot add conflict with empty field path
Given I have a conflict report
When I try to add a conflict with empty field path
Then a merge conflict error should be raised with message containing "field_path cannot be empty"
Scenario: Resolve conflict using merged version
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
And I resolve conflict for field "timeout" using "merged"
Then the resolved value for field "timeout" should be 500
Scenario: Get resolved value with manual resolution raises error
Given parent plan modifies field "timeout" to 500
And subplan modifies field "timeout" to 700
When I detect conflicts between ancestor, parent, and subplan
And I resolve conflict for field "timeout" using "manual"
And I try to get resolved value for field "timeout"
Then a merge conflict error should be raised with message containing "Unknown resolution strategy"
features/steps/cli_steps.py
+20 -2
View File
@@ -20,6 +20,24 @@ from cleveragents.cli.main import (
)
def _context_cli_output(context: Any) -> str:
"""Return CLI output captured by any shared step family."""
for attr in ("output", "command_output"):
value = getattr(context, attr, "")
if value:
return str(value)
cli_result = getattr(context, "cli_result", None)
value = getattr(cli_result, "output", "") if cli_result is not None else ""
if value:
return str(value)
result = getattr(context, "result", None)
if isinstance(result, dict):
return str(result.get("output", ""))
return str(getattr(result, "output", "") if result is not None else "")
def _run_cli(context: Any, args: Sequence[str]) -> None:
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer):
@@ -79,13 +97,13 @@ def step_run_cli_with_args(context, args):
@then('the CLI output should contain "{text}"')
def step_cli_output_contains(context, text):
output = getattr(context, "output", "") or context.result.get("output", "")
output = _context_cli_output(context)
assert text in output, f"Expected '{text}' in output: {output}"
@then('the CLI output should not contain "{text}"')
def step_cli_output_not_contains(context, text):
output = getattr(context, "output", "") or context.result.get("output", "")
output = _context_cli_output(context)
assert text not in output, f"Did not expect '{text}' in output: {output}"
features/steps/plan_merge_conflict_detection_steps.py
+370
View File
@@ -0,0 +1,370 @@
"""Step definitions for three-way merge conflict detection."""
from behave import given, then, when
from cleveragents.domain.models.planconfig import (
ConflictContext,
ConflictReport,
ConflictResolution,
ConflictType,
ThreeWayMergeConflictDetector,
)
@given("I have a conflict detector initialized")
def step_initialize_detector(context):
"""Initialize the conflict detector."""
context.detector = ThreeWayMergeConflictDetector()
context.ancestor = {}
context.parent = {}
context.subplan = {}
context.report = None
@given("I have ancestor plan data with fields:")
def step_set_ancestor_data(context):
"""Set ancestor plan data from table."""
context.ancestor = {}
for row in context.table:
field_name = row["field_name"]
value = row["value"]
# Parse value
if value == "[base]":
value = ["base"]
elif value.startswith("[") and value.endswith("]"):
value = value[1:-1].split(", ")
elif value.isdigit():
value = int(value)
context.ancestor[field_name] = value
@given('parent plan modifies field "{field}" to {value}')
def step_parent_modifies_field(context, field, value):
"""Set parent plan field value."""
value = _parse_value(value)
context.parent[field] = value
@given('subplan modifies field "{field}" to {value}')
def step_subplan_modifies_field(context, field, value):
"""Set subplan field value."""
value = _parse_value(value)
context.subplan[field] = value
def _parse_value(value):
if value == "None":
return None
if len(value) >= 2 and value[0] == value[-1] == '"':
return value[1:-1]
elif value.isdigit():
return int(value)
elif value.startswith("[") and value.endswith("]"):
return value[1:-1].split(", ")
return value
@given('parent plan does not modify field "{field}"')
def step_parent_does_not_modify(context, field):
"""Ensure parent doesn't modify field (keep ancestor value)."""
if field in context.ancestor:
context.parent[field] = context.ancestor[field]
@given('subplan does not modify field "{field}"')
def step_subplan_does_not_modify(context, field):
"""Ensure subplan doesn't modify field (keep ancestor value)."""
if field in context.ancestor:
context.subplan[field] = context.ancestor[field]
@given('parent plan deletes field "{field}"')
def step_parent_deletes_field(context, field):
"""Mark field as deleted in parent (set to None)."""
context.parent[field] = None
@given('subplan deletes field "{field}"')
def step_subplan_deletes_field(context, field):
"""Mark field as deleted in subplan (set to None)."""
context.subplan[field] = None
@given("I have a conflict report")
def step_create_empty_report(context):
"""Create an empty conflict report."""
context.report = ConflictReport(plan_id="plan1", subplan_id="subplan1")
@given("I have a conflict report with no conflicts")
def step_create_report_no_conflicts(context):
"""Create a conflict report with no conflicts."""
context.report = ConflictReport(plan_id="plan1", subplan_id="subplan1")
@given("ancestor plan data is empty")
def step_empty_ancestor(context):
"""Set ancestor to empty."""
context.ancestor = {}
@given('parent plan has field "{field}" with value "{value}"')
def step_parent_has_field(context, field, value):
"""Set parent field value."""
if value == "None":
value = None
elif value.isdigit():
value = int(value)
context.parent[field] = value
@given('subplan has field "{field}" with value "{value}"')
def step_subplan_has_field(context, field, value):
"""Set subplan field value."""
if value == "None":
value = None
elif value.isdigit():
value = int(value)
context.subplan[field] = value
@given('ancestor plan has field "{field}" with value "{value}"')
def step_ancestor_has_field(context, field, value):
"""Set ancestor field value."""
if value == "None":
value = None
elif value.isdigit():
value = int(value)
elif value == "{}":
value = {}
context.ancestor[field] = value
@when("I detect conflicts between ancestor, parent, and subplan")
def step_detect_conflicts(context):
"""Detect conflicts using the detector."""
context.report = context.detector.detect_conflicts(
plan_id="plan1",
subplan_id="subplan1",
ancestor=context.ancestor,
parent=context.parent,
subplan=context.subplan,
)
@when('I resolve conflict for field "{field}" using "{resolution}"')
def step_resolve_conflict(context, field, resolution):
"""Resolve a conflict."""
resolution_enum = ConflictResolution[resolution.upper()]
context.report.resolve_conflict(field, resolution_enum)
@when("I auto-resolve conflicts")
def step_auto_resolve(context):
"""Auto-resolve conflicts."""
context.report = context.detector.auto_resolve_conflicts(context.report)
@when("I try to add a conflict with empty field path")
def step_try_add_empty_field_path(context):
"""Try to add a conflict with no field path set."""
ctx = ConflictContext()
try:
context.report.add_conflict(ctx)
context.error = None
except ValueError as e:
context.error = str(e)
@when('I try to add a conflict for field "{field}" twice')
def step_try_add_duplicate(context, field):
"""Try to add duplicate conflict."""
conflict = ConflictContext(
field_path=field,
ancestor_value="value1",
parent_value="value2",
subplan_value="value3",
conflict_type=ConflictType.MODIFY_MODIFY,
)
context.report.add_conflict(conflict)
try:
context.report.add_conflict(conflict)
context.error = None
except ValueError as e:
context.error = str(e)
@when('I try to resolve conflict for field "{field}"')
def step_try_resolve_nonexistent(context, field):
"""Try to resolve non-existent conflict."""
try:
context.report.resolve_conflict(field, ConflictResolution.PARENT)
context.error = None
except ValueError as e:
context.error = str(e)
@when('I try to get resolved value for field "{field}"')
def step_try_get_unresolved_value(context, field):
"""Try to get value for unresolved conflict."""
try:
context.report.get_resolved_value(field)
context.error = None
except ValueError as e:
context.error = str(e)
@then("no plan merge conflicts should be detected")
def step_no_conflicts(context):
"""Assert no conflicts detected."""
assert context.report.conflict_count == 0, (
f"Expected 0 conflicts, got {context.report.conflict_count}"
)
@then("{count} plan merge conflict should be detected")
@then("{count} plan merge conflicts should be detected")
def step_conflict_count(context, count):
"""Assert conflict count."""
count = int(count)
assert context.report.conflict_count == count, (
f"Expected {count} conflicts, got {context.report.conflict_count}"
)
@then('the conflict should have type "{conflict_type}"')
def step_conflict_type(context, conflict_type):
"""Assert conflict type."""
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
assert conflict.conflict_type.presentation_value == conflict_type, (
f"Expected {conflict_type}, got {conflict.conflict_type.presentation_value}"
)
@then("the conflict should have ancestor value {value}")
def step_conflict_ancestor_value(context, value):
"""Assert ancestor value in conflict."""
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
value = _parse_value(value)
assert conflict.ancestor_value == value, (
f"Expected {value}, got {conflict.ancestor_value}"
)
@then("the conflict should have parent value {value}")
def step_conflict_parent_value(context, value):
"""Assert parent value in conflict."""
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
value = _parse_value(value)
assert conflict.parent_value == value, (
f"Expected {value}, got {conflict.parent_value}"
)
@then("the conflict should have subplan value {value}")
def step_conflict_subplan_value(context, value):
"""Assert subplan value in conflict."""
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
value = _parse_value(value)
assert conflict.subplan_value == value, (
f"Expected {value}, got {conflict.subplan_value}"
)
@then('conflict for field "{field}" should have type "{conflict_type}"')
def step_field_conflict_type(context, field, conflict_type):
"""Assert conflict type for specific field."""
conflict = next(
(c for c in context.report.conflicts if c.field_path == field), None
)
assert conflict is not None, f"No conflict found for field '{field}'"
assert conflict.conflict_type.presentation_value == conflict_type, (
f"Expected {conflict_type}, got {conflict.conflict_type.presentation_value}"
)
@then("the conflict report should have plan_id set")
def step_report_has_plan_id(context):
"""Assert report has plan_id."""
assert context.report.plan_id is not None, "plan_id not set"
assert context.report.plan_id == "plan1", (
f"Expected 'plan1', got {context.report.plan_id}"
)
@then("the conflict report should have subplan_id set")
def step_report_has_subplan_id(context):
"""Assert report has subplan_id."""
assert context.report.subplan_id is not None, "subplan_id not set"
assert context.report.subplan_id == "subplan1", (
f"Expected 'subplan1', got {context.report.subplan_id}"
)
@then("the conflict report should have has_conflicts as true")
def step_report_has_conflicts(context):
"""Assert report has_conflicts is true."""
assert context.report.has_conflicts is True, "has_conflicts should be True"
@then("the conflict report should have conflict_count as {count}")
def step_report_conflict_count(context, count):
"""Assert report conflict_count."""
count = int(count)
assert context.report.conflict_count == count, (
f"Expected {count}, got {context.report.conflict_count}"
)
@then('the resolved value for field "{field}" should be {value}')
def step_resolved_value(context, field, value):
"""Assert resolved value."""
value = _parse_value(value)
resolved = context.report.get_resolved_value(field)
assert resolved == value, f"Expected {value}, got {resolved}"
@then('conflict for field "{field}" should be resolved with "{resolution}"')
def step_conflict_resolved_with(context, field, resolution):
"""Assert conflict is resolved with specific strategy."""
assert field in context.report.resolutions, f"Field '{field}' not resolved"
actual = context.report.resolutions[field].value
expected = resolution.lower()
assert actual == expected, f"Expected {expected}, got {actual}"
@then('conflict for field "{field}" should not be resolved')
def step_conflict_not_resolved(context, field):
"""Assert conflict is not resolved."""
assert field not in context.report.resolutions, (
f"Field '{field}' should not be resolved"
)
@then("unresolved conflicts count should be {count}")
def step_unresolved_count(context, count):
"""Assert unresolved conflicts count."""
count = int(count)
actual = len(context.report.unresolved_conflicts)
assert actual == count, f"Expected {count} unresolved, got {actual}"
@then("resolved conflicts count should be {count}")
def step_resolved_count(context, count):
"""Assert resolved conflicts count."""
count = int(count)
actual = len(context.report.resolved_conflicts)
assert actual == count, f"Expected {count} resolved, got {actual}"
@then('a merge conflict error should be raised with message containing "{message}"')
def step_error_raised(context, message):
"""Assert error was raised with message."""
assert context.error is not None, "Expected error but none was raised"
assert message in context.error, (
f"Expected message containing '{message}', got '{context.error}'"
)
src/cleveragents/domain/models/planconfig/__init__.py
+2
View File
@@ -3,6 +3,7 @@
from .merge_conflict import (
ConflictContext,
ConflictReport,
ConflictResolution,
ConflictType,
MergeResult,
ThreeWayMergeConflictDetector,
@@ -12,6 +13,7 @@ from .plan_config import * # noqa: F403
__all__ = [
"ConflictContext",
"ConflictReport",
"ConflictResolution",
"ConflictType",
"MergeResult",
"ThreeWayMergeConflictDetector",
src/cleveragents/domain/models/planconfig/merge_conflict.py
+290 -94
View File
@@ -5,13 +5,13 @@ from __future__ import annotations
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from pydantic import ConfigDict, Field
from cleveragents.domain.models.base import DomainBaseModel
class ConflictType(StrEnum):
"""Categorises the nature of a merge conflict between three versions."""
"""Types of conflicts that can occur during a three-way merge."""
MODIFY_MODIFY = "MODIFY_MODIFY"
MODIFY_DELETE = "MODIFY_DELETE"
@@ -19,26 +19,61 @@ class ConflictType(StrEnum):
DELETE_DELETE = "DELETE_DELETE"
ADD_ADD = "ADD_ADD"
def __str__(self) -> str:
"""Render legacy summaries with the enum member name."""
return self.name
class ConflictContext(BaseModel):
"""Holds the three competing versions of a single field during merge."""
@property
def presentation_value(self) -> str:
"""Return the lowercase presentation value used by aggregate reports."""
return self.value.lower()
model_config = ConfigDict(frozen=True)
ancestor_value: str | None = None
parent_value: str | None = None
subplan_value: str | None = None
class ConflictResolution(StrEnum):
"""Resolution strategies for conflicts."""
PARENT = "parent"
SUBPLAN = "subplan"
MANUAL = "manual"
MERGED = "merged"
class ConflictContext(DomainBaseModel):
"""Context information for a conflicting field."""
model_config = ConfigDict(use_enum_values=False)
field_path: str | None = None
ancestor_value: Any = None
parent_value: Any = None
subplan_value: Any = None
conflict_type: ConflictType = ConflictType.MODIFY_MODIFY
class ConflictReport(DomainBaseModel):
"""Structured report of all conflicts found during a three-way merge."""
"""Structured conflict report.
key: str
conflict_type: ConflictType
context: ConflictContext
The model supports both a single-field conflict report used by ``detect()``
and an aggregate report used by ``detect_conflicts()``.
"""
model_config = ConfigDict(use_enum_values=False)
key: str | None = None
conflict_type: ConflictType | None = None
context: ConflictContext | None = None
resolution_strategy: str | None = None
plan_id: str | None = None
subplan_id: str | None = None
conflicts: list[ConflictContext] = Field(default_factory=list)
resolutions: dict[str, ConflictResolution] = Field(default_factory=dict)
@property
def resolved_value(self) -> str | None:
def resolved_value(self) -> Any:
"""Return the automatic single-conflict value, if one is known."""
if self.context is None or self.conflict_type is None: # pragma: no cover
return None
ctx = self.context
if self.conflict_type == ConflictType.MODIFY_DELETE:
return ctx.parent_value
@@ -52,32 +87,101 @@ class ConflictReport(DomainBaseModel):
return ctx.parent_value
return None
@property
def has_conflicts(self) -> bool:
"""Return whether the aggregate report contains any conflicts."""
return bool(self.conflicts)
@property
def conflict_count(self) -> int:
"""Return the number of conflicts in the aggregate report."""
return len(self.conflicts)
@property
def unresolved_conflicts(self) -> list[ConflictContext]:
"""Return aggregate conflicts that do not have a recorded resolution."""
return [
conflict
for conflict in self.conflicts
if conflict.field_path not in self.resolutions
]
@property
def resolved_conflicts(self) -> list[ConflictContext]:
"""Return aggregate conflicts that have a recorded resolution."""
return [
conflict
for conflict in self.conflicts
if conflict.field_path in self.resolutions
]
def add_conflict(self, conflict: ConflictContext) -> None:
"""Add a conflict to the aggregate report."""
if not conflict.field_path:
raise ValueError("field_path cannot be empty")
if any(c.field_path == conflict.field_path for c in self.conflicts):
raise ValueError(
f"Conflict for field '{conflict.field_path}' already exists"
)
self.conflicts.append(conflict)
def resolve_conflict(self, field_path: str, resolution: ConflictResolution) -> None:
"""Record a resolution strategy for a conflict."""
if not any(c.field_path == field_path for c in self.conflicts):
raise ValueError(f"No conflict found for field '{field_path}'")
self.resolutions[field_path] = resolution
def get_resolved_value(self, field_path: str) -> Any:
"""Return the value selected by an aggregate conflict resolution."""
if field_path not in self.resolutions:
raise ValueError(f"Field '{field_path}' is not resolved")
conflict = next((c for c in self.conflicts if c.field_path == field_path), None)
if conflict is None: # pragma: no cover
raise ValueError(f"No conflict found for field '{field_path}'")
resolution = self.resolutions[field_path]
if resolution == ConflictResolution.PARENT:
return conflict.parent_value
if resolution == ConflictResolution.SUBPLAN:
return conflict.subplan_value
if resolution == ConflictResolution.MERGED:
return conflict.parent_value
raise ValueError(f"Unknown resolution strategy: {resolution}")
class MergeResult(DomainBaseModel):
"""The output of a complete three-way merge operation."""
merged_config: dict[str, Any] = Field(default_factory=dict)
conflicts: list[ConflictReport] = Field(default_factory=list)
auto_resolved: dict[str, str | None] = Field(default_factory=dict)
auto_resolved: dict[str, Any] = Field(default_factory=dict)
class ThreeWayMergeConflictDetector:
"""Detects and resolves conflicts during a three-way plan merge."""
def __init__(
self, ancestor: dict[str, Any], parent: dict[str, Any], subplan: dict[str, Any]
self,
ancestor: dict[str, Any] | None = None,
parent: dict[str, Any] | None = None,
subplan: dict[str, Any] | None = None,
) -> None:
self.ancestor = ancestor or {}
self.parent = parent or {}
self.subplan = subplan or {} # -- Public API --
self.subplan = subplan or {}
def detect(self) -> MergeResult:
"""Run the three-way merge and return conflicts plus merged result."""
merged: dict[str, Any] = {}
conflicts: list[ConflictReport] = []
auto_resolved: dict[str, str | None] = {}
auto_resolved: dict[str, Any] = {}
for key in sorted(set(self.parent.keys()) | set(self.subplan.keys())):
for key in sorted(
set(self.ancestor.keys())
| set(self.parent.keys())
| set(self.subplan.keys())
):
ancestor_has_key = key in self.ancestor
parent_has_key = key in self.parent
subplan_has_key = key in self.subplan
@@ -86,30 +190,35 @@ class ThreeWayMergeConflictDetector:
parent_val = self.parent.get(key) if parent_has_key else None
subplan_val = self.subplan.get(key) if subplan_has_key else None
ancestor_s = str(ancestor_val) if ancestor_val is not None else None
parent_s = str(parent_val) if parent_val is not None else None
subplan_s = str(subplan_val) if subplan_val is not None else None
if ancestor_has_key and not parent_has_key and not subplan_has_key:
continue
if parent_s == subplan_s and parent_s == ancestor_s:
merged[key] = parent_s
if (
ancestor_has_key
and parent_has_key
and subplan_has_key
and parent_val == subplan_val == ancestor_val
):
merged[key] = _legacy_stringify(parent_val)
continue
conflict = self._analyse_field(
key,
ancestor_s,
parent_s,
subplan_s,
ancestor_val,
parent_val,
subplan_val,
ancestor_has_key=ancestor_has_key,
parent_has_key=parent_has_key,
subplan_has_key=subplan_has_key,
)
rv = conflict.resolved_value
if rv is not None:
merged[key] = rv
auto_resolved[key] = rv
resolved = conflict.resolved_value
if resolved is not None:
legacy_resolved = _legacy_stringify(resolved)
merged[key] = legacy_resolved
auto_resolved[key] = legacy_resolved
elif conflict.conflict_type == ConflictType.DELETE_DELETE:
pass # Both deleted it -- absent from output
continue
else:
conflicts.append(conflict)
@@ -119,6 +228,93 @@ class ThreeWayMergeConflictDetector:
auto_resolved=auto_resolved,
)
@staticmethod
def detect_conflicts(
plan_id: str,
subplan_id: str,
ancestor: dict[str, Any],
parent: dict[str, Any],
subplan: dict[str, Any],
) -> ConflictReport:
"""Detect conflicts and return an aggregate report."""
report = ConflictReport(plan_id=plan_id, subplan_id=subplan_id)
all_keys = set()
all_keys.update(ancestor.keys() if ancestor else [])
all_keys.update(parent.keys() if parent else [])
all_keys.update(subplan.keys() if subplan else [])
for key in sorted(all_keys):
ancestor_val = ancestor.get(key) if ancestor else None
parent_val = parent.get(key) if parent else None
subplan_val = subplan.get(key) if subplan else None
conflict = ThreeWayMergeConflictDetector._detect_field_conflict(
key, ancestor_val, parent_val, subplan_val
)
if conflict is not None:
report.add_conflict(conflict)
return report
@staticmethod
def _detect_field_conflict(
field_path: str,
ancestor_val: Any,
parent_val: Any,
subplan_val: Any,
) -> ConflictContext | None:
"""Detect aggregate-report conflict details for a single field."""
parent_changed = ancestor_val != parent_val
subplan_changed = ancestor_val != subplan_val
if not parent_changed and not subplan_changed:
return None
if parent_changed and not subplan_changed:
return None
if subplan_changed and not parent_changed:
return None
if parent_val == subplan_val:
return None
parent_deleted = parent_val is None
subplan_deleted = subplan_val is None
if parent_deleted and subplan_deleted: # pragma: no cover
conflict_type = ConflictType.DELETE_DELETE
elif parent_deleted:
conflict_type = ConflictType.DELETE_MODIFY
elif subplan_deleted:
conflict_type = ConflictType.MODIFY_DELETE
elif ancestor_val is None:
conflict_type = ConflictType.ADD_ADD
else:
conflict_type = ConflictType.MODIFY_MODIFY
return ConflictContext(
field_path=field_path,
ancestor_value=ancestor_val,
parent_value=parent_val,
subplan_value=subplan_val,
conflict_type=conflict_type,
)
@staticmethod
def auto_resolve_conflicts(report: ConflictReport) -> ConflictReport:
"""Attempt to automatically resolve non-manual aggregate conflicts."""
for conflict in report.conflicts:
if conflict.conflict_type == ConflictType.DELETE_DELETE: # pragma: no cover
continue
if conflict.conflict_type == ConflictType.MODIFY_DELETE:
report.resolve_conflict(
conflict.field_path or "", ConflictResolution.PARENT
)
elif conflict.conflict_type == ConflictType.DELETE_MODIFY:
report.resolve_conflict(
conflict.field_path or "", ConflictResolution.SUBPLAN
)
return report
def detect_diff_text(self) -> str:
"""Human-readable summary of all conflicts."""
lines: list[str] = []
@@ -135,27 +331,23 @@ class ThreeWayMergeConflictDetector:
lines.append("")
for cr in result.conflicts:
if cr.context is None: # pragma: no cover
continue
ctx = cr.context
lines.append(f" Key: {cr.key}")
lines.append(f" Type: {cr.conflict_type}")
an = " Ancestor: " + (
ctx.ancestor_value if ctx.ancestor_value is not None else "(absent)"
)
pa = " Parent: " + (
ctx.parent_value if ctx.parent_value is not None else "(absent)"
)
su = " Subplan: " + (
ctx.subplan_value if ctx.subplan_value is not None else "(absent)"
)
lines.extend([an, pa, su, ""])
lines.append(f" Ancestor: {_format_value(ctx.ancestor_value)}")
lines.append(f" Parent: {_format_value(ctx.parent_value)}")
lines.append(f" Subplan: {_format_value(ctx.subplan_value)}")
lines.append("")
for k, v in result.auto_resolved.items():
lines.append(f" * Auto-resolved {k}: {v}")
for key, value in result.auto_resolved.items():
lines.append(f" * Auto-resolved {key}: {value}")
if result.merged_config:
lines.append("Merged config:")
for k, v in sorted(result.merged_config.items()):
lines.append(f" {k}: {v}")
for key, value in sorted(result.merged_config.items()):
lines.append(f" {key}: {value}")
return "\n".join(lines)
def describe_all(self) -> list[dict[str, Any]]:
@@ -163,13 +355,14 @@ class ThreeWayMergeConflictDetector:
result = self.detect()
summary: list[dict[str, Any]] = []
all_keys = sorted(
set(result.merged_config.keys()) | {cr.key for cr in result.conflicts}
set(result.merged_config.keys())
| {cr.key for cr in result.conflicts if cr.key}
)
for key in all_keys:
entry: dict[str, Any] = {"key": key}
cr = next((c for c in result.conflicts if c.key == key), None)
if cr is not None:
if cr is not None and cr.context is not None:
entry.update(
{
"status": "CONFLICT",
@@ -196,69 +389,72 @@ class ThreeWayMergeConflictDetector:
def _analyse_field(
self,
key: str,
ancestor_val: str | None,
parent_val: str | None,
subplan_val: str | None,
ancestor_val: Any,
parent_val: Any,
subplan_val: Any,
*,
ancestor_has_key: bool,
parent_has_key: bool,
subplan_has_key: bool,
) -> ConflictReport:
"""Classify a single field into a conflict type."""
p_changed = (ancestor_has_key != parent_has_key) or (
ancestor_has_key and parent_has_key and str(ancestor_val) != str(parent_val)
parent_changed = (ancestor_has_key != parent_has_key) or (
ancestor_has_key and parent_has_key and ancestor_val != parent_val
)
s_changed = (ancestor_has_key != subplan_has_key) or (
ancestor_has_key
and subplan_has_key
and str(ancestor_val) != str(subplan_val)
subplan_changed = (ancestor_has_key != subplan_has_key) or (
ancestor_has_key and subplan_has_key and ancestor_val != subplan_val
)
if not p_changed and not s_changed:
if ancestor_has_key and not parent_has_key and not subplan_has_key:
ct = ConflictType.DELETE_DELETE
else:
ct = ConflictType.ADD_ADD
return ConflictReport(
key=key,
conflict_type=ct,
context=ConflictContext(
ancestor_value=ancestor_val,
parent_value=parent_val,
subplan_value=subplan_val,
),
if not parent_changed and not subplan_changed: # pragma: no cover
conflict_type = (
ConflictType.DELETE_DELETE
if ancestor_has_key and not parent_has_key and not subplan_has_key
else ConflictType.ADD_ADD
)
# Both sides changed independently from ancestor.
if p_changed and s_changed:
elif parent_changed and subplan_changed:
if not ancestor_has_key:
ct = ConflictType.ADD_ADD
elif parent_val is None:
ct = ConflictType.DELETE_MODIFY
elif subplan_val is None:
ct = ConflictType.MODIFY_DELETE
elif str(parent_val) == str(subplan_val):
ct = ConflictType.ADD_ADD
conflict_type = ConflictType.ADD_ADD
elif not parent_has_key:
conflict_type = ConflictType.DELETE_MODIFY
elif not subplan_has_key:
conflict_type = ConflictType.MODIFY_DELETE
elif parent_val == subplan_val:
conflict_type = ConflictType.ADD_ADD
else:
ct = ConflictType.MODIFY_MODIFY
elif p_changed and not s_changed:
if ancestor_val is not None and parent_val is None:
ct = ConflictType.DELETE_DELETE
else:
ct = ConflictType.MODIFY_DELETE
conflict_type = ConflictType.MODIFY_MODIFY
elif parent_changed:
conflict_type = (
ConflictType.DELETE_DELETE
if ancestor_has_key and not parent_has_key
else ConflictType.MODIFY_DELETE
)
else:
if ancestor_val is not None and subplan_val is None:
ct = ConflictType.DELETE_DELETE
else:
ct = ConflictType.DELETE_MODIFY
conflict_type = (
ConflictType.DELETE_DELETE
if ancestor_has_key and not subplan_has_key
else ConflictType.DELETE_MODIFY
)
return ConflictReport(
key=key,
conflict_type=ct,
conflict_type=conflict_type,
context=ConflictContext(
ancestor_value=ancestor_val,
parent_value=parent_val,
subplan_value=subplan_val,
field_path=key,
ancestor_value=_legacy_stringify(ancestor_val),
parent_value=_legacy_stringify(parent_val),
subplan_value=_legacy_stringify(subplan_val),
conflict_type=conflict_type,
),
)
def _legacy_stringify(value: Any) -> str | None:
if value is None:
return None
return str(value)
def _format_value(value: Any) -> str:
if value is None:
return "(absent)"
return str(value)
tests/domain/models/planconfig/test_merge_conflict_legacy_contract.py
+396
View File
@@ -0,0 +1,396 @@
import pytest
from cleveragents.domain.models.planconfig import (
ConflictContext,
ConflictReport,
ConflictResolution,
ConflictType,
ThreeWayMergeConflictDetector,
)
def test_detect_serializes_conflict_type_with_legacy_uppercase_value() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"timeout": 100},
parent={"timeout": 200},
subplan={"timeout": 300},
).detect()
dumped = result.model_dump(mode="json")
assert dumped["conflicts"][0]["conflict_type"] == "MODIFY_MODIFY"
assert dumped["conflicts"][0]["context"]["conflict_type"] == "MODIFY_MODIFY"
def test_detect_auto_resolved_timeout_uses_legacy_string_value() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"timeout": 100},
parent={"timeout": 500},
subplan={},
).detect()
assert result.auto_resolved["timeout"] == "500"
assert result.model_dump(mode="json")["auto_resolved"]["timeout"] == "500"
def test_detect_auto_resolves_delete_modify_with_subplan_value() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"tags": ["base"]},
parent={},
subplan={"tags": ["subplan"]},
).detect()
assert result.auto_resolved["tags"] == "['subplan']"
assert result.merged_config["tags"] == "['subplan']"
def test_detect_auto_resolves_identical_add_add_value() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={},
parent={"owner": "team-a"},
subplan={"owner": "team-a"},
).detect()
assert result.auto_resolved["owner"] == "team-a"
assert result.merged_config["owner"] == "team-a"
def test_detect_clean_merged_timeout_uses_legacy_string_value() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"timeout": 300},
parent={"timeout": 300},
subplan={"timeout": 300},
).detect()
assert result.merged_config["timeout"] == "300"
assert result.model_dump(mode="json")["merged_config"]["timeout"] == "300"
def test_detect_skips_both_deleted_field() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"retries": 3},
parent={},
subplan={},
).detect()
assert result.merged_config == {}
assert result.auto_resolved == {}
assert result.conflicts == []
def test_detect_reports_manual_modify_modify_conflict() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"timeout": 100},
parent={"timeout": 200},
subplan={"timeout": 300},
).detect()
conflict = result.conflicts[0]
assert conflict.key == "timeout"
assert conflict.conflict_type == ConflictType.MODIFY_MODIFY
assert str(conflict.conflict_type) == "MODIFY_MODIFY"
assert conflict.resolved_value is None
def test_detect_classifies_parent_only_delete_as_delete_delete() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"retries": 3},
parent={},
subplan={"retries": 3},
).detect()
assert result.conflicts == []
assert result.auto_resolved == {}
assert result.merged_config == {}
def test_detect_classifies_subplan_only_delete_as_delete_delete() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"retries": 3},
parent={"retries": 3},
subplan={},
).detect()
assert result.conflicts == []
assert result.auto_resolved == {}
assert result.merged_config == {}
def test_detect_classifies_matching_modifications_as_add_add_auto_resolved() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"timeout": 100},
parent={"timeout": 200},
subplan={"timeout": 200},
).detect()
assert result.auto_resolved["timeout"] == "200"
assert result.merged_config["timeout"] == "200"
def test_detect_conflicts_keeps_native_aggregate_values() -> None:
report = ThreeWayMergeConflictDetector.detect_conflicts(
plan_id="plan1",
subplan_id="subplan1",
ancestor={"timeout": 100},
parent={"timeout": 500},
subplan={"timeout": 600},
)
conflict = report.conflicts[0]
assert conflict.ancestor_value == 100
assert conflict.parent_value == 500
assert conflict.subplan_value == 600
assert conflict.conflict_type.presentation_value == "modify_modify"
def test_detect_conflicts_ignores_one_sided_changes() -> None:
parent_only = ThreeWayMergeConflictDetector.detect_conflicts(
plan_id="plan1",
subplan_id="sub1",
ancestor={"timeout": 100},
parent={"timeout": 200},
subplan={"timeout": 100},
)
subplan_only = ThreeWayMergeConflictDetector.detect_conflicts(
plan_id="plan1",
subplan_id="sub1",
ancestor={"timeout": 100},
parent={"timeout": 100},
subplan={"timeout": 200},
)
same_value = ThreeWayMergeConflictDetector.detect_conflicts(
plan_id="plan1",
subplan_id="sub1",
ancestor={"timeout": 100},
parent={"timeout": 200},
subplan={"timeout": 200},
)
assert parent_only.conflicts == []
assert subplan_only.conflicts == []
assert same_value.conflicts == []
def test_detect_conflicts_classifies_delete_modify_modify_delete_and_add_add() -> None:
report = ThreeWayMergeConflictDetector.detect_conflicts(
plan_id="plan1",
subplan_id="sub1",
ancestor={"tags": ["base"], "retries": 3},
parent={"tags": None, "retries": 10, "owner": "parent"},
subplan={"tags": ["subplan"], "retries": None, "owner": "subplan"},
)
by_field = {conflict.field_path: conflict for conflict in report.conflicts}
assert by_field["tags"].conflict_type == ConflictType.DELETE_MODIFY
assert by_field["retries"].conflict_type == ConflictType.MODIFY_DELETE
assert by_field["owner"].conflict_type == ConflictType.ADD_ADD
def test_conflict_report_resolved_value_returns_none_without_context() -> None:
report = ConflictReport(
key="x",
conflict_type=ConflictType.MODIFY_MODIFY,
context=None,
)
assert report.resolved_value is None
def test_add_conflict_raises_on_empty_field_path() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
ctx = ConflictContext()
with pytest.raises(ValueError, match="field_path cannot be empty"):
report.add_conflict(ctx)
def test_get_resolved_value_raises_when_conflict_missing_from_conflicts() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
report.resolutions["ghost"] = ConflictResolution.PARENT
with pytest.raises(ValueError, match="No conflict found"):
report.get_resolved_value("ghost")
def test_get_resolved_value_merged_returns_parent_value() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
ctx = ConflictContext(
field_path="timeout",
ancestor_value=100,
parent_value=200,
subplan_value=300,
conflict_type=ConflictType.MODIFY_MODIFY,
)
report.add_conflict(ctx)
report.resolve_conflict("timeout", ConflictResolution.MERGED)
assert report.get_resolved_value("timeout") == 200
def test_get_resolved_value_parent_and_subplan_return_selected_value() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
ctx = ConflictContext(
field_path="timeout",
ancestor_value=100,
parent_value=200,
subplan_value=300,
conflict_type=ConflictType.MODIFY_MODIFY,
)
report.add_conflict(ctx)
report.resolve_conflict("timeout", ConflictResolution.PARENT)
assert report.get_resolved_value("timeout") == 200
report.resolve_conflict("timeout", ConflictResolution.SUBPLAN)
assert report.get_resolved_value("timeout") == 300
def test_get_resolved_value_manual_raises_value_error() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
ctx = ConflictContext(
field_path="timeout",
ancestor_value=100,
parent_value=200,
subplan_value=300,
conflict_type=ConflictType.MODIFY_MODIFY,
)
report.add_conflict(ctx)
report.resolve_conflict("timeout", ConflictResolution.MANUAL)
with pytest.raises(ValueError, match="Unknown resolution strategy"):
report.get_resolved_value("timeout")
def test_detect_conflicts_unchanged_field_produces_no_conflict() -> None:
report = ThreeWayMergeConflictDetector.detect_conflicts(
plan_id="plan1",
subplan_id="sub1",
ancestor={"timeout": 300},
parent={"timeout": 300},
subplan={"timeout": 300},
)
assert not report.has_conflicts
def test_auto_resolve_skips_delete_delete_conflict() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
ctx = ConflictContext(
field_path="timeout",
ancestor_value=300,
parent_value=None,
subplan_value=None,
conflict_type=ConflictType.DELETE_DELETE,
)
report.add_conflict(ctx)
result = ThreeWayMergeConflictDetector.auto_resolve_conflicts(report)
assert "timeout" not in result.resolutions
def test_auto_resolve_resolves_modify_delete_and_delete_modify() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
report.add_conflict(
ConflictContext(
field_path="retries",
ancestor_value=3,
parent_value=10,
subplan_value=None,
conflict_type=ConflictType.MODIFY_DELETE,
)
)
report.add_conflict(
ConflictContext(
field_path="tags",
ancestor_value=["base"],
parent_value=None,
subplan_value=["subplan"],
conflict_type=ConflictType.DELETE_MODIFY,
)
)
result = ThreeWayMergeConflictDetector.auto_resolve_conflicts(report)
assert result.resolutions["retries"] == ConflictResolution.PARENT
assert result.resolutions["tags"] == ConflictResolution.SUBPLAN
def test_detect_diff_text_reports_conflicts_auto_resolved_and_merged_values() -> None:
summary = ThreeWayMergeConflictDetector(
ancestor={"name": "base", "timeout": 100, "retries": 3},
parent={"name": "parent", "timeout": 200, "retries": 3},
subplan={"name": "subplan", "timeout": 200, "retries": 3},
).detect_diff_text()
assert "Three-way merge: 1 conflict(s), 1 auto-resolved." in summary
assert "Key: name" in summary
assert "Type: MODIFY_MODIFY" in summary
assert "Auto-resolved timeout: 200" in summary
assert "Merged config:" in summary
assert "retries: 3" in summary
def test_detect_diff_text_reports_perfect_merge() -> None:
summary = ThreeWayMergeConflictDetector(
ancestor={"timeout": 100},
parent={"timeout": 100},
subplan={"timeout": 100},
).detect_diff_text()
assert summary == "No conflicts detected -- perfect merge."
def test_detect_diff_text_formats_absent_values() -> None:
summary = ThreeWayMergeConflictDetector(
ancestor={},
parent={"timeout": 100},
subplan={"timeout": 200},
).detect_diff_text()
assert "Ancestor: (absent)" in summary
def test_conflict_report_aggregate_properties_and_errors() -> None:
report = ConflictReport(plan_id="plan1", subplan_id="sub1")
ctx = ConflictContext(
field_path="timeout",
ancestor_value=100,
parent_value=200,
subplan_value=300,
conflict_type=ConflictType.MODIFY_MODIFY,
)
report.add_conflict(ctx)
assert report.conflict_count == 1
assert report.unresolved_conflicts == [ctx]
with pytest.raises(ValueError, match="already exists"):
report.add_conflict(ctx)
with pytest.raises(ValueError, match="No conflict found"):
report.resolve_conflict("missing", ConflictResolution.PARENT)
with pytest.raises(ValueError, match="not resolved"):
report.get_resolved_value("timeout")
report.resolve_conflict("timeout", ConflictResolution.PARENT)
assert report.resolved_conflicts == [ctx]
def test_describe_all_reports_conflict_auto_resolved_and_clean_statuses() -> None:
summary = ThreeWayMergeConflictDetector(
ancestor={"name": "base", "timeout": 100, "retries": 3},
parent={"name": "parent", "timeout": 200, "retries": 3},
subplan={"name": "subplan", "timeout": 200, "retries": 3},
).describe_all()
by_key = {entry["key"]: entry for entry in summary}
assert by_key["name"] == {
"key": "name",
"status": "CONFLICT",
"conflict_type": "MODIFY_MODIFY",
"ancestor": "base",
"parent": "parent",
"subplan": "subplan",
}
assert by_key["timeout"] == {
"key": "timeout",
"status": "AUTO_RESOLVED",
"resolved_to": "200",
}
assert by_key["retries"] == {
"key": "retries",
"status": "MERGED_OK",
"value": "3",
}