diff --git a/benchmarks/subplan_execution_bench.py b/benchmarks/subplan_execution_bench.py new file mode 100644 index 000000000..a5764e04e --- /dev/null +++ b/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={}) diff --git a/docs/reference/subplans.md b/docs/reference/subplans.md new file mode 100644 index 000000000..d6d4457b9 --- /dev/null +++ b/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. diff --git a/features/steps/subplan_execution_steps.py b/features/steps/subplan_execution_steps.py new file mode 100644 index 000000000..9fdb134ac --- /dev/null +++ b/features/steps/subplan_execution_steps.py @@ -0,0 +1,1118 @@ +"""Step definitions for subplan execution and merge scenarios.""" + +from __future__ import annotations + +import threading +import time +from collections.abc import Callable + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.subplan_execution_service import ( + SubplanExecutionOutput, + SubplanExecutionResult, + SubplanExecutionService, +) +from cleveragents.application.services.subplan_merge_service import ( + MergeConflictError, + SubplanMergeResult, + SubplanMergeService, +) +from cleveragents.domain.models.core.plan import ( + ExecutionMode, + ProcessingState, + SubplanConfig, + SubplanMergeStrategy, + SubplanStatus, +) + +_S1 = "01HGZ6FE0AQDYTR4BXVQZ6EA00" +_S2 = "01HGZ6FE0AQDYTR4BXVQZ6EB00" +_S3 = "01HGZ6FE0AQDYTR4BXVQZ6EC00" +_S4 = "01HGZ6FE0AQDYTR4BXVQZ6ED00" +_S5 = "01HGZ6FE0AQDYTR4BXVQZ6EE00" + + +def _make_status( + subplan_id: str, + action: str = "local/sub-action", +) -> SubplanStatus: + return SubplanStatus( + subplan_id=subplan_id, + action_name=action, + ) + + +def _success_executor(status: SubplanStatus) -> SubplanExecutionOutput: + return SubplanExecutionOutput( + subplan_id=status.subplan_id, + success=True, + files={"src/main.py": f"# modified by {status.subplan_id}\n"}, + files_changed=1, + changeset_summary="Modified main.py", + ) + + +def _ordered_executor( + order_list: list[str], + lock: threading.Lock, +) -> Callable[[SubplanStatus], SubplanExecutionOutput]: + def _exec(status: SubplanStatus) -> SubplanExecutionOutput: + with lock: + order_list.append(status.subplan_id) + time.sleep(0.01) + return SubplanExecutionOutput( + subplan_id=status.subplan_id, + success=True, + files={"src/main.py": f"# by {status.subplan_id}\n"}, + files_changed=1, + ) + + return _exec + + +# --- Sequential execution --- + + +@given("a parent plan with {n:d} subplans in sequential mode") +def step_parent_sequential(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL) + context.executor_fn = _success_executor + context.fail_map: dict[str, str | None] = {} + context.retry_map: dict[str, int] = {} + context.execution_order: list[str] = [] + context.order_lock = threading.Lock() + + +@given("a parent plan with {n:d} subplans in sequential mode with fail_fast") +def step_parent_sequential_failfast(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.SEQUENTIAL, + fail_fast=True, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given("a parent plan with {n:d} subplan in sequential mode with retry enabled") +def step_parent_sequential_retry(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.SEQUENTIAL, + retry_failed=True, + max_retries=2, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given('the second subplan will fail with "{error}"') +def step_second_fails(context: Context, error: str) -> None: + context.fail_map[_S2] = error + + +@given('the subplan will fail once with "{error}" then succeed') +def step_fail_once_then_succeed(context: Context, error: str) -> None: + context.fail_map[_S1] = error + context.retry_map[_S1] = 1 # fail once, succeed on retry + + +@given('the first subplan will fail with "{error}"') +def step_first_fails(context: Context, error: str) -> None: + context.fail_map[_S1] = error + + +# --- Parallel execution --- + + +@given("a parent plan with {n:d} subplans in parallel mode with max_parallel {mp:d}") +def step_parent_parallel(context: Context, n: int, mp: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=mp, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given("a parent plan with {n:d} subplans in parallel mode with fail_fast") +def step_parent_parallel_failfast(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=n, + fail_fast=True, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +# --- Dependency-ordered execution --- + + +@given("a parent plan with {n:d} subplans in dependency_ordered mode") +def step_parent_dependency(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.DEPENDENCY_ORDERED, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + context.dependency_graph: dict[str, list[str]] = {} + + +@given("subplan C depends on subplan B which depends on subplan A") +def step_linear_deps(context: Context) -> None: + context.dependency_graph = { + _S1: [], + _S2: [_S1], + _S3: [_S2], + } + + +@given("the subplans have circular dependencies") +def step_circular_deps(context: Context) -> None: + context.dependency_graph = { + _S1: [_S2], + _S2: [_S1], + } + + +# --- When: execution --- + + +def _build_executor( + context: Context, +) -> Callable[[SubplanStatus], SubplanExecutionOutput]: + call_counts: dict[str, int] = {} + + def _exec(status: SubplanStatus) -> SubplanExecutionOutput: + sid = status.subplan_id + call_counts[sid] = call_counts.get(sid, 0) + 1 + with context.order_lock: + context.execution_order.append(sid) + + # Track concurrency when configured + concurrency: dict[str, int] | None = getattr( + context, "concurrency_counter", None + ) + concurrency_lock: threading.Lock | None = getattr( + context, "concurrency_lock", None + ) + if concurrency is not None and concurrency_lock is not None: + with concurrency_lock: + concurrency["current"] += 1 + concurrency["max"] = max(concurrency["max"], concurrency["current"]) + + try: + # Check if executor should raise an exception + raise_map: dict[str, str] = getattr(context, "raise_map", {}) + raise_counts: dict[str, int] = getattr(context, "raise_counts", {}) + if sid in raise_map and call_counts[sid] <= raise_counts.get(sid, 0): + error_type = raise_map[sid] + raise RuntimeError(f"{error_type}: simulated executor exception") + + # Check retry logic + max_fail = context.retry_map.get(sid, 0) + if sid in context.fail_map and call_counts[sid] <= max_fail: + error = context.fail_map[sid] + return SubplanExecutionOutput( + subplan_id=sid, + success=False, + error=error, + ) + if sid in context.fail_map and sid not in context.retry_map: + error = context.fail_map[sid] + return SubplanExecutionOutput( + subplan_id=sid, + success=False, + error=error, + ) + + # Per-subplan delay (supports staggered completion / timeout tests) + delay_map: dict[str, float] = getattr(context, "delay_map", {}) + delay = delay_map.get(sid, 0.01) + time.sleep(delay) + + override: dict[str, dict[str, str]] = getattr(context, "override_files", {}) + if sid in override: + files = override[sid] + else: + files = {f"src/{sid[-4:]}.py": f"# modified by {sid}\n"} + return SubplanExecutionOutput( + subplan_id=sid, + success=True, + files=files, + files_changed=len(files), + changeset_summary=f"Modified by {sid}", + ) + finally: + if concurrency is not None and concurrency_lock is not None: + with concurrency_lock: + concurrency["current"] -= 1 + + return _exec + + +@when("the subplans are executed") +def step_execute_subplans(context: Context) -> None: + executor_fn = _build_executor(context) + service = SubplanExecutionService( + config=context.subplan_cfg, + executor_fn=executor_fn, + ) + dep_graph = getattr(context, "dependency_graph", None) + base_files: dict[str, str] = getattr(context, "base_files_for_exec", {}) + context.exec_result = service.execute_all( + subplan_statuses=context.subplan_statuses, + base_files=base_files, + dependency_graph=dep_graph, + ) + + +@when("the subplans are executed expecting an error") +def step_execute_expecting_error(context: Context) -> None: + executor_fn = _build_executor(context) + service = SubplanExecutionService( + config=context.subplan_cfg, + executor_fn=executor_fn, + ) + dep_graph = getattr(context, "dependency_graph", None) + try: + context.exec_result = service.execute_all( + subplan_statuses=context.subplan_statuses, + base_files={}, + dependency_graph=dep_graph, + ) + context.exec_error = None + except Exception as exc: + context.exec_error = exc + context.exec_result = None + + +@when("the subplans are executed without a dependency graph") +def step_execute_no_dep_graph(context: Context) -> None: + executor_fn = _build_executor(context) + service = SubplanExecutionService( + config=context.subplan_cfg, + executor_fn=executor_fn, + ) + try: + context.exec_result = service.execute_all( + subplan_statuses=context.subplan_statuses, + base_files={}, + dependency_graph=None, + ) + context.exec_error = None + except Exception as exc: + context.exec_error = exc + context.exec_result = None + + +# --- Then: execution assertions --- + + +@then("all {n:d} subplans should complete successfully") +def step_all_succeeded(context: Context, n: int) -> None: + result: SubplanExecutionResult = context.exec_result + completed = [s for s in result.statuses if s.status == ProcessingState.COMPLETE] + assert len(completed) == n, ( + f"Expected {n} completed, got {len(completed)}: " + f"{[(s.subplan_id, s.status) for s in result.statuses]}" + ) + + +@then("the subplans should have been executed in order") +def step_executed_in_order(context: Context) -> None: + expected = [s.subplan_id for s in context.subplan_statuses] + # Check the execution_order list + actual = context.execution_order + # Sequential: first execution of each should be in order + seen: list[str] = [] + for sid in actual: + if sid not in seen: + seen.append(sid) + assert seen == expected, f"Expected order {expected}, got {seen}" + + +@then("the execution result should report all succeeded") +def step_result_all_succeeded(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert result.all_succeeded, ( + f"Expected all succeeded but got failures: {result.failed_subplan_ids}" + ) + + +@then("the first subplan should complete successfully") +def step_first_complete(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + first = result.statuses[0] + assert first.status == ProcessingState.COMPLETE, ( + f"Expected COMPLETE, got {first.status}" + ) + + +@then("the first subplan should be errored") +def step_first_errored(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + first = result.statuses[0] + assert first.status == ProcessingState.ERRORED, ( + f"Expected ERRORED, got {first.status}" + ) + + +@then("the second subplan should be errored") +def step_second_errored(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + second = result.statuses[1] + assert second.status == ProcessingState.ERRORED, ( + f"Expected ERRORED, got {second.status}" + ) + + +@then("the third subplan should be cancelled") +def step_third_cancelled(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + third = result.statuses[2] + assert third.status == ProcessingState.CANCELLED, ( + f"Expected CANCELLED, got {third.status}" + ) + + +@then("the subplan should complete successfully") +def step_single_complete(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert result.statuses[0].status == ProcessingState.COMPLETE + + +@then("the subplan should have {n:d} previous attempt recorded") +def step_previous_attempts(context: Context, n: int) -> None: + result: SubplanExecutionResult = context.exec_result + status = result.statuses[0] + assert len(status.previous_attempts) == n, ( + f"Expected {n} previous attempts, got {len(status.previous_attempts)}" + ) + + +@then("at least one subplan should be errored") +def step_at_least_one_errored(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + errored = [s for s in result.statuses if s.status == ProcessingState.ERRORED] + assert len(errored) >= 1, "Expected at least one errored subplan" + + +@then("the execution result should have failed subplan ids") +def step_has_failed_ids(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert len(result.failed_subplan_ids) > 0 + + +@then("subplan A should have completed before subplan B") +def step_a_before_b(context: Context) -> None: + order = context.execution_order + first_a = next(i for i, sid in enumerate(order) if sid == _S1) + first_b = next(i for i, sid in enumerate(order) if sid == _S2) + assert first_a < first_b, f"A at {first_a}, B at {first_b}" + + +@then("subplan B should have completed before subplan C") +def step_b_before_c(context: Context) -> None: + order = context.execution_order + first_b = next(i for i, sid in enumerate(order) if sid == _S2) + first_c = next(i for i, sid in enumerate(order) if sid == _S3) + assert first_b < first_c, f"B at {first_b}, C at {first_c}" + + +@then("a circular dependency error should be raised") +def step_circular_error(context: Context) -> None: + assert context.exec_error is not None + assert "Circular dependency" in str(context.exec_error) + + +@then("a missing dependency graph error should be raised") +def step_missing_graph_error(context: Context) -> None: + assert context.exec_error is not None + assert "dependency_graph is required" in str(context.exec_error) + + +# --- Merge strategy steps --- + + +@given("subplan outputs with non-overlapping changes to the same file") +def step_non_overlapping_outputs(context: Context) -> None: + context.base_files = {"src/main.py": "line1\nline2\nline3\n"} + context.subplan_outputs = [ + (_S1, {"src/main.py": "line1\nline2\nline3\nnew_line_a\n"}), + (_S2, {"src/main.py": "new_line_b\nline1\nline2\nline3\n"}), + ] + + +@given("subplan outputs with overlapping changes to the same file") +def step_overlapping_outputs(context: Context) -> None: + context.base_files = {"src/main.py": "original content\n"} + context.subplan_outputs = [ + (_S1, {"src/main.py": "first version\n"}), + (_S2, {"src/main.py": "second version\n"}), + ] + + +@given("subplan outputs with conflicting changes to the same file") +def step_conflicting_outputs(context: Context) -> None: + context.base_files = {"src/main.py": "base line\n"} + context.subplan_outputs = [ + (_S1, {"src/main.py": "alpha change\n"}), + (_S2, {"src/main.py": "beta change\n"}), + ] + + +@given("a single subplan output with file changes") +def step_single_output(context: Context) -> None: + context.base_files = {"src/main.py": "original\n"} + context.subplan_outputs = [ + (_S1, {"src/main.py": "modified by subplan\n"}), + ] + + +@given("no subplan outputs") +def step_no_outputs(context: Context) -> None: + context.base_files = {} + context.subplan_outputs = [] + + +@given("the merge strategy is {strategy}") +def step_merge_strategy(context: Context, strategy: str) -> None: + strategy_map = { + "git_three_way": SubplanMergeStrategy.GIT_THREE_WAY, + "sequential_apply": SubplanMergeStrategy.SEQUENTIAL_APPLY, + "fail_on_conflict": SubplanMergeStrategy.FAIL_ON_CONFLICT, + "last_wins": SubplanMergeStrategy.LAST_WINS, + } + context.merge_strategy = strategy_map[strategy] + + +@when("the subplan outputs are merged") +def step_merge_outputs(context: Context) -> None: + service = SubplanMergeService(context.merge_strategy) + context.merge_result = service.merge(context.base_files, context.subplan_outputs) + + +@when("the subplan outputs are merged expecting a conflict error") +def step_merge_expecting_conflict(context: Context) -> None: + service = SubplanMergeService(context.merge_strategy) + try: + context.merge_result = service.merge( + context.base_files, context.subplan_outputs + ) + context.merge_error = None + except MergeConflictError as exc: + context.merge_error = exc + context.merge_result = None + + +@when("the subplan outputs are merged expecting a validation error") +def step_merge_expecting_validation(context: Context) -> None: + service = SubplanMergeService(context.merge_strategy) + try: + context.merge_result = service.merge( + context.base_files, context.subplan_outputs + ) + context.merge_error = None + except ValueError as exc: + context.merge_error = exc + context.merge_result = None + + +@then("the merge should succeed") +def step_merge_succeed(context: Context) -> None: + result: SubplanMergeResult = context.merge_result + assert result.success, f"Merge failed with conflicts: {result.conflict_files}" + + +@then("the merged file should contain changes from both subplans") +def step_merged_both_changes(context: Context) -> None: + result: SubplanMergeResult = context.merge_result + content = result.merged_files[0].content + # Non-overlapping changes: should contain the new lines from both + assert "new_line_a" in content and "new_line_b" in content, ( + f"Expected content from both subplans, got: {content}" + ) + + +@then("the final content should reflect sequential application") +def step_sequential_content(context: Context) -> None: + result: SubplanMergeResult = context.merge_result + content = result.merged_files[0].content + # Sequential apply: last subplan's content wins + assert "second version" in content, f"Expected second version, got: {content}" + + +@then("a merge conflict error should be raised") +def step_conflict_error_raised(context: Context) -> None: + assert context.merge_error is not None + assert isinstance(context.merge_error, MergeConflictError) + + +@then("the merged content should be from the last subplan") +def step_last_wins_content(context: Context) -> None: + result: SubplanMergeResult = context.merge_result + content = result.merged_files[0].content + assert "second version" in content, f"Expected last wins, got: {content}" + + +@then("the merged content should match the single subplan output") +def step_single_output_match(context: Context) -> None: + result: SubplanMergeResult = context.merge_result + assert result.merged_files[0].content == "modified by subplan\n" + + +@then("an empty outputs error should be raised") +def step_empty_outputs_error(context: Context) -> None: + assert context.merge_error is not None + assert "must not be empty" in str(context.merge_error) + + +# --- Integration steps --- + + +@given("a parent plan with {n:d} subplans in sequential mode with git merge") +def step_parent_seq_git(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.SEQUENTIAL, + merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given("both subplans produce non-overlapping file changes") +def step_non_overlapping_changes(context: Context) -> None: + context.base_files_for_exec = {} + + +@given("a parent plan with {n:d} subplans in parallel mode with last_wins merge") +def step_parent_par_last_wins(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=n, + merge_strategy=SubplanMergeStrategy.LAST_WINS, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given("both subplans modify the same file differently") +def step_same_file_different(context: Context) -> None: + context.base_files_for_exec = {} + + +@then("the execution result should include a merge result") +def step_has_merge_result(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert result.merge_result is not None, "Expected merge result" + + +@then("the merge result should have no conflicts") +def step_no_conflicts(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert result.merge_result is not None + assert len(result.merge_result.conflict_files) == 0 + + +@then("the merge result should succeed") +def step_merge_result_succeeds(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert result.merge_result is not None + assert result.merge_result.success + + +# --- Validation steps --- + + +@when("a SubplanExecutionService is created with None config") +def step_none_config(context: Context) -> None: + try: + SubplanExecutionService( + config=None, # type: ignore[arg-type] + executor_fn=_success_executor, + ) + context.validation_error = None + except ValueError as exc: + context.validation_error = exc + + +@when("a SubplanExecutionService is created with None executor") +def step_none_executor(context: Context) -> None: + try: + SubplanExecutionService( + config=SubplanConfig(), + executor_fn=None, # type: ignore[arg-type] + ) + context.validation_error = None + except ValueError as exc: + context.validation_error = exc + + +@when("a SubplanMergeService is created with None strategy") +def step_none_strategy(context: Context) -> None: + try: + SubplanMergeService(strategy=None) # type: ignore[arg-type] + context.validation_error = None + except ValueError as exc: + context.validation_error = exc + + +@then("a config validation error should be raised") +def step_config_error(context: Context) -> None: + assert context.validation_error is not None + assert "config" in str(context.validation_error) + + +@then("an executor validation error should be raised") +def step_executor_error(context: Context) -> None: + assert context.validation_error is not None + assert "executor_fn" in str(context.validation_error) + + +@then("a strategy validation error should be raised") +def step_strategy_error(context: Context) -> None: + assert context.validation_error is not None + assert "strategy" in str(context.validation_error) + + +@given("a valid SubplanExecutionService") +def step_valid_service(context: Context) -> None: + context.exec_service = SubplanExecutionService( + config=SubplanConfig(), + executor_fn=_success_executor, + ) + + +@when("execute_all is called with empty subplan statuses") +def step_empty_statuses(context: Context) -> None: + try: + context.exec_service.execute_all( + subplan_statuses=[], + base_files={}, + ) + context.validation_error = None + except ValueError as exc: + context.validation_error = exc + + +@then("an empty statuses error should be raised") +def step_empty_statuses_error(context: Context) -> None: + assert context.validation_error is not None + assert "must not be empty" in str(context.validation_error) + + +# --- Property accessor steps --- + + +@then("the service config property should return the configured config") +def step_config_property(context: Context) -> None: + service: SubplanExecutionService = context.exec_service + assert service.config is not None + assert service.config.execution_mode == ExecutionMode.SEQUENTIAL + + +@then("the service merge_service property should return a merge service") +def step_merge_service_property(context: Context) -> None: + service: SubplanExecutionService = context.exec_service + assert service.merge_service is not None + assert isinstance(service.merge_service, SubplanMergeService) + + +@given("a SubplanMergeService with git_three_way strategy") +def step_merge_service_instance(context: Context) -> None: + context.merge_svc = SubplanMergeService(SubplanMergeStrategy.GIT_THREE_WAY) + + +@then("the merge strategy property should return git_three_way") +def step_strategy_property(context: Context) -> None: + assert context.merge_svc.strategy == SubplanMergeStrategy.GIT_THREE_WAY + + +# --- Retry edge-case steps --- + + +@given('the subplan will fail with "{error}"') +def step_subplan_always_fail(context: Context, error: str) -> None: + context.fail_map[_S1] = error + + +@given("a parent plan with {n:d} subplan in sequential mode with max_retries {mr:d}") +def step_parent_sequential_max_retries(context: Context, n: int, mr: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.SEQUENTIAL, + retry_failed=True, + max_retries=mr, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given('the subplan will always fail with "{error}"') +def step_subplan_always_fail_retriable(context: Context, error: str) -> None: + context.fail_map[_S1] = error + context.retry_map[_S1] = 999 # always fail (never enough retries) + + +@given('the executor will raise "{error_type}" once then succeed') +def step_executor_raises_once(context: Context, error_type: str) -> None: + context.raise_map = {_S1: error_type} + context.raise_counts = {_S1: 1} # raise once then succeed + + +@given('the executor will raise "{error_type}" permanently') +def step_executor_raises_permanent(context: Context, error_type: str) -> None: + context.raise_map = {_S1: error_type} + context.raise_counts = {_S1: 999} # always raise + + +@then("the subplan should be errored") +def step_single_errored(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert result.statuses[0].status == ProcessingState.ERRORED + + +@then("the subplan attempt number should be {n:d}") +def step_attempt_number(context: Context, n: int) -> None: + result: SubplanExecutionResult = context.exec_result + status = result.statuses[0] + assert status.attempt_number == n, ( + f"Expected attempt_number={n}, got {status.attempt_number}" + ) + + +# --- Timeout enforcement steps --- + + +@given( + "a parent plan with {n:d} subplan in sequential mode with a {t:d} second timeout" +) +def step_parent_seq_timeout(context: Context, n: int, t: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.SEQUENTIAL, + timeout_per_subplan_seconds=t, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given("the subplan executor will block for {s:d} seconds") +def step_slow_executor_all(context: Context, s: int) -> None: + context.delay_map = { + _S1: float(s), + _S2: float(s), + _S3: float(s), + _S4: float(s), + _S5: float(s), + } + + +@given("a parent plan with {n:d} subplans in parallel mode with a {t:d} second timeout") +def step_parent_par_timeout(context: Context, n: int, t: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=n, + timeout_per_subplan_seconds=t, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given("the first subplan executor will block for {s:d} seconds") +def step_first_slow(context: Context, s: int) -> None: + context.delay_map = {_S1: float(s)} + + +@then("the subplan error should mention timeout") +def step_error_mentions_timeout(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + status = result.statuses[0] + assert status.status == ProcessingState.ERRORED + assert "timeout" in (status.error or "").lower(), ( + f"Expected timeout in error message, got: {status.error}" + ) + + +@then("at least one subplan error should mention timeout") +def step_at_least_one_timeout(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + errored = [s for s in result.statuses if s.status == ProcessingState.ERRORED] + assert len(errored) > 0, "Expected at least one errored subplan" + assert any("timeout" in (s.error or "").lower() for s in errored), ( + f"No errored subplan mentions timeout: " + f"{[(s.subplan_id, s.error) for s in errored]}" + ) + + +# --- Completion order steps --- + + +@given("a parent plan with {n:d} subplans in parallel mode with staggered completion") +def step_parent_parallel_staggered(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=n, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + # S1 takes longest, S3 shortest -> completion order: S3, S2, S1 + context.delay_map = {_S1: 0.3, _S2: 0.2, _S3: 0.1} + + +@then("the execution result statuses should be in completion order") +def step_completion_order(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + completed_ids = [ + s.subplan_id for s in result.statuses if s.status == ProcessingState.COMPLETE + ] + # With staggered delays: S3 (0.1s) first, S2 (0.2s), S1 (0.3s) last + assert completed_ids[0] == _S3, f"Expected {_S3} first, got {completed_ids[0]}" + assert completed_ids[-1] == _S1, f"Expected {_S1} last, got {completed_ids[-1]}" + + +# --- Cancelled status steps --- + + +@given( + "a parent plan with {n:d} subplans in parallel mode" + " with max_parallel {mp:d} and fail_fast" +) +def step_parent_par_mp_failfast(context: Context, n: int, mp: int) -> None: + ids = [_S1, _S2, _S3, _S4, _S5][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=mp, + fail_fast=True, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@then("the remaining subplans should have CANCELLED status") +def step_remaining_cancelled(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + for status in result.statuses: + if status.subplan_id == _S1: + continue # Skip the errored first subplan + assert status.status == ProcessingState.CANCELLED, ( + f"Expected CANCELLED for {status.subplan_id}, got {status.status}" + ) + + +# --- Dependency-ordered concurrent execution steps --- + + +@given( + "a parent plan with {n:d} subplans in dependency_ordered mode" + " with concurrency tracking" +) +def step_parent_dep_concurrent(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.DEPENDENCY_ORDERED, + max_parallel=n, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + context.dependency_graph = {} + context.concurrency_counter = {"current": 0, "max": 0} + context.concurrency_lock = threading.Lock() + # Give A and B enough time to overlap + context.delay_map = {_S1: 0.15, _S2: 0.15, _S3: 0.05} + + +@given("subplans A and B are independent while C depends on both") +def step_ab_independent_c_depends(context: Context) -> None: + context.dependency_graph = { + _S1: [], + _S2: [], + _S3: [_S1, _S2], + } + + +@then("the peak concurrent execution count should be at least {n:d}") +def step_peak_concurrency(context: Context, n: int) -> None: + peak = context.concurrency_counter["max"] + assert peak >= n, f"Expected peak concurrency >= {n}, got {peak}" + + +@then("subplan C should have started after A and B completed") +def step_c_after_ab(context: Context) -> None: + order = context.execution_order + c_first = next(i for i, sid in enumerate(order) if sid == _S3) + a_first = next(i for i, sid in enumerate(order) if sid == _S1) + b_first = next(i for i, sid in enumerate(order) if sid == _S2) + assert a_first < c_first, f"A({a_first}) should execute before C({c_first})" + assert b_first < c_first, f"B({b_first}) should execute before C({c_first})" + + +# --- Dependency-ordered fail_fast steps --- + + +@given("a parent plan with {n:d} subplans in dependency_ordered mode with fail_fast") +def step_parent_dep_failfast(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.DEPENDENCY_ORDERED, + fail_fast=True, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + context.dependency_graph = {} + + +@then("the second and third subplans should have CANCELLED status") +def step_second_third_cancelled(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + non_first = [s for s in result.statuses if s.subplan_id != _S1] + for status in non_first: + assert status.status == ProcessingState.CANCELLED, ( + f"Expected CANCELLED for {status.subplan_id}, got {status.status}" + ) + + +# --- Dependency-ordered timeout steps --- + + +@given( + "a parent plan with {n:d} subplans in dependency_ordered mode" + " with a {t:d} second timeout" +) +def step_parent_dep_timeout(context: Context, n: int, t: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.DEPENDENCY_ORDERED, + timeout_per_subplan_seconds=t, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + context.dependency_graph = {} + + +@given("subplan B depends on subplan A") +def step_b_depends_a(context: Context) -> None: + context.dependency_graph = { + _S1: [], + _S2: [_S1], + } + + +@given( + "a parent plan with {n:d} subplans in dependency_ordered mode" + " with concurrency and a {t:d} second timeout" +) +def step_parent_dep_concurrent_timeout(context: Context, n: int, t: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.DEPENDENCY_ORDERED, + max_parallel=n, + timeout_per_subplan_seconds=t, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + context.dependency_graph = {} + + +# --- Parallel executor exception steps --- + + +@given('the executor will raise "{error_type}" for the first subplan in parallel') +def step_executor_raises_parallel(context: Context, error_type: str) -> None: + context.raise_map = {_S1: error_type} + context.raise_counts = {_S1: 999} # always raise + + +# --- Integration: merge conflict during execution --- + + +@given( + "a parent plan with {n:d} subplans in sequential mode with fail_on_conflict merge" +) +def step_parent_seq_fail_on_conflict(context: Context, n: int) -> None: + ids = [_S1, _S2, _S3][:n] + context.subplan_statuses = [_make_status(sid) for sid in ids] + context.subplan_cfg = SubplanConfig( + execution_mode=ExecutionMode.SEQUENTIAL, + merge_strategy=SubplanMergeStrategy.FAIL_ON_CONFLICT, + ) + context.fail_map = {} + context.retry_map = {} + context.execution_order = [] + context.order_lock = threading.Lock() + + +@given("both subplans produce conflicting file changes") +def step_conflicting_exec_outputs(context: Context) -> None: + context.base_files_for_exec = {"src/main.py": "base content\n"} + context.override_files = { + _S1: {"src/main.py": "alpha conflict\n"}, + _S2: {"src/main.py": "beta conflict\n"}, + } + + +@then("the execution result should report not all succeeded") +def step_not_all_succeeded(context: Context) -> None: + result: SubplanExecutionResult = context.exec_result + assert not result.all_succeeded, "Expected not all_succeeded but got True" diff --git a/features/subplan_execution.feature b/features/subplan_execution.feature new file mode 100644 index 000000000..08bf4c1e3 --- /dev/null +++ b/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 diff --git a/robot/helper_subplan_execution.py b/robot/helper_subplan_execution.py new file mode 100644 index 000000000..1362e77e2 --- /dev/null +++ b/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() diff --git a/robot/subplan_execution.robot b/robot/subplan_execution.robot new file mode 100644 index 000000000..93d5fbbe1 --- /dev/null +++ b/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 diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 8ed7916f4..dbe7f4ea8 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/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", diff --git a/src/cleveragents/application/services/subplan_execution_service.py b/src/cleveragents/application/services/subplan_execution_service.py new file mode 100644 index 000000000..2fa3467ed --- /dev/null +++ b/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 diff --git a/src/cleveragents/application/services/subplan_merge_service.py b/src/cleveragents/application/services/subplan_merge_service.py new file mode 100644 index 000000000..a9d2b149f --- /dev/null +++ b/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, + )