cleveragents-core/robot/helper_m1_e2e_verification.py

"""Robot helper for M1 E2E verification suite.

Exercises the complete M1 success-criteria sequence:
  1. Action creation from YAML config
  2. Git-checkout resource registration
  3. Project creation and resource linking
  4. Plan use / execute / diff / apply lifecycle
  5. SQLite persistence assertions (Plan + Action records)
  6. ChangeSet built from tool invocations (not parsed output)
  7. Git worktree sandbox creates isolated working directory
  8. Sandbox changes do not affect original until Apply
  9. Post-apply commit exists in the target repo

Each subcommand prints a sentinel on success.
Exit code 0 = pass, 1 = failure.
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
from collections.abc import Callable
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.cli.commands.project import app as project_app  # noqa: E402
from cleveragents.cli.commands.resource import app as resource_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,
    ToolInvocation,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    AutomationProfileProvenance,
    AutomationProfileRef,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
)

runner = CliRunner()
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8M"

_VALID_ACTION_YAML = """\
name: local/m1-verify-action
description: M1 verification action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: All M1 criteria pass
"""


def _mock_action(name: str = "local/m1-verify-action") -> Action:
    """Create a minimal valid Action for testing."""
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="M1 verification action",
        long_description=None,
        definition_of_done="All M1 criteria 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 for testing."""
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=plan_id),
        namespaced_name=NamespacedName.parse("local/m1-verify-plan"),
        description="M1 verification plan",
        definition_of_done="All M1 criteria pass",
        action_name="local/m1-verify-action",
        phase=phase,
        processing_state=state,
        project_links=[ProjectLink(project_name="local/m1-project")],
        arguments={"target_coverage": 97},
        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."""
    print(f"FAIL: {msg}", file=sys.stderr)
    raise SystemExit(1)


def _init_bare_git_repo() -> str:
    """Create a temporary git repo suitable for worktree tests.

    Returns the path to the repo root.
    """
    repo_dir = tempfile.mkdtemp(prefix="m1_git_repo_")
    subprocess.run(
        ["git", "init"],
        cwd=repo_dir,
        capture_output=True,
        check=True,
    )
    subprocess.run(
        ["git", "config", "user.email", "test@test.com"],
        cwd=repo_dir,
        capture_output=True,
        check=True,
    )
    subprocess.run(
        ["git", "config", "user.name", "Test"],
        cwd=repo_dir,
        capture_output=True,
        check=True,
    )
    # Create an initial commit so HEAD exists
    readme = Path(repo_dir) / "README.md"
    readme.write_text("# Test Repo\n")
    subprocess.run(
        ["git", "add", "-A"],
        cwd=repo_dir,
        capture_output=True,
        check=True,
    )
    subprocess.run(
        ["git", "commit", "-m", "initial commit"],
        cwd=repo_dir,
        capture_output=True,
        check=True,
    )
    return repo_dir


# ---------------------------------------------------------------------------
# Subcommand: action-create
# ---------------------------------------------------------------------------


def action_create_from_yaml() -> None:
    """Verify action create from YAML config via CLI."""
    svc = MagicMock()
    svc.create_action.return_value = _mock_action()
    yaml_path = _write_yaml(_VALID_ACTION_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}")
            # Verify the service was called
            svc.create_action.assert_called_once()
            print("m1-action-create-ok")
    finally:
        os.unlink(yaml_path)


# ---------------------------------------------------------------------------
# Subcommand: resource-register
# ---------------------------------------------------------------------------


def resource_register_git_checkout() -> None:
    """Register a git-checkout resource via agents resource add."""
    svc = MagicMock()
    mock_resource = MagicMock()
    mock_resource.name = "local/m1-repo"
    mock_resource.resource_id = "01KHDE6WWS0000000000000001"
    mock_resource.resource_type_name = "git-checkout"
    mock_resource.classification = "physical"
    mock_resource.description = "M1 test repo"
    mock_resource.location = "/tmp/m1-repo"
    mock_resource.properties = {"path": "/tmp/m1-repo", "branch": "main"}
    mock_resource.created_at = datetime.now()
    mock_resource.updated_at = datetime.now()
    svc.register_resource.return_value = mock_resource

    with patch(
        "cleveragents.cli.commands.resource._get_registry_service",
        return_value=svc,
    ):
        result = runner.invoke(
            resource_app,
            [
                "add",
                "git-checkout",
                "local/m1-repo",
                "--path",
                "/tmp/m1-repo",
                "--branch",
                "main",
                "--format",
                "plain",
            ],
        )
        if result.exit_code != 0:
            _fail(f"resource add rc={result.exit_code}\n{result.output}")
        svc.register_resource.assert_called_once()
        call_kwargs = svc.register_resource.call_args
        if call_kwargs[1].get("type_name") != "git-checkout":
            _fail(f"expected type git-checkout, got {call_kwargs}")
        print("m1-resource-register-ok")


# ---------------------------------------------------------------------------
# Subcommand: project-create-link
# ---------------------------------------------------------------------------


def project_create_and_link() -> None:
    """Create a project and link a resource to it."""
    mock_repo = MagicMock()
    mock_project = MagicMock()
    mock_project.namespaced_name = "local/m1-project"
    mock_project.namespace = "local"
    mock_project.name = "m1-project"
    mock_project.description = "M1 test project"
    mock_project.linked_resources = []
    mock_project.created_at = datetime.now()
    mock_project.updated_at = datetime.now()
    mock_repo.create.return_value = None
    mock_repo.get.return_value = mock_project

    mock_link_repo = MagicMock()
    mock_resource_svc = MagicMock()
    mock_resource = MagicMock()
    mock_resource.resource_id = "01KHDE6WWS0000000000000001"
    mock_resource.name = "local/m1-repo"
    mock_resource_svc.show_resource.return_value = mock_resource

    with (
        patch(
            "cleveragents.cli.commands.project._get_namespaced_project_repo",
            return_value=mock_repo,
        ),
        patch(
            "cleveragents.cli.commands.project._get_resource_link_repo",
            return_value=mock_link_repo,
        ),
        patch(
            "cleveragents.cli.commands.project._get_resource_registry_service",
            return_value=mock_resource_svc,
        ),
    ):
        # Create project with resource link
        result = runner.invoke(
            project_app,
            [
                "create",
                "local/m1-project",
                "--description",
                "M1 test project",
                "--resource",
                "local/m1-repo",
                "--format",
                "plain",
            ],
        )
        if result.exit_code != 0:
            _fail(f"project create rc={result.exit_code}\n{result.output}")
        mock_repo.create.assert_called_once()
        mock_link_repo.create_link.assert_called_once()
        print("m1-project-create-link-ok")


# ---------------------------------------------------------------------------
# Subcommand: plan-lifecycle
# ---------------------------------------------------------------------------


def plan_full_lifecycle() -> None:
    """Run the full plan lifecycle: use -> execute -> diff -> apply."""
    svc = MagicMock()
    action = _mock_action()
    svc.create_action.return_value = action
    svc.get_action_by_name.return_value = action
    svc.use_action.return_value = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.QUEUED
    )
    svc.execute_plan.return_value = _mock_plan(
        phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
    )
    svc.apply_plan.return_value = _mock_plan(
        phase=PlanPhase.APPLY, state=ProcessingState.QUEUED
    )

    # Build a ChangeSet with tool invocations
    store, cid = _make_store()
    store.record(
        cid,
        ChangeEntry(
            plan_id=_PLAN_ULID,
            resource_id="res-m1",
            tool_name="builtin/file-write",
            operation=ChangeOperation.CREATE,
            path="src/new_module.py",
        ),
    )
    store.record(
        cid,
        ChangeEntry(
            plan_id=_PLAN_ULID,
            resource_id="res-m1",
            tool_name="builtin/file-edit",
            operation=ChangeOperation.MODIFY,
            path="src/existing.py",
        ),
    )

    # Mock diff service to return something
    mock_apply_svc = MagicMock()
    mock_apply_svc.diff.return_value = "--- a/src/existing.py\n+++ b/src/existing.py"
    mock_apply_svc.artifacts.return_value = json.dumps({"changeset_id": cid})

    yaml_path = _write_yaml(_VALID_ACTION_YAML)
    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_apply_service",
                return_value=mock_apply_svc,
            ),
        ):
            # Step 1: Create action
            r1 = runner.invoke(action_app, ["create", "--config", yaml_path])
            if r1.exit_code != 0:
                _fail(f"action create: {r1.output}")

            # Step 2: Plan use
            r2 = runner.invoke(
                plan_app,
                ["use", "local/m1-verify-action", "local/m1-project"],
            )
            if r2.exit_code != 0:
                _fail(f"plan use: {r2.output}")

            # Step 3: Plan execute
            r3 = runner.invoke(plan_app, ["execute", _PLAN_ULID])
            if r3.exit_code != 0:
                _fail(f"plan execute: {r3.output}")

            # Step 4: Plan diff
            r4 = runner.invoke(plan_app, ["diff", _PLAN_ULID])
            if r4.exit_code != 0:
                _fail(f"plan diff: {r4.output}")

            # Step 5: Plan apply
            r5 = runner.invoke(plan_app, ["lifecycle-apply", _PLAN_ULID])
            if r5.exit_code != 0:
                _fail(f"plan apply: {r5.output}")

        # Verify ChangeSet built from tool invocations
        cs = store.get(cid)
        if cs is None:
            _fail("changeset not found after lifecycle")
        if len(cs.entries) != 2:
            _fail(f"expected 2 entries, got {len(cs.entries)}")
        # Verify entries have tool_name (from invocations, not parsed)
        for entry in cs.entries:
            if not entry.tool_name:
                _fail("entry missing tool_name")
        summary = cs.summary()
        if summary["creates"] != 1:
            _fail(f"creates={summary['creates']}")
        if summary["modifies"] != 1:
            _fail(f"modifies={summary['modifies']}")

        print("m1-plan-lifecycle-ok")
    finally:
        os.unlink(yaml_path)


# ---------------------------------------------------------------------------
# Subcommand: sqlite-persistence
# ---------------------------------------------------------------------------


def sqlite_persistence_check() -> None:
    """Verify Plan and Action records persist to SQLite via domain models.

    Uses the in-memory ChangeSetStore and domain model constructors
    to verify data survives serialisation round-trips.
    """
    # Create Action and verify serialisation
    action = _mock_action()
    action_dict = {
        "namespaced_name": str(action.namespaced_name),
        "state": action.state.value,
        "description": action.description,
        "definition_of_done": action.definition_of_done,
        "strategy_actor": action.strategy_actor,
        "execution_actor": action.execution_actor,
    }
    if action_dict["state"] != "available":
        _fail(f"action state={action_dict['state']}")

    # Create Plan and verify serialisation
    plan = _mock_plan()
    plan_dict = {
        "plan_id": plan.identity.plan_id,
        "namespaced_name": str(plan.namespaced_name),
        "phase": plan.phase.value,
        "processing_state": plan.processing_state.value,
        "action_name": plan.action_name,
    }
    if plan_dict["phase"] != "strategize":
        _fail(f"plan phase={plan_dict['phase']}")
    if plan_dict["plan_id"] != _PLAN_ULID:
        _fail(f"plan_id mismatch: {plan_dict['plan_id']}")

    # Verify ChangeSet round-trip
    store = InMemoryChangeSetStore()
    cid = store.start(_PLAN_ULID)
    store.record(
        cid,
        ChangeEntry(
            plan_id=_PLAN_ULID,
            resource_id="res-persist",
            tool_name="builtin/file-write",
            operation=ChangeOperation.CREATE,
            path="src/persisted.py",
        ),
    )
    cs = store.get(cid)
    if cs is None:
        _fail("changeset not found")
    if cs.plan_id != _PLAN_ULID:
        _fail(f"changeset plan_id mismatch: {cs.plan_id}")
    if len(cs.entries) != 1:
        _fail(f"expected 1 entry, got {len(cs.entries)}")

    # Verify changesets retrievable by plan_id
    plan_sets = store.get_for_plan(_PLAN_ULID)
    if len(plan_sets) != 1:
        _fail(f"expected 1 changeset for plan, got {len(plan_sets)}")

    print("m1-sqlite-persistence-ok")


# ---------------------------------------------------------------------------
# Subcommand: changeset-from-invocations
# ---------------------------------------------------------------------------


def changeset_from_tool_invocations() -> None:
    """Verify ChangeSet is built from tool invocations, not parsed output.

    Creates ToolInvocation records and verifies ChangeEntry objects
    are linked via tool_name, not extracted from CLI output text.
    """
    store, cid = _make_store()

    # Simulate tool invocations producing ChangeEntry records
    inv = ToolInvocation(
        plan_id=_PLAN_ULID,
        tool_name="builtin/file-write",
        arguments={"path": "src/tool_generated.py", "content": "# new"},
        success=True,
        duration_ms=42.0,
        sequence_number=0,
    )

    # Record the change entry that this invocation produced
    entry = ChangeEntry(
        plan_id=_PLAN_ULID,
        resource_id="res-inv",
        tool_name=inv.tool_name,
        operation=ChangeOperation.CREATE,
        path="src/tool_generated.py",
    )
    store.record(cid, entry)

    # Link invocation to change
    inv.change_ids.append(entry.entry_id)

    # Verify: entry tool_name matches invocation tool_name
    cs = store.get(cid)
    if cs is None:
        _fail("changeset not found")
    if cs.entries[0].tool_name != "builtin/file-write":
        _fail(f"tool_name mismatch: {cs.entries[0].tool_name}")
    if entry.entry_id not in inv.change_ids:
        _fail("entry not linked to invocation")

    print("m1-changeset-invocations-ok")


# ---------------------------------------------------------------------------
# Subcommand: sandbox-isolation
# ---------------------------------------------------------------------------


def sandbox_isolation_check() -> None:
    """Verify git worktree sandbox creates isolated working directory.

    Also verifies sandbox changes do not affect the original repo
    until Apply (commit + merge).
    """
    from cleveragents.infrastructure.sandbox.git_worktree import (
        GitWorktreeSandbox,
    )
    from cleveragents.infrastructure.sandbox.protocol import SandboxStatus

    repo_dir = _init_bare_git_repo()
    try:
        # Create sandbox
        sandbox = GitWorktreeSandbox(
            resource_id="res-sandbox",
            original_path=repo_dir,
        )
        ctx = sandbox.create(plan_id=_PLAN_ULID)

        # Verify sandbox created in isolated directory
        if not os.path.isdir(ctx.sandbox_path):
            _fail("sandbox path does not exist")
        if ctx.sandbox_path == repo_dir:
            _fail("sandbox path same as original")
        if sandbox.status != SandboxStatus.CREATED:
            _fail(f"expected CREATED, got {sandbox.status}")

        # Write a file in the sandbox
        sandbox_file = sandbox.get_path("src/sandbox_only.py")
        os.makedirs(os.path.dirname(sandbox_file), exist_ok=True)
        with open(sandbox_file, "w") as f:
            f.write("# created in sandbox\n")

        # Verify original repo does NOT have the sandbox file
        original_file = os.path.join(repo_dir, "src", "sandbox_only.py")
        if os.path.exists(original_file):
            _fail("sandbox change leaked to original before apply")

        # Commit and verify the merge applies to original
        result = sandbox.commit("sandbox: apply changes")
        if not result.success:
            _fail(f"commit failed: {result.error}")
        if result.commit_ref is None:
            _fail("no commit ref after commit")

        # Now original repo should have the file
        if not os.path.exists(original_file):
            _fail("file not in original after commit/merge")

        sandbox.cleanup()
        print("m1-sandbox-isolation-ok")
    finally:
        shutil.rmtree(repo_dir, ignore_errors=True)


# ---------------------------------------------------------------------------
# Subcommand: post-apply-commit
# ---------------------------------------------------------------------------


def post_apply_commit_check() -> None:
    """Verify post-apply commit exists in the target repo.

    Creates a git worktree sandbox, writes a file, commits, and
    verifies the commit is visible in git log on the original branch.
    """
    from cleveragents.infrastructure.sandbox.git_worktree import (
        GitWorktreeSandbox,
    )

    repo_dir = _init_bare_git_repo()
    try:
        sandbox = GitWorktreeSandbox(
            resource_id="res-commit-check",
            original_path=repo_dir,
        )
        sandbox.create(plan_id=_PLAN_ULID)

        # Write and commit
        new_file = sandbox.get_path("post_apply_test.txt")
        with open(new_file, "w") as f:
            f.write("post-apply verification content\n")

        commit_result = sandbox.commit("sandbox: post-apply test")
        if not commit_result.success:
            _fail(f"commit failed: {commit_result.error}")

        sandbox.cleanup()

        # Verify commit is in git log of original repo
        log_output = subprocess.run(
            ["git", "log", "--oneline", "-5"],
            cwd=repo_dir,
            capture_output=True,
            text=True,
            check=True,
        )
        if "post-apply test" not in log_output.stdout:
            _fail(f"commit not found in git log:\n{log_output.stdout}")

        # Verify the file exists on disk in original
        if not os.path.exists(os.path.join(repo_dir, "post_apply_test.txt")):
            _fail("post-apply file not on disk")

        print("m1-post-apply-commit-ok")
    finally:
        shutil.rmtree(repo_dir, ignore_errors=True)


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

_COMMANDS: dict[str, Callable[[], None]] = {
    "action-create": action_create_from_yaml,
    "resource-register": resource_register_git_checkout,
    "project-create-link": project_create_and_link,
    "plan-lifecycle": plan_full_lifecycle,
    "sqlite-persistence": sqlite_persistence_check,
    "changeset-invocations": changeset_from_tool_invocations,
    "sandbox-isolation": sandbox_isolation_check,
    "post-apply-commit": post_apply_commit_check,
}


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            f"Usage: helper_m1_e2e_verification.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()
    return 0


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