"""Step definitions for M6 autonomy guardrails, profiles, and guard enforcement. Split from ``m6_autonomy_acceptance_steps.py`` to stay under the project's 500-line guideline. All step names keep the ``m6 smoke`` prefix to avoid ``AmbiguousStep`` conflicts. """ from __future__ import annotations import json import tempfile from pathlib import Path import yaml from behave import given, then, when from behave.runner import Context from cleveragents.application.services.automation_profile_service import ( AutomationProfileService, ) from cleveragents.domain.models.core.automation_guard import ( AutomationGuard, ) from cleveragents.domain.models.core.automation_profile import ( BUILTIN_PROFILES, AutomationProfile, ) _FIXTURES_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "m6" # ----------------------------------------------------------------------- # Fixture loading — guardrails and profiles # ----------------------------------------------------------------------- @when("I m6 smoke load the autonomy guardrails fixture") def step_m6_smoke_load_guardrails(context: Context) -> None: with open(_FIXTURES_DIR / "autonomy_guardrails.json") as f: context.m6_guardrails_data = json.load(f) @then("the m6 smoke guardrails fixture should have a denylist entry") def step_m6_smoke_guardrails_denylist(context: Context) -> None: names = [f["name"] for f in context.m6_guardrails_data["fixtures"]] assert "denylist_guard" in names @then("the m6 smoke guardrails fixture should have an allowlist entry") def step_m6_smoke_guardrails_allowlist(context: Context) -> None: names = [f["name"] for f in context.m6_guardrails_data["fixtures"]] assert "allowlist_guard" in names @then("the m6 smoke guardrails fixture should have a cost budget entry") def step_m6_smoke_guardrails_budget(context: Context) -> None: names = [f["name"] for f in context.m6_guardrails_data["fixtures"]] assert "cost_budget_guard" in names @when("I m6 smoke load the automation profiles fixture") def step_m6_smoke_load_profiles_fixture(context: Context) -> None: with open(_FIXTURES_DIR / "automation_profiles.json") as f: context.m6_profiles_data = json.load(f) @then("the m6 smoke profiles fixture should list all 8 built-in names") def step_m6_smoke_profiles_builtin_count(context: Context) -> None: entry = next( f for f in context.m6_profiles_data["fixtures"] if f["name"] == "builtin_profiles" ) assert len(entry["expected_names"]) == 8 @then("the m6 smoke profiles fixture should have a custom profile entry") def step_m6_smoke_profiles_custom(context: Context) -> None: names = [f["name"] for f in context.m6_profiles_data["fixtures"]] assert "custom_profile" in names # ----------------------------------------------------------------------- # Automation profiles built-in # ----------------------------------------------------------------------- @when("I m6 smoke list all built-in profiles") def step_m6_smoke_list_builtins(context: Context) -> None: context.m6_profiles_list = list(BUILTIN_PROFILES.keys()) @then("the m6 smoke profile count should be {count:d}") def step_m6_smoke_profile_count(context: Context, count: int) -> None: assert len(context.m6_profiles_list) == count @then('the m6 smoke profiles should include "{name}"') def step_m6_smoke_profiles_include(context: Context, name: str) -> None: assert name in context.m6_profiles_list @when('I m6 smoke load built-in profile "{name}"') def step_m6_smoke_load_builtin(context: Context, name: str) -> None: context.m6_profile = BUILTIN_PROFILES[name] @then("the m6 smoke profile decompose_task should be {value:g}") def step_m6_smoke_profile_decompose_task( context: Context, value: float, ) -> None: assert context.m6_profile.decompose_task == value @then("the m6 smoke profile create_tool should be {value:g}") def step_m6_smoke_profile_create_tool( context: Context, value: float, ) -> None: assert context.m6_profile.create_tool == value @then("the m6 smoke profile select_tool should be {value:g}") def step_m6_smoke_profile_select_tool( context: Context, value: float, ) -> None: assert context.m6_profile.select_tool == value @then("the m6 smoke profile require_sandbox should be true") def step_m6_smoke_profile_sandbox_true(context: Context) -> None: assert context.m6_profile.safety.require_sandbox is True @then("the m6 smoke profile require_sandbox should be false") def step_m6_smoke_profile_sandbox_false(context: Context) -> None: assert context.m6_profile.safety.require_sandbox is False @then("the m6 smoke profile allow_unsafe_tools should be true") def step_m6_smoke_profile_unsafe_true(context: Context) -> None: assert context.m6_profile.safety.allow_unsafe_tools is True # ----------------------------------------------------------------------- # Automation profile creation and validation # ----------------------------------------------------------------------- @when('I m6 smoke create a profile named "{name}" with select_tool {value:g}') def step_m6_smoke_create_profile( context: Context, name: str, value: float, ) -> None: context.m6_profile = AutomationProfile(name=name, select_tool=value) @then('the m6 smoke created profile name should be "{name}"') def step_m6_smoke_created_name(context: Context, name: str) -> None: assert context.m6_profile.name == name @then("the m6 smoke created profile select_tool should be {value:g}") def step_m6_smoke_created_select_tool(context: Context, value: float) -> None: assert context.m6_profile.select_tool == value @when('I m6 smoke create a profile with invalid name "{name}"') def step_m6_smoke_invalid_name(context: Context, name: str) -> None: try: AutomationProfile(name=name) context.m6_error = None except ValueError as exc: context.m6_error = exc @then("the m6 smoke creation should raise ValueError") def step_m6_smoke_error_value(context: Context) -> None: assert isinstance(context.m6_error, (ValueError,)) @when("I m6 smoke create a profile with decompose_task {value:g}") def step_m6_smoke_invalid_threshold(context: Context, value: float) -> None: try: AutomationProfile(name="test-bad-threshold", decompose_task=value) context.m6_error = None except ValueError as exc: context.m6_error = exc @given("a m6 smoke temporary profile YAML file") def step_m6_smoke_yaml_file(context: Context) -> None: config = { "name": "test-yaml-profile", "description": "A profile loaded from YAML", "decompose_task": 0.5, "create_tool": 0.5, "select_tool": 1.0, } with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False, ) as tmp: yaml.dump(config, tmp) context.m6_yaml_path = tmp.name @when("I m6 smoke load profile from the temp YAML") def step_m6_smoke_load_yaml(context: Context) -> None: context.m6_profile = AutomationProfile.from_yaml(context.m6_yaml_path) @then('the m6 smoke loaded profile name should be "{name}"') def step_m6_smoke_loaded_name(context: Context, name: str) -> None: assert context.m6_profile.name == name # ----------------------------------------------------------------------- # Guard enforcement # ----------------------------------------------------------------------- @given('a m6 smoke profile with denylist guard for "{tool}"') def step_m6_smoke_denylist_guard(context: Context, tool: str) -> None: guard = AutomationGuard(tool_denylist=[tool]) context.m6_profile = AutomationProfile(name="test-deny", guards=guard) @when('I m6 smoke check guard for tool "{tool}"') def step_m6_smoke_check_guard(context: Context, tool: str) -> None: context.m6_guard_result = context.m6_profile.check_guard(tool_name=tool) @then("the m6 smoke guard result should not be allowed") def step_m6_smoke_guard_not_allowed(context: Context) -> None: assert context.m6_guard_result.allowed is False @then("the m6 smoke guard result should be allowed") def step_m6_smoke_guard_allowed(context: Context) -> None: assert context.m6_guard_result.allowed is True @then('the m6 smoke guard reason should contain "{text}"') def step_m6_smoke_guard_reason(context: Context, text: str) -> None: assert context.m6_guard_result.reason is not None assert text in context.m6_guard_result.reason @given('a m6 smoke profile with allowlist guard for "{t1}" and "{t2}"') def step_m6_smoke_allowlist_guard( context: Context, t1: str, t2: str, ) -> None: guard = AutomationGuard(tool_allowlist=[t1, t2]) context.m6_profile = AutomationProfile(name="test-allow", guards=guard) @given("a m6 smoke profile with max {limit:d} tool calls per step") def step_m6_smoke_max_calls_guard(context: Context, limit: int) -> None: guard = AutomationGuard(max_tool_calls_per_step=limit) context.m6_profile = AutomationProfile(name="test-maxcall", guards=guard) @when('I m6 smoke check guard for tool "{tool}" with {calls:d} calls so far') def step_m6_smoke_check_guard_calls( context: Context, tool: str, calls: int, ) -> None: context.m6_guard_result = context.m6_profile.check_guard( tool_name=tool, calls_so_far=calls, ) @given("a m6 smoke profile with max cost {budget:g}") def step_m6_smoke_cost_guard(context: Context, budget: float) -> None: guard = AutomationGuard(max_total_cost=budget) context.m6_profile = AutomationProfile(name="test-budget", guards=guard) @when('I m6 smoke check guard for tool "{tool}" with cost {cost:g}') def step_m6_smoke_check_guard_cost( context: Context, tool: str, cost: float, ) -> None: context.m6_guard_result = context.m6_profile.check_guard( tool_name=tool, cost_so_far=cost, ) @given("a m6 smoke profile with write approval required") def step_m6_smoke_write_guard(context: Context) -> None: guard = AutomationGuard(require_approval_for_writes=True) context.m6_profile = AutomationProfile( name="test-writeapproval", guards=guard, ) @when('I m6 smoke check guard for tool "{tool}" as a write operation') def step_m6_smoke_check_guard_write(context: Context, tool: str) -> None: context.m6_guard_result = context.m6_profile.check_guard( tool_name=tool, is_write=True, ) @given("a m6 smoke profile with apply approval required") def step_m6_smoke_apply_guard(context: Context) -> None: guard = AutomationGuard(require_approval_for_apply=True) context.m6_profile = AutomationProfile( name="test-applyapproval", guards=guard, ) @given("a m6 smoke profile with no guards") def step_m6_smoke_no_guards(context: Context) -> None: context.m6_profile = AutomationProfile(name="test-noguards") # ----------------------------------------------------------------------- # Profile service resolution # ----------------------------------------------------------------------- @given("a m6 smoke automation profile service") def step_m6_smoke_profile_service(context: Context) -> None: context.m6_profile_service = AutomationProfileService() @when( "I m6 smoke resolve profile with plan " '"{plan}" action "{action}" project "{project}"' ) def step_m6_smoke_resolve( context: Context, plan: str, action: str, project: str, ) -> None: context.m6_profile = context.m6_profile_service.resolve_profile( plan_profile=plan if plan != "null" else None, action_profile=action if action != "null" else None, project_profile=project if project != "null" else None, ) @when('I m6 smoke resolve profile with plan null action "{action}" project "{project}"') def step_m6_smoke_resolve_no_plan( context: Context, action: str, project: str, ) -> None: context.m6_profile = context.m6_profile_service.resolve_profile( plan_profile=None, action_profile=action if action != "null" else None, project_profile=project if project != "null" else None, ) @when("I m6 smoke resolve profile with plan null action null project null") def step_m6_smoke_resolve_all_null(context: Context) -> None: context.m6_profile = context.m6_profile_service.resolve_profile( plan_profile=None, action_profile=None, project_profile=None, ) @then('the m6 smoke resolved profile name should be "{name}"') def step_m6_smoke_resolved_name(context: Context, name: str) -> None: assert context.m6_profile.name == name # ----------------------------------------------------------------------- # Profile service guard evaluation # ----------------------------------------------------------------------- @when('I m6 smoke evaluate guard for profile "{profile}" and tool "{tool}"') def step_m6_smoke_evaluate_guard( context: Context, profile: str, tool: str, ) -> None: context.m6_guard_result = context.m6_profile_service.evaluate_guard( profile_name=profile, tool_name=tool, )