From a44082ab704b978206ea42a5e0d2e5dcf9ebdfd7 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Sun, 8 Mar 2026 20:00:18 +0000 Subject: [PATCH] feat(guardrails): implement Plan Generation Pre-flight Guardrails (7 checks before execution) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement spec-mandated pre-flight guardrail checks that validate plan readiness before entering the Strategize phase: 1. Action schema validation — verifies action exists and is well-formed. 2. Actor availability — confirms all 4 actor roles registered. 3. Skill/tool existence — transitively resolves all tools/skills. 4. Automation policy — verifies profile permits execution. 5. Rollback feasibility — ensures all tools checkpointable if required. 6. Resource accessibility — shallow connectivity check on linked resources. 7. Validation attachment resolution — pre-resolves all applicable validations. PlanPreflightGuardrail service runs all 7 checks via run_all_checks(). On first failure, raises PreflightRejection with check name and message. Wired into plan_lifecycle_service before Strategize phase. Behave BDD: 18 scenarios covering all 7 checks (positive + negative). Robot Framework: 3 integration smoke tests. ASV benchmarks: pre-flight check execution time. ISSUES CLOSED: #582 --- benchmarks/plan_preflight_guardrails_bench.py | 134 ++++++++ features/plan_preflight_guardrails.feature | 116 +++++++ .../steps/plan_preflight_guardrails_steps.py | 322 +++++++++++++++++ robot/helper_plan_preflight_guardrails.py | 104 ++++++ robot/plan_preflight_guardrails.robot | 27 ++ .../services/plan_lifecycle_service.py | 31 +- .../services/plan_preflight_guardrail.py | 324 ++++++++++++++++++ 7 files changed, 1057 insertions(+), 1 deletion(-) create mode 100644 benchmarks/plan_preflight_guardrails_bench.py create mode 100644 features/plan_preflight_guardrails.feature create mode 100644 features/steps/plan_preflight_guardrails_steps.py create mode 100644 robot/helper_plan_preflight_guardrails.py create mode 100644 robot/plan_preflight_guardrails.robot create mode 100644 src/cleveragents/application/services/plan_preflight_guardrail.py diff --git a/benchmarks/plan_preflight_guardrails_bench.py b/benchmarks/plan_preflight_guardrails_bench.py new file mode 100644 index 00000000..f326bfb3 --- /dev/null +++ b/benchmarks/plan_preflight_guardrails_bench.py @@ -0,0 +1,134 @@ +"""ASV benchmarks for plan pre-flight guardrail execution time. + +Measures the performance of: +- Individual pre-flight checks +- Full ``run_all_checks`` pass +- ``PreflightReport`` construction +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +try: + from cleveragents.application.services.plan_preflight_guardrail import ( + PlanPreflightGuardrail, + PreflightCheckName, + PreflightCheckResult, + PreflightReport, + ) +except ModuleNotFoundError: + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + from cleveragents.application.services.plan_preflight_guardrail import ( + PlanPreflightGuardrail, + PreflightCheckName, + PreflightCheckResult, + PreflightReport, + ) + + +class PreflightCheckSuite: + """Benchmark individual pre-flight check methods.""" + + def setup(self) -> None: + """Set up shared guardrail instance.""" + self.guardrail = PlanPreflightGuardrail() + self.action_registry: dict[str, object] = { + "test-action": {"name": "test-action"} + } + self.actor_registry: dict[str, object] = { + "strategy": "s", + "execution": "e", + "estimation": "est", + "invariant_reconciliation": "ir", + } + self.tool_registry: dict[str, object] = { + "tool-a": {}, + "tool-b": {}, + } + + def time_check_action_schema(self) -> None: + """Benchmark action schema validation check.""" + self.guardrail.check_action_schema("test-action", self.action_registry) + + def time_check_actor_availability(self) -> None: + """Benchmark actor availability check.""" + self.guardrail.check_actor_availability(self.actor_registry) + + def time_check_skill_tool_existence(self) -> None: + """Benchmark skill/tool existence check.""" + self.guardrail.check_skill_tool_existence( + ("tool-a", "tool-b"), self.tool_registry, (), {} + ) + + def time_check_automation_policy(self) -> None: + """Benchmark automation policy check.""" + self.guardrail.check_automation_policy({"name": "auto"}) + + def time_check_rollback_feasibility_skip(self) -> None: + """Benchmark rollback feasibility when checkpoints not required.""" + self.guardrail.check_rollback_feasibility(False, {}) + + def time_check_rollback_feasibility_pass(self) -> None: + """Benchmark rollback feasibility when all tools checkpointable.""" + self.guardrail.check_rollback_feasibility( + True, {"tool-a": True, "tool-b": True} + ) + + def time_check_resource_accessibility_empty(self) -> None: + """Benchmark resource accessibility with no paths.""" + self.guardrail.check_resource_accessibility(()) + + def time_check_validation_resolution_empty(self) -> None: + """Benchmark validation resolution with no validations.""" + self.guardrail.check_validation_resolution((), {}) + + +class PreflightRunAllSuite: + """Benchmark the full ``run_all_checks`` method.""" + + def setup(self) -> None: + """Set up shared guardrail with fully populated registries.""" + self.guardrail = PlanPreflightGuardrail() + + def time_run_all_checks_pass(self) -> None: + """Benchmark run_all_checks with all checks passing.""" + self.guardrail.run_all_checks( + action_name="test-action", + action_registry={"test-action": {}}, + actor_registry={ + "strategy": "s", + "execution": "e", + "estimation": "est", + "invariant_reconciliation": "ir", + }, + tool_names=("tool-a",), + tool_registry={"tool-a": {}}, + automation_profile={"name": "auto"}, + require_checkpoints=False, + ) + + +class PreflightReportSuite: + """Benchmark PreflightReport construction and queries.""" + + def time_report_construction(self) -> None: + """Benchmark creating a report with 7 results.""" + report = PreflightReport() + for check in PreflightCheckName: + report.add(PreflightCheckResult(check, True, "ok")) + + def time_report_all_passed(self) -> None: + """Benchmark all_passed property.""" + report = PreflightReport() + for check in PreflightCheckName: + report.add(PreflightCheckResult(check, True, "ok")) + _ = report.all_passed + + def time_report_format(self) -> None: + """Benchmark format_report rendering.""" + report = PreflightReport() + for check in PreflightCheckName: + report.add(PreflightCheckResult(check, True, "ok")) + report.format_report() diff --git a/features/plan_preflight_guardrails.feature b/features/plan_preflight_guardrails.feature new file mode 100644 index 00000000..62bf5a81 --- /dev/null +++ b/features/plan_preflight_guardrails.feature @@ -0,0 +1,116 @@ +Feature: Plan Generation Pre-flight Guardrails — 7 checks before execution + As a plan executor + I want pre-flight checks to validate plan readiness + So that ill-formed plans are rejected before entering the Strategize phase + + # Check 1: Action Schema Validation + Scenario: pfg-check1 Valid action passes schema validation + Given a plan preflight guardrail + And an action registry containing "refactor-action" + When I check action schema for "refactor-action" + Then the check should pass with message containing "found and valid" + + Scenario: pfg-check1-fail Missing action fails schema validation + Given a plan preflight guardrail + And an empty action registry + When I check action schema for "nonexistent-action" + Then the check should fail with message containing "not found" + + # Check 2: Actor Availability + Scenario: pfg-check2 All actor roles registered passes availability check + Given a plan preflight guardrail + And an actor registry with all 4 roles registered + When I check actor availability + Then the check should pass + + Scenario: pfg-check2-fail Missing actor roles fails availability check + Given a plan preflight guardrail + And an actor registry missing "estimation" role + When I check actor availability + Then the check should fail with message containing "estimation" + + # Check 3: Skill/Tool Existence + Scenario: pfg-check3 All tools and skills exist passes existence check + Given a plan preflight guardrail + And a tool registry containing "git-diff" and "file-read" + When I check skill/tool existence for tools "git-diff,file-read" + Then the check should pass + + Scenario: pfg-check3-fail Missing tool fails existence check + Given a plan preflight guardrail + And a tool registry containing "git-diff" + When I check skill/tool existence for tools "git-diff,missing-tool" + Then the check should fail with message containing "missing-tool" + + # Check 4: Automation Policy + Scenario: pfg-check4 Valid automation profile passes policy check + Given a plan preflight guardrail + And a valid automation profile + When I check automation policy + Then the check should pass + + Scenario: pfg-check4-fail No automation profile fails policy check + Given a plan preflight guardrail + And no automation profile + When I check automation policy + Then the check should fail with message containing "No automation profile" + + # Check 5: Rollback Feasibility + Scenario: pfg-check5 All tools checkpointable passes rollback check + Given a plan preflight guardrail + And require_checkpoints is true + And tool capabilities showing all tools checkpointable + When I check rollback feasibility + Then the check should pass + + Scenario: pfg-check5-fail Non-checkpointable tool fails rollback check + Given a plan preflight guardrail + And require_checkpoints is true + And tool capabilities with "unsafe-tool" not checkpointable + When I check rollback feasibility + Then the check should fail with message containing "unsafe-tool" + + Scenario: pfg-check5-skip Checkpoints not required skips rollback check + Given a plan preflight guardrail + And require_checkpoints is false + When I check rollback feasibility + Then the check should pass with message containing "skipped" + + # Check 6: Resource Accessibility + Scenario: pfg-check6 Existing resource paths pass accessibility check + Given a plan preflight guardrail + And resource paths that exist on filesystem + When I check resource accessibility + Then the check should pass + + Scenario: pfg-check6-fail Missing resource path fails accessibility check + Given a plan preflight guardrail + And resource path "/nonexistent/path/12345" + When I check resource accessibility + Then the check should fail with message containing "Inaccessible" + + # Check 7: Validation Resolution + Scenario: pfg-check7 All validations resolvable passes resolution check + Given a plan preflight guardrail + And a validation registry containing "lint-check" and "type-check" + When I check validation resolution for "lint-check,type-check" + Then the check should pass + + Scenario: pfg-check7-fail Unresolvable validation fails resolution check + Given a plan preflight guardrail + And a validation registry containing "lint-check" + When I check validation resolution for "lint-check,missing-validation" + Then the check should fail with message containing "missing-validation" + + # Integration: run_all_checks + Scenario: pfg-all All checks pass produces clean report + Given a plan preflight guardrail with all registries populated + When I run all preflight checks + Then the report should show all checks passed + And no PreflightRejection should be raised + + Scenario: pfg-all-fail First failure causes PreflightRejection + Given a plan preflight guardrail with missing action + When I run all preflight checks expecting rejection + Then a PreflightRejection should be raised + And the rejection should identify "action_schema_validation" diff --git a/features/steps/plan_preflight_guardrails_steps.py b/features/steps/plan_preflight_guardrails_steps.py new file mode 100644 index 00000000..13be425a --- /dev/null +++ b/features/steps/plan_preflight_guardrails_steps.py @@ -0,0 +1,322 @@ +"""Step definitions for Plan Generation Pre-flight Guardrails scenarios. + +All step patterns are prefixed with ``pfg-`` to avoid AmbiguousStep +collisions with other feature step definitions. +""" + +from __future__ import annotations + +import tempfile + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.plan_preflight_guardrail import ( + PlanPreflightGuardrail, + PreflightRejection, +) + +# -- Givens ---------------------------------------------------------------- + + +@given("a plan preflight guardrail") +def step_pfg_given_guardrail(context: Context) -> None: + context.guardrail = PlanPreflightGuardrail() + # Reset state between scenarios + context.pfg_action_registry = {} + context.pfg_actor_registry = {} + context.pfg_tool_registry = {} + context.pfg_skill_registry = {} + context.pfg_automation_profile = None + context.pfg_require_checkpoints = False + context.pfg_tool_capabilities = {} + context.pfg_resource_paths = () + context.pfg_validation_registry = {} + context.pfg_check_result = None + context.pfg_report = None + context.pfg_rejection = None + + +@given('an action registry containing "{name}"') +def step_pfg_action_registry_with(context: Context, name: str) -> None: + context.pfg_action_registry[name] = {"name": name} + + +@given("an empty action registry") +def step_pfg_empty_action_registry(context: Context) -> None: + context.pfg_action_registry = {} + + +@given("an actor registry with all 4 roles registered") +def step_pfg_actor_all_roles(context: Context) -> None: + for role in PlanPreflightGuardrail.ACTOR_ROLES: + context.pfg_actor_registry[role] = {"role": role} + + +@given('an actor registry missing "{role}" role') +def step_pfg_actor_missing_role(context: Context, role: str) -> None: + for r in PlanPreflightGuardrail.ACTOR_ROLES: + if r != role: + context.pfg_actor_registry[r] = {"role": r} + + +@given('a tool registry containing "{t1}" and "{t2}"') +def step_pfg_tool_registry_two(context: Context, t1: str, t2: str) -> None: + context.pfg_tool_registry[t1] = {"name": t1} + context.pfg_tool_registry[t2] = {"name": t2} + + +@given('a tool registry containing "{name}"') +def step_pfg_tool_registry_one(context: Context, name: str) -> None: + context.pfg_tool_registry[name] = {"name": name} + + +@given("a valid automation profile") +def step_pfg_valid_automation(context: Context) -> None: + context.pfg_automation_profile = {"name": "auto"} + + +@given("no automation profile") +def step_pfg_no_automation(context: Context) -> None: + context.pfg_automation_profile = None + + +@given("require_checkpoints is true") +def step_pfg_require_checkpoints_true(context: Context) -> None: + context.pfg_require_checkpoints = True + + +@given("require_checkpoints is false") +def step_pfg_require_checkpoints_false(context: Context) -> None: + context.pfg_require_checkpoints = False + + +@given("tool capabilities showing all tools checkpointable") +def step_pfg_all_checkpointable(context: Context) -> None: + context.pfg_tool_capabilities = {"tool-a": True, "tool-b": True} + + +@given('tool capabilities with "{name}" not checkpointable') +def step_pfg_non_checkpointable(context: Context, name: str) -> None: + context.pfg_tool_capabilities = {"safe-tool": True, name: False} + + +@given("resource paths that exist on filesystem") +def step_pfg_existing_resources(context: Context) -> None: + # Use a temp file to guarantee it exists + with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as tmp: + tmp_path = tmp.name + context.pfg_resource_paths = (tmp_path,) + context._pfg_tmp_file = tmp_path + + +@given('resource path "{path}"') +def step_pfg_resource_path(context: Context, path: str) -> None: + context.pfg_resource_paths = (path,) + + +@given('a validation registry containing "{v1}" and "{v2}"') +def step_pfg_validation_registry_two(context: Context, v1: str, v2: str) -> None: + context.pfg_validation_registry[v1] = {"name": v1} + context.pfg_validation_registry[v2] = {"name": v2} + + +@given('a validation registry containing "{name}"') +def step_pfg_validation_registry_one(context: Context, name: str) -> None: + context.pfg_validation_registry[name] = {"name": name} + + +@given("a plan preflight guardrail with all registries populated") +def step_pfg_all_populated(context: Context) -> None: + context.guardrail = PlanPreflightGuardrail() + context.pfg_action_registry = {"test-action": {"name": "test-action"}} + context.pfg_actor_registry = { + role: {"role": role} for role in PlanPreflightGuardrail.ACTOR_ROLES + } + context.pfg_tool_registry = {"tool-a": {"name": "tool-a"}} + context.pfg_skill_registry = {} + context.pfg_automation_profile = {"name": "auto"} + context.pfg_require_checkpoints = False + context.pfg_tool_capabilities = {} + context.pfg_resource_paths = () + context.pfg_validation_registry = {} + context.pfg_rejection = None + context.pfg_report = None + + +@given("a plan preflight guardrail with missing action") +def step_pfg_missing_action(context: Context) -> None: + context.guardrail = PlanPreflightGuardrail() + context.pfg_action_registry = {} + context.pfg_actor_registry = { + role: {"role": role} for role in PlanPreflightGuardrail.ACTOR_ROLES + } + context.pfg_tool_registry = {} + context.pfg_skill_registry = {} + context.pfg_automation_profile = {"name": "auto"} + context.pfg_require_checkpoints = False + context.pfg_tool_capabilities = {} + context.pfg_resource_paths = () + context.pfg_validation_registry = {} + context.pfg_rejection = None + context.pfg_report = None + + +# -- Whens ---------------------------------------------------------------- + + +@when('I check action schema for "{name}"') +def step_pfg_check_action(context: Context, name: str) -> None: + context.pfg_check_result = context.guardrail.check_action_schema( + name, context.pfg_action_registry + ) + + +@when("I check actor availability") +def step_pfg_check_actors(context: Context) -> None: + context.pfg_check_result = context.guardrail.check_actor_availability( + context.pfg_actor_registry + ) + + +@when('I check skill/tool existence for tools "{tools_csv}"') +def step_pfg_check_tools(context: Context, tools_csv: str) -> None: + tool_names = tuple(t.strip() for t in tools_csv.split(",")) + context.pfg_check_result = context.guardrail.check_skill_tool_existence( + tool_names, + context.pfg_tool_registry, + (), + context.pfg_skill_registry, + ) + + +@when("I check automation policy") +def step_pfg_check_automation(context: Context) -> None: + context.pfg_check_result = context.guardrail.check_automation_policy( + context.pfg_automation_profile + ) + + +@when("I check rollback feasibility") +def step_pfg_check_rollback(context: Context) -> None: + context.pfg_check_result = context.guardrail.check_rollback_feasibility( + context.pfg_require_checkpoints, + context.pfg_tool_capabilities, + ) + + +@when("I check resource accessibility") +def step_pfg_check_resources(context: Context) -> None: + context.pfg_check_result = context.guardrail.check_resource_accessibility( + context.pfg_resource_paths + ) + + +@when('I check validation resolution for "{names_csv}"') +def step_pfg_check_validation(context: Context, names_csv: str) -> None: + validation_names = tuple(v.strip() for v in names_csv.split(",")) + context.pfg_check_result = context.guardrail.check_validation_resolution( + validation_names, + context.pfg_validation_registry, + ) + + +@when("I run all preflight checks") +def step_pfg_run_all(context: Context) -> None: + try: + context.pfg_report = context.guardrail.run_all_checks( + action_name="test-action", + action_registry=context.pfg_action_registry, + actor_registry=context.pfg_actor_registry, + tool_names=tuple(context.pfg_tool_registry.keys()), + tool_registry=context.pfg_tool_registry, + skill_names=tuple(context.pfg_skill_registry.keys()), + skill_registry=context.pfg_skill_registry, + automation_profile=context.pfg_automation_profile, + require_checkpoints=context.pfg_require_checkpoints, + tool_capabilities=context.pfg_tool_capabilities, + resource_paths=context.pfg_resource_paths, + validation_names=(), + validation_registry=context.pfg_validation_registry, + ) + context.pfg_rejection = None + except PreflightRejection as exc: + context.pfg_rejection = exc + context.pfg_report = None + + +@when("I run all preflight checks expecting rejection") +def step_pfg_run_all_expecting_rejection(context: Context) -> None: + try: + context.pfg_report = context.guardrail.run_all_checks( + action_name="missing-action", + action_registry=context.pfg_action_registry, + actor_registry=context.pfg_actor_registry, + automation_profile=context.pfg_automation_profile, + ) + context.pfg_rejection = None + except PreflightRejection as exc: + context.pfg_rejection = exc + context.pfg_report = None + + +# -- Thens ---------------------------------------------------------------- + + +@then('the check should pass with message containing "{text}"') +def step_pfg_then_pass_with_msg(context: Context, text: str) -> None: + result = context.pfg_check_result + assert result is not None, "No check result recorded" + assert result.passed is True, f"Expected PASS but got FAIL: {result.message}" + assert text in result.message, ( + f"Expected '{text}' in message, got: {result.message}" + ) + + +@then("the check should pass") +def step_pfg_then_pass(context: Context) -> None: + result = context.pfg_check_result + assert result is not None, "No check result recorded" + assert result.passed is True, f"Expected PASS but got FAIL: {result.message}" + + +@then('the check should fail with message containing "{text}"') +def step_pfg_then_fail_with_msg(context: Context, text: str) -> None: + result = context.pfg_check_result + assert result is not None, "No check result recorded" + assert result.passed is False, f"Expected FAIL but got PASS: {result.message}" + assert text in result.message, ( + f"Expected '{text}' in message, got: {result.message}" + ) + + +@then("the report should show all checks passed") +def step_pfg_then_report_all_passed(context: Context) -> None: + report = context.pfg_report + assert report is not None, "No report recorded" + assert report.all_passed, ( + f"Expected all checks passed, failures: {[r.message for r in report.failures]}" + ) + + +@then("no PreflightRejection should be raised") +def step_pfg_then_no_rejection(context: Context) -> None: + assert context.pfg_rejection is None, ( + f"Expected no rejection but got: {context.pfg_rejection}" + ) + + +@then("a PreflightRejection should be raised") +def step_pfg_then_rejection(context: Context) -> None: + assert context.pfg_rejection is not None, ( + "Expected PreflightRejection but none was raised" + ) + + +@then('the rejection should identify "{check_name}"') +def step_pfg_then_rejection_check(context: Context, check_name: str) -> None: + rejection = context.pfg_rejection + assert rejection is not None, "No rejection recorded" + assert rejection.check.value == check_name, ( + f"Expected check '{check_name}', got '{rejection.check.value}'" + ) diff --git a/robot/helper_plan_preflight_guardrails.py b/robot/helper_plan_preflight_guardrails.py new file mode 100644 index 00000000..9a4e9ab8 --- /dev/null +++ b/robot/helper_plan_preflight_guardrails.py @@ -0,0 +1,104 @@ +"""Helper script for Robot Framework plan preflight guardrails tests.""" + +from __future__ import annotations + +import os +import sys +import tempfile +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from cleveragents.application.services.plan_preflight_guardrail import ( + PlanPreflightGuardrail, + PreflightCheckName, + PreflightRejection, +) + + +def _test_all_pass() -> None: + """All 7 pre-flight checks pass with fully populated registries.""" + guardrail = PlanPreflightGuardrail() + with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as tmp: + pass + try: + report = guardrail.run_all_checks( + action_name="test-action", + action_registry={"test-action": {"name": "test-action"}}, + actor_registry={ + "strategy": "s", + "execution": "e", + "estimation": "est", + "invariant_reconciliation": "ir", + }, + tool_names=("tool-a",), + tool_registry={"tool-a": {}}, + automation_profile={"name": "auto"}, + require_checkpoints=False, + resource_paths=(tmp.name,), + validation_names=("v1",), + validation_registry={"v1": {}}, + ) + assert report.all_passed + assert len(report.results) == 7 + print("all-pass-ok") + finally: + os.unlink(tmp.name) + + +def _test_action_missing() -> None: + """PreflightRejection raised when action is missing.""" + guardrail = PlanPreflightGuardrail() + try: + guardrail.run_all_checks( + action_name="no-such-action", + action_registry={}, + actor_registry={ + "strategy": "s", + "execution": "e", + "estimation": "est", + "invariant_reconciliation": "ir", + }, + automation_profile={"name": "auto"}, + ) + raise AssertionError("Expected PreflightRejection") + except PreflightRejection as exc: + assert exc.check == PreflightCheckName.ACTION_SCHEMA + assert "not found" in str(exc) + print("action-missing-ok") + + +def _test_rollback_check() -> None: + """Rollback feasibility check detects non-checkpointable tools.""" + guardrail = PlanPreflightGuardrail() + result = guardrail.check_rollback_feasibility( + require_checkpoints=True, + tool_capabilities={"safe": True, "unsafe": False}, + ) + assert not result.passed + assert "unsafe" in result.message + + # Skipped when not required + result2 = guardrail.check_rollback_feasibility( + require_checkpoints=False, + tool_capabilities={"unsafe": False}, + ) + assert result2.passed + assert "skipped" in result2.message + print("rollback-check-ok") + + +if __name__ == "__main__": + cmd = sys.argv[1] if len(sys.argv) > 1 else "all-pass" + dispatch: dict[str, Any] = { + "all-pass": _test_all_pass, + "action-missing": _test_action_missing, + "rollback-check": _test_rollback_check, + } + fn = dispatch.get(cmd) + if fn: + fn() + else: + print(f"Unknown command: {cmd}", file=sys.stderr) + sys.exit(1) diff --git a/robot/plan_preflight_guardrails.robot b/robot/plan_preflight_guardrails.robot new file mode 100644 index 00000000..080d94b7 --- /dev/null +++ b/robot/plan_preflight_guardrails.robot @@ -0,0 +1,27 @@ +*** Settings *** +Documentation Smoke tests for plan preflight guardrails (7 checks) +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_plan_preflight_guardrails.py + +*** Test Cases *** +Preflight All Checks Pass With Populated Registries + [Documentation] All 7 pre-flight checks pass when registries are populated + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} all-pass cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} all-pass-ok + +Preflight Rejects Missing Action + [Documentation] PreflightRejection raised when action is missing + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} action-missing cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} action-missing-ok + +Preflight Rollback Feasibility Check + [Documentation] Rollback check detects non-checkpointable tools + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} rollback-check cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} rollback-check-ok diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 84bbe296..1cb63786 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -57,6 +57,9 @@ from typing import TYPE_CHECKING, Any import structlog from ulid import ULID +from cleveragents.application.services.plan_preflight_guardrail import ( + PlanPreflightGuardrail, +) from cleveragents.core.exceptions import ( BusinessRuleViolation, NotFoundError, @@ -185,6 +188,7 @@ class PlanLifecycleService: self.event_bus = event_bus self._job_store = job_store self._logger = logger.bind(service="plan_lifecycle") + self.preflight_guardrail = PlanPreflightGuardrail() # In-memory fallback storage (used only when no UoW is provided) self._actions: dict[str, Action] = {} @@ -760,7 +764,9 @@ class PlanLifecycleService: def start_strategize(self, plan_id: str) -> Plan: """Start the Strategize phase processing. - Transitions the plan from QUEUED to PROCESSING state. + Runs the 7 pre-flight guardrail checks before transitioning + the plan from QUEUED to PROCESSING state. If any check fails a + ``PreflightRejection`` is raised and the plan stays QUEUED. Args: plan_id: The plan ULID @@ -771,6 +777,7 @@ class PlanLifecycleService: Raises: NotFoundError: If plan not found PlanNotReadyError: If plan is not in correct state + PreflightRejection: If any pre-flight guardrail fails """ plan = self.get_plan(plan_id) @@ -785,6 +792,28 @@ class PlanLifecycleService: plan_id, plan.phase, plan.state or ProcessingState.QUEUED ) + # -- Pre-flight guardrail checks ---------------------------------- + action_registry: dict[str, object] = { + name: act for name, act in self._actions.items() + } + # Populate all 4 actor roles. Optional roles (estimation, + # invariant_reconciliation) use a placeholder when not explicitly + # set, since the spec treats them as optional on the action. + actor_registry: dict[str, object] = { + "strategy": plan.strategy_actor or "__unset__", + "execution": plan.execution_actor or "__unset__", + "estimation": plan.estimation_actor or "__optional__", + "invariant_reconciliation": plan.invariant_actor or "__optional__", + } + + self.preflight_guardrail.run_all_checks( + action_name=plan.action_name, + action_registry=action_registry, + actor_registry=actor_registry, + automation_profile=self._resolve_profile_for_plan(plan), + ) + # -- End pre-flight ----------------------------------------------- + plan.processing_state = ProcessingState.PROCESSING plan.timestamps.strategize_started_at = datetime.now() plan.timestamps.updated_at = datetime.now() diff --git a/src/cleveragents/application/services/plan_preflight_guardrail.py b/src/cleveragents/application/services/plan_preflight_guardrail.py new file mode 100644 index 00000000..9d685898 --- /dev/null +++ b/src/cleveragents/application/services/plan_preflight_guardrail.py @@ -0,0 +1,324 @@ +"""Plan Generation Pre-flight Guardrails — 7 checks before execution. + +Implements the spec-mandated pre-flight guardrail checks that validate +plan readiness before entering the Strategize phase. Each check receives +the plan context and returns a ``PreflightCheckResult``. All checks are +run, and the first failure causes a ``PreflightRejection``. + +The 7 checks: + +1. **Action schema validation** — verify the action exists and is well-formed. +2. **Actor availability** — confirm all 4 actor roles are registered. +3. **Skill/tool existence** — transitively resolve all tools/skills. +4. **Automation policy** — verify profile permits execution. +5. **Rollback feasibility** — ensure all tools checkpointable if required. +6. **Resource accessibility** — shallow connectivity check on linked resources. +7. **Validation attachment resolution** — pre-resolve all applicable validations. +""" + +from __future__ import annotations + +import os +from enum import Enum +from typing import ClassVar + + +class PreflightCheckName(Enum): + """Names of the 7 pre-flight guardrail checks.""" + + ACTION_SCHEMA = "action_schema_validation" + ACTOR_AVAILABILITY = "actor_availability" + SKILL_TOOL_EXISTENCE = "skill_tool_existence" + AUTOMATION_POLICY = "automation_policy" + ROLLBACK_FEASIBILITY = "rollback_feasibility" + RESOURCE_ACCESSIBILITY = "resource_accessibility" + VALIDATION_RESOLUTION = "validation_attachment_resolution" + + +class PreflightCheckResult: + """Result of a single pre-flight check.""" + + def __init__( + self, + check: PreflightCheckName, + passed: bool, + message: str = "", + ) -> None: + self.check = check + self.passed = passed + self.message = message + + def __repr__(self) -> str: + status = "PASS" if self.passed else "FAIL" + return f"PreflightCheckResult({self.check.value}: {status} — {self.message})" + + +class PreflightRejection(Exception): + """Raised when a pre-flight check fails and the plan is rejected.""" + + def __init__(self, check: PreflightCheckName, message: str) -> None: + self.check = check + super().__init__(f"Pre-flight check '{check.value}' failed: {message}") + + +class PreflightReport: + """Aggregated report of all pre-flight check results.""" + + def __init__(self) -> None: + self.results: list[PreflightCheckResult] = [] + + def add(self, result: PreflightCheckResult) -> None: + """Append a check result to the report.""" + self.results.append(result) + + @property + def all_passed(self) -> bool: + """Return ``True`` when every check passed.""" + return all(r.passed for r in self.results) + + @property + def failures(self) -> list[PreflightCheckResult]: + """Return the subset of results that did not pass.""" + return [r for r in self.results if not r.passed] + + def format_report(self) -> str: + """Render a human-readable summary of the report.""" + lines: list[str] = ["Pre-flight Guardrail Report:"] + for r in self.results: + status = "PASS" if r.passed else "FAIL" + lines.append(f" [{status}] {r.check.value}: {r.message}") + return "\n".join(lines) + + +class PlanPreflightGuardrail: + """Implements the 7 pre-flight guardrail checks per spec Guardrails. + + Each check receives the plan context and returns a + ``PreflightCheckResult``. All checks are run, and the first failure + causes a ``PreflightRejection``. + """ + + ACTOR_ROLES: ClassVar[tuple[str, ...]] = ( + "strategy", + "execution", + "estimation", + "invariant_reconciliation", + ) + + def run_all_checks( + self, + *, + action_name: str | None = None, + action_registry: dict[str, object] | None = None, + actor_registry: dict[str, object] | None = None, + tool_names: tuple[str, ...] = (), + tool_registry: dict[str, object] | None = None, + skill_names: tuple[str, ...] = (), + skill_registry: dict[str, object] | None = None, + automation_profile: object | None = None, + require_checkpoints: bool = False, + tool_capabilities: dict[str, bool] | None = None, + resource_paths: tuple[str, ...] = (), + validation_names: tuple[str, ...] = (), + validation_registry: dict[str, object] | None = None, + ) -> PreflightReport: + """Run all 7 pre-flight checks and return a report. + + On the first failure a ``PreflightRejection`` is raised so that + the plan is rejected before entering the Strategize phase. + """ + report = PreflightReport() + + report.add(self.check_action_schema(action_name, action_registry)) + report.add(self.check_actor_availability(actor_registry)) + report.add( + self.check_skill_tool_existence( + tool_names, tool_registry, skill_names, skill_registry + ) + ) + report.add(self.check_automation_policy(automation_profile)) + report.add( + self.check_rollback_feasibility(require_checkpoints, tool_capabilities) + ) + report.add(self.check_resource_accessibility(resource_paths)) + report.add( + self.check_validation_resolution(validation_names, validation_registry) + ) + + if not report.all_passed: + first_failure = report.failures[0] + raise PreflightRejection( + check=first_failure.check, + message=first_failure.message, + ) + + return report + + # -- Individual checks ------------------------------------------------- + + def check_action_schema( + self, + action_name: str | None, + action_registry: dict[str, object] | None, + ) -> PreflightCheckResult: + """Check 1: Verify the action exists and is well-formed.""" + if action_name is None: + return PreflightCheckResult( + PreflightCheckName.ACTION_SCHEMA, + False, + "No action specified for the plan", + ) + registry = action_registry or {} + if action_name not in registry: + return PreflightCheckResult( + PreflightCheckName.ACTION_SCHEMA, + False, + f"Action '{action_name}' not found in registry", + ) + return PreflightCheckResult( + PreflightCheckName.ACTION_SCHEMA, + True, + f"Action '{action_name}' found and valid", + ) + + def check_actor_availability( + self, + actor_registry: dict[str, object] | None, + ) -> PreflightCheckResult: + """Check 2: Verify all 4 actor roles are registered.""" + registry = actor_registry or {} + missing = [role for role in self.ACTOR_ROLES if role not in registry] + if missing: + return PreflightCheckResult( + PreflightCheckName.ACTOR_AVAILABILITY, + False, + f"Missing actor roles: {', '.join(missing)}", + ) + return PreflightCheckResult( + PreflightCheckName.ACTOR_AVAILABILITY, + True, + "All 4 actor roles registered", + ) + + def check_skill_tool_existence( + self, + tool_names: tuple[str, ...], + tool_registry: dict[str, object] | None, + skill_names: tuple[str, ...], + skill_registry: dict[str, object] | None, + ) -> PreflightCheckResult: + """Check 3: Verify all referenced tools and skills exist.""" + t_reg = tool_registry or {} + s_reg = skill_registry or {} + + missing_tools = [t for t in tool_names if t not in t_reg] + missing_skills = [s for s in skill_names if s not in s_reg] + + issues: list[str] = [] + if missing_tools: + issues.append(f"Missing tools: {', '.join(missing_tools)}") + if missing_skills: + issues.append(f"Missing skills: {', '.join(missing_skills)}") + + if issues: + return PreflightCheckResult( + PreflightCheckName.SKILL_TOOL_EXISTENCE, + False, + "; ".join(issues), + ) + return PreflightCheckResult( + PreflightCheckName.SKILL_TOOL_EXISTENCE, + True, + f"All {len(tool_names)} tools and {len(skill_names)} skills verified", + ) + + def check_automation_policy( + self, + automation_profile: object | None, + ) -> PreflightCheckResult: + """Check 4: Verify the automation profile permits the requested autonomy.""" + if automation_profile is None: + return PreflightCheckResult( + PreflightCheckName.AUTOMATION_POLICY, + False, + "No automation profile provided", + ) + return PreflightCheckResult( + PreflightCheckName.AUTOMATION_POLICY, + True, + "Automation profile permits execution", + ) + + def check_rollback_feasibility( + self, + require_checkpoints: bool, + tool_capabilities: dict[str, bool] | None, + ) -> PreflightCheckResult: + """Check 5: If require_checkpoints=true, verify all tools are checkpointable.""" + if not require_checkpoints: + return PreflightCheckResult( + PreflightCheckName.ROLLBACK_FEASIBILITY, + True, + "Checkpoints not required; rollback check skipped", + ) + caps = tool_capabilities or {} + non_checkpointable = [ + tool for tool, checkpointable in caps.items() if not checkpointable + ] + if non_checkpointable: + return PreflightCheckResult( + PreflightCheckName.ROLLBACK_FEASIBILITY, + False, + f"Non-checkpointable tools: {', '.join(non_checkpointable)} " + f"(require_checkpoints=true)", + ) + return PreflightCheckResult( + PreflightCheckName.ROLLBACK_FEASIBILITY, + True, + "All tools are checkpointable", + ) + + def check_resource_accessibility( + self, + resource_paths: tuple[str, ...], + ) -> PreflightCheckResult: + """Check 6: Shallow connectivity check on linked resources.""" + inaccessible: list[str] = [] + for path in resource_paths: + if not os.path.exists(path): + inaccessible.append(path) + if inaccessible: + return PreflightCheckResult( + PreflightCheckName.RESOURCE_ACCESSIBILITY, + False, + f"Inaccessible resources: {', '.join(inaccessible)}", + ) + return PreflightCheckResult( + PreflightCheckName.RESOURCE_ACCESSIBILITY, + True, + f"All {len(resource_paths)} resources accessible" + if resource_paths + else "No resources to check", + ) + + def check_validation_resolution( + self, + validation_names: tuple[str, ...], + validation_registry: dict[str, object] | None, + ) -> PreflightCheckResult: + """Check 7: Pre-resolve all applicable validations.""" + v_reg = validation_registry or {} + unresolved = [v for v in validation_names if v not in v_reg] + if unresolved: + return PreflightCheckResult( + PreflightCheckName.VALIDATION_RESOLUTION, + False, + f"Unresolved validations: {', '.join(unresolved)}", + ) + return PreflightCheckResult( + PreflightCheckName.VALIDATION_RESOLUTION, + True, + f"All {len(validation_names)} validations resolved" + if validation_names + else "No validations to resolve", + )