temp/features/steps/subplan_execution_coverage_boost_steps.py

"""Step definitions for subplan execution defensive code-path coverage.

Targets the following uncovered lines in subplan_execution_service.py:
  - 298-303: _execute_parallel generic Exception handler
  - 331-334: _execute_parallel safety-net for uncaptured futures
  - 381:     _execute_dependency_ordered empty-ready break
  - 573-575: _execute_wave CancelledError handler
  - 576-578: _execute_wave generic Exception handler
"""

from __future__ import annotations

from concurrent.futures import CancelledError
from unittest.mock import patch

from behave import given, then, when
from behave.runner import Context

from cleveragents.application.services.subplan_execution_service import (
    SubplanExecutionOutput,
    SubplanExecutionService,
)
from cleveragents.domain.models.core.plan import (
    ExecutionMode,
    ProcessingState,
    SubplanConfig,
    SubplanStatus,
)

# Stable fake subplan IDs (valid ULIDs)
_P1 = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
_P2 = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
_P3 = "01HGZ6FE0AQDYTR4BXVQZ6EC00"


def _make_status(subplan_id: str) -> SubplanStatus:
    return SubplanStatus(subplan_id=subplan_id, action_name="test/action")


def _ok_output(status: SubplanStatus) -> SubplanExecutionOutput:
    return SubplanExecutionOutput(
        subplan_id=status.subplan_id,
        success=True,
        files={f"src/{status.subplan_id[-4:]}.py": f"# ok {status.subplan_id}\n"},
        files_changed=1,
        changeset_summary="ok",
    )


# ========================================================================
# Scenario: Parallel execution catches generic exception (lines 298-303)
# ========================================================================


@given("a subplan execution service in parallel mode with 2 subplans")
def step_parallel_2_subplans(context: Context) -> None:
    context.cb_statuses = [_make_status(_P1), _make_status(_P2)]
    context.cb_config = SubplanConfig(
        execution_mode=ExecutionMode.PARALLEL,
        max_parallel=2,
    )


@given(
    "the internal retry method is patched to raise a RuntimeError for the first subplan"
)
def step_patch_retry_raises_runtime(context: Context) -> None:
    """Record that we want _execute_one_with_retry to raise for _P1."""
    context.raise_for_subplan_id = _P1
    context.raise_exception = RuntimeError("unexpected internal failure")


@when("parallel execution is invoked")
def step_invoke_parallel(context: Context) -> None:
    service = SubplanExecutionService(
        config=context.cb_config,
        executor_fn=_ok_output,
    )

    target_id = context.raise_for_subplan_id
    target_exc = context.raise_exception
    original_method = service._execute_one_with_retry

    def _patched_retry(status: SubplanStatus):
        if status.subplan_id == target_id:
            raise target_exc
        return original_method(status)

    with patch.object(service, "_execute_one_with_retry", side_effect=_patched_retry):
        context.cb_result = service._execute_parallel(context.cb_statuses)


@then("the first subplan should be marked ERRORED with the RuntimeError message")
def step_first_errored_runtime(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    p1_status = next(s for s in updated_statuses if s.subplan_id == _P1)
    assert p1_status.status == ProcessingState.ERRORED, (
        f"Expected ERRORED, got {p1_status.status}"
    )
    assert "unexpected internal failure" in (p1_status.error or ""), (
        f"Expected RuntimeError message in error, got: {p1_status.error}"
    )


@then("the second subplan should complete normally")
def step_second_completes(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    p2_status = next(s for s in updated_statuses if s.subplan_id == _P2)
    assert p2_status.status == ProcessingState.COMPLETE, (
        f"Expected COMPLETE, got {p2_status.status}"
    )


# ========================================================================
# Scenario: Parallel safety-net for uncaptured futures (lines 331-334)
# ========================================================================


@given("a subplan execution service in parallel fail-fast mode with 3 subplans")
def step_parallel_failfast_3(context: Context) -> None:
    context.cb_statuses = [_make_status(_P1), _make_status(_P2), _make_status(_P3)]
    context.cb_config = SubplanConfig(
        execution_mode=ExecutionMode.PARALLEL,
        max_parallel=3,
        fail_fast=True,
    )


@given("as_completed is patched to skip the third future and the first subplan fails")
def step_patch_as_completed_skip_third(context: Context) -> None:
    """Record that the first subplan will fail and as_completed should skip _P3's future."""
    context.skip_subplan_id = _P3
    context.fail_subplan_id = _P1


@when("parallel execution is invoked with patched as_completed")
def step_invoke_parallel_patched_as_completed(context: Context) -> None:
    skip_id = context.skip_subplan_id
    fail_id = context.fail_subplan_id

    def _failing_executor(status: SubplanStatus) -> SubplanExecutionOutput:
        if status.subplan_id == fail_id:
            return SubplanExecutionOutput(
                subplan_id=status.subplan_id,
                success=False,
                error="ValidationError: forced failure",
            )
        return _ok_output(status)

    service = SubplanExecutionService(
        config=context.cb_config,
        executor_fn=_failing_executor,
    )

    # We need to intercept as_completed to skip the future for _P3.
    # We patch it at the module level where it's imported.
    import cleveragents.application.services.subplan_execution_service as svc_mod

    original_as_completed = svc_mod.as_completed

    def _filtered_as_completed(fs, **kwargs):
        """Yield all futures except the one for the skipped subplan."""
        for future in original_as_completed(fs, **kwargs):
            # fs is the future_to_id dict; check if this future maps to skip_id
            if isinstance(fs, dict) and fs.get(future) == skip_id:
                continue
            yield future

    with patch.object(svc_mod, "as_completed", _filtered_as_completed):
        context.cb_result = service._execute_parallel(context.cb_statuses)


@then("the third subplan should be marked CANCELLED by the safety net")
def step_third_cancelled_safety_net(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    p3_status = next(s for s in updated_statuses if s.subplan_id == _P3)
    assert p3_status.status == ProcessingState.CANCELLED, (
        f"Expected CANCELLED for P3 (safety net), got {p3_status.status}"
    )


@then("the results should contain all 3 subplans")
def step_results_contain_all_3(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    ids = {s.subplan_id for s in updated_statuses}
    assert _P1 in ids and _P2 in ids and _P3 in ids, (
        f"Expected all 3 subplan IDs, got {ids}"
    )


# ========================================================================
# Scenario: Dependency-ordered empty ready list break (line 381)
# ========================================================================


@given("a subplan execution service in dependency-ordered mode with 2 subplans")
def step_dep_ordered_2(context: Context) -> None:
    context.cb_statuses = [_make_status(_P1), _make_status(_P2)]
    context.cb_config = SubplanConfig(
        execution_mode=ExecutionMode.DEPENDENCY_ORDERED,
    )


@given("topological sort is patched to accept a circular dependency graph")
def step_patch_topo_sort(context: Context) -> None:
    """We'll provide a circular graph but patch _topological_sort to not raise."""
    context.circular_graph = {
        _P1: [_P2],
        _P2: [_P1],
    }


@when("dependency-ordered execution is invoked with the circular graph")
def step_invoke_dep_ordered_circular(context: Context) -> None:
    service = SubplanExecutionService(
        config=context.cb_config,
        executor_fn=_ok_output,
    )

    # Patch _topological_sort to accept the circular graph (return node_ids as-is)
    def _fake_topo_sort(node_ids, graph):
        return list(node_ids)

    with patch.object(
        SubplanExecutionService, "_topological_sort", staticmethod(_fake_topo_sort)
    ):
        context.cb_result = service._execute_dependency_ordered(
            context.cb_statuses, context.circular_graph
        )


@then("execution should complete without raising an error")
def step_no_error(context: Context) -> None:
    # If we got here, no exception was raised
    assert context.cb_result is not None


@then("the number of completed subplans should be less than 2")
def step_fewer_completed(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    # With a circular graph and patched topo sort, the while loop should break
    # when ready becomes empty. Neither _P1 nor _P2 can ever become ready because
    # each depends on the other (remaining_deps is never emptied).
    assert len(updated_statuses) < 2, (
        f"Expected fewer than 2 completed statuses, got {len(updated_statuses)}"
    )


# ========================================================================
# Scenario: Wave CancelledError (lines 573-575)
# ========================================================================


@given(
    "a subplan execution service in dependency-ordered mode with 3 independent subplans"
)
def step_dep_ordered_3_independent(context: Context) -> None:
    context.cb_statuses = [_make_status(_P1), _make_status(_P2), _make_status(_P3)]
    context.cb_config = SubplanConfig(
        execution_mode=ExecutionMode.DEPENDENCY_ORDERED,
        max_parallel=3,
    )
    # All independent (no dependencies) -> single wave of 3
    context.dep_graph = {_P1: [], _P2: [], _P3: []}


@given(
    "the internal retry method is patched to raise CancelledError "
    "for the second subplan in wave"
)
def step_patch_retry_cancelled(context: Context) -> None:
    context.wave_cancel_subplan_id = _P2


@when("dependency-ordered execution is invoked for the wave scenario")
def step_invoke_wave_cancelled(context: Context) -> None:
    cancel_id = context.wave_cancel_subplan_id

    service = SubplanExecutionService(
        config=context.cb_config,
        executor_fn=_ok_output,
    )

    original_method = service._execute_one_with_retry

    def _patched_retry(status: SubplanStatus):
        if status.subplan_id == cancel_id:
            raise CancelledError("simulated cancellation")
        return original_method(status)

    with patch.object(service, "_execute_one_with_retry", side_effect=_patched_retry):
        context.cb_result = service._execute_dependency_ordered(
            context.cb_statuses, context.dep_graph
        )


@then("the second subplan should be marked CANCELLED")
def step_second_cancelled(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    p2_status = next(s for s in updated_statuses if s.subplan_id == _P2)
    assert p2_status.status == ProcessingState.CANCELLED, (
        f"Expected CANCELLED, got {p2_status.status}"
    )


@then("the other subplans in the wave should complete successfully")
def step_others_complete_wave(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    for status in updated_statuses:
        if status.subplan_id == _P2:
            continue
        assert status.status == ProcessingState.COMPLETE, (
            f"Expected COMPLETE for {status.subplan_id}, got {status.status}"
        )


# ========================================================================
# Scenario: Wave generic Exception (lines 576-578)
# ========================================================================


@given(
    "a subplan execution service in dependency-ordered mode "
    "with 3 independent subplans for exception test"
)
def step_dep_ordered_3_independent_exc(context: Context) -> None:
    context.cb_statuses = [_make_status(_P1), _make_status(_P2), _make_status(_P3)]
    context.cb_config = SubplanConfig(
        execution_mode=ExecutionMode.DEPENDENCY_ORDERED,
        max_parallel=3,
    )
    context.dep_graph = {_P1: [], _P2: [], _P3: []}


@given(
    "the internal retry method is patched to raise ValueError "
    "for the second subplan in wave"
)
def step_patch_retry_valueerror(context: Context) -> None:
    context.wave_error_subplan_id = _P2
    context.wave_error_exception = ValueError("bad value in wave")


@when("dependency-ordered execution is invoked for the wave exception scenario")
def step_invoke_wave_exception(context: Context) -> None:
    error_id = context.wave_error_subplan_id
    error_exc = context.wave_error_exception

    service = SubplanExecutionService(
        config=context.cb_config,
        executor_fn=_ok_output,
    )

    original_method = service._execute_one_with_retry

    def _patched_retry(status: SubplanStatus):
        if status.subplan_id == error_id:
            raise error_exc
        return original_method(status)

    with patch.object(service, "_execute_one_with_retry", side_effect=_patched_retry):
        context.cb_result = service._execute_dependency_ordered(
            context.cb_statuses, context.dep_graph
        )


@then("the second subplan should be marked ERRORED with the ValueError message")
def step_second_errored_valueerror(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    p2_status = next(s for s in updated_statuses if s.subplan_id == _P2)
    assert p2_status.status == ProcessingState.ERRORED, (
        f"Expected ERRORED, got {p2_status.status}"
    )
    assert "bad value in wave" in (p2_status.error or ""), (
        f"Expected ValueError message in error, got: {p2_status.error}"
    )


@then("the other wave subplans should still complete")
def step_others_complete_wave_exc(context: Context) -> None:
    updated_statuses, _outputs = context.cb_result
    for status in updated_statuses:
        if status.subplan_id == _P2:
            continue
        assert status.status == ProcessingState.COMPLETE, (
            f"Expected COMPLETE for {status.subplan_id}, got {status.status}"
        )