"""Step definitions for Plan Actor Integration tests.""" from __future__ import annotations from typing import Any from behave import given, then, when from behave.runner import Context from cleveragents.application.services.plan_executor import ( ExecuteResult, PlanExecutor, StrategizeResult, StrategyDecision, ) from cleveragents.application.services.plan_lifecycle_service import ( PlanLifecycleService, ) from cleveragents.config.settings import Settings from cleveragents.core.exceptions import PlanError, ValidationError from cleveragents.domain.models.core.plan import ( ProcessingState, ) from cleveragents.tool.builtins.changeset import ChangeSet from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner @given("I have a plan executor service") def step_create_executor_service(context: Context) -> None: """Create a plan executor service with in-memory lifecycle.""" settings = Settings() context.lifecycle_service = PlanLifecycleService(settings=settings) context.tool_registry = ToolRegistry() context.tool_runner = ToolRunner(registry=context.tool_registry) context.executor = PlanExecutor( lifecycle_service=context.lifecycle_service, tool_runner=context.tool_runner, ) context.error = None context.strategize_result = None context.execute_result = None context.stream_events: list[tuple[str, dict[str, Any]]] = [] @given("I have a plan executor with sandbox root") def step_create_executor_with_sandbox(context: Context) -> None: """Create a plan executor with sandbox root configured.""" import tempfile context.sandbox_dir = tempfile.mkdtemp() settings = Settings() context.lifecycle_service = PlanLifecycleService(settings=settings) context.tool_registry = ToolRegistry() context.tool_runner = ToolRunner(registry=context.tool_registry) context.executor = PlanExecutor( lifecycle_service=context.lifecycle_service, tool_runner=context.tool_runner, sandbox_root=context.sandbox_dir, ) context.error = None context.strategize_result = None context.execute_result = None context.stream_events = [] def _create_plan_in_strategize(context: Context, definition: str) -> str: """Helper to create a plan in strategize/queued state.""" action = context.lifecycle_service.create_action( name="local/test-executor-action", description="Test action for executor", definition_of_done=definition, strategy_actor="local/stub-strategize", execution_actor="local/stub-execute", ) plan = context.lifecycle_service.use_action( action_name=str(action.namespaced_name), ) context.plan = plan context.plan_id = plan.identity.plan_id return plan.identity.plan_id @given('I have a plan in strategize queued state with definition "{definition}"') def step_plan_strategize_queued(context: Context, definition: str) -> None: """Create a plan in strategize/queued state.""" # Unescape newlines from feature file definition = definition.replace("\\n", "\n") _create_plan_in_strategize(context, definition) @given("I have a plan in strategize queued state with empty definition") def step_plan_strategize_empty_definition(context: Context) -> None: """Create a plan in strategize/queued state with empty definition. The Action model requires non-empty definition_of_done, so we create the action with a minimal definition, then clear it on the plan after creation to test the empty-definition path in the strategize actor. """ plan_id = _create_plan_in_strategize(context, "placeholder") plan = context.lifecycle_service.get_plan(plan_id) plan.definition_of_done = None context.lifecycle_service._commit_plan(plan) @given("I have a plan with invariants in strategize queued state") def step_plan_with_invariants(context: Context) -> None: """Create a plan with invariants in strategize/queued state.""" from cleveragents.domain.models.core.plan import ( InvariantSource, PlanInvariant, ) action = context.lifecycle_service.create_action( name="local/test-invariant-action", description="Test action with invariants", definition_of_done="Tests pass", strategy_actor="local/stub-strategize", execution_actor="local/stub-execute", invariants=["Must not exceed 100 lines", "Must have docstrings"], ) plan = context.lifecycle_service.use_action( action_name=str(action.namespaced_name), invariants=[ PlanInvariant(text="Plan-level constraint", source=InvariantSource.PLAN) ], ) context.plan = plan context.plan_id = plan.identity.plan_id @given("I have a plan that completed strategize") def step_plan_completed_strategize(context: Context) -> None: """Create a plan that has completed the strategize phase.""" _create_plan_in_strategize(context, "Tests pass\nCoverage met") context.strategize_result = context.executor.run_strategize(context.plan_id) # Transition to execute phase context.lifecycle_service.execute_plan(context.plan_id) context.plan = context.lifecycle_service.get_plan(context.plan_id) @given("I have a plan in execute phase for executor") def step_plan_in_execute_phase(context: Context) -> None: """Create a plan already in execute phase.""" _create_plan_in_strategize(context, "Tests pass") context.executor.run_strategize(context.plan_id) context.lifecycle_service.execute_plan(context.plan_id) context.plan = context.lifecycle_service.get_plan(context.plan_id) @given("I have a plan in execute queued state without decisions") def step_plan_execute_no_decisions(context: Context) -> None: """Create a plan in execute/queued without decision tree.""" _create_plan_in_strategize(context, "Tests pass") # Manually force through strategize without setting decision_root_id context.lifecycle_service.start_strategize(context.plan_id) context.lifecycle_service.complete_strategize(context.plan_id) context.lifecycle_service.execute_plan(context.plan_id) context.plan = context.lifecycle_service.get_plan(context.plan_id) # Ensure no decision_root_id context.plan.decision_root_id = None context.lifecycle_service._commit_plan(context.plan) @given("the strategize actor is configured to fail") def step_strategize_actor_fails(context: Context) -> None: """Configure the strategize actor to raise an error.""" def _failing_execute(*args: Any, **kwargs: Any) -> Any: raise RuntimeError("Strategize actor deliberate failure") context.executor._strategize_actor.execute = _failing_execute # type: ignore[method-assign] @given("the execute actor is configured to fail") def step_execute_actor_fails(context: Context) -> None: """Configure the execute actor to raise an error.""" def _failing_execute(*args: Any, **kwargs: Any) -> Any: raise RuntimeError("Execute actor deliberate failure") context.executor._execute_actor.execute = _failing_execute # type: ignore[method-assign] @given("I have a stream callback registered") def step_register_stream_callback(context: Context) -> None: """Register a stream callback.""" context.stream_events = [] def _callback(event_type: str, data: dict[str, Any]) -> None: context.stream_events.append((event_type, data)) context.stream_callback = _callback # When steps @when("I run the strategize phase") def step_run_strategize(context: Context) -> None: """Run the strategize phase.""" context.strategize_result = context.executor.run_strategize(context.plan_id) context.plan = context.lifecycle_service.get_plan(context.plan_id) @when("I run the execute phase") def step_run_execute(context: Context) -> None: """Run the execute phase.""" context.execute_result = context.executor.run_execute(context.plan_id) context.plan = context.lifecycle_service.get_plan(context.plan_id) @when("I try to run the execute phase") def step_try_run_execute_wrong_phase(context: Context) -> None: """Try to run execute when plan is in wrong phase.""" context.error = None try: context.lifecycle_service.execute_plan(context.plan_id) except (PlanError, Exception) as exc: context.error = exc @when("I try to run the execute phase on the executor") def step_try_run_execute_on_executor(context: Context) -> None: """Try to run execute on the executor.""" context.error = None try: context.executor.run_execute(context.plan_id) except (PlanError, Exception) as exc: context.error = exc @when("I try to run the strategize phase") def step_try_run_strategize(context: Context) -> None: """Try to run strategize when plan is in wrong phase.""" context.error = None try: context.executor.run_strategize(context.plan_id) except (PlanError, Exception) as exc: context.error = exc @when("I try to run the strategize phase expecting failure") def step_try_run_strategize_failure(context: Context) -> None: """Try to run strategize expecting an actor failure.""" context.error = None try: context.executor.run_strategize(context.plan_id) except Exception as exc: context.error = exc context.plan = context.lifecycle_service.get_plan(context.plan_id) @when("I try to run the execute phase expecting failure") def step_try_run_execute_failure(context: Context) -> None: """Try to run execute expecting an actor failure.""" context.error = None try: context.executor.run_execute(context.plan_id) except Exception as exc: context.error = exc context.plan = context.lifecycle_service.get_plan(context.plan_id) @when("I run the strategize phase with streaming") def step_run_strategize_streaming(context: Context) -> None: """Run strategize with streaming callback.""" context.strategize_result = context.executor.run_strategize( context.plan_id, stream_callback=context.stream_callback ) context.plan = context.lifecycle_service.get_plan(context.plan_id) @when("I run the execute phase with streaming") def step_run_execute_streaming(context: Context) -> None: """Run execute with streaming callback.""" context.execute_result = context.executor.run_execute( context.plan_id, stream_callback=context.stream_callback ) context.plan = context.lifecycle_service.get_plan(context.plan_id) @when("I try to run strategize with empty plan id") def step_try_strategize_empty_id(context: Context) -> None: """Try to run strategize with empty plan id.""" context.error = None try: context.executor.run_strategize("") except ValidationError as exc: context.error = exc @when("I try to run execute with empty plan id") def step_try_execute_empty_id(context: Context) -> None: """Try to run execute with empty plan id.""" context.error = None try: context.executor.run_execute("") except ValidationError as exc: context.error = exc @when("I try to create a PlanExecutor with None lifecycle service") def step_try_create_executor_none(context: Context) -> None: """Try to create a PlanExecutor with None lifecycle service.""" context.error = None try: PlanExecutor(lifecycle_service=None) except ValidationError as exc: context.error = exc @when("I create a strategy decision with valid fields") def step_create_strategy_decision(context: Context) -> None: """Create a StrategyDecision with valid fields.""" context.decision = StrategyDecision( decision_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", step_text="Run unit tests", sequence=0, ) @when("I create a strategize result with 3 decisions") def step_create_strategize_result(context: Context) -> None: """Create a StrategizeResult with 3 decisions.""" from ulid import ULID root_id = str(ULID()) decisions = [ StrategyDecision( decision_id=root_id if i == 0 else str(ULID()), step_text=f"Step {i + 1}", sequence=i, parent_id=root_id if i > 0 else None, ) for i in range(3) ] context.strategize_result = StrategizeResult( decision_root_id=root_id, decisions=decisions, ) @when("I create an execute result with changeset") def step_create_execute_result(context: Context) -> None: """Create an ExecuteResult with a changeset.""" from ulid import ULID cs_id = str(ULID()) changeset = ChangeSet(plan_id="test-plan", entries=[]) context.execute_result = ExecuteResult( changeset_id=cs_id, changeset=changeset, tool_calls_count=5, sandbox_refs=["/tmp/sandbox"], ) @when("I transition the plan to execute") def step_transition_to_execute(context: Context) -> None: """Transition the plan from strategize complete to execute.""" context.lifecycle_service.execute_plan(context.plan_id) context.plan = context.lifecycle_service.get_plan(context.plan_id) # Then steps @then("strategize should produce {count:d} decisions") def step_check_decision_count(context: Context, count: int) -> None: """Verify the number of decisions produced.""" assert context.strategize_result is not None assert len(context.strategize_result.decisions) == count, ( f"Expected {count} decisions, got {len(context.strategize_result.decisions)}" ) @then("the decision root id should be set on the plan") def step_check_decision_root_id(context: Context) -> None: """Verify that decision_root_id is set on the plan.""" plan = context.lifecycle_service.get_plan(context.plan_id) assert plan.decision_root_id is not None, "decision_root_id should be set" @then("the plan should be in strategize complete state") def step_check_strategize_complete(context: Context) -> None: """Verify plan is in strategize/complete state.""" plan = context.lifecycle_service.get_plan(context.plan_id) # After complete_strategize, auto-progress may advance to execute assert plan.processing_state in ( ProcessingState.COMPLETE, ProcessingState.QUEUED, ), f"Expected complete or queued, got {plan.processing_state}" @then('the first decision should be "{expected}"') def step_check_first_decision(context: Context, expected: str) -> None: """Verify the first decision text.""" assert context.strategize_result is not None assert len(context.strategize_result.decisions) > 0 assert context.strategize_result.decisions[0].step_text == expected @then("the strategize result should include invariant records") def step_check_invariant_records(context: Context) -> None: """Verify invariant records are present.""" assert context.strategize_result is not None assert len(context.strategize_result.invariant_records) > 0 @then("each invariant record should have enforcement status") def step_check_invariant_enforcement(context: Context) -> None: """Verify each invariant record has enforcement info.""" assert context.strategize_result is not None for record in context.strategize_result.invariant_records: assert "enforced" in record assert "enforcement_note" in record @then("no changeset should be produced during strategize") def step_check_no_changeset_strategize(context: Context) -> None: """Verify strategize did not produce a changeset.""" plan = context.lifecycle_service.get_plan(context.plan_id) assert plan.changeset_id is None, "Strategize should not produce a changeset" @then("the plan sandbox refs should be empty") def step_check_sandbox_refs_empty(context: Context) -> None: """Verify sandbox refs are empty.""" plan = context.lifecycle_service.get_plan(context.plan_id) assert len(plan.sandbox_refs) == 0, "Sandbox refs should be empty" @then("the changeset id should be set on the plan") def step_check_changeset_id(context: Context) -> None: """Verify changeset_id is set on the plan.""" plan = context.lifecycle_service.get_plan(context.plan_id) assert plan.changeset_id is not None, "changeset_id should be set" @then("the plan should be in execute complete state") def step_check_execute_complete(context: Context) -> None: """Verify plan is in execute/complete state.""" plan = context.lifecycle_service.get_plan(context.plan_id) # After complete_execute, auto-progress may advance to apply assert plan.processing_state in ( ProcessingState.COMPLETE, ProcessingState.QUEUED, ), f"Expected complete or queued, got {plan.processing_state}" @then("the plan should have execution metadata") def step_check_execution_metadata(context: Context) -> None: """Verify execution metadata is present.""" assert context.execute_result is not None assert context.execute_result.changeset_id is not None @then("the tool calls count should be recorded") def step_check_tool_calls_count(context: Context) -> None: """Verify tool calls count is recorded.""" assert context.execute_result is not None assert context.execute_result.tool_calls_count >= 0 @then("the plan sandbox refs should not be empty") def step_check_sandbox_refs_not_empty(context: Context) -> None: """Verify sandbox refs are not empty.""" assert context.execute_result is not None assert len(context.execute_result.sandbox_refs) > 0 @then("a plan error should be raised about phase mismatch") def step_check_plan_error_phase(context: Context) -> None: """Verify a PlanError was raised about phase.""" assert context.error is not None, "Expected a PlanError but none was raised" @then("a plan error should be raised about missing decisions") def step_check_plan_error_decisions(context: Context) -> None: """Verify a PlanError was raised about missing decisions.""" assert context.error is not None, "Expected a PlanError but none was raised" assert ( "decision" in str(context.error).lower() or "strategize" in str(context.error).lower() ), f"Error should mention decisions: {context.error}" @then("the plan should be in errored state") def step_check_plan_errored(context: Context) -> None: """Verify the plan is in errored state.""" plan = context.lifecycle_service.get_plan(context.plan_id) assert plan.processing_state == ProcessingState.ERRORED, ( f"Expected errored, got {plan.processing_state}" ) @then("the plan should have error details") def step_check_error_details(context: Context) -> None: """Verify the plan has error details.""" plan = context.lifecycle_service.get_plan(context.plan_id) assert plan.error_details is not None, "Plan should have error_details" @then("the stream callback should have received strategize events") def step_check_strategize_stream_events(context: Context) -> None: """Verify strategize streaming events were received.""" assert len(context.stream_events) > 0, "Should have received stream events" @then("the stream events should include strategize_started") def step_check_strategize_started_event(context: Context) -> None: """Verify strategize_started event was emitted.""" event_types = [e[0] for e in context.stream_events] assert "strategize_started" in event_types @then("the stream events should include strategize_complete") def step_check_strategize_complete_event(context: Context) -> None: """Verify strategize_complete event was emitted.""" event_types = [e[0] for e in context.stream_events] assert "strategize_complete" in event_types @then("the stream callback should have received execute events") def step_check_execute_stream_events(context: Context) -> None: """Verify execute streaming events were received.""" assert len(context.stream_events) > 0, "Should have received stream events" @then("the stream events should include execute_started") def step_check_execute_started_event(context: Context) -> None: """Verify execute_started event was emitted.""" event_types = [e[0] for e in context.stream_events] assert "execute_started" in event_types @then("the stream events should include execute_complete") def step_check_execute_complete_event(context: Context) -> None: """Verify execute_complete event was emitted.""" event_types = [e[0] for e in context.stream_events] assert "execute_complete" in event_types @then("a validation error should be raised for empty plan id") def step_check_validation_error_empty(context: Context) -> None: """Verify a ValidationError was raised for empty plan id.""" assert context.error is not None assert isinstance(context.error, ValidationError) @then("a validation error should be raised for None lifecycle service") def step_check_validation_error_none(context: Context) -> None: """Verify a ValidationError was raised for None lifecycle service.""" assert context.error is not None assert isinstance(context.error, ValidationError) @then("the strategy decision should have the correct attributes") def step_check_decision_attributes(context: Context) -> None: """Verify StrategyDecision attributes.""" assert context.decision.decision_id == "01ARZ3NDEKTSV4RRFFQ69G5FAV" assert context.decision.step_text == "Run unit tests" assert context.decision.sequence == 0 assert context.decision.parent_id is None @then("the result should contain {count:d} decisions") def step_check_result_decisions(context: Context, count: int) -> None: """Verify the result contains expected decision count.""" assert context.strategize_result is not None assert len(context.strategize_result.decisions) == count @then("the result should have a root id") def step_check_result_root_id(context: Context) -> None: """Verify the result has a root id.""" assert context.strategize_result is not None assert context.strategize_result.decision_root_id is not None @then("the result should have a changeset id") def step_check_result_changeset_id(context: Context) -> None: """Verify the result has a changeset id.""" assert context.execute_result is not None assert context.execute_result.changeset_id is not None @then("the result should have a changeset object") def step_check_result_changeset_obj(context: Context) -> None: """Verify the result has a changeset object.""" assert context.execute_result is not None assert context.execute_result.changeset is not None assert isinstance(context.execute_result.changeset, ChangeSet) # Custom actor injection tests ------------------------------------------- class _SpyStrategizeActor: """A strategize actor that records calls for test assertions.""" def __init__(self) -> None: self.called = False self.call_count = 0 def execute( self, plan_id: str, definition_of_done: str | None, invariants: Any = None, stream_callback: Any = None, ) -> StrategizeResult: self.called = True self.call_count += 1 from ulid import ULID root_id = str(ULID()) return StrategizeResult( decision_root_id=root_id, decisions=[ StrategyDecision( decision_id=root_id, step_text=definition_of_done or "Complete the plan objectives", sequence=0, ), ], invariant_records=[], ) class _SpyExecuteActor: """An execute actor that records calls for test assertions.""" def __init__(self) -> None: self.called = False self.call_count = 0 def execute( self, plan_id: str, decisions: list[StrategyDecision], tool_runner: Any = None, sandbox_root: str | None = None, stream_callback: Any = None, *, read_only: bool = False, ) -> ExecuteResult: self.called = True self.call_count += 1 from ulid import ULID return ExecuteResult( changeset_id=str(ULID()), changeset=ChangeSet(plan_id=plan_id, entries=[]), tool_calls_count=0, sandbox_refs=[], ) @given("I have a plan executor with a custom strategize actor") def step_executor_with_custom_strategize(context: Context) -> None: """Create a PlanExecutor with a spy strategize actor.""" from cleveragents.config.settings import Settings settings = Settings() context.lifecycle_service = PlanLifecycleService(settings=settings) context.custom_strategize_actor = _SpyStrategizeActor() context.executor = PlanExecutor( lifecycle_service=context.lifecycle_service, strategize_actor=context.custom_strategize_actor, ) context.error = None context.strategize_result = None context.execute_result = None context.stream_events = [] @given("I have a plan executor with a custom execute actor") def step_executor_with_custom_execute(context: Context) -> None: """Create a PlanExecutor with a spy execute actor.""" from cleveragents.config.settings import Settings settings = Settings() context.lifecycle_service = PlanLifecycleService(settings=settings) context.custom_execute_actor = _SpyExecuteActor() context.executor = PlanExecutor( lifecycle_service=context.lifecycle_service, execute_actor=context.custom_execute_actor, ) context.error = None context.strategize_result = None context.execute_result = None context.stream_events = [] @then("the custom strategize actor should have been called") def step_check_custom_strategize_called(context: Context) -> None: """Verify the custom strategize actor was called.""" assert hasattr(context, "custom_strategize_actor"), ( "Expected custom_strategize_actor on context" ) assert context.custom_strategize_actor.called, ( "Custom strategize actor should have been called" ) @then("the custom execute actor should have been called") def step_check_custom_execute_called(context: Context) -> None: """Verify the custom execute actor was called.""" assert hasattr(context, "custom_execute_actor"), ( "Expected custom_execute_actor on context" ) assert context.custom_execute_actor.called, ( "Custom execute actor should have been called" )