CHANGELOG.md

@@ -114,6 +114,20 @@
   `database_url` resolves inside `CLEVERAGENTS_HOME`, not the current
   working directory.  Includes Robot Framework integration tests with a
   helper script exercising the same resolution path via subprocess.  (#1034)
+- Added integration Robot Framework test for Specification Workflow Example 7:
+  CI/CD Integration — Automated PR Review and Fix.  Exercises the `ci`
+  automation profile (headless, non-interactive) covering: ci-profile
+  configuration (automation-profile, format, log level), idempotent
+  resource and project registration with duplicate-detection assertions,
+  three validation tools (`ci-lint`, `ci-typecheck`, `ci-tests`) registration
+  and resource attachment via `ToolRegistryService`, action creation with
+  typed arguments and invariants per spec Step 2, plan lifecycle with
+  explicit phase/state transition assertions across all phases
+  (strategize, execute, apply) plus terminal `applied` and `cancelled`
+  path checks, and JSON output structure verification
+  including `plan_id`, `phase`, `state`, `action`, `projects`, and
+  `arguments` fields.
+  (`robot/wf07_cicd_integration.robot`, `robot/helper_wf07_cicd.py`)  (#771)
 - Added volatile in-memory `audit_log` to `ReactiveEventBus` — every emitted
   `DomainEvent` is appended to a volatile in-memory log accessible via the
   `audit_log` property (defensive copy).  Emit ordering now follows the
@@ -455,7 +469,7 @@
   under the 500-line limit.  `Verify Plan In List` and `Full Flow Apply Step`
   keywords use hard assertions instead of WARN fallbacks.  Profile Precedence
   test documents that action > global precedence requires production wiring
-   not yet present in `PlanLifecycleService.use_action`.
+  not yet present in `PlanLifecycleService.use_action`.
   (`robot/e2e/m6_acceptance.robot`, `robot/e2e/common_e2e.resource`)  (#746)
 - Added E2E Robot Framework test for Specification Workflow Example 7: CI/CD
   Integration — Automated PR Review and Fix.  Exercises the `ci` automation

benchmarks/decision_correction_model_bench.py

@@ -106,7 +106,7 @@ class CorrectionServiceSuite:
         """Benchmark creating a correction request."""
         self.service.request_correction(
             plan_id="plan-bench",
-            decision_id="DEC-001",
+            target_decision_id="DEC-001",
             mode=CorrectionMode.REVERT,
             guidance="Benchmark correction",
         )
@@ -115,7 +115,7 @@ class CorrectionServiceSuite:
         """Benchmark impact analysis (stub)."""
         req = self.service.request_correction(
             plan_id="plan-bench",
-            decision_id="DEC-001",
+            target_decision_id="DEC-001",
             mode=CorrectionMode.REVERT,
             guidance="Benchmark analysis",
         )
@@ -125,7 +125,7 @@ class CorrectionServiceSuite:
         """Benchmark correction execution (stub)."""
         req = self.service.request_correction(
             plan_id="plan-bench",
-            decision_id="DEC-001",
+            target_decision_id="DEC-001",
             mode=CorrectionMode.REVERT,
             guidance="Benchmark execution",
         )

benchmarks/decision_di_bench.py

@@ -44,7 +44,7 @@ except ModuleNotFoundError:
     from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 
 
-_PLAN_ID = "01HV00000000000000DIBENCH1"
+_PLAN_ID = "01HV00000000000000DJBENCH1"
 
 
 def _make_uow() -> UnitOfWork:

benchmarks/phase_reversion_bench.py

@@ -119,16 +119,16 @@ class AutoRevertSuite:
         )
         plan_id = plan.identity.plan_id
         self.service.start_strategize(plan_id)
+        # complete_strategize auto-progresses to EXECUTE with "ci" profile
         self.service.complete_strategize(plan_id)
         plan = self.service.get_plan(plan_id)
         if plan.phase != PlanPhase.EXECUTE:
             self.service.execute_plan(plan_id)
         self.service.start_execute(plan_id)
+        # complete_execute auto-progresses to APPLY with "ci" profile
         self.service.complete_execute(plan_id)
-        self.service.apply_plan(plan_id)
         self.service.start_apply(plan_id)
-        plan = self.service.get_plan(plan_id)
-        plan.processing_state = ProcessingState.CONSTRAINED
+        self.service.constrain_apply(plan_id, "benchmark constraint")
         self.service.try_auto_revert_from_apply(plan_id, "benchmark")
 
 

benchmarks/resource_cli_tree_bench.py

@@ -75,10 +75,10 @@ class ResourceTreeSuite:
         """Set up mock service."""
         root = _mock_resource()
         child1 = _mock_resource(
-            "01HBENCH0000000000CHILD001", "local/child-1", "fs-directory"
+            "01HBENCH0000000000CHJKD001", "local/child-1", "fs-directory"
         )
         child2 = _mock_resource(
-            "01HBENCH0000000000CHILD002", "local/child-2", "fs-directory"
+            "01HBENCH0000000000CHJKD002", "local/child-2", "fs-directory"
         )
         tree = [
             _mock_tree_node(

benchmarks/session_list_bench.py

@@ -61,6 +61,11 @@ class SessionListDISuite:
             bind=self._engine,
             expire_on_commit=False,
         )
+        # Pre-initialise attributes used by per-method setup hooks so that
+        # benchmarks still work even when the ASV runner skips per-method
+        # setup_<name>() callbacks (e.g. in fork-server mode).
+        self._empty_svc = self._make_service()
+        self._fresh_engine()
 
     def teardown(self) -> None:
         self._engine.dispose()

benchmarks/unified_context_models_bench.py

@@ -84,7 +84,7 @@ class TimeUnifiedModelCreation:
     def time_context_payload(self) -> None:
         """Create core ContextPayload (inherits CRP AssembledContext)."""
         for _ in range(1000):
-            ContextPayload(plan_id="01JQBENCHPN00000000000000AA")
+            ContextPayload(plan_id="01JQBENCHPN0000000000000AA")
 
 
 class TimeIsinstanceChecks:
@@ -104,7 +104,7 @@ class TimeIsinstanceChecks:
         )
         self.budget = ContextBudget(max_tokens=4096, reserved_tokens=512)
         self.payload = ContextPayload(
-            plan_id="01JQBENCHPN00000000000000AA",
+            plan_id="01JQBENCHPN0000000000000AA",
         )
 
     def time_isinstance_provenance(self) -> None:

noxfile.py

@@ -577,6 +577,12 @@ def integration_tests(session: nox.Session):
     # race on CI runners with high core counts.
     session.run("python", "-m", "compileall", "-q", "src/")
 
+    # Build a pre-migrated template DB so helper scripts that call
+    # setup_workspace() can copy it instead of running 25+ Alembic
+    # migrations per test — critical for parallel pabot execution.
+    template_path = _create_template_db(session)
+    session.env["CLEVERAGENTS_TEMPLATE_DB"] = template_path
+
     pabot_args, robot_args = _split_pabot_args(session.posargs)
     parallel_args = _pabot_parallel_args(pabot_args)
 

robot/helper_e2e_common.py

@@ -79,8 +79,15 @@ def setup_workspace(prefix: str = "e2e_") -> str:
     """Create an isolated workspace directory with a ready database.
 
     Sets ``CLEVERAGENTS_HOME`` and ``CLEVERAGENTS_DATABASE_URL`` to the
-    workspace and runs Alembic migrations so all tables are available
-    for every CLI command.
+    workspace so all tables are available for every CLI command.
+
+    When a pre-migrated template database is available (either via the
+    ``CLEVERAGENTS_TEMPLATE_DB`` environment variable or at the default
+    ``build/.template-migrated.db`` path), the template is copied
+    instead of running full Alembic migrations.  This reduces per-test
+    setup from ~1-3 s to ~1 ms, which is critical for parallel
+    execution under pabot where many workers set up workspaces
+    concurrently.
 
     Returns the absolute path to the workspace.
     """
@@ -90,14 +97,28 @@ def setup_workspace(prefix: str = "e2e_") -> str:
     db_url = f"sqlite:///{db_path}"
     os.environ["CLEVERAGENTS_DATABASE_URL"] = db_url
 
-    # Run Alembic migrations so resource_types / projects / etc. tables
-    # exist before any CLI subprocess touches the database.
-    from cleveragents.infrastructure.database.migration_runner import (
-        MigrationRunner,
-    )
+    # Fast path: copy pre-migrated template DB instead of running
+    # 25+ Alembic migrations (avoids I/O contention under pabot).
+    template = os.environ.get("CLEVERAGENTS_TEMPLATE_DB")
+    if not template:
+        default_template = os.path.join(
+            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            "build",
+            ".template-migrated.db",
+        )
+        if os.path.isfile(default_template):
+            template = default_template
 
-    runner = MigrationRunner(db_url)
-    runner.init_or_upgrade(require_confirmation=False)
+    if template and os.path.isfile(template):
+        shutil.copy2(template, db_path)
+    else:
+        # Fallback: run Alembic migrations (slow path).
+        from cleveragents.infrastructure.database.migration_runner import (
+            MigrationRunner,
+        )
+
+        runner = MigrationRunner(db_url)
+        runner.init_or_upgrade(require_confirmation=False)
 
     return workspace
 

robot/helper_wf07_cicd.py

@@ -0,0 +1,500 @@
+"""Helper script for wf07_cicd_integration.robot tests.
+
+Each subcommand is a self-contained integration check that prints a sentinel
+on success and exercises specification workflow example 7.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import tempfile
+from collections.abc import Callable
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+# Ensure local source tree is importable
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from sqlalchemy import create_engine  # noqa: E402
+from sqlalchemy.exc import IntegrityError  # noqa: E402
+from sqlalchemy.orm import sessionmaker  # noqa: E402
+
+from cleveragents.application.services.automation_profile_service import (  # noqa: E402
+    AutomationProfileService,
+)
+from cleveragents.application.services.config_service import (  # noqa: E402
+    ConfigService,
+)
+from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
+    PlanLifecycleService,
+)
+from cleveragents.application.services.resource_registry_service import (  # noqa: E402
+    ResourceRegistryService,
+)
+from cleveragents.application.services.tool_registry_service import (  # noqa: E402
+    ToolRegistryService,
+)
+from cleveragents.application.services.validation_pipeline import (  # noqa: E402
+    ValidationCommand,
+    ValidationPipeline,
+)
+from cleveragents.config.settings import Settings  # noqa: E402
+from cleveragents.core.exceptions import DatabaseError  # noqa: E402
+from cleveragents.domain.models.core.action import (  # noqa: E402
+    ActionArgument,
+    ArgumentRequirement,
+    ArgumentType,
+)
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    InvariantSource,
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.domain.models.core.project import (  # noqa: E402
+    NamespacedProject,
+    parse_namespaced_name,
+)
+from cleveragents.domain.models.core.tool import ValidationMode  # noqa: E402
+from cleveragents.infrastructure.database.models import Base  # noqa: E402
+from cleveragents.infrastructure.database.repositories import (  # noqa: E402
+    NamespacedProjectRepository,
+    ToolRegistryRepository,
+    ValidationAttachmentRepository,
+)
+
+# ---------------------------------------------------------------------------
+# Database helpers (in-memory SQLite)
+# ---------------------------------------------------------------------------
+
+
+class _NoCloseSession:
+    """Wrapper that prevents the session from being closed by service code."""
+
+    def __init__(self, session: object) -> None:
+        object.__setattr__(self, "_s", session)
+
+    def close(self) -> None:
+        pass
+
+    def __setattr__(self, name: str, value: object) -> None:
+        setattr(object.__getattribute__(self, "_s"), name, value)
+
+    def __getattr__(self, name: str) -> object:
+        return getattr(object.__getattribute__(self, "_s"), name)
+
+
+def _setup_db() -> tuple[Any, Any]:
+    """Create an in-memory SQLite DB and return (session_factory, session)."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine, expire_on_commit=False)()
+    wrapper = _NoCloseSession(session)
+
+    def factory() -> object:
+        return wrapper
+
+    return factory, session
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def config_ci_profile() -> None:
+    """Set and get ci profile, json format, and log level via ConfigService."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        svc = ConfigService(config_dir=Path(tmpdir))
+        svc.set_value("core.automation-profile", "ci")
+        resolved = svc.resolve("core.automation-profile")
+        assert resolved.value == "ci", f"Expected 'ci', got '{resolved.value}'"
+        svc.set_value("core.format", "json")
+        resolved = svc.resolve("core.format")
+        assert resolved.value == "json", f"Expected 'json', got '{resolved.value}'"
+        svc.set_value("core.log.level", "WARN")
+        resolved = svc.resolve("core.log.level")
+        assert resolved.value == "WARN", f"Expected 'WARN', got '{resolved.value}'"
+    print("wf07-config-ci-profile-ok")
+
+
+def resource_idempotent() -> None:
+    """Register a git-checkout resource twice; verify only one exists."""
+    factory, _session = _setup_db()
+    svc = ResourceRegistryService(session_factory=factory)
+    svc.bootstrap_builtin_types()
+    # First registration — includes branch per spec Step 3
+    with tempfile.TemporaryDirectory() as repo_dir:
+        svc.register_resource(
+            type_name="git-checkout",
+            name="local/cicd-repo",
+            location=repo_dir,
+            description="CI/CD repository",
+            properties={"branch": "fix/handle-null-users"},
+        )
+    # Second registration mirrors spec idempotency; CI handles duplicates with || true.
+    duplicate_raised = False
+    with tempfile.TemporaryDirectory() as dup_dir:
+        try:
+            svc.register_resource(
+                type_name="git-checkout",
+                name="local/cicd-repo",
+                location=dup_dir,
+                description="CI/CD repository (updated)",
+            )
+        except (IntegrityError, DatabaseError):
+            duplicate_raised = True
+    # The DB unique constraint should prevent a second resource
+    assert duplicate_raised, (
+        "Expected IntegrityError or DatabaseError on duplicate resource registration"
+    )
+    resources = svc.list_resources(type_name="git-checkout")
+    assert len(resources) == 1, f"Expected 1 resource, got {len(resources)}"
+    assert resources[0].description == "CI/CD repository"
+    assert resources[0].properties.get("branch") == "fix/handle-null-users", (
+        f"Expected branch 'fix/handle-null-users', got {resources[0].properties}"
+    )
+    print("wf07-resource-idempotent-ok")
+
+
+def project_idempotent() -> None:
+    """Create a project twice; verify it exists exactly once."""
+    factory, _session = _setup_db()
+    proj_repo = NamespacedProjectRepository(session_factory=factory)
+    parsed = parse_namespaced_name("cicd-project")
+    proj = NamespacedProject(
+        name=parsed.name,
+        namespace=parsed.namespace,
+        description="CI/CD integration project",
+    )
+    proj_repo.create(proj)
+    # Second creation with same name: expect DatabaseError (integrity violation).
+    # NOTE: NamespacedProjectRepository.create has @database_retry (3 attempts,
+    # 0.5s wait) so the expected IntegrityError→DatabaseError adds ~1s overhead.
+    duplicate_raised = False
+    try:
+        proj_dup = NamespacedProject(
+            name=parsed.name,
+            namespace=parsed.namespace,
+            description="CI/CD integration project (dup)",
+        )
+        proj_repo.create(proj_dup)
+    except (DatabaseError, IntegrityError):
+        duplicate_raised = True
+    assert duplicate_raised, (
+        "Expected DatabaseError or IntegrityError on duplicate project creation"
+    )
+    projects = proj_repo.list_projects()
+    assert len(projects) == 1, f"Expected exactly 1 project, got {len(projects)}"
+    fetched = proj_repo.get("local/cicd-project")
+    assert fetched.name == "cicd-project", (
+        f"Expected name 'cicd-project', got '{fetched.name}'"
+    )
+    print("wf07-project-idempotent-ok")
+
+
+def validation_attach() -> None:
+    """Register validation tools, attach to a resource, and run pipeline."""
+    factory, _session = _setup_db()
+    tool_repo = ToolRegistryRepository(session_factory=factory)
+    attachment_repo = ValidationAttachmentRepository(session_factory=factory)
+    svc = ToolRegistryService(
+        tool_repo=tool_repo,
+        attachment_repo=attachment_repo,
+    )
+    # Spec Step 3 registers 3 validations: ci-lint, ci-typecheck, ci-tests
+    now = datetime.now(tz=timezone.utc).isoformat()  # noqa: UP017
+    validation_names = ["local/ci-lint", "local/ci-typecheck", "local/ci-tests"]
+    for vname in validation_names:
+        svc.register_tool(
+            {
+                "name": vname,
+                "description": f"CI validation: {vname.rsplit('/', 1)[-1]}",
+                "tool_type": "validation",
+                "source": "builtin",
+                "timeout": 300,
+                "created_at": now,
+                "updated_at": now,
+                "resource_bindings": [],
+                "mode": "required",
+            }
+        )
+    # Verify all 3 tools are registered
+    for vname in validation_names:
+        registered = svc.get_tool(vname)
+        assert registered is not None, f"Validation tool {vname} must be registered"
+    # --- Step 2: Attach to resource (spec: ``validation attach``) ---
+    # Register a resource so the attachment references a real entity.
+    res_svc = ResourceRegistryService(session_factory=factory)
+    res_svc.bootstrap_builtin_types()
+    with tempfile.TemporaryDirectory(prefix="val-test-repo-") as val_tmpdir:
+        registered_resource = res_svc.register_resource(
+            type_name="git-checkout",
+            name="local/cicd-repo",
+            location=val_tmpdir,
+            description="Validation test repo",
+        )
+        resource_id = registered_resource.resource_id
+        # Attach all 3 validations per spec Step 3
+        for vname in validation_names:
+            attachment = svc.attach_validation(
+                validation_name=vname,
+                resource_id=resource_id,
+                mode="required",
+                project_name="local/ci-workspace",
+            )
+            assert attachment is not None, f"Attachment for {vname} must be created"
+        # Verify all 3 attachments exist
+        attachments = svc.list_validations_for_resource(resource_id)
+        assert len(attachments) == 3, f"Expected 3 attachments, got {len(attachments)}"
+        # --- Step 3: Verify pipeline execution with the validations ---
+        commands = [
+            ValidationCommand(
+                validation_name=vn,
+                resource_id=resource_id,
+                resource_name="local/cicd-repo",
+                mode=ValidationMode.REQUIRED,
+                arguments={"language": "python"},
+                timeout_seconds=30.0,
+            )
+            for vn in validation_names
+        ]
+
+        def mock_executor(
+            validation_name: str,
+            arguments: dict[str, Any],
+        ) -> dict[str, Any]:
+            return {"passed": True, "message": f"{validation_name} passed"}
+
+        pipeline = ValidationPipeline(
+            commands=commands,
+            executor=mock_executor,
+            max_workers=1,
+        )
+        summary = pipeline.run()
+        assert summary.total == 3, f"Expected 3 validations, got {summary.total}"
+        assert summary.all_required_passed, "Expected all required validations to pass"
+    print("wf07-validation-attach-ok")
+
+
+def ci_plan_lifecycle() -> None:
+    """Create action, verify phase transitions, and assert terminal states."""
+    settings = Settings()
+    service = PlanLifecycleService(settings=settings)
+    # Create action for PR review (spec Step 2).  Actor names use local/
+    # prefixes (no real API connectivity in integration tests).
+    action = service.create_action(
+        name="local/review-pr",
+        description="Automatically review a PR and fix issues",
+        definition_of_done=(
+            "- All lint issues are resolved\n"
+            "- Type checking passes\n"
+            "- Test coverage does not decrease\n"
+            "- Security scan passes\n"
+            "- All fixes are committed to the PR branch"
+        ),
+        strategy_actor="local/ci-planner",
+        execution_actor="local/ci-executor",
+        automation_profile="ci",
+        reusable=True,
+        arguments=[
+            ActionArgument(
+                name="pr_branch",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.REQUIRED,
+                description="Branch name of the PR",
+            ),
+            ActionArgument(
+                name="base_branch",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="Base branch to compare against",
+                default_value="main",
+            ),
+        ],
+        invariants=[
+            "Only modify files that are already changed in the PR",
+            "Do not change the intent of any code"
+            " — only fix style, types, and test issues",
+            "All fixes must include a comment explaining what was changed and why",
+        ],
+    )
+    assert str(action.namespaced_name) == "local/review-pr"  # L6: correctness
+    assert action.automation_profile == "ci", (
+        f"Expected profile 'ci', got '{action.automation_profile}'"
+    )
+    assert len(action.arguments) == 2
+    assert len(action.invariants) == 3
+    assert action.reusable is True, "Expected reusable=True"
+    # Use the action to create a plan with spec-required arguments
+    plan = service.use_action(
+        action_name=str(action.namespaced_name),
+        project_links=[ProjectLink(project_name="cicd-project")],
+        created_by="ci-pipeline",
+        arguments={"pr_branch": "fix/handle-null-users", "base_branch": "main"},
+    )
+    assert plan.phase == PlanPhase.STRATEGIZE
+    assert plan.state == ProcessingState.QUEUED
+    plan_id = plan.identity.plan_id
+    assert plan_id, "Plan must have a plan_id"
+    # Verify arguments flowed to the plan
+    assert plan.arguments == {
+        "pr_branch": "fix/handle-null-users",
+        "base_branch": "main",
+    }, f"Plan arguments mismatch: {plan.arguments}"
+    # Verify invariants flowed from action to plan — count and content (M5)
+    action_invariants = [
+        inv for inv in plan.invariants if inv.source == InvariantSource.ACTION
+    ]
+    assert len(action_invariants) == 3, (
+        f"Expected 3 action invariants on plan, got {len(action_invariants)}"
+    )
+    expected_inv_texts = sorted(
+        [
+            "Only modify files that are already changed in the PR",
+            "Do not change the intent of any code"
+            " — only fix style, types, and test issues",
+            "All fixes must include a comment explaining what was changed and why",
+        ]
+    )
+    actual_inv_texts = sorted(inv.text for inv in action_invariants)
+    assert actual_inv_texts == expected_inv_texts, (
+        f"Invariant text mismatch: {actual_inv_texts}"
+    )
+    fetched = service.get_plan(plan_id)
+    assert fetched.identity.plan_id == plan_id
+    assert fetched.action_name == str(action.namespaced_name)
+    # Verify the automation profile resolved to ci
+    profile_svc = AutomationProfileService(repo=None)
+    ci_profile = profile_svc.resolve_profile(plan_profile="ci")
+    assert ci_profile.name == "ci"
+    # NOTE: TODO(#1060) - use_action() does not propagate action profile to plan.
+    # auto_progress then resolves to manual; keep phase guards until #1060 is fixed.
+    # --- Phase-by-phase plan completion (spec Step 3) ---
+    service.start_strategize(plan_id)
+    p = service.get_plan(plan_id)
+    assert p.state == ProcessingState.PROCESSING, (
+        f"Expected PROCESSING after start_strategize, got {p.state}"
+    )
+    service.complete_strategize(plan_id)
+    p = service.get_plan(plan_id)
+    assert (p.phase, p.state) in {
+        (PlanPhase.STRATEGIZE, ProcessingState.COMPLETE),
+        (PlanPhase.EXECUTE, ProcessingState.QUEUED),
+    }, (
+        "After complete_strategize expected strategize/complete or execute/queued, "
+        f"got {p.phase.value}/{p.state.value}"
+    )
+    if p.phase == PlanPhase.STRATEGIZE:
+        service.execute_plan(plan_id)
+    # Execute phase
+    service.start_execute(plan_id)
+    service.complete_execute(plan_id)
+    p = service.get_plan(plan_id)
+    assert (p.phase, p.state) in {
+        (PlanPhase.EXECUTE, ProcessingState.COMPLETE),
+        (PlanPhase.APPLY, ProcessingState.QUEUED),
+    }, (
+        "After complete_execute expected execute/complete or apply/queued, "
+        f"got {p.phase.value}/{p.state.value}"
+    )
+    if p.phase == PlanPhase.EXECUTE:
+        service.apply_plan(plan_id)
+    # Apply phase
+    service.start_apply(plan_id)
+    service.complete_apply(plan_id)
+    # Verify terminal state — the polling loop exits on 'applied'
+    final = service.get_plan(plan_id)
+    assert final.state == ProcessingState.APPLIED, (
+        f"Expected APPLIED, got {final.state}"
+    )
+    cancelled_plan = service.use_action(
+        action_name=str(action.namespaced_name),
+        project_links=[ProjectLink(project_name="cicd-project")],
+        created_by="ci-pipeline",
+        arguments={"pr_branch": "fix/handle-null-users", "base_branch": "main"},
+    )
+    cancelled = service.cancel_plan(cancelled_plan.identity.plan_id, reason="ci cancel")
+    assert cancelled.state == ProcessingState.CANCELLED
+    assert cancelled.is_terminal
+    print("wf07-ci-plan-lifecycle-ok")
+
+
+def json_output() -> None:
+    """Verify JSON dict from Action and Plan models via as_cli_dict()."""
+    settings = Settings()
+    service = PlanLifecycleService(settings=settings)
+    # Create an action with arguments to verify full JSON round-trip
+    action = service.create_action(
+        name="local/json-test",
+        description="JSON output test action",
+        definition_of_done="Verify JSON output",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        arguments=[
+            ActionArgument(
+                name="pr_branch",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.REQUIRED,
+                description="Branch name of the PR",
+            ),
+            ActionArgument(
+                name="base_branch",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="Base branch to compare against",
+                default_value="main",
+            ),
+        ],
+    )
+    # Get CLI dict representation (simulates --format json output)
+    parsed = json.loads(json.dumps(action.as_cli_dict()))
+    assert isinstance(parsed, dict), "Parsed JSON must be a dict"
+    assert parsed.get("name") == "local/json-test"
+    assert parsed.get("state") == "available"
+    assert "description" in parsed, "JSON must contain 'description' field"
+    # Create a plan with arguments to verify full JSON output per spec Step 3
+    plan = service.use_action(
+        action_name=str(action.namespaced_name),
+        project_links=[ProjectLink(project_name="json-project")],
+        arguments={"pr_branch": "fix/test", "base_branch": "main"},
+    )
+    plan_parsed = json.loads(json.dumps(plan.as_cli_dict()))
+    # NOTE: as_cli_dict() uses "arguments"/"projects" (plural/dict) while
+    # the spec sample JSON (line 39199) uses "args"/"project" (singular).
+    # as_cli_dict() also omits "attempt" and "resources" (CLI-level fields)
+    # and "automation_profile" (not yet propagated — TODO #1060).
+    assert "plan_id" in plan_parsed
+    assert plan_parsed.get("phase") == "strategize"
+    assert plan_parsed.get("state") == "queued"
+    assert plan_parsed.get("action") == "local/json-test"
+    assert plan_parsed["projects"][0]["name"] == "json-project"
+    assert plan_parsed["arguments"]["pr_branch"] == "fix/test"
+    print("wf07-json-output-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, Callable[[], None]] = {
+    "config-ci-profile": config_ci_profile,
+    "resource-idempotent": resource_idempotent,
+    "project-idempotent": project_idempotent,
+    "validation-attach": validation_attach,
+    "ci-plan-lifecycle": ci_plan_lifecycle,
+    "json-output": json_output,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(
+            f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    _COMMANDS[sys.argv[1]]()

robot/wf07_cicd_integration.robot

@@ -0,0 +1,74 @@
+*** Settings ***
+Documentation    Integration test for Workflow Example 7: CI/CD integration,
+...              automated PR review and fix (ci profile).
+...
+...              Validates the CI/CD automation workflow covering: ci-profile
+...              configuration, idempotent resource and project registration,
+...              validation registration and attachment, plan lifecycle with
+...              ci automation profile and phase-by-phase completion through
+...              all phases until terminal applied state, and JSON output
+...              structure verification.
+Library          Process
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment With Database Isolation
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_wf07_cicd.py
+
+*** Test Cases ***
+WF07 CI Profile Config Set And Get
+    [Documentation]    Set and verify ci automation profile, json format, and log level via ConfigService.
+    [Tags]    cicd    integration    workflow7
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    config-ci-profile    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf07-config-ci-profile-ok
+
+WF07 Idempotent Resource Registration
+    [Documentation]    Register a git-checkout resource twice and verify only one exists.
+    [Tags]    cicd    integration    workflow7
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    resource-idempotent    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf07-resource-idempotent-ok
+
+WF07 Idempotent Project Registration
+    [Documentation]    Create a project twice and verify it exists exactly once.
+    [Tags]    cicd    integration    workflow7
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    project-idempotent    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf07-project-idempotent-ok
+
+WF07 Validation Registration And Attachment
+    [Documentation]    Register three validation tools, attach to a resource, and verify pipeline execution.
+    [Tags]    cicd    integration    workflow7
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    validation-attach    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf07-validation-attach-ok
+
+WF07 CI Plan Lifecycle
+    [Documentation]    Create an action with arguments and invariants, run plan lifecycle with
+    ...                phase-by-phase completion through all phases, then verify applied and cancelled terminal states.
+    [Tags]    cicd    integration    workflow7
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    ci-plan-lifecycle    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf07-ci-plan-lifecycle-ok
+
+WF07 JSON Output Parsing
+    [Documentation]    Run commands with JSON output and verify structure including plan_id,
+    ...                phase, state, and action fields.
+    [Tags]    cicd    integration    workflow7
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    json-output    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf07-json-output-ok