cleveragents-core/robot/helper_m2_e2e_verification.py

"""Robot Framework helper for M2 E2E verification tests.

Exercises the complete M2 success criteria sequence:
  1. Actor YAML creation and loading
  2. Actor compilation to LangGraph StateGraphs
  3. Tool router external tool resolution
  4. Validation runner with required/informational modes
  5. Multi-file generation producing a correct ChangeSet

CLI-facing tests (actor-add-config-cli, action-create, plan-use-execute)
invoke the real ``agents`` CLI via subprocess without mocking.
Domain-level tests exercise real application code directly.

Each subcommand prints a sentinel string on success and exits 0.
On failure it prints a diagnostic and exits 1.

Usage:
    python robot/helper_m2_e2e_verification.py <command> [args...]
"""

from __future__ import annotations

import json
import os
import re
import shutil
import sys
import tempfile
from pathlib import Path
from typing import Any, NoReturn

# Ensure src is importable when run from workspace root
_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,
    is_expected_provider_unavailable,
    run_cli,
    setup_workspace,
    write_yaml,
)
from helpers_common import reset_global_state  # noqa: E402

from cleveragents.actor.compiler import (  # noqa: E402
    ActorCompilationError,
    CompiledActor,
    compile_actor,
)
from cleveragents.actor.loader import ActorLoader  # noqa: E402
from cleveragents.actor.schema import (  # noqa: E402
    ActorConfigSchema,
    ActorType,
)
from cleveragents.domain.models.core.tool import (  # noqa: E402
    Validation,
    ValidationMode,
)
from cleveragents.tool.builtins.changeset import (  # noqa: E402
    ChangeSet,
    ChangeSetCapture,
)
from cleveragents.tool.builtins.file_tools import FILE_WRITE_SPEC  # noqa: E402
from cleveragents.tool.registry import ToolRegistry  # noqa: E402
from cleveragents.tool.router import (  # noqa: E402
    ProviderFormat,
    ToolCallRouter,
    detect_provider_format,
    normalize_tool_call,
)
from cleveragents.tool.runner import ToolRunner  # noqa: E402
from cleveragents.tool.runtime import ToolSpec  # noqa: E402


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


def _extract_plan_id(output: str) -> str | None:
    """Extract a ULID plan_id from plain CLI output."""
    match = re.search(r"\b([0-9A-Z]{26})\b", output)
    return match.group(1) if match else None


# -----------------------------------------------------------------------
# Helpers for domain-level tests (unchanged)
# -----------------------------------------------------------------------


def _echo(inputs: dict[str, Any]) -> dict[str, Any]:
    return dict(inputs)


def _validation_pass(inputs: dict[str, Any]) -> dict[str, Any]:
    return {"passed": True, "message": "validation passed"}


def _validation_fail(inputs: dict[str, Any]) -> dict[str, Any]:
    return {"passed": False, "message": "validation failed"}


# -----------------------------------------------------------------------
# Subcommand: actor-yaml-create-load (domain-level, no CLI)
# -----------------------------------------------------------------------


def actor_yaml_create_load() -> None:
    """Create a temporary actor YAML and load it via ActorLoader."""
    reset_global_state()
    tmp = Path(tempfile.mkdtemp(prefix="m2_actor_"))
    yaml_content = (
        "name: local/m2-test-actor\n"
        "type: graph\n"
        "description: M2 E2E test actor\n"
        'version: "1.0"\n'
        "provider: openai\n"
        "model: gpt-4\n"
        "route:\n"
        "  nodes:\n"
        "    - id: planner\n"
        "      type: agent\n"
        "      name: Planner\n"
        "      description: Plans the task\n"
        "      config:\n"
        "        model: gpt-4\n"
        "        prompt: Plan the task\n"
        "    - id: executor\n"
        "      type: tool\n"
        "      name: Executor\n"
        "      description: Executes the plan\n"
        "      config:\n"
        "        tool_name: builtins/echo\n"
        "  edges:\n"
        "    - from_node: planner\n"
        "      to_node: executor\n"
        "  entry_node: planner\n"
        "  exit_nodes:\n"
        "    - executor\n"
    )
    yaml_path = tmp / "m2_actor.yaml"
    yaml_path.write_text(yaml_content)

    loader = ActorLoader(search_roots=[tmp])
    actors = loader.discover()
    assert len(actors) == 1, f"Expected 1 actor, got {len(actors)}"
    actor = actors[0]
    assert actor.name == "local/m2-test-actor", f"Wrong name: {actor.name}"
    assert actor.type == ActorType.GRAPH, f"Wrong type: {actor.type}"
    print("m2-actor-yaml-create-load-ok")


# -----------------------------------------------------------------------
# Subcommand: actor-add-config-cli (CLI via subprocess)
# -----------------------------------------------------------------------


def actor_add_config_cli() -> None:
    """Add an actor via ``agents actor add --config`` using subprocess."""
    reset_global_state()
    workspace = setup_workspace(prefix="m2_actor_cli_")
    config_dir = tempfile.mkdtemp(prefix="m2_actor_cfg_")
    config_path = os.path.join(config_dir, "actor.json")
    with open(config_path, "w") as f:
        json.dump(
            {
                "name": "local/m2-cli-actor",
                "provider": "openai",
                "model": "gpt-4",
                "options": {"temperature": 0.5},
            },
            f,
        )
    try:
        result = run_cli(
            "actor",
            "add",
            "local/m2-cli-actor",
            "--config",
            config_path,
            "--format",
            "plain",
            workspace=workspace,
        )
        if result.returncode != 0:
            _fail(
                f"actor add rc={result.returncode}\n"
                f"stdout: {result.stdout}\nstderr: {result.stderr}"
            )
        print("m2-actor-add-config-cli-ok")
    finally:
        shutil.rmtree(config_dir, ignore_errors=True)
        cleanup_workspace(workspace)


# -----------------------------------------------------------------------
# Subcommand: action-create (CLI via subprocess)
# -----------------------------------------------------------------------


_ACTION_YAML = """\
name: local/m2-test-action
description: M2 test action
definition_of_done: All tests pass
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
"""


def action_create() -> None:
    """Create an action via real CLI subprocess."""
    reset_global_state()
    workspace = setup_workspace(prefix="m2_action_")
    yaml_path = write_yaml(_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"
                f"stdout: {result.stdout}\nstderr: {result.stderr}"
            )
        # Verify persistence: action should be retrievable
        show = run_cli(
            "action",
            "show",
            "local/m2-test-action",
            "--format",
            "plain",
            workspace=workspace,
        )
        if show.returncode != 0:
            _fail(f"action show rc={show.returncode}\nstderr: {show.stderr}")
        if "local/m2-test-action" not in show.stdout:
            _fail(f"action not found in show output:\n{show.stdout}")
        print("m2-action-create-ok")
    finally:
        os.unlink(yaml_path)
        cleanup_workspace(workspace)


# -----------------------------------------------------------------------
# Subcommand: plan-use-execute (CLI via subprocess)
# -----------------------------------------------------------------------


def plan_use_execute() -> None:
    """Run plan use and plan execute via real CLI subprocesses."""
    reset_global_state()
    workspace = setup_workspace(prefix="m2_plan_")
    yaml_path = write_yaml(_ACTION_YAML)
    try:
        # 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
        r2 = run_cli(
            "plan",
            "use",
            "local/m2-test-action",
            "--format",
            "plain",
            workspace=workspace,
        )
        if r2.returncode != 0:
            _fail(f"plan use: {r2.stderr}")
        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: Plan execute — plan is strategize/queued so it
        # correctly rejects with "not ready" (controlled abort).
        r3 = run_cli("plan", "execute", plan_id, workspace=workspace)
        combined = r3.stdout + r3.stderr
        if "Traceback" in combined:
            _fail(f"plan execute crashed:\n{combined}")
        if "INTERNAL" in combined and not is_expected_provider_unavailable(combined):
            _fail(f"plan execute crashed:\n{combined}")

        print("m2-plan-use-execute-ok")
    finally:
        os.unlink(yaml_path)
        cleanup_workspace(workspace)


# -----------------------------------------------------------------------
# Subcommand: actor-yaml-parse-validate (domain-level, no CLI)
# -----------------------------------------------------------------------


def actor_yaml_parse_validate() -> None:
    """Verify actor YAML files parse and validate correctly."""
    reset_global_state()
    # Test valid graph YAML
    tmp = Path(tempfile.mkdtemp(prefix="m2_parse_"))
    valid_yaml = (
        "name: local/valid-actor\n"
        "type: graph\n"
        "description: Valid actor\n"
        'version: "1.0"\n'
        "provider: openai\n"
        "model: gpt-4\n"
        "route:\n"
        "  nodes:\n"
        "    - id: start\n"
        "      type: agent\n"
        "      name: Start\n"
        "      description: Start node\n"
        "      config:\n"
        "        model: gpt-4\n"
        "        prompt: Begin\n"
        "  edges: []\n"
        "  entry_node: start\n"
        "  exit_nodes:\n"
        "    - start\n"
    )
    (tmp / "valid.yaml").write_text(valid_yaml)
    config = ActorConfigSchema.from_yaml_file(str(tmp / "valid.yaml"))
    assert config.name == "local/valid-actor"
    assert config.type == ActorType.GRAPH
    assert config.route is not None
    assert len(config.route.nodes) == 1

    # Test valid LLM YAML
    llm_yaml = (
        "name: local/llm-actor\n"
        "type: llm\n"
        "description: LLM actor\n"
        'version: "1.0"\n'
        "provider: openai\n"
        "model: gpt-4\n"
    )
    (tmp / "llm.yaml").write_text(llm_yaml)
    llm_config = ActorConfigSchema.from_yaml_file(str(tmp / "llm.yaml"))
    assert llm_config.type == ActorType.LLM

    # Test invalid YAML is rejected
    bad_yaml = "name: [\ninvalid yaml\n"
    (tmp / "bad.yaml").write_text(bad_yaml)
    try:
        ActorConfigSchema.from_yaml_file(str(tmp / "bad.yaml"))
        print("FAIL: bad YAML not rejected")
        sys.exit(1)
    except Exception:
        pass  # Expected

    print("m2-actor-yaml-parse-validate-ok")


# -----------------------------------------------------------------------
# Subcommand: actor-compile-stategraph (domain-level, no CLI)
# -----------------------------------------------------------------------


def actor_compile_stategraph() -> None:
    """Verify actors compile to LangGraph StateGraphs."""
    reset_global_state()
    project_root = Path(__file__).resolve().parents[1]
    yaml_path = project_root / "examples" / "actors" / "simple_graph.yaml"
    config = ActorConfigSchema.from_yaml_file(str(yaml_path))
    compiled = compile_actor(config)

    assert isinstance(compiled, CompiledActor)
    assert compiled.name == config.name
    assert len(compiled.nodes) > 0, "Expected at least one node"
    assert len(compiled.edges) > 0, "Expected at least one edge"
    assert compiled.entry_point != "", "Entry point must not be empty"
    assert len(compiled.metadata.node_ids) > 0
    assert compiled.metadata.entry_node != ""

    # Verify LLM actors are rejected
    tmp = Path(tempfile.mkdtemp(prefix="m2_compile_"))
    llm_yaml = (
        "name: local/llm-only\n"
        "type: llm\n"
        "description: LLM only actor\n"
        'version: "1.0"\n'
        "provider: openai\n"
        "model: gpt-4\n"
    )
    (tmp / "llm.yaml").write_text(llm_yaml)
    llm_config = ActorConfigSchema.from_yaml_file(str(tmp / "llm.yaml"))
    try:
        compile_actor(llm_config)
        print("FAIL: LLM actor should not compile")
        sys.exit(1)
    except ActorCompilationError:
        pass  # Expected

    print("m2-actor-compile-stategraph-ok")


# -----------------------------------------------------------------------
# Subcommand: tool-router-resolve-external (domain-level, no CLI)
# -----------------------------------------------------------------------


def tool_router_resolve_external() -> None:
    """Verify tool router can resolve external tools."""
    reset_global_state()
    registry = ToolRegistry()
    ext_spec = ToolSpec(
        name="external/custom-tool",
        description="External custom tool",
        handler=_echo,
        source="custom",
    )
    registry.register(ext_spec)

    tool_runner_inst = ToolRunner(registry)
    router = ToolCallRouter(
        registry=registry, runner=tool_runner_inst, plan_id="m2-plan-1"
    )

    # Route an OpenAI-format call to the external tool
    result = router.route(
        {
            "name": "external/custom-tool",
            "arguments": json.dumps({"key": "value"}),
        }
    )
    assert result.result.success, f"Expected success, got error: {result.result.error}"
    assert result.provider_format == ProviderFormat.OPENAI

    # Route an Anthropic-format call
    result2 = router.route(
        {
            "name": "external/custom-tool",
            "input": {"key": "value"},
        }
    )
    assert result2.result.success
    assert result2.provider_format == ProviderFormat.ANTHROPIC

    # Verify detection and normalization
    fmt = detect_provider_format({"name": "x/y", "arguments": '{"a": 1}'})
    assert fmt == ProviderFormat.OPENAI
    req = normalize_tool_call(
        {"name": "external/custom-tool", "arguments": '{"k":"v"}'}
    )
    assert req.tool_name == "external/custom-tool"

    print("m2-tool-router-resolve-external-ok")


# -----------------------------------------------------------------------
# Subcommand: validation-runner-execute (domain-level, no CLI)
# -----------------------------------------------------------------------


def validation_runner_execute() -> None:
    """Verify validation runner executes required/informational validations."""
    reset_global_state()
    # Create a required validation
    required_val = Validation.from_config(
        {
            "name": "local/coverage-check",
            "description": "Check code coverage",
            "source": "custom",
            "mode": "required",
            "code": (
                "def run(inputs): return {'passed': inputs.get('coverage', 0) >= 80}"
            ),
        }
    )
    assert required_val.mode == ValidationMode.REQUIRED

    # Create an informational validation
    info_val = Validation.from_config(
        {
            "name": "local/lint-check",
            "description": "Lint check",
            "source": "custom",
            "mode": "informational",
            "code": (
                "def run(inputs): return {'passed': False, 'message': 'lint warnings'}"
            ),
        }
    )
    assert info_val.mode == ValidationMode.INFORMATIONAL

    # Register validation tools in registry and execute through runner
    registry = ToolRegistry()
    required_spec = ToolSpec(
        name="local/coverage-check",
        description="Coverage validation",
        handler=_validation_pass,
    )
    info_spec = ToolSpec(
        name="local/lint-check",
        description="Lint validation",
        handler=_validation_fail,
    )
    registry.register(required_spec)
    registry.register(info_spec)
    tool_runner_inst = ToolRunner(registry)

    # Execute required validation (pass)
    req_result = tool_runner_inst.execute("local/coverage-check", {"coverage": 95})
    assert req_result.success, f"Required validation failed: {req_result.error}"
    assert req_result.output.get("passed") is True

    # Execute informational validation (fail -- should not block)
    info_result = tool_runner_inst.execute("local/lint-check", {"code": "x"})
    assert info_result.success, (
        f"Informational validation execution failed: {info_result.error}"
    )
    assert info_result.output.get("passed") is False

    # Verify as_cli_dict renders mode
    cli_dict = required_val.as_cli_dict()
    assert cli_dict["mode"] == "required"
    info_cli = info_val.as_cli_dict()
    assert info_cli["mode"] == "informational"

    print("m2-validation-runner-execute-ok")


# -----------------------------------------------------------------------
# Subcommand: changeset-multifile (domain-level, no CLI)
# -----------------------------------------------------------------------


def changeset_multifile() -> None:
    """Verify multi-file generation produces correct ChangeSet."""
    reset_global_state()
    tmpdir = tempfile.mkdtemp(prefix="m2_changeset_")
    capture = ChangeSetCapture(
        plan_id="m2-plan-cs",
        resource_id="res-m2",
        sandbox_root=tmpdir,
    )
    wrapped = capture.wrap_tool(FILE_WRITE_SPEC)

    # Write multiple files
    wrapped.handler(
        {
            "path": "file_a.py",
            "content": "# File A\nprint('hello')\n",
            "sandbox_root": tmpdir,
        }
    )
    wrapped.handler(
        {
            "path": "file_b.py",
            "content": "# File B\nprint('world')\n",
            "sandbox_root": tmpdir,
        }
    )
    wrapped.handler(
        {
            "path": "README.md",
            "content": "# Readme\nSample project\n",
            "sandbox_root": tmpdir,
        }
    )

    cs = capture.get_changeset()
    assert isinstance(cs, ChangeSet)
    assert len(cs.entries) == 3, f"Expected 3 entries, got {len(cs.entries)}"

    # Verify each entry
    paths = [e.path for e in cs.entries]
    assert "file_a.py" in paths
    assert "file_b.py" in paths
    assert "README.md" in paths

    # All operations should be 'create'
    for entry in cs.entries:
        assert entry.operation == "create", (
            f"Expected 'create', got '{entry.operation}' for {entry.path}"
        )

    # Verify summary is a non-empty string describing 3 changes
    summary_str = cs.summary()
    assert "3" in summary_str, f"Expected '3' in summary: {summary_str}"
    assert "create" in summary_str, f"Expected 'create' in summary: {summary_str}"

    # Also verify via the domain-level SpecChangeSet summary (returns dict)
    spec_cs = capture.to_spec_changeset()
    spec_summary = spec_cs.summary()
    assert spec_summary["total"] == 3
    assert spec_summary["creates"] == 3

    print("m2-changeset-multifile-ok")


# -----------------------------------------------------------------------
# Dispatcher
# -----------------------------------------------------------------------

_COMMANDS: dict[str, object] = {
    "actor-yaml-create-load": actor_yaml_create_load,
    "actor-add-config-cli": actor_add_config_cli,
    "action-create": action_create,
    "plan-use-execute": plan_use_execute,
    "actor-yaml-parse-validate": actor_yaml_parse_validate,
    "actor-compile-stategraph": actor_compile_stategraph,
    "tool-router-resolve-external": tool_router_resolve_external,
    "validation-runner-execute": validation_runner_execute,
    "changeset-multifile": changeset_multifile,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
        sys.exit(1)
    fn = _COMMANDS[sys.argv[1]]
    fn()  # type: ignore[operator]