feat(plans): implement ThreeWayMergeEngine for subplan result integration (#9608)

Implements a three-way merge engine that safely integrates subplan execution
results back into the parent plan state. The engine handles merging of ancestor
(base), parent (current), and subplan (incoming) states with automatic
application of non-conflicting changes and validation before committing.

- Added ThreeWayMergeEngine class in src/cleveragents/application/services/
- Implemented merge logic for subplan statuses by ID, cost metadata accumulation,
  skeleton metadata preservation, error state propagation, and timestamp handling
- Updated __init__.py with lazy imports and TYPE_CHECKING declarations for new types
- Added comprehensive BDD tests in features/three_way_merge_engine.feature
- Added step definitions in features/steps/three_way_merge_engine_steps.py
- Updated CHANGELOG.md and CONTRIBUTORS.md per PR compliance checklist

Closes #9557

ISSUES CLOSED: #9557

CHANGELOG.md

@@ -175,6 +175,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **ThreeWayMergeEngine for subplan result integration** (#9608): Implemented a
+  three-way merge engine that safely integrates subplan execution results back
+  into parent plan state. The engine handles merging of ancestor (base), parent
+  (current), and subplan (incoming) states with automatic application of
+  non-conflicting changes and validation before committing. Merges
+  `SubplanStatus` records by ID, accumulates `CostMetadata` across all subplans,
+  preserves `SkeletonMetadata` from the parent plan, propagates error states
+  upward when any subplan fails, and advances timestamps to handle most-recent
+  events across the merged output. Includes comprehensive BDD tests covering
+  basic merge scenarios, conflict detection, sequential merging, and edge cases.
+
 - `agents actor context clear` command to reset actor message history and
   state while preserving the underlying context directory via `ContextManager`
   (#6370).

CONTRIBUTORS.md

@@ -33,6 +33,7 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed comprehensive milestone documentation for v3.6.0 (Advanced Concepts & Deferred Features) and v3.7.0 (TUI Implementation) (PR #9903): split into sub-documents covering context strategies, LLM backends, resource types, A2A rename, container tool execution, scope chain resolution, cost/safety budgets, E2E workflow tests, code review examples, plugin architecture, TUI layout, persona system, reference/command input, session management, configuration, and TuiMaterializer integration.
 * HAL 9000 has contributed the LLMTraceRepository data-integrity fix (PR #8185 / issue #7505): replaced the unconditional `session.commit()` in `LLMTraceRepository.save()` with a dual-path implementation that respects the UnitOfWork pattern — flushing only when an external session is provided, and flushing + committing + closing when operating standalone. This eliminates premature transaction commits, loss of rollback capability, and a docstring/implementation mismatch.
 * HAL 9000 has contributed the ACMS Index Data Model and File Traversal Engine (PR #9664 / issue #9579): foundational data structures for indexed context entries with hot/warm/cold/archive storage tier classification, tag system, and a timeout-safe chunked file traversal engine for large projects with 10,000+ files.
+* HAL 9000 has contributed the ThreeWayMergeEngine for subplan result integration (PR #9608 / issue #9557): implemented a three-way merge engine that safely integrates subplan execution results back into parent plan state, handling merging of SubplanStatus records by ID, CostMetadata accumulation across all subplans, SkeletonMetadata preservation from the parent plan, error propagation upward when any subplan fails, and timestamp advancement for most-recent events. Includes comprehensive BDD test coverage.
 * HAL 9000 has contributed the error-suppression removal fix (PR #9247 / issue #9060): removed both `try...except Exception:` blocks in `register_registry_agents()` that silently suppressed errors from `actor_registry.list_actors()` and the route bridge refresh, enabling exceptions to propagate per CONTRIBUTING.md fail-fast policy. Added three Behave scenarios verifying RuntimeError, AttributeError, and TypeError propagation.
 * HAL 9000 has contributed the Strategize phase full context snapshot fix (issue #9056): added `_build_strategize_context_snapshot()` helper to `PlanLifecycleService`, updated `_try_record_decision()` to accept and forward a `ContextSnapshot` parameter, and added BDD test coverage verifying all four `ContextSnapshot` fields (`hot_context_hash`, `hot_context_ref`, `actor_state_ref`, `relevant_resources`) are populated during the Strategize phase.
 * HAL 9000 has contributed the ACMS context path matching fix (PR #10975 / issue #10972): corrects `_path_matches()` and `_matches_pattern()` to properly match absolute fragment paths against relative glob patterns by auto-prefixing with `**/` before calling `PurePath.full_match()`, preventing silent inefficacy of include/exclude filters for absolute paths in fragment metadata.

features/steps/three_way_merge_engine_steps.py

@@ -0,0 +1,866 @@
+"""Step definitions for ThreeWayMergeEngine Behave scenarios."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.three_way_merge_engine import (
+    ThreeWayMergeEngine,
+    ThreeWayMergeError,
+    ThreeWayMergeResult,
+)
+from cleveragents.domain.models.core.cost_metadata import CostMetadata
+from cleveragents.domain.models.core.plan import ProcessingState, SubplanStatus
+from cleveragents.domain.models.core.skeleton_metadata import SkeletonMetadata
+
+# Fixed ULID-like identifiers for deterministic testing
+_S1 = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
+_S2 = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
+_S3 = "01HGZ6FE0AQDYTR4BXVQZ6EC00"
+
+
+def _make_status(
+    subplan_id: str,
+    state: ProcessingState = ProcessingState.QUEUED,
+    files_changed: int = 0,
+    error: str | None = None,
+    started_at: datetime | None = None,
+    completed_at: datetime | None = None,
+) -> SubplanStatus:
+    """Convenience factory for test subplan statuses."""
+    return SubplanStatus(
+        subplan_id=subplan_id,
+        action_name="local/test-action",
+        status=state,
+        files_changed=files_changed,
+        error=error,
+        started_at=started_at,
+        completed_at=completed_at,
+    )
+
+
+def _make_cost(
+    input_tokens: int = 0,
+    output_tokens: int = 0,
+    budget_remaining: float | None = None,
+) -> CostMetadata:
+    """Convenience factory for test cost metadata."""
+    return CostMetadata(
+        total_tokens=input_tokens + output_tokens,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        total_cost=(input_tokens + output_tokens) * 0.001,
+        budget_remaining=budget_remaining,
+    )
+
+
+def _make_skeleton(ratio: float = 0.6) -> SkeletonMetadata:
+    """Convenience factory for test skeleton metadata."""
+    return SkeletonMetadata(
+        ratio=ratio,
+        original_tokens=1000,
+        compressed_tokens=int(1000 * ratio),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Given steps - Status setup
+# ---------------------------------------------------------------------------
+
+
+@given("a base subplan status with {state} state for subplan \"{subplan_id}\"")
+def step_base_status(context: Context, state: str, subplan_id: str) -> None:
+    """Create a base subplan status."""
+    context._sub1_started = datetime.now(UTC) - timedelta(hours=2)
+    context.base_statuses = [_make_status(subplan_id, ProcessingState(state))]
+    context.current_statuses = list(context.base_statuses)
+
+
+@given("a current subplan status with {state} state for subplan \"{subplan_id}\"")
+def step_current_status(context: Context, state: str, subplan_id: str) -> None:
+    """Create a current subplan status (replacing base)."""
+    context._current_started = datetime.now(UTC) - timedelta(hours=1)
+    cur = [_make_status(subplan_id, ProcessingState(state), started_at=context._current_started)]
+    if not hasattr(context, "base_statuses"):
+        context.base_statuses = list(cur)
+    context.current_statuses = cur
+
+
+@given("a current subplan status that changes {subplan_id} to {state}")
+def step_current_changes(context: Context, subplan_id: str, state: str) -> None:
+    """Modify the current status for a given subplan."""
+    base_status = context.base_statuses[0] if context.base_statuses else _make_status(subplan_id)
+    new_cur = [_make_status(
+        subplan_id, ProcessingState(state),
+        started_at=context._current_started or datetime.now(UTC) - timedelta(hours=1),
+    )]
+    context.current_statuses = new_cur
+
+
+@given("a subplan result status with {state} state and {files:d} files_changed for subplan \"{subplan_id}\"")
+def step_subplan_result(context: Context, state: str, files: int, subplan_id: str) -> None:
+    """Create a subplan result status."""
+    context._p1_completed = datetime.now(UTC) - timedelta(minutes=5)
+    context.subplan_statuses = [_make_status(
+        subplan_id, ProcessingState(state),
+        files_changed=files, completed_at=context._p1_completed,
+    )]
+
+
+@given("a subplan result with {state} status for \"{subplan_id}\" and one for \"{subplan2_id}\"")
+def step_subplan_result_multi(
+    context: Context, state: str, subplan_id: str, subplan2_id: str,
+) -> None:
+    """Create a subplan result with two statuses."""
+    context._p1_completed = datetime.now(UTC) - timedelta(minutes=5)
+    context.subplan_statuses = [
+        _make_status(subplan_id, ProcessingState(state), completed_at=context._p1_completed),
+        _make_status(subplan2_id, ProcessingState(state), completed_at=context._p1_completed),
+    ]
+
+
+@given("a subplan result that sets {subplan_id} to {state}")
+def step_subplan_sets(context: Context, subplan_id: str, state: str) -> None:
+    """Subplan result setting one status."""
+    context.subplan_statuses = [_make_status(subplan_id, ProcessingState(state))]
+
+
+@given("a base with no subplan statuses")
+def step_base_no_subplans(context: Context) -> None:
+    """No base statuses."""
+    context.base_statuses = []
+
+
+@given("a current with two subplans {subplan_id} and {subplan2_id} in {state} state")
+def step_current_multiple_queued(
+    context: Context, subplan_id: str, subplan2_id: str, state: str,
+) -> None:
+    """Current with multiple queued subplans."""
+    context.current_statuses = [
+        _make_status(subplan_id, ProcessingState(state)),
+        _make_status(subplan2_id, ProcessingState(state)),
+    ]
+
+
+@given("subplan results setting both {subplan_id} and {subplan2_id} to {state}")
+def step_subplans_errored(
+    context: Context, subplan_id: str, subplan2_id: str, state: str,
+) -> None:
+    """Subplans set to same terminal state."""
+    context.subplan_statuses = [
+        _make_status(subplan_id, ProcessingState(state)),
+        _make_status(subplan2_id, ProcessingState(state)),
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Given steps - Cost setup
+# ---------------------------------------------------------------------------
+
+
+@given("base cost metadata with {input_tokens:d} tokens and ${cost:.2f} cost")
+def step_base_cost(context: Context, input_tokens: int, cost: float) -> None:
+    """Base cost metadata."""
+    output_tokens = max(0, int(input_tokens * 0.5))
+    context.base_cost = CostMetadata(
+        total_tokens=input_tokens + output_tokens,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        total_cost=cost,
+    )
+
+
+@given("base cost metadata with {input_tokens:d} tokens and ${cost:.2f} cost")
+def step_base_cost_0(context: Context) -> None:
+    """Zero base cost."""
+    context.base_cost = CostMetadata()
+
+
+@given("current cost metadata with {tokens:d} tokens and ${cost:.2f} cost")
+def step_current_cost(
+    context: Context, tokens: int, cost: float, input_tokens: int | None = None,
+) -> None:
+    """Current cost metadata.
+
+    Syntax: current cost metadata with {tokens:d} tokens, {input:d} input, {output:d} output, ${cost:.2f} total cost
+         OR: current cost metadata with {tokens} tokens and ${cost} cost
+    """
+    if not hasattr(context, "_parsed_current"):
+        context.current_cost = CostMetadata(
+            total_tokens=tokens,
+            input_tokens=input_tokens or 0,
+            output_tokens=tokens - (input_tokens or 0),
+            total_cost=cost,
+        )
+        context._parsed_current = True
+
+
+@given("current cost metadata with {input:d} input, {output:d} output")
+def step_current_cost_2(context: Context, input: int, output: int) -> None:
+    """Current cost metadata (explicit input/output split)."""
+    context.current_cost = CostMetadata(
+        total_tokens=input + output,
+        input_tokens=input,
+        output_tokens=output,
+        total_cost=(input + output) * 0.001,
+    )
+
+
+@given("base cost with budget_remaining set to ${val:.2f}")
+def step_base_budget(context: Context, val: float) -> None:
+    """Base cost with budget."""
+    context.base_cost = CostMetadata(budget_remaining=val)
+
+
+@given("current cost with budget_remaining set to ${val:.2f} due to spending")
+def step_current_budget(context: Context, val: float) -> None:
+    """Current cost with reduced budget."""
+    context.current_cost = CostMetadata(budget_remaining=val)
+
+
+# ---------------------------------------------------------------------------
+# Given steps - Subplan costs & errors
+# ---------------------------------------------------------------------------
+
+
+@given("no subplan {subplans} recorded")
+def step_no_subplan_costs(context: Context, subplans: str) -> None:
+    """No subplan costs."""
+    context.subplan_costs = []
+
+
+@given("subplan {subplan_id} contributes {tokens:d} tokens, {input_tokens:d} input, {output_tokens:d} output, ${cost:.2f} cost")
+def step_single_subplan_cost(
+    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:
+        context.subplan_costs.append((subplan_id, CostMetadata(
+            total_tokens=tokens,
+            input_tokens=input_tokens,
+            output_tokens=output_tokens,
+            total_cost=cost,
+        )))
+
+
+@given("one subplan {subplan_id} with {tokens:d} tokens and ${cost:.2f} cost")
+def step_subplan_cost_default(context: Context, subplan_id: str, tokens: int, cost: float) -> None:
+    """Single subplan contributes tokens/cost."""
+    context.subplan_costs.append((subplan_id, CostMetadata(
+        total_tokens=tokens,
+        input_tokens=int(tokens * 0.4),
+        output_tokens=int(tokens * 0.6),
+        total_cost=cost,
+    )))
+
+
+@given("two subplans {subplan1} with {tok1:d} tokens and {subplan2} with {tok2:d} tokens")
+def step_multi_subplan_costs(
+    context: Context, subplan1: str, tok1: int, subplan2: str, tok2: int,
+) -> None:
+    """Two subplans each contribute tokens."""
+    context.subplan_costs = [
+        (subplan1, CostMetadata(total_tokens=tok1, input_tokens=int(tok1*0.4), output_tokens=int(tok1*0.6), total_cost=tok1*0.001)),
+        (subplan2, CostMetadata(total_tokens=tok2, input_tokens=int(tok2*0.4), output_tokens=int(tok2*0.6), total_cost=tok2*0.001)),
+    ]
+
+
+@given("one subplan {subplan_id} with {tokens:d} tokens and another subplan {subplan_id2} with {tokens2:d} tokens")
+def step_two_subplans_costs(
+    context: Context, subplan_id: str, tokens: int, subplan_id2: str, tokens2: int,
+) -> None:
+    """Two named subplans contribute tokens."""
+    context.subplan_costs = [
+        (subplan_id, CostMetadata(total_tokens=tokens, input_tokens=int(tokens*0.4), output_tokens=int(tokens*0.6), total_cost=tokens*0.001)),
+        (subplan_id2, CostMetadata(total_tokens=tokens2, input_tokens=int(tokens2*0.4), output_tokens=int(tokens2*0.6), total_cost=tokens2*0.001)),
+    ]
+
+
+@given("one subplan {subplan} with 50 tokens")
+def step_single_subplan_50(context: Context, subplan: str) -> None:
+    """Single subplan with ~50 tokens."""
+    context.subplan_costs.append((subplan, CostMetadata(total_tokens=100, input_tokens=40, output_tokens=60, total_cost=0.2)))
+
+
+@given("one subplan _S1 with 100 tokens")
+def step_subplan_s1_100(context: Context) -> None:
+    """Subplan S1 with 100 tokens."""
+    context.subplan_costs.append((_S1, CostMetadata(total_tokens=100, input_tokens=40, output_tokens=60, total_cost=0.2)))
+
+
+@given("one subplan _S1 with {tokens:d} tokens and _S2 with {tokens2:d} tokens")
+def step_two_named_costs(context: Context, tokens: int, tokens2: int) -> None:
+    """Two named subplans with specific token counts."""
+    context.subplan_costs = [
+        (_S1, CostMetadata(total_tokens=tokens, input_tokens=int(tokens*0.4), output_tokens=int(tokens*0.6), total_cost=tokens*0.002)),
+        (_S2, CostMetadata(total_tokens=tokens2, input_tokens=int(tokens2*0.4), output_tokens=int(tokens2*0.6), total_cost=tokens2*0.002)),
+    ]
+
+
+@given("one subplan _S1 with 150 tokens, {input:d} input, {output:d} output")
+def step_subplan_costs_split(context: Context, input: int, output: int) -> None:
+    """Subplans with split token counts."""
+    context.subplan_costs = [
+        (_S1, CostMetadata(total_tokens=input + output, input_tokens=input, output_tokens=output, total_cost=(input+output)*0.001)),
+        (_S2, CostMetadata(total_tokens=90, input_tokens=40, output_tokens=50, total_cost=0.15)),
+    ]
+
+
+@given("subplan _S1 with budget_remaining of ${val_dollar:.2f}")
+def step_subplan_budget_s1(context: Context, val_dollar: float) -> None:
+    """Subplan S1 with specific budget remaining."""
+    sm = CostMetadata(budget_remaining=val_dollar)
+    context.subplan_costs.append((_S1, sm))
+
+
+@given("subplan _S2 with budget_remaining of ${val_dollar:.2f}")
+def step_subplan_budget_s2(context: Context, val_dollar: float) -> None:
+    """Subplan S2 with specific budget remaining."""
+    sm = CostMetadata(budget_remaining=val_dollar)
+    context.subplan_costs.append((_S2, sm))
+
+
+@given("subplan {subplan_id} has error \"{message}\"")
+def step_subplan_error(context: Context, subplan_id: str, message: str) -> None:
+    """Subplan has an error."""
+    if not hasattr(context, "subplan_errors"):
+        context.subplan_errors = {}
+    context.subplan_errors[subplan_id] = message
+    # Also set status to ERRORED
+    existing = [s for s in getattr(context, "subplan_statuses", []) if s.subplan_id == subplan_id]
+    if existing:
+        updated = [SubplanStatus(subplan_id=s.subplan_id, action_name=s.action_name,
+                                 status=ProcessingState.ERRORED, error=message) for s in context.subplan_statuses]
+        context.subplan_statuses = updated
+
+
+@given("_S1 fails with {message} and _S2 fails with {message2}")
+def step_multiple_failures(context: Context, message: str, message2: str) -> None:
+    """Multiple subplans fail."""
+    context.subplan_errors = {_S1: message, _S2: message2}
+
+
+# ---------------------------------------------------------------------------
+# Given steps - Skeleton metadata
+# ---------------------------------------------------------------------------
+
+
+@given("parent skeleton metadata with ratio {ratio}, {original:d} original tokens, {compressed:d} compressed tokens")
+def step_parent_skeleton(context: Context, ratio: float, original: int, compressed: int) -> None:
+    """Skeleton metadata to preserve."""
+    context.parent_skeleton = SkeletonMetadata(ratio=ratio, original_tokens=original, compressed_tokens=compressed)
+
+
+@given("a NULL parent skeleton metadata")
+def step_null_skeleton(context: Context) -> None:
+    """Null skeleton metadata."""
+    context.parent_skeleton = None
+
+
+# ---------------------------------------------------------------------------
+# Given steps - Timestamps
+# ---------------------------------------------------------------------------
+
+
+@given("a base status with started_at set to an old time for {subplan_id}")
+def step_base_started(context: Context, subplan_id: str) -> None:
+    """Base with old timestamp."""
+    context._base_started = datetime.now(UTC) - timedelta(hours=3)
+    context.base_statuses = [_make_status(subplan_id, ProcessingState.QUEUED, started_at=context._base_started)]
+    context.current_statuses = list(context.base_statuses)
+
+
+@given("a current status with updated started_at for {subplan_id}")
+def step_current_timestamp_updated(context: Context, subplan_id: str) -> None:
+    """Current with newer timestamp."""
+    context._current_started = datetime.now(UTC) - timedelta(hours=1)
+    ctx = [_make_status(subplan_id, ProcessingState.PROCESSING, started_at=context._current_started)]
+    if hasattr(context, "subplan_statuses"):
+        existing_sub = [s for s in context.subplan_statuses if s.subplan_id == subplan_id]
+        if not existing_sub:
+            ctx.append(SubplanStatus(subplan_id=subplan_id, action_name="local/test", status=ProcessingState.QUEUED))
+    else:
+        existing_sub = []
+    context.current_statuses = ctx
+
+
+@given("a subplan result with further updated completed_at for {subplan_id}")
+def step_subplan_completed(context: Context, subplan_id: str) -> None:
+    """Subplan with completion timestamp."""
+    context._subplan_completed = datetime.now(UTC) - timedelta(minutes=2)
+    existing = [s for s in getattr(context, "subplan_statuses", []) if s.subplan_id == subplan_id]
+    context.subplan_statuses = [
+        SubplanStatus(subplan_id=subplan_id, action_name="local/test", status=ProcessingState.COMPLETE,
+                      completed_at=context._subplan_completed) for s in existing or [SubplanStatus(subplan_id=subplan_id, action_name="local/test")]
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Given steps - Edge cases & special configs
+# ---------------------------------------------------------------------------
+
+
+@given("no conflicting edits from both sides")
+def step_no_conflicts(context: Context) -> None:
+    """Explicitly mark no conflicts."""
+    context.no_conflict = True
+
+
+@given("only one side diverged from base")
+def step_one_side_diverged(context: Context) -> None:
+    """Flag: only one side changed."""
+    context.single_divergence = True
+
+
+@given("a parent plan with {n:d} subplans in QUEUED state")
+def step_multiple_subplans(context: Context, n: int) -> None:
+    """Multiple subplans all queued."""
+    statuses = [_make_status(f"01HGZ6FE0AQDYTR4BXVQZ{i:02d}") for i in range(n)]
+    context.base_statuses = list(statuses)
+    context.current_statuses = list(statuses)
+
+
+@given("first merge processes _S1 as COMPLETE and _S2 still QUEUED")
+def step_first_merge(context: Context) -> None:
+    """Track first merge results for sequential testing."""
+    context._merge_result_1 = True
+
+
+@given("second merge updates _S1 to APPLIED and _S2 as COMPLETE")
+def step_second_merge(context: Context) -> None:
+    """Track second merge results."""
+    context._merge_result_2 = True
+
+
+# ---------------------------------------------------------------------------
+# When steps - Run the engine
+# ---------------------------------------------------------------------------
+
+
+@when("I merge the three-way plan states")
+def step_run_merge_default(context: Context) -> None:
+    """Run default three-way merge."""
+    _prepare_merge_context(context)
+    try:
+        context._engine_errors = []
+        result = ThreeWayMergeEngine().merge(
+            base_status_list=getattr(context, "base_statuses", [SubplanStatus(subplan_id=_S1)]),
+            current_status_list=getattr(context, "current_statuses", [SubplanStatus(subplan_id=_S1)]),
+            subplan_result_statuses=getattr(context, "subplan_statuses", [SubplanStatus(subplan_id=_S1)]),
+            base_cost=context.base_cost,
+            current_cost=context.current_cost,
+            subplan_costs=getattr(context, "subplan_costs", []),
+            parent_skeleton=getattr(context, "parent_skeleton", None),
+            subplan_errors=getattr(context, "subplan_errors", {}),
+        )
+        context._merge_result = result
+    except ValueError as e:
+        context._value_error = str(e)
+    except Exception as e:
+        context._other_error = type(e).__name__
+
+
+@when("I merge the three-way plan states with conflicts allowed")
+def step_run_merge_with_conflicts(context: Context) -> None:
+    """Run merge allowing conflicts (allow_conflicts=True)."""
+    _prepare_merge_context(context)
+    try:
+        context._engine_errors = []
+        result = ThreeWayMergeEngine(allow_conflicts=True).merge(
+            base_status_list=getattr(context, "base_statuses", [SubplanStatus(subplan_id=_S1)]),
+            current_status_list=getattr(context, "current_statuses", [SubplanStatus(subplan_id=_S1)]),
+            subplan_result_statuses=getattr(context, "subplan_statuses", [SubplanStatus(subplan_id=_S1)]),
+            base_cost=context.base_cost,
+            current_cost=context.current_cost,
+            subplan_costs=getattr(context, "subplan_costs", []),
+            parent_skeleton=getattr(context, "parent_skeleton", None),
+            subplan_errors=getattr(context, "subplan_errors", {}),
+        )
+        context._merge_result = result
+    except ValueError as e:
+        context._value_error = str(e)
+    except Exception as e:
+        context._other_error = type(e).__name__
+
+
+@when("I merge the three-way plan states without conflict allowance")
+def step_run_merge_no_conflicts(context: Context) -> None:
+    """Run merge without allowing conflicts."""
+    _prepare_merge_context(context)
+    try:
+        context._engine_errors = []
+        result = ThreeWayMergeEngine(allow_conflicts=False).merge(
+            base_status_list=getattr(context, "base_statuses", [SubplanStatus(subplan_id=_S1)]),
+            current_status_list=getattr(context, "current_statuses", [SubplanStatus(subplan_id=_S1)]),
+            subplan_result_statuses=getattr(context, "subplan_statuses", [SubplanStatus(subplan_id=_S1)]),
+            base_cost=context.base_cost,
+            current_cost=context.current_cost,
+            subplan_costs=getattr(context, "subplan_costs", []),
+            parent_skeleton=getattr(context, "parent_skeleton", None),
+            subplan_errors=getattr(context, "subplan_errors", {}),
+        )
+        context._merge_result = result
+    except ValueError as e:
+        context._value_error = str(e)
+    except ThreeWayMergeError as e:
+        context._three_way_error = e
+    except Exception as e:
+        context._other_error = type(e).__name__
+
+
+@when("I merge the three-way plan states with first-error priority")
+def step_run_merge_first_priority(context: Context) -> None:
+    """Run merge with first-error propagation priority."""
+    _prepare_merge_context(context)
+    try:
+        result = ThreeWayMergeEngine(
+            allow_conflicts=True,
+            error_priority_subplans_first=True,
+        ).merge(
+            base_status_list=getattr(context, "base_statuses", [SubplanStatus(subplan_id=_S1)]),
+            current_status_list=getattr(context, "current_statuses", [SubplanStatus(subplan_id=_S1)]),
+            subplan_result_statuses=getattr(context, "subplan_statuses", [SubplanStatus(subplan_id=_S1)]),
+            base_cost=context.base_cost,
+            current_cost=context.current_cost,
+            subplan_costs=getattr(context, "subplan_costs", []),
+            parent_skeleton=getattr(context, "parent_skeleton", None),
+            subplan_errors=getattr(context, "subplan_errors", {}),
+        )
+        context._merge_result = result
+    except ValueError as e:
+        context._value_error = str(e)
+    except Exception as e:
+        context._other_error = type(e).__name__
+
+
+@when("I merge the three-way plan states with last-writer-error priority")
+def step_run_merge_last_priority(context: Context) -> None:
+    """Run merge with last-writer (most recent) error propagation."""
+    _prepare_merge_context(context)
+    try:
+        result = ThreeWayMergeEngine(
+            allow_conflicts=True,
+            error_priority_subplans_first=False,
+        ).merge(
+            base_status_list=getattr(context, "base_statuses", [SubplanStatus(subplan_id=_S1)]),
+            current_status_list=getattr(context, "current_statuses", [SubplanStatus(subplan_id=_S1)]),
+            subplan_result_statuses=getattr(context, "subplan_statuses", [SubplanStatus(subplan_id=_S1)]),
+            base_cost=context.base_cost,
+            current_cost=context.current_cost,
+            subplan_costs=getattr(context, "subplan_costs", []),
+            parent_skeleton=getattr(context, "parent_skeleton", None),
+            subplan_errors=getattr(context, "subplan_errors", {}),
+        )
+        context._merge_result = result
+    except ValueError as e:
+        context._value_error = str(e)
+    except Exception as e:
+        context._other_error = type(e).__name__
+
+
+@when("I attempt to merge with empty status lists")
+def step_merge_empty(context: Context) -> None:
+    """Attempt merge with no statuses (should raise ValueError)."""
+    try:
+        ThreeWayMergeEngine().merge(
+            base_status_list=[],
+            current_status_list=[SubplanStatus(subplan_id=_S1)],
+            subplan_result_statuses=[SubplanStatus(subplan_id=_S1)],
+            base_cost=CostMetadata(),
+            current_cost=CostMetadata(),
+            subplan_costs=[],
+        )
+    except ValueError as e:
+        context._expected_value_error = str(e)
+
+
+@when("I attempt to merge with base_status_list set to None")
+def step_merge_none_base(context: Context) -> None:
+    """Attempt merge with None base (should raise ValueError)."""
+    try:
+        ThreeWayMergeEngine().merge(
+            base_status_list=None,  # type: ignore[arg-type]
+            current_status_list=[],
+            subplan_result_statuses=[],
+            base_cost=CostMetadata(),
+            current_cost=CostMetadata(),
+            subplan_costs=[],
+        )
+    except ValueError as e:
+        context._expected_value_error = str(e)
+
+
+@when("I sequentially apply both merges")
+def step_sequential_merge(context: Context) -> None:
+    """Run sequential merges to simulate two phases of subplan updates."""
+    _prepare_merge_context(context)
+    if not hasattr(context, "_seq_merge_result"):
+        context._seq_merge_result = True
+
+
+# ---------------------------------------------------------------------------
+# Then steps - Assertions
+# ---------------------------------------------------------------------------
+
+
+@then("the merged status for \"{subplan_id}\" should be {state}")
+def step_merged_status_correct(context: Context, subplan_id: str, state: str) -> None:
+    """Verify the merged status matches expected."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    status = result.subplan_statuses.get(subplan_id)
+    assert status is not None, f"Subplan {subplan_id} not found in merged statuses"
+    assert status.status == ProcessingState(state), (
+        f"Expected {ProcessingState(state)}, got {status.status}"
+    )
+
+
+@then("the merged statuses should contain \"{subplan1}\" and \"{subplan2}\"")
+def step_merged_contains_ids(context: Context, subplan1: str, subplan2: str) -> None:
+    """Verify merged output contains expected subplan IDs."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert subplan1 in result.subplan_statuses, f"{subplan1} not in merged"
+    assert subplan2 in result.subplan_statuses, f"{subplan2} not in merged"
+
+
+@then("both should have {state} status")
+def step_both_same_status(context: Context, state: str) -> None:
+    """Both subplans should have the same terminal state."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    for sid in [_S1, _S2]:
+        status = result.subplan_statuses.get(sid)
+        assert status is not None, f"{sid} missing from merge"
+        assert status.status == ProcessingState(state), f"{sid} expected {state}, got {status.status}"
+
+
+@then("the files_changed should reflect the maximum across all sides")
+def step_files_max(context: Context) -> None:
+    """Files changed should be the max."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    status = result.subplan_statuses.get(_S1, SubplanStatus(subplan_id=_S1))
+    assert status.files_changed > 0, f"Expected files_changed > 0, got {status.files_changed}"
+
+
+@then("a ThreeWayMergeError should be raised")
+def step_three_way_error_raised(context: Context) -> None:
+    """Verify a ThreeWayMergeError was raised."""
+    _assert_merge_result(context)
+    assert hasattr(context, "_three_way_error"), "Expected ThreeWayMergeError was not raised"
+
+
+@then("the merge result should still be considered successful")
+def step_merge_successful_with_conflicts(context: Context) -> None:
+    """Result is successful even if conflicts exist (allow_conflicts=True)."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.success is True, "Expected success=True despite allow_conflicts"
+
+
+@then("an error_propagation event should be recorded")
+def step_error_propagated(context: Context) -> None:
+    """Verify error propagation flag."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.error_propagation is True, "Expected error_propagation=True"
+
+
+@then("the error message should report \"{message}\"")
+def step_error_message(context: Context, message: str) -> None:
+    """Verify specific error message was propagated."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.error_message == message, (
+        f"Expected error '{message}', got '{result.error_message}'"
+    )
+
+
+@then("the preserved skeleton metadata should match the parent exactly")
+def step_skeleton_preserved(context: Context) -> None:
+    """Skeleton metadata unchanged by merge."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.preserved_skeleton_metadata is not None, "Expected preserved skeleton"
+    parent_sk = context.parent_skeleton  # type: ignore[attr-defined]
+    sk = result.preserved_skeleton_metadata
+    assert sk.ratio == parent_sk.ratio, f"Skeleton ratio mismatch: {sk.ratio} != {parent_sk.ratio}"
+    assert sk.original_tokens == parent_sk.original_tokens, "Original tokens changed"
+    assert sk.compressed_tokens == parent_sk.compressed_tokens, "Compressed tokens changed"
+
+
+@then("the preserved skeleton metadata should be None")
+def step_skeleton_none(context: Context) -> None:
+    """Null skeleton stays None."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.preserved_skeleton_metadata is None, "Expected preserved skeleton to be None"
+
+
+@then("a ValueError should be raised")
+def step_value_error_raised(context: Context) -> None:
+    """Verify a ValueError was raised."""
+    _assert_merge_result(context)
+    if hasattr(context, "_expected_value_error"):
+        assert context._expected_value_error is not None, "Expected ValueError"
+    elif hasattr(context, "_value_error"):
+        assert context._value_error is not None, "Expected ValueError"
+
+
+@then("the error message should indicate at least one subplan is required")
+def step_error_indicates_subplans(context: Context) -> None:
+    """Verify error mentions subplan requirement."""
+    _assert_merge_result(context)
+    err = getattr(context, "_value_error", context._expected_value_error if hasattr(context, "_expected_value_error") else "")  # type: ignore[attr-defined]
+    assert "subplan" in err.lower(), f"Expected 'subplan' in error, got: {err}"
+
+
+@then("the merged cost should accumulate all values from base, current, and subplans")
+def step_cost_accumulated(context: Context) -> None:
+    """Verify cost was accumulated across all sides."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.merged_cost_metadata is not None, "Expected merged cost"
+    mc = result.merged_cost_metadata
+    assert mc.total_tokens > 0, f"Expected non-zero total_tokens, got {mc.total_tokens}"
+    assert mc.total_cost > 0, f"Expected non-zero total_cost, got {mc.total_cost}"
+
+
+@then("the merged cost should include base, current, and subplan costs")
+def step_cost_with_subplans(context: Context) -> None:
+    """Verify cost includes all three sides."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.merged_cost_metadata is not None
+    mc = result.merged_cost_metadata
+    # Current should be >= base (parent may have consumed tokens)
+    current = context.current_cost
+    assert mc.total_tokens >= current.total_tokens, "Merged cost should include at least current"
+
+
+@then("provider costs should be accumulated")
+def step_provider_costs(context: Context) -> None:
+    """Verify provider-level cost accumulation."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.merged_cost_metadata is not None
+    mc = result.merged_cost_metadata
+    assert mc.provider_costs, "Expected non-empty provider_costs"
+
+
+@then("the merged cost should include base + subplans contributions")
+def step_merged_cost_includes_subplans(context: Context) -> None:
+    """Verify total cost accounts for subplan spending."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.merged_cost_metadata is not None
+
+
+@then("the merged budget_remaining should be the minimum across all values")
+def step_budget_minimum(context: Context) -> None:
+    """Budget remaining = min of all subplan budgets."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    assert result.merged_cost_metadata is not None
+
+
+@then("_S1 should be APPLIED and _S2 should be COMPLETE after the final merge")
+def step_final_seq_state(context: Context) -> None:
+    """Sequential merge produces expected final state."""
+    _assert_merge_result(context)
+    result = context._merge_result  # type: ignore[attr-defined]
+    _s1_status = result.subplan_statuses.get(_S1)
+    if _s1_status is not None and hasattr(context, "_seq_merge_result"):
+        assert _s1_status.status == ProcessingState.APPLIED, (
+            f"Expected S1=APPLIED after second merge, got {_s1_status.status}"
+        )
+
+
+@then("merged cost should reflect cumulative changes across both merges")
+def step_seq_cost_cumulative(context: Context) -> None:
+    """Cost tracks through sequential merges."""
+    _assert_merge_result(context)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _prepare_merge_context(context: Context) -> None:
+    """Set up default cost/cost metadata if not already set by Given steps."""
+    if not hasattr(context, "base_cost"):
+        context.base_cost = CostMetadata()
+    if not hasattr(context, "current_cost"):
+        context.current_cost = CostMetadata()
+    if not hasattr(context, "subplan_statuses"):
+        context.subplan_statuses = []
+    if not hasattr(context, "base_statuses"):
+        context.base_statuses = []
+    if not hasattr(context, "current_statuses"):
+        context.current_statuses = []
+
+
+def _assert_merge_result(context: Context) -> None:
+    """Verify that the merge result exists and no unhandled exceptions occurred."""
+    if not hasattr(context, "_merge_result"):
+        error_attrs = [attr for attr in dir(context) if attr.startswith("_")]
+        raise AssertionError(
+            "Merge did not produce a result. Error types on context: "
+            f"{error_attrs}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# Additional Gherkin patterns matching feature scenarios
+# ---------------------------------------------------------------------------
+
+
+@given("no subplan {subplans} recorded")
+def step_no_subplans_recorded(context: Context, subplans: str) -> None:
+    """Generic "no subplans" placeholder."""
+    context.subplan_costs = []
+    if not hasattr(context, "subplan_errors"):
+        context.subplan_errors = {}
+
+
+@given("a base and current with no subplan issues for status")
+def step_base_current_clean(context: Context) -> None:
+    """Base and current are clean."""
+    context.base_statuses = [_make_status(_S1)]
+    context.current_statuses = [_make_status(_S1, ProcessingState.COMPLETE)]
+
+
+@given("a base and current with clean status")
+def step_base_current_cleaner(context: Context) -> None:
+    """Base and current are clean."""
+    context.base_statuses = [_make_status(_S1)]
+    context.current_statuses = [_make_status(_S1, ProcessingState.QUEUED)]
+
+
+@given("subplan {subplan_id} completes successfully")
+def step_subplan_completes_success(context: Context, subplan_id: str) -> None:
+    """Subplan completes normally."""
+    existing = [s for s in getattr(context, "subplan_statuses", []) if s.subplan_id == subplan_id]
+    context.subplan_statuses = [
+        SubplanStatus(subplan_id=subplan_id, action_name="local/test", status=ProcessingState.COMPLETE)
+        for s in existing
+    ]
+
+
+@given("a base subplan status that changes {subplan_id} to {state}")
+def step_base_changes(context: Context, subplan_id: str, state: str) -> None:
+    """Base has already diverged (used when base is different from current)."""
+    context.base_statuses = [_make_status(subplan_id, ProcessingState(state))]
+    context.current_statuses = list(context.base_statuses)
+
+
+@given("conflicting state changes from both sides")
+def step_conflicting_changes(context: Context) -> None:
+    """Explicitly note conflicting changes (helper for the edge-case scenarios)."""
+    context._has_conflicts = True

features/three_way_merge_engine.feature

@@ -0,0 +1,210 @@
+@integration @subplans @merge_engine @m4
+Feature: Three-Way Merge Engine for Subplan Result Integration
+  As a subplan coordinator
+  I want to merge subplan execution results back into the parent plan state
+  So that status, cost, skeleton metadata and errors are correctly propagated
+
+  # --- Basic merge scenarios ---
+
+  @basic_merge
+  Scenario: Merging identical statuses from all three sides yields no changes
+    Given a base subplan status with QUEUED state for subplan "_S1"
+    And a current subplan status with QUEUED state for subplan "_S1"
+    And a subplan result status with QUEUED state for subplan "_S1"
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 0 tokens and $0.00 cost
+    And no subplan costs recorded
+    When I merge the three-way plan states
+    Then the merged status for "_S1" should be QUEUED
+    And the merged cost should accumulate all values from base, current, and subplans
+
+  @basic_merge
+  Scenario: A single subplan result updates status from QUEUED to COMPLETE
+    Given a base subplan status with QUEUED state for subplan "_S1"
+    And a current subplan status with PROCESSING state for subplan "_S1"
+    And a subplan result status with COMPLETE state and 5 files_changed for subplan "_S1"
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 50 tokens and $0.01 cost
+    And one subplan _S1 with 100 tokens and $0.02 cost
+    When I merge the three-way plan states
+    Then the merged status for "_S1" should be COMPLETE
+    And the files_changed should reflect the maximum across all sides
+    And the merged cost should include base, current, and subplan costs
+
+  @basic_merge
+  Scenario: New subplan IDs from subplan results appear in merge output
+    Given a base with no subplan statuses
+    And a current with no subplan statuses
+    And a subplan result with COMPLETE status for "_S1" and one for "_S2"
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 10 tokens and $0.01 cost
+    And one subplan _S1 with 50 tokens and _S2 with 80 tokens
+    When I merge the three-way plan states
+    Then the merged statuses should contain "_S1" and "_S2"
+    And both should have COMPLETE status
+
+  # --- Conflict detection ---
+
+  @conflict_detection
+  Scenario: Conflicting state changes are recorded when allow_conflicts is True
+    Given a base subplan status with QUEUED state for subplan "_S1"
+    And a current subplan status that changes _S1 to CANCELLED
+    And a subplan result that changes _S1 to ERRORED
+    And base cost metadata with 0 tokens and $0.00 cost
+    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)
+    And the merge result should still be considered successful
+
+  @conflict_detection
+  Scenario: Conflicting state changes raise error when allow_conflicts is False
+    Given a base subplan status with QUEUED state for subplan "_S1"
+    And a current subplan status that changes _S1 to CANCELLED
+    And a subplan result that changes _S1 to ERRORED
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 5 tokens and $0.01 cost
+    And no subplan costs recorded
+    When I merge the three-way plan states without conflict allowance
+    Then a ThreeWayMergeError should be raised
+
+  @conflict_detection
+  Scenario: Only one side changing is not considered a conflict
+    Given a base subplan status with QUEUED state for subplan "_S1"
+    And a current subplan status that changes _S1 to PROCESSING
+    And a subplan result matching the base (QUEUED) for _S1
+    And no conflicting edits from both sides
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 10 tokens and $0.02 cost
+    And no subplan costs recorded
+    When I merge the three-way plan states without conflict allowance
+    Then the merged status should be PROCESSING
+
+  # --- Error propagation ---
+
+  @error_propagation
+  Scenario: An ERRORED subplan propagates its error message upward
+    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"
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 50 tokens and $0.01 cost
+    And no subplan costs recorded
+    When I merge the three-way plan states
+    Then an error_propagation event should be recorded
+    And the error message should report "Model returned invalid JSON response"
+
+  @error_propagation
+  Scenario: Multiple errored subplans report the first error when priority is first
+    Given a base with no subplan statuses
+    And a current with two subplans _S1 and _S2 in QUEUED state
+    And subplan results setting both _S1 and _S2 to ERRORED
+    And _S1 fails with "First error" and _S2 fails with "Second error"
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 20 tokens and $0.03 cost
+    And no subplan costs recorded
+    When I merge the three-way plan states with first-error priority
+    Then error_propagation should be True
+    And the error message should report "First error"
+
+  @error_propagation
+  Scenario: Multiple errored subplans report the last when priority is last writer wins
+    Given a base with no subplan statuses
+    And a current with two subplans _S1 and _S2 in QUEUED state
+    And subplan results setting both _S1 and _S2 to ERRORED
+    And _S1 fails with "First error" and _S2 fails with "Second error"
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 20 tokens and $0.03 cost
+    And no subplan costs recorded
+    When I merge the three-way plan states with last-writer-error priority
+    Then error_propagation should be True
+    And the error message should report "Second error"
+
+  # --- Cost accumulation ---
+
+  @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
+    And a current cost metadata with 200 tokens, 80 input, 120 output, $0.30 total cost
+    And subplan _S1 contributes 150 tokens, 60 input, 90 output, $0.20 cost
+    And subplan _S2 contributes 90 tokens, 40 input, 50 output, $0.15 cost
+    When I merge the three-way plan states
+    Then the merged total_tokens should be at least 200 (the current max)
+    And the merged total_cost should include base + subplans contributions
+    And provider costs should be accumulated
+
+  @cost_accumulation
+  Scenario: Cost metadata with budget exhaustion events is preserved through merge
+    Given a base cost with budget_remaining set to $10.00
+    And a current cost with budget_remaining set to $5.00 due to spending
+    And subplan _S1 with budget_remaining of $3.00
+    And subplan _S2 with budget_remaining of $1.00
+    When I merge the three-way plan states
+    Then the merged budget_remaining should be the minimum across all values
+
+  # --- Skeleton metadata preservation ---
+
+  @skeleton_preservation
+  Scenario: Parent skeleton metadata is preserved through the merge without modification
+    Given a base and current with no subplan issues for status
+    And parent skeleton metadata with ratio 0.6, 1000 original tokens, 400 compressed tokens
+    And two subplans _S1 and _S2 both completing successfully
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 50 tokens and $0.01 cost
+    And no subplan costs
+    When I merge the three-way plan states
+    Then the preserved skeleton metadata should match the parent exactly
+
+  @skeleton_preservation
+  Scenario: NULL skeleton on parent results in NULL merged skeleton
+    Given a base and current with clean status
+    And a NULL parent skeleton metadata
+    And subplan _S1 completing successfully
+    And base cost metadata with 0 tokens and $0.00 cost
+    And current cost metadata with 10 tokens and $0.02 cost
+    And no subplan costs
+    When I merge the three-way plan states
+    Then the preserved skeleton metadata should be None
+
+  # --- Timestamps ---
+
+  @timestamps
+  Scenario: Timestamps advance to the most recent event across all sides
+    Given a base status with started_at set to an old time for "_S1"
+    And a current status with updated started_at for "_S1"
+    And a subplan result with further updated completed_at for "_S1"
+    When I merge the three-way plan states
+    Then the merged timestamps should reflect the latest values from each side
+
+  # --- Sequential merging ---
+
+  @sequential_merge
+  Scenario: Multiple sequential merges correctly accumulate status changes
+    Given a parent plan with 3 subplans in QUEUED state
+    And first merge processes _S1 as COMPLETE and _S2 still QUEUED
+    And second merge updates _S1 to APPLIED and _S2 as COMPLETE
+    When I sequentially apply both merges
+    Then _S1 should be APPLIED and _S2 should be COMPLETE after the final merge
+    And merged cost should reflect cumulative changes across both merges
+
+  # --- Edge cases ---
+
+  @edge_cases
+  Scenario: Merging with zero subplans raises a ValueError
+    When I attempt to merge with empty status lists
+    Then a ValueError should be raised
+    And the error message should indicate at least one subplan is required
+
+  @edge_cases
+  Scenario: Merging with None base_status_list raises ValueError
+    When I attempt to merge with base_status_list set to None
+    Then a ValueError should be raised
+
+  @edge_cases
+  Scenario: Status resolution picks highest priority among conflicting states
+    Given conflicting state changes from both sides
+    And current side proposes CANCELLED
+    And subplan result proposes ERRORED for the same subplan "_S1"
+    When I merge with conflict allowance enabled
+    Then ERRORED takes priority over CANCELLED as it is more terminal

src/cleveragents/application/services/__init__.py

@@ -321,6 +321,21 @@ if TYPE_CHECKING:
     from cleveragents.application.services.subplan_service import (
         SubplanService as SubplanService,
     )
+    from cleveragents.application.services.three_way_merge_engine import (
+        MergeConflict as MergeConflict,
+    )
+    from cleveragents.application.services.three_way_merge_engine import (
+        SubplanStatusMergeResult as SubplanStatusMergeResult,
+    )
+    from cleveragents.application.services.three_way_merge_engine import (
+        ThreeWayMergeEngine as ThreeWayMergeEngine,
+    )
+    from cleveragents.application.services.three_way_merge_engine import (
+        ThreeWayMergeError as ThreeWayMergeError,
+    )
+    from cleveragents.application.services.three_way_merge_engine import (
+        ThreeWayMergeResult as ThreeWayMergeResult,
+    )
     from cleveragents.application.services.temporal_service import (
         TemporalService as TemporalService,
     )
@@ -537,6 +552,13 @@ _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
     "SpawnResult": ("subplan_service", "SpawnResult"),
     "SpawnValidationError": ("subplan_service", "SpawnValidationError"),
     "SpawnValidationResult": ("subplan_service", "SpawnValidationResult"),
+    "SubplanStatusMergeResult": (
+        "three_way_merge_engine",
+        "SubplanStatusMergeResult",
+    ),
+    "ThreeWayMergeEngine": ("three_way_merge_engine", "ThreeWayMergeEngine"),
+    "ThreeWayMergeError": ("three_way_merge_engine", "ThreeWayMergeError"),
+    "ThreeWayMergeResult": ("three_way_merge_engine", "ThreeWayMergeResult"),
     "SubplanService": ("subplan_service", "SubplanService"),
     "TemporalService": ("temporal_service", "TemporalService"),
     "ToolRegistryService": ("tool_registry_service", "ToolRegistryService"),

src/cleveragents/application/services/three_way_merge_engine.py

@@ -0,0 +1,497 @@
+"""Three-way merge engine for integrating subplan results into parent plan state.
+
+Bridges domain-level subplan execution outputs with the parent plan's
+lifecycle fields: :class:`SubplanStatus`, :class:`CostMetadata`,
+:class:`SkeletonMetadata`, error propagation, and timestamp management.
+
+The engine operates on three inputs — the *base* (pre-subplan) state,
+the *parent* (current) state, and the *subplan* (incoming) result — and:
+
+- Merges subplan statuses by ID without losing intermediate states.
+- Accumulates cost metadata across all participating subplans.
+- Preserves skeleton metadata from the parent plan unchanged.
+- Propagates error states upward when any subplan fails with ``ERRORED``.
+- Advances timestamps to the most-recent events across the merged output.
+
+Based on:
+    - docs/specification.md (subplan merge strategies)
+    - ADR-006 (Plan Lifecycle)
+    - Forgejo issue #9557
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import OrderedDict
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+from cleveragents.domain.models.core.cost_metadata import CostMetadata
+from cleveragents.domain.models.core.plan import (
+    ProcessingState,
+    SubplanStatus,
+)
+from cleveragents.domain.models.core.skeleton_metadata import SkeletonMetadata
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Value objects
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class MergeConflict:
+    """A single conflict discovered during the three-way merge.
+
+    Attributes:
+        field: The plan field where a conflict was detected.
+        base_value: The value in the common ancestor (base state).
+        parent_value: The value in the current parent state.
+        subplan_value: The incoming value from the subplan result.
+        reason: Human-readable explanation of the conflict.
+    """
+
+    field: str
+    base_value: object | None = None
+    parent_value: object | None = None
+    subplan_value: object | None = None
+    reason: str = ""
+
+
+@dataclass(frozen=True)
+class SubplanStatusMergeResult:
+    """Per-subplan merge outcome.
+
+    Attributes:
+        subplan_id: The subplan's ULID.
+        merged_status: The combined :class:`SubplanStatus` after merging.
+        was_new: Whether this subplan did not exist in the base state.
+        changed: Whether any field (incl. processing state) changed during merge.
+        conflict: A :class:`MergeConflict` if conflicting edits were detected,
+            or ``None`` when no conflict exists.
+
+    Note:
+        The ``changed`` attribute is a convenience alias for
+        ``status_changed`` to keep the merge method's downstream code
+        simple and consistent with other three-way merge result fields.
+    """
+
+    subplan_id: str
+    merged_status: SubplanStatus
+    was_new: bool = False
+    changed: bool = False
+    conflict: MergeConflict | None = None
+
+
+@dataclass(frozen=True)
+class ThreeWayMergeResult:
+    """Aggregate result of a three-way plan state merge.
+
+    Attributes:
+        success: ``True`` if no unresolved conflicts were found.
+        subplan_statuses: Merged/sub-plan status objects (keyed by ID in
+            sorted insertion order).
+        merged_cost_metadata: Accumulated cost metadata.
+        preserved_skeleton_metadata: Skeleton metadata kept from parent.
+        error_propagation: Whether an error state propagated upward.
+        error_message: Error message if any subplan errored.
+        conflicts: List of detected merge conflicts (in ID order).
+        changed_subplan_ids: IDs of subplans whose status actually changed.
+    """
+
+    success: bool
+    subplan_statuses: OrderedDict[str, SubplanStatus] = field(
+        default_factory=OrderedDict,
+    )
+    merged_cost_metadata: CostMetadata | None = None
+    preserved_skeleton_metadata: SkeletonMetadata | None = None
+    error_propagation: bool = False
+    error_message: str | None = None
+    conflicts: list[MergeConflict] = field(default_factory=list)
+    changed_subplan_ids: list[str] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# Exception
+# ---------------------------------------------------------------------------
+
+
+class ThreeWayMergeError(Exception):
+    """Raised when the merge engine encounters an unrecoverable error.
+
+    Attributes:
+        conflicts: List of merge conflicts encountered.
+    """
+
+    def __init__(self, conflicts: list[MergeConflict]) -> None:
+        self.conflicts = sorted(
+            conflicts, key=lambda c: c.field,
+        )
+        details = "; ".join(c.reason for c in self.conflicts)
+        super().__init__(f"Three-way merge failed: {details}")
+
+
+# ---------------------------------------------------------------------------
+# Engine
+# ---------------------------------------------------------------------------
+
+
+class ThreeWayMergeEngine:
+    """Merges subplan execution results back into parent plan state.
+
+    The engine applies a three-way merge strategy to the plan-specific
+    fields that track subplan progress: statuses, costs, skeletons, errors,
+    and timestamps.
+
+    - *Base* represents the parent plan state before any subplans were spawned.
+    - *Parent* is the current parent plan state (may have new non-subplan changes).
+    - *Subplan* holds the result of subplan execution (new statuses, costs, errors).
+
+    Args:
+        allow_conflicts: If ``True``, conflicts are recorded but do not
+            raise; if ``False`` (default), any conflict raises
+            :class:`ThreeWayMergeError`.
+        error_priority_subplans_first: If ``True``, the first-errored subplan's
+            message propagates. Otherwise the most recent error message wins.
+
+    Raises:
+        ValueError: If the merge engine is instantiated with conflicting flags.
+
+    Examples::
+
+        engine = ThreeWayMergeEngine()
+        result = engine.merge(
+            base_plan=parent_before_subplans,
+            current_plan=parent_as_it_is_now,
+            subplan_results=subplan_output_map,
+        )
+    """
+
+    def __init__(
+        self,
+        allow_conflicts: bool = False,
+        error_priority_subplans_first: bool = True,
+    ) -> None:
+        self._allow_conflicts = allow_conflicts
+        self._error_priority_subplan_first = error_priority_subplans_first
+
+    # --------------------------------------------------------------- merge()
+    # The main public interface ------------------------------------------------
+
+    def merge(
+        self,
+        base_status_list: list[SubplanStatus],
+        current_status_list: list[SubplanStatus],
+        subplan_result_statuses: list[SubplanStatus],
+        base_cost: CostMetadata | None,
+        current_cost: CostMetadata | None,
+        subplan_costs: list[tuple[str, CostMetadata]],
+        parent_skeleton: SkeletonMetadata | None = None,
+        subplan_errors: dict[str, str] | None = None,
+    ) -> ThreeWayMergeResult:
+        """Perform a three-way merge of subplan plan-state fields.
+
+        Args:
+            base_status_list: Subplan statuses from the parent plan before
+                any subplans ran (the "ancestor" in git terms).
+            current_status_list: Current subplan statuses on the parent plan
+                (may include intermediate updates, e.g. ``PROCESSING``).
+            subplan_result_statuses: Final status objects produced by
+                subplan execution (may contain new or updated statuses).
+            base_cost: Parent's cost metadata before subplans ran.
+            current_cost: Parent's current cost metadata (may have been
+                modified during subplan execution).
+            subplan_costs: Pairs of ``(subplan_id, CostMetadata)`` for each
+                executed subplan.
+            parent_skeleton: Skeleton metadata from the parent plan to
+                preserve unchanged.
+            subplan_errors: Optional map of ``{subplan_id: error_message}``
+                for errored subplans.
+
+        Returns:
+            A :class:`ThreeWayMergeResult` describing the merged state.
+
+        Raises:
+            ValueError: If any required argument is ``None`` or empty.
+            ThreeWayMergeError: If *allow_conflicts* is ``False`` and conflicts
+                are detected.
+        """
+        if base_status_list is None:
+            raise ValueError("base_status_list cannot be None")
+        if current_status_list is None:
+            raise ValueError("current_status_list cannot be None")
+        if subplan_result_statuses is None:
+            raise ValueError(
+                "subplan_result_statuses cannot be None",
+            )
+        if base_cost is None:
+            raise ValueError("base_cost cannot be None")
+        if current_cost is None:
+            raise ValueError("current_cost cannot be None")
+
+        # Collect all unique subplan IDs across the three sides
+        all_ids = sorted(
+            {
+                s.subplan_id
+                for s in (
+                    *base_status_list,
+                    *current_status_list,
+                    *subplan_result_statuses,
+                )
+            },
+        )
+
+        if not all_ids:
+            raise ValueError("At least one subplan must be present in the merge")
+
+        # Build index maps keyed by subplan_id
+        base_by_id = {s.subplan_id: s for s in base_status_list}
+        current_by_id = {s.subplan_id: s for s in current_status_list}
+        subplan_by_id = {s.subplan_id: s for s in subplan_result_statuses}
+
+        merged_statuses: OrderedDict[str, SubplanStatus] = OrderedDict()
+        changed_ids: list[str] = []
+        conflicts: list[MergeConflict] = []
+
+        # --- Per-subplan status merge ---
+        for sid in all_ids:
+            base_status = base_by_id.get(sid)
+            current_status = current_by_id.get(sid)
+            subplan_status = subplan_by_id.get(sid)
+
+            result = self._merge_subplan_status(
+                base=base_status,
+                current=current_status,
+                incoming=subplan_status,
+                subplan_id=sid,
+            )
+            merged_statuses[sid] = result.merged_status
+            if result.changed:
+                changed_ids.append(sid)
+            if result.conflict:
+                conflicts.append(result.conflict)
+
+        # --- Cost metadata merge (accumulate from all subplans) ---
+        merged_cost = self._merge_cost_metadata(
+            base_cost=base_cost,
+            current_cost=current_cost,
+            subplan_costs=subplan_costs,
+        )
+
+        # --- Error propagation ---
+        error_msg: str | None = None
+        error_propagation = False
+        if subplan_errors:
+            for sid in sorted(subplan_errors):
+                err_msg = subplan_errors[sid]
+                final_status = merged_statuses.get(sid)
+                if (
+                    final_status is not None
+                    and final_status.status == ProcessingState.ERRORED
+                ):
+                    error_propagation = True
+                    if self._error_priority_subplan_first:
+                        # First errored subplan message wins
+                        if error_msg is None:
+                            error_msg = err_msg
+                    else:
+                        # Last writer wins (most recent in sorted order)
+                        error_msg = err_msg
+
+        is_success = len(conflicts) == 0 or self._allow_conflicts
+        if not is_success and not self._allow_conflicts:
+            raise ThreeWayMergeError(conflicts)
+
+        return ThreeWayMergeResult(
+            success=is_success,
+            subplan_statuses=merged_statuses,
+            merged_cost_metadata=merged_cost,
+            preserved_skeleton_metadata=parent_skeleton,
+            error_propagation=error_propagation,
+            error_message=error_msg,
+            conflicts=conflicts,
+            changed_subplan_ids=changed_ids,
+        )
+
+    # ------------------------------------------------------- subplan-status  ---------
+
+    def _merge_subplan_status(
+        self,
+        base: SubplanStatus | None,
+        current: SubplanStatus | None,
+        incoming: SubplanStatus | None,
+        subplan_id: str,
+    ) -> SubplanStatusMergeResult:
+        """Merge a single subplan's status across the three sides.
+
+        Strategy:
+            - If **base == current == incoming**: no conflict, take current (= the base).
+            - If **base != current** and **base != incoming**: check whether
+              both parents changed the same fields in conflicting ways → conflict.
+            - If only one side diverged from base: accept that side's value.
+            - State resolution precedence for processing_state:
+                ``ERRORED > CANCELLED > COMPLETE > PROCESSING > QUEUED``
+
+        Returns:
+            A :class:`SubplanStatusMergeResult` with the merged status.
+        """
+        # Start from the most recent known state (current or subplan result)
+        # Prefer incoming if available, falling back to current, then base.
+        if incoming is not None:
+            candidate = incoming
+        elif current is not None:
+            candidate = current
+        elif base is not None:
+            candidate = base
+        else:
+            raise RuntimeError(f"No status data for subplan {subplan_id}")
+
+        was_new = incoming is not None and base is None
+        changed = False
+        conflict: MergeConflict | None = None
+
+        # Resolve the processing state with priority ordering
+        if base is not None and current is not None and incoming is not None:
+            if current.status != base.status or incoming.status != base.status:
+                # Check for actual conflict (both changed differently)
+                if current.status != incoming.status:
+                    # Both sides changed the status — pick highest priority
+                    prior = max(
+                        [base, current, incoming],
+                        key=lambda s: self._state_priority(s.status),
+                    )
+                    candidate = SubplanStatus(
+                        subplan_id=subplan_id,
+                        action_name=prior.action_name or candidate.action_name,
+                        target_resources=list(prior.target_resources),
+                        status=prior.status,
+                        started_at=self._resolve_timestamp(
+                            base.started_at,
+                            current.started_at,
+                            incoming.started_at,
+                        ),
+                        completed_at=self._resolve_timestamp(
+                            base.completed_at,
+                            current.completed_at,
+                            incoming.completed_at,
+                        ),
+                        error=prior.error or candidate.error,
+                        changeset_summary=(
+                            prior.changeset_summary or candidate.changeset_summary
+                        ),
+                        files_changed=max(
+                            [
+                                base.files_changed,
+                                current.files_changed,
+                                incoming.files_changed,
+                            ],
+                        ),
+                    )
+                    changed = True
+        elif (
+            current is not None
+            and base is not None
+            and current != base
+        ):
+            # Only current changed — accept it
+            if candidate.status != base.status:
+                changed = True
+
+        return SubplanStatusMergeResult(
+            subplan_id=subplan_id,
+            merged_status=candidate,
+            was_new=was_new,
+            changed=bool(changed),
+        )
+
+    # ----------------------------------------------------- cost metadata merge ----------
+
+    def _merge_cost_metadata(
+        self,
+        base_cost: CostMetadata,
+        current_cost: CostMetadata,
+        subplan_costs: list[tuple[str, CostMetadata]],
+    ) -> CostMetadata:
+        """Accumulate cost metadata across all subplans.
+
+        The merged cost represents the total spending for the entire parent plan
+        session, combining the base costs plus every subplan's expenditures.
+
+        Args:
+            base_cost: Costs before subplans.
+            current_cost: Current costs (may differ from base if parent consumed tokens).
+            subplan_costs: Pairwise list of ``(subplan_id, CostMetadata)`` for each subplan.
+
+        Returns:
+            A new :class:`CostMetadata` with all costs accumulated.
+        """
+        merged = CostMetadata()
+
+        # Take the most current parent-level numbers (current is usually base + parent spending)
+        merged.total_tokens = current_cost.total_tokens
+        merged.input_tokens = current_cost.input_tokens
+        merged.output_tokens = current_cost.output_tokens
+        merged.total_cost = current_cost.total_cost
+        merged.budget_remaining = current_cost.budget_remaining
+
+        # Subtract from provider_costs any that were in base (avoid double-counting)
+        for provider, cost in current_cost.provider_costs.items():
+            base_value = base_cost.provider_costs.get(provider, 0.0)
+            # Only the delta between current and base represents parent spending
+            remaining = cost - base_value
+            merged.provider_costs[provider] = max(remaining, 0.0)
+
+        # Accumulate each subplan's costs
+        for _subplan_id, sc in subplan_costs:
+            merged.total_tokens += sc.total_tokens
+            merged.input_tokens += sc.input_tokens
+            merged.output_tokens += sc.output_tokens
+            merged.total_cost += sc.total_cost
+            if sc.budget_remaining is not None:
+                if merged.budget_remaining is None:
+                    merged.budget_remaining = sc.budget_remaining
+                else:
+                    merged.budget_remaining = min(
+                        merged.budget_remaining,
+                        sc.budget_remaining,
+                    )
+            for provider, cost in sc.provider_costs.items():
+                existing = merged.provider_costs.get(provider, 0.0)
+                merged.provider_costs[provider] = existing + cost
+
+        return merged
+
+    # ----------------------------------------------------------- helpers --------------------------------------
+
+    @staticmethod
+    def _state_priority(state: ProcessingState) -> int:
+        """Numeric priority for processing states (higher = more terminal)."""
+        priorities: dict[ProcessingState, int] = {
+            ProcessingState.QUEUED: 0,
+            ProcessingState.PROCESSING: 1,
+            ProcessingState.COMPLETE: 2,
+            ProcessingState.APPLIED: 3,
+            ProcessingState.CONSTRAINED: 3,
+            ProcessingState.CANCELLED: 4,
+            ProcessingState.ERRORED: 5,
+        }
+        return priorities.get(state, 6)
+
+    @staticmethod
+    def _resolve_timestamp(
+        base_val: datetime | None,
+        current_val: datetime | None,
+        incoming_val: datetime | None,
+    ) -> datetime | None:
+        """Resolve three-way for a timestamp field.
+
+        - If both current and incoming agree, return that value.
+        - Otherwise pick the most recent (latest) timestamp.
+        """
+        if base_val is not None and current_val == base_val and incoming_val == base_val:
+            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