feat(subplan): execute and merge subplans

Implement SubplanExecutionService and SubplanMergeService to enable
parent plans to decompose work into coordinated child subplans with
configurable execution modes (sequential, parallel, dependency-ordered)
and merge strategies (git_three_way, sequential_apply, fail_on_conflict,
last_wins).

- SubplanExecutionService: schedules subplan execution with retry
  support via SubplanFailureHandler, max_parallel limits, fail_fast
  semantics, and topological DAG ordering for dependency mode
- SubplanMergeService: wraps sandbox merge infrastructure to combine
  subplan sandbox outputs using the configured merge strategy
- BDD tests: 21 scenarios covering all execution modes, merge
  strategies, validation, and integration flows
- Robot tests: 11 integration test cases with helper module
- ASV benchmarks: performance benchmarks for execution and merge
- Documentation: reference guide in docs/reference/subplans.md

ISSUES CLOSED: #184

benchmarks/subplan_execution_bench.py

@@ -0,0 +1,156 @@
+"""Airspeed Velocity benchmarks for subplan execution scheduler overhead.
+
+Measures scheduling, merge, and integrated execution+merge performance for
+SubplanExecutionService and SubplanMergeService across all execution modes
+and merge strategies.
+"""
+
+from __future__ import annotations
+
+from cleveragents.application.services.subplan_execution_service import (
+    SubplanExecutionOutput,
+    SubplanExecutionService,
+)
+from cleveragents.application.services.subplan_merge_service import (
+    SubplanMergeService,
+)
+from cleveragents.domain.models.core.plan import (
+    ExecutionMode,
+    ProcessingState,
+    SubplanConfig,
+    SubplanMergeStrategy,
+    SubplanStatus,
+)
+
+_S1 = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
+_S2 = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
+_S3 = "01HGZ6FE0AQDYTR4BXVQZ6EC00"
+
+
+def _make_status(subplan_id: str) -> SubplanStatus:
+    return SubplanStatus(
+        subplan_id=subplan_id,
+        action_name="local/bench-sub",
+    )
+
+
+def _noop_executor(status: SubplanStatus) -> SubplanExecutionOutput:
+    return SubplanExecutionOutput(
+        subplan_id=status.subplan_id,
+        success=True,
+        files={f"src/{status.subplan_id[-4:]}.py": f"# {status.subplan_id}\n"},
+        files_changed=1,
+    )
+
+
+class SubplanExecutionSchedulerSuite:
+    """Benchmark SubplanExecutionService scheduling overhead."""
+
+    def setup(self) -> None:
+        """Prepare fixtures for scheduler benchmarks."""
+        self.statuses = [_make_status(_S1), _make_status(_S2), _make_status(_S3)]
+        self.seq_config = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL)
+        self.par_config = SubplanConfig(
+            execution_mode=ExecutionMode.PARALLEL, max_parallel=3
+        )
+        self.dep_config = SubplanConfig(execution_mode=ExecutionMode.DEPENDENCY_ORDERED)
+        self.dep_graph = {_S1: [], _S2: [_S1], _S3: [_S2]}
+
+    def time_sequential_execution(self) -> None:
+        """Time sequential execution of 3 subplans."""
+        service = SubplanExecutionService(
+            config=self.seq_config, executor_fn=_noop_executor
+        )
+        service.execute_all(subplan_statuses=self.statuses, base_files={})
+
+    def time_parallel_execution(self) -> None:
+        """Time parallel execution of 3 subplans."""
+        service = SubplanExecutionService(
+            config=self.par_config, executor_fn=_noop_executor
+        )
+        service.execute_all(subplan_statuses=self.statuses, base_files={})
+
+    def time_dependency_ordered_execution(self) -> None:
+        """Time dependency-ordered execution of 3 subplans."""
+        service = SubplanExecutionService(
+            config=self.dep_config, executor_fn=_noop_executor
+        )
+        service.execute_all(
+            subplan_statuses=self.statuses,
+            base_files={},
+            dependency_graph=self.dep_graph,
+        )
+
+    def time_service_construction(self) -> None:
+        """Time constructing SubplanExecutionService."""
+        SubplanExecutionService(config=self.seq_config, executor_fn=_noop_executor)
+
+
+class SubplanMergeStrategySuite:
+    """Benchmark SubplanMergeService strategy overhead."""
+
+    def setup(self) -> None:
+        """Prepare fixtures for merge benchmarks."""
+        self.base_files = {"src/main.py": "line1\nline2\nline3\n"}
+        self.outputs = [
+            (_S1, {"src/main.py": "line1\nline2\nline3\nnew_a\n"}),
+            (_S2, {"src/main.py": "new_b\nline1\nline2\nline3\n"}),
+        ]
+        self.overlapping_outputs = [
+            (_S1, {"src/main.py": "first version\n"}),
+            (_S2, {"src/main.py": "second version\n"}),
+        ]
+
+    def time_git_three_way_merge(self) -> None:
+        """Time git three-way merge of non-overlapping changes."""
+        service = SubplanMergeService(SubplanMergeStrategy.GIT_THREE_WAY)
+        service.merge(self.base_files, self.outputs)
+
+    def time_sequential_apply_merge(self) -> None:
+        """Time sequential apply merge."""
+        service = SubplanMergeService(SubplanMergeStrategy.SEQUENTIAL_APPLY)
+        service.merge(self.base_files, self.overlapping_outputs)
+
+    def time_last_wins_merge(self) -> None:
+        """Time last-wins merge."""
+        service = SubplanMergeService(SubplanMergeStrategy.LAST_WINS)
+        service.merge(self.base_files, self.overlapping_outputs)
+
+    def time_merge_service_construction(self) -> None:
+        """Time constructing SubplanMergeService."""
+        SubplanMergeService(SubplanMergeStrategy.GIT_THREE_WAY)
+
+    def time_single_file_merge(self) -> None:
+        """Time merging a single file output."""
+        service = SubplanMergeService(SubplanMergeStrategy.LAST_WINS)
+        service.merge(
+            {"src/main.py": "base\n"},
+            [(_S1, {"src/main.py": "modified\n"})],
+        )
+
+
+class SubplanIntegrationSuite:
+    """Benchmark integrated execution+merge overhead."""
+
+    def setup(self) -> None:
+        """Prepare fixtures for integration benchmarks."""
+        self.statuses = [_make_status(_S1), _make_status(_S2)]
+
+    def time_sequential_with_last_wins(self) -> None:
+        """Time sequential execution with last-wins merge."""
+        config = SubplanConfig(
+            execution_mode=ExecutionMode.SEQUENTIAL,
+            merge_strategy=SubplanMergeStrategy.LAST_WINS,
+        )
+        service = SubplanExecutionService(config=config, executor_fn=_noop_executor)
+        service.execute_all(subplan_statuses=self.statuses, base_files={})
+
+    def time_parallel_with_git_merge(self) -> None:
+        """Time parallel execution with git three-way merge."""
+        config = SubplanConfig(
+            execution_mode=ExecutionMode.PARALLEL,
+            max_parallel=2,
+            merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY,
+        )
+        service = SubplanExecutionService(config=config, executor_fn=_noop_executor)
+        service.execute_all(subplan_statuses=self.statuses, base_files={})

docs/reference/subplans.md

@@ -0,0 +1,207 @@
+# Subplans: Execution and Merge
+
+Subplans allow a parent plan to decompose work into coordinated child plans.
+Each child plan runs in its own sandbox, and results are merged back into the
+parent plan using a configurable merge strategy.
+
+## Execution Modes
+
+The `SubplanConfig.execution_mode` field controls how child plans are
+scheduled:
+
+| Mode                  | Description                                      |
+|-----------------------|--------------------------------------------------|
+| `sequential`          | Execute one at a time in order                   |
+| `parallel`            | Execute concurrently (up to `max_parallel`)      |
+| `dependency_ordered`  | Respect DAG dependencies via topological sort    |
+
+### Sequential Mode
+
+Subplans execute one at a time in the order they appear in
+`subplan_statuses`. If a subplan fails and `fail_fast` is enabled,
+remaining subplans are cancelled.
+
+### Parallel Mode
+
+Subplans execute concurrently using a thread pool. The `max_parallel`
+setting (default: 5, range: 1–50) caps the number of concurrent workers.
+When `fail_fast` is enabled, remaining subplans are cancelled on first
+failure.  Cancelled futures are reported with `CANCELLED` status (not
+`ERRORED`).  Results are returned in completion order, which affects
+ordering-sensitive merge strategies such as `sequential_apply`.
+
+### Dependency-Ordered Mode
+
+Subplans are sorted topologically based on a dependency graph.
+Independent subplans (those whose dependencies are all satisfied)
+within the same topological wave are executed concurrently, respecting
+the `max_parallel` limit.  Subsequent waves wait for the previous wave
+to finish.  A circular dependency raises a `ValueError`.  The
+dependency graph maps each `subplan_id` to the list of `subplan_id`s
+it depends on.
+
+## Merge Strategies
+
+After subplans complete, their sandbox outputs are merged using the
+configured `SubplanMergeStrategy`:
+
+| Strategy              | Description                                      |
+|-----------------------|--------------------------------------------------|
+| `git_three_way`       | Three-way merge via `git merge-file`             |
+| `sequential_apply`    | Apply changes in completion order                |
+| `fail_on_conflict`    | Raise `MergeConflictError` on any conflict       |
+| `last_wins`           | Final subplan's output overwrites earlier ones   |
+
+### Git Three-Way Merge
+
+Uses `git merge-file` to perform content-level merging. Non-overlapping
+changes from different subplans are combined automatically. Overlapping
+changes produce conflict markers in the output.
+
+### Sequential Apply
+
+Each subplan's output is applied in the order subplans completed, with
+each becoming the new base for the next.  In sequential mode, this
+matches the input order.  In parallel mode, outputs are ordered by
+actual completion time, so the subplan that finishes last has its
+changes applied last.  Equivalent to replaying edits sequentially.
+
+### Fail on Conflict
+
+Same as git three-way merge, but raises `MergeConflictError` if any
+file has unresolved conflicts. Use this when conflicts must be
+resolved manually.
+
+### Last Wins
+
+The final subplan's version of each file overwrites all earlier
+versions unconditionally. Simple but may discard concurrent work.
+
+## Merge Outcomes
+
+The `SubplanMergeResult` reports:
+
+| Field             | Description                                          |
+|-------------------|------------------------------------------------------|
+| `success`         | `True` if all files merged without conflicts         |
+| `merged_files`    | Per-file outcomes with content and conflict status   |
+| `conflict_files`  | Paths of files with unresolved conflicts             |
+| `total_files`     | Total unique files across all subplan outputs        |
+
+Each `FileMergeOutcome` includes:
+
+| Field                | Description                                       |
+|----------------------|---------------------------------------------------|
+| `path`               | Relative file path                                |
+| `content`            | Merged content (may contain conflict markers)     |
+| `has_conflict`       | Whether this file has unresolved conflicts        |
+| `source_subplan_ids` | IDs of subplans that modified this file           |
+
+## Timeout Enforcement
+
+When `timeout_per_subplan_seconds` is set on `SubplanConfig`, each
+subplan execution is subject to a wall-clock timeout.  If a subplan
+(including any retries) does not complete within the configured number
+of seconds, it is marked `ERRORED` with a `SubplanTimeoutError`
+message.  Timeouts are enforced in all execution modes (sequential,
+parallel, and dependency-ordered).
+
+The timeout applies per-subplan, not per-attempt.  The entire retry
+loop for a subplan must complete within the deadline.  If the timeout
+expires during a retry, no further retries are attempted and the
+subplan is marked as errored immediately.
+
+## Failure Handling
+
+The `SubplanFailureHandler` determines retry and stop behavior:
+
+- **`fail_fast`**: Stop all subplans on first failure (any mode).
+- **Sequential mode**: Always stops on failure (even without `fail_fast`).
+- **Parallel mode**: Other subplans continue unless `fail_fast` is set.
+- **Retry**: Retriable errors (`TimeoutError`, `ValidationError`,
+  `TemporaryResourceError`, `MergeConflictError`) are retried up to
+  `max_retries` times.
+- **Non-retriable**: `ConfigurationError`, `AuthenticationError`,
+  `MissingResourceError`, `CircularDependencyError` are never retried.
+
+## Configuration Reference
+
+```yaml
+subplan_config:
+  execution_mode: parallel        # sequential | parallel | dependency_ordered
+  merge_strategy: git_three_way   # git_three_way | sequential_apply | fail_on_conflict | last_wins
+  max_parallel: 5                 # 1-50, for parallel mode
+  fail_fast: false                # stop all on first failure
+  timeout_per_subplan_seconds: ~  # optional per-subplan timeout
+  retry_failed: true              # auto-retry failed subplans
+  max_retries: 2                  # 0-5, max retry attempts
+```
+
+## Service API
+
+### SubplanExecutionService
+
+```python
+from cleveragents.application.services import (
+    SubplanExecutionService,
+    SubplanExecutionOutput,
+)
+
+service = SubplanExecutionService(
+    config=plan.subplan_config,
+    executor_fn=my_executor,  # Callable[[SubplanStatus], SubplanExecutionOutput]
+)
+
+result = service.execute_all(
+    subplan_statuses=plan.subplan_statuses,
+    base_files={"src/main.py": original_content},
+    dependency_graph={"sub-2": ["sub-1"]},  # for dependency_ordered mode
+)
+```
+
+### SubplanMergeService
+
+```python
+from cleveragents.application.services import SubplanMergeService
+
+merge_service = SubplanMergeService(strategy=SubplanMergeStrategy.GIT_THREE_WAY)
+
+result = merge_service.merge(
+    base_files={"src/main.py": "original content"},
+    subplan_outputs=[
+        ("sub-1", {"src/main.py": "modified by sub-1"}),
+        ("sub-2", {"src/main.py": "modified by sub-2"}),
+    ],
+)
+```
+
+## Design Decisions and Specification Alignment
+
+This section documents intentional deviations from the specification
+(`docs/specification.md`) and the rationale behind them.
+
+### Execution mode via configuration vs. decision tree
+
+The specification describes execution mode as emerging from the decision
+tree (`subplan_spawn` vs `subplan_parallel_spawn` decision types). This
+implementation uses a configurable `ExecutionMode` enum on `SubplanConfig`
+instead, which simplifies the service interface and allows the execution
+mode to be set declaratively without coupling to decision-tree internals.
+
+### Single merge strategy per plan vs. per-resource-type
+
+The specification describes merge strategies as resource-type-dependent
+(e.g., git three-way for git resources, sequential for databases). This
+implementation uses a single `SubplanMergeStrategy` per `SubplanConfig`,
+applied uniformly to all files. A per-resource-type merge dispatch can be
+layered on top by callers that select the strategy based on the resource
+type before constructing the service.
+
+### Global concurrency and depth limits
+
+The specification defines `plan.concurrency` (default 4) and
+`plan.max-child-depth` (default 5) as global limits. These are not
+enforced within this service because they are cross-cutting concerns that
+belong at the orchestration layer above `SubplanExecutionService`. The
+`max_parallel` setting on `SubplanConfig` controls per-plan concurrency
+locally.

features/subplan_execution.feature

@@ -0,0 +1,309 @@
+@phase1 @subplan @execution
+Feature: Subplan Execution and Merge
+  As a system orchestrating parallel workloads
+  I want to execute subplans in sequential, parallel, and dependency-ordered modes
+  And merge their sandbox outputs using configurable strategies
+  So that parent plans can decompose work into coordinated child plans
+
+  # --- Sequential execution ---
+
+  @sequential
+  Scenario: Sequential execution completes all subplans in order
+    Given a parent plan with 3 subplans in sequential mode
+    When the subplans are executed
+    Then all 3 subplans should complete successfully
+    And the subplans should have been executed in order
+    And the execution result should report all succeeded
+
+  @sequential
+  Scenario: Sequential execution stops on failure when fail_fast is enabled
+    Given a parent plan with 3 subplans in sequential mode with fail_fast
+    And the second subplan will fail with "ValidationError: bad input"
+    When the subplans are executed
+    Then the first subplan should complete successfully
+    And the second subplan should be errored
+    And the third subplan should be cancelled
+
+  @sequential
+  Scenario: Sequential execution retries retriable failures
+    Given a parent plan with 1 subplan in sequential mode with retry enabled
+    And the subplan will fail once with "TimeoutError: timed out" then succeed
+    When the subplans are executed
+    Then the subplan should complete successfully
+    And the subplan should have 1 previous attempt recorded
+
+  # --- Parallel execution ---
+
+  @parallel
+  Scenario: Parallel execution runs subplans concurrently
+    Given a parent plan with 3 subplans in parallel mode with max_parallel 3
+    When the subplans are executed
+    Then all 3 subplans should complete successfully
+    And the execution result should report all succeeded
+
+  @parallel
+  Scenario: Parallel execution respects max_parallel limit
+    Given a parent plan with 5 subplans in parallel mode with max_parallel 2
+    When the subplans are executed
+    Then all 5 subplans should complete successfully
+
+  @parallel
+  Scenario: Parallel execution with fail_fast cancels remaining on failure
+    Given a parent plan with 3 subplans in parallel mode with fail_fast
+    And the first subplan will fail with "ValidationError: schema mismatch"
+    When the subplans are executed
+    Then at least one subplan should be errored
+    And the execution result should have failed subplan ids
+
+  # --- Dependency-ordered execution ---
+
+  @dependency_ordered
+  Scenario: Dependency-ordered execution respects DAG ordering
+    Given a parent plan with 3 subplans in dependency_ordered mode
+    And subplan C depends on subplan B which depends on subplan A
+    When the subplans are executed
+    Then all 3 subplans should complete successfully
+    And subplan A should have completed before subplan B
+    And subplan B should have completed before subplan C
+
+  @dependency_ordered
+  Scenario: Dependency-ordered execution rejects circular dependencies
+    Given a parent plan with 2 subplans in dependency_ordered mode
+    And the subplans have circular dependencies
+    When the subplans are executed expecting an error
+    Then a circular dependency error should be raised
+
+  @dependency_ordered
+  Scenario: Dependency-ordered mode requires a dependency graph
+    Given a parent plan with 2 subplans in dependency_ordered mode
+    When the subplans are executed without a dependency graph
+    Then a missing dependency graph error should be raised
+
+  # --- Merge strategies ---
+
+  @merge @git_three_way
+  Scenario: Git three-way merge combines non-overlapping changes
+    Given subplan outputs with non-overlapping changes to the same file
+    And the merge strategy is git_three_way
+    When the subplan outputs are merged
+    Then the merge should succeed
+    And the merged file should contain changes from both subplans
+
+  @merge @sequential_apply
+  Scenario: Sequential apply merges changes in order
+    Given subplan outputs with overlapping changes to the same file
+    And the merge strategy is sequential_apply
+    When the subplan outputs are merged
+    Then the merge should succeed
+    And the final content should reflect sequential application
+
+  @merge @fail_on_conflict
+  Scenario: Fail-on-conflict raises error when conflicts exist
+    Given subplan outputs with conflicting changes to the same file
+    And the merge strategy is fail_on_conflict
+    When the subplan outputs are merged expecting a conflict error
+    Then a merge conflict error should be raised
+
+  @merge @last_wins
+  Scenario: Last-wins takes the final subplan output
+    Given subplan outputs with overlapping changes to the same file
+    And the merge strategy is last_wins
+    When the subplan outputs are merged
+    Then the merge should succeed
+    And the merged content should be from the last subplan
+
+  @merge
+  Scenario: Merge with single subplan output returns unchanged content
+    Given a single subplan output with file changes
+    And the merge strategy is git_three_way
+    When the subplan outputs are merged
+    Then the merge should succeed
+    And the merged content should match the single subplan output
+
+  @merge
+  Scenario: Merge with empty subplan outputs raises error
+    Given no subplan outputs
+    And the merge strategy is git_three_way
+    When the subplan outputs are merged expecting a validation error
+    Then an empty outputs error should be raised
+
+  # --- Integration: execution + merge ---
+
+  @integration
+  Scenario: Sequential execution with git merge produces combined result
+    Given a parent plan with 2 subplans in sequential mode with git merge
+    And both subplans produce non-overlapping file changes
+    When the subplans are executed
+    Then the execution result should include a merge result
+    And the merge result should have no conflicts
+
+  @integration
+  Scenario: Parallel execution with last-wins merge resolves overlaps
+    Given a parent plan with 2 subplans in parallel mode with last_wins merge
+    And both subplans modify the same file differently
+    When the subplans are executed
+    Then the execution result should include a merge result
+    And the merge result should succeed
+
+  # --- Service validation ---
+
+  @validation
+  Scenario: SubplanExecutionService rejects None config
+    When a SubplanExecutionService is created with None config
+    Then a config validation error should be raised
+
+  @validation
+  Scenario: SubplanExecutionService rejects None executor
+    When a SubplanExecutionService is created with None executor
+    Then an executor validation error should be raised
+
+  @validation
+  Scenario: SubplanMergeService rejects None strategy
+    When a SubplanMergeService is created with None strategy
+    Then a strategy validation error should be raised
+
+  @validation
+  Scenario: SubplanExecutionService rejects empty subplan list
+    Given a valid SubplanExecutionService
+    When execute_all is called with empty subplan statuses
+    Then an empty statuses error should be raised
+
+  # --- Property accessors ---
+
+  @accessor
+  Scenario: SubplanExecutionService exposes config and merge_service properties
+    Given a valid SubplanExecutionService
+    Then the service config property should return the configured config
+    And the service merge_service property should return a merge service
+
+  @accessor
+  Scenario: SubplanMergeService exposes strategy property
+    Given a SubplanMergeService with git_three_way strategy
+    Then the merge strategy property should return git_three_way
+
+  # --- Retry edge cases ---
+
+  @retry
+  Scenario: Non-retriable error is not retried
+    Given a parent plan with 1 subplan in sequential mode with retry enabled
+    And the subplan will fail with "ConfigurationError: invalid config"
+    When the subplans are executed
+    Then the subplan should be errored
+    And the subplan attempt number should be 1
+
+  @retry
+  Scenario: Retry exhaustion marks subplan as permanently failed
+    Given a parent plan with 1 subplan in sequential mode with max_retries 1
+    And the subplan will always fail with "TimeoutError: timed out"
+    When the subplans are executed
+    Then the subplan should be errored
+    And the subplan should have 2 previous attempt recorded
+
+  @retry
+  Scenario: Executor exception triggers retry for retriable errors
+    Given a parent plan with 1 subplan in sequential mode with retry enabled
+    And the executor will raise "TimeoutError" once then succeed
+    When the subplans are executed
+    Then the subplan should complete successfully
+    And the subplan should have 1 previous attempt recorded
+
+  @retry
+  Scenario: Executor exception with non-retriable error is not retried
+    Given a parent plan with 1 subplan in sequential mode with retry enabled
+    And the executor will raise "ConfigurationError" permanently
+    When the subplans are executed
+    Then the subplan should be errored
+    And the subplan should have 1 previous attempt recorded
+
+  # --- Timeout enforcement ---
+
+  @timeout
+  Scenario: Sequential execution enforces per-subplan timeout
+    Given a parent plan with 1 subplan in sequential mode with a 1 second timeout
+    And the subplan executor will block for 3 seconds
+    When the subplans are executed
+    Then the subplan should be errored
+    And the subplan error should mention timeout
+
+  @timeout
+  Scenario: Parallel execution enforces per-subplan timeout
+    Given a parent plan with 2 subplans in parallel mode with a 1 second timeout
+    And the first subplan executor will block for 3 seconds
+    When the subplans are executed
+    Then at least one subplan error should mention timeout
+
+  # --- Completion order for parallel merge ---
+
+  @parallel @completion_order
+  Scenario: Parallel execution returns outputs in completion order
+    Given a parent plan with 3 subplans in parallel mode with staggered completion
+    When the subplans are executed
+    Then the execution result statuses should be in completion order
+
+  # --- Cancelled status in parallel fail_fast ---
+
+  @parallel @cancel_status
+  Scenario: Parallel fail_fast marks unstarted futures as CANCELLED not ERRORED
+    Given a parent plan with 3 subplans in parallel mode with max_parallel 1 and fail_fast
+    And the first subplan will fail with "ValidationError: schema mismatch"
+    When the subplans are executed
+    Then the first subplan should be errored
+    And the remaining subplans should have CANCELLED status
+
+  # --- Dependency-ordered concurrent execution ---
+
+  @dependency_ordered @concurrent
+  Scenario: Dependency-ordered mode runs independent subplans concurrently
+    Given a parent plan with 3 subplans in dependency_ordered mode with concurrency tracking
+    And subplans A and B are independent while C depends on both
+    When the subplans are executed
+    Then all 3 subplans should complete successfully
+    And the peak concurrent execution count should be at least 2
+    And subplan C should have started after A and B completed
+
+  # --- Dependency-ordered fail_fast ---
+
+  @dependency_ordered @fail_fast
+  Scenario: Dependency-ordered fail_fast cancels subsequent waves
+    Given a parent plan with 3 subplans in dependency_ordered mode with fail_fast
+    And subplan C depends on subplan B which depends on subplan A
+    And the first subplan will fail with "ValidationError: bad input"
+    When the subplans are executed
+    Then the first subplan should be errored
+    And the second and third subplans should have CANCELLED status
+
+  # --- Dependency-ordered timeout enforcement ---
+
+  @dependency_ordered @timeout
+  Scenario: Dependency-ordered mode enforces timeout on single-node wave
+    Given a parent plan with 2 subplans in dependency_ordered mode with a 1 second timeout
+    And subplan B depends on subplan A
+    And the first subplan executor will block for 3 seconds
+    When the subplans are executed
+    Then at least one subplan error should mention timeout
+
+  @dependency_ordered @timeout
+  Scenario: Dependency-ordered mode enforces timeout on multi-node wave
+    Given a parent plan with 3 subplans in dependency_ordered mode with concurrency and a 1 second timeout
+    And subplans A and B are independent while C depends on both
+    And the subplan executor will block for 3 seconds
+    When the subplans are executed
+    Then at least one subplan error should mention timeout
+
+  # --- Parallel executor exception bubbling ---
+
+  @parallel @error
+  Scenario: Parallel execution handles executor exception as ERRORED
+    Given a parent plan with 2 subplans in parallel mode with max_parallel 2
+    And the executor will raise "RuntimeError" for the first subplan in parallel
+    When the subplans are executed
+    Then at least one subplan should be errored
+
+  # --- Integration: merge conflict during execution ---
+
+  @integration
+  Scenario: Execution with merge conflict marks result as not all succeeded
+    Given a parent plan with 2 subplans in sequential mode with fail_on_conflict merge
+    And both subplans produce conflicting file changes
+    When the subplans are executed
+    Then the execution result should report not all succeeded

robot/helper_subplan_execution.py

@@ -0,0 +1,343 @@
+"""Helper script for subplan execution Robot Framework smoke tests.
+
+Exercises SubplanExecutionService, SubplanMergeService, and integrated
+execution+merge workflows without requiring the full service layer.
+"""
+
+from __future__ import annotations
+
+import sys
+import threading
+
+from cleveragents.application.services.subplan_execution_service import (
+    SubplanExecutionOutput,
+    SubplanExecutionService,
+)
+from cleveragents.application.services.subplan_merge_service import (
+    MergeConflictError,
+    SubplanMergeService,
+)
+from cleveragents.domain.models.core.plan import (
+    ExecutionMode,
+    ProcessingState,
+    SubplanConfig,
+    SubplanMergeStrategy,
+    SubplanStatus,
+)
+
+_S1 = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
+_S2 = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
+_S3 = "01HGZ6FE0AQDYTR4BXVQZ6EC00"
+
+
+def _make_status(subplan_id: str) -> SubplanStatus:
+    return SubplanStatus(
+        subplan_id=subplan_id,
+        action_name="local/robot-sub-action",
+    )
+
+
+def _success_executor(status: SubplanStatus) -> SubplanExecutionOutput:
+    return SubplanExecutionOutput(
+        subplan_id=status.subplan_id,
+        success=True,
+        files={f"src/{status.subplan_id[-4:]}.py": f"# {status.subplan_id}\n"},
+        files_changed=1,
+        changeset_summary="Robot test output",
+    )
+
+
+def _sequential_all() -> None:
+    """Verify sequential execution completes all subplans."""
+    statuses = [_make_status(_S1), _make_status(_S2), _make_status(_S3)]
+    config = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL)
+    service = SubplanExecutionService(config=config, executor_fn=_success_executor)
+    result = service.execute_all(subplan_statuses=statuses, base_files={})
+    assert result.all_succeeded, f"Not all succeeded: {result.failed_subplan_ids}"
+    completed = [s for s in result.statuses if s.status == ProcessingState.COMPLETE]
+    assert len(completed) == 3, f"Expected 3, got {len(completed)}"
+    print("sequential-all-ok")
+
+
+def _sequential_failfast() -> None:
+    """Verify sequential fail_fast stops after first failure."""
+    fail_set = {_S2}
+
+    def executor(status: SubplanStatus) -> SubplanExecutionOutput:
+        if status.subplan_id in fail_set:
+            return SubplanExecutionOutput(
+                subplan_id=status.subplan_id,
+                success=False,
+                error="ValidationError: bad input",
+            )
+        return _success_executor(status)
+
+    statuses = [_make_status(_S1), _make_status(_S2), _make_status(_S3)]
+    config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+        fail_fast=True,
+    )
+    service = SubplanExecutionService(config=config, executor_fn=executor)
+    result = service.execute_all(subplan_statuses=statuses, base_files={})
+    assert result.statuses[0].status == ProcessingState.COMPLETE
+    assert result.statuses[1].status == ProcessingState.ERRORED
+    assert result.statuses[2].status == ProcessingState.CANCELLED
+    print("sequential-failfast-ok")
+
+
+def _parallel_all() -> None:
+    """Verify parallel execution completes all subplans."""
+    statuses = [_make_status(_S1), _make_status(_S2), _make_status(_S3)]
+    config = SubplanConfig(execution_mode=ExecutionMode.PARALLEL, max_parallel=3)
+    service = SubplanExecutionService(config=config, executor_fn=_success_executor)
+    result = service.execute_all(subplan_statuses=statuses, base_files={})
+    assert result.all_succeeded
+    print("parallel-all-ok")
+
+
+def _dep_ordered() -> None:
+    """Verify dependency-ordered execution follows topological order."""
+    order: list[str] = []
+    lock = threading.Lock()
+
+    def executor(status: SubplanStatus) -> SubplanExecutionOutput:
+        with lock:
+            order.append(status.subplan_id)
+        return _success_executor(status)
+
+    statuses = [_make_status(_S1), _make_status(_S2), _make_status(_S3)]
+    config = SubplanConfig(execution_mode=ExecutionMode.DEPENDENCY_ORDERED)
+    dep_graph = {_S1: [], _S2: [_S1], _S3: [_S2]}
+    service = SubplanExecutionService(config=config, executor_fn=executor)
+    result = service.execute_all(
+        subplan_statuses=statuses,
+        base_files={},
+        dependency_graph=dep_graph,
+    )
+    assert result.all_succeeded
+    # Verify order: first occurrences should be A, B, C
+    seen: list[str] = []
+    for sid in order:
+        if sid not in seen:
+            seen.append(sid)
+    assert seen == [_S1, _S2, _S3], f"Expected [A,B,C], got {seen}"
+    print("dep-ordered-ok")
+
+
+def _merge_git_clean() -> None:
+    """Verify git three-way merge with non-overlapping changes."""
+    service = SubplanMergeService(SubplanMergeStrategy.GIT_THREE_WAY)
+    base = {"src/main.py": "line1\nline2\nline3\n"}
+    outputs = [
+        (_S1, {"src/main.py": "line1\nline2\nline3\nnew_a\n"}),
+        (_S2, {"src/main.py": "new_b\nline1\nline2\nline3\n"}),
+    ]
+    result = service.merge(base, outputs)
+    assert result.success, f"Merge failed: {result.conflict_files}"
+    print("merge-git-clean-ok")
+
+
+def _merge_sequential() -> None:
+    """Verify sequential apply merge."""
+    service = SubplanMergeService(SubplanMergeStrategy.SEQUENTIAL_APPLY)
+    base = {"src/main.py": "original\n"}
+    outputs = [
+        (_S1, {"src/main.py": "first\n"}),
+        (_S2, {"src/main.py": "second\n"}),
+    ]
+    result = service.merge(base, outputs)
+    assert result.success
+    assert "second" in result.merged_files[0].content
+    print("merge-sequential-ok")
+
+
+def _merge_last_wins() -> None:
+    """Verify last-wins merge takes final output."""
+    service = SubplanMergeService(SubplanMergeStrategy.LAST_WINS)
+    base = {"src/main.py": "original\n"}
+    outputs = [
+        (_S1, {"src/main.py": "first\n"}),
+        (_S2, {"src/main.py": "second\n"}),
+    ]
+    result = service.merge(base, outputs)
+    assert result.success
+    assert result.merged_files[0].content == "second\n"
+    print("merge-last-wins-ok")
+
+
+def _merge_fail_conflict() -> None:
+    """Verify fail-on-conflict raises MergeConflictError."""
+    service = SubplanMergeService(SubplanMergeStrategy.FAIL_ON_CONFLICT)
+    base = {"src/main.py": "base line\n"}
+    outputs = [
+        (_S1, {"src/main.py": "alpha\n"}),
+        (_S2, {"src/main.py": "beta\n"}),
+    ]
+    try:
+        service.merge(base, outputs)
+        raise AssertionError("Expected MergeConflictError")
+    except MergeConflictError:
+        pass
+    print("merge-fail-conflict-ok")
+
+
+def _exec_merge_integration() -> None:
+    """Verify end-to-end execution + merge integration."""
+    statuses = [_make_status(_S1), _make_status(_S2)]
+    config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+        merge_strategy=SubplanMergeStrategy.LAST_WINS,
+    )
+    service = SubplanExecutionService(config=config, executor_fn=_success_executor)
+    result = service.execute_all(subplan_statuses=statuses, base_files={})
+    assert result.all_succeeded
+    assert result.merge_result is not None
+    assert result.merge_result.success
+    print("exec-merge-integration-ok")
+
+
+def _validation_guards() -> None:
+    """Verify service validation guards."""
+    # None config
+    try:
+        SubplanExecutionService(
+            config=None,  # type: ignore[arg-type]
+            executor_fn=_success_executor,
+        )
+        raise AssertionError("Expected ValueError for None config")
+    except ValueError:
+        pass
+
+    # None executor
+    try:
+        SubplanExecutionService(
+            config=SubplanConfig(),
+            executor_fn=None,  # type: ignore[arg-type]
+        )
+        raise AssertionError("Expected ValueError for None executor")
+    except ValueError:
+        pass
+
+    # None strategy
+    try:
+        SubplanMergeService(strategy=None)  # type: ignore[arg-type]
+        raise AssertionError("Expected ValueError for None strategy")
+    except ValueError:
+        pass
+
+    # Empty statuses
+    svc = SubplanExecutionService(config=SubplanConfig(), executor_fn=_success_executor)
+    try:
+        svc.execute_all(subplan_statuses=[], base_files={})
+        raise AssertionError("Expected ValueError for empty statuses")
+    except ValueError:
+        pass
+
+    print("validation-guards-ok")
+
+
+def _exec_summary() -> None:
+    """Verify execution result summary report."""
+    statuses = [_make_status(_S1), _make_status(_S2)]
+    config = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL)
+    service = SubplanExecutionService(config=config, executor_fn=_success_executor)
+    result = service.execute_all(subplan_statuses=statuses, base_files={})
+    assert result.all_succeeded
+    assert result.total_duration_ms >= 0
+    assert len(result.statuses) == 2
+    assert len(result.failed_subplan_ids) == 0
+    print("exec-summary-ok")
+
+
+def _retry_non_retriable() -> None:
+    """Verify non-retriable error is not retried."""
+
+    def executor(status: SubplanStatus) -> SubplanExecutionOutput:
+        return SubplanExecutionOutput(
+            subplan_id=status.subplan_id,
+            success=False,
+            error="ConfigurationError: invalid config",
+        )
+
+    config = SubplanConfig(retry_failed=True, max_retries=2)
+    service = SubplanExecutionService(config=config, executor_fn=executor)
+    result = service.execute_all(subplan_statuses=[_make_status(_S1)], base_files={})
+    assert not result.all_succeeded
+    assert result.statuses[0].attempt_number == 1
+    print("retry-non-retriable-ok")
+
+
+def _retry_exception_then_succeed() -> None:
+    """Verify executor exception triggers retry and then succeeds."""
+    call_count: list[int] = [0]
+
+    def executor(status: SubplanStatus) -> SubplanExecutionOutput:
+        call_count[0] += 1
+        if call_count[0] <= 1:
+            raise RuntimeError("TimeoutError: simulated exception")
+        return SubplanExecutionOutput(
+            subplan_id=status.subplan_id,
+            success=True,
+            files={"src/main.py": "# recovered\n"},
+            files_changed=1,
+        )
+
+    config = SubplanConfig(retry_failed=True, max_retries=2)
+    service = SubplanExecutionService(config=config, executor_fn=executor)
+    result = service.execute_all(subplan_statuses=[_make_status(_S1)], base_files={})
+    assert result.all_succeeded
+    assert result.statuses[0].attempt_number == 2
+    print("retry-exception-ok")
+
+
+def _merge_conflict_exec() -> None:
+    """Verify merge conflict during execution marks not all succeeded."""
+
+    def executor(status: SubplanStatus) -> SubplanExecutionOutput:
+        return SubplanExecutionOutput(
+            subplan_id=status.subplan_id,
+            success=True,
+            files={"src/main.py": f"conflict-{status.subplan_id}\n"},
+            files_changed=1,
+        )
+
+    config = SubplanConfig(merge_strategy=SubplanMergeStrategy.FAIL_ON_CONFLICT)
+    service = SubplanExecutionService(config=config, executor_fn=executor)
+    result = service.execute_all(
+        subplan_statuses=[_make_status(_S1), _make_status(_S2)],
+        base_files={"src/main.py": "base\n"},
+    )
+    assert not result.all_succeeded
+    print("merge-conflict-exec-ok")
+
+
+def main() -> None:
+    """Dispatch command from sys.argv."""
+    if len(sys.argv) < 2:
+        raise SystemExit("Expected command argument")
+    command = sys.argv[1]
+    commands: dict[str, object] = {
+        "sequential-all": _sequential_all,
+        "sequential-failfast": _sequential_failfast,
+        "parallel-all": _parallel_all,
+        "dep-ordered": _dep_ordered,
+        "merge-git-clean": _merge_git_clean,
+        "merge-sequential": _merge_sequential,
+        "merge-last-wins": _merge_last_wins,
+        "merge-fail-conflict": _merge_fail_conflict,
+        "exec-merge-integration": _exec_merge_integration,
+        "validation-guards": _validation_guards,
+        "exec-summary": _exec_summary,
+        "retry-non-retriable": _retry_non_retriable,
+        "retry-exception": _retry_exception_then_succeed,
+        "merge-conflict-exec": _merge_conflict_exec,
+    }
+    if command not in commands:
+        raise SystemExit(f"Unknown command: {command}")
+    func = commands[command]
+    if callable(func):
+        func()
+
+
+if __name__ == "__main__":
+    main()

robot/subplan_execution.robot

@@ -0,0 +1,108 @@
+*** Settings ***
+Documentation     Smoke tests for subplan execution scheduler, merge strategies,
+...               and integrated execution+merge workflows.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    robot/helper_subplan_execution.py
+
+*** Test Cases ***
+Sequential Execution Completes All Subplans
+    [Documentation]    Verify sequential mode executes all subplans in order
+    [Tags]    subplan    execution    sequential
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    sequential-all    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    sequential-all-ok
+
+Sequential Execution Stops On Fail Fast
+    [Documentation]    Verify sequential fail_fast stops after first failure
+    [Tags]    subplan    execution    sequential    failfast
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    sequential-failfast    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    sequential-failfast-ok
+
+Parallel Execution Completes All Subplans
+    [Documentation]    Verify parallel mode executes all subplans
+    [Tags]    subplan    execution    parallel
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    parallel-all    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    parallel-all-ok
+
+Dependency Ordered Execution Respects DAG
+    [Documentation]    Verify dependency-ordered mode follows topological order
+    [Tags]    subplan    execution    dependency
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    dep-ordered    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    dep-ordered-ok
+
+Git Three Way Merge Non Overlapping
+    [Documentation]    Verify git three-way merge combines non-overlapping changes
+    [Tags]    subplan    merge    git
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    merge-git-clean    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    merge-git-clean-ok
+
+Sequential Apply Merge
+    [Documentation]    Verify sequential apply merge strategy
+    [Tags]    subplan    merge    sequential
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    merge-sequential    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    merge-sequential-ok
+
+Last Wins Merge Strategy
+    [Documentation]    Verify last-wins merge strategy takes final output
+    [Tags]    subplan    merge    lastwins
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    merge-last-wins    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    merge-last-wins-ok
+
+Fail On Conflict Merge Raises Error
+    [Documentation]    Verify fail-on-conflict strategy raises error on conflicts
+    [Tags]    subplan    merge    conflict
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    merge-fail-conflict    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    merge-fail-conflict-ok
+
+Execution With Merge Integration
+    [Documentation]    Verify end-to-end execution and merge integration
+    [Tags]    subplan    execution    merge    integration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    exec-merge-integration    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    exec-merge-integration-ok
+
+Service Validation Guards
+    [Documentation]    Verify service rejects invalid inputs
+    [Tags]    subplan    validation
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    validation-guards    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    validation-guards-ok
+
+Execution Summary Report
+    [Documentation]    Verify execution result reports correct summary
+    [Tags]    subplan    execution    summary
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    exec-summary    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    exec-summary-ok
+
+Non Retriable Error Is Not Retried
+    [Documentation]    Verify non-retriable errors are not retried
+    [Tags]    subplan    execution    retry
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    retry-non-retriable    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    retry-non-retriable-ok
+
+Executor Exception Triggers Retry Then Succeeds
+    [Documentation]    Verify executor exception triggers retry for retriable errors
+    [Tags]    subplan    execution    retry
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    retry-exception    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    retry-exception-ok
+
+Merge Conflict During Execution
+    [Documentation]    Verify merge conflict during execution marks result as failed
+    [Tags]    subplan    execution    merge    conflict
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    merge-conflict-exec    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    merge-conflict-exec-ok

src/cleveragents/application/services/__init__.py

@@ -60,6 +60,17 @@ from cleveragents.application.services.session_service import (
 from cleveragents.application.services.skill_registry_service import (
     SkillRegistryService,
 )
+from cleveragents.application.services.subplan_execution_service import (
+    SubplanExecutionOutput,
+    SubplanExecutionResult,
+    SubplanExecutionService,
+)
+from cleveragents.application.services.subplan_merge_service import (
+    FileMergeOutcome,
+    MergeConflictError,
+    SubplanMergeResult,
+    SubplanMergeService,
+)
 from cleveragents.application.services.tool_registry_service import (
     ToolRegistryService,
 )
@@ -95,7 +106,9 @@ __all__ = [
     "DefaultValidationRunner",
     "DependencyCycleRule",
     "DuplicateImportRule",
+    "FileMergeOutcome",
     "InvariantService",
+    "MergeConflictError",
     "MissingImportRule",
     "MissingSymbolRule",
     "NormalisedOutputDict",
@@ -113,6 +126,11 @@ __all__ = [
     "SemanticValidationService",
     "SemanticValidationSeverity",
     "SkillRegistryService",
+    "SubplanExecutionOutput",
+    "SubplanExecutionResult",
+    "SubplanExecutionService",
+    "SubplanMergeResult",
+    "SubplanMergeService",
     "SyntaxCheckRule",
     "ToolRegistryService",
     "ValidationAttachment",

src/cleveragents/application/services/subplan_execution_service.py

@@ -0,0 +1,665 @@
+"""Subplan execution scheduler service.
+
+Orchestrates the execution of child subplans within a parent plan, supporting
+three scheduling modes (sequential, parallel, dependency-ordered) and four
+merge strategies for combining subplan outputs.
+
+The service delegates individual subplan execution to a caller-provided
+executor callable and uses :class:`SubplanMergeService` for combining
+sandbox outputs.  Failure handling follows :class:`SubplanFailureHandler`
+semantics (retry, stop-others).
+
+Design decisions:
+    - Dependency injection: executor and merge service are injected.
+    - Thread-based parallelism: uses ``concurrent.futures.ThreadPoolExecutor``
+      for PARALLEL mode (consistent with the synchronous codebase).
+    - Stateless: all state is carried in the ``SubplanConfig`` and
+      ``SubplanStatus`` objects on the parent ``Plan``.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections import deque
+from collections.abc import Callable
+from concurrent.futures import CancelledError, Future, ThreadPoolExecutor, as_completed
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+from cleveragents.application.services.subplan_merge_service import (
+    MergeConflictError,
+    SubplanMergeResult,
+    SubplanMergeService,
+)
+from cleveragents.domain.models.core.plan import (
+    ExecutionMode,
+    ProcessingState,
+    SubplanAttempt,
+    SubplanConfig,
+    SubplanFailureHandler,
+    SubplanStatus,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Value objects
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class SubplanExecutionOutput:
+    """Output produced by executing a single subplan.
+
+    Attributes:
+        subplan_id: The ULID of the executed subplan.
+        success: Whether execution succeeded.
+        files: Mapping of relative file paths to their content.
+        error: Error message if execution failed.
+        files_changed: Number of files modified.
+        changeset_summary: Brief description of changes.
+    """
+
+    subplan_id: str
+    success: bool
+    files: dict[str, str] = field(default_factory=dict)
+    error: str | None = None
+    files_changed: int = 0
+    changeset_summary: str | None = None
+
+
+# ---------------------------------------------------------------------------
+# Types
+# ---------------------------------------------------------------------------
+
+SubplanExecutorFn = Callable[[SubplanStatus], SubplanExecutionOutput]
+"""Callable that executes a single subplan and returns its output."""
+
+
+@dataclass(frozen=True)
+class SubplanExecutionResult:
+    """Aggregate result of executing all subplans for a parent plan.
+
+    Attributes:
+        all_succeeded: ``True`` if every subplan completed successfully.
+        statuses: Updated status objects for each subplan.
+        merge_result: Result of merging subplan outputs (None if merge
+            was not attempted, e.g. all subplans failed).
+        total_duration_ms: Wall-clock time for the entire execution in ms.
+        failed_subplan_ids: IDs of subplans that failed.
+    """
+
+    all_succeeded: bool
+    statuses: list[SubplanStatus]
+    merge_result: SubplanMergeResult | None = None
+    total_duration_ms: int = 0
+    failed_subplan_ids: list[str] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# Service
+# ---------------------------------------------------------------------------
+
+
+class SubplanExecutionService:
+    """Execute and merge subplans according to a parent plan's configuration.
+
+    Args:
+        config: The parent plan's subplan configuration.
+        executor_fn: Callable that executes a single subplan.
+        merge_service: Service for merging subplan outputs.  If ``None``,
+            a default merge service is created from the config's strategy.
+        failure_handler: Handler for failure decisions.  If ``None``,
+            a default handler is created.
+
+    Raises:
+        ValueError: If *config* or *executor_fn* is ``None``.
+    """
+
+    def __init__(
+        self,
+        config: SubplanConfig,
+        executor_fn: SubplanExecutorFn,
+        merge_service: SubplanMergeService | None = None,
+        failure_handler: SubplanFailureHandler | None = None,
+    ) -> None:
+        if config is None:
+            raise ValueError("config must not be None")
+        if executor_fn is None:
+            raise ValueError("executor_fn must not be None")
+
+        self._config: SubplanConfig = config
+        self._executor_fn: SubplanExecutorFn = executor_fn
+        self._merge_service: SubplanMergeService = merge_service or SubplanMergeService(
+            config.merge_strategy
+        )
+        self._failure_handler: SubplanFailureHandler = (
+            failure_handler or SubplanFailureHandler()
+        )
+
+    @property
+    def config(self) -> SubplanConfig:
+        """The subplan execution configuration."""
+        return self._config
+
+    @property
+    def merge_service(self) -> SubplanMergeService:
+        """The merge service used for combining outputs."""
+        return self._merge_service
+
+    def execute_all(
+        self,
+        subplan_statuses: list[SubplanStatus],
+        base_files: dict[str, str],
+        dependency_graph: dict[str, list[str]] | None = None,
+    ) -> SubplanExecutionResult:
+        """Execute all subplans and merge their outputs.
+
+        Args:
+            subplan_statuses: Status objects for each subplan to execute.
+            base_files: File contents before subplans (for merge base).
+            dependency_graph: For DEPENDENCY_ORDERED mode, maps each
+                subplan_id to the list of subplan_ids it depends on.
+
+        Returns:
+            A :class:`SubplanExecutionResult` with updated statuses
+            and merge outcome.
+
+        Raises:
+            ValueError: If *subplan_statuses* is empty.
+            ValueError: If DEPENDENCY_ORDERED mode but no dependency_graph.
+        """
+        if not subplan_statuses:
+            raise ValueError("subplan_statuses must not be empty")
+
+        if (
+            self._config.execution_mode == ExecutionMode.DEPENDENCY_ORDERED
+            and dependency_graph is None
+        ):
+            raise ValueError("dependency_graph is required for DEPENDENCY_ORDERED mode")
+
+        start = time.monotonic()
+
+        mode = self._config.execution_mode
+        if mode == ExecutionMode.SEQUENTIAL:
+            statuses, outputs = self._execute_sequential(subplan_statuses)
+        elif mode == ExecutionMode.PARALLEL:
+            statuses, outputs = self._execute_parallel(subplan_statuses)
+        else:
+            statuses, outputs = self._execute_dependency_ordered(
+                subplan_statuses, dependency_graph or {}
+            )
+
+        duration_ms = int((time.monotonic() - start) * 1000)
+
+        failed_ids = [
+            s.subplan_id for s in statuses if s.status == ProcessingState.ERRORED
+        ]
+        all_succeeded = len(failed_ids) == 0
+
+        # Merge outputs if any subplans produced results
+        merge_result: SubplanMergeResult | None = None
+        successful_outputs = [(sid, files) for sid, files in outputs if files]
+        if successful_outputs:
+            try:
+                merge_result = self._merge_service.merge(base_files, successful_outputs)
+            except MergeConflictError:
+                logger.warning("Merge conflict after subplan execution")
+                all_succeeded = False
+
+        return SubplanExecutionResult(
+            all_succeeded=all_succeeded,
+            statuses=statuses,
+            merge_result=merge_result,
+            total_duration_ms=duration_ms,
+            failed_subplan_ids=failed_ids,
+        )
+
+    # -- scheduling modes ----------------------------------------------------
+
+    def _execute_sequential(
+        self,
+        statuses: list[SubplanStatus],
+    ) -> tuple[list[SubplanStatus], list[tuple[str, dict[str, str]]]]:
+        """Execute subplans one at a time in order.
+
+        When ``timeout_per_subplan_seconds`` is configured, each subplan
+        is executed with a per-subplan wall-clock timeout.
+        """
+        updated: list[SubplanStatus] = []
+        outputs: list[tuple[str, dict[str, str]]] = []
+        timeout = self._config.timeout_per_subplan_seconds
+
+        for status in statuses:
+            if timeout is not None:
+                result_status, output = self._execute_with_timeout(status, timeout)
+            else:
+                result_status, output = self._execute_one_with_retry(status)
+            updated.append(result_status)
+            outputs.append((result_status.subplan_id, output))
+
+            if (
+                result_status.status == ProcessingState.ERRORED
+                and self._failure_handler.should_stop_others(
+                    self._config, result_status
+                )
+            ):
+                # Cancel remaining subplans
+                for remaining in statuses[len(updated) :]:
+                    cancelled = self._cancel_status(remaining)
+                    updated.append(cancelled)
+                    outputs.append((cancelled.subplan_id, {}))
+                break
+
+        return updated, outputs
+
+    def _execute_parallel(
+        self,
+        statuses: list[SubplanStatus],
+    ) -> tuple[list[SubplanStatus], list[tuple[str, dict[str, str]]]]:
+        """Execute subplans concurrently up to max_parallel.
+
+        Outputs are returned in completion order so that ordering-sensitive
+        merge strategies (e.g. ``SEQUENTIAL_APPLY``) apply changes in the
+        order subplans actually finished.  Futures cancelled via fail-fast
+        are reported with ``CANCELLED`` status, not ``ERRORED``.
+
+        When ``timeout_per_subplan_seconds`` is configured, each subplan
+        is executed with a per-subplan wall-clock timeout.
+        """
+        max_workers = min(self._config.max_parallel, len(statuses))
+        results_map: dict[str, tuple[SubplanStatus, dict[str, str]]] = {}
+        completion_order: list[str] = []
+        stop_flag = False
+        timeout = self._config.timeout_per_subplan_seconds
+
+        with ThreadPoolExecutor(max_workers=max_workers) as pool:
+            future_to_id: dict[Future[tuple[SubplanStatus, dict[str, str]]], str] = {}
+            for status in statuses:
+                if timeout is not None:
+                    future: Future[tuple[SubplanStatus, dict[str, str]]] = pool.submit(
+                        self._execute_with_timeout, status, timeout
+                    )
+                else:
+                    future = pool.submit(self._execute_one_with_retry, status)
+                future_to_id[future] = status.subplan_id
+
+            for future in as_completed(future_to_id):
+                subplan_id = future_to_id[future]
+                try:
+                    result_status, output = future.result()
+                except CancelledError:
+                    result_status = self._cancel_status(
+                        next(s for s in statuses if s.subplan_id == subplan_id)
+                    )
+                    output = {}
+                except Exception as exc:  # pragma: no cover - defensive
+                    result_status = self._error_status(
+                        next(s for s in statuses if s.subplan_id == subplan_id),
+                        str(exc),
+                    )
+                    output = {}
+
+                results_map[subplan_id] = (result_status, output)
+                completion_order.append(subplan_id)
+
+                if (
+                    result_status.status == ProcessingState.ERRORED
+                    and self._failure_handler.should_stop_others(
+                        self._config, result_status
+                    )
+                ):
+                    stop_flag = True
+                    # Cancel remaining futures
+                    for f in future_to_id:
+                        if not f.done():
+                            f.cancel()
+
+        # Build results in completion order (not original input order)
+        updated: list[SubplanStatus] = []
+        outputs: list[tuple[str, dict[str, str]]] = []
+        for subplan_id in completion_order:
+            s, o = results_map[subplan_id]
+            updated.append(s)
+            outputs.append((s.subplan_id, o))
+
+        # Safety net: append any subplans not captured by as_completed
+        if stop_flag:  # pragma: no branch - as_completed always yields all futures
+            for status in statuses:
+                if status.subplan_id not in results_map:  # pragma: no cover - defensive
+                    cancelled = self._cancel_status(status)
+                    updated.append(cancelled)
+                    outputs.append((cancelled.subplan_id, {}))
+
+        return updated, outputs
+
+    def _execute_dependency_ordered(
+        self,
+        statuses: list[SubplanStatus],
+        dependency_graph: dict[str, list[str]],
+    ) -> tuple[list[SubplanStatus], list[tuple[str, dict[str, str]]]]:
+        """Execute subplans respecting dependency order with concurrent waves.
+
+        Independent subplans (those whose dependencies are all satisfied)
+        within the same topological wave are executed concurrently using
+        :meth:`_execute_wave`.  Subsequent waves wait for the previous
+        wave to finish.
+        """
+        node_ids = [s.subplan_id for s in statuses]
+        # Validate no cycles (raises ValueError if cycle detected)
+        self._topological_sort(node_ids, dependency_graph)
+
+        status_map = {s.subplan_id: s for s in statuses}
+        timeout = self._config.timeout_per_subplan_seconds
+
+        # Build dependency tracking
+        node_set = set(node_ids)
+        remaining_deps: dict[str, set[str]] = {}
+        dependents: dict[str, list[str]] = {nid: [] for nid in node_ids}
+
+        for nid in node_ids:
+            deps = dependency_graph.get(nid, [])
+            remaining_deps[nid] = {d for d in deps if d in node_set}
+            for dep in deps:
+                if dep in node_set:
+                    dependents[dep].append(nid)
+
+        updated: list[SubplanStatus] = []
+        outputs: list[tuple[str, dict[str, str]]] = []
+        completed: set[str] = set()
+        stop_flag = False
+
+        while len(completed) < len(node_ids):
+            ready = [
+                nid
+                for nid in node_ids
+                if nid not in completed and len(remaining_deps[nid]) == 0
+            ]
+            if not ready:  # pragma: no cover - defensive
+                break  # Shouldn't happen after _topological_sort validation
+
+            if stop_flag:
+                for nid in ready:
+                    cancelled = self._cancel_status(status_map[nid])
+                    updated.append(cancelled)
+                    outputs.append((cancelled.subplan_id, {}))
+                    completed.add(nid)
+                    for dep_nid in dependents[nid]:
+                        remaining_deps[dep_nid].discard(nid)
+                continue
+
+            wave_results = self._execute_wave(
+                [status_map[nid] for nid in ready], timeout
+            )
+            for result_status, output in wave_results:
+                nid = result_status.subplan_id
+                updated.append(result_status)
+                outputs.append((nid, output))
+                completed.add(nid)
+                for dep_nid in dependents[nid]:
+                    remaining_deps[dep_nid].discard(nid)
+
+                if (
+                    result_status.status == ProcessingState.ERRORED
+                    and self._failure_handler.should_stop_others(
+                        self._config, result_status
+                    )
+                ):
+                    stop_flag = True
+
+        return updated, outputs
+
+    # -- single subplan execution with retry ---------------------------------
+
+    def _execute_one_with_retry(
+        self,
+        status: SubplanStatus,
+    ) -> tuple[SubplanStatus, dict[str, str]]:
+        """Execute a single subplan with retry logic.
+
+        Returns:
+            Tuple of (updated_status, output_files).
+        """
+        current_status = status.model_copy(
+            update={
+                "status": ProcessingState.PROCESSING,
+                "started_at": datetime.now(tz=UTC),
+            }
+        )
+
+        while True:
+            attempt_start = datetime.now(tz=UTC)
+            try:
+                output = self._executor_fn(current_status)
+                completed_status = current_status.model_copy(
+                    update={
+                        "status": (
+                            ProcessingState.COMPLETE
+                            if output.success
+                            else ProcessingState.ERRORED
+                        ),
+                        "completed_at": datetime.now(tz=UTC),
+                        "error": output.error,
+                        "files_changed": output.files_changed,
+                        "changeset_summary": output.changeset_summary,
+                    }
+                )
+
+                if not output.success:
+                    completed_status = self._record_attempt(
+                        completed_status, attempt_start, output.error
+                    )
+                    if self._failure_handler.should_retry(
+                        self._config, completed_status
+                    ):
+                        current_status = completed_status.model_copy(
+                            update={
+                                "status": ProcessingState.PROCESSING,
+                                "attempt_number": (completed_status.attempt_number + 1),
+                                "error": None,
+                                "completed_at": None,
+                            }
+                        )
+                        continue
+                    return completed_status, output.files
+
+                return completed_status, output.files
+
+            except Exception as exc:
+                error_msg = f"{type(exc).__name__}: {exc}"
+                errored_status = current_status.model_copy(
+                    update={
+                        "status": ProcessingState.ERRORED,
+                        "completed_at": datetime.now(tz=UTC),
+                        "error": error_msg,
+                    }
+                )
+                errored_status = self._record_attempt(
+                    errored_status, attempt_start, error_msg
+                )
+
+                if self._failure_handler.should_retry(self._config, errored_status):
+                    current_status = errored_status.model_copy(
+                        update={
+                            "status": ProcessingState.PROCESSING,
+                            "attempt_number": errored_status.attempt_number + 1,
+                            "error": None,
+                            "completed_at": None,
+                        }
+                    )
+                    continue
+                return errored_status, {}
+
+    # -- timeout & wave helpers -----------------------------------------------
+
+    def _execute_with_timeout(
+        self,
+        status: SubplanStatus,
+        timeout_seconds: int,
+    ) -> tuple[SubplanStatus, dict[str, str]]:
+        """Execute a single subplan with a per-subplan wall-clock timeout.
+
+        Submits the retry loop to a dedicated thread and waits up to
+        *timeout_seconds*.  If the deadline expires the subplan is marked
+        ``ERRORED`` with a descriptive timeout message.
+
+        Args:
+            status: The subplan status to execute.
+            timeout_seconds: Maximum wall-clock seconds for this subplan.
+
+        Returns:
+            Tuple of (updated_status, output_files).
+        """
+        pool = ThreadPoolExecutor(max_workers=1)
+        future: Future[tuple[SubplanStatus, dict[str, str]]] = pool.submit(
+            self._execute_one_with_retry, status
+        )
+        try:
+            return future.result(timeout=timeout_seconds)
+        except TimeoutError:
+            future.cancel()
+            error_msg = (
+                f"SubplanTimeoutError: subplan timed out after {timeout_seconds}s"
+            )
+            return self._error_status(status, error_msg), {}
+        finally:
+            pool.shutdown(wait=False, cancel_futures=True)
+
+    def _execute_wave(
+        self,
+        statuses: list[SubplanStatus],
+        timeout: int | None,
+    ) -> list[tuple[SubplanStatus, dict[str, str]]]:
+        """Execute a wave of independent subplans concurrently.
+
+        Used by :meth:`_execute_dependency_ordered` to run all subplans
+        whose dependencies have been satisfied in parallel, respecting
+        the ``max_parallel`` limit.
+
+        Args:
+            statuses: Independent subplans that can execute in parallel.
+            timeout: Optional per-subplan timeout in seconds.
+
+        Returns:
+            List of ``(status, output_files)`` tuples.
+        """
+        if len(statuses) == 1:
+            if timeout is not None:
+                return [self._execute_with_timeout(statuses[0], timeout)]
+            return [self._execute_one_with_retry(statuses[0])]
+
+        max_workers = min(self._config.max_parallel, len(statuses))
+        results: list[tuple[SubplanStatus, dict[str, str]]] = []
+
+        with ThreadPoolExecutor(max_workers=max_workers) as pool:
+            future_map: dict[
+                Future[tuple[SubplanStatus, dict[str, str]]], SubplanStatus
+            ] = {}
+            for status in statuses:
+                if timeout is not None:
+                    future: Future[tuple[SubplanStatus, dict[str, str]]] = pool.submit(
+                        self._execute_with_timeout, status, timeout
+                    )
+                else:
+                    future = pool.submit(self._execute_one_with_retry, status)
+                future_map[future] = status
+
+            for future in as_completed(future_map):
+                original_status = future_map[future]
+                try:
+                    result_status, output = future.result()
+                except CancelledError:  # pragma: no cover - defensive
+                    result_status = self._cancel_status(original_status)
+                    output = {}
+                except Exception as exc:  # pragma: no cover - defensive
+                    result_status = self._error_status(original_status, str(exc))
+                    output = {}
+                results.append((result_status, output))
+
+        return results
+
+    # -- helpers -------------------------------------------------------------
+
+    @staticmethod
+    def _record_attempt(
+        status: SubplanStatus,
+        started_at: datetime,
+        error: str | None,
+    ) -> SubplanStatus:
+        """Record a failed attempt in the status's previous_attempts."""
+        attempt = SubplanAttempt(
+            attempt_number=status.attempt_number,
+            started_at=started_at,
+            completed_at=datetime.now(tz=UTC),
+            error=error,
+            was_retried=True,
+        )
+        new_attempts = [*status.previous_attempts, attempt]
+        return status.model_copy(update={"previous_attempts": new_attempts})
+
+    @staticmethod
+    def _cancel_status(status: SubplanStatus) -> SubplanStatus:
+        """Mark a subplan status as cancelled."""
+        return status.model_copy(
+            update={
+                "status": ProcessingState.CANCELLED,
+                "completed_at": datetime.now(tz=UTC),
+            }
+        )
+
+    @staticmethod
+    def _error_status(status: SubplanStatus, error: str) -> SubplanStatus:
+        """Mark a subplan status as errored."""
+        return status.model_copy(
+            update={
+                "status": ProcessingState.ERRORED,
+                "completed_at": datetime.now(tz=UTC),
+                "error": error,
+            }
+        )
+
+    @staticmethod
+    def _topological_sort(
+        node_ids: list[str],
+        graph: dict[str, list[str]],
+    ) -> list[str]:
+        """Topological sort of subplan IDs based on dependency graph.
+
+        Args:
+            node_ids: All subplan IDs to sort.
+            graph: Maps subplan_id to list of subplan_ids it depends on.
+
+        Returns:
+            Topologically sorted list of subplan IDs.
+
+        Raises:
+            ValueError: If a cycle is detected.
+        """
+        in_degree: dict[str, int] = {nid: 0 for nid in node_ids}
+        adjacency: dict[str, list[str]] = {nid: [] for nid in node_ids}
+        node_set = set(node_ids)
+
+        for nid in node_ids:
+            deps = graph.get(nid, [])
+            for dep in deps:
+                if dep in node_set:
+                    adjacency[dep].append(nid)
+                    in_degree[nid] += 1
+
+        queue: deque[str] = deque(nid for nid in node_ids if in_degree[nid] == 0)
+        result: list[str] = []
+
+        while queue:
+            node = queue.popleft()
+            result.append(node)
+            for neighbor in adjacency[node]:
+                in_degree[neighbor] -= 1
+                if in_degree[neighbor] == 0:
+                    queue.append(neighbor)
+
+        if len(result) != len(node_ids):
+            raise ValueError("Circular dependency detected in subplan dependency graph")
+
+        return result

src/cleveragents/application/services/subplan_merge_service.py

@@ -0,0 +1,269 @@
+"""Merge service for combining subplan sandbox outputs.
+
+Bridges the domain-level ``SubplanMergeStrategy`` enum to the infrastructure
+merge implementations (``GitMergeStrategy``, ``SequentialMergeStrategy``, etc.)
+and adds the ``FAIL_ON_CONFLICT`` and ``LAST_WINS`` strategies that are specific
+to the subplan execution workflow.
+
+Each merge operation works on a *file map* — a ``dict[str, str]`` mapping
+relative file paths to their text content.  The service merges each file
+individually using the configured strategy and returns a combined result.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+
+from cleveragents.domain.models.core.plan import SubplanMergeStrategy
+from cleveragents.infrastructure.sandbox.merge import (
+    GitMergeStrategy,
+    MergeResult,
+    SequentialMergeStrategy,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Value objects
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class FileMergeOutcome:
+    """Result of merging a single file across subplan outputs.
+
+    Attributes:
+        path: Relative file path.
+        content: The merged content (may contain conflict markers).
+        has_conflict: Whether unresolved conflicts exist.
+        source_subplan_ids: IDs of subplans that touched this file.
+    """
+
+    path: str
+    content: str
+    has_conflict: bool = False
+    source_subplan_ids: list[str] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class SubplanMergeResult:
+    """Aggregate result of merging all subplan outputs.
+
+    Attributes:
+        success: ``True`` if every file merged cleanly.
+        merged_files: Per-file merge outcomes.
+        conflict_files: Paths of files with unresolved conflicts.
+        total_files: Total number of unique files across all subplans.
+    """
+
+    success: bool
+    merged_files: list[FileMergeOutcome]
+    conflict_files: list[str] = field(default_factory=list)
+    total_files: int = 0
+
+
+# ---------------------------------------------------------------------------
+# Exceptions
+# ---------------------------------------------------------------------------
+
+
+class MergeConflictError(Exception):
+    """Raised by FAIL_ON_CONFLICT when any file has a conflict."""
+
+    def __init__(self, conflict_paths: list[str]) -> None:
+        self.conflict_paths = conflict_paths
+        paths_str = ", ".join(conflict_paths)
+        super().__init__(f"Merge conflicts in: {paths_str}")
+
+
+# ---------------------------------------------------------------------------
+# Service
+# ---------------------------------------------------------------------------
+
+
+class SubplanMergeService:
+    """Merge subplan sandbox outputs using a configured strategy.
+
+    The service accepts ordered sandbox output maps (one per subplan)
+    and a base content map, then merges them file by file.
+
+    Strategies:
+        - **GIT_THREE_WAY**: Three-way merge via ``git merge-file``.
+        - **SEQUENTIAL_APPLY**: Apply each subplan's changes in order
+          (last writer wins per file).
+        - **FAIL_ON_CONFLICT**: Same as GIT_THREE_WAY but raises
+          :class:`MergeConflictError` if any conflict is detected.
+        - **LAST_WINS**: The final subplan's version overwrites all
+          earlier versions unconditionally.
+
+    Args:
+        strategy: The merge strategy to use.
+
+    Raises:
+        ValueError: If *strategy* is ``None``.
+    """
+
+    def __init__(self, strategy: SubplanMergeStrategy) -> None:
+        if strategy is None:
+            raise ValueError("strategy must not be None")
+        self._strategy: SubplanMergeStrategy = strategy
+        self._git_merge = GitMergeStrategy()
+        self._seq_merge = SequentialMergeStrategy()
+
+    @property
+    def strategy(self) -> SubplanMergeStrategy:
+        """The configured merge strategy."""
+        return self._strategy
+
+    def merge(
+        self,
+        base_files: dict[str, str],
+        subplan_outputs: Sequence[tuple[str, dict[str, str]]],
+    ) -> SubplanMergeResult:
+        """Merge outputs from multiple subplans against a base.
+
+        Args:
+            base_files: File contents before any subplan ran (path -> content).
+            subplan_outputs: Ordered sequence of ``(subplan_id, file_map)``
+                tuples.  Order matters for SEQUENTIAL_APPLY and LAST_WINS.
+
+        Returns:
+            A :class:`SubplanMergeResult` describing the merged state.
+
+        Raises:
+            ValueError: If *subplan_outputs* is empty.
+            MergeConflictError: If strategy is FAIL_ON_CONFLICT and conflicts
+                are detected.
+        """
+        if not subplan_outputs:
+            raise ValueError("subplan_outputs must not be empty")
+
+        # Collect all unique file paths and which subplans touched them
+        all_paths: dict[str, list[str]] = {}
+        for subplan_id, file_map in subplan_outputs:
+            for path in file_map:
+                if path not in all_paths:
+                    all_paths[path] = []
+                all_paths[path].append(subplan_id)
+
+        merged_files: list[FileMergeOutcome] = []
+        conflict_files: list[str] = []
+
+        for path, subplan_ids in sorted(all_paths.items()):
+            base_content = base_files.get(path, "")
+            outcome = self._merge_file(path, base_content, subplan_ids, subplan_outputs)
+            merged_files.append(outcome)
+            if outcome.has_conflict:
+                conflict_files.append(path)
+
+        if self._strategy == SubplanMergeStrategy.FAIL_ON_CONFLICT and conflict_files:
+            raise MergeConflictError(conflict_files)
+
+        return SubplanMergeResult(
+            success=len(conflict_files) == 0,
+            merged_files=merged_files,
+            conflict_files=conflict_files,
+            total_files=len(all_paths),
+        )
+
+    def _merge_file(
+        self,
+        path: str,
+        base_content: str,
+        subplan_ids: list[str],
+        subplan_outputs: Sequence[tuple[str, dict[str, str]]],
+    ) -> FileMergeOutcome:
+        """Merge a single file across contributing subplans.
+
+        Args:
+            path: The file path being merged.
+            base_content: Original content before subplans.
+            subplan_ids: IDs of subplans that modified this file.
+            subplan_outputs: All subplan outputs for content lookup.
+
+        Returns:
+            A :class:`FileMergeOutcome` for this file.
+        """
+        # Collect ordered contents from subplans that touched this file
+        contents: list[str] = []
+        for _subplan_id, file_map in subplan_outputs:
+            if path in file_map:
+                contents.append(file_map[path])
+
+        if not contents:
+            return FileMergeOutcome(
+                path=path,
+                content=base_content,
+                has_conflict=False,
+                source_subplan_ids=subplan_ids,
+            )
+
+        if len(contents) == 1:
+            # Only one subplan touched this file -- no merge needed
+            return FileMergeOutcome(
+                path=path,
+                content=contents[0],
+                has_conflict=False,
+                source_subplan_ids=subplan_ids,
+            )
+
+        # Multiple subplans touched the same file -- strategy-dependent merge
+        if self._strategy == SubplanMergeStrategy.LAST_WINS:
+            return FileMergeOutcome(
+                path=path,
+                content=contents[-1],
+                has_conflict=False,
+                source_subplan_ids=subplan_ids,
+            )
+
+        if self._strategy == SubplanMergeStrategy.SEQUENTIAL_APPLY:
+            return self._sequential_apply(path, base_content, contents, subplan_ids)
+
+        # GIT_THREE_WAY and FAIL_ON_CONFLICT both use three-way merge
+        return self._git_three_way(path, base_content, contents, subplan_ids)
+
+    def _sequential_apply(
+        self,
+        path: str,
+        base_content: str,
+        contents: list[str],
+        subplan_ids: list[str],
+    ) -> FileMergeOutcome:
+        """Apply changes sequentially (each becomes the new base)."""
+        current = base_content
+        for content in contents:
+            result: MergeResult = self._seq_merge.merge(base_content, current, content)
+            current = result.content
+        return FileMergeOutcome(
+            path=path,
+            content=current,
+            has_conflict=False,
+            source_subplan_ids=subplan_ids,
+        )
+
+    def _git_three_way(
+        self,
+        path: str,
+        base_content: str,
+        contents: list[str],
+        subplan_ids: list[str],
+    ) -> FileMergeOutcome:
+        """Iteratively three-way merge each subplan's changes."""
+        current = contents[0]
+        has_conflict = False
+
+        for content in contents[1:]:
+            result: MergeResult = self._git_merge.merge(base_content, current, content)
+            current = result.content
+            if result.has_conflicts:
+                has_conflict = True
+
+        return FileMergeOutcome(
+            path=path,
+            content=current,
+            has_conflict=has_conflict,
+            source_subplan_ids=subplan_ids,
+        )