cleveragents-core/features/steps/subplan_execution_steps.py

"""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)
        # Use the un-patched sleep so that ordering/timing tests see
        # real wall-clock delays rather than the global 10ms cap.
        _real_sleep = getattr(time, "_original_sleep", time.sleep)
        _real_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).
            # Use the un-patched sleep so that timeout scenarios see real
            # wall-clock delays rather than the global 10ms cap.
            _real_sleep = getattr(time, "_original_sleep", time.sleep)
            delay_map: dict[str, float] = getattr(context, "delay_map", {})
            delay = delay_map.get(sid, 0.01)
            _real_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 — use generous delays so that
    # concurrency is detected even under heavy CPU contention from the
    # parallel test runner.
    context.delay_map = {_S1: 0.5, _S2: 0.5, _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"