"""Step definitions for plan_executor_subplan_spawning.feature. Tests that PlanExecutor correctly wires SubplanService and SubplanExecutionService into the Execute phase so that subplan_spawn and subplan_parallel_spawn decisions are realised as actual child plan executions. """ from __future__ import annotations from unittest.mock import MagicMock from behave import given, then, when 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, SubplanConfig, SubplanStatus, ) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _PLAN_ID = "01KNFMGJSG67S6RG9TVXV205TQ" _ROOT_ID = "01KNFMGJSH67S6RG9TVXV205TR" _DEC_ID = "01KNFMGJSH67S6RG9TVXV205TS" _SUBPLAN_ID = "01KNFMGJSH67S6RG9TVXV205TT" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_plan( *, phase: PlanPhase = PlanPhase.EXECUTE, state: ProcessingState = ProcessingState.QUEUED, subplan_config: SubplanConfig | None = None, ) -> Plan: """Build a minimal Plan domain object for testing.""" 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 subplan spawning", action_name="local/test-action", phase=phase, processing_state=state, subplan_config=subplan_config, ) def _make_spawn_decision( decision_type: DecisionType = DecisionType.SUBPLAN_SPAWN, ) -> Decision: """Build a spawn-type 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_subplan_status( status: ProcessingState = ProcessingState.COMPLETE, error: str | None = None, ) -> SubplanStatus: """Build a SubplanStatus for testing.""" return SubplanStatus( subplan_id=_SUBPLAN_ID, action_name="local/sub-action", status=status, error=error, ) def _make_spawn_result( execution_mode: str = ExecutionMode.SEQUENTIAL, status: ProcessingState = ProcessingState.QUEUED, ) -> SpawnResult: """Build a SpawnResult with one subplan status.""" sub_status = SubplanStatus( subplan_id=_SUBPLAN_ID, action_name="local/sub-action", status=status, ) return SpawnResult( spawned_statuses=[sub_status], metadata={}, total_spawned=1, execution_mode=execution_mode, child_plans=[], ) def _make_exec_result( all_succeeded: bool = True, failed_ids: list[str] | None = None, ) -> SubplanExecutionResult: """Build a SubplanExecutionResult.""" failed = failed_ids or [] status = ProcessingState.COMPLETE if all_succeeded else ProcessingState.ERRORED sub_status = SubplanStatus( subplan_id=_SUBPLAN_ID, action_name="local/sub-action", status=status, error=None if all_succeeded else "ChildExecError: timeout", ) return SubplanExecutionResult( all_succeeded=all_succeeded, statuses=[sub_status], merge_result=None, total_duration_ms=10, failed_subplan_ids=failed, ) def _make_lifecycle(plan: Plan) -> MagicMock: """Build a mock lifecycle service that returns the given plan.""" 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 def _make_subplan_service( decisions: list[Decision] | None = None, spawn_result: SpawnResult | None = None, ) -> MagicMock: """Build a mock SubplanService.""" 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_subplan_execution_service( exec_result: SubplanExecutionResult | None = None, ) -> MagicMock: """Build a mock SubplanExecutionService.""" svc = MagicMock(spec=SubplanExecutionService) svc.execute_all.return_value = exec_result or _make_exec_result() return svc def _make_executor( plan: Plan, subplan_service: MagicMock | None = None, subplan_execution_service: MagicMock | None = None, *, runtime_mode: bool = False, ) -> tuple[PlanExecutor, MagicMock]: """Build a PlanExecutor with mocked dependencies.""" lcs = _make_lifecycle(plan) if runtime_mode: # Minimal execution context mock exec_ctx = MagicMock() exec_ctx.changeset_store = MagicMock() exec_ctx.decision_root_id = None exec_ctx.sandbox_root = None exec_ctx.sandbox_manager = None tool_runner = MagicMock() tool_runner.discover.return_value = [] # Mock RuntimeExecuteActor to avoid real LLM calls runtime_actor_result = MagicMock() runtime_actor_result.changeset_id = "01JSPAWN0000000000000CS0001" runtime_actor_result.sandbox_refs = [] runtime_actor_result.tool_call_count = 0 runtime_actor_result.decision_ids_processed = [] runtime_actor_result.execution_duration_ms = 5 import unittest.mock as _mock with _mock.patch( "cleveragents.application.services.plan_executor.RuntimeExecuteActor" ) as _MockActor: _MockActor.return_value.execute.return_value = runtime_actor_result executor = PlanExecutor( lifecycle_service=lcs, tool_runner=tool_runner, execution_context=exec_ctx, subplan_service=subplan_service, subplan_execution_service=subplan_execution_service, ) # Store the mock actor class for later assertions executor._mock_runtime_actor_cls = _MockActor else: # Stub mode: patch ExecuteStubActor.execute to avoid real work 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=subplan_service, subplan_execution_service=subplan_execution_service, ) return executor, lcs # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("a PlanExecutor configured with SubplanService and SubplanExecutionService") def step_given_executor_with_subplan_services(context: Context) -> None: """Set up a PlanExecutor with both subplan services mocked.""" decision = _make_spawn_decision(DecisionType.SUBPLAN_SPAWN) spawn_result = _make_spawn_result() exec_result = _make_exec_result(all_succeeded=True) context.spawn_decision = decision context.spawn_result = spawn_result context.exec_result = exec_result context.mock_subplan_svc = _make_subplan_service( decisions=[decision], spawn_result=spawn_result, ) context.mock_exec_svc = _make_subplan_execution_service(exec_result=exec_result) plan = _make_plan() plan.decision_root_id = _ROOT_ID context.plan = plan executor, lcs = _make_executor( plan, subplan_service=context.mock_subplan_svc, subplan_execution_service=context.mock_exec_svc, ) context.executor = executor context.lcs = lcs @given( "a PlanExecutor configured with SubplanService and SubplanExecutionService " "in runtime mode" ) def step_given_executor_with_subplan_services_runtime(context: Context) -> None: """Set up a PlanExecutor in runtime mode with both subplan services mocked.""" decision = _make_spawn_decision(DecisionType.SUBPLAN_SPAWN) spawn_result = _make_spawn_result() exec_result = _make_exec_result(all_succeeded=True) context.spawn_decision = decision context.spawn_result = spawn_result context.exec_result = exec_result context.mock_subplan_svc = _make_subplan_service( decisions=[decision], spawn_result=spawn_result, ) context.mock_exec_svc = _make_subplan_execution_service(exec_result=exec_result) plan = _make_plan() plan.decision_root_id = _ROOT_ID context.plan = plan context.runtime_mode = True @given("a parent plan in Execute phase with a subplan_spawn decision") def step_given_plan_with_spawn_decision(context: Context) -> None: """Ensure the plan has a subplan_spawn decision (already set in Given above).""" # Plan is already configured in the previous Given step assert context.plan is not None assert context.plan.phase == PlanPhase.EXECUTE @given("a parent plan in Execute phase with a subplan_parallel_spawn decision") def step_given_plan_with_parallel_spawn_decision(context: Context) -> None: """Set up a plan with a subplan_parallel_spawn decision.""" decision = _make_spawn_decision(DecisionType.SUBPLAN_PARALLEL_SPAWN) spawn_result = _make_spawn_result(execution_mode=ExecutionMode.PARALLEL) exec_result = _make_exec_result(all_succeeded=True) context.spawn_decision = decision context.spawn_result = spawn_result context.exec_result = exec_result context.mock_subplan_svc = _make_subplan_service( decisions=[decision], spawn_result=spawn_result, ) context.mock_exec_svc = _make_subplan_execution_service(exec_result=exec_result) plan = _make_plan() plan.decision_root_id = _ROOT_ID context.plan = plan executor, lcs = _make_executor( plan, subplan_service=context.mock_subplan_svc, subplan_execution_service=context.mock_exec_svc, ) context.executor = executor context.lcs = lcs @given("a parent plan in Execute phase with no spawn decisions") def step_given_plan_with_no_spawn_decisions(context: Context) -> None: """Set up a plan with no spawn decisions.""" context.mock_subplan_svc = _make_subplan_service(decisions=[]) context.mock_exec_svc = _make_subplan_execution_service() plan = _make_plan() plan.decision_root_id = _ROOT_ID context.plan = plan executor, lcs = _make_executor( plan, subplan_service=context.mock_subplan_svc, subplan_execution_service=context.mock_exec_svc, ) context.executor = executor context.lcs = lcs @given("a PlanExecutor with no SubplanService configured") def step_given_executor_without_subplan_service(context: Context) -> None: """Set up a PlanExecutor without SubplanService.""" plan = _make_plan() plan.decision_root_id = _ROOT_ID context.plan = plan executor, lcs = _make_executor( plan, subplan_service=None, subplan_execution_service=None, ) context.executor = executor context.lcs = lcs context.mock_subplan_svc = None context.mock_exec_svc = None @given("the SubplanExecutionService will report a failed subplan") def step_given_exec_svc_reports_failure(context: Context) -> None: """Reconfigure the execution service to report a failure.""" failed_exec_result = _make_exec_result( all_succeeded=False, failed_ids=[_SUBPLAN_ID], ) context.exec_result = failed_exec_result context.mock_exec_svc.execute_all.return_value = failed_exec_result # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when("I call run_execute on the parent plan") def step_when_run_execute(context: Context) -> None: """Call run_execute on the executor.""" try: context.execute_result = context.executor.run_execute( plan_id=_PLAN_ID, ) context.execute_error = None except Exception as exc: context.execute_result = None context.execute_error = exc @when("I call run_execute on the parent plan in runtime mode") def step_when_run_execute_runtime(context: Context) -> None: """Call run_execute in runtime mode.""" import unittest.mock as _mock plan = context.plan lcs = _make_lifecycle(plan) exec_ctx = MagicMock() exec_ctx.changeset_store = MagicMock() exec_ctx.decision_root_id = None exec_ctx.sandbox_root = None exec_ctx.sandbox_manager = None tool_runner = MagicMock() tool_runner.discover.return_value = [] runtime_actor_result = MagicMock() runtime_actor_result.changeset_id = "01JSPAWN0000000000000CS0001" runtime_actor_result.sandbox_refs = [] runtime_actor_result.tool_call_count = 0 runtime_actor_result.decision_ids_processed = [] runtime_actor_result.execution_duration_ms = 5 with _mock.patch( "cleveragents.application.services.plan_executor.RuntimeExecuteActor" ) as MockActor: MockActor.return_value.execute.return_value = runtime_actor_result executor = PlanExecutor( lifecycle_service=lcs, tool_runner=tool_runner, execution_context=exec_ctx, subplan_service=context.mock_subplan_svc, subplan_execution_service=context.mock_exec_svc, ) try: context.execute_result = executor.run_execute(plan_id=_PLAN_ID) context.execute_error = None except Exception as exc: context.execute_result = None context.execute_error = exc context.executor = executor context.lcs = lcs # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then("SubplanService.get_spawn_decisions should have been called") def step_then_get_spawn_decisions_called(context: Context) -> None: """Verify get_spawn_decisions was called with the plan ID.""" assert context.execute_error is None, f"run_execute raised: {context.execute_error}" context.mock_subplan_svc.get_spawn_decisions.assert_called_once_with(_PLAN_ID) @then("SubplanService.spawn should have been called with the spawn entries") def step_then_spawn_called(context: Context) -> None: """Verify spawn was called.""" context.mock_subplan_svc.spawn.assert_called_once() @then("SubplanExecutionService.execute_all should have been called") def step_then_execute_all_called(context: Context) -> None: """Verify execute_all was called.""" context.mock_exec_svc.execute_all.assert_called_once() @then("the parent plan subplan_statuses should be updated") def step_then_subplan_statuses_updated(context: Context) -> None: """Verify the plan's subplan_statuses were updated via _commit_plan.""" context.lcs._commit_plan.assert_called() @then("SubplanService.spawn should NOT have been called") def step_then_spawn_not_called(context: Context) -> None: """Verify spawn was NOT called (no-op path).""" assert context.execute_error is None, f"run_execute raised: {context.execute_error}" context.mock_subplan_svc.spawn.assert_not_called() @then("SubplanExecutionService.execute_all should NOT have been called") def step_then_execute_all_not_called(context: Context) -> None: """Verify execute_all was NOT called (no-op path).""" context.mock_exec_svc.execute_all.assert_not_called() @then("the execute phase completes without error") def step_then_execute_completes_without_error(context: Context) -> None: """Verify run_execute completed without raising.""" assert context.execute_error is None, ( f"Expected no error but got: {context.execute_error}" ) @then("no subplan spawning occurs") def step_then_no_subplan_spawning(context: Context) -> None: """Verify no subplan spawning occurred (no SubplanService configured).""" # No mock_subplan_svc to assert on — just verify no error assert context.execute_error is None, ( f"Expected no error but got: {context.execute_error}" ) @then("the parent plan error_details should contain failed_subplan_ids") def step_then_error_details_has_failed_ids(context: Context) -> None: """Verify error_details contains failed_subplan_ids.""" assert context.execute_error is None, f"run_execute raised: {context.execute_error}" # The _commit_plan call should have been made with a plan that has # error_details containing failed_subplan_ids commit_calls = context.lcs._commit_plan.call_args_list assert commit_calls, "Expected _commit_plan to be called" # Find the call where error_details was set with failed_subplan_ids found = False for c in commit_calls: committed_plan = c[0][0] details = getattr(committed_plan, "error_details", None) or {} if "failed_subplan_ids" in details: found = True break assert found, ( "Expected error_details to contain 'failed_subplan_ids' in a " f"_commit_plan call. Calls: {commit_calls}" ) @then("the parent plan error_details should contain subplan_execution_failed true") def step_then_error_details_has_exec_failed(context: Context) -> None: """Verify error_details contains subplan_execution_failed=true.""" commit_calls = context.lcs._commit_plan.call_args_list found = False for c in commit_calls: committed_plan = c[0][0] details = getattr(committed_plan, "error_details", None) or {} if details.get("subplan_execution_failed") == "true": found = True break assert found, ( "Expected error_details to contain 'subplan_execution_failed'='true' " f"in a _commit_plan call. Calls: {commit_calls}" )