cleveragents-core/robot/helper_cli_lifecycle.py

"""Robot helper for CLI lifecycle E2E with sandbox + ChangeSet verification.

Each subcommand is self-contained and prints a sentinel on success.
Exit code 0 = pass, 1 = failure.
"""

from __future__ import annotations

import os
import shutil
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from typing import NoReturn
from unittest.mock import MagicMock, patch

# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from typer.testing import CliRunner  # noqa: E402

from cleveragents.cli.commands.action import app as action_app  # noqa: E402
from cleveragents.cli.commands.plan import app as plan_app  # noqa: E402
from cleveragents.domain.models.core.action import Action, ActionState  # noqa: E402
from cleveragents.domain.models.core.change import (  # noqa: E402
    ChangeEntry,
    ChangeOperation,
    InMemoryChangeSetStore,
    SpecChangeSet,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    AutomationProfileProvenance,
    AutomationProfileRef,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
)

runner = CliRunner()
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8R"

_VALID_YAML = """\
name: local/lifecycle-action
description: Lifecycle test action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: All lifecycle tests pass
"""


def _mock_action(name: str = "local/lifecycle-action") -> Action:
    """Create a minimal valid Action."""
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="Lifecycle test action",
        long_description=None,
        definition_of_done="All lifecycle tests pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        state=ActionState.AVAILABLE,
        reusable=True,
        read_only=False,
        created_at=datetime.now(),
        updated_at=datetime.now(),
        created_by=None,
    )


def _mock_plan(
    phase: PlanPhase = PlanPhase.STRATEGIZE,
    state: ProcessingState = ProcessingState.QUEUED,
    plan_id: str = _PLAN_ULID,
) -> Plan:
    """Create a minimal valid Plan."""
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=plan_id),
        namespaced_name=NamespacedName.parse("local/lifecycle-plan"),
        description="Lifecycle test plan",
        definition_of_done="Tests pass",
        action_name="local/lifecycle-action",
        phase=phase,
        processing_state=state,
        project_links=[ProjectLink(project_name="proj-a")],
        arguments={"target_coverage": 80},
        arguments_order=["target_coverage"],
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        ),
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        reusable=True,
        read_only=False,
        created_by=None,
        timestamps=PlanTimestamps(created_at=now, updated_at=now),
    )


def _write_yaml(content: str) -> str:
    """Write YAML content to a temp file and return the path."""
    fd, path = tempfile.mkstemp(suffix=".yaml")
    with os.fdopen(fd, "w") as fh:
        fh.write(content)
    return path


def _make_store() -> tuple[InMemoryChangeSetStore, str]:
    """Create an InMemoryChangeSetStore with one changeset started."""
    store = InMemoryChangeSetStore()
    cid = store.start(_PLAN_ULID)
    return store, cid


def _fail(msg: str) -> NoReturn:
    """Print failure message and exit (never returns)."""
    print(f"FAIL: {msg}", file=sys.stderr)
    raise SystemExit(1)


# ---------------------------------------------------------------------------
# Positive subcommands
# ---------------------------------------------------------------------------


def action_create() -> None:
    """Verify action create from config file."""
    svc = MagicMock()
    svc.create_action.return_value = _mock_action()
    yaml_path = _write_yaml(_VALID_YAML)
    try:
        with patch(
            "cleveragents.cli.commands.action._get_lifecycle_service", return_value=svc
        ):
            result = runner.invoke(action_app, ["create", "--config", yaml_path])
            if result.exit_code != 0:
                _fail(f"action create rc={result.exit_code}\n{result.output}")
            print("robot-lifecycle-action-create-ok")
    finally:
        os.unlink(yaml_path)


def plan_use_sandbox() -> None:
    """Verify plan use with simulated git_worktree sandbox workspace."""
    svc = MagicMock()
    svc.get_action_by_name.return_value = _mock_action()
    svc.use_action.return_value = _mock_plan()

    sandbox_dir = tempfile.mkdtemp(prefix="robot_sandbox_")
    try:
        sb = Path(sandbox_dir)
        (sb / ".git").mkdir()
        (sb / "src").mkdir()
        (sb / "src" / "main.py").write_text("# placeholder\n")

        with patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
        ):
            result = runner.invoke(
                plan_app, ["use", "local/lifecycle-action", "proj-a"]
            )
            if result.exit_code != 0:
                _fail(f"plan use rc={result.exit_code}\n{result.output}")
        if not sb.exists():
            _fail("sandbox directory missing")
        print("robot-lifecycle-plan-use-sandbox-ok")
    finally:
        shutil.rmtree(sandbox_dir, ignore_errors=True)


def plan_execute_changeset() -> None:
    """Verify plan execute and ChangeSet capture records entries."""
    svc = MagicMock()
    svc.execute_plan.return_value = _mock_plan(phase=PlanPhase.EXECUTE)
    svc.get_plan.return_value = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE
    )
    store, cid = _make_store()

    store.record(
        cid,
        ChangeEntry(
            plan_id=_PLAN_ULID,
            resource_id="res-a",
            tool_name="builtin/file-write",
            operation=ChangeOperation.CREATE,
            path="src/new_module.py",
        ),
    )

    with (
        patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
        ),
        patch(
            "cleveragents.cli.commands.plan._get_plan_executor",
            return_value=MagicMock(),
        ),
    ):
        result = runner.invoke(plan_app, ["execute", _PLAN_ULID])
        if result.exit_code != 0:
            _fail(f"plan execute rc={result.exit_code}\n{result.output}")

    cs: SpecChangeSet | None = store.get(cid)
    if cs is None:
        _fail("changeset not found")
    if len(cs.entries) != 1:
        _fail(f"changeset entry count={len(cs.entries)}")
    if cs.entries[0].operation != ChangeOperation.CREATE:
        _fail(f"expected CREATE, got {cs.entries[0].operation}")
    print("robot-lifecycle-plan-execute-changeset-ok")


def plan_apply_changeset() -> None:
    """Verify plan apply with ChangeSet summary verification."""
    svc = MagicMock()
    svc.apply_plan.return_value = _mock_plan(phase=PlanPhase.APPLY)
    store, cid = _make_store()

    store.record(
        cid,
        ChangeEntry(
            plan_id=_PLAN_ULID,
            resource_id="res-a",
            tool_name="builtin/file-write",
            operation=ChangeOperation.CREATE,
            path="src/new_module.py",
        ),
    )
    store.record(
        cid,
        ChangeEntry(
            plan_id=_PLAN_ULID,
            resource_id="res-a",
            tool_name="builtin/file-edit",
            operation=ChangeOperation.MODIFY,
            path="src/existing.py",
        ),
    )

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
    ):
        result = runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID])
        if result.exit_code != 0:
            _fail(f"plan apply rc={result.exit_code}\n{result.output}")

    summary = store.summarize(cid)
    if summary.get("total") != 2:
        _fail(f"total={summary.get('total')}")
    if summary.get("creates") != 1:
        _fail(f"creates={summary.get('creates')}")
    if summary.get("modifies") != 1:
        _fail(f"modifies={summary.get('modifies')}")
    print("robot-lifecycle-plan-apply-changeset-ok")


def full_lifecycle() -> None:
    """End-to-end: action create -> plan use -> execute -> apply with sandbox."""
    svc = MagicMock()
    svc.create_action.return_value = _mock_action()
    svc.get_action_by_name.return_value = _mock_action()
    svc.use_action.return_value = _mock_plan()
    svc.execute_plan.return_value = _mock_plan(phase=PlanPhase.EXECUTE)
    svc.get_plan.return_value = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE
    )
    svc.apply_plan.return_value = _mock_plan(phase=PlanPhase.APPLY)

    sandbox_dir = tempfile.mkdtemp(prefix="robot_full_sandbox_")
    yaml_path = _write_yaml(_VALID_YAML)
    store, cid = _make_store()
    store.record(
        cid,
        ChangeEntry(
            plan_id=_PLAN_ULID,
            resource_id="res-a",
            tool_name="builtin/file-write",
            operation=ChangeOperation.CREATE,
            path="src/generated.py",
        ),
    )

    try:
        with (
            patch(
                "cleveragents.cli.commands.action._get_lifecycle_service",
                return_value=svc,
            ),
            patch(
                "cleveragents.cli.commands.plan._get_lifecycle_service",
                return_value=svc,
            ),
            patch(
                "cleveragents.cli.commands.plan._get_plan_executor",
                return_value=MagicMock(),
            ),
        ):
            r1 = runner.invoke(action_app, ["create", "--config", yaml_path])
            assert r1.exit_code == 0, f"action create: {r1.output}"
            r2 = runner.invoke(plan_app, ["use", "local/lifecycle-action", "proj-a"])
            assert r2.exit_code == 0, f"plan use: {r2.output}"
            r3 = runner.invoke(plan_app, ["execute", _PLAN_ULID])
            assert r3.exit_code == 0, f"plan execute: {r3.output}"
            r4 = runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID])
            assert r4.exit_code == 0, f"plan apply: {r4.output}"

        cs = store.get(cid)
        assert cs is not None, "changeset not found"
        assert len(cs.entries) == 1, f"entries={len(cs.entries)}"
        print("robot-lifecycle-full-e2e-sandbox-ok")
    except AssertionError as exc:
        _fail(str(exc))
    finally:
        os.unlink(yaml_path)
        shutil.rmtree(sandbox_dir, ignore_errors=True)


# ---------------------------------------------------------------------------
# Negative subcommands
# ---------------------------------------------------------------------------


def invalid_project_name() -> None:
    """Verify invalid project name format is handled."""
    svc = MagicMock()
    svc.get_action_by_name.return_value = _mock_action()
    svc.use_action.side_effect = ValueError("Invalid project name")

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
    ):
        result = runner.invoke(plan_app, ["use", "local/lifecycle-action", "bad name!"])
        if (
            result.exit_code != 0
            or "error" in result.output.lower()
            or svc.use_action.called
        ):
            print("robot-lifecycle-invalid-project-name-ok")
        else:
            _fail("expected error for invalid project name")


def missing_resource() -> None:
    """Verify missing action/resource returns appropriate error."""
    from cleveragents.core.exceptions import NotFoundError

    svc = MagicMock()
    svc.get_action_by_name.side_effect = NotFoundError(
        "Action 'local/nonexistent' not found"
    )

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
    ):
        result = runner.invoke(plan_app, ["use", "local/nonexistent", "proj-a"])
        out = result.output.lower()
        if result.exit_code != 0 or "error" in out or "not found" in out:
            print("robot-lifecycle-missing-resource-ok")
        else:
            _fail(f"expected error, rc={result.exit_code}\n{result.output}")


def invalid_arg_format() -> None:
    """Verify --arg without '=' separator is rejected."""
    svc = MagicMock()
    svc.get_action_by_name.return_value = _mock_action()

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
    ):
        result = runner.invoke(
            plan_app,
            ["use", "local/lifecycle-action", "proj-a", "--arg", "no_equals"],
        )
        out = result.output.lower()
        if (
            result.exit_code != 0
            or "invalid argument format" in out
            or "aborted" in out
        ):
            print("robot-lifecycle-invalid-arg-format-ok")
        else:
            _fail(f"expected rejection, rc={result.exit_code}\n{result.output}")


# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, object] = {
    "action-create": action_create,
    "plan-use-sandbox": plan_use_sandbox,
    "plan-execute-changeset": plan_execute_changeset,
    "plan-apply-changeset": plan_apply_changeset,
    "full-lifecycle": full_lifecycle,
    "invalid-project-name": invalid_project_name,
    "missing-resource": missing_resource,
    "invalid-arg-format": invalid_arg_format,
}


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(f"Usage: helper_cli_lifecycle.py <{'|'.join(_COMMANDS)}>")
        return 1
    command = sys.argv[1]
    handler = _COMMANDS.get(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1
    handler()  # type: ignore[operator]
    return 0


if __name__ == "__main__":
    sys.exit(main())