feat(plans): implement conflict detection and structured conflict report for three-way merge (#11000) #11240

Merged
HAL9000 merged 4 commits from pr_fix-11000-conflict-report into master 2026-05-28 17:14:16 +00:00
4 changed files with 779 additions and 0 deletions
@@ -0,0 +1,360 @@
"""Steps for three_way_merge_conflict.feature."""
from __future__ import annotations
from behave import given, then, when
from cleveragents.domain.models.planconfig import (
ConflictContext,
ConflictReport,
ConflictType,
MergeResult,
ThreeWayMergeConflictDetector,
)
_CT = {
"MODIFY_MODIFY": ConflictType.MODIFY_MODIFY,
"MODIFY_DELETE": ConflictType.MODIFY_DELETE,
"DELETE_MODIFY": ConflictType.DELETE_MODIFY,
"DELETE_DELETE": ConflictType.DELETE_DELETE,
"ADD_ADD": ConflictType.ADD_ADD,
}
# ── detector setup ─────────────────────────────────────────────────────────────
@given('a three-way merge where all versions agree on key "{key}" value "{value}"')
def step_all_agree(context, key, value):
context.detector = ThreeWayMergeConflictDetector(
ancestor={key: value},
parent={key: value},
subplan={key: value},
)
@given(
'a three-way merge with ancestor "{k1}"="{v1}", parent "{k2}"="{v2}", subplan "{k3}"="{v3}"'
)
def step_three_modify(context, k1, v1, k2, v2, k3, v3):
context.detector = ThreeWayMergeConflictDetector(
ancestor={k1: v1},
parent={k2: v2},
subplan={k3: v3},
)
@given(
'a three-way merge with key "{key}", ancestor "{av}", parent changed to "{pv}", subplan deleted it'
)
def step_modify_delete(context, key, av, pv):
context.detector = ThreeWayMergeConflictDetector(
ancestor={key: av},
parent={key: pv},
subplan={},
)
@given(
'a three-way merge with key "{key}", ancestor "{av}", parent deleted it, subplan changed to "{sv}"'
)
def step_delete_modify(context, key, av, sv):
context.detector = ThreeWayMergeConflictDetector(
ancestor={key: av},
parent={},
subplan={key: sv},
)
@given(
'a three-way merge where ancestor lacks "{key}", parent and subplan both add it as "{value}"'
)
def step_add_same(context, key, value):
context.detector = ThreeWayMergeConflictDetector(
ancestor={},
parent={key: value},
subplan={key: value},
)
@given(
'a three-way merge where ancestor lacks "{key}", parent adds "{pv}" and subplan adds "{sv}"'
)
def step_add_different(context, key, pv, sv):
context.detector = ThreeWayMergeConflictDetector(
ancestor={},
parent={key: pv},
subplan={key: sv},
)
@given(
'a three-way merge where parent deletes "{key}" and subplan keeps it at "{value}"'
)
def step_parent_deletes_subplan_keeps(context, key, value):
context.detector = ThreeWayMergeConflictDetector(
ancestor={key: value},
parent={},
subplan={key: value},
)
@given(
'a three-way merge where subplan deletes "{key}" and parent keeps it at "{value}"'
)
def step_subplan_deletes_parent_keeps(context, key, value):
context.detector = ThreeWayMergeConflictDetector(
ancestor={key: value},
parent={key: value},
subplan={},
)
@given(
'a three-way merge where only parent changed key "{key}" from "{av}" to "{pv}", subplan kept "{sv}"'
)
def step_only_parent_changed(context, key, av, pv, sv):
context.detector = ThreeWayMergeConflictDetector(
ancestor={key: av},
parent={key: pv},
subplan={key: sv},
)
@given(
'a three-way merge where only subplan changed key "{key}" from "{av}" to "{sv}", parent kept "{pv}"'
)
def step_only_subplan_changed(context, key, av, sv, pv):
context.detector = ThreeWayMergeConflictDetector(
ancestor={key: av},
parent={key: pv},
subplan={key: sv},
)
@given("a mixed three-way merge scenario")
def step_mixed_scenario(context):
# conflict_key: MODIFY_MODIFY (both changed, differently)
# auto_key: MODIFY_DELETE auto-resolved to "new" (parent changed, subplan deleted)
# ok_key: unchanged in all three → MERGED_OK
context.detector = ThreeWayMergeConflictDetector(
ancestor={"conflict_key": "a", "auto_key": "old", "ok_key": "same"},
parent={"conflict_key": "b", "auto_key": "new", "ok_key": "same"},
subplan={"conflict_key": "c", "ok_key": "same"},
)
@given("a three-way merge detector initialized with all None inputs")
def step_none_inputs(context):
context.detector = ThreeWayMergeConflictDetector(None, None, None)
@given("I directly call analyse_field with all-False has_key flags and None values")
def step_analyse_field_direct(context):
detector = ThreeWayMergeConflictDetector({}, {}, {})
context.analyse_result = detector._analyse_field(
"x",
None,
None,
None,
ancestor_has_key=False,
parent_has_key=False,
subplan_has_key=False,
)
@given("I import the planconfig package and call its internal helper")
def step_import_planconfig(context):
from cleveragents.domain.models.planconfig import _get_all_from_plan_config
context.helper_result = _get_all_from_plan_config()
# Verify MergeResult is importable from the package (exercises __init__ exports)
assert MergeResult is not None
# ── ConflictReport setup ───────────────────────────────────────────────────────
@given(
'a ConflictReport of type "{ctype}" with ancestor "{av}" parent "{pv}" subplan "{sv}"'
)
def step_cr_all_values(context, ctype, av, pv, sv):
context.report = ConflictReport(
key="k",
conflict_type=_CT[ctype],
context=ConflictContext(ancestor_value=av, parent_value=pv, subplan_value=sv),
)
@given(
'a ConflictReport of type "{ctype}" with ancestor "{av}" parent "{pv}" and no subplan'
)
def step_cr_no_subplan(context, ctype, av, pv):
context.report = ConflictReport(
key="k",
conflict_type=_CT[ctype],
context=ConflictContext(ancestor_value=av, parent_value=pv, subplan_value=None),
)
@given(
'a ConflictReport of type "{ctype}" with ancestor "{av}" no parent and subplan "{sv}"'
)
def step_cr_no_parent(context, ctype, av, sv):
context.report = ConflictReport(
key="k",
conflict_type=_CT[ctype],
context=ConflictContext(ancestor_value=av, parent_value=None, subplan_value=sv),
)
@given(
'a ConflictReport of type "{ctype}" with no ancestor parent "{pv}" subplan "{sv}"'
)
def step_cr_no_ancestor(context, ctype, pv, sv):
context.report = ConflictReport(
key="k",
conflict_type=_CT[ctype],
context=ConflictContext(ancestor_value=None, parent_value=pv, subplan_value=sv),
)
# ── action steps ───────────────────────────────────────────────────────────────
@when("I run the three-way merge detector")
def step_run_detector(context):
context.result = context.detector.detect()
@when("I call detect_diff_text")
def step_call_diff_text(context):
context.diff_text = context.detector.detect_diff_text()
@when("I call describe_all")
def step_call_describe_all(context):
context.summary = context.detector.describe_all()
@when("I check the resolved_value")
def step_check_resolved(context):
context.resolved = context.report.resolved_value
# ── assertion steps ────────────────────────────────────────────────────────────
@then("the merge should have no conflicts")
def step_no_conflicts(context):
assert len(context.result.conflicts) == 0, (
f"Expected no conflicts but got: {context.result.conflicts}"
)
@then('the merged config should contain key "{key}" with value "{value}"')
def step_merged_contains(context, key, value):
assert key in context.result.merged_config, (
f"Key '{key}' not in merged_config: {context.result.merged_config}"
)
assert context.result.merged_config[key] == value, (
f"Expected '{value}' but got '{context.result.merged_config[key]}'"
)
@then("the merge should have {n:d} conflict")
def step_n_conflicts(context, n):
actual = len(context.result.conflicts)
assert actual == n, f"Expected {n} conflict(s) but got {actual}"
@then('the conflict key is "{key}" with type "{ctype}"')
def step_conflict_key_type(context, key, ctype):
assert any(
c.key == key and str(c.conflict_type) == ctype for c in context.result.conflicts
), f"No conflict key='{key}' type='{ctype}' in {context.result.conflicts}"
@then("the conflict has no auto-resolved entries")
def step_no_auto_resolved(context):
assert len(context.result.auto_resolved) == 0, (
f"Expected empty auto_resolved but got: {context.result.auto_resolved}"
)
@then('key "{key}" is auto-resolved to "{value}"')
def step_auto_resolved_to(context, key, value):
assert key in context.result.auto_resolved, (
f"Key '{key}' not in auto_resolved: {context.result.auto_resolved}"
)
assert context.result.auto_resolved[key] == value, (
f"Expected '{value}' but got '{context.result.auto_resolved[key]}'"
)
@then('the merged config should not contain key "{key}"')
def step_not_in_merged(context, key):
assert key not in context.result.merged_config, (
f"Key '{key}' unexpectedly present in merged_config: {context.result.merged_config}"
)
@then('the diff text should say "No conflicts detected"')
def step_diff_no_conflicts(context):
assert "No conflicts detected" in context.diff_text, (
f"Expected 'No conflicts detected' in:\n{context.diff_text!r}"
)
@then('the diff text should contain "{text}"')
def step_diff_contains(context, text):
assert text in context.diff_text, (
f"Expected '{text}' in diff text:\n{context.diff_text}"
)
@then("the summary should contain a CONFLICT entry")
def step_summary_conflict(context):
assert any(e["status"] == "CONFLICT" for e in context.summary), (
f"No CONFLICT entry in: {context.summary}"
)
@then("the summary should contain an AUTO_RESOLVED entry")
def step_summary_auto_resolved(context):
assert any(e["status"] == "AUTO_RESOLVED" for e in context.summary), (
f"No AUTO_RESOLVED entry in: {context.summary}"
)
@then("the summary should contain a MERGED_OK entry")
def step_summary_merged_ok(context):
assert any(e["status"] == "MERGED_OK" for e in context.summary), (
f"No MERGED_OK entry in: {context.summary}"
)
@then("the resolved_value should be None")
def step_resolved_none(context):
assert context.resolved is None, f"Expected None but got: {context.resolved!r}"
@then('the resolved_value should equal "{value}"')
def step_resolved_equals(context, value):
assert context.resolved == value, (
f"Expected '{value}' but got: {context.resolved!r}"
)
@then('the analyse_field result type should be "{ctype}"')
def step_analyse_result_type(context, ctype):
assert str(context.analyse_result.conflict_type) == ctype, (
f"Expected '{ctype}' but got: {context.analyse_result.conflict_type!r}"
)
@then("the helper returns a non-empty list")
def step_helper_non_empty(context):
assert isinstance(context.helper_result, list) and len(context.helper_result) > 0, (
f"Expected non-empty list but got: {context.helper_result!r}"
)
+134
View File
@@ -0,0 +1,134 @@
Feature: Three-Way Merge Conflict Detector
Exercises ThreeWayMergeConflictDetector, ConflictReport, ConflictContext,
MergeResult, and ConflictType for structured plan config merge conflict reporting.
Scenario: Identical values produce a clean merge
Given a three-way merge where all versions agree on key "x" value "hello"
When I run the three-way merge detector
Then the merge should have no conflicts
And the merged config should contain key "x" with value "hello"
Scenario: Both sides modify a key independently raises MODIFY_MODIFY
Given a three-way merge with ancestor "x"="a", parent "x"="b", subplan "x"="c"
When I run the three-way merge detector
Then the merge should have 1 conflict
And the conflict key is "x" with type "MODIFY_MODIFY"
And the conflict has no auto-resolved entries
Scenario: Both sides converge on the same new value is auto-resolved as ADD_ADD
Given a three-way merge with ancestor "x"="old", parent "x"="new", subplan "x"="new"
When I run the three-way merge detector
Then the merge should have no conflicts
And key "x" is auto-resolved to "new"
Scenario: Parent modifies and subplan deletes is auto-resolved to parent value
Given a three-way merge with key "x", ancestor "old", parent changed to "new", subplan deleted it
When I run the three-way merge detector
Then the merge should have no conflicts
And key "x" is auto-resolved to "new"
Scenario: Parent deletes and subplan modifies is auto-resolved to subplan value
Given a three-way merge with key "x", ancestor "old", parent deleted it, subplan changed to "new"
When I run the three-way merge detector
Then the merge should have no conflicts
And key "x" is auto-resolved to "new"
Scenario: Both sides add the same new key is auto-resolved
Given a three-way merge where ancestor lacks "x", parent and subplan both add it as "same"
When I run the three-way merge detector
Then the merge should have no conflicts
And key "x" is auto-resolved to "same"
Scenario: Both sides add different values for the same new key raises ADD_ADD conflict
Given a three-way merge where ancestor lacks "x", parent adds "val1" and subplan adds "val2"
When I run the three-way merge detector
Then the merge should have 1 conflict
And the conflict key is "x" with type "ADD_ADD"
Scenario: Parent deletes a key the subplan kept unchanged removes it from output
Given a three-way merge where parent deletes "x" and subplan keeps it at "val"
When I run the three-way merge detector
Then the merge should have no conflicts
And the merged config should not contain key "x"
Scenario: Subplan deletes a key the parent kept unchanged removes it from output
Given a three-way merge where subplan deletes "x" and parent keeps it at "val"
When I run the three-way merge detector
Then the merge should have no conflicts
And the merged config should not contain key "x"
Scenario: Only parent changes a key is auto-resolved to parent value
Given a three-way merge where only parent changed key "y" from "old" to "new", subplan kept "old"
When I run the three-way merge detector
Then the merge should have no conflicts
And key "y" is auto-resolved to "new"
Scenario: Only subplan changes a key is auto-resolved to subplan value
Given a three-way merge where only subplan changed key "y" from "old" to "new", parent kept "old"
When I run the three-way merge detector
Then the merge should have no conflicts
And key "y" is auto-resolved to "new"
Scenario: detect_diff_text returns a clean message when there are no conflicts
Given a three-way merge where all versions agree on key "z" value "ok"
When I call detect_diff_text
Then the diff text should say "No conflicts detected"
Scenario: detect_diff_text includes absent ancestor in ADD_ADD conflict report
Given a three-way merge where ancestor lacks "m", parent adds "v1" and subplan adds "v2"
When I call detect_diff_text
Then the diff text should contain "ADD_ADD"
And the diff text should contain "(absent)"
And the diff text should contain "Key: m"
Scenario: detect_diff_text includes auto-resolved and merged config sections
Given a mixed three-way merge scenario
When I call detect_diff_text
Then the diff text should contain "MODIFY_MODIFY"
And the diff text should contain "Auto-resolved"
And the diff text should contain "Merged config"
Scenario: describe_all returns CONFLICT AUTO_RESOLVED and MERGED_OK entries
Given a mixed three-way merge scenario
When I call describe_all
Then the summary should contain a CONFLICT entry
And the summary should contain an AUTO_RESOLVED entry
And the summary should contain a MERGED_OK entry
Scenario: ConflictReport resolved_value is None for MODIFY_MODIFY
Given a ConflictReport of type "MODIFY_MODIFY" with ancestor "a" parent "b" subplan "c"
When I check the resolved_value
Then the resolved_value should be None
Scenario: ConflictReport resolved_value returns parent value for MODIFY_DELETE
Given a ConflictReport of type "MODIFY_DELETE" with ancestor "a" parent "new_val" and no subplan
When I check the resolved_value
Then the resolved_value should equal "new_val"
Scenario: ConflictReport resolved_value returns subplan value for DELETE_MODIFY
Given a ConflictReport of type "DELETE_MODIFY" with ancestor "a" no parent and subplan "new_val"
When I check the resolved_value
Then the resolved_value should equal "new_val"
Scenario: ConflictReport resolved_value is None for ADD_ADD with mismatched values
Given a ConflictReport of type "ADD_ADD" with no ancestor parent "p1" subplan "p2"
When I check the resolved_value
Then the resolved_value should be None
Scenario: ConflictReport resolved_value returns parent for ADD_ADD with matching values
Given a ConflictReport of type "ADD_ADD" with no ancestor parent "same" subplan "same"
When I check the resolved_value
Then the resolved_value should equal "same"
Scenario: ThreeWayMergeConflictDetector handles None inputs gracefully
Given a three-way merge detector initialized with all None inputs
When I run the three-way merge detector
Then the merge should have no conflicts
Scenario: _analyse_field direct call covers the not-p-changed not-s-changed ADD_ADD branch
Given I directly call analyse_field with all-False has_key flags and None values
Then the analyse_field result type should be "ADD_ADD"
Scenario: planconfig package __init__ exports are importable and helpers callable
Given I import the planconfig package and call its internal helper
Then the helper returns a non-empty list
@@ -1,3 +1,24 @@
"""Plan Config models."""
from .merge_conflict import (
ConflictContext,
ConflictReport,
ConflictType,
MergeResult,
ThreeWayMergeConflictDetector,
)
from .plan_config import * # noqa: F403
__all__ = [
"ConflictContext",
"ConflictReport",
"ConflictType",
"MergeResult",
"ThreeWayMergeConflictDetector",
]
def _get_all_from_plan_config():
from .plan_config import AutoModeType, ConfigSetting, PlanConfig
return ["AutoModeType", "ConfigSetting", "PlanConfig"]
@@ -0,0 +1,264 @@
"""Structured conflict detection and reporting for three-way plan merges."""
from __future__ import annotations
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from cleveragents.domain.models.base import DomainBaseModel
class ConflictType(StrEnum):
"""Categorises the nature of a merge conflict between three versions."""
MODIFY_MODIFY = "MODIFY_MODIFY"
MODIFY_DELETE = "MODIFY_DELETE"
DELETE_MODIFY = "DELETE_MODIFY"
DELETE_DELETE = "DELETE_DELETE"
ADD_ADD = "ADD_ADD"
class ConflictContext(BaseModel):
"""Holds the three competing versions of a single field during merge."""
model_config = ConfigDict(frozen=True)
ancestor_value: str | None = None
parent_value: str | None = None
subplan_value: str | None = None
class ConflictReport(DomainBaseModel):
"""Structured report of all conflicts found during a three-way merge."""
key: str
conflict_type: ConflictType
context: ConflictContext
resolution_strategy: str | None = None
@property
def resolved_value(self) -> str | None:
ctx = self.context
if self.conflict_type == ConflictType.MODIFY_DELETE:
return ctx.parent_value
if self.conflict_type == ConflictType.DELETE_MODIFY:
return ctx.subplan_value
if (
self.conflict_type == ConflictType.ADD_ADD
and ctx.parent_value is not None
and ctx.parent_value == ctx.subplan_value
):
return ctx.parent_value
return None
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)
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]
) -> None:
self.ancestor = ancestor or {}
self.parent = parent or {}
self.subplan = subplan or {} # -- Public API --
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] = {}
for key in sorted(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
ancestor_val = self.ancestor.get(key) if ancestor_has_key else None
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 parent_s == subplan_s and parent_s == ancestor_s:
merged[key] = parent_s
continue
conflict = self._analyse_field(
key,
ancestor_s,
parent_s,
subplan_s,
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
elif conflict.conflict_type == ConflictType.DELETE_DELETE:
pass # Both deleted it -- absent from output
else:
conflicts.append(conflict)
return MergeResult(
merged_config=merged,
conflicts=conflicts,
auto_resolved=auto_resolved,
)
def detect_diff_text(self) -> str:
"""Human-readable summary of all conflicts."""
lines: list[str] = []
result = self.detect()
if not result.conflicts:
return "No conflicts detected -- perfect merge."
n_conflicts = len(result.conflicts)
n_resolved = len(result.auto_resolved)
lines.append(
f"Three-way merge: {n_conflicts} conflict(s), {n_resolved} auto-resolved."
)
lines.append("")
for cr in result.conflicts:
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, ""])
for k, v in result.auto_resolved.items():
lines.append(f" * Auto-resolved {k}: {v}")
if result.merged_config:
lines.append("Merged config:")
for k, v in sorted(result.merged_config.items()):
lines.append(f" {k}: {v}")
return "\n".join(lines)
def describe_all(self) -> list[dict[str, Any]]:
"""Serializable summary of every field's merge outcome."""
result = self.detect()
summary: list[dict[str, Any]] = []
all_keys = sorted(
set(result.merged_config.keys()) | {cr.key for cr in result.conflicts}
)
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:
entry.update(
{
"status": "CONFLICT",
"conflict_type": str(cr.conflict_type),
"ancestor": cr.context.ancestor_value,
"parent": cr.context.parent_value,
"subplan": cr.context.subplan_value,
}
)
elif key in result.auto_resolved:
entry.update(
{
"status": "AUTO_RESOLVED",
"resolved_to": result.auto_resolved[key],
}
)
else:
entry.update(
{"status": "MERGED_OK", "value": result.merged_config.get(key)}
)
summary.append(entry)
return summary
def _analyse_field(
self,
key: str,
ancestor_val: str | None,
parent_val: str | None,
subplan_val: str | None,
*,
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)
)
s_changed = (ancestor_has_key != subplan_has_key) or (
ancestor_has_key
and subplan_has_key
and str(ancestor_val) != str(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,
),
)
# Both sides changed independently from ancestor.
if p_changed and s_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
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
else:
if ancestor_val is not None and subplan_val is None:
ct = ConflictType.DELETE_DELETE
else:
ct = ConflictType.DELETE_MODIFY
return ConflictReport(
key=key,
conflict_type=ct,
context=ConflictContext(
ancestor_value=ancestor_val,
parent_value=parent_val,
subplan_value=subplan_val,
),
)