From 348c230bc570e233b04fe35463030957660d00dc Mon Sep 17 00:00:00 2001 From: "Brent E. Edwards" Date: Fri, 27 Feb 2026 20:31:15 +0000 Subject: [PATCH 1/2] test(e2e): add M6 autonomy acceptance suite Add comprehensive M6 autonomy acceptance test suites covering the ACP local-mode facade, autonomy guardrails, automation profile resolution, event queue pub/sub, HTTP transport stub, and version negotiation. Behave suite (52 scenarios): - ACP facade dispatch for all 11 operations - Guard enforcement (denylist, allowlist, budget, call limit, write/apply) - Automation profile built-in validation and custom creation - Profile resolution precedence (plan > action > project > global) - Event queue lifecycle (publish, subscribe, unsubscribe, close) - HTTP transport stub rejection in local mode - ACP version negotiation (accept/reject) - Model validation (AcpRequest, AcpResponse, AcpEvent, AcpErrorDetail) Robot integration suite (11 tests): - Facade session/plan lifecycle, unknown operation error - Event queue publish/subscribe, transport stub, version negotiation - Guard denylist/budget enforcement, profile resolution - Fixture loading, full end-to-end flow ASV benchmarks (5 suites): - Facade dispatch, guard evaluation, profile resolution - Event queue operations, fixture loading Fixtures: acp_facade_flows.json, autonomy_guardrails.json, automation_profiles.json Closes #211 --- benchmarks/m6_acceptance_bench.py | 231 ++++++ docs/development/testing.md | 114 +++ features/fixtures/m6/acp_facade_flows.json | 42 + features/fixtures/m6/automation_profiles.json | 56 ++ features/fixtures/m6/autonomy_guardrails.json | 83 ++ features/m6_autonomy_acceptance.feature | 309 +++++++ .../steps/m6_autonomy_acceptance_steps.py | 774 ++++++++++++++++++ robot/helper_m6_autonomy_acceptance.py | 356 ++++++++ robot/m6_autonomy_acceptance.robot | 97 +++ 9 files changed, 2062 insertions(+) create mode 100644 benchmarks/m6_acceptance_bench.py create mode 100644 features/fixtures/m6/acp_facade_flows.json create mode 100644 features/fixtures/m6/automation_profiles.json create mode 100644 features/fixtures/m6/autonomy_guardrails.json create mode 100644 features/m6_autonomy_acceptance.feature create mode 100644 features/steps/m6_autonomy_acceptance_steps.py create mode 100644 robot/helper_m6_autonomy_acceptance.py create mode 100644 robot/m6_autonomy_acceptance.robot diff --git a/benchmarks/m6_acceptance_bench.py b/benchmarks/m6_acceptance_bench.py new file mode 100644 index 000000000..0640daf1c --- /dev/null +++ b/benchmarks/m6_acceptance_bench.py @@ -0,0 +1,231 @@ +"""ASV benchmarks for M6 autonomy acceptance suite runtime. + +Measures the performance of: +- ACP local facade dispatch operations +- Automation guard evaluation +- AutomationProfileService resolution precedence +- ACP event queue publish/subscribe +- Fixture loading overhead +""" + +from __future__ import annotations + +import importlib +import json +import sys +from pathlib import Path + +# Ensure the local *source* tree is importable even when ASV has an +# older build of the package installed. +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +import cleveragents # noqa: E402 + +importlib.reload(cleveragents) + +from cleveragents.acp.events import AcpEventQueue # noqa: E402 +from cleveragents.acp.facade import AcpLocalFacade # noqa: E402 +from cleveragents.acp.models import AcpEvent, AcpRequest # noqa: E402 +from cleveragents.acp.versioning import AcpVersionNegotiator # noqa: E402 +from cleveragents.application.services.automation_profile_service import ( # noqa: E402 + AutomationProfileService, +) +from cleveragents.domain.models.core.automation_guard import ( # noqa: E402 + AutomationGuard, +) +from cleveragents.domain.models.core.automation_profile import ( # noqa: E402 + AutomationProfile, +) + +_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m6" + + +class M6FacadeDispatchSuite: + """Benchmark ACP local facade dispatch operations.""" + + def setup(self) -> None: + self._facade = AcpLocalFacade() + + def time_session_create(self) -> None: + """Benchmark session.create dispatch.""" + self._facade.dispatch(AcpRequest(operation="session.create", params={})) + + def time_plan_create(self) -> None: + """Benchmark plan.create dispatch.""" + self._facade.dispatch(AcpRequest(operation="plan.create", params={})) + + def time_plan_execute(self) -> None: + """Benchmark plan.execute dispatch.""" + self._facade.dispatch( + AcpRequest( + operation="plan.execute", + params={"plan_id": "01M6SM0KE00000000000000001"}, + ) + ) + + def time_list_operations(self) -> None: + """Benchmark list_operations call.""" + self._facade.list_operations() + + +class M6GuardEvaluationSuite: + """Benchmark automation guard evaluation.""" + + def setup(self) -> None: + self._profile_denylist = AutomationProfile( + name="bench-deny", + guards=AutomationGuard( + tool_denylist=["rm_rf", "drop_database", "format_disk"], + ), + ) + self._profile_allowlist = AutomationProfile( + name="bench-allow", + guards=AutomationGuard( + tool_allowlist=["read_file", "search", "list_dir"], + ), + ) + self._profile_budget = AutomationProfile( + name="bench-budget", + guards=AutomationGuard( + max_total_cost=100.0, + max_tool_calls_per_step=10, + ), + ) + self._profile_no_guards = AutomationProfile(name="bench-noguard") + + def time_denylist_check_allowed(self) -> None: + """Benchmark denylist check for allowed tool.""" + self._profile_denylist.check_guard(tool_name="read_file") + + def time_denylist_check_denied(self) -> None: + """Benchmark denylist check for denied tool.""" + self._profile_denylist.check_guard(tool_name="rm_rf") + + def time_allowlist_check_allowed(self) -> None: + """Benchmark allowlist check for allowed tool.""" + self._profile_allowlist.check_guard(tool_name="read_file") + + def time_allowlist_check_denied(self) -> None: + """Benchmark allowlist check for unlisted tool.""" + self._profile_allowlist.check_guard(tool_name="write_file") + + def time_budget_check_under(self) -> None: + """Benchmark budget check under limit.""" + self._profile_budget.check_guard( + tool_name="llm_call", + cost_so_far=10.0, + calls_so_far=3, + ) + + def time_no_guards_check(self) -> None: + """Benchmark check with no guards configured.""" + self._profile_no_guards.check_guard(tool_name="anything") + + +class M6ProfileResolutionSuite: + """Benchmark AutomationProfileService resolution.""" + + def setup(self) -> None: + self._service = AutomationProfileService() + + def time_resolve_plan_level(self) -> None: + """Benchmark resolution with plan-level override.""" + self._service.resolve_profile( + plan_profile="ci", + action_profile="auto", + project_profile="manual", + ) + + def time_resolve_action_level(self) -> None: + """Benchmark resolution with action-level override.""" + self._service.resolve_profile( + plan_profile=None, + action_profile="auto", + project_profile="manual", + ) + + def time_resolve_global_default(self) -> None: + """Benchmark resolution falling back to global default.""" + self._service.resolve_profile( + plan_profile=None, + action_profile=None, + project_profile=None, + ) + + def time_get_builtin_profile(self) -> None: + """Benchmark looking up a built-in profile.""" + self._service.get_profile("manual") + + def time_list_profiles(self) -> None: + """Benchmark listing all profiles.""" + self._service.list_profiles() + + def time_evaluate_guard(self) -> None: + """Benchmark guard evaluation through service.""" + self._service.evaluate_guard( + profile_name="manual", + tool_name="read_file", + ) + + def time_version_negotiation(self) -> None: + """Benchmark ACP version negotiation.""" + negotiator = AcpVersionNegotiator() + negotiator.negotiate("1.0") + + +class M6EventQueueSuite: + """Benchmark ACP event queue operations.""" + + def setup(self) -> None: + self._queue = AcpEventQueue() + + def teardown(self) -> None: + if not self._queue.is_closed: + self._queue.close() + + def time_publish_event(self) -> None: + """Benchmark publishing a single event.""" + self._queue.publish(AcpEvent(event_type="plan.progress", data={"step": 1})) + + def time_subscribe_and_publish(self) -> None: + """Benchmark subscribe + publish cycle.""" + q = AcpEventQueue() + sub_id = q.subscribe_local(lambda _e: None) + q.publish(AcpEvent(event_type="test.event", data={})) + q.unsubscribe(sub_id) + q.close() + + def time_get_events(self) -> None: + """Benchmark retrieving events from queue.""" + self._queue.get_events(limit=50) + + +class M6FixtureLoadSuite: + """Benchmark loading M6 fixture files.""" + + def time_load_acp_facade_flows(self) -> None: + """Benchmark loading acp_facade_flows.json.""" + with open(_FIXTURES_DIR / "acp_facade_flows.json") as f: + json.load(f) + + def time_load_autonomy_guardrails(self) -> None: + """Benchmark loading autonomy_guardrails.json.""" + with open(_FIXTURES_DIR / "autonomy_guardrails.json") as f: + json.load(f) + + def time_load_automation_profiles(self) -> None: + """Benchmark loading automation_profiles.json.""" + with open(_FIXTURES_DIR / "automation_profiles.json") as f: + json.load(f) + + def time_load_all_fixtures(self) -> None: + """Benchmark loading all M6 fixture files.""" + for fname in ( + "acp_facade_flows.json", + "autonomy_guardrails.json", + "automation_profiles.json", + ): + with open(_FIXTURES_DIR / fname) as f: + json.load(f) diff --git a/docs/development/testing.md b/docs/development/testing.md index aa4e0739d..0187d5e74 100644 --- a/docs/development/testing.md +++ b/docs/development/testing.md @@ -1295,3 +1295,117 @@ nox -s benchmark - **Coverage drops**: The M4 smoke tests cover correction CLI commands, SubplanFailureHandler logic, and fixture loading. Check `build/htmlcov/index.html` for uncovered lines in correction/plan CLI code. + +## M6 Autonomy Acceptance Suite + +E2E fixtures covering ACP facade flows and autonomy guardrails. The suite +exercises guardrail enforcement (max steps, tool budget, confirmations) and +audit trail persistence in integrated scenarios. + +### Overview + +The M6 acceptance suite validates the autonomy hardening layer: + +- **ACP Local Facade** — local-mode dispatch of all 11 supported operations + (session, plan, registry, context, event). +- **Autonomy Guardrails** — denylist/allowlist enforcement, cost budget caps, + tool call limits, write approval gates, apply approval gates. +- **Automation Profiles** — 8 built-in profiles, custom namespaced profiles, + YAML loading, threshold validation, and four-level resolution precedence + (plan > action > project > global). +- **ACP Event Queue** — publish/subscribe, local callbacks, unsubscribe, close + semantics, remote stub rejection. +- **ACP HTTP Transport Stub** — all mutating methods raise + `AcpNotAvailableError` in local mode. +- **ACP Version Negotiation** — accept supported versions, reject unsupported. +- **ACP Model Validation** — non-empty operation, valid status values, + non-empty event types, non-empty error codes. + +### Fixtures (`features/fixtures/m6/`) + +| File | Purpose | +|------|---------| +| `acp_facade_flows.json` | ACP operation flows (session, plan, registry, context, event) | +| `autonomy_guardrails.json` | Guard configurations (denylist, allowlist, budget, write/apply approval) | +| `automation_profiles.json` | Built-in profile assertions, custom profile config, resolution precedence cases | + +### Behave Suite: `features/m6_autonomy_acceptance.feature` + +- **Fixture loading** — 3 scenarios verifying JSON structure +- **ACP facade dispatch** — 10 scenarios covering all 11 operations +- **ACP error handling** — 2 scenarios (unknown operation, invalid request type) +- **Service registration** — 2 scenarios (register + list operations) +- **Event queue** — 5 scenarios (publish, subscribe, unsubscribe, close, remote reject) +- **HTTP transport stub** — 4 scenarios (send, connect, disconnect, is_connected) +- **Version negotiation** — 3 scenarios (accept, reject, is_supported) +- **Built-in profiles** — 3 scenarios (list, manual thresholds, full-auto thresholds) +- **Profile creation** — 4 scenarios (custom, invalid name, bad threshold, YAML load) +- **Guard enforcement** — 8 scenarios (denylist, allowlist, call limit, budget, write, apply, no guards) +- **Profile resolution** — 3 scenarios (plan precedence, action precedence, global default) +- **Service guard evaluation** — 1 scenario +- **Model validation** — 4 scenarios (AcpRequest, AcpResponse, AcpEvent, AcpErrorDetail) + +Step implementations: `features/steps/m6_autonomy_acceptance_steps.py` + +### Robot Suite: `robot/m6_autonomy_acceptance.robot` + +| Test Case | What it checks | +|-----------|---------------| +| M6 ACP Facade Session Lifecycle | session.create + session.close dispatch | +| M6 ACP Facade Plan Lifecycle | plan create/execute/status/diff/apply | +| M6 ACP Facade Unknown Operation Error | AcpOperationNotFoundError on bad op | +| M6 ACP Event Queue Publish Subscribe | publish, subscribe, unsubscribe, close | +| M6 ACP Transport Stub Rejects All | send/connect/disconnect raise AcpNotAvailableError | +| M6 ACP Version Negotiation | accept 1.0, reject 2.0 | +| M6 Guard Denylist Enforcement | denylist blocks denied tools | +| M6 Guard Budget Enforcement | cost budget and call limit guards | +| M6 Profile Resolution Precedence | plan > action > project > global | +| M6 Fixture Loading | all 3 fixture files load and validate | +| M6 Full Autonomy Acceptance Flow | end-to-end facade + guard + resolution | + +Helper: `robot/helper_m6_autonomy_acceptance.py` + +### ASV Benchmarks: `benchmarks/m6_acceptance_bench.py` + +| Suite | Methods | +|-------|---------| +| `M6FacadeDispatchSuite` | `time_session_create`, `time_plan_create`, `time_plan_execute`, `time_list_operations` | +| `M6GuardEvaluationSuite` | `time_denylist_check_allowed`, `time_denylist_check_denied`, `time_allowlist_check_allowed`, `time_allowlist_check_denied`, `time_budget_check_under`, `time_no_guards_check` | +| `M6ProfileResolutionSuite` | `time_resolve_plan_level`, `time_resolve_action_level`, `time_resolve_global_default`, `time_get_builtin_profile`, `time_list_profiles`, `time_evaluate_guard`, `time_version_negotiation` | +| `M6EventQueueSuite` | `time_publish_event`, `time_subscribe_and_publish`, `time_get_events` | +| `M6FixtureLoadSuite` | `time_load_acp_facade_flows`, `time_load_autonomy_guardrails`, `time_load_automation_profiles`, `time_load_all_fixtures` | + +### Running the M6 Autonomy Acceptance Suites + +```bash +# Behave only +nox -s behave -- features/m6_autonomy_acceptance.feature + +# Robot only +nox -s robot -- robot/m6_autonomy_acceptance.robot + +# ASV benchmarks +nox -s benchmark + +# Everything (default nox sessions) +nox +``` + +### Failure Triage Tips + +- **`AmbiguousStep` errors**: All M6 smoke steps are prefixed with `m6 smoke`. + If ambiguous, check that no other step file defines a conflicting pattern. +- **Fixture file not found**: Verify `features/fixtures/m6/` contains all three + fixture files (`acp_facade_flows.json`, `autonomy_guardrails.json`, + `automation_profiles.json`). +- **Robot `FAIL` sentinel**: Each Robot helper subcommand prints a detailed + `FAIL:` line with exit code and output when something goes wrong. +- **AcpNotAvailableError expected**: The transport stub and remote event + subscribe are *supposed* to raise this error. If a test fails here, it + likely means the stub was changed to not raise. +- **Guard evaluation failures**: Check that the `AutomationGuard` model fields + match the fixture data. Denylist/allowlist/budget fields are all optional. +- **Coverage drops**: The M6 acceptance tests cover the `acp/` package facade, + events, transport, versioning, errors, and models, plus + `automation_profile.py` and `automation_profile_service.py`. Check + `build/htmlcov/index.html` for uncovered lines. diff --git a/features/fixtures/m6/acp_facade_flows.json b/features/fixtures/m6/acp_facade_flows.json new file mode 100644 index 000000000..d88b3cb17 --- /dev/null +++ b/features/fixtures/m6/acp_facade_flows.json @@ -0,0 +1,42 @@ +{ + "description": "ACP local-mode facade operation flows for M6 autonomy acceptance tests", + "fixtures": [ + { + "name": "session_lifecycle", + "operations": [ + {"operation": "session.create", "params": {}, "expected_status": "ok", "expected_keys": ["session_id", "status"]}, + {"operation": "session.close", "params": {}, "expected_status": "ok", "expected_keys": ["status"]} + ] + }, + { + "name": "plan_lifecycle", + "operations": [ + {"operation": "plan.create", "params": {}, "expected_status": "ok", "expected_keys": ["plan_id", "status"]}, + {"operation": "plan.execute", "params": {"plan_id": "01M6SM0KE00000000000000001"}, "expected_status": "ok", "expected_keys": ["plan_id", "status"]}, + {"operation": "plan.status", "params": {"plan_id": "01M6SM0KE00000000000000001"}, "expected_status": "ok", "expected_keys": ["plan_id", "phase"]}, + {"operation": "plan.diff", "params": {"plan_id": "01M6SM0KE00000000000000001"}, "expected_status": "ok", "expected_keys": ["plan_id", "changes"]}, + {"operation": "plan.apply", "params": {"plan_id": "01M6SM0KE00000000000000001"}, "expected_status": "ok", "expected_keys": ["plan_id", "status"]} + ] + }, + { + "name": "registry_queries", + "operations": [ + {"operation": "registry.list_tools", "params": {}, "expected_status": "ok", "expected_keys": ["tools"]}, + {"operation": "registry.list_resources", "params": {}, "expected_status": "ok", "expected_keys": ["resources"]} + ] + }, + { + "name": "context_and_events", + "operations": [ + {"operation": "context.get", "params": {}, "expected_status": "ok", "expected_keys": ["context"]}, + {"operation": "event.subscribe", "params": {}, "expected_status": "ok", "expected_keys": ["subscription_id", "status"]} + ] + }, + { + "name": "unknown_operation", + "operations": [ + {"operation": "nonexistent.op", "params": {}, "expected_error": "AcpOperationNotFoundError"} + ] + } + ] +} diff --git a/features/fixtures/m6/automation_profiles.json b/features/fixtures/m6/automation_profiles.json new file mode 100644 index 000000000..693e7c06d --- /dev/null +++ b/features/fixtures/m6/automation_profiles.json @@ -0,0 +1,56 @@ +{ + "description": "Automation profile configurations for M6 acceptance tests", + "fixtures": [ + { + "name": "builtin_profiles", + "expected_names": ["manual", "review", "supervised", "cautious", "trusted", "auto", "ci", "full-auto"], + "manual_thresholds": { + "auto_strategize": 1.0, + "auto_execute": 1.0, + "auto_apply": 1.0, + "require_sandbox": true, + "require_checkpoints": true, + "allow_unsafe_tools": false + }, + "full_auto_thresholds": { + "auto_strategize": 0.0, + "auto_execute": 0.0, + "auto_apply": 0.0, + "require_sandbox": false, + "require_checkpoints": false, + "allow_unsafe_tools": true + } + }, + { + "name": "custom_profile", + "config": { + "name": "acme/strict", + "description": "Strict profile for ACME Corp", + "schema_version": "1.0", + "auto_strategize": 1.0, + "auto_execute": 1.0, + "auto_apply": 1.0, + "auto_decisions_strategize": 1.0, + "auto_decisions_execute": 1.0, + "auto_validation_fix": 1.0, + "auto_strategy_revision": 1.0, + "auto_reversion_from_apply": 1.0, + "auto_child_plans": 1.0, + "auto_retry_transient": 1.0, + "auto_checkpoint_restore": 1.0, + "require_sandbox": true, + "require_checkpoints": true, + "allow_unsafe_tools": false + } + }, + { + "name": "profile_resolution_precedence", + "cases": [ + {"plan": "ci", "action": "auto", "project": "manual", "expected": "ci"}, + {"plan": null, "action": "auto", "project": "manual", "expected": "auto"}, + {"plan": null, "action": null, "project": "manual", "expected": "manual"}, + {"plan": null, "action": null, "project": null, "expected": "manual"} + ] + } + ] +} diff --git a/features/fixtures/m6/autonomy_guardrails.json b/features/fixtures/m6/autonomy_guardrails.json new file mode 100644 index 000000000..03403c189 --- /dev/null +++ b/features/fixtures/m6/autonomy_guardrails.json @@ -0,0 +1,83 @@ +{ + "description": "Autonomy guardrail configurations for M6 acceptance tests", + "fixtures": [ + { + "name": "denylist_guard", + "profile": { + "name": "test-denylist", + "description": "Profile with tool denylist guard", + "auto_strategize": 0.0, + "auto_execute": 0.0, + "auto_apply": 1.0, + "guards": { + "tool_denylist": ["rm_rf", "drop_database", "format_disk"], + "require_approval_for_writes": false, + "require_approval_for_apply": false + } + }, + "checks": [ + {"tool": "read_file", "is_write": false, "expected_allowed": true}, + {"tool": "rm_rf", "is_write": false, "expected_allowed": false, "expected_reason": "denylist"}, + {"tool": "drop_database", "is_write": false, "expected_allowed": false, "expected_reason": "denylist"} + ] + }, + { + "name": "allowlist_guard", + "profile": { + "name": "test-allowlist", + "description": "Profile with tool allowlist guard", + "auto_strategize": 0.0, + "auto_execute": 0.0, + "auto_apply": 1.0, + "guards": { + "tool_allowlist": ["read_file", "list_dir", "search"], + "require_approval_for_writes": false, + "require_approval_for_apply": false + } + }, + "checks": [ + {"tool": "read_file", "is_write": false, "expected_allowed": true}, + {"tool": "write_file", "is_write": false, "expected_allowed": false, "expected_reason": "allowlist"}, + {"tool": "search", "is_write": false, "expected_allowed": true} + ] + }, + { + "name": "cost_budget_guard", + "profile": { + "name": "test-budget", + "description": "Profile with cost budget guard", + "auto_strategize": 0.0, + "auto_execute": 0.0, + "auto_apply": 1.0, + "guards": { + "max_total_cost": 10.0, + "max_tool_calls_per_step": 5 + } + }, + "checks": [ + {"tool": "llm_call", "cost_so_far": 3.0, "calls_so_far": 2, "expected_allowed": true}, + {"tool": "llm_call", "cost_so_far": 10.0, "calls_so_far": 2, "expected_allowed": false, "expected_reason": "budget"}, + {"tool": "llm_call", "cost_so_far": 3.0, "calls_so_far": 5, "expected_allowed": false, "expected_reason": "limit"} + ] + }, + { + "name": "write_approval_guard", + "profile": { + "name": "test-write-approval", + "description": "Profile requiring write approval", + "auto_strategize": 0.0, + "auto_execute": 0.0, + "auto_apply": 1.0, + "guards": { + "require_approval_for_writes": true, + "require_approval_for_apply": true + } + }, + "checks": [ + {"tool": "read_file", "is_write": false, "expected_allowed": true}, + {"tool": "write_file", "is_write": true, "expected_allowed": false, "expected_reason": "Write operations"}, + {"tool": "__apply__", "is_write": false, "expected_allowed": false, "expected_reason": "Apply phase"} + ] + } + ] +} diff --git a/features/m6_autonomy_acceptance.feature b/features/m6_autonomy_acceptance.feature new file mode 100644 index 000000000..b2d519c5c --- /dev/null +++ b/features/m6_autonomy_acceptance.feature @@ -0,0 +1,309 @@ +Feature: M6 autonomy acceptance smoke tests + As a developer working with the CleverAgents M6 milestone + I want to verify ACP facade flows, autonomy guardrails, and audit trails + So that the autonomy hardening layer enforces safety constraints correctly + + Background: + Given a m6 smoke test runner + And a m6 smoke ACP local facade + + # --- Fixture loading --- + + Scenario: M6 smoke load ACP facade flows fixture + When I m6 smoke load the ACP facade flows fixture + Then the m6 smoke facade fixture should have a session lifecycle entry + And the m6 smoke facade fixture should have a plan lifecycle entry + + Scenario: M6 smoke load autonomy guardrails fixture + When I m6 smoke load the autonomy guardrails fixture + Then the m6 smoke guardrails fixture should have a denylist entry + And the m6 smoke guardrails fixture should have an allowlist entry + And the m6 smoke guardrails fixture should have a cost budget entry + + Scenario: M6 smoke load automation profiles fixture + When I m6 smoke load the automation profiles fixture + Then the m6 smoke profiles fixture should list all 8 built-in names + And the m6 smoke profiles fixture should have a custom profile entry + + # --- ACP facade session operations --- + + Scenario: M6 smoke ACP session create returns session id + When I m6 smoke dispatch "session.create" with params {} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "session_id" + And the m6 smoke response data should contain key "status" + + Scenario: M6 smoke ACP session close returns closed status + When I m6 smoke dispatch "session.close" with params {} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data "status" should equal "closed" + + # --- ACP facade plan operations --- + + Scenario: M6 smoke ACP plan create returns plan id + When I m6 smoke dispatch "plan.create" with params {} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "plan_id" + + Scenario: M6 smoke ACP plan execute queues plan + When I m6 smoke dispatch "plan.execute" with params {"plan_id": "01M6SM0KE00000000000000001"} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data "status" should equal "queued" + + Scenario: M6 smoke ACP plan status returns phase + When I m6 smoke dispatch "plan.status" with params {"plan_id": "01M6SM0KE00000000000000001"} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "phase" + + Scenario: M6 smoke ACP plan diff returns changes list + When I m6 smoke dispatch "plan.diff" with params {"plan_id": "01M6SM0KE00000000000000001"} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "changes" + + Scenario: M6 smoke ACP plan apply returns applied status + When I m6 smoke dispatch "plan.apply" with params {"plan_id": "01M6SM0KE00000000000000001"} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data "status" should equal "applied" + + # --- ACP facade registry and context operations --- + + Scenario: M6 smoke ACP registry list tools returns empty list + When I m6 smoke dispatch "registry.list_tools" with params {} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "tools" + + Scenario: M6 smoke ACP registry list resources returns empty list + When I m6 smoke dispatch "registry.list_resources" with params {} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "resources" + + Scenario: M6 smoke ACP context get returns context dict + When I m6 smoke dispatch "context.get" with params {} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "context" + + Scenario: M6 smoke ACP event subscribe returns subscription id + When I m6 smoke dispatch "event.subscribe" with params {} + Then the m6 smoke response status should be "ok" + And the m6 smoke response data should contain key "subscription_id" + + # --- ACP facade error handling --- + + Scenario: M6 smoke ACP unknown operation raises error + When I m6 smoke dispatch unknown operation "nonexistent.op" + Then the m6 smoke facade should raise AcpOperationNotFoundError + + Scenario: M6 smoke ACP dispatch with invalid request type raises TypeError + When I m6 smoke dispatch with a non-AcpRequest object + Then the m6 smoke facade should raise TypeError + + # --- ACP facade service registration --- + + Scenario: M6 smoke ACP register service stores service + When I m6 smoke register service "plan_service" on the facade + Then the m6 smoke facade should have service "plan_service" + + Scenario: M6 smoke ACP list operations returns all supported + When I m6 smoke list facade operations + Then the m6 smoke operations should include "session.create" + And the m6 smoke operations should include "plan.execute" + And the m6 smoke operations should include "event.subscribe" + And the m6 smoke operations count should be 11 + + # --- ACP event queue --- + + Scenario: M6 smoke ACP event queue publish and retrieve + Given a m6 smoke ACP event queue + When I m6 smoke publish an event with type "plan.progress" + Then the m6 smoke event queue should have 1 event + And the m6 smoke last event type should be "plan.progress" + + Scenario: M6 smoke ACP event queue subscribe local callback + Given a m6 smoke ACP event queue + When I m6 smoke subscribe a local callback + And I m6 smoke publish an event with type "plan.complete" + Then the m6 smoke callback should have been called once + + Scenario: M6 smoke ACP event queue unsubscribe + Given a m6 smoke ACP event queue + When I m6 smoke subscribe a local callback + And I m6 smoke unsubscribe the callback + And I m6 smoke publish an event with type "plan.complete" + Then the m6 smoke callback should not have been called + + Scenario: M6 smoke ACP event queue close prevents publish + Given a m6 smoke ACP event queue + When I m6 smoke close the event queue + Then the m6 smoke publishing should raise RuntimeError + + Scenario: M6 smoke ACP event queue remote subscribe raises error + Given a m6 smoke ACP event queue + When I m6 smoke attempt remote subscribe to "https://example.com/events" + Then the m6 smoke facade should raise AcpNotAvailableError + + # --- ACP HTTP transport stub --- + + Scenario: M6 smoke ACP transport send raises not available + When I m6 smoke attempt transport send + Then the m6 smoke facade should raise AcpNotAvailableError + + Scenario: M6 smoke ACP transport connect raises not available + When I m6 smoke attempt transport connect to "https://example.com/acp" + Then the m6 smoke facade should raise AcpNotAvailableError + + Scenario: M6 smoke ACP transport disconnect raises not available + When I m6 smoke attempt transport disconnect + Then the m6 smoke facade should raise AcpNotAvailableError + + Scenario: M6 smoke ACP transport is_connected returns false + When I m6 smoke check transport is_connected + Then the m6 smoke transport should not be connected + + # --- ACP version negotiation --- + + Scenario: M6 smoke ACP version negotiation accepts 1.0 + When I m6 smoke negotiate ACP version "1.0" + Then the m6 smoke negotiated version should be "1.0" + + Scenario: M6 smoke ACP version negotiation rejects unsupported + When I m6 smoke negotiate ACP version "2.0" + Then the m6 smoke facade should raise AcpVersionMismatchError + + Scenario: M6 smoke ACP version is_supported returns correct result + When I m6 smoke check if version "1.0" is supported + Then the m6 smoke version support should be true + When I m6 smoke check if version "99.0" is supported + Then the m6 smoke version support should be false + + # --- Automation profiles built-in --- + + Scenario: M6 smoke all 8 built-in profiles exist + When I m6 smoke list all built-in profiles + Then the m6 smoke profile count should be 8 + And the m6 smoke profiles should include "manual" + And the m6 smoke profiles should include "full-auto" + + Scenario: M6 smoke manual profile has all thresholds at 1.0 + When I m6 smoke load built-in profile "manual" + Then the m6 smoke profile auto_strategize should be 1.0 + And the m6 smoke profile auto_execute should be 1.0 + And the m6 smoke profile auto_apply should be 1.0 + And the m6 smoke profile require_sandbox should be true + + Scenario: M6 smoke full-auto profile has no gates + When I m6 smoke load built-in profile "full-auto" + Then the m6 smoke profile auto_strategize should be 0.0 + And the m6 smoke profile auto_execute should be 0.0 + And the m6 smoke profile auto_apply should be 0.0 + And the m6 smoke profile require_sandbox should be false + And the m6 smoke profile allow_unsafe_tools should be true + + # --- Automation profile creation and validation --- + + Scenario: M6 smoke create custom namespaced profile + When I m6 smoke create a profile named "acme/strict" with auto_apply 1.0 + Then the m6 smoke created profile name should be "acme/strict" + And the m6 smoke created profile auto_apply should be 1.0 + + Scenario: M6 smoke profile name validation rejects invalid + When I m6 smoke create a profile with invalid name "has spaces" + Then the m6 smoke creation should raise ValueError + + Scenario: M6 smoke profile threshold validation rejects out of range + When I m6 smoke create a profile with auto_strategize 1.5 + Then the m6 smoke creation should raise ValueError + + Scenario: M6 smoke profile from_yaml loads correctly + Given a m6 smoke temporary profile YAML file + When I m6 smoke load profile from the temp YAML + Then the m6 smoke loaded profile name should be "test-yaml-profile" + + # --- Guard enforcement --- + + Scenario: M6 smoke guard denylist blocks denied tool + Given a m6 smoke profile with denylist guard for "rm_rf" + When I m6 smoke check guard for tool "rm_rf" + Then the m6 smoke guard result should not be allowed + And the m6 smoke guard reason should contain "denylist" + + Scenario: M6 smoke guard denylist allows non-denied tool + Given a m6 smoke profile with denylist guard for "rm_rf" + When I m6 smoke check guard for tool "read_file" + Then the m6 smoke guard result should be allowed + + Scenario: M6 smoke guard allowlist blocks unlisted tool + Given a m6 smoke profile with allowlist guard for "read_file" and "search" + When I m6 smoke check guard for tool "write_file" + Then the m6 smoke guard result should not be allowed + And the m6 smoke guard reason should contain "allowlist" + + Scenario: M6 smoke guard max tool calls blocks at limit + Given a m6 smoke profile with max 5 tool calls per step + When I m6 smoke check guard for tool "llm_call" with 5 calls so far + Then the m6 smoke guard result should not be allowed + And the m6 smoke guard reason should contain "limit" + + Scenario: M6 smoke guard cost budget blocks at cap + Given a m6 smoke profile with max cost 10.0 + When I m6 smoke check guard for tool "llm_call" with cost 10.0 + Then the m6 smoke guard result should not be allowed + And the m6 smoke guard reason should contain "budget" + + Scenario: M6 smoke guard write approval blocks write operations + Given a m6 smoke profile with write approval required + When I m6 smoke check guard for tool "write_file" as a write operation + Then the m6 smoke guard result should not be allowed + And the m6 smoke guard reason should contain "Write operations" + + Scenario: M6 smoke guard apply approval blocks apply phase + Given a m6 smoke profile with apply approval required + When I m6 smoke check guard for tool "__apply__" + Then the m6 smoke guard result should not be allowed + And the m6 smoke guard reason should contain "Apply phase" + + Scenario: M6 smoke guard with no guards allows everything + Given a m6 smoke profile with no guards + When I m6 smoke check guard for tool "anything" + Then the m6 smoke guard result should be allowed + + # --- Profile service resolution precedence --- + + Scenario: M6 smoke profile resolution plan takes precedence + Given a m6 smoke automation profile service + When I m6 smoke resolve profile with plan "ci" action "auto" project "manual" + Then the m6 smoke resolved profile name should be "ci" + + Scenario: M6 smoke profile resolution action takes precedence over project + Given a m6 smoke automation profile service + When I m6 smoke resolve profile with plan null action "auto" project "manual" + Then the m6 smoke resolved profile name should be "auto" + + Scenario: M6 smoke profile resolution falls back to global default + Given a m6 smoke automation profile service + When I m6 smoke resolve profile with plan null action null project null + Then the m6 smoke resolved profile name should be "manual" + + # --- Profile service guard evaluation --- + + Scenario: M6 smoke service evaluate guard delegates to profile + Given a m6 smoke automation profile service + When I m6 smoke evaluate guard for profile "manual" and tool "read_file" + Then the m6 smoke guard result should be allowed + + # --- ACP model validation --- + + Scenario: M6 smoke AcpRequest validates non-empty operation + When I m6 smoke create AcpRequest with empty operation + Then the m6 smoke creation should raise ValueError + + Scenario: M6 smoke AcpResponse validates status values + When I m6 smoke create AcpResponse with invalid status "maybe" + Then the m6 smoke creation should raise ValueError + + Scenario: M6 smoke AcpEvent validates non-empty event_type + When I m6 smoke create AcpEvent with empty event_type + Then the m6 smoke creation should raise ValueError + + Scenario: M6 smoke AcpErrorDetail validates non-empty fields + When I m6 smoke create AcpErrorDetail with empty code + Then the m6 smoke creation should raise ValueError diff --git a/features/steps/m6_autonomy_acceptance_steps.py b/features/steps/m6_autonomy_acceptance_steps.py new file mode 100644 index 000000000..22a256e0c --- /dev/null +++ b/features/steps/m6_autonomy_acceptance_steps.py @@ -0,0 +1,774 @@ +"""Step definitions for M6 autonomy acceptance smoke tests. + +All step names are prefixed with ``m6 smoke`` to avoid ``AmbiguousStep`` +conflicts with existing steps. +""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path +from unittest.mock import MagicMock + +import yaml +from behave import given, then, when +from behave.runner import Context + +from cleveragents.acp.errors import ( + AcpNotAvailableError, + AcpOperationNotFoundError, + AcpVersionMismatchError, +) +from cleveragents.acp.events import AcpEventQueue +from cleveragents.acp.facade import AcpLocalFacade +from cleveragents.acp.models import ( + AcpErrorDetail, + AcpEvent, + AcpRequest, + AcpResponse, +) +from cleveragents.acp.transport import AcpHttpTransport +from cleveragents.acp.versioning import AcpVersionNegotiator +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" + + +# ----------------------------------------------------------------------- +# Background +# ----------------------------------------------------------------------- + + +@given("a m6 smoke test runner") +def step_m6_smoke_runner(context: Context) -> None: + """Initialise the m6 smoke test context.""" + context.m6_result = None + context.m6_response = None + context.m6_error = None + context.m6_profile = None + context.m6_guard_result = None + context.m6_operations = None + context.m6_event_queue = None + context.m6_callback_calls = [] + context.m6_subscription_id = None + context.m6_version_result = None + context.m6_version_supported = None + context.m6_transport_connected = None + context.m6_profiles_list = [] + + +@given("a m6 smoke ACP local facade") +def step_m6_smoke_facade(context: Context) -> None: + """Create an AcpLocalFacade for testing.""" + context.m6_facade = AcpLocalFacade() + + +# ----------------------------------------------------------------------- +# Fixture loading +# ----------------------------------------------------------------------- + + +@when("I m6 smoke load the ACP facade flows fixture") +def step_m6_smoke_load_facade_fixture(context: Context) -> None: + with open(_FIXTURES_DIR / "acp_facade_flows.json") as f: + context.m6_fixture_data = json.load(f) + + +@then("the m6 smoke facade fixture should have a session lifecycle entry") +def step_m6_smoke_facade_session(context: Context) -> None: + names = [f["name"] for f in context.m6_fixture_data["fixtures"]] + assert "session_lifecycle" in names + + +@then("the m6 smoke facade fixture should have a plan lifecycle entry") +def step_m6_smoke_facade_plan(context: Context) -> None: + names = [f["name"] for f in context.m6_fixture_data["fixtures"]] + assert "plan_lifecycle" in names + + +@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 + + +# ----------------------------------------------------------------------- +# ACP facade dispatch operations +# ----------------------------------------------------------------------- + + +@when('I m6 smoke dispatch "{operation}" with params {params_json}') +def step_m6_smoke_dispatch( + context: Context, + operation: str, + params_json: str, +) -> None: + params = json.loads(params_json) + request = AcpRequest(operation=operation, params=params) + context.m6_response = context.m6_facade.dispatch(request) + + +@then('the m6 smoke response status should be "{status}"') +def step_m6_smoke_response_status(context: Context, status: str) -> None: + assert context.m6_response is not None + assert context.m6_response.status == status + + +@then('the m6 smoke response data should contain key "{key}"') +def step_m6_smoke_response_key(context: Context, key: str) -> None: + assert context.m6_response is not None + assert key in context.m6_response.data + + +@then('the m6 smoke response data "{key}" should equal "{value}"') +def step_m6_smoke_response_value( + context: Context, + key: str, + value: str, +) -> None: + assert context.m6_response is not None + assert context.m6_response.data[key] == value + + +# ----------------------------------------------------------------------- +# ACP facade error handling +# ----------------------------------------------------------------------- + + +@when('I m6 smoke dispatch unknown operation "{operation}"') +def step_m6_smoke_dispatch_unknown(context: Context, operation: str) -> None: + request = AcpRequest(operation=operation, params={}) + try: + context.m6_facade.dispatch(request) + context.m6_error = None + except AcpOperationNotFoundError as exc: + context.m6_error = exc + + +@then("the m6 smoke facade should raise AcpOperationNotFoundError") +def step_m6_smoke_error_op_not_found(context: Context) -> None: + assert isinstance(context.m6_error, AcpOperationNotFoundError) + + +@when("I m6 smoke dispatch with a non-AcpRequest object") +def step_m6_smoke_dispatch_invalid(context: Context) -> None: + try: + context.m6_facade.dispatch("not a request") # type: ignore[arg-type] + context.m6_error = None + except TypeError as exc: + context.m6_error = exc + + +@then("the m6 smoke facade should raise TypeError") +def step_m6_smoke_error_type(context: Context) -> None: + assert isinstance(context.m6_error, TypeError) + + +# ----------------------------------------------------------------------- +# ACP facade service registration and list operations +# ----------------------------------------------------------------------- + + +@when('I m6 smoke register service "{name}" on the facade') +def step_m6_smoke_register_service(context: Context, name: str) -> None: + mock_service = MagicMock() + context.m6_facade.register_service(name, mock_service) + + +@then('the m6 smoke facade should have service "{name}"') +def step_m6_smoke_has_service(context: Context, name: str) -> None: + assert name in context.m6_facade._services + + +@when("I m6 smoke list facade operations") +def step_m6_smoke_list_ops(context: Context) -> None: + context.m6_operations = context.m6_facade.list_operations() + + +@then('the m6 smoke operations should include "{operation}"') +def step_m6_smoke_ops_include(context: Context, operation: str) -> None: + assert operation in context.m6_operations + + +@then("the m6 smoke operations count should be {count:d}") +def step_m6_smoke_ops_count(context: Context, count: int) -> None: + assert len(context.m6_operations) == count + + +# ----------------------------------------------------------------------- +# ACP event queue +# ----------------------------------------------------------------------- + + +@given("a m6 smoke ACP event queue") +def step_m6_smoke_event_queue(context: Context) -> None: + context.m6_event_queue = AcpEventQueue() + context.m6_callback_calls = [] + + +@when('I m6 smoke publish an event with type "{event_type}"') +def step_m6_smoke_publish_event(context: Context, event_type: str) -> None: + event = AcpEvent(event_type=event_type, data={"test": True}) + context.m6_event_queue.publish(event) + + +@then("the m6 smoke event queue should have {count:d} event") +def step_m6_smoke_event_count(context: Context, count: int) -> None: + events = context.m6_event_queue.get_events() + assert len(events) == count + + +@then('the m6 smoke last event type should be "{event_type}"') +def step_m6_smoke_last_event_type(context: Context, event_type: str) -> None: + events = context.m6_event_queue.get_events() + assert events[-1].event_type == event_type + + +@when("I m6 smoke subscribe a local callback") +def step_m6_smoke_subscribe_local(context: Context) -> None: + def _callback(event: AcpEvent) -> None: + context.m6_callback_calls.append(event) + + context.m6_subscription_id = context.m6_event_queue.subscribe_local(_callback) + + +@then("the m6 smoke callback should have been called once") +def step_m6_smoke_callback_called(context: Context) -> None: + assert len(context.m6_callback_calls) == 1 + + +@when("I m6 smoke unsubscribe the callback") +def step_m6_smoke_unsubscribe(context: Context) -> None: + context.m6_event_queue.unsubscribe(context.m6_subscription_id) + + +@then("the m6 smoke callback should not have been called") +def step_m6_smoke_callback_not_called(context: Context) -> None: + assert len(context.m6_callback_calls) == 0 + + +@when("I m6 smoke close the event queue") +def step_m6_smoke_close_queue(context: Context) -> None: + context.m6_event_queue.close() + + +@then("the m6 smoke publishing should raise RuntimeError") +def step_m6_smoke_publish_after_close(context: Context) -> None: + event = AcpEvent(event_type="after.close", data={}) + try: + context.m6_event_queue.publish(event) + raise AssertionError("Expected RuntimeError") + except RuntimeError: + pass + + +@when('I m6 smoke attempt remote subscribe to "{endpoint}"') +def step_m6_smoke_remote_subscribe(context: Context, endpoint: str) -> None: + try: + context.m6_event_queue.subscribe_remote(endpoint) + context.m6_error = None + except AcpNotAvailableError as exc: + context.m6_error = exc + + +@then("the m6 smoke facade should raise AcpNotAvailableError") +def step_m6_smoke_error_not_available(context: Context) -> None: + assert isinstance(context.m6_error, AcpNotAvailableError) + + +# ----------------------------------------------------------------------- +# ACP HTTP transport stub +# ----------------------------------------------------------------------- + + +@when("I m6 smoke attempt transport send") +def step_m6_smoke_transport_send(context: Context) -> None: + transport = AcpHttpTransport() + request = AcpRequest(operation="plan.create", params={}) + try: + transport.send(request) + context.m6_error = None + except AcpNotAvailableError as exc: + context.m6_error = exc + + +@when('I m6 smoke attempt transport connect to "{url}"') +def step_m6_smoke_transport_connect(context: Context, url: str) -> None: + transport = AcpHttpTransport() + try: + transport.connect(url) + context.m6_error = None + except AcpNotAvailableError as exc: + context.m6_error = exc + + +@when("I m6 smoke attempt transport disconnect") +def step_m6_smoke_transport_disconnect(context: Context) -> None: + transport = AcpHttpTransport() + try: + transport.disconnect() + context.m6_error = None + except AcpNotAvailableError as exc: + context.m6_error = exc + + +@when("I m6 smoke check transport is_connected") +def step_m6_smoke_transport_connected(context: Context) -> None: + transport = AcpHttpTransport() + context.m6_transport_connected = transport.is_connected() + + +@then("the m6 smoke transport should not be connected") +def step_m6_smoke_transport_not_connected(context: Context) -> None: + assert context.m6_transport_connected is False + + +# ----------------------------------------------------------------------- +# ACP version negotiation +# ----------------------------------------------------------------------- + + +@when('I m6 smoke negotiate ACP version "{version}"') +def step_m6_smoke_negotiate(context: Context, version: str) -> None: + negotiator = AcpVersionNegotiator() + try: + context.m6_version_result = negotiator.negotiate(version) + context.m6_error = None + except AcpVersionMismatchError as exc: + context.m6_error = exc + + +@then('the m6 smoke negotiated version should be "{version}"') +def step_m6_smoke_negotiated(context: Context, version: str) -> None: + assert context.m6_version_result == version + + +@then("the m6 smoke facade should raise AcpVersionMismatchError") +def step_m6_smoke_error_version(context: Context) -> None: + assert isinstance(context.m6_error, AcpVersionMismatchError) + + +@when('I m6 smoke check if version "{version}" is supported') +def step_m6_smoke_version_supported(context: Context, version: str) -> None: + negotiator = AcpVersionNegotiator() + context.m6_version_supported = negotiator.is_supported(version) + + +@then("the m6 smoke version support should be true") +def step_m6_smoke_version_true(context: Context) -> None: + assert context.m6_version_supported is True + + +@then("the m6 smoke version support should be false") +def step_m6_smoke_version_false(context: Context) -> None: + assert context.m6_version_supported is False + + +# ----------------------------------------------------------------------- +# 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 auto_strategize should be {value:g}") +def step_m6_smoke_profile_auto_strategize( + context: Context, + value: float, +) -> None: + assert context.m6_profile.auto_strategize == value + + +@then("the m6 smoke profile auto_execute should be {value:g}") +def step_m6_smoke_profile_auto_execute( + context: Context, + value: float, +) -> None: + assert context.m6_profile.auto_execute == value + + +@then("the m6 smoke profile auto_apply should be {value:g}") +def step_m6_smoke_profile_auto_apply( + context: Context, + value: float, +) -> None: + assert context.m6_profile.auto_apply == 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.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.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.allow_unsafe_tools is True + + +# ----------------------------------------------------------------------- +# Automation profile creation and validation +# ----------------------------------------------------------------------- + + +@when('I m6 smoke create a profile named "{name}" with auto_apply {value:g}') +def step_m6_smoke_create_profile( + context: Context, + name: str, + value: float, +) -> None: + context.m6_profile = AutomationProfile(name=name, auto_apply=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 auto_apply should be {value:g}") +def step_m6_smoke_created_auto_apply(context: Context, value: float) -> None: + assert context.m6_profile.auto_apply == 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 auto_strategize {value:g}") +def step_m6_smoke_invalid_threshold(context: Context, value: float) -> None: + try: + AutomationProfile(name="test-bad-threshold", auto_strategize=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", + "auto_strategize": 0.5, + "auto_execute": 0.5, + "auto_apply": 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, + ) + + +# ----------------------------------------------------------------------- +# ACP model validation +# ----------------------------------------------------------------------- + + +@when("I m6 smoke create AcpRequest with empty operation") +def step_m6_smoke_invalid_request(context: Context) -> None: + try: + AcpRequest(operation="") + context.m6_error = None + except ValueError as exc: + context.m6_error = exc + + +@when('I m6 smoke create AcpResponse with invalid status "{status}"') +def step_m6_smoke_invalid_response(context: Context, status: str) -> None: + try: + AcpResponse(request_id="test", status=status) + context.m6_error = None + except ValueError as exc: + context.m6_error = exc + + +@when("I m6 smoke create AcpEvent with empty event_type") +def step_m6_smoke_invalid_event(context: Context) -> None: + try: + AcpEvent(event_type="") + context.m6_error = None + except ValueError as exc: + context.m6_error = exc + + +@when("I m6 smoke create AcpErrorDetail with empty code") +def step_m6_smoke_invalid_error_detail(context: Context) -> None: + try: + AcpErrorDetail(code="", message="test") + context.m6_error = None + except ValueError as exc: + context.m6_error = exc diff --git a/robot/helper_m6_autonomy_acceptance.py b/robot/helper_m6_autonomy_acceptance.py new file mode 100644 index 000000000..dabe1345b --- /dev/null +++ b/robot/helper_m6_autonomy_acceptance.py @@ -0,0 +1,356 @@ +"""Helper script for m6_autonomy_acceptance.robot E2E tests. + +Each subcommand is a self-contained check that prints a sentinel on success. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +# Ensure local source tree is importable +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from cleveragents.acp.errors import ( # noqa: E402 + AcpNotAvailableError, + AcpOperationNotFoundError, + AcpVersionMismatchError, +) +from cleveragents.acp.events import AcpEventQueue # noqa: E402 +from cleveragents.acp.facade import AcpLocalFacade # noqa: E402 +from cleveragents.acp.models import AcpEvent, AcpRequest # noqa: E402 +from cleveragents.acp.transport import AcpHttpTransport # noqa: E402 +from cleveragents.acp.versioning import AcpVersionNegotiator # noqa: E402 +from cleveragents.application.services.automation_profile_service import ( # noqa: E402 + AutomationProfileService, +) +from cleveragents.domain.models.core.automation_guard import ( # noqa: E402 + AutomationGuard, +) +from cleveragents.domain.models.core.automation_profile import ( # noqa: E402 + AutomationProfile, +) + +_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m6" + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def facade_session() -> None: + """Dispatch session.create and session.close.""" + facade = AcpLocalFacade() + + resp_create = facade.dispatch(AcpRequest(operation="session.create", params={})) + assert resp_create.status == "ok", f"Expected ok, got {resp_create.status}" + assert "session_id" in resp_create.data + + resp_close = facade.dispatch(AcpRequest(operation="session.close", params={})) + assert resp_close.status == "ok" + assert resp_close.data["status"] == "closed" + + print("m6-facade-session-ok") + + +def facade_plan() -> None: + """Dispatch plan create/execute/status/diff/apply.""" + facade = AcpLocalFacade() + + resp = facade.dispatch(AcpRequest(operation="plan.create", params={})) + assert resp.status == "ok" + plan_id = resp.data["plan_id"] + assert plan_id + + resp = facade.dispatch( + AcpRequest(operation="plan.execute", params={"plan_id": plan_id}) + ) + assert resp.status == "ok" + assert resp.data["status"] == "queued" + + resp = facade.dispatch( + AcpRequest(operation="plan.status", params={"plan_id": plan_id}) + ) + assert resp.status == "ok" + assert "phase" in resp.data + + resp = facade.dispatch( + AcpRequest(operation="plan.diff", params={"plan_id": plan_id}) + ) + assert resp.status == "ok" + assert "changes" in resp.data + + resp = facade.dispatch( + AcpRequest(operation="plan.apply", params={"plan_id": plan_id}) + ) + assert resp.status == "ok" + assert resp.data["status"] == "applied" + + print("m6-facade-plan-ok") + + +def facade_unknown_op() -> None: + """Verify unknown operation raises AcpOperationNotFoundError.""" + facade = AcpLocalFacade() + try: + facade.dispatch(AcpRequest(operation="nonexistent.op", params={})) + print("FAIL: expected AcpOperationNotFoundError") + sys.exit(1) + except AcpOperationNotFoundError: + pass + + print("m6-facade-unknown-op-ok") + + +def event_queue() -> None: + """Publish events and verify local subscriber receives them.""" + queue = AcpEventQueue() + received = [] + + sub_id = queue.subscribe_local(lambda e: received.append(e)) + queue.publish(AcpEvent(event_type="plan.progress", data={"step": 1})) + assert len(received) == 1 + assert received[0].event_type == "plan.progress" + + # Unsubscribe and verify no more callbacks + queue.unsubscribe(sub_id) + queue.publish(AcpEvent(event_type="plan.complete", data={})) + assert len(received) == 1 # still 1 + + # Verify get_events returns both + events = queue.get_events() + assert len(events) == 2 + + # Close and verify publish raises + queue.close() + try: + queue.publish(AcpEvent(event_type="after.close", data={})) + print("FAIL: expected RuntimeError") + sys.exit(1) + except RuntimeError: + pass + + print("m6-event-queue-ok") + + +def transport_stub() -> None: + """Verify HTTP transport stub raises AcpNotAvailableError.""" + transport = AcpHttpTransport() + + # send + try: + transport.send(AcpRequest(operation="plan.create", params={})) + print("FAIL: expected AcpNotAvailableError on send") + sys.exit(1) + except AcpNotAvailableError: + pass + + # connect + try: + transport.connect("https://example.com/acp") + print("FAIL: expected AcpNotAvailableError on connect") + sys.exit(1) + except AcpNotAvailableError: + pass + + # disconnect + try: + transport.disconnect() + print("FAIL: expected AcpNotAvailableError on disconnect") + sys.exit(1) + except AcpNotAvailableError: + pass + + # is_connected + assert transport.is_connected() is False + + print("m6-transport-stub-ok") + + +def version_negotiation() -> None: + """Negotiate supported and unsupported ACP versions.""" + negotiator = AcpVersionNegotiator() + + # Supported + result = negotiator.negotiate("1.0") + assert result == "1.0" + assert negotiator.is_supported("1.0") is True + assert negotiator.get_current() == "1.0" + + # Unsupported + try: + negotiator.negotiate("2.0") + print("FAIL: expected AcpVersionMismatchError") + sys.exit(1) + except AcpVersionMismatchError: + pass + + assert negotiator.is_supported("99.0") is False + + print("m6-version-negotiation-ok") + + +def guard_denylist() -> None: + """Verify denylist guard blocks denied tools.""" + guard = AutomationGuard(tool_denylist=["rm_rf", "drop_database"]) + profile = AutomationProfile(name="test-deny", guards=guard) + + # Allowed tool + result = profile.check_guard(tool_name="read_file") + assert result.allowed is True + + # Denied tool + result = profile.check_guard(tool_name="rm_rf") + assert result.allowed is False + assert result.requires_approval is True + assert "denylist" in (result.reason or "") + + print("m6-guard-denylist-ok") + + +def guard_budget() -> None: + """Verify cost budget and call limit guards.""" + guard = AutomationGuard(max_total_cost=10.0, max_tool_calls_per_step=5) + profile = AutomationProfile(name="test-budget", guards=guard) + + # Under budget and limit + result = profile.check_guard(tool_name="llm_call", cost_so_far=3.0, calls_so_far=2) + assert result.allowed is True + + # Over budget + result = profile.check_guard(tool_name="llm_call", cost_so_far=10.0, calls_so_far=2) + assert result.allowed is False + assert ( + "budget" in (result.reason or "").lower() + or "cost" in (result.reason or "").lower() + ) + + # Over call limit + result = profile.check_guard(tool_name="llm_call", cost_so_far=3.0, calls_so_far=5) + assert result.allowed is False + assert "limit" in (result.reason or "").lower() + + print("m6-guard-budget-ok") + + +def profile_resolution() -> None: + """Verify plan > action > project > global resolution.""" + service = AutomationProfileService() + + # Plan takes precedence + p = service.resolve_profile( + plan_profile="ci", action_profile="auto", project_profile="manual" + ) + assert p.name == "ci" + + # Action takes precedence over project + p = service.resolve_profile( + plan_profile=None, action_profile="auto", project_profile="manual" + ) + assert p.name == "auto" + + # Project takes precedence over global + p = service.resolve_profile( + plan_profile=None, action_profile=None, project_profile="review" + ) + assert p.name == "review" + + # Falls back to global default (manual) + p = service.resolve_profile( + plan_profile=None, action_profile=None, project_profile=None + ) + assert p.name == "manual" + + print("m6-profile-resolution-ok") + + +def fixture_loading() -> None: + """Load all M6 fixture files and verify structure.""" + for fname in ( + "acp_facade_flows.json", + "autonomy_guardrails.json", + "automation_profiles.json", + ): + fpath = _FIXTURES_DIR / fname + with open(fpath) as f: + data = json.load(f) + assert "fixtures" in data, f"Missing 'fixtures' key in {fname}" + assert len(data["fixtures"]) > 0, f"Empty fixtures in {fname}" + print("m6-fixture-loading-ok") + + +def full_flow() -> None: + """End-to-end: facade dispatch + guard check + profile resolution.""" + # Step 1: Facade dispatch + facade = AcpLocalFacade() + resp = facade.dispatch(AcpRequest(operation="session.create", params={})) + assert resp.status == "ok" + + resp = facade.dispatch(AcpRequest(operation="plan.create", params={})) + assert resp.status == "ok" + plan_id = resp.data["plan_id"] + + resp = facade.dispatch( + AcpRequest(operation="plan.execute", params={"plan_id": plan_id}) + ) + assert resp.status == "ok" + + # Step 2: Guard check + guard = AutomationGuard( + tool_denylist=["dangerous_tool"], + max_tool_calls_per_step=3, + ) + profile = AutomationProfile(name="test-full-flow", guards=guard) + + result = profile.check_guard(tool_name="safe_tool", calls_so_far=1) + assert result.allowed is True + + result = profile.check_guard(tool_name="dangerous_tool") + assert result.allowed is False + + # Step 3: Profile resolution + service = AutomationProfileService() + p = service.resolve_profile(plan_profile="ci") + assert p.name == "ci" + + # Step 4: Version negotiation + negotiator = AcpVersionNegotiator() + assert negotiator.negotiate("1.0") == "1.0" + + # Step 5: Event queue + queue = AcpEventQueue() + queue.publish(AcpEvent(event_type="flow.complete", data={})) + assert len(queue.get_events()) == 1 + queue.close() + + print("m6-full-flow-ok") + + +# --------------------------------------------------------------------------- +# Dispatcher +# --------------------------------------------------------------------------- + +_COMMANDS: dict[str, object] = { + "facade-session": facade_session, + "facade-plan": facade_plan, + "facade-unknown-op": facade_unknown_op, + "event-queue": event_queue, + "transport-stub": transport_stub, + "version-negotiation": version_negotiation, + "guard-denylist": guard_denylist, + "guard-budget": guard_budget, + "profile-resolution": profile_resolution, + "fixture-loading": fixture_loading, + "full-flow": full_flow, +} + +if __name__ == "__main__": + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>") + sys.exit(1) + fn = _COMMANDS[sys.argv[1]] + fn() # type: ignore[operator] diff --git a/robot/m6_autonomy_acceptance.robot b/robot/m6_autonomy_acceptance.robot new file mode 100644 index 000000000..b9c439791 --- /dev/null +++ b/robot/m6_autonomy_acceptance.robot @@ -0,0 +1,97 @@ +*** Settings *** +Documentation M6 autonomy acceptance E2E smoke tests — ACP facade, guardrails, audit +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_m6_autonomy_acceptance.py + +*** Test Cases *** +M6 ACP Facade Session Lifecycle + [Documentation] Dispatch session.create and session.close via local facade + ${result}= Run Process ${PYTHON} ${HELPER} facade-session cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-facade-session-ok + +M6 ACP Facade Plan Lifecycle + [Documentation] Dispatch plan create/execute/status/diff/apply operations + ${result}= Run Process ${PYTHON} ${HELPER} facade-plan cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-facade-plan-ok + +M6 ACP Facade Unknown Operation Error + [Documentation] Verify unknown operations raise AcpOperationNotFoundError + ${result}= Run Process ${PYTHON} ${HELPER} facade-unknown-op cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-facade-unknown-op-ok + +M6 ACP Event Queue Publish Subscribe + [Documentation] Publish events and verify local subscriber receives them + ${result}= Run Process ${PYTHON} ${HELPER} event-queue cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-event-queue-ok + +M6 ACP Transport Stub Rejects All + [Documentation] Verify HTTP transport stub raises AcpNotAvailableError + ${result}= Run Process ${PYTHON} ${HELPER} transport-stub cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-transport-stub-ok + +M6 ACP Version Negotiation + [Documentation] Negotiate supported and unsupported ACP versions + ${result}= Run Process ${PYTHON} ${HELPER} version-negotiation cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-version-negotiation-ok + +M6 Guard Denylist Enforcement + [Documentation] Verify denylist guard blocks denied tools + ${result}= Run Process ${PYTHON} ${HELPER} guard-denylist cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-guard-denylist-ok + +M6 Guard Budget Enforcement + [Documentation] Verify cost budget and call limit guards work + ${result}= Run Process ${PYTHON} ${HELPER} guard-budget cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-guard-budget-ok + +M6 Profile Resolution Precedence + [Documentation] Verify plan > action > project > global resolution + ${result}= Run Process ${PYTHON} ${HELPER} profile-resolution cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-profile-resolution-ok + +M6 Fixture Loading + [Documentation] Load all M6 fixture files and verify structure + ${result}= Run Process ${PYTHON} ${HELPER} fixture-loading cwd=${WORKSPACE} timeout=30s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-fixture-loading-ok + +M6 Full Autonomy Acceptance Flow + [Documentation] End-to-end: facade dispatch + guard check + profile resolution + ${result}= Run Process ${PYTHON} ${HELPER} full-flow cwd=${WORKSPACE} timeout=60s + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m6-full-flow-ok -- 2.52.0 From c129f4c3f091cf6596b4bd3835c96e2d3f483064 Mon Sep 17 00:00:00 2001 From: "Brent E. Edwards" Date: Fri, 27 Feb 2026 22:33:31 +0000 Subject: [PATCH 2/2] fix(test): address PR #470 review feedback - Add CHANGELOG.md entry for M6 autonomy acceptance suite - Split m6_autonomy_acceptance_steps.py (774 lines) into m6_facade_steps.py (398 lines) and m6_guardrails_steps.py (399 lines) to comply with the project's 500-line guideline Closes #211 --- CHANGELOG.md | 8 + ...acceptance_steps.py => m6_facade_steps.py} | 386 +---------------- features/steps/m6_guardrails_steps.py | 399 ++++++++++++++++++ 3 files changed, 412 insertions(+), 381 deletions(-) rename features/steps/{m6_autonomy_acceptance_steps.py => m6_facade_steps.py} (50%) create mode 100644 features/steps/m6_guardrails_steps.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e5fd83136..dca4e6b0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +- Added comprehensive M6 autonomy acceptance test suite covering ACP local facade dispatch + (session/plan/registry/context/event operations), event queue pub/sub with local callbacks + and close semantics, HTTP transport stub rejection, version negotiation, automation profile + built-ins (8 profiles), custom profile creation/validation/YAML loading, guard enforcement + (denylist, allowlist, call-limit, cost-budget, write-approval, apply-approval), and profile + service 4-level resolution precedence. Includes Behave BDD scenarios (52), Robot Framework + integration tests (11), ASV performance benchmarks (5 suites), JSON fixtures, and + documentation update. (#211) - Added MCP refresh hooks to wire `notifications/tools/list_changed` events from MCP servers to `SkillRegistry`. Introduced `SkillRegistry.refresh(name)` and `refresh_all()` to recompute flattened tool sets on demand. Added `MCPRefreshHook` with configurable debounce diff --git a/features/steps/m6_autonomy_acceptance_steps.py b/features/steps/m6_facade_steps.py similarity index 50% rename from features/steps/m6_autonomy_acceptance_steps.py rename to features/steps/m6_facade_steps.py index 22a256e0c..5f3bc8efa 100644 --- a/features/steps/m6_autonomy_acceptance_steps.py +++ b/features/steps/m6_facade_steps.py @@ -1,17 +1,16 @@ -"""Step definitions for M6 autonomy acceptance smoke tests. +"""Step definitions for M6 ACP facade, event queue, transport, and version tests. -All step names are prefixed with ``m6 smoke`` to avoid ``AmbiguousStep`` -conflicts with existing steps. +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 from unittest.mock import MagicMock -import yaml from behave import given, then, when from behave.runner import Context @@ -30,16 +29,6 @@ from cleveragents.acp.models import ( ) from cleveragents.acp.transport import AcpHttpTransport from cleveragents.acp.versioning import AcpVersionNegotiator -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" @@ -74,7 +63,7 @@ def step_m6_smoke_facade(context: Context) -> None: # ----------------------------------------------------------------------- -# Fixture loading +# Fixture loading — ACP facade flows # ----------------------------------------------------------------------- @@ -96,52 +85,6 @@ def step_m6_smoke_facade_plan(context: Context) -> None: assert "plan_lifecycle" in names -@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 - - # ----------------------------------------------------------------------- # ACP facade dispatch operations # ----------------------------------------------------------------------- @@ -414,325 +357,6 @@ def step_m6_smoke_version_false(context: Context) -> None: assert context.m6_version_supported is False -# ----------------------------------------------------------------------- -# 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 auto_strategize should be {value:g}") -def step_m6_smoke_profile_auto_strategize( - context: Context, - value: float, -) -> None: - assert context.m6_profile.auto_strategize == value - - -@then("the m6 smoke profile auto_execute should be {value:g}") -def step_m6_smoke_profile_auto_execute( - context: Context, - value: float, -) -> None: - assert context.m6_profile.auto_execute == value - - -@then("the m6 smoke profile auto_apply should be {value:g}") -def step_m6_smoke_profile_auto_apply( - context: Context, - value: float, -) -> None: - assert context.m6_profile.auto_apply == 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.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.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.allow_unsafe_tools is True - - -# ----------------------------------------------------------------------- -# Automation profile creation and validation -# ----------------------------------------------------------------------- - - -@when('I m6 smoke create a profile named "{name}" with auto_apply {value:g}') -def step_m6_smoke_create_profile( - context: Context, - name: str, - value: float, -) -> None: - context.m6_profile = AutomationProfile(name=name, auto_apply=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 auto_apply should be {value:g}") -def step_m6_smoke_created_auto_apply(context: Context, value: float) -> None: - assert context.m6_profile.auto_apply == 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 auto_strategize {value:g}") -def step_m6_smoke_invalid_threshold(context: Context, value: float) -> None: - try: - AutomationProfile(name="test-bad-threshold", auto_strategize=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", - "auto_strategize": 0.5, - "auto_execute": 0.5, - "auto_apply": 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, - ) - - # ----------------------------------------------------------------------- # ACP model validation # ----------------------------------------------------------------------- diff --git a/features/steps/m6_guardrails_steps.py b/features/steps/m6_guardrails_steps.py new file mode 100644 index 000000000..244827fae --- /dev/null +++ b/features/steps/m6_guardrails_steps.py @@ -0,0 +1,399 @@ +"""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 auto_strategize should be {value:g}") +def step_m6_smoke_profile_auto_strategize( + context: Context, + value: float, +) -> None: + assert context.m6_profile.auto_strategize == value + + +@then("the m6 smoke profile auto_execute should be {value:g}") +def step_m6_smoke_profile_auto_execute( + context: Context, + value: float, +) -> None: + assert context.m6_profile.auto_execute == value + + +@then("the m6 smoke profile auto_apply should be {value:g}") +def step_m6_smoke_profile_auto_apply( + context: Context, + value: float, +) -> None: + assert context.m6_profile.auto_apply == 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.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.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.allow_unsafe_tools is True + + +# ----------------------------------------------------------------------- +# Automation profile creation and validation +# ----------------------------------------------------------------------- + + +@when('I m6 smoke create a profile named "{name}" with auto_apply {value:g}') +def step_m6_smoke_create_profile( + context: Context, + name: str, + value: float, +) -> None: + context.m6_profile = AutomationProfile(name=name, auto_apply=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 auto_apply should be {value:g}") +def step_m6_smoke_created_auto_apply(context: Context, value: float) -> None: + assert context.m6_profile.auto_apply == 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 auto_strategize {value:g}") +def step_m6_smoke_invalid_threshold(context: Context, value: float) -> None: + try: + AutomationProfile(name="test-bad-threshold", auto_strategize=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", + "auto_strategize": 0.5, + "auto_execute": 0.5, + "auto_apply": 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, + ) -- 2.52.0