feat(plans): implement conflict detection and structured conflict report for three-way merge (#11000) #11240
@@ -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}"
|
||||
)
|
||||
@@ -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,
|
||||
),
|
||||
)
|
||||
Reference in New Issue
Block a user