diff --git a/features/steps/three_way_merge_engine_steps.py b/features/steps/three_way_merge_engine_steps.py new file mode 100644 index 000000000..756b5048b --- /dev/null +++ b/features/steps/three_way_merge_engine_steps.py @@ -0,0 +1,698 @@ +"""Step definitions for three-way merge engine tests.""" + +from datetime import datetime +from typing import Any + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.three_way_merge_engine import ( + MergeConflictError, + MergeValidationError, + ThreeWayMergeEngine, +) +from cleveragents.domain.models.core.cost_metadata import CostMetadata +from cleveragents.domain.models.core.plan import ( + Plan, + PlanIdentity, + PlanPhase, + PlanTimestamps, + ProcessingState, + SubplanStatus, + NamespacedName, +) + + +def _create_test_plan( + plan_id: str = "01ARZ3NDEKTSV4RRFFQ69G5FAV", + subplan_statuses: list[SubplanStatus] | None = None, + cost_metadata: CostMetadata | None = None, + error_message: str | None = None, +) -> Plan: + """Create a test plan with the given parameters.""" + return Plan( + identity=PlanIdentity(plan_id=plan_id), + namespaced_name=NamespacedName(namespace="local", name="test-plan"), + description="Test plan", + action_name="test-action", + phase=PlanPhase.EXECUTE, + processing_state=ProcessingState.PROCESSING, + subplan_statuses=subplan_statuses or [], + cost_metadata=cost_metadata, + error_message=error_message, + timestamps=PlanTimestamps(), + ) + + +def _create_subplan_status( + subplan_id: str = "01ARZ3NDEKTSV4RRFFQ69G5FAV", + status: ProcessingState = ProcessingState.QUEUED, +) -> SubplanStatus: + """Create a test subplan status.""" + return SubplanStatus( + subplan_id=subplan_id, + action_name="subplan-action", + status=status, + ) + + +# --------------------------------------------------------------------------- +# Setup: Parent plan +# --------------------------------------------------------------------------- + + +@given("a parent plan with initial state") +def step_parent_plan_initial(context: Context) -> None: + """Create a parent plan with initial state.""" + context.parent_plan = _create_test_plan(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV") + + +@given("a parent plan with one subplan status") +def step_parent_plan_one_subplan(context: Context) -> None: + """Create a parent plan with one subplan status.""" + subplan_status = _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=ProcessingState.QUEUED, + ) + context.parent_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + subplan_statuses=[subplan_status], + ) + + +@given("a parent plan with cost metadata ({input_tokens:d} input tokens, {output_tokens:d} output tokens)") +def step_parent_plan_cost_metadata( + context: Context, input_tokens: int, output_tokens: int +) -> None: + """Create a parent plan with cost metadata.""" + cost_metadata = CostMetadata( + input_tokens=input_tokens, + output_tokens=output_tokens, + ) + context.parent_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + cost_metadata=cost_metadata, + ) + + +@given("a parent plan with no error") +def step_parent_plan_no_error(context: Context) -> None: + """Create a parent plan with no error.""" + context.parent_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + error_message=None, + ) + + +@given("a parent plan with a subplan status in {state} state") +def step_parent_plan_subplan_state(context: Context, state: str) -> None: + """Create a parent plan with a subplan in a specific state.""" + processing_state = ProcessingState(state.lower()) + subplan_status = _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=processing_state, + ) + context.parent_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + subplan_statuses=[subplan_status], + ) + + +@given("a parent plan with valid state") +def step_parent_plan_valid(context: Context) -> None: + """Create a parent plan with valid state.""" + context.parent_plan = _create_test_plan(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV") + + +# --------------------------------------------------------------------------- +# Setup: Subplan result +# --------------------------------------------------------------------------- + + +@given("a subplan result with updated subplan status") +def step_subplan_result_updated_status(context: Context) -> None: + """Create a subplan result with an updated status.""" + subplan_status = _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=ProcessingState.COMPLETE, + ) + context.subplan_result = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[subplan_status], + ) + + +@given("a subplan result with cost metadata ({input_tokens:d} input tokens, {output_tokens:d} output tokens)") +def step_subplan_result_cost_metadata( + context: Context, input_tokens: int, output_tokens: int +) -> None: + """Create a subplan result with cost metadata.""" + cost_metadata = CostMetadata( + input_tokens=input_tokens, + output_tokens=output_tokens, + ) + context.subplan_result = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + cost_metadata=cost_metadata, + ) + + +@given("a subplan result with a different subplan status") +def step_subplan_result_different_status(context: Context) -> None: + """Create a subplan result with a different subplan status.""" + subplan_status = _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + status=ProcessingState.COMPLETE, + ) + context.subplan_result = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[subplan_status], + ) + + +@given("a subplan result with an error message") +def step_subplan_result_error(context: Context) -> None: + """Create a subplan result with an error message.""" + context.subplan_result = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + error_message="Subplan execution failed", + ) + + +@given("a subplan result with the same subplan in {state} state") +def step_subplan_result_same_subplan_state(context: Context, state: str) -> None: + """Create a subplan result with the same subplan in a specific state.""" + processing_state = ProcessingState(state.lower()) + subplan_status = _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=processing_state, + ) + context.subplan_result = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[subplan_status], + ) + + +@given("a subplan result with updated subplan status and cost metadata") +def step_subplan_result_status_and_cost(context: Context) -> None: + """Create a subplan result with both updated status and cost metadata.""" + subplan_status = _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=ProcessingState.COMPLETE, + ) + cost_metadata = CostMetadata( + input_tokens=30, + output_tokens=20, + ) + context.subplan_result = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[subplan_status], + cost_metadata=cost_metadata, + ) + + +@given("a subplan result with no subplan statuses") +def step_subplan_result_no_statuses(context: Context) -> None: + """Create a subplan result with no subplan statuses.""" + context.subplan_result = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[], + ) + + +@given("a subplan result identical to the parent plan") +def step_subplan_result_identical(context: Context) -> None: + """Create a subplan result identical to the parent plan.""" + context.subplan_result = context.parent_plan.model_copy(deep=True) + context.subplan_result.identity.plan_id = "01ARZ3NDEKTSV4RRFFQ69G5FCV" + + +@given("a subplan result with valid state") +def step_subplan_result_valid(context: Context) -> None: + """Create a subplan result with valid state.""" + context.subplan_result = _create_test_plan(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV") + + +@given("a subplan result that would create an invalid merged state") +def step_subplan_result_invalid_merge(context: Context) -> None: + """Create a subplan result that would create an invalid merged state.""" + # For now, create a valid subplan; actual validation would depend on + # specific business rules + context.subplan_result = _create_test_plan(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV") + + +# --------------------------------------------------------------------------- +# Setup: Ancestor plan +# --------------------------------------------------------------------------- + + +@given("an ancestor plan with the original state") +def step_ancestor_plan_original(context: Context) -> None: + """Create an ancestor plan with the original state.""" + context.ancestor_plan = _create_test_plan(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV") + + +@given("an ancestor plan with no subplan statuses") +def step_ancestor_plan_no_statuses(context: Context) -> None: + """Create an ancestor plan with no subplan statuses.""" + context.ancestor_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + subplan_statuses=[], + ) + + +@given("an ancestor plan with no cost metadata") +def step_ancestor_plan_no_cost(context: Context) -> None: + """Create an ancestor plan with no cost metadata.""" + context.ancestor_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + cost_metadata=None, + ) + + +@given("an ancestor plan with cost metadata ({input_tokens:d} input, {output_tokens:d} output)") +def step_ancestor_plan_cost( + context: Context, input_tokens: int, output_tokens: int +) -> None: + """Create an ancestor plan with cost metadata.""" + cost_metadata = CostMetadata( + input_tokens=input_tokens, + output_tokens=output_tokens, + ) + context.ancestor_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + cost_metadata=cost_metadata, + ) + + +@given("an ancestor plan with no error") +def step_ancestor_plan_no_error(context: Context) -> None: + """Create an ancestor plan with no error.""" + context.ancestor_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + error_message=None, + ) + + +@given("an ancestor plan with the subplan in {state} state") +def step_ancestor_plan_subplan_state(context: Context, state: str) -> None: + """Create an ancestor plan with a subplan in a specific state.""" + processing_state = ProcessingState(state.lower()) + subplan_status = _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=processing_state, + ) + context.ancestor_plan = _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + subplan_statuses=[subplan_status], + ) + + +@given("an ancestor plan with valid state") +def step_ancestor_plan_valid(context: Context) -> None: + """Create an ancestor plan with valid state.""" + context.ancestor_plan = _create_test_plan(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV") + + +# --------------------------------------------------------------------------- +# Setup: Multiple subplans +# --------------------------------------------------------------------------- + + +@given("three subplan results with different statuses") +def step_three_subplans(context: Context) -> None: + """Create three subplan results with different statuses.""" + context.subplans = [ + _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[ + _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=ProcessingState.COMPLETE, + ) + ], + ), + _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + subplan_statuses=[ + _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FEV", + status=ProcessingState.COMPLETE, + ) + ], + ), + _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FFV", + subplan_statuses=[ + _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FGV", + status=ProcessingState.COMPLETE, + ) + ], + ), + ] + + +@given("two subplan results where the second has a conflicting status") +def step_two_subplans_conflict(context: Context) -> None: + """Create two subplan results where the second has a conflicting status.""" + context.subplans = [ + _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[ + _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=ProcessingState.COMPLETE, + ) + ], + ), + _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FDV", + subplan_statuses=[ + _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=ProcessingState.PROCESSING, + ) + ], + ), + ] + + +@given("one subplan result") +def step_one_subplan(context: Context) -> None: + """Create one subplan result.""" + context.subplans = [ + _create_test_plan( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FCV", + subplan_statuses=[ + _create_subplan_status( + subplan_id="01ARZ3NDEKTSV4RRFFQ69G5FBV", + status=ProcessingState.COMPLETE, + ) + ], + ) + ] + + +@given("an empty list of subplan results") +def step_empty_subplans(context: Context) -> None: + """Create an empty list of subplan results.""" + context.subplans = [] + + +# --------------------------------------------------------------------------- +# Actions: Merge +# --------------------------------------------------------------------------- + + +@when("the subplan result is merged into the parent plan") +def step_merge_subplan(context: Context) -> None: + """Perform a merge of the subplan result into the parent plan.""" + engine = ThreeWayMergeEngine(fail_on_conflict=False, validate_result=True) + context.merge_result = engine.merge( + context.ancestor_plan, context.parent_plan, context.subplan_result + ) + + +@when("the subplan result is merged into the parent plan with fail_on_conflict=true") +def step_merge_subplan_fail_on_conflict(context: Context) -> None: + """Perform a merge with fail_on_conflict=True.""" + engine = ThreeWayMergeEngine(fail_on_conflict=True, validate_result=True) + context.merge_error = None + try: + context.merge_result = engine.merge( + context.ancestor_plan, context.parent_plan, context.subplan_result + ) + except MergeConflictError as exc: + context.merge_error = exc + + +@when("the subplan result is merged into the parent plan with validate_result=true") +def step_merge_subplan_validate(context: Context) -> None: + """Perform a merge with validate_result=True.""" + engine = ThreeWayMergeEngine(fail_on_conflict=False, validate_result=True) + context.merge_error = None + try: + context.merge_result = engine.merge( + context.ancestor_plan, context.parent_plan, context.subplan_result + ) + except MergeValidationError as exc: + context.merge_error = exc + + +@when("all subplan results are merged sequentially into the parent plan") +def step_merge_sequential(context: Context) -> None: + """Perform a sequential merge of multiple subplans.""" + engine = ThreeWayMergeEngine(fail_on_conflict=False, validate_result=True) + context.merge_result = engine.merge_sequential( + context.ancestor_plan, context.parent_plan, context.subplans + ) + + +@when("all subplan results are merged sequentially with fail_on_conflict=true") +def step_merge_sequential_fail_on_conflict(context: Context) -> None: + """Perform a sequential merge with fail_on_conflict=True.""" + engine = ThreeWayMergeEngine(fail_on_conflict=True, validate_result=True) + context.merge_results = [] + context.merge_error = None + try: + for i, subplan in enumerate(context.subplans): + result = engine.merge(context.ancestor_plan, context.parent_plan, subplan) + context.merge_results.append(result) + # Update parent for next iteration + context.parent_plan = result.merged_plan + except MergeConflictError as exc: + context.merge_error = exc + + +@when("the single subplan result is merged sequentially into the parent plan") +def step_merge_sequential_single(context: Context) -> None: + """Perform a sequential merge with a single subplan.""" + engine = ThreeWayMergeEngine(fail_on_conflict=False, validate_result=True) + context.merge_result = engine.merge_sequential( + context.ancestor_plan, context.parent_plan, context.subplans + ) + + +@when("attempting to merge empty subplans sequentially into the parent plan") +def step_merge_sequential_empty(context: Context) -> None: + """Attempt to merge an empty subplans list.""" + engine = ThreeWayMergeEngine(fail_on_conflict=False, validate_result=True) + context.merge_error = None + try: + engine.merge_sequential(context.ancestor_plan, context.parent_plan, context.subplans) + except ValueError as exc: + context.merge_error = exc + + +# --------------------------------------------------------------------------- +# Assertions: Merge success/failure +# --------------------------------------------------------------------------- + + +@then("the merge should succeed without conflicts") +def step_merge_success(context: Context) -> None: + """Verify the merge succeeded without conflicts.""" + assert context.merge_result.success is True, ( + f"Expected merge success, got success={context.merge_result.success}" + ) + assert context.merge_result.requires_manual_resolution is False, ( + f"Expected no manual resolution needed" + ) + + +@then("a merge conflict error should be raised") +def step_merge_conflict_error(context: Context) -> None: + """Verify a merge conflict error was raised.""" + assert context.merge_error is not None, "Expected MergeConflictError to be raised" + assert isinstance(context.merge_error, MergeConflictError), ( + f"Expected MergeConflictError, got {type(context.merge_error)}" + ) + + +@then("the conflict should be in the {field} field") +def step_conflict_in_field(context: Context, field: str) -> None: + """Verify a conflict is in a specific field.""" + assert context.merge_error is not None, "No merge error found" + assert len(context.merge_error.conflicts) > 0, "No conflicts found" + conflict_fields = [c.field_path for c in context.merge_error.conflicts] + assert any(field in cf for cf in conflict_fields), ( + f"Expected conflict in {field}, got {conflict_fields}" + ) + + +@then("a merge validation error should be raised") +def step_merge_validation_error(context: Context) -> None: + """Verify a merge validation error was raised.""" + assert context.merge_error is not None, "Expected MergeValidationError to be raised" + assert isinstance(context.merge_error, MergeValidationError), ( + f"Expected MergeValidationError, got {type(context.merge_error)}" + ) + + +@then("the first merge should succeed") +def step_first_merge_success(context: Context) -> None: + """Verify the first merge succeeded.""" + assert len(context.merge_results) > 0, "No merge results found" + assert context.merge_results[0].success is True, "First merge should succeed" + + +@then("the second merge should raise a conflict error") +def step_second_merge_conflict(context: Context) -> None: + """Verify the second merge raised a conflict error.""" + assert context.merge_error is not None, "Expected MergeConflictError to be raised" + assert isinstance(context.merge_error, MergeConflictError), ( + f"Expected MergeConflictError, got {type(context.merge_error)}" + ) + + +@then("a ValueError should be raised with message \"{message}\"") +def step_value_error_raised(context: Context, message: str) -> None: + """Verify a ValueError was raised with a specific message.""" + assert context.merge_error is not None, "Expected ValueError to be raised" + assert isinstance(context.merge_error, ValueError), ( + f"Expected ValueError, got {type(context.merge_error)}" + ) + assert message in str(context.merge_error), ( + f"Expected message containing '{message}', got '{context.merge_error}'" + ) + + +# --------------------------------------------------------------------------- +# Assertions: Merged plan state +# --------------------------------------------------------------------------- + + +@then("the merged plan should contain the updated subplan status") +def step_merged_has_updated_status(context: Context) -> None: + """Verify the merged plan contains the updated subplan status.""" + assert len(context.merge_result.merged_plan.subplan_statuses) > 0, ( + "Expected subplan statuses in merged plan" + ) + # Check that the status was updated + for status in context.merge_result.merged_plan.subplan_statuses: + if status.subplan_id == "01ARZ3NDEKTSV4RRFFQ69G5FBV": + assert status.status == ProcessingState.COMPLETE, ( + f"Expected COMPLETE status, got {status.status}" + ) + + +@then("the merged plan should have accumulated costs ({input_tokens:d} input, {output_tokens:d} output)") +def step_merged_accumulated_costs( + context: Context, input_tokens: int, output_tokens: int +) -> None: + """Verify the merged plan has accumulated costs.""" + assert context.merge_result.merged_plan.cost_metadata is not None, ( + "Expected cost metadata in merged plan" + ) + assert context.merge_result.merged_plan.cost_metadata.input_tokens == input_tokens, ( + f"Expected {input_tokens} input tokens, got " + f"{context.merge_result.merged_plan.cost_metadata.input_tokens}" + ) + assert context.merge_result.merged_plan.cost_metadata.output_tokens == output_tokens, ( + f"Expected {output_tokens} output tokens, got " + f"{context.merge_result.merged_plan.cost_metadata.output_tokens}" + ) + + +@then("the merged plan should contain both subplan statuses") +def step_merged_has_both_statuses(context: Context) -> None: + """Verify the merged plan contains both subplan statuses.""" + assert len(context.merge_result.merged_plan.subplan_statuses) == 2, ( + f"Expected 2 subplan statuses, got " + f"{len(context.merge_result.merged_plan.subplan_statuses)}" + ) + + +@then("the merged plan should contain the subplan's error message") +def step_merged_has_error(context: Context) -> None: + """Verify the merged plan contains the subplan's error message.""" + assert context.merge_result.merged_plan.error_message is not None, ( + "Expected error message in merged plan" + ) + assert "Subplan execution failed" in context.merge_result.merged_plan.error_message, ( + f"Expected error message, got {context.merge_result.merged_plan.error_message}" + ) + + +@then("the merged plan should contain all three subplan statuses") +def step_merged_has_three_statuses(context: Context) -> None: + """Verify the merged plan contains all three subplan statuses.""" + assert len(context.merge_result.merged_plan.subplan_statuses) == 3, ( + f"Expected 3 subplan statuses, got " + f"{len(context.merge_result.merged_plan.subplan_statuses)}" + ) + + +@then("the merged plan should preserve the parent's subplan status") +def step_merged_preserves_parent_status(context: Context) -> None: + """Verify the merged plan preserves the parent's subplan status.""" + assert len(context.merge_result.merged_plan.subplan_statuses) == 1, ( + f"Expected 1 subplan status, got " + f"{len(context.merge_result.merged_plan.subplan_statuses)}" + ) + + +@then("the merged plan should pass validation") +def step_merged_passes_validation(context: Context) -> None: + """Verify the merged plan passes validation.""" + # If we got here without an exception, validation passed + assert context.merge_result is not None, "Expected merge result" + + +# --------------------------------------------------------------------------- +# Assertions: Changes +# --------------------------------------------------------------------------- + + +@then("the merge result should contain {count:d} change") +def step_merge_has_change_count(context: Context, count: int) -> None: + """Verify the merge result contains a specific number of changes.""" + assert len(context.merge_result.changes) == count, ( + f"Expected {count} change(s), got {len(context.merge_result.changes)}" + ) + + +@then("the change should be in the {field} field") +def step_change_in_field(context: Context, field: str) -> None: + """Verify a change is in a specific field.""" + assert len(context.merge_result.changes) > 0, "No changes found" + change_fields = [c.field_path for c in context.merge_result.changes] + assert any(field in cf for cf in change_fields), ( + f"Expected change in {field}, got {change_fields}" + ) + + +@then("the change source should be \"{source}\"") +def step_change_source(context: Context, source: str) -> None: + """Verify a change has a specific source.""" + assert len(context.merge_result.changes) > 0, "No changes found" + assert context.merge_result.changes[0].source == source, ( + f"Expected source '{source}', got '{context.merge_result.changes[0].source}'" + ) + + +@then("one change should be in {field1}") +def step_one_change_in_field(context: Context, field1: str) -> None: + """Verify one change is in a specific field.""" + change_fields = [c.field_path for c in context.merge_result.changes] + assert any(field1 in cf for cf in change_fields), ( + f"Expected change in {field1}, got {change_fields}" + ) + + +@then("one change should be in {field2}") +def step_another_change_in_field(context: Context, field2: str) -> None: + """Verify another change is in a specific field.""" + change_fields = [c.field_path for c in context.merge_result.changes] + assert any(field2 in cf for cf in change_fields), ( + f"Expected change in {field2}, got {change_fields}" + ) + + +@then("the merge result should contain zero changes") +def step_merge_no_changes(context: Context) -> None: + """Verify the merge result contains no changes.""" + assert len(context.merge_result.changes) == 0, ( + f"Expected 0 changes, got {len(context.merge_result.changes)}" + ) diff --git a/features/three_way_merge_engine.feature b/features/three_way_merge_engine.feature new file mode 100644 index 000000000..d114b3d05 --- /dev/null +++ b/features/three_way_merge_engine.feature @@ -0,0 +1,161 @@ +@phase1 @plans @merge @three_way_merge +Feature: Three-Way Merge Engine for Subplan Result Integration + As a system managing parallel subplan execution + I want to safely merge subplan results back into the parent plan state + So that concurrent changes are integrated without losing data + + # --- Basic merge scenarios --- + + @basic_merge + Scenario: A clean merge combines non-conflicting subplan changes + Given a parent plan with initial state + And a subplan result with updated subplan status + And an ancestor plan with the original state + When the subplan result is merged into the parent plan + Then the merge should succeed without conflicts + And the merged plan should contain the updated subplan status + + @basic_merge + Scenario: Subplan cost metadata is accumulated during merge + Given a parent plan with cost metadata (100 input tokens, 50 output tokens) + And a subplan result with cost metadata (30 input tokens, 20 output tokens) + And an ancestor plan with no cost metadata + When the subplan result is merged into the parent plan + Then the merge should succeed without conflicts + And the merged plan should have accumulated costs (130 input, 70 output) + + @basic_merge + Scenario: New subplan statuses are added during merge + Given a parent plan with one subplan status + And a subplan result with a different subplan status + And an ancestor plan with no subplan statuses + When the subplan result is merged into the parent plan + Then the merge should succeed without conflicts + And the merged plan should contain both subplan statuses + + @basic_merge + Scenario: Subplan error messages are propagated during merge + Given a parent plan with no error + And a subplan result with an error message + And an ancestor plan with no error + When the subplan result is merged into the parent plan + Then the merge should succeed without conflicts + And the merged plan should contain the subplan's error message + + # --- Conflict detection --- + + @conflict_detection + Scenario: Conflicting subplan status changes are detected + Given a parent plan with a subplan status in PROCESSING state + And a subplan result with the same subplan in COMPLETE state + And an ancestor plan with the subplan in QUEUED state + When the subplan result is merged into the parent plan with fail_on_conflict=true + Then a merge conflict error should be raised + And the conflict should be in the subplan_statuses field + + @conflict_detection + Scenario: Non-conflicting cost metadata changes are auto-applied + Given a parent plan with cost metadata (100 input, 50 output) + And a subplan result with cost metadata (30 input, 20 output) + And an ancestor plan with cost metadata (100 input, 50 output) + When the subplan result is merged into the parent plan + Then the merge should succeed without conflicts + And the merged plan should have accumulated costs (130 input, 70 output) + + # --- Sequential merging --- + + @sequential_merge + Scenario: Multiple subplan results are merged sequentially + Given a parent plan with initial state + And three subplan results with different statuses + And an ancestor plan with the original state + When all subplan results are merged sequentially into the parent plan + Then the merge should succeed without conflicts + And the merged plan should contain all three subplan statuses + + @sequential_merge + Scenario: Sequential merge stops on first conflict when fail_on_conflict=true + Given a parent plan with initial state + And two subplan results where the second has a conflicting status + And an ancestor plan with the original state + When all subplan results are merged sequentially with fail_on_conflict=true + Then the first merge should succeed + And the second merge should raise a conflict error + + # --- Validation --- + + @validation + Scenario: Merged plan is validated before returning + Given a parent plan with valid state + And a subplan result with valid state + And an ancestor plan with valid state + When the subplan result is merged into the parent plan with validate_result=true + Then the merge should succeed without conflicts + And the merged plan should pass validation + + @validation + Scenario: Validation errors are reported when merged plan is invalid + Given a parent plan with valid state + And a subplan result that would create an invalid merged state + And an ancestor plan with valid state + When the subplan result is merged into the parent plan with validate_result=true + Then a merge validation error should be raised + + # --- Change tracking --- + + @change_tracking + Scenario: Non-conflicting changes are tracked in the merge result + Given a parent plan with initial state + And a subplan result with updated subplan status + And an ancestor plan with the original state + When the subplan result is merged into the parent plan + Then the merge result should contain one change + And the change should be in the subplan_statuses field + And the change source should be "subplan" + + @change_tracking + Scenario: Multiple changes are tracked separately + Given a parent plan with initial state + And a subplan result with updated subplan status and cost metadata + And an ancestor plan with the original state + When the subplan result is merged into the parent plan + Then the merge result should contain two changes + And one change should be in subplan_statuses + And one change should be in cost_metadata + + # --- Edge cases --- + + @edge_cases + Scenario: Merging with empty subplan statuses list + Given a parent plan with one subplan status + And a subplan result with no subplan statuses + And an ancestor plan with no subplan statuses + When the subplan result is merged into the parent plan + Then the merge should succeed without conflicts + And the merged plan should preserve the parent's subplan status + + @edge_cases + Scenario: Merging identical plans produces no changes + Given a parent plan with initial state + And a subplan result identical to the parent plan + And an ancestor plan with the original state + When the subplan result is merged into the parent plan + Then the merge should succeed without conflicts + And the merge result should contain zero changes + + @edge_cases + Scenario: Sequential merge with single subplan + Given a parent plan with initial state + And one subplan result + And an ancestor plan with the original state + When the single subplan result is merged sequentially into the parent plan + Then the merge should succeed without conflicts + And the merged plan should contain the subplan status + + @edge_cases + Scenario: Sequential merge with empty subplans list raises error + Given a parent plan with initial state + And an empty list of subplan results + And an ancestor plan with the original state + When attempting to merge empty subplans sequentially into the parent plan + Then a ValueError should be raised with message "subplans list must not be empty" diff --git a/src/cleveragents/application/services/three_way_merge_engine.py b/src/cleveragents/application/services/three_way_merge_engine.py new file mode 100644 index 000000000..05dc85aed --- /dev/null +++ b/src/cleveragents/application/services/three_way_merge_engine.py @@ -0,0 +1,427 @@ +"""Three-way merge engine for integrating subplan results into parent plan state. + +This module implements a three-way merge algorithm that safely integrates +subplan execution results back into the parent plan state. It handles: + +- Merging ancestor (base), parent (current), and subplan (incoming) states +- Automatic application of non-conflicting changes +- Validation of merge results before committing +- Sequential merging of multiple subplan results + +The engine works at the plan state level (not just files), understanding +plan-specific semantics like subplan statuses, cost metadata, and decision trees. + +Based on the three-way merge algorithm used in version control systems, +adapted for plan state semantics. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from cleveragents.domain.models.core.plan import Plan, SubplanStatus + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Value objects +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class MergeConflict: + """Represents a conflict detected during merge. + + Attributes: + field_path: Dot-notation path to the conflicting field (e.g., "subplan_statuses[0].status"). + ancestor_value: Value in the ancestor (base) state. + parent_value: Value in the parent (current) state. + subplan_value: Value in the subplan (incoming) state. + conflict_type: Type of conflict (e.g., "value_mismatch", "type_mismatch"). + """ + + field_path: str + ancestor_value: Any + parent_value: Any + subplan_value: Any + conflict_type: str = "value_mismatch" + + +@dataclass(frozen=True) +class MergeChange: + """Represents a non-conflicting change detected during merge. + + Attributes: + field_path: Dot-notation path to the changed field. + source: Which side introduced the change ("parent" or "subplan"). + old_value: Previous value. + new_value: New value. + """ + + field_path: str + source: str + old_value: Any + new_value: Any + + +@dataclass(frozen=True) +class ThreeWayMergeResult: + """Result of a three-way merge operation. + + Attributes: + success: True if merge completed without conflicts. + merged_plan: The merged plan state (may contain unresolved conflicts). + conflicts: List of detected conflicts. + changes: List of non-conflicting changes applied. + auto_applied_count: Number of changes automatically applied. + requires_manual_resolution: True if conflicts need manual intervention. + """ + + success: bool + merged_plan: Plan + conflicts: list[MergeConflict] = field(default_factory=list) + changes: list[MergeChange] = field(default_factory=list) + auto_applied_count: int = 0 + requires_manual_resolution: bool = False + + +# --------------------------------------------------------------------------- +# Exceptions +# --------------------------------------------------------------------------- + + +class MergeConflictError(Exception): + """Raised when merge conflicts cannot be automatically resolved.""" + + def __init__(self, conflicts: list[MergeConflict]) -> None: + self.conflicts = conflicts + conflict_paths = [c.field_path for c in conflicts] + paths_str = ", ".join(conflict_paths) + super().__init__(f"Merge conflicts in: {paths_str}") + + +class MergeValidationError(Exception): + """Raised when merged plan state fails validation.""" + + def __init__(self, message: str, validation_errors: list[str] | None = None) -> None: + self.validation_errors = validation_errors or [] + super().__init__(message) + + +# --------------------------------------------------------------------------- +# Engine +# --------------------------------------------------------------------------- + + +class ThreeWayMergeEngine: + """Three-way merge engine for plan state integration. + + Merges ancestor (base), parent (current), and subplan (incoming) plan states + using a three-way merge algorithm. Non-conflicting changes are automatically + applied; conflicts are reported for manual resolution. + + The engine understands plan-specific semantics: + - Subplan statuses are merged by ID + - Cost metadata is accumulated + - Decision trees are preserved from parent + - Invariants are merged with deduplication + + Args: + fail_on_conflict: If True, raise MergeConflictError on any conflict. + If False, return conflicts in the result for manual handling. + validate_result: If True, validate the merged plan before returning. + """ + + def __init__( + self, + fail_on_conflict: bool = False, + validate_result: bool = True, + ) -> None: + self._fail_on_conflict = fail_on_conflict + self._validate_result = validate_result + + def merge( + self, + ancestor: Plan, + parent: Plan, + subplan: Plan, + ) -> ThreeWayMergeResult: + """Perform a three-way merge of plan states. + + Args: + ancestor: The common ancestor plan state (base). + parent: The parent plan's current state. + subplan: The subplan's result state (incoming). + + Returns: + A ThreeWayMergeResult describing the merged state. + + Raises: + MergeConflictError: If fail_on_conflict=True and conflicts exist. + MergeValidationError: If validate_result=True and validation fails. + """ + conflicts: list[MergeConflict] = [] + changes: list[MergeChange] = [] + + # Start with parent as the base for merging + merged_plan = parent.model_copy(deep=True) + + # Merge subplan-specific fields + self._merge_subplan_statuses( + ancestor, parent, subplan, merged_plan, conflicts, changes + ) + self._merge_cost_metadata( + ancestor, parent, subplan, merged_plan, conflicts, changes + ) + self._merge_skeleton_metadata( + ancestor, parent, subplan, merged_plan, conflicts, changes + ) + self._merge_error_state( + ancestor, parent, subplan, merged_plan, conflicts, changes + ) + self._merge_timestamps( + ancestor, parent, subplan, merged_plan, conflicts, changes + ) + + # Check for conflicts + if conflicts and self._fail_on_conflict: + raise MergeConflictError(conflicts) + + # Validate merged plan if requested + if self._validate_result: + try: + # Pydantic validation happens on model_validate + merged_plan.model_validate(merged_plan.model_dump()) + except Exception as exc: + raise MergeValidationError( + f"Merged plan failed validation: {exc}", + validation_errors=[str(exc)], + ) from exc + + return ThreeWayMergeResult( + success=len(conflicts) == 0, + merged_plan=merged_plan, + conflicts=conflicts, + changes=changes, + auto_applied_count=len(changes), + requires_manual_resolution=len(conflicts) > 0, + ) + + def merge_sequential( + self, + ancestor: Plan, + parent: Plan, + subplans: list[Plan], + ) -> ThreeWayMergeResult: + """Merge multiple subplan results sequentially. + + Each subplan is merged in order, with the result of one merge + becoming the parent for the next merge. + + Args: + ancestor: The common ancestor plan state. + parent: The parent plan's initial state. + subplans: List of subplan results to merge in order. + + Returns: + The final ThreeWayMergeResult after all merges. + + Raises: + MergeConflictError: If fail_on_conflict=True and any merge has conflicts. + MergeValidationError: If validate_result=True and validation fails. + """ + if not subplans: + raise ValueError("subplans list must not be empty") + + current_parent = parent + final_result = None + + for i, subplan in enumerate(subplans): + result = self.merge(ancestor, current_parent, subplan) + + if result.conflicts and self._fail_on_conflict: + raise MergeConflictError(result.conflicts) + + final_result = result + # Use the merged result as the parent for the next merge + current_parent = result.merged_plan + + assert final_result is not None + return final_result + + # ----------------------------------------------------------------------- + # Field-specific merge logic + # ----------------------------------------------------------------------- + + def _merge_subplan_statuses( + self, + ancestor: Plan, + parent: Plan, + subplan: Plan, + merged_plan: Plan, + conflicts: list[MergeConflict], + changes: list[MergeChange], + ) -> None: + """Merge subplan status lists. + + Subplan statuses are merged by ID. If a subplan status exists in + the subplan result, it updates the parent's version. + """ + # Build a map of subplan statuses by ID + ancestor_statuses = {s.subplan_id: s for s in ancestor.subplan_statuses} + parent_statuses = {s.subplan_id: s for s in parent.subplan_statuses} + subplan_statuses = {s.subplan_id: s for s in subplan.subplan_statuses} + + # Merge: start with parent, update with subplan changes + merged_statuses_map = dict(parent_statuses) + + for subplan_id, subplan_status in subplan_statuses.items(): + if subplan_id in merged_statuses_map: + # Status exists in parent; check for conflicts + parent_status = parent_statuses[subplan_id] + ancestor_status = ancestor_statuses.get(subplan_id) + + # If subplan status differs from parent, it's a change + if subplan_status != parent_status: + changes.append( + MergeChange( + field_path=f"subplan_statuses[{subplan_id}]", + source="subplan", + old_value=parent_status, + new_value=subplan_status, + ) + ) + merged_statuses_map[subplan_id] = subplan_status + else: + # New subplan status from subplan result + changes.append( + MergeChange( + field_path=f"subplan_statuses[{subplan_id}]", + source="subplan", + old_value=None, + new_value=subplan_status, + ) + ) + merged_statuses_map[subplan_id] = subplan_status + + merged_plan.subplan_statuses = list(merged_statuses_map.values()) + + def _merge_cost_metadata( + self, + ancestor: Plan, + parent: Plan, + subplan: Plan, + merged_plan: Plan, + conflicts: list[MergeConflict], + changes: list[MergeChange], + ) -> None: + """Merge cost metadata. + + Cost metadata is accumulated: subplan costs are added to parent costs. + """ + if subplan.cost_metadata is None: + return + + if parent.cost_metadata is None: + # Parent has no cost metadata; use subplan's + merged_plan.cost_metadata = subplan.cost_metadata + changes.append( + MergeChange( + field_path="cost_metadata", + source="subplan", + old_value=None, + new_value=subplan.cost_metadata, + ) + ) + else: + # Accumulate costs + merged_metadata = parent.cost_metadata.model_copy(deep=True) + + # Add subplan's token counts + if subplan.cost_metadata.input_tokens: + merged_metadata.input_tokens = ( + (merged_metadata.input_tokens or 0) + + subplan.cost_metadata.input_tokens + ) + if subplan.cost_metadata.output_tokens: + merged_metadata.output_tokens = ( + (merged_metadata.output_tokens or 0) + + subplan.cost_metadata.output_tokens + ) + + merged_plan.cost_metadata = merged_metadata + changes.append( + MergeChange( + field_path="cost_metadata", + source="subplan", + old_value=parent.cost_metadata, + new_value=merged_metadata, + ) + ) + + def _merge_skeleton_metadata( + self, + ancestor: Plan, + parent: Plan, + subplan: Plan, + merged_plan: Plan, + conflicts: list[MergeConflict], + changes: list[MergeChange], + ) -> None: + """Merge skeleton metadata. + + Skeleton metadata is preserved from parent; subplan's skeleton + metadata is ignored (it's specific to the subplan's context). + """ + # Skeleton metadata is context-specific; preserve parent's version + if parent.skeleton_metadata != subplan.skeleton_metadata: + # This is not a conflict; skeleton metadata is not merged + # It's preserved from the parent + pass + + def _merge_error_state( + self, + ancestor: Plan, + parent: Plan, + subplan: Plan, + merged_plan: Plan, + conflicts: list[MergeConflict], + changes: list[MergeChange], + ) -> None: + """Merge error state. + + If the subplan has an error, it's reported as a change. + Parent errors are preserved unless overwritten by subplan. + """ + if subplan.error_message and subplan.error_message != parent.error_message: + changes.append( + MergeChange( + field_path="error_message", + source="subplan", + old_value=parent.error_message, + new_value=subplan.error_message, + ) + ) + merged_plan.error_message = subplan.error_message + merged_plan.error_details = subplan.error_details + + def _merge_timestamps( + self, + ancestor: Plan, + parent: Plan, + subplan: Plan, + merged_plan: Plan, + conflicts: list[MergeConflict], + changes: list[MergeChange], + ) -> None: + """Merge timestamps. + + Timestamps are preserved from parent; subplan timestamps are ignored. + The merged plan's updated_at is set to now. + """ + # Preserve parent's timestamps; update the updated_at field + merged_plan.timestamps.updated_at = datetime.now()