cleveragents-core/robot/helper_m4_e2e_verification.py

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

Exercises the M4 success criteria for subplans and parallel execution:
  1. Plan spawns multiple subplans during Execute
  2. View subplan tree via plan tree
  3. Verify merged results via plan diff
  4. Parallel subplan execution respects max_parallel
  5. Three-way merge combines non-conflicting changes
  6. Merge conflicts are surfaced correctly
  7. Parent plan tracks all subplan statuses
  8. CLI plan use creates plan with subplan config
  9. CLI plan execute transitions plan with subplans
  10. CLI plan tree displays subplan hierarchy
  11. CLI plan execute aborts on read-only plan (error path)
  12. CLI plan use aborts on unavailable action (error path)
  13. CLI plan diff aborts on missing changeset (error path)
  14. CLI plan tree handles zero decisions (error path)

CLI-facing tests (plan-diff, cli-plan-use, cli-plan-execute, cli-plan-tree)
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 to stderr and exits 1.

Usage:
    python robot/helper_m4_e2e_verification.py <command>
"""

from __future__ import annotations

import importlib
import sys
from collections.abc import Callable
from pathlib import Path

# 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.append(_SRC)

from helpers_common import reset_global_state  # noqa: E402

# Command → (module_name, function_name) dispatch table.
# Modules are imported lazily so that running a domain-only command
# (e.g. ``merge-clean``) does not load the CLI module's heavy imports
# (``typer.testing``, ``cleveragents.cli.commands.plan``).
_COMMAND_MAP: dict[str, tuple[str, str]] = {
    # Domain tests
    "spawn-subplans": ("helper_m4_e2e_domain", "spawn_subplans"),
    "plan-tree": ("helper_m4_e2e_domain", "plan_tree"),
    "parallel-max": ("helper_m4_e2e_domain", "parallel_max"),
    "parent-tracking": ("helper_m4_e2e_domain", "parent_tracking"),
    # Merge tests
    "merge-clean": ("helper_m4_e2e_merge", "merge_clean"),
    "merge-conflict": ("helper_m4_e2e_merge", "merge_conflict"),
    # CLI happy-path tests
    "plan-diff": ("helper_m4_e2e_cli", "plan_diff"),
    "cli-plan-use": ("helper_m4_e2e_cli", "cli_plan_use"),
    "cli-plan-execute": ("helper_m4_e2e_cli", "cli_plan_execute"),
    "cli-plan-tree": ("helper_m4_e2e_cli", "cli_plan_tree"),
    # CLI error-path tests
    "cli-plan-execute-readonly": (
        "helper_m4_e2e_cli_errors",
        "cli_plan_execute_readonly",
    ),
    "cli-plan-use-not-found": ("helper_m4_e2e_cli_errors", "cli_plan_use_not_found"),
    "cli-plan-diff-no-changeset": (
        "helper_m4_e2e_cli_errors",
        "cli_plan_diff_no_changeset",
    ),
    "cli-plan-tree-empty": ("helper_m4_e2e_cli_errors", "cli_plan_tree_empty"),
}


def _resolve_handler(command: str) -> Callable[[], None] | None:
    """Lazily import the handler for *command*.

    Only the module containing the requested command is loaded,
    avoiding unnecessary import overhead for unrelated modules.
    """
    entry = _COMMAND_MAP.get(command)
    if entry is None:
        return None
    module_name, func_name = entry
    module = importlib.import_module(module_name)
    handler: Callable[[], None] = getattr(module, func_name)
    return handler


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            f"Usage: helper_m4_e2e_verification.py <{'|'.join(_COMMAND_MAP)}>",
        )
        return 1
    command = sys.argv[1]
    handler = _resolve_handler(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1
    reset_global_state()
    handler()
    return 0


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