Implements a three-way merge engine that safely integrates subplan execution
results back into the parent plan state. The engine 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, understanding plan-specific semantics:
- Subplan statuses are merged by ID
- Cost metadata is accumulated
- Decision trees are preserved from parent
- Invariants are merged with deduplication

Includes comprehensive BDD tests covering:
- Basic merge scenarios
- Conflict detection
- Sequential merging
- Validation
- Change tracking
- Edge cases

Closes #9557

@@ -0,0 +247,4 @@
+@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, (

@@ -0,0 +1,866 @@
+"""Step definitions for ThreeWayMergeEngine Behave scenarios."""

@@ -0,0 +238,4 @@
+    context: Context, subplan_id: str, tokens: int, input_tokens: int, output_tokens: int, cost: float,
+) -> None:
+    """Single subplan contributes specific cost."""
+    if subplan_id not in context._subplan_costs_map:

@@ -0,0 +821,4 @@
+# ---------------------------------------------------------------------------
+
+
+@given("no subplan {subplans} recorded")

@@ -0,0 +1,516 @@
+"""Three-way merge engine for integrating subplan results into parent plan state.

@@ -0,0 +430,4 @@
+                    merged.provider_costs.get(provider, 0.0) + cost
+                )
+
+        return merged

@@ -0,0 +472,4 @@
+    @staticmethod
+    def _state_priority(state: ProcessingState) -> int:
+        """Numeric priority for processing states (higher = more terminal)."""
+        priorites = {

@@ -0,0 +508,4 @@
+            return base_val  # no divergence
+
+        candidates = [v for v in (base_val, current_val, incoming_val) if v is not None]
+        return max(candidates) if candidates else None

@@ -0,0 +1,866 @@
+"""Step definitions for ThreeWayMergeEngine Behave scenarios."""

@@ -0,0 +1,516 @@
+"""Three-way merge engine for integrating subplan results into parent plan state.

@@ -0,0 +54,4 @@
+    And current cost metadata with 5 tokens and $0.01 cost
+    And no subplan costs recorded
+    When I merge the three-way plan states with conflicts allowed
+    Then the merged status for "_S1" should be ERRORED (highest priority)

@@ -0,0 +84,4 @@
+
+  @error_propagation
+  Scenario: An ERRORED subplan propagates its error message upward
+    Given a base subplan status with QUEUED state for "_S1"

@@ -0,0 +87,4 @@
+    Given a base subplan status with QUEUED state for "_S1"
+    And a current subplan status with COMPLETE state for "_S1"
+    And a subplan result that sets _S1 to ERRORED
+    And the subplan _S1 has error "Model returned invalid JSON response"

@@ -0,0 +125,4 @@
+
+  @cost_accumulation
+  Scenario: Costs are correctly accumulated across base, current, and subplans
+    Given a base cost metadata with 100 tokens, 50 input, 50 output, $0.10 total cost

@@ -0,0 +163,4 @@
+def skeleton_preserved_ok() -> None:
+    """Verify parent skeleton metadata preserved."""
+    engine = ThreeWayMergeEngine()
+    subplan_statuses = [_mk_status(_S1, ProcessingState.COMPLETE)]

@@ -0,0 +331,4 @@
+                    merged.budget_remaining = min(
+                        merged.budget_remaining, sc.budget_remaining
+                    )
+                    continue

@@ -0,0 +276,4 @@
+            if candidate.status != base.status:
+                changed = True
+
+        return SubplanStatusMergeResult(