robot/fixtures/wf08/cloud-metrics.json

@@ -0,0 +1,44 @@
+{
+  "metrics": [
+    {
+      "resource_id": "i-0abc123def456",
+      "resource_type": "aws_instance",
+      "resource_name": "web",
+      "metric": "CPUUtilization",
+      "period_days": 30,
+      "average": 12.3,
+      "peak": 28.7,
+      "unit": "Percent"
+    },
+    {
+      "resource_id": "i-0abc123def456",
+      "resource_type": "aws_instance",
+      "resource_name": "web",
+      "metric": "NetworkIn",
+      "period_days": 30,
+      "average": 1024000,
+      "peak": 5120000,
+      "unit": "Bytes"
+    },
+    {
+      "resource_id": "i-0def456ghi789",
+      "resource_type": "aws_instance",
+      "resource_name": "api",
+      "metric": "CPUUtilization",
+      "period_days": 30,
+      "average": 65.2,
+      "peak": 89.1,
+      "unit": "Percent"
+    },
+    {
+      "resource_id": "prod-logs-bucket",
+      "resource_type": "aws_s3_bucket",
+      "resource_name": "logs",
+      "metric": "BucketSizeBytes",
+      "period_days": 30,
+      "average": 10737418240,
+      "peak": 10737418240,
+      "unit": "Bytes"
+    }
+  ]
+}

robot/fixtures/wf08/main.tf

@@ -0,0 +1,52 @@
+terraform {
+  required_version = ">= 1.5.0"
+
+  required_providers {
+    aws = {
+      source  = "hashicorp/aws"
+      version = "~> 5.0"
+    }
+  }
+
+  backend "s3" {
+    bucket = "terraform-state-prod"
+    key    = "infrastructure/terraform.tfstate"
+    region = "us-east-1"
+  }
+}
+
+provider "aws" {
+  region = "us-east-1"
+}
+
+resource "aws_instance" "web" {
+  ami           = "ami-0c55b159cbfafe1f0"
+  instance_type = "m5.2xlarge"
+
+  tags = {
+    Name        = "web-server"
+    Environment = "production"
+    critical    = "false"
+  }
+}
+
+resource "aws_instance" "api" {
+  ami           = "ami-0c55b159cbfafe1f0"
+  instance_type = "m5.xlarge"
+
+  tags = {
+    Name        = "api-server"
+    Environment = "production"
+    critical    = "true"
+  }
+}
+
+resource "aws_s3_bucket" "logs" {
+  bucket = "prod-logs-bucket"
+
+  tags = {
+    Name        = "log-storage"
+    Environment = "production"
+    critical    = "false"
+  }
+}

robot/fixtures/wf08/terraform-state.yaml

@@ -0,0 +1,23 @@
+name: local/terraform-state
+description: "Terraform state file resource type"
+resource_kind: physical
+sandbox_strategy: copy_on_write
+user_addable: true
+handler: local/terraform-state-handler
+capabilities:
+  read: true
+  write: true
+  sandbox: true
+  checkpoint: true
+cli_args:
+  - name: state-path
+    type: string
+    required: true
+    description: "Path to terraform.tfstate or remote state configuration"
+  - name: workspace
+    type: string
+    required: false
+    default: "default"
+    description: "Terraform workspace name"
+child_types:
+  - fs-file

robot/fixtures/wf08/terraform.tfstate.json

@@ -0,0 +1,70 @@
+{
+  "version": 4,
+  "terraform_version": "1.5.7",
+  "serial": 42,
+  "lineage": "abc123-def456-ghi789",
+  "outputs": {},
+  "resources": [
+    {
+      "mode": "managed",
+      "type": "aws_instance",
+      "name": "web",
+      "provider": "provider[\"registry.terraform.io/hashicorp/aws\"]",
+      "instances": [
+        {
+          "attributes": {
+            "id": "i-0abc123def456",
+            "instance_type": "m5.2xlarge",
+            "tags": {
+              "Name": "web-server",
+              "Environment": "production",
+              "critical": "false"
+            },
+            "cpu_core_count": 8,
+            "memory": 32768
+          }
+        }
+      ]
+    },
+    {
+      "mode": "managed",
+      "type": "aws_instance",
+      "name": "api",
+      "provider": "provider[\"registry.terraform.io/hashicorp/aws\"]",
+      "instances": [
+        {
+          "attributes": {
+            "id": "i-0def456ghi789",
+            "instance_type": "m5.xlarge",
+            "tags": {
+              "Name": "api-server",
+              "Environment": "production",
+              "critical": "true"
+            },
+            "cpu_core_count": 4,
+            "memory": 16384
+          }
+        }
+      ]
+    },
+    {
+      "mode": "managed",
+      "type": "aws_s3_bucket",
+      "name": "logs",
+      "provider": "provider[\"registry.terraform.io/hashicorp/aws\"]",
+      "instances": [
+        {
+          "attributes": {
+            "id": "prod-logs-bucket",
+            "bucket": "prod-logs-bucket",
+            "tags": {
+              "Name": "log-storage",
+              "Environment": "production",
+              "critical": "false"
+            }
+          }
+        }
+      ]
+    }
+  ]
+}

robot/helper_wf08_cloud_infra_supervised.py

@@ -0,0 +1,672 @@
+"""Helper for Workflow Example 8: Cloud Infrastructure Management.
+
+Integration test exercising the ``supervised`` automation profile with:
+- Custom resource type (local/terraform-state) registration (Step 1)
+- Custom skill (local/terraform-ops) with tool composition (Step 1)
+- Supervised automation profile behavior verification
+- Action creation with invariants + plan use (Step 3)
+- Infrastructure analysis with mocked Terraform operations
+- Invariant enforcement verification
+
+Spec reference: docs/specification.md, Example 8 (~line 39315)
+Issue: #772
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# Shared DB setup
+# ---------------------------------------------------------------------------
+
+
+def _setup_db() -> Any:
+    """Create in-memory SQLite, return session factory."""
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    from cleveragents.infrastructure.database.models import Base
+
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine, expire_on_commit=False)()
+
+    class _NoClose:
+        """Keep in-memory SQLite alive across session.close() calls."""
+
+        __slots__ = ("_s",)
+
+        def __init__(self, s: object) -> None:
+            object.__setattr__(self, "_s", s)
+
+        def close(self) -> None:
+            pass
+
+        def __getattr__(self, n: str) -> object:
+            return getattr(object.__getattribute__(self, "_s"), n)
+
+    wrapper = _NoClose(session)
+    return lambda: wrapper
+
+
+# ---------------------------------------------------------------------------
+# Fixture directory
+# ---------------------------------------------------------------------------
+
+_FIXTURES = Path(__file__).resolve().parent / "fixtures" / "wf08"
+
+
+# ---------------------------------------------------------------------------
+# Mock Terraform operations (deterministic, no real Terraform needed)
+# ---------------------------------------------------------------------------
+
+
+def _mock_terraform_show(resource_address: str) -> dict[str, Any]:
+    """Mock ``terraform show`` returning resource state from fixture."""
+    tfstate_path = _FIXTURES / "terraform.tfstate.json"
+    state = json.loads(tfstate_path.read_text())
+    for resource in state.get("resources", []):
+        addr = f"{resource['type']}.{resource['name']}"
+        if addr == resource_address:
+            return {
+                "address": addr,
+                "type": resource["type"],
+                "name": resource["name"],
+                "values": resource["instances"][0]["attributes"],
+            }
+    return {"error": f"Resource {resource_address} not found"}
+
+
+def _mock_cloud_metrics(resource_id: str, metric: str) -> dict[str, Any]:
+    """Mock ``cloud_metrics`` fetching monitoring data from fixture."""
+    metrics_path = _FIXTURES / "cloud-metrics.json"
+    data = json.loads(metrics_path.read_text())
+    for entry in data.get("metrics", []):
+        if entry["resource_id"] == resource_id and entry["metric"] == metric:
+            return entry
+    return {"error": f"No metric {metric} for {resource_id}"}
+
+
+def _mock_terraform_plan(target: str | None = None) -> dict[str, Any]:
+    """Mock ``terraform plan`` returning a plan summary."""
+    changes: list[dict[str, str]] = []
+    if target is None or target == "aws_instance.web":
+        changes.append(
+            {
+                "action": "update",
+                "resource": "aws_instance.web",
+                "detail": "instance_type: m5.2xlarge -> m5.large (right-sizing)",
+            }
+        )
+    return {
+        "plan_summary": {
+            "add": 0,
+            "change": len(changes),
+            "destroy": 0,
+        },
+        "changes": changes,
+    }
+
+
+def _analyze_infrastructure() -> dict[str, Any]:
+    """Run mocked infrastructure analysis producing optimization recommendations.
+
+    Simulates the LLM-driven analysis of Terraform state and cloud metrics
+    to produce cost-saving recommendations.
+    """
+    # Step 1: Show current web instance state
+    web_state = _mock_terraform_show("aws_instance.web")
+    assert "error" not in web_state, f"Failed to show web instance: {web_state}"
+
+    # Step 2: Get CPU metrics for the web instance
+    web_cpu = _mock_cloud_metrics("i-0abc123def456", "CPUUtilization")
+    assert "error" not in web_cpu, f"Failed to get web CPU metrics: {web_cpu}"
+
+    # Step 3: Check API instance (should NOT be recommended for changes — critical)
+    api_state = _mock_terraform_show("aws_instance.api")
+    assert api_state["values"]["tags"]["critical"] == "true"
+
+    # Step 4: Generate plan for right-sizing
+    plan = _mock_terraform_plan("aws_instance.web")
+
+    # Step 5: Build optimization recommendations
+    recommendations: list[dict[str, Any]] = []
+
+    # Web instance is over-provisioned: avg CPU 12.3%, peak 28.7% on m5.2xlarge
+    if web_cpu["average"] < 30.0 and web_cpu["peak"] < 70.0:
+        recommendations.append(
+            {
+                "resource": "aws_instance.web",
+                "type": "right-sizing",
+                "current": web_state["values"]["instance_type"],
+                "recommended": "m5.large",
+                "reason": (
+                    f"Average CPU {web_cpu['average']}%, "
+                    f"peak {web_cpu['peak']}% — "
+                    "well within m5.large capacity"
+                ),
+                "estimated_monthly_savings_usd": 148.0,
+            }
+        )
+
+    return {
+        "recommendations": recommendations,
+        "plan": plan,
+        "analyzed_resources": 3,
+        "skipped_critical": ["aws_instance.api"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# 1. Register custom resource type (Spec Step 1)
+# ---------------------------------------------------------------------------
+
+
+def _register_terraform_resource_type() -> None:
+    """Register local/terraform-state resource type via YAML fixture."""
+    from cleveragents.application.services.resource_registry_service import (
+        ResourceRegistryService,
+    )
+
+    factory = _setup_db()
+    svc = ResourceRegistryService(session_factory=factory)
+
+    yaml_path = str(_FIXTURES / "terraform-state.yaml")
+    spec = svc.register_type(yaml_path)
+
+    assert spec.name == "local/terraform-state", (
+        f"Expected name 'local/terraform-state', got '{spec.name}'"
+    )
+    assert str(spec.sandbox_strategy) == "copy_on_write", (
+        f"Expected sandbox_strategy 'copy_on_write', got '{spec.sandbox_strategy}'"
+    )
+    assert spec.capabilities.get("read") is True
+    assert spec.capabilities.get("write") is True
+    assert spec.capabilities.get("sandbox") is True
+    assert spec.capabilities.get("checkpoint") is True
+
+    # Verify cli_args (state-path required, workspace optional)
+    cli_args = spec.cli_args
+    assert len(cli_args) == 2, f"Expected 2 cli_args, got {len(cli_args)}"
+    state_path_arg = next(a for a in cli_args if a.name == "state-path")
+    assert state_path_arg.required is True
+    workspace_arg = next(a for a in cli_args if a.name == "workspace")
+    assert workspace_arg.required is False
+    assert workspace_arg.default == "default"
+
+    # Verify show round-trip
+    shown = svc.show_type("local/terraform-state")
+    assert shown.name == "local/terraform-state"
+
+    # Register an instance of this type
+    with tempfile.TemporaryDirectory(prefix="wf08-tf-") as tmpdir:
+        resource = svc.register_resource(
+            type_name="local/terraform-state",
+            name="local/prod-tfstate",
+            location=tmpdir,
+            description="Production Terraform state",
+            properties={
+                "state-path": f"{tmpdir}/terraform.tfstate",
+                "workspace": "production",
+            },
+        )
+        assert resource.name == "local/prod-tfstate"
+        assert resource.properties["workspace"] == "production"
+
+    print("register-terraform-resource-type-ok")
+
+
+# ---------------------------------------------------------------------------
+# 2. Register skill with tool composition (Spec Step 1)
+# ---------------------------------------------------------------------------
+
+
+def _register_terraform_skill() -> None:
+    """Register local/terraform-ops skill with 3 tools and include_skills."""
+    from cleveragents.application.services.skill_service import SkillService
+    from cleveragents.domain.models.core.skill import Skill
+
+    skill_svc = SkillService()
+
+    # Build skill from config dict matching spec YAML
+    skill_config: dict[str, Any] = {
+        "name": "local/terraform-ops",
+        "description": "Terraform infrastructure operations",
+        "anonymous_tools": [
+            {
+                "description": "Run terraform plan and return the execution plan",
+                "source": "custom",
+                "input_schema": {
+                    "type": "object",
+                    "properties": {
+                        "target": {
+                            "type": "string",
+                            "description": "Specific resource to target",
+                        },
+                    },
+                    "required": [],
+                },
+                "capability": {
+                    "writes": False,
+                    "checkpointable": False,
+                },
+            },
+            {
+                "description": "Show current state of a Terraform resource",
+                "source": "custom",
+                "input_schema": {
+                    "type": "object",
+                    "properties": {
+                        "resource_address": {
+                            "type": "string",
+                            "description": "Terraform resource address",
+                        },
+                    },
+                    "required": ["resource_address"],
+                },
+                "capability": {
+                    "writes": False,
+                    "checkpointable": False,
+                },
+            },
+            {
+                "description": "Fetch cloud monitoring metrics for a resource",
+                "source": "custom",
+                "input_schema": {
+                    "type": "object",
+                    "properties": {
+                        "resource_id": {"type": "string"},
+                        "metric": {
+                            "type": "string",
+                            "description": "e.g., CPUUtilization",
+                        },
+                        "period_days": {
+                            "type": "integer",
+                            "default": 30,
+                        },
+                    },
+                    "required": ["resource_id", "metric"],
+                },
+                "capability": {
+                    "writes": False,
+                    "checkpointable": False,
+                },
+            },
+        ],
+        # Skill composition — includes local/file-ops per spec
+        "includes": ["local/file-ops"],
+    }
+
+    skill = Skill.from_config(skill_config)
+    assert skill.name == "local/terraform-ops", (
+        f"Expected name 'local/terraform-ops', got '{skill.name}'"
+    )
+    assert skill.description == "Terraform infrastructure operations"
+
+    # Verify 3 anonymous tools (terraform_plan, terraform_show, cloud_metrics)
+    assert len(skill.anonymous_tools) == 3, (
+        f"Expected 3 anonymous tools, got {len(skill.anonymous_tools)}"
+    )
+
+    # Verify all tools are read-only (writes=False)
+    for tool in skill.anonymous_tools:
+        cap = tool.capability
+        assert cap is not None, "Tool capability must not be None"
+        assert cap.writes is False, (
+            f"Tool '{tool.description}' should have writes=False"
+        )
+        assert cap.checkpointable is False, (
+            f"Tool '{tool.description}' should have checkpointable=False"
+        )
+
+    # Verify skill composition (includes)
+    assert len(skill.includes) == 1, f"Expected 1 include, got {len(skill.includes)}"
+    assert skill.includes[0].name == "local/file-ops"
+
+    # Verify skill_count tracks
+    assert skill_svc.skill_count() >= 0
+
+    print("register-terraform-skill-ok")
+
+
+# ---------------------------------------------------------------------------
+# 3. Supervised profile behavior (Spec Step 3 automation)
+# ---------------------------------------------------------------------------
+
+
+def _supervised_profile_behavior() -> None:
+    """Verify supervised profile: gates at strategize->execute and execute->apply."""
+    from cleveragents.application.services.automation_profile_service import (
+        AutomationProfileService,
+    )
+    from cleveragents.application.services.plan_lifecycle_service import (
+        PlanLifecycleService,
+    )
+    from cleveragents.config.settings import Settings
+    from cleveragents.domain.models.core.plan import (
+        PlanPhase,
+        ProcessingState,
+        ProjectLink,
+    )
+
+    # Verify supervised profile properties
+    profile_svc = AutomationProfileService(repo=None)
+    supervised = profile_svc.resolve_profile(plan_profile="supervised")
+    assert supervised.name == "supervised"
+    assert supervised.create_tool == 1.0, (
+        f"Expected create_tool=1.0, got {supervised.create_tool}"
+    )
+    assert supervised.select_tool == 1.0, (
+        f"Expected select_tool=1.0, got {supervised.select_tool}"
+    )
+    assert supervised.safety.require_sandbox is True
+    assert supervised.safety.require_checkpoints is True
+
+    # Verify supervised profile gates plan transitions
+    service = PlanLifecycleService(settings=Settings())
+    service.create_action(
+        name="local/supervised-gate-test",
+        description="Test supervised profile gates",
+        definition_of_done="Gates verified",
+        strategy_actor="local/stub",
+        execution_actor="local/stub",
+        automation_profile="supervised",
+    )
+    plan = service.use_action(
+        action_name="local/supervised-gate-test",
+        project_links=[ProjectLink(project_name="test")],
+    )
+    pid = plan.identity.plan_id
+
+    # Strategize phase
+    service.start_strategize(pid)
+    service.complete_strategize(pid)
+    plan = service.get_plan(pid)
+
+    # Supervised: create_tool=1.0 means should_auto_progress is False
+    assert plan.phase == PlanPhase.STRATEGIZE
+    assert plan.processing_state == ProcessingState.COMPLETE
+    assert service.should_auto_progress(plan) is False, (
+        "Supervised profile should NOT auto-progress from strategize"
+    )
+
+    # Explicit execute
+    service.execute_plan(pid)
+    service.start_execute(pid)
+    plan = service.complete_execute(pid)
+
+    # Supervised: select_tool=1.0 means should_auto_progress is False
+    assert plan.phase == PlanPhase.EXECUTE
+    assert plan.processing_state == ProcessingState.COMPLETE
+    assert service.should_auto_progress(plan) is False, (
+        "Supervised profile should NOT auto-progress from execute"
+    )
+
+    # Explicit apply works
+    service.apply_plan(pid)
+    service.start_apply(pid)
+    plan = service.complete_apply(pid)
+    assert plan.processing_state == ProcessingState.APPLIED
+    assert plan.is_terminal
+
+    print("supervised-profile-behavior-ok")
+
+
+# ---------------------------------------------------------------------------
+# 4. Action with supervised profile, args, invariants + plan use (Spec Step 3)
+# ---------------------------------------------------------------------------
+
+
+def _create_infra_optimize_action() -> None:
+    """Create infra-optimize action and verify plan creation with invariants."""
+    from cleveragents.application.services.plan_lifecycle_service import (
+        PlanLifecycleService,
+    )
+    from cleveragents.config.settings import Settings
+    from cleveragents.domain.models.core.action import (
+        ActionArgument,
+        ArgumentRequirement,
+        ArgumentType,
+    )
+    from cleveragents.domain.models.core.plan import (
+        InvariantSource,
+        PlanPhase,
+        ProcessingState,
+        ProjectLink,
+    )
+
+    service = PlanLifecycleService(settings=Settings())
+    action = service.create_action(
+        name="local/infra-optimize",
+        description="Analyze and optimize cloud infrastructure costs",
+        definition_of_done=(
+            "- Unused resources identified and removal plans generated\n"
+            "- Over-provisioned resources identified with right-sizing recs\n"
+            "- terraform validate passes on all changes\n"
+            "- terraform plan shows no unexpected resource deletions\n"
+            "- Estimated monthly cost savings documented"
+        ),
+        strategy_actor="local/refactoring-strategist",
+        execution_actor="local/infra-executor",
+        automation_profile="supervised",
+        reusable=True,
+        arguments=[
+            ActionArgument(
+                name="optimization_targets",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="Comma-separated: compute, storage, all",
+                default_value="all",
+            ),
+            ActionArgument(
+                name="min_savings_threshold",
+                arg_type=ArgumentType.FLOAT,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="Min monthly savings (USD) to include",
+                default_value="10.0",
+            ),
+        ],
+        invariants=[
+            "Changes to production databases require manual approval",
+            (
+                "Instance type downgrades must verify"
+                " current peak CPU < 70% of new capacity"
+            ),
+        ],
+    )
+
+    assert str(action.namespaced_name) == "local/infra-optimize"
+    assert action.automation_profile == "supervised"
+    assert action.reusable is True
+    assert len(action.arguments) == 2
+    assert len(action.invariants) == 2
+
+    # Use the action to create a plan
+    plan = service.use_action(
+        action_name="local/infra-optimize",
+        project_links=[ProjectLink(project_name="local/infra-prod")],
+        arguments={
+            "optimization_targets": "compute,storage",
+            "min_savings_threshold": 25.0,
+        },
+    )
+    assert plan.phase == PlanPhase.STRATEGIZE
+    assert plan.state == ProcessingState.QUEUED
+
+    # Verify arguments flowed to plan
+    assert plan.arguments["optimization_targets"] == "compute,storage"
+    assert plan.arguments["min_savings_threshold"] == 25.0
+
+    # Verify invariants propagated from action
+    action_invariants = [
+        inv for inv in plan.invariants if inv.source == InvariantSource.ACTION
+    ]
+    assert len(action_invariants) == 2, (
+        f"Expected 2 action invariants on plan, got {len(action_invariants)}"
+    )
+    inv_texts = sorted(inv.text for inv in action_invariants)
+    expected = sorted(
+        [
+            "Changes to production databases require manual approval",
+            (
+                "Instance type downgrades must verify"
+                " current peak CPU < 70% of new capacity"
+            ),
+        ]
+    )
+    assert inv_texts == expected, f"Invariant texts mismatch: {inv_texts}"
+
+    print("create-infra-optimize-action-ok")
+
+
+# ---------------------------------------------------------------------------
+# 5. Infrastructure analysis produces optimization recommendations
+# ---------------------------------------------------------------------------
+
+
+def _infra_analysis_recommendations() -> None:
+    """Verify mocked infrastructure analysis produces valid recommendations."""
+    result = _analyze_infrastructure()
+
+    # Must have recommendations
+    recs = result["recommendations"]
+    assert len(recs) >= 1, f"Expected at least 1 recommendation, got {len(recs)}"
+
+    # First recommendation should be right-sizing the web instance
+    web_rec = recs[0]
+    assert web_rec["resource"] == "aws_instance.web"
+    assert web_rec["type"] == "right-sizing"
+    assert web_rec["current"] == "m5.2xlarge"
+    assert web_rec["recommended"] == "m5.large"
+    assert web_rec["estimated_monthly_savings_usd"] > 0
+
+    # Plan should show no destroys (safe optimization)
+    plan = result["plan"]
+    assert plan["plan_summary"]["destroy"] == 0, (
+        "Optimization plan should have zero destroys"
+    )
+
+    # Critical resources should be skipped
+    assert "aws_instance.api" in result["skipped_critical"]
+
+    # Total analyzed resources
+    assert result["analyzed_resources"] == 3
+
+    print("infra-analysis-recommendations-ok")
+
+
+# ---------------------------------------------------------------------------
+# 6. Invariant enforcement
+# ---------------------------------------------------------------------------
+
+
+def _invariant_enforcement() -> None:
+    """Verify invariants block critical resource modifications.
+
+    Simulates the invariant check: "Never delete or modify resources tagged
+    with 'critical: true'" — a project-level invariant from the spec.
+    """
+    from cleveragents.application.services.plan_lifecycle_service import (
+        PlanLifecycleService,
+    )
+    from cleveragents.config.settings import Settings
+    from cleveragents.domain.models.core.plan import (
+        InvariantSource,
+        ProjectLink,
+    )
+
+    service = PlanLifecycleService(settings=Settings())
+    action = service.create_action(
+        name="local/invariant-test-infra",
+        description="Invariant enforcement test",
+        definition_of_done="Invariants block critical changes",
+        strategy_actor="local/stub",
+        execution_actor="local/stub",
+        automation_profile="supervised",
+        invariants=[
+            "Never delete or modify resources tagged with 'critical: true'",
+            "All changes must be backward-compatible with running services",
+            "Cost changes must be documented in the PR description",
+        ],
+    )
+    assert len(action.invariants) == 3
+
+    plan = service.use_action(
+        action_name="local/invariant-test-infra",
+        project_links=[ProjectLink(project_name="local/infra-prod")],
+    )
+
+    # All invariants should flow to the plan
+    action_invariants = [
+        inv for inv in plan.invariants if inv.source == InvariantSource.ACTION
+    ]
+    assert len(action_invariants) == 3, (
+        f"Expected 3 action invariants, got {len(action_invariants)}"
+    )
+
+    # Simulate invariant check: critical resource modification blocked
+    api_state = _mock_terraform_show("aws_instance.api")
+    is_critical = api_state["values"]["tags"].get("critical") == "true"
+    assert is_critical, "API instance must be tagged as critical"
+
+    # Verify that the invariant text matches what we'd check against
+    inv_texts = [inv.text for inv in action_invariants]
+    critical_invariant = next((t for t in inv_texts if "critical" in t.lower()), None)
+    assert critical_invariant is not None, (
+        "Must have invariant referencing critical resources"
+    )
+
+    # Mock enforcement: if resource is critical, block the change
+    change_blocked = is_critical and critical_invariant is not None
+    assert change_blocked, (
+        "Critical resource modification should be blocked by invariant"
+    )
+
+    # Non-critical resource should NOT be blocked
+    web_state = _mock_terraform_show("aws_instance.web")
+    web_is_critical = web_state["values"]["tags"].get("critical") == "true"
+    assert not web_is_critical, "Web instance should NOT be critical"
+    web_blocked = web_is_critical and critical_invariant is not None
+    assert not web_blocked, "Non-critical resource modification should NOT be blocked"
+
+    print("invariant-enforcement-ok")
+
+
+# ---------------------------------------------------------------------------
+# Command dispatch
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, Any] = {
+    "register-terraform-resource-type": _register_terraform_resource_type,
+    "register-terraform-skill": _register_terraform_skill,
+    "supervised-profile-behavior": _supervised_profile_behavior,
+    "create-infra-optimize-action": _create_infra_optimize_action,
+    "infra-analysis-recommendations": _infra_analysis_recommendations,
+    "invariant-enforcement": _invariant_enforcement,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <command>")
+        print(f"Commands: {', '.join(sorted(_COMMANDS))}")
+        sys.exit(1)
+
+    cmd = sys.argv[1]
+    fn = _COMMANDS.get(cmd)
+    if fn is None:
+        print(f"Unknown command: {cmd}")
+        sys.exit(1)
+
+    try:
+        fn()
+    except Exception as exc:
+        print(f"FAIL [{cmd}]: {type(exc).__name__}: {exc}")
+        import traceback
+
+        traceback.print_exc()
+        sys.exit(1)

robot/wf08_cloud_infra_supervised.robot

@@ -0,0 +1,90 @@
+*** Settings ***
+Documentation    Integration test for Workflow Example 8: Cloud Infrastructure
+...              Management with the supervised automation profile.
+...
+...              Validates custom resource types (local/terraform-state), custom
+...              skills with tool composition (local/terraform-ops), supervised
+...              profile gating, infrastructure analysis with mocked Terraform
+...              operations, optimization recommendations, and invariant
+...              enforcement.
+...
+...              Spec reference: docs/specification.md, Example 8
+...              Issue: #772
+Library          Process
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_wf08_cloud_infra_supervised.py
+
+*** Test Cases ***
+WF08 Register Terraform Resource Type
+    [Documentation]    Register local/terraform-state custom resource type
+    ...                from YAML fixture with filesystem_copy sandbox strategy,
+    ...                cli_args, and resource instance creation.
+    [Tags]    wf08    resource    registration    integration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    register-terraform-resource-type    cwd=${WORKSPACE}    timeout=60s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0    Registration failed: ${result.stderr}
+    Should Contain    ${result.stdout}    register-terraform-resource-type-ok
+
+WF08 Register Terraform Skill With Tool Composition
+    [Documentation]    Register local/terraform-ops skill with 3 anonymous tools
+    ...                (terraform_plan, terraform_show, cloud_metrics) and skill
+    ...                composition via includes (local/file-ops).
+    [Tags]    wf08    skill    composition    integration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    register-terraform-skill    cwd=${WORKSPACE}    timeout=60s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0    Skill registration failed: ${result.stderr}
+    Should Contain    ${result.stdout}    register-terraform-skill-ok
+
+WF08 Supervised Profile Gating Behavior
+    [Documentation]    Verify the supervised automation profile gates both
+    ...                strategize-to-execute and execute-to-apply transitions,
+    ...                requiring explicit human approval at each step.
+    [Tags]    wf08    profile    supervised    integration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    supervised-profile-behavior    cwd=${WORKSPACE}    timeout=60s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0    Profile check failed: ${result.stderr}
+    Should Contain    ${result.stdout}    supervised-profile-behavior-ok
+
+WF08 Create Infra Optimize Action With Invariants
+    [Documentation]    Create local/infra-optimize action with supervised profile,
+    ...                two arguments (optimization_targets, min_savings_threshold),
+    ...                two invariants, and verify plan creation with correct state
+    ...                and invariant propagation.
+    [Tags]    wf08    action    plan    invariants    integration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    create-infra-optimize-action    cwd=${WORKSPACE}    timeout=60s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0    Action/plan failed: ${result.stderr}
+    Should Contain    ${result.stdout}    create-infra-optimize-action-ok
+
+WF08 Infrastructure Analysis Produces Optimization Recommendations
+    [Documentation]    Run mocked infrastructure analysis using Terraform state
+    ...                and cloud metrics fixtures. Verify that over-provisioned
+    ...                resources receive right-sizing recommendations and that
+    ...                critical resources are skipped.
+    [Tags]    wf08    analysis    recommendations    integration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    infra-analysis-recommendations    cwd=${WORKSPACE}    timeout=60s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0    Analysis failed: ${result.stderr}
+    Should Contain    ${result.stdout}    infra-analysis-recommendations-ok
+
+WF08 Invariant Enforcement Blocks Critical Resource Deletion
+    [Documentation]    Verify that action invariants propagate to plans and
+    ...                that the invariant "Never delete or modify resources
+    ...                tagged with 'critical: true'" blocks modifications to
+    ...                critical infrastructure while allowing changes to
+    ...                non-critical resources.
+    [Tags]    wf08    invariants    enforcement    integration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    invariant-enforcement    cwd=${WORKSPACE}    timeout=60s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0    Invariant enforcement failed: ${result.stderr}
+    Should Contain    ${result.stdout}    invariant-enforcement-ok