cleveragents-core/features/steps/plan_execution_hierarchical_steps.py

"""Step definitions for plan_execution_hierarchical.feature.

Tests that PlanExecutor lazily creates SubplanExecutionService when
SubplanService is wired but SubplanExecutionService is not pre-configured
(Forgejo #10268).
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

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

from cleveragents.application.services.plan_executor import PlanExecutor
from cleveragents.application.services.subplan_execution_service import (
    SubplanExecutionResult,
    SubplanExecutionService,
)
from cleveragents.application.services.subplan_service import (
    SpawnEntry,
    SpawnResult,
    SubplanService,
)
from cleveragents.domain.models.core.decision import (
    ContextSnapshot,
    Decision,
    DecisionType,
)
from cleveragents.domain.models.core.plan import (
    ExecutionMode,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    ProcessingState,
    SubplanStatus,
)

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

_PLAN_ID = "01KNFMGJSG67S6RG9TVXV205TQ"
_ROOT_ID = "01KNFMGJSH67S6RG9TVXV205TR"
_DEC_ID = "01KNFMGJSH67S6RG9TVXV205TS"
_SUBPLAN_ID = "01KNFMGJSH67S6RG9TVXV205TT"


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_plan(*, phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED):
    return Plan(
        identity=PlanIdentity(plan_id=_PLAN_ID, root_plan_id=_ROOT_ID),
        namespaced_name=NamespacedName(namespace="local", name="test-plan"),
        description="Test plan for hierarchical execution",
        action_name="local/test-action",
        phase=phase,
        processing_state=state,
    )


def _make_spawn_decision(
    decision_type: DecisionType = DecisionType.SUBPLAN_SPAWN,
) -> Decision:
    return Decision(
        decision_id=_DEC_ID,
        plan_id=_PLAN_ID,
        decision_type=decision_type,
        sequence_number=0,
        question="Spawn a child plan?",
        chosen_option="local/sub-action",
        context_snapshot=ContextSnapshot(relevant_resources=[]),
    )


def _make_spawn_result(
    execution_mode: str = ExecutionMode.SEQUENTIAL,
) -> SpawnResult:
    sub_status = SubplanStatus(
        subplan_id=_SUBPLAN_ID,
        action_name="local/sub-action",
    )
    return SpawnResult(
        spawned_statuses=[sub_status],
        metadata={},
        total_spawned=1,
        execution_mode=execution_mode,
        child_plans=[],
    )


def _make_subplan_service(
    decisions: list[Decision] | None = None,
    spawn_result: SpawnResult | None = None,
) -> MagicMock:
    svc = MagicMock(spec=SubplanService)
    svc.get_spawn_decisions.return_value = decisions or []
    svc.build_spawn_entries.return_value = [
        SpawnEntry(decision=d, action_name="local/sub-action")
        for d in (decisions or [])
    ]
    svc.spawn.return_value = spawn_result or _make_spawn_result()
    return svc


def _make_lifecycle(plan: Plan) -> MagicMock:
    lcs = MagicMock()
    lcs.get_plan.return_value = plan
    lcs.start_execute = MagicMock()
    lcs.complete_execute = MagicMock()
    lcs.fail_execute = MagicMock()
    lcs._commit_plan = MagicMock()
    return lcs


# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------


@given("a PlanExecutor with SubplanService but no SubplanExecutionService")
def step_given_executor_with_subplan_service_only(context: Context) -> None:
    decision = _make_spawn_decision(DecisionType.SUBPLAN_SPAWN)
    spawn_result = _make_spawn_result()
    context.spawn_decision = decision
    context.spawn_result = spawn_result

    mock_subplan_svc = _make_subplan_service(
        decisions=[decision], spawn_result=spawn_result
    )
    plan = _make_plan()
    plan.decision_root_id = _ROOT_ID
    context.plan = plan

    lcs = _make_lifecycle(plan)
    execute_actor = MagicMock()
    execute_result = MagicMock()
    execute_result.changeset_id = "01JSPAWN0000000000000CS0001"
    execute_result.sandbox_refs = []
    execute_result.tool_calls_count = 0
    execute_actor.execute.return_value = execute_result

    executor = PlanExecutor(
        lifecycle_service=lcs,
        execute_actor=execute_actor,
        subplan_service=mock_subplan_svc,
        subplan_execution_service=None,
    )

    sub_status = SubplanStatus(
        subplan_id=_SUBPLAN_ID,
        action_name="local/sub-action",
        status=ProcessingState.COMPLETE,
    )
    patcher = patch.object(SubplanExecutionService, "execute_all")
    mock_exec_all = patcher.start()
    mock_exec_all.return_value = SubplanExecutionResult(
        all_succeeded=True,
        statuses=[sub_status],
        merge_result=None,
        total_duration_ms=5,
        failed_subplan_ids=[],
    )
    context._superb_patcher = patcher
    context._mock_execute_all = mock_exec_all

    context.executor = executor
    context.lcs = lcs
    context.mock_subplan_svc = mock_subplan_svc


@given("a child plan ready for execution in the lifecycle service")
def step_given_child_plan_ready(context: Context) -> None:
    child_plan = Plan(
        identity=PlanIdentity(
            plan_id=_SUBPLAN_ID,
            parent_plan_id=_PLAN_ID,
            root_plan_id=_ROOT_ID,
        ),
        namespaced_name=NamespacedName(namespace="local", name="child-plan"),
        description="Child plan for testing",
        action_name="local/sub-action",
        phase=PlanPhase.STRATEGIZE,
        processing_state=ProcessingState.QUEUED,
        definition_of_done="- [ ] Step one",
    )
    context.lcs.get_plan.side_effect = lambda pid: (
        child_plan if pid == _SUBPLAN_ID else context.plan
    )
    context.child_plan = child_plan


# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------


@then("SubplanExecutionService.execute_all should have been called via lazy creation")
def step_then_execute_all_called_via_lazy(context: Context) -> None:
    try:
        context._mock_execute_all.assert_called()
    finally:
        context._superb_patcher.stop()