cleveragents-core/robot/helper_m4_e2e_cli.py

"""M4 E2E CLI integration tests (happy-path).

Tests the ``agents plan`` CLI subcommands (diff, use, execute, tree) by
invoking them via Typer's ``CliRunner`` with mocked services.

Error-path tests live in ``helper_m4_e2e_cli_errors`` to keep both
files under the 500-line CONTRIBUTING.md limit.

Usage (via dispatcher):
    python robot/helper_m4_e2e_verification.py plan-diff
    python robot/helper_m4_e2e_verification.py cli-plan-use
    python robot/helper_m4_e2e_verification.py cli-plan-execute
    python robot/helper_m4_e2e_verification.py cli-plan-tree
"""

from __future__ import annotations

import json
import re
import sys
from pathlib import Path
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.append(_SRC)

from helper_m4_e2e_common import (  # noqa: E402
    _CHILD_A_ULID,
    _CHILD_B_ULID,
    _ROOT_ULID,
    _assert_exit_code,
    _assert_mock_called_once,
    _assert_mock_called_once_with,
    _fail,
    _make_subplan_statuses,
    _mock_parent_plan,
)
from helpers_common import reset_global_state  # noqa: E402
from typer.testing import CliRunner  # noqa: E402

from cleveragents.cli.commands.plan import app as plan_app  # noqa: E402
from cleveragents.domain.models.core.decision import (  # noqa: E402
    Decision,
    DecisionType,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    NamespacedName,
    PlanPhase,
    ProcessingState,
    ProjectLink,
)

_runner = CliRunner()

# Decision ULIDs with descriptive names indicating their decision type.
_DECISION_PROMPT_DEF = "01KHDE6WWS2171PWW3GJEBXZ01"
_DECISION_STRATEGY = "01KHDE6WWS2171PWW3GJEBXZ02"
_DECISION_PARALLEL_SPAWN = "01KHDE6WWS2171PWW3GJEBXZ03"
_DECISION_SPAWN_API = "01KHDE6WWS2171PWW3GJEBXZ04"
_DECISION_SPAWN_UI = "01KHDE6WWS2171PWW3GJEBXZ05"


# ---------------------------------------------------------------------------
# plan-diff
# ---------------------------------------------------------------------------


def plan_diff() -> None:
    """Verify merged results via ``agents plan diff`` (mocked service)."""
    reset_global_state()
    mock_apply_svc = MagicMock()
    mock_apply_svc.diff.return_value = (
        "--- a/src/api.py\n"
        "+++ b/src/api.py\n"
        "@@ -1,3 +1,5 @@\n"
        " # API module\n"
        "+from fastapi import FastAPI\n"
        "+app = FastAPI()\n"
        " # existing code\n"
        "--- a/src/ui.py\n"
        "+++ b/src/ui.py\n"
        "@@ -1,2 +1,3 @@\n"
        " # UI module\n"
        "+import react\n"
    )

    with patch(
        "cleveragents.cli.commands.plan._get_apply_service",
        return_value=mock_apply_svc,
    ):
        result = _runner.invoke(plan_app, ["diff", _ROOT_ULID])
        _assert_exit_code(result, "plan diff")

        # Verify diff was called with the plan_id (kwargs + positional
        # fallback to avoid IndexError if called as keyword argument).
        _assert_mock_called_once(mock_apply_svc.diff, "diff")
        call_args = mock_apply_svc.diff.call_args
        plan_id_arg = call_args.kwargs.get("plan_id")
        if plan_id_arg is None and call_args.args:
            plan_id_arg = call_args.args[0]
        if plan_id_arg != _ROOT_ULID:
            _fail(f"diff called with wrong plan_id: {plan_id_arg}")

        # Verify the ``fmt`` kwarg was forwarded.  The production CLI
        # calls ``service.diff(plan_id, fmt=fmt)``; a regression that
        # dropped the ``fmt`` argument must be detected.  The default
        # value is ``"rich"`` (from the CLI ``--format`` option), so
        # both presence and value are asserted strictly.
        fmt_arg = call_args.kwargs.get("fmt")
        if fmt_arg is None and len(call_args.args) > 1:
            fmt_arg = call_args.args[1]
        if fmt_arg is None:
            _fail(
                "diff was not called with 'fmt' argument — "
                "expected fmt='rich' (default)"
            )
        if fmt_arg != "rich":
            _fail(f"diff called with unexpected fmt: {fmt_arg!r} — expected 'rich'")

        # Verify diff content is rendered in CLI output (M4 criterion:
        # "plan diff shows merged results").
        if "api.py" not in result.output:
            _fail(
                f"diff output should contain 'api.py' from merged results\n"
                f"output={result.output}"
            )
        if "ui.py" not in result.output:
            _fail(
                f"diff output should contain 'ui.py' from merged results\n"
                f"output={result.output}"
            )

    print("m4-plan-diff-ok")


# ---------------------------------------------------------------------------
# cli-plan-use
# ---------------------------------------------------------------------------


def cli_plan_use() -> None:
    """Verify ``agents plan use`` CLI creates a plan with subplan config.

    Mocks the lifecycle service and invokes the actual CLI command via
    Typer's CliRunner.  Asserts the service receives the correct action
    name and project, and the CLI exits cleanly.
    """
    reset_global_state()
    mock_service = MagicMock()

    # Mock action lookup -- use_action needs get_action_by_name first
    mock_action = MagicMock()
    mock_action.namespaced_name = NamespacedName.parse("local/refactor-action")
    mock_service.get_action_by_name.return_value = mock_action

    # Mock use_action to return a plan in Strategize phase
    plan = _mock_parent_plan(
        phase=PlanPhase.STRATEGIZE,
        state=ProcessingState.QUEUED,
    )
    mock_service.use_action.return_value = plan

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = _runner.invoke(
            plan_app,
            ["use", "local/refactor-action", "local/monorepo"],
        )
        _assert_exit_code(result, "plan use")

        # Verify the action was looked up
        _assert_mock_called_once_with(
            mock_service.get_action_by_name,
            "get_action_by_name",
            "local/refactor-action",
        )

        # Verify use_action was called with correct action and project
        _assert_mock_called_once(mock_service.use_action, "use_action")
        call_kwargs = mock_service.use_action.call_args

        # action_name: kwargs with positional fallback
        action_arg = call_kwargs.kwargs.get("action_name")
        if action_arg is None and call_kwargs.args:
            action_arg = call_kwargs.args[0]
        if action_arg != "local/refactor-action":
            _fail(f"use_action called with action={action_arg}")

        # project_links: kwargs with positional fallback (symmetric
        # with action_name to avoid asymmetric arg extraction).
        project_links = call_kwargs.kwargs.get("project_links")
        if project_links is None and len(call_kwargs.args) > 1:
            project_links = call_kwargs.args[1]
        if project_links is None:
            _fail("use_action not called with project_links argument")

        # Type guard: verify all items are ProjectLink instances
        if not all(isinstance(pl, ProjectLink) for pl in project_links):
            _fail(
                f"project_links should be list of ProjectLink, "
                f"got {[type(p).__name__ for p in project_links]}"
            )

        project_names = [pl.project_name for pl in project_links]
        if "local/monorepo" not in project_names:
            _fail(f"use_action project_links missing 'local/monorepo': {project_names}")

    print("m4-cli-plan-use-ok")


# ---------------------------------------------------------------------------
# cli-plan-execute
# ---------------------------------------------------------------------------


def cli_plan_execute() -> None:
    """Verify ``agents plan execute`` CLI transitions plan to Execute.

    Mocks the lifecycle service and invokes the actual CLI command.
    Asserts the plan transitions to Execute phase and the CLI renders
    the plan details.
    """
    reset_global_state()
    mock_service = MagicMock()

    # get_plan is called for the read-only guard
    pre_plan = _mock_parent_plan(
        phase=PlanPhase.STRATEGIZE,
        state=ProcessingState.COMPLETE,
    )
    mock_service.get_plan.return_value = pre_plan

    # execute_plan returns the plan in Execute phase with subplans
    statuses = _make_subplan_statuses()
    post_plan = _mock_parent_plan(
        phase=PlanPhase.EXECUTE,
        state=ProcessingState.PROCESSING,
        subplan_statuses=statuses,
    )
    mock_service.execute_plan.return_value = post_plan

    with (
        patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service",
            return_value=mock_service,
        ),
        patch(
            "cleveragents.cli.commands.plan._get_plan_executor",
            return_value=MagicMock(),
        ),
    ):
        result = _runner.invoke(plan_app, ["execute", _ROOT_ULID])
        _assert_exit_code(result, "plan execute")

        # Verify execute_plan was called with the plan ID
        _assert_mock_called_once_with(
            mock_service.execute_plan, "execute_plan", _ROOT_ULID
        )

        # Verify CLI output references the plan (rich rendering)
        if _ROOT_ULID not in result.output:
            _fail(
                f"CLI output should contain plan ID {_ROOT_ULID}\n"
                f"output={result.output}"
            )

        # Verify CLI output reflects the Execute phase transition.
        # Use word-boundary regex to match "execute" as a distinct word,
        # avoiding false positives from incidental substrings.
        if not re.search(r"\bexecute\b", result.output, re.IGNORECASE):
            _fail(f"CLI output should mention 'execute' phase\noutput={result.output}")

    # NOTE: Domain-level post_plan assertions (phase, subplan_statuses,
    # has_subplans) that were previously here have been removed.  They
    # verified values the test itself constructed via _mock_parent_plan,
    # constituting a tautology that could never fail.  The CLI does not
    # render subplan_count or has_subplans in its output, so these
    # properties cannot be meaningfully verified from CLI output alone.
    # Domain-level coverage for these properties is provided by the
    # spawn-subplans and parent-tracking domain tests.

    print("m4-cli-plan-execute-ok")


# ---------------------------------------------------------------------------
# cli-plan-tree
# ---------------------------------------------------------------------------


def cli_plan_tree() -> None:
    """Verify ``agents plan tree`` CLI displays subplan hierarchy.

    Creates real Decision objects forming a subplan tree:
      - prompt_definition (root)
        - strategy_choice
          - subplan_parallel_spawn
            - subplan_spawn (child A)
            - subplan_spawn (child B)

    Mocks the DI container and DecisionService, then invokes the
    CLI command with ``--format json`` and verifies the JSON output
    contains the expected decision types and hierarchy.

    NOTE: This test patches ``get_container`` at its definition site
    rather than ``_get_lifecycle_service`` because the production
    ``tree`` command resolves the ``DecisionService`` directly from
    the DI container (not via the lifecycle service helper).
    """
    reset_global_state()
    # Build a realistic decision tree with subplan decisions
    decisions = [
        Decision(
            decision_id=_DECISION_PROMPT_DEF,
            plan_id=_ROOT_ULID,
            parent_decision_id=None,
            sequence_number=0,
            decision_type=DecisionType.PROMPT_DEFINITION,
            question="What is the plan prompt?",
            chosen_option="Refactor code across monorepo modules",
            confidence_score=1.0,
        ),
        Decision(
            decision_id=_DECISION_STRATEGY,
            plan_id=_ROOT_ULID,
            parent_decision_id=_DECISION_PROMPT_DEF,
            sequence_number=1,
            decision_type=DecisionType.STRATEGY_CHOICE,
            question="How to decompose the refactoring?",
            chosen_option="Parallel subplans per module",
            confidence_score=0.85,
        ),
        Decision(
            decision_id=_DECISION_PARALLEL_SPAWN,
            plan_id=_ROOT_ULID,
            parent_decision_id=_DECISION_STRATEGY,
            sequence_number=2,
            decision_type=DecisionType.SUBPLAN_PARALLEL_SPAWN,
            question="Which subplans to run in parallel?",
            chosen_option="Spawn API and UI subplans concurrently",
            downstream_plan_ids=[_CHILD_A_ULID, _CHILD_B_ULID],
        ),
        Decision(
            decision_id=_DECISION_SPAWN_API,
            plan_id=_ROOT_ULID,
            parent_decision_id=_DECISION_PARALLEL_SPAWN,
            sequence_number=3,
            decision_type=DecisionType.SUBPLAN_SPAWN,
            question="Spawn subplan for API module?",
            chosen_option="Spawn API refactor subplan",
            downstream_plan_ids=[_CHILD_A_ULID],
        ),
        Decision(
            decision_id=_DECISION_SPAWN_UI,
            plan_id=_ROOT_ULID,
            parent_decision_id=_DECISION_PARALLEL_SPAWN,
            sequence_number=4,
            decision_type=DecisionType.SUBPLAN_SPAWN,
            question="Spawn subplan for UI module?",
            chosen_option="Spawn UI refactor subplan",
            downstream_plan_ids=[_CHILD_B_ULID],
        ),
    ]

    # Mock the DI container and DecisionService.
    # The production ``tree`` command calls ``container.decision_service()``
    # (not ``container.resolve()``), so the mock must match that API.
    mock_container = MagicMock()
    mock_decision_svc = MagicMock()
    mock_decision_svc.list_decisions.return_value = decisions
    mock_container.decision_service.return_value = mock_decision_svc

    with patch(
        "cleveragents.application.container.get_container",
        return_value=mock_container,
    ):
        result = _runner.invoke(
            plan_app,
            ["tree", _ROOT_ULID, "--format", "json"],
        )
        _assert_exit_code(result, "plan tree")

        # Verify the decision service was queried
        _assert_mock_called_once_with(
            mock_decision_svc.list_decisions,
            "list_decisions",
            _ROOT_ULID,
        )

        # Parse the JSON output and verify tree structure
        try:
            tree_data: list[dict[str, object]] = json.loads(result.output)
        except json.JSONDecodeError:
            _fail(f"plan tree output is not valid JSON:\n{result.output}")
            return  # unreachable; _fail raises SystemExit

        # Verify the tree is a list with exactly 1 root node
        if not isinstance(tree_data, list):
            _fail(f"tree output should be a list, got {type(tree_data).__name__}")
        if len(tree_data) != 1:
            _fail(f"tree should have 1 root, got {len(tree_data)}")

        # Root must be prompt_definition
        root = tree_data[0]
        if root.get("type") != "prompt_definition":
            _fail(f"root type should be prompt_definition, got {root.get('type')}")
        if root.get("decision_id") != _DECISION_PROMPT_DEF:
            _fail(f"root decision_id mismatch: {root.get('decision_id')}")

        # Root -> strategy_choice
        root_children = root.get("children", [])
        if not isinstance(root_children, list) or len(root_children) != 1:
            _fail(f"root should have 1 child, got {root_children!r}")
        strategy_node = root_children[0]
        if strategy_node.get("type") != "strategy_choice":
            _fail(
                f"root child should be strategy_choice, got {strategy_node.get('type')}"
            )

        # strategy_choice -> subplan_parallel_spawn
        strategy_children = strategy_node.get("children", [])
        if not isinstance(strategy_children, list) or len(strategy_children) != 1:
            _fail(f"strategy_choice should have 1 child, got {strategy_children!r}")
        parallel_node = strategy_children[0]
        if parallel_node.get("type") != "subplan_parallel_spawn":
            _fail(
                f"strategy child should be subplan_parallel_spawn, "
                f"got {parallel_node.get('type')}"
            )

        # subplan_parallel_spawn -> 2x subplan_spawn
        spawn_children = parallel_node.get("children", [])
        if not isinstance(spawn_children, list) or len(spawn_children) != 2:
            _fail(
                f"subplan_parallel_spawn should have 2 children, got {spawn_children!r}"
            )
        for sc in spawn_children:
            if sc.get("type") != "subplan_spawn":
                _fail(
                    f"parallel spawn child should be subplan_spawn, "
                    f"got {sc.get('type')}"
                )

        # Verify total node count: 5 decisions in the tree
        def _count_nodes(nodes: list[dict[str, object]]) -> int:
            total = 0
            for n in nodes:
                total += 1
                children = n.get("children")
                if isinstance(children, list):
                    total += _count_nodes(children)
            return total

        node_count = _count_nodes(tree_data)
        if node_count != 5:
            _fail(f"tree should have 5 nodes, got {node_count}")

    print("m4-cli-plan-tree-ok")