"""Step definitions for plan_executor_coverage_boost.feature. These steps target specific uncovered lines in plan_executor.py: - Lines 328-330: _try_emit_metric success path (emitter configured) - Lines 331-335: _try_emit_metric exception path (emitter raises) - Line 384: _try_create_checkpoint returns None (no sandbox) - Line 422: _try_rollback_to_last_checkpoint returns False (no sandbox) - Line 458: _resolve_sandbox_for_checkpoint via execution_context - Lines 598-599: _enforce_guardrails raises PlanError (step limit) - Line 612: _enforce_guardrails_per_step returns early (guardrails None) - Line 619: _enforce_guardrails_per_step raises PlanError (wall-clock) """ from types import SimpleNamespace from unittest.mock import MagicMock from behave import given, then, when from ulid import ULID from cleveragents.application.services.autonomy_guardrail_service import ( AutonomyGuardrailService, ) from cleveragents.application.services.plan_executor import ( PlanExecutor, ) from cleveragents.core.exceptions import PlanError from cleveragents.domain.models.core.autonomy_guardrails import AutonomyGuardrails from cleveragents.domain.models.observability.metrics import OperationalMetricKey from cleveragents.infrastructure.sandbox.checkpoint import CheckpointManager # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- # A valid 26-char ULID for use as plan_id in metrics and other calls. _VALID_PLAN_ID = str(ULID()) def _make_lifecycle_stub(): """Return a minimal mock lifecycle service.""" return MagicMock(name="lifecycle_service") # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the plan executor module is imported") def step_plan_executor_module_imported(context): """Verify that PlanExecutor is importable.""" assert PlanExecutor is not None # --------------------------------------------------------------------------- # _try_emit_metric success path (lines 328-330) # --------------------------------------------------------------------------- @given("a PlanExecutor with a mock metrics emitter") def step_executor_with_mock_emitter(context): """Create a PlanExecutor with a mocked metrics_emitter.""" context.mock_emitter = MagicMock(name="metrics_emitter") context.executor = PlanExecutor( lifecycle_service=_make_lifecycle_stub(), metrics_emitter=context.mock_emitter, ) @when("I trigger metric emission with a valid key and value") def step_trigger_metric_emission(context): """Call _try_emit_metric with a valid key, plan_id, and value.""" context.emit_error = None try: context.executor._try_emit_metric( key=OperationalMetricKey.PLAN_DURATION_MS, plan_id=_VALID_PLAN_ID, value=42.5, ) except Exception as exc: context.emit_error = exc @then("the metrics emitter should have received exactly one entry") def step_verify_emitter_called(context): """Verify the mock emitter's emit() was called once.""" assert context.emit_error is None, f"Unexpected error: {context.emit_error!r}" assert context.mock_emitter.emit.call_count == 1 # --------------------------------------------------------------------------- # _try_emit_metric exception path (lines 331-335) # --------------------------------------------------------------------------- @given("a PlanExecutor with a failing metrics emitter") def step_executor_with_failing_emitter(context): """Create a PlanExecutor whose emitter.emit() raises.""" failing_emitter = MagicMock(name="failing_emitter") failing_emitter.emit.side_effect = RuntimeError("emit boom") context.executor = PlanExecutor( lifecycle_service=_make_lifecycle_stub(), metrics_emitter=failing_emitter, ) @when("I trigger metric emission that causes an exception") def step_trigger_failing_metric_emission(context): """Call _try_emit_metric knowing the emitter will fail.""" context.emit_error = None try: context.executor._try_emit_metric( key=OperationalMetricKey.PLAN_DURATION_MS, plan_id=_VALID_PLAN_ID, value=99.0, ) except Exception as exc: context.emit_error = exc @then("no exception should propagate from metric emission") def step_verify_no_emission_exception(context): """Verify the exception was swallowed (best-effort emission).""" assert context.emit_error is None, f"Expected no error, got {context.emit_error!r}" # --------------------------------------------------------------------------- # _try_create_checkpoint returns None when sandbox is None (line 384) # --------------------------------------------------------------------------- @given("a PlanExecutor with a checkpoint manager but no sandbox source") def step_executor_with_checkpoint_manager_no_sandbox(context): """Create a PlanExecutor with a checkpoint manager but no sandbox_root and no execution_context, so _resolve_sandbox_for_checkpoint returns None. """ mock_cp_manager = MagicMock(spec=CheckpointManager, name="cp_manager") context.executor = PlanExecutor( lifecycle_service=_make_lifecycle_stub(), checkpoint_manager=mock_cp_manager, sandbox_root=None, execution_context=None, ) @when("I attempt to create a plan executor checkpoint") def step_attempt_create_checkpoint(context): """Call _try_create_checkpoint on the executor.""" context.checkpoint_result = context.executor._try_create_checkpoint( plan_id=_VALID_PLAN_ID, phase="pre_execute", ) @then("the checkpoint result should be None") def step_verify_checkpoint_none(context): """Verify the checkpoint result is None.""" assert context.checkpoint_result is None # --------------------------------------------------------------------------- # _try_rollback_to_last_checkpoint returns False (line 422) # --------------------------------------------------------------------------- @when("I attempt to rollback to the last checkpoint") def step_attempt_rollback(context): """Call _try_rollback_to_last_checkpoint on the executor.""" context.rollback_result = context.executor._try_rollback_to_last_checkpoint( plan_id=_VALID_PLAN_ID, ) @then("the rollback result should be False") def step_verify_rollback_false(context): """Verify the rollback result is False.""" assert context.rollback_result is False # --------------------------------------------------------------------------- # _resolve_sandbox_for_checkpoint via execution_context (line 458) # --------------------------------------------------------------------------- @given("a PlanExecutor with an execution context that has a sandbox manager") def step_executor_with_execution_context_sandbox(context): """Create a PlanExecutor with an execution_context whose sandbox_manager.list_sandboxes returns a sandbox object. """ mock_sandbox = SimpleNamespace( sandbox_id="sandbox-from-ctx", context=SimpleNamespace(sandbox_path="/tmp/sandbox-ctx"), ) mock_sandbox_mgr = MagicMock(name="sandbox_manager") mock_sandbox_mgr.list_sandboxes.return_value = [mock_sandbox] mock_exec_ctx = SimpleNamespace(sandbox_manager=mock_sandbox_mgr) context.executor = PlanExecutor( lifecycle_service=_make_lifecycle_stub(), execution_context=mock_exec_ctx, ) context.expected_sandbox = mock_sandbox @when("I resolve a sandbox for checkpointing") def step_resolve_sandbox(context): """Call _resolve_sandbox_for_checkpoint.""" context.resolved_sandbox = context.executor._resolve_sandbox_for_checkpoint( plan_id=_VALID_PLAN_ID, ) @then("the resolved sandbox should come from the execution context") def step_verify_resolved_sandbox(context): """Verify the returned sandbox matches the one from the execution context.""" assert context.resolved_sandbox is context.expected_sandbox assert context.resolved_sandbox.sandbox_id == "sandbox-from-ctx" # --------------------------------------------------------------------------- # _enforce_guardrails raises PlanError on step limit (lines 598-599) # --------------------------------------------------------------------------- @given("a PlanExecutor with a guardrail service that blocks step limit") def step_executor_guardrail_blocks_step_limit(context): """Create a PlanExecutor with a guardrail service that: - get_guardrails returns guardrails with start_time already set - check_wall_clock returns True (ok) - check_step_limit returns False (blocked) """ guardrail_svc = MagicMock(spec=AutonomyGuardrailService, name="guardrail_svc") mock_guardrails = MagicMock(spec=AutonomyGuardrails) mock_guardrails.start_time = "2025-01-01T00:00:00Z" mock_guardrails.step_count = 100 guardrail_svc.get_guardrails.return_value = mock_guardrails guardrail_svc.check_wall_clock.return_value = True guardrail_svc.check_step_limit.return_value = False context.executor = PlanExecutor( lifecycle_service=_make_lifecycle_stub(), guardrail_service=guardrail_svc, ) @when("I enforce guardrails for a plan") def step_enforce_guardrails(context): """Call _enforce_guardrails, expecting a PlanError.""" context.guardrail_error = None try: context.executor._enforce_guardrails("plan-step-limit-test") except PlanError as exc: context.guardrail_error = exc @then("a PlanError about step limit should be raised") def step_verify_step_limit_error(context): """Verify that a PlanError mentioning step limit was raised.""" assert context.guardrail_error is not None, "Expected PlanError but none raised" assert "step limit already reached" in str(context.guardrail_error).lower(), ( f"Unexpected error message: {context.guardrail_error}" ) # --------------------------------------------------------------------------- # _enforce_guardrails_per_step returns early when guardrails is None (line 612) # --------------------------------------------------------------------------- @given("a PlanExecutor with a guardrail service that returns no guardrails") def step_executor_guardrail_returns_none(context): """Create a PlanExecutor with a guardrail service that returns None from get_guardrails (no guardrails configured for this plan). """ guardrail_svc = MagicMock(spec=AutonomyGuardrailService, name="guardrail_svc_none") guardrail_svc.get_guardrails.return_value = None context.executor = PlanExecutor( lifecycle_service=_make_lifecycle_stub(), guardrail_service=guardrail_svc, ) @when("I enforce per-step guardrails") def step_enforce_per_step_guardrails(context): """Call _enforce_guardrails_per_step with guardrails=None.""" context.per_step_error = None try: context.executor._enforce_guardrails_per_step("plan-no-guardrails") except Exception as exc: context.per_step_error = exc @then("no exception should be raised from per-step enforcement") def step_verify_no_per_step_error(context): """Verify no exception was raised (early return).""" assert context.per_step_error is None, ( f"Expected no error, got {context.per_step_error!r}" ) # --------------------------------------------------------------------------- # _enforce_guardrails_per_step raises PlanError on wall-clock (line 619) # --------------------------------------------------------------------------- @given("a PlanExecutor with a guardrail service that blocks wall-clock per step") def step_executor_guardrail_blocks_wallclock_per_step(context): """Create a PlanExecutor with a guardrail service that: - get_guardrails returns valid guardrails - check_step_limit returns True (step OK) - check_wall_clock returns False (wall-clock exceeded) """ guardrail_svc = MagicMock(spec=AutonomyGuardrailService, name="guardrail_svc_wc") mock_guardrails = MagicMock(spec=AutonomyGuardrails) mock_guardrails.step_count = 5 guardrail_svc.get_guardrails.return_value = mock_guardrails guardrail_svc.check_step_limit.return_value = True guardrail_svc.check_wall_clock.return_value = False context.executor = PlanExecutor( lifecycle_service=_make_lifecycle_stub(), guardrail_service=guardrail_svc, ) @when("I enforce per-step guardrails expecting wall-clock error") def step_enforce_per_step_wallclock(context): """Call _enforce_guardrails_per_step, expecting a wall-clock PlanError.""" context.per_step_wc_error = None try: context.executor._enforce_guardrails_per_step("plan-wc-exceeded") except PlanError as exc: context.per_step_wc_error = exc @then("a PlanError about wall-clock limit should be raised from per-step") def step_verify_wallclock_per_step_error(context): """Verify that a PlanError about wall-clock was raised.""" assert context.per_step_wc_error is not None, "Expected PlanError but none raised" assert "wall-clock limit exceeded" in str(context.per_step_wc_error).lower(), ( f"Unexpected error message: {context.per_step_wc_error}" )