feat(guardrails): implement Plan Generation Pre-flight Guardrails (7 checks before execution)

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

@@ -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()

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"

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}'"
+    )

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)

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

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()

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",
+        )