"""Step definitions for autonomy guardrail service coverage-boost scenarios. Targets uncovered lines in autonomy_guardrail_service.py: 173, 191-194 — check_tool_budget blocked path 211 — check_wall_clock with no guardrails 259 — check_actor_tool_calls with no guardrails 266, 285-288 — check_actor_tool_calls blocked path 310 — check_retries_per_failure with no guardrails 371-372 — check_confirmation_required granted path 404-405 — record_enforcement public method 517 — _record_enforcement_locked creates new trail """ from __future__ import annotations from behave import given, then, when from behave.runner import Context from cleveragents.application.services.autonomy_guardrail_service import ( AutonomyGuardrailService, ) from cleveragents.domain.models.core.autonomy_guardrails import ( ActorLimits, AutonomyGuardrails, GuardrailAuditEntry, GuardrailEventType, GuardrailResult, ) # ---- Shared setup ---- @given("a fresh guardrail service for coverage") def step_given_fresh_service(context: Context) -> None: context.cov_service = AutonomyGuardrailService() @given('a guardrail service with tool_budget {budget:g} for plan "{plan_id}"') def step_given_service_with_budget( context: Context, budget: float, plan_id: str ) -> None: context.cov_service = AutonomyGuardrailService() context.cov_service.configure_guardrails( plan_id, AutonomyGuardrails(tool_budget=budget), ) @given('a guardrail service with actor tool-call limit {limit:d} for plan "{plan_id}"') def step_given_service_with_actor_limit( context: Context, limit: int, plan_id: str ) -> None: context.cov_service = AutonomyGuardrailService() context.cov_service.configure_guardrails( plan_id, AutonomyGuardrails( actor_limits=ActorLimits(max_tool_calls_per_invocation=limit), ), ) @given('a guardrail service with confirmations "{confirmations}" for plan "{plan_id}"') def step_given_service_with_confirmations( context: Context, confirmations: str, plan_id: str ) -> None: context.cov_service = AutonomyGuardrailService() context.cov_service.configure_guardrails( plan_id, AutonomyGuardrails( required_confirmations=confirmations.split(","), ), ) # ---- check_tool_budget: blocked path (lines 173, 191-194) ---- @when( 'I check tool budget via the coverage service for plan "{plan_id}" with cost {cost:g}' ) def step_cov_check_budget(context: Context, plan_id: str, cost: float) -> None: context.cov_budget_result = context.cov_service.check_tool_budget(plan_id, cost) @when('I successfully check tool budget for plan "{plan_id}" with cost {cost:g}') def step_cov_check_budget_success(context: Context, plan_id: str, cost: float) -> None: result = context.cov_service.check_tool_budget(plan_id, cost) assert result is True, f"Expected budget check to succeed, got {result}" @then("the coverage budget check should return False") def step_cov_budget_false(context: Context) -> None: assert context.cov_budget_result is False @then( 'the audit trail for coverage plan "{plan_id}" should record a budget_blocked event' ) def step_cov_trail_budget_blocked(context: Context, plan_id: str) -> None: trail = context.cov_service.get_audit_trail(plan_id) blocked = [ e for e in trail.entries if e.event_type == GuardrailEventType.BUDGET_BLOCKED ] assert len(blocked) >= 1, ( f"Expected at least one BUDGET_BLOCKED event, found {len(blocked)}" ) # ---- check_wall_clock: no guardrails (line 211) ---- @when('I check wall-clock via the coverage service for plan "{plan_id}"') def step_cov_check_wall_clock(context: Context, plan_id: str) -> None: context.cov_wall_clock_result = context.cov_service.check_wall_clock(plan_id) @then("the coverage wall-clock check should return True") def step_cov_wall_clock_true(context: Context) -> None: assert context.cov_wall_clock_result is True # ---- check_actor_tool_calls: no guardrails (line 259) / blocked (266, 285-288) ---- @when( 'I check actor tool calls via the coverage service for plan "{plan_id}" with {calls:d} calls' ) def step_cov_check_actor(context: Context, plan_id: str, calls: int) -> None: context.cov_actor_result = context.cov_service.check_actor_tool_calls( plan_id, calls ) @then("the coverage actor check should return True") def step_cov_actor_true(context: Context) -> None: assert context.cov_actor_result is True @then("the coverage actor check should return False") def step_cov_actor_false(context: Context) -> None: assert context.cov_actor_result is False @then( 'the audit trail for coverage plan "{plan_id}" should record an actor_limit_blocked event' ) def step_cov_trail_actor_blocked(context: Context, plan_id: str) -> None: trail = context.cov_service.get_audit_trail(plan_id) blocked = [ e for e in trail.entries if e.event_type == GuardrailEventType.ACTOR_LIMIT_BLOCKED ] assert len(blocked) >= 1, ( f"Expected at least one ACTOR_LIMIT_BLOCKED event, found {len(blocked)}" ) # ---- check_retries_per_failure: no guardrails (line 310) ---- @when( 'I check retries via the coverage service for plan "{plan_id}" with {retries:d} retries' ) def step_cov_check_retries(context: Context, plan_id: str, retries: int) -> None: context.cov_retry_result = context.cov_service.check_retries_per_failure( plan_id, retries ) @then("the coverage retry check should return True") def step_cov_retry_true(context: Context) -> None: assert context.cov_retry_result is True # ---- check_confirmation_required: granted path (lines 371-372) ---- @when( 'I check confirmation via the coverage service for plan "{plan_id}" operation "{operation}"' ) def step_cov_check_confirmation(context: Context, plan_id: str, operation: str) -> None: context.cov_confirm_result = context.cov_service.check_confirmation_required( plan_id, operation ) @then("the coverage confirmation check should return False") def step_cov_confirm_false(context: Context) -> None: assert context.cov_confirm_result is False @then( 'the audit trail for coverage plan "{plan_id}" should record a confirmation_granted event' ) def step_cov_trail_confirm_granted(context: Context, plan_id: str) -> None: trail = context.cov_service.get_audit_trail(plan_id) granted = [ e for e in trail.entries if e.event_type == GuardrailEventType.CONFIRMATION_GRANTED ] assert len(granted) >= 1, ( f"Expected at least one CONFIRMATION_GRANTED event, found {len(granted)}" ) # ---- record_enforcement public method (lines 404-405, 517) ---- @when('I call record_enforcement for plan "{plan_id}" with a step_allowed entry') def step_cov_record_enforcement_step(context: Context, plan_id: str) -> None: entry = GuardrailAuditEntry( event_type=GuardrailEventType.STEP_ALLOWED, guard_name="step_limit", result=GuardrailResult.ALLOWED, ) context.cov_service.record_enforcement(plan_id, entry) @when('I call record_enforcement for plan "{plan_id}" with a budget_allowed entry') def step_cov_record_enforcement_budget(context: Context, plan_id: str) -> None: entry = GuardrailAuditEntry( event_type=GuardrailEventType.BUDGET_ALLOWED, guard_name="tool_budget", result=GuardrailResult.ALLOWED, ) context.cov_service.record_enforcement(plan_id, entry) @then('the audit trail for coverage plan "{plan_id}" should have {count:d} entries') def step_cov_trail_count(context: Context, plan_id: str, count: int) -> None: trail = context.cov_service.get_audit_trail(plan_id) assert len(trail.entries) == count, ( f"Expected {count} entries, found {len(trail.entries)}" ) @then('the first entry for coverage plan "{plan_id}" should be budget_allowed') def step_cov_first_entry_budget_allowed(context: Context, plan_id: str) -> None: trail = context.cov_service.get_audit_trail(plan_id) assert len(trail.entries) > 0, "Audit trail is empty" first = trail.entries[0] assert first.event_type == GuardrailEventType.BUDGET_ALLOWED, ( f"Expected BUDGET_ALLOWED, got {first.event_type}" )