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.

CLI-facing tests (1-4) invoke the real ``agents`` CLI via subprocess
without mocking the service layer.  Domain-level tests (5-9) exercise
real application code directly.
"""

from __future__ import annotations

import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from datetime import datetime
from pathlib import Path
from typing import NoReturn

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

# Ensure robot/ is on the import path for helper_e2e_common.
_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
    sys.path.insert(0, _ROBOT)

from helper_e2e_common import (  # noqa: E402
    cleanup_workspace,
    init_bare_git_repo,
    is_expected_provider_unavailable,
    run_cli,
    setup_workspace,
    write_yaml,
)

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

_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 _extract_plan_id(output: str) -> str | None:
    """Extract a ULID plan_id from plain CLI output.

    Searches for a 26-character ULID pattern (uppercase alphanumeric)
    which is the format used by ``python-ulid``.
    """
    import re

    match = re.search(r"\b([0-9A-Z]{26})\b", output)
    return match.group(1) if match else None


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


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


def action_create_from_yaml() -> None:
    """Verify action create from YAML config via real CLI subprocess."""
    workspace = setup_workspace(prefix="m1_action_")
    yaml_path = write_yaml(_VALID_ACTION_YAML)
    try:
        result = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
        if result.returncode != 0:
            _fail(f"action create rc={result.returncode}\n{result.stderr}")
        # Verify persistence: the action should be retrievable
        show = run_cli(
            "action",
            "show",
            "local/m1-verify-action",
            "--format",
            "plain",
            workspace=workspace,
        )
        if show.returncode != 0:
            _fail(f"action show rc={show.returncode}\n{show.stderr}")
        if "local/m1-verify-action" not in show.stdout:
            _fail(f"action not found in show output:\n{show.stdout}")
        print("m1-action-create-ok")
    finally:
        os.unlink(yaml_path)
        cleanup_workspace(workspace)


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


def resource_register_git_checkout() -> None:
    """Register a git-checkout resource via real CLI subprocess."""
    workspace = setup_workspace(prefix="m1_resource_")
    repo_dir = init_bare_git_repo()
    try:
        result = run_cli(
            "resource",
            "add",
            "git-checkout",
            "local/m1-repo",
            "--path",
            repo_dir,
            "--branch",
            "main",
            "--format",
            "plain",
            workspace=workspace,
        )
        if result.returncode != 0:
            _fail(f"resource add rc={result.returncode}\n{result.stderr}")
        # Verify persistence: the resource should be retrievable
        show = run_cli(
            "resource",
            "show",
            "local/m1-repo",
            "--format",
            "plain",
            workspace=workspace,
        )
        if show.returncode != 0:
            _fail(f"resource show rc={show.returncode}\n{show.stderr}")
        if "git-checkout" not in show.stdout:
            _fail(f"resource type not git-checkout:\n{show.stdout}")
        print("m1-resource-register-ok")
    finally:
        shutil.rmtree(repo_dir, ignore_errors=True)
        cleanup_workspace(workspace)


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


def project_create_and_link() -> None:
    """Create a project and link a resource via real CLI subprocess."""
    workspace = setup_workspace(prefix="m1_project_")
    repo_dir = init_bare_git_repo()
    try:
        # First register a resource (dependency for --resource flag)
        r1 = run_cli(
            "resource",
            "add",
            "git-checkout",
            "local/m1-repo",
            "--path",
            repo_dir,
            "--branch",
            "main",
            "--format",
            "plain",
            workspace=workspace,
        )
        if r1.returncode != 0:
            _fail(f"resource add rc={r1.returncode}\n{r1.stderr}")
        # Create project with resource link
        r2 = run_cli(
            "project",
            "create",
            "local/m1-project",
            "--description",
            "M1 test project",
            "--resource",
            "local/m1-repo",
            "--format",
            "plain",
            workspace=workspace,
        )
        if r2.returncode != 0:
            _fail(f"project create rc={r2.returncode}\n{r2.stderr}")
        # Verify persistence
        r3 = run_cli(
            "project",
            "show",
            "local/m1-project",
            "--format",
            "plain",
            workspace=workspace,
        )
        if r3.returncode != 0:
            _fail(f"project show rc={r3.returncode}\n{r3.stderr}")
        if "local/m1-project" not in r3.stdout:
            _fail(f"project not found in show output:\n{r3.stdout}")
        print("m1-project-create-link-ok")
    finally:
        shutil.rmtree(repo_dir, ignore_errors=True)
        cleanup_workspace(workspace)


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


def plan_full_lifecycle() -> None:
    """Run the plan lifecycle via real CLI subprocesses.

    Invokes action create -> plan use -> plan show -> plan execute ->
    plan diff -> plan lifecycle-apply using the real ``agents`` CLI
    binary.  Without an AI backend the plan stays in ``strategize /
    queued``, so ``execute``, ``diff``, and ``lifecycle-apply``
    return graceful "not ready" messages rather than crashing.  The
    test verifies the full CLI wiring (DI, persistence, env vars)
    end-to-end.  ChangeSet verification uses the domain-level tests
    (``changeset-invocations``), not this CLI lifecycle test.
    """
    workspace = setup_workspace(prefix="m1_plan_")
    repo_dir = init_bare_git_repo()
    yaml_path = write_yaml(_VALID_ACTION_YAML)
    try:
        # Setup: register resource + create project
        r_res = run_cli(
            "resource",
            "add",
            "git-checkout",
            "local/m1-repo",
            "--path",
            repo_dir,
            "--branch",
            "main",
            workspace=workspace,
        )
        if r_res.returncode != 0:
            _fail(f"resource add: {r_res.stderr}")
        r_proj = run_cli(
            "project",
            "create",
            "local/m1-project",
            "--resource",
            "local/m1-repo",
            workspace=workspace,
        )
        if r_proj.returncode != 0:
            _fail(f"project create: {r_proj.stderr}")

        # Step 1: Create action
        r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
        if r1.returncode != 0:
            _fail(f"action create: {r1.stderr}")

        # Step 2: Plan use (--format plain to extract plan_id)
        r2 = run_cli(
            "plan",
            "use",
            "local/m1-verify-action",
            "local/m1-project",
            "--format",
            "plain",
            workspace=workspace,
        )
        if r2.returncode != 0:
            _fail(f"plan use: {r2.stderr}")
        # Extract plan_id from plain output
        plan_id = _extract_plan_id(r2.stdout)
        if not plan_id:
            _fail(f"could not extract plan_id from plan use output:\n{r2.stdout}")

        # Step 3: Verify plan persisted — status must find it
        r3 = run_cli(
            "plan",
            "status",
            plan_id,
            "--format",
            "plain",
            workspace=workspace,
        )
        if r3.returncode != 0:
            _fail(f"plan status: {r3.stderr}")
        if plan_id not in r3.stdout:
            _fail(f"plan_id not in plan status output:\n{r3.stdout}")

        # Step 4: Plan execute — plan is strategize/queued without
        # an AI backend so it correctly rejects with "not ready".
        # The test verifies no internal server error (500); a
        # controlled abort with a user-facing message is expected.
        r4 = run_cli("plan", "execute", plan_id, workspace=workspace)
        combined4 = r4.stdout + r4.stderr
        if "Traceback" in combined4:
            _fail(f"plan execute crashed:\n{combined4}")
        if "INTERNAL" in combined4 and not is_expected_provider_unavailable(combined4):
            _fail(f"plan execute crashed:\n{combined4}")

        # Step 5: Plan diff — similar: plan not in execute phase
        r5 = run_cli("plan", "diff", plan_id, workspace=workspace)
        combined5 = r5.stdout + r5.stderr
        if "Traceback" in combined5:
            _fail(f"plan diff crashed:\n{combined5}")
        if "INTERNAL" in combined5 and not is_expected_provider_unavailable(combined5):
            _fail(f"plan diff crashed:\n{combined5}")

        # Step 6: Plan lifecycle-apply — similar: plan not ready
        r6 = run_cli("plan", "lifecycle-apply", "--yes", plan_id, workspace=workspace)
        combined6 = r6.stdout + r6.stderr
        if "Traceback" in combined6:
            _fail(f"plan apply crashed:\n{combined6}")
        if "INTERNAL" in combined6 and not is_expected_provider_unavailable(combined6):
            _fail(f"plan apply crashed:\n{combined6}")

        print("m1-plan-lifecycle-ok")
    finally:
        os.unlink(yaml_path)
        shutil.rmtree(repo_dir, ignore_errors=True)
        cleanup_workspace(workspace)


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