CHANGELOG.md

@@ -260,9 +260,13 @@
   tests and correction/subplan smoke tests pass against the final v3.3.0 implementation.
   Added CLI-exercising integration tests for `plan use`, `plan execute`, and `plan tree`
   commands to verify the milestone success criteria through actual Typer CLI invocations.
-  Fixed CONTRIBUTORS.md alphabetical ordering. Full nox quality gates pass: lint, format,
-  typecheck, unit tests, integration tests, coverage (97%), security scan, dead code
-  detection, docs build, wheel build, and ASV benchmarks. (#495)
+  Split 1074-line helper into six focused modules (`_common`, `_domain`, `_merge`, `_cli`,
+  `_cli_errors`, dispatcher) under the 500-line limit.  Added CLI error-path tests for
+  read-only plan execute, unavailable action, missing changeset, and empty decision tree.
+  Extracted `_make_subplan_status` factory, `_assert_exit_code` and
+  `_assert_mock_called_once*` wrappers, frozen timestamp constant, and `shutil.which`
+  git pre-check.  Removed tautological domain assertions in `plan_tree` and
+  `parallel_max`.  Fixed CONTRIBUTORS.md alphabetical ordering. (#495)
 - Added minimal LSP server stub with `agents lsp serve` CLI command supporting the
   `initialize`, `shutdown`, and `exit` lifecycle handshake over JSON-RPC stdin/stdout
   transport with Content-Length header framing. Unsupported methods return `MethodNotFound`

robot/helper_m4_e2e_cli.py

@@ -0,0 +1,453 @@
+"""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 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)."""
+    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.
+    """
+    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.
+    """
+    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,
+    ):
+        result = _runner.invoke(plan_app, ["execute", _ROOT_ULID])
+        _assert_exit_code(result, "plan execute")
+
+        # Verify get_plan was called for the read-only guard.
+        _assert_mock_called_once_with(mock_service.get_plan, "get_plan", _ROOT_ULID)
+
+        # 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).
+    """
+    # 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")

robot/helper_m4_e2e_cli_errors.py

@@ -0,0 +1,197 @@
+"""M4 E2E CLI error-path tests.
+
+Tests that the ``agents plan`` CLI subcommands handle error conditions
+gracefully: read-only plans, unavailable actions, missing changesets,
+and empty decision lists.
+
+Usage (via dispatcher):
+    python robot/helper_m4_e2e_verification.py cli-plan-execute-readonly
+    python robot/helper_m4_e2e_verification.py cli-plan-use-not-found
+    python robot/helper_m4_e2e_verification.py cli-plan-diff-no-changeset
+    python robot/helper_m4_e2e_verification.py cli-plan-tree-empty
+"""
+
+from __future__ import annotations
+
+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
+    _ROOT_ULID,
+    _assert_exit_code,
+    _fail,
+    _mock_parent_plan,
+)
+from typer.testing import CliRunner  # noqa: E402
+
+from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
+    ActionNotAvailableError,
+)
+from cleveragents.cli.commands.plan import app as plan_app  # noqa: E402
+from cleveragents.core.exceptions import PlanError  # noqa: E402
+from cleveragents.domain.models.core.action import ActionState  # noqa: E402
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    PlanPhase,
+    ProcessingState,
+)
+
+_runner = CliRunner()
+
+
+# ---------------------------------------------------------------------------
+# cli-plan-execute-readonly (error path)
+# ---------------------------------------------------------------------------
+
+
+def cli_plan_execute_readonly() -> None:
+    """Verify ``plan execute`` aborts on a read-only plan.
+
+    The ``execute`` command calls ``service.get_plan(plan_id)`` as a
+    fail-fast read-only guard.  When ``plan.read_only is True``, the
+    CLI should print an error and abort without calling ``execute_plan``.
+    """
+    mock_service = MagicMock()
+    plan = _mock_parent_plan(
+        phase=PlanPhase.STRATEGIZE,
+        state=ProcessingState.COMPLETE,
+        read_only=True,
+    )
+    mock_service.get_plan.return_value = plan
+
+    with patch(
+        "cleveragents.cli.commands.plan._get_lifecycle_service",
+        return_value=mock_service,
+    ):
+        result = _runner.invoke(plan_app, ["execute", _ROOT_ULID])
+        if result.exit_code == 0:
+            _fail(
+                f"plan execute should fail for read-only plan\noutput={result.output}"
+            )
+        output_lower = result.output.lower()
+        if "read-only" not in output_lower and "read_only" not in output_lower:
+            _fail(f"output should mention read-only\noutput={result.output}")
+
+        # execute_plan must not have been called
+        mock_service.execute_plan.assert_not_called()
+
+    print("m4-cli-plan-execute-readonly-ok")
+
+
+# ---------------------------------------------------------------------------
+# cli-plan-use-not-found (error path)
+# ---------------------------------------------------------------------------
+
+
+def cli_plan_use_not_found() -> None:
+    """Verify ``plan use`` aborts when the action is not available.
+
+    Raises ``ActionNotAvailableError`` from ``get_action_by_name``,
+    which the CLI catches and prints a user-friendly error message.
+    """
+    mock_service = MagicMock()
+    mock_service.get_action_by_name.side_effect = ActionNotAvailableError(
+        "nonexistent/action", ActionState.ARCHIVED
+    )
+
+    with patch(
+        "cleveragents.cli.commands.plan._get_lifecycle_service",
+        return_value=mock_service,
+    ):
+        result = _runner.invoke(
+            plan_app, ["use", "nonexistent/action", "local/monorepo"]
+        )
+        if result.exit_code == 0:
+            _fail(
+                f"plan use should fail for unavailable action\noutput={result.output}"
+            )
+        output_lower = result.output.lower()
+        # Assert command-specific message fragment to avoid accepting
+        # unrelated error paths.  The production CLI prints:
+        #   "[red]Action not available:[/red] {e}"
+        if "action not available" not in output_lower:
+            _fail(
+                f"output should contain 'Action not available' "
+                f"(got generic or wrong error)\noutput={result.output}"
+            )
+
+        # use_action must not have been called
+        mock_service.use_action.assert_not_called()
+
+    print("m4-cli-plan-use-not-found-ok")
+
+
+# ---------------------------------------------------------------------------
+# cli-plan-diff-no-changeset (error path)
+# ---------------------------------------------------------------------------
+
+
+def cli_plan_diff_no_changeset() -> None:
+    """Verify ``plan diff`` aborts when the plan has no changeset.
+
+    Raises ``PlanError`` from the apply service ``diff()`` method,
+    which the CLI catches and prints ``[red]Diff Error:[/red]``.
+    """
+    mock_apply_svc = MagicMock()
+    mock_apply_svc.diff.side_effect = PlanError(
+        f"Plan {_ROOT_ULID} has no ChangeSet. "
+        "Execute phase must complete before viewing diff."
+    )
+
+    with patch(
+        "cleveragents.cli.commands.plan._get_apply_service",
+        return_value=mock_apply_svc,
+    ):
+        result = _runner.invoke(plan_app, ["diff", _ROOT_ULID])
+        if result.exit_code == 0:
+            _fail(
+                f"plan diff should fail when no changeset exists\n"
+                f"output={result.output}"
+            )
+        output_lower = result.output.lower()
+        # Assert command-specific message fragment to avoid accepting
+        # unrelated error paths.  The production CLI prints:
+        #   "[red]Diff Error:[/red] {e.message}"
+        # where the PlanError message contains "has no ChangeSet".
+        if "diff error" not in output_lower and "has no changeset" not in output_lower:
+            _fail(
+                f"output should contain 'Diff Error' or 'has no ChangeSet' "
+                f"(got generic or wrong error)\noutput={result.output}"
+            )
+
+    print("m4-cli-plan-diff-no-changeset-ok")
+
+
+# ---------------------------------------------------------------------------
+# cli-plan-tree-empty (error path)
+# ---------------------------------------------------------------------------
+
+
+def cli_plan_tree_empty() -> None:
+    """Verify ``plan tree`` handles zero decisions gracefully.
+
+    When ``DecisionService.list_decisions()`` returns an empty list, the
+    CLI should print an informational message and exit 0 (not crash).
+    """
+    # 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 = []
+    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])
+        _assert_exit_code(result, "plan tree (empty)")
+        if "no decisions" not in result.output.lower():
+            _fail(f"output should mention 'No decisions'\noutput={result.output}")
+
+    print("m4-cli-plan-tree-empty-ok")

robot/helper_m4_e2e_common.py

@@ -0,0 +1,244 @@
+"""Shared constants, utilities, and mock factories for M4 E2E verification tests.
+
+Provides reusable test infrastructure for both domain-model tests
+(``helper_m4_e2e_domain``), merge tests (``helper_m4_e2e_merge``),
+and CLI integration tests (``helper_m4_e2e_cli``, ``helper_m4_e2e_cli_errors``).
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import TYPE_CHECKING, NoReturn
+from unittest.mock import MagicMock
+
+# Ensure the src directory is on the import path.  Uses ``append`` rather
+# than ``insert(0, ...)`` to avoid shadowing stdlib modules if a file
+# in ``src/`` were ever named identically to a stdlib module.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.append(_SRC)
+
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    ExecutionMode,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+    ProjectLink,
+    SubplanAttempt,
+    SubplanConfig,
+    SubplanMergeStrategy,
+    SubplanStatus,
+)
+
+if TYPE_CHECKING:
+    # Typer's CliRunner wraps Click's Result; importing via click.testing
+    # is the canonical path since typer.testing re-exports CliRunner but
+    # does not re-export Result.
+    from click.testing import Result as CliResult
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+_ROOT_ULID = "01KHDE6WWS2171PWW3GJEBXZ8R"
+_CHILD_A_ULID = "01KHDE6WWS2171PWW3GJEBXZ8A"
+_CHILD_B_ULID = "01KHDE6WWS2171PWW3GJEBXZ8B"
+_CHILD_C_ULID = "01KHDE6WWS2171PWW3GJEBXZ8C"
+
+# Frozen timestamp for deterministic fixtures.  Avoids non-deterministic
+# ``datetime.now()`` calls that produce subtly inconsistent timestamps
+# across fixture objects within a single test.
+_FROZEN_NOW = datetime(2026, 3, 1, 12, 0, 0)
+
+# ---------------------------------------------------------------------------
+# Shared test helpers
+# ---------------------------------------------------------------------------
+
+
+def _fail(msg: str) -> NoReturn:
+    """Print failure message to stderr and exit 1."""
+    print(f"FAIL: {msg}", file=sys.stderr)
+    raise SystemExit(1)
+
+
+def _assert_mock_called_once(mock: MagicMock, label: str) -> None:
+    """Wrap ``assert_called_once`` with standardised ``_fail()`` diagnostics.
+
+    Raw ``AssertionError`` from ``unittest.mock`` bypasses the ``FAIL:``
+    prefix pattern used by Robot Framework test evaluation; this wrapper
+    converts it.
+    """
+    try:
+        mock.assert_called_once()
+    except AssertionError as exc:
+        _fail(f"{label}: {exc}")
+
+
+def _assert_mock_called_once_with(
+    mock: MagicMock,
+    label: str,
+    *args: object,
+    **kwargs: object,
+) -> None:
+    """Wrap ``assert_called_once_with`` with ``_fail()`` diagnostics."""
+    try:
+        mock.assert_called_once_with(*args, **kwargs)
+    except AssertionError as exc:
+        _fail(f"{label}: {exc}")
+
+
+def _assert_exit_code(result: CliResult, label: str) -> None:
+    """Verify CLI exit code is 0, failing with full diagnostic output."""
+    if result.exit_code != 0:
+        _fail(
+            f"{label} rc={result.exit_code}\n"
+            f"output={result.output}\n"
+            f"exception={result.exception}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# SubplanStatus factory (DRY extraction)
+# ---------------------------------------------------------------------------
+
+
+def _make_subplan_status(
+    subplan_id: str,
+    *,
+    action_name: str = "local/m4-verify-action",
+    target_resources: list[str] | None = None,
+    status: ProcessingState = ProcessingState.QUEUED,
+    started_at: datetime | None = None,
+    completed_at: datetime | None = None,
+    changeset_summary: str | None = None,
+    files_changed: int = 0,
+    error: str | None = None,
+    attempt_number: int = 1,
+    previous_attempts: list[SubplanAttempt] | None = None,
+) -> SubplanStatus:
+    """Create a ``SubplanStatus`` with sensible test defaults."""
+    return SubplanStatus(
+        subplan_id=subplan_id,
+        action_name=action_name,
+        target_resources=target_resources or [],
+        status=status,
+        started_at=started_at,
+        completed_at=completed_at,
+        changeset_summary=changeset_summary,
+        files_changed=files_changed,
+        error=error,
+        attempt_number=attempt_number,
+        previous_attempts=previous_attempts or [],
+    )
+
+
+def _make_subplan_statuses() -> list[SubplanStatus]:
+    """Create three SubplanStatus entries (A=COMPLETE, B=COMPLETE, C=QUEUED).
+
+    This is the default three-child status set used by many M4 tests.
+    """
+    return [
+        _make_subplan_status(
+            _CHILD_A_ULID,
+            target_resources=["res-api"],
+            status=ProcessingState.COMPLETE,
+            started_at=_FROZEN_NOW,
+            completed_at=_FROZEN_NOW,
+            changeset_summary="Updated API endpoints",
+            files_changed=3,
+        ),
+        _make_subplan_status(
+            _CHILD_B_ULID,
+            target_resources=["res-ui"],
+            status=ProcessingState.COMPLETE,
+            started_at=_FROZEN_NOW,
+            completed_at=_FROZEN_NOW,
+            changeset_summary="Updated UI components",
+            files_changed=5,
+        ),
+        _make_subplan_status(
+            _CHILD_C_ULID,
+            target_resources=["res-db"],
+        ),
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Plan mock factories
+# ---------------------------------------------------------------------------
+
+
+def _mock_parent_plan(
+    phase: PlanPhase = PlanPhase.EXECUTE,
+    state: ProcessingState = ProcessingState.PROCESSING,
+    subplan_statuses: list[SubplanStatus] | None = None,
+    subplan_config: SubplanConfig | None = None,
+    *,
+    read_only: bool = False,
+) -> Plan:
+    """Create a parent plan with subplan config."""
+    return Plan(
+        identity=PlanIdentity(plan_id=_ROOT_ULID),
+        namespaced_name=NamespacedName.parse("local/m4-parent-plan"),
+        description="M4 verification parent plan with subplans",
+        definition_of_done="All subplans complete successfully",
+        action_name="local/m4-verify-action",
+        phase=phase,
+        processing_state=state,
+        project_links=[ProjectLink(project_name="local/m4-project")],
+        arguments={},
+        arguments_order=[],
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        timestamps=PlanTimestamps(created_at=_FROZEN_NOW, updated_at=_FROZEN_NOW),
+        created_by=None,
+        reusable=True,
+        read_only=read_only,
+        subplan_config=subplan_config
+        or SubplanConfig(
+            execution_mode=ExecutionMode.PARALLEL,
+            merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY,
+            max_parallel=3,
+            fail_fast=False,
+            retry_failed=True,
+            max_retries=2,
+        ),
+        subplan_statuses=subplan_statuses or [],
+    )
+
+
+def _mock_child_plan(
+    child_id: str,
+    parent_id: str = _ROOT_ULID,
+    root_id: str = _ROOT_ULID,
+    phase: PlanPhase = PlanPhase.EXECUTE,
+    state: ProcessingState = ProcessingState.QUEUED,
+) -> Plan:
+    """Create a child subplan."""
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=child_id,
+            parent_plan_id=parent_id,
+            root_plan_id=root_id,
+        ),
+        namespaced_name=NamespacedName.parse(f"local/m4-child-{child_id[-1].lower()}"),
+        description=f"M4 child subplan {child_id[-1]}",
+        definition_of_done="Subplan criteria pass",
+        action_name="local/m4-verify-action",
+        phase=phase,
+        processing_state=state,
+        project_links=[ProjectLink(project_name="local/m4-project")],
+        arguments={},
+        arguments_order=[],
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        timestamps=PlanTimestamps(created_at=_FROZEN_NOW, updated_at=_FROZEN_NOW),
+        created_by=None,
+        reusable=True,
+        read_only=False,
+    )

robot/helper_m4_e2e_domain.py

@@ -0,0 +1,385 @@
+"""M4 E2E domain-model verification tests.
+
+Tests subplan spawning, plan tree field completeness, parallel execution
+constraints, and parent plan subplan status tracking.
+
+Tree *construction* logic is exercised by the CLI-level ``cli_plan_tree()``
+test (which invokes ``build_decision_tree()``); this module validates the
+``SubplanStatus`` and ``PlanIdentity`` data models that feed into that tree.
+
+Merge tests (``merge-clean``, ``merge-conflict``) live in the separate
+``helper_m4_e2e_merge`` module to keep this file well under the 500-line
+limit.
+
+Usage (via dispatcher):
+    python robot/helper_m4_e2e_verification.py spawn-subplans
+    python robot/helper_m4_e2e_verification.py plan-tree
+    python robot/helper_m4_e2e_verification.py parallel-max
+    python robot/helper_m4_e2e_verification.py parent-tracking
+"""
+
+from __future__ import annotations
+
+import sys
+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 helper_m4_e2e_common import (  # noqa: E402
+    _CHILD_A_ULID,
+    _CHILD_B_ULID,
+    _CHILD_C_ULID,
+    _FROZEN_NOW,
+    _ROOT_ULID,
+    _fail,
+    _make_subplan_status,
+    _make_subplan_statuses,
+    _mock_child_plan,
+    _mock_parent_plan,
+)
+
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    ExecutionMode,
+    ProcessingState,
+    SubplanAttempt,
+    SubplanConfig,
+    SubplanFailureHandler,
+)
+
+# ---------------------------------------------------------------------------
+# spawn-subplans
+# ---------------------------------------------------------------------------
+
+
+def spawn_subplans() -> None:
+    """Verify a parent plan can spawn multiple subplans during Execute."""
+    statuses = _make_subplan_statuses()
+    parent = _mock_parent_plan(subplan_statuses=statuses)
+
+    # Verify parent has subplan config
+    if parent.subplan_config is None:
+        _fail("parent plan missing subplan_config")
+    if parent.subplan_config.execution_mode != ExecutionMode.PARALLEL:
+        _fail(f"expected PARALLEL mode, got {parent.subplan_config.execution_mode}")
+
+    # Verify subplans are tracked
+    if not parent.has_subplans:
+        _fail("parent should have subplans")
+    if len(parent.subplan_statuses) != 3:
+        _fail(f"expected 3 subplans, got {len(parent.subplan_statuses)}")
+
+    # Create child plans and verify identity hierarchy
+    child_a = _mock_child_plan(_CHILD_A_ULID)
+    child_b = _mock_child_plan(_CHILD_B_ULID)
+    child_c = _mock_child_plan(_CHILD_C_ULID)
+
+    for child in [child_a, child_b, child_c]:
+        if not child.is_subplan:
+            _fail(f"child {child.identity.plan_id} should be a subplan")
+        if child.identity.parent_plan_id != _ROOT_ULID:
+            _fail(f"child parent_plan_id mismatch: {child.identity.parent_plan_id}")
+        if child.identity.root_plan_id != _ROOT_ULID:
+            _fail(f"child root_plan_id mismatch: {child.identity.root_plan_id}")
+
+    # Verify parent is root
+    if not parent.is_root_plan:
+        _fail("parent should be root plan")
+    if parent.depth != 0:
+        _fail(f"parent depth should be 0, got {parent.depth}")
+
+    # Verify child depth placeholder
+    if child_a.depth != -1:
+        _fail(f"child depth should be -1 (placeholder), got {child_a.depth}")
+
+    # Verify CLI dict includes subplan_count
+    cli_dict = parent.as_cli_dict()
+    actual = cli_dict.get("subplan_count")
+    if actual != 3:
+        _fail(f"as_cli_dict subplan_count should be 3, got {actual}")
+
+    print("m4-spawn-subplans-ok")
+
+
+# ---------------------------------------------------------------------------
+# plan-tree  (SubplanStatus field completeness + parent linkage)
+# ---------------------------------------------------------------------------
+
+
+def plan_tree() -> None:
+    """Verify SubplanStatus field completeness and parent→child linkage.
+
+    This test validates that the ``SubplanStatus`` data model used by
+    ``plan tree`` has all expected fields populated (``subplan_id``,
+    ``action_name``, ``changeset_summary``, ``files_changed``) and that
+    child ``PlanIdentity`` objects correctly reference the parent.
+
+    Tree *construction* logic (``build_decision_tree()``) is exercised
+    by the CLI-level ``cli_plan_tree()`` test, which verifies the full
+    structural hierarchy via JSON output.
+    """
+    statuses = _make_subplan_statuses()
+    parent = _mock_parent_plan(subplan_statuses=statuses)
+
+    child_a = _mock_child_plan(_CHILD_A_ULID)
+    child_b = _mock_child_plan(_CHILD_B_ULID)
+    child_c = _mock_child_plan(_CHILD_C_ULID)
+
+    # Verify the parent reports the correct number of subplans.
+    if len(parent.subplan_statuses) != 3:
+        _fail(f"expected 3 subplan statuses, got {len(parent.subplan_statuses)}")
+
+    # Verify each subplan status has required fields populated.
+    for status in parent.subplan_statuses:
+        if not status.subplan_id:
+            _fail("subplan status missing subplan_id")
+        if not status.action_name:
+            _fail("subplan status missing action_name")
+
+    # Verify completed subplans have changeset summaries.
+    completed = [
+        s for s in parent.subplan_statuses if s.status == ProcessingState.COMPLETE
+    ]
+    if len(completed) != 2:
+        _fail(f"expected 2 completed subplans, got {len(completed)}")
+    for s in completed:
+        if not s.changeset_summary:
+            _fail(f"completed subplan {s.subplan_id} missing changeset_summary")
+        if s.files_changed <= 0:
+            _fail(f"completed subplan {s.subplan_id} has no files_changed")
+
+    # Verify queued subplan has no changeset.
+    queued = [s for s in parent.subplan_statuses if s.status == ProcessingState.QUEUED]
+    if len(queued) != 1:
+        _fail(f"expected 1 queued subplan, got {len(queued)}")
+    if queued[0].changeset_summary is not None:
+        _fail("queued subplan should not have changeset_summary")
+
+    # Verify parent → child reverse link via PlanIdentity.
+    for child in [child_a, child_b, child_c]:
+        if child.identity.parent_plan_id != parent.identity.plan_id:
+            _fail(f"child {child.identity.plan_id} parent mismatch")
+
+    print("m4-plan-tree-ok")
+
+
+# ---------------------------------------------------------------------------
+# parallel-max
+# ---------------------------------------------------------------------------
+
+
+def parallel_max() -> None:
+    """Verify parallel subplan execution respects max_parallel."""
+    config = SubplanConfig(
+        execution_mode=ExecutionMode.PARALLEL,
+        max_parallel=3,
+        fail_fast=False,
+        retry_failed=True,
+        max_retries=2,
+    )
+
+    # Verify max_parallel constraints (1 <= max_parallel <= 50)
+    try:
+        SubplanConfig(execution_mode=ExecutionMode.PARALLEL, max_parallel=0)
+        _fail("max_parallel=0 should be rejected")
+    except ValueError:
+        pass  # Expected: ge=1 constraint
+
+    try:
+        SubplanConfig(execution_mode=ExecutionMode.PARALLEL, max_parallel=51)
+        _fail("max_parallel=51 should be rejected")
+    except ValueError:
+        pass  # Expected: le=50 constraint
+
+    # Verify a parent plan with parallel config and multiple subplans
+    statuses = [
+        _make_subplan_status(
+            _CHILD_A_ULID,
+            target_resources=["res-1"],
+            status=ProcessingState.PROCESSING,
+            started_at=_FROZEN_NOW,
+        ),
+        _make_subplan_status(
+            _CHILD_B_ULID,
+            target_resources=["res-2"],
+            status=ProcessingState.PROCESSING,
+            started_at=_FROZEN_NOW,
+        ),
+        _make_subplan_status(
+            _CHILD_C_ULID,
+            target_resources=["res-3"],
+        ),
+    ]
+
+    parent = _mock_parent_plan(subplan_config=config, subplan_statuses=statuses)
+
+    # Count running subplans -- should not exceed max_parallel
+    running = [
+        s for s in parent.subplan_statuses if s.status == ProcessingState.PROCESSING
+    ]
+    if len(running) > config.max_parallel:
+        _fail(
+            f"running subplans ({len(running)}) exceeds "
+            f"max_parallel ({config.max_parallel})"
+        )
+
+    # Verify all execution modes are valid
+    for mode in ExecutionMode:
+        cfg = SubplanConfig(execution_mode=mode, max_parallel=5)
+        if cfg.execution_mode != mode:
+            _fail(f"execution_mode mismatch: {cfg.execution_mode} != {mode}")
+
+    # Verify SubplanFailureHandler with parallel config
+    handler = SubplanFailureHandler()
+    failed_status = _make_subplan_status(
+        _CHILD_A_ULID,
+        status=ProcessingState.ERRORED,
+        error="ValidationError: tests failed",
+    )
+
+    # fail_fast=False and parallel => should NOT stop others
+    if handler.should_stop_others(config, failed_status):
+        _fail("parallel + fail_fast=False should not stop others")
+
+    # fail_fast=True => should stop others
+    ff_config = SubplanConfig(
+        execution_mode=ExecutionMode.PARALLEL,
+        max_parallel=3,
+        fail_fast=True,
+    )
+    if not handler.should_stop_others(ff_config, failed_status):
+        _fail("fail_fast=True should stop others")
+
+    # Retry check: ValidationError is retriable
+    if not handler.should_retry(config, failed_status):
+        _fail("ValidationError should be retriable")
+
+    print("m4-parallel-max-ok")
+
+
+# ---------------------------------------------------------------------------
+# parent-tracking
+# ---------------------------------------------------------------------------
+
+
+def parent_tracking() -> None:
+    """Verify parent plan tracks all subplan statuses correctly."""
+    statuses = [
+        _make_subplan_status(
+            _CHILD_A_ULID,
+            target_resources=["res-api"],
+            status=ProcessingState.COMPLETE,
+            started_at=_FROZEN_NOW,
+            completed_at=_FROZEN_NOW,
+            changeset_summary="Updated API",
+            files_changed=3,
+        ),
+        _make_subplan_status(
+            _CHILD_B_ULID,
+            target_resources=["res-ui"],
+            status=ProcessingState.ERRORED,
+            started_at=_FROZEN_NOW,
+            completed_at=_FROZEN_NOW,
+            error="TimeoutError: execution timed out",
+        ),
+        _make_subplan_status(
+            _CHILD_C_ULID,
+            target_resources=["res-db"],
+            status=ProcessingState.PROCESSING,
+            started_at=_FROZEN_NOW,
+            attempt_number=2,
+            previous_attempts=[
+                SubplanAttempt(
+                    attempt_number=1,
+                    started_at=_FROZEN_NOW,
+                    completed_at=_FROZEN_NOW,
+                    error="ValidationError: schema mismatch",
+                    was_retried=True,
+                ),
+            ],
+        ),
+    ]
+
+    parent = _mock_parent_plan(subplan_statuses=statuses)
+
+    # Verify parent tracks all three statuses
+    if len(parent.subplan_statuses) != 3:
+        _fail(f"expected 3 statuses, got {len(parent.subplan_statuses)}")
+
+    # Verify completed subplan
+    completed = [
+        s for s in parent.subplan_statuses if s.status == ProcessingState.COMPLETE
+    ]
+    if len(completed) != 1:
+        _fail(f"expected 1 completed, got {len(completed)}")
+    if completed[0].files_changed != 3:
+        _fail(f"completed files_changed should be 3, got {completed[0].files_changed}")
+
+    # Verify errored subplan
+    errored = [
+        s for s in parent.subplan_statuses if s.status == ProcessingState.ERRORED
+    ]
+    if len(errored) != 1:
+        _fail(f"expected 1 errored, got {len(errored)}")
+    if errored[0].error is None:
+        _fail("errored subplan should have error message")
+    if "TimeoutError" not in (errored[0].error or ""):
+        _fail(f"error should contain TimeoutError, got {errored[0].error}")
+
+    # Verify retry tracking on third subplan
+    retried = [
+        s for s in parent.subplan_statuses if s.status == ProcessingState.PROCESSING
+    ]
+    if len(retried) != 1:
+        _fail(f"expected 1 processing, got {len(retried)}")
+    if retried[0].attempt_number != 2:
+        _fail(
+            f"retried subplan should be on attempt 2, got {retried[0].attempt_number}"
+        )
+    if len(retried[0].previous_attempts) != 1:
+        _fail(f"expected 1 previous attempt, got {len(retried[0].previous_attempts)}")
+    prev = retried[0].previous_attempts[0]
+    if not prev.was_retried:
+        _fail("previous attempt should have was_retried=True")
+    if prev.error is None:
+        _fail("previous attempt should have error message")
+
+    # Verify SubplanFailureHandler retry logic
+    handler = SubplanFailureHandler()
+    config = parent.subplan_config
+    if config is None:
+        _fail("parent should have subplan_config")
+
+    # TimeoutError is retriable
+    if not handler.should_retry(config, errored[0]):
+        _fail("TimeoutError should be retriable")
+
+    # Non-retriable error should not retry
+    non_retriable_status = _make_subplan_status(
+        _CHILD_A_ULID,
+        status=ProcessingState.ERRORED,
+        error="ConfigurationError: invalid config",
+    )
+    if handler.should_retry(config, non_retriable_status):
+        _fail("ConfigurationError should NOT be retriable")
+
+    # Exhausted retries should not retry
+    exhausted_status = _make_subplan_status(
+        _CHILD_A_ULID,
+        status=ProcessingState.ERRORED,
+        error="ValidationError: tests failed",
+        attempt_number=3,
+    )
+    if handler.should_retry(config, exhausted_status):
+        _fail("attempt_number > max_retries should not retry")
+
+    # Verify as_cli_dict on parent includes subplan tracking
+    cli_dict = parent.as_cli_dict()
+    if "subplan_count" not in cli_dict:
+        _fail("as_cli_dict should include subplan_count")
+    if cli_dict["subplan_count"] != 3:
+        _fail(f"subplan_count should be 3, got {cli_dict['subplan_count']}")
+
+    print("m4-parent-tracking-ok")

robot/helper_m4_e2e_merge.py

@@ -0,0 +1,146 @@
+"""M4 E2E merge verification tests.
+
+Tests three-way merge for non-conflicting changes and conflict surfacing.
+Extracted from ``helper_m4_e2e_domain`` to keep that module well under
+the 500-line limit (CONTRIBUTING.md compliance).
+
+Usage (via dispatcher):
+    python robot/helper_m4_e2e_verification.py merge-clean
+    python robot/helper_m4_e2e_verification.py merge-conflict
+"""
+
+from __future__ import annotations
+
+import shutil
+import sys
+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 helper_m4_e2e_common import _fail  # noqa: E402
+
+from cleveragents.domain.models.core.plan import SubplanMergeStrategy  # noqa: E402
+from cleveragents.infrastructure.sandbox.merge import (  # noqa: E402
+    GitMergeStrategy,
+    MergeResult,
+    SequentialMergeStrategy,
+)
+
+
+def _assert_git_available() -> None:
+    """Pre-check that ``git`` is available on ``$PATH``.
+
+    ``GitMergeStrategy.merge()`` shells out to ``git merge-file``.  If
+    ``git`` is absent, the fallback to ``SequentialMergeStrategy``
+    produces results that fail the merge assertions with misleading
+    diagnostics (e.g. "missing ours change" instead of "git not found").
+    """
+    if shutil.which("git") is None:
+        _fail("git is required on PATH for merge tests")
+
+
+# ---------------------------------------------------------------------------
+# merge-clean
+# ---------------------------------------------------------------------------
+
+
+def merge_clean() -> None:
+    """Verify three-way merge combines non-conflicting changes."""
+    _assert_git_available()
+
+    strategy = GitMergeStrategy()
+
+    base = "line1\nline2\nline3\nline4\nline5\n"
+    ours = "line1-modified-by-subplan-A\nline2\nline3\nline4\nline5\n"
+    theirs = "line1\nline2\nline3\nline4\nline5-modified-by-subplan-B\n"
+
+    result = strategy.merge(base, ours, theirs)
+
+    if not result.success:
+        _fail(f"clean merge should succeed, got conflicts: {result.has_conflicts}")
+    if result.has_conflicts:
+        _fail("clean merge should have no conflicts")
+    if "line1-modified-by-subplan-A" not in result.content:
+        _fail("merged content missing ours change")
+    if "line5-modified-by-subplan-B" not in result.content:
+        _fail("merged content missing theirs change")
+    if result.conflict_markers:
+        _fail(f"should have no conflict markers, got {result.conflict_markers}")
+
+    # Also verify SequentialMergeStrategy (last-wins)
+    seq_strategy = SequentialMergeStrategy()
+    seq_result = seq_strategy.merge(base, ours, theirs)
+    if not seq_result.success:
+        _fail("sequential merge should always succeed")
+    if seq_result.content != theirs:
+        _fail("sequential merge should return theirs")
+
+    # Verify merge result dataclass fields
+    clean_result = MergeResult(success=True, content="merged", has_conflicts=False)
+    if clean_result.success is not True:
+        _fail("MergeResult success field incorrect")
+    if clean_result.has_conflicts is not False:
+        _fail("MergeResult has_conflicts field incorrect")
+
+    print("m4-merge-clean-ok")
+
+
+# ---------------------------------------------------------------------------
+# merge-conflict
+# ---------------------------------------------------------------------------
+
+
+def merge_conflict() -> None:
+    """Verify merge conflicts are surfaced correctly."""
+    _assert_git_available()
+
+    strategy = GitMergeStrategy()
+
+    base = "line1\nline2\nline3\n"
+    ours = "line1\nline2-ours\nline3\n"
+    theirs = "line1\nline2-theirs\nline3\n"
+
+    result = strategy.merge(base, ours, theirs)
+
+    if result.success:
+        _fail("conflicting merge should not succeed")
+    if not result.has_conflicts:
+        _fail("conflicting merge should have conflicts")
+    if not result.conflict_markers:
+        _fail("conflict markers should be present")
+
+    # Verify conflict markers contain standard git markers
+    if "<<<<<<<" not in result.content:
+        _fail("merged content should contain <<<<<<< marker")
+    if ">>>>>>>" not in result.content:
+        _fail("merged content should contain >>>>>>> marker")
+    if "=======" not in result.content:
+        _fail("merged content should contain ======= marker")
+
+    # Verify conflict_markers list has at least one region
+    for start, end in result.conflict_markers:
+        if start < 1:
+            _fail(f"conflict marker start should be >= 1, got {start}")
+        if end < start:
+            _fail(f"conflict marker end ({end}) should be >= start ({start})")
+
+    # Verify SubplanMergeStrategy enum values
+    if SubplanMergeStrategy.FAIL_ON_CONFLICT != "fail_on_conflict":
+        _fail("FAIL_ON_CONFLICT value mismatch")
+
+    expected_strategies = {
+        "git_three_way",
+        "sequential_apply",
+        "fail_on_conflict",
+        "last_wins",
+    }
+    actual_strategies = {s.value for s in SubplanMergeStrategy}
+    if actual_strategies != expected_strategies:
+        _fail(
+            f"merge strategies mismatch: {actual_strategies} != {expected_strategies}"
+        )
+
+    print("m4-merge-conflict-ok")

robot/helper_m4_e2e_verification.py

@@ -11,6 +11,10 @@ Exercises the M4 success criteria for subplans and parallel execution:
   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.
@@ -25,831 +29,74 @@ Usage:
 
 from __future__ import annotations
 
-import os
-import re
+import importlib
 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)
+    sys.path.append(_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,
-    run_cli,
-    setup_workspace,
-    write_yaml,
-)
 from helpers_common import reset_global_state  # noqa: E402
 
-from cleveragents.application.services.decision_service import (  # noqa: E402
-    DecisionService,
-)
-from cleveragents.config.settings import Settings  # noqa: E402
-from cleveragents.domain.models.core.decision import (  # noqa: E402
-    ContextSnapshot,
-    DecisionType,
-    ResourceRef,
-)
-from cleveragents.domain.models.core.plan import (  # noqa: E402
-    ExecutionMode,
-    NamespacedName,
-    Plan,
-    PlanIdentity,
-    PlanPhase,
-    PlanTimestamps,
-    ProcessingState,
-    ProjectLink,
-    SubplanAttempt,
-    SubplanConfig,
-    SubplanFailureHandler,
-    SubplanMergeStrategy,
-    SubplanStatus,
-)
-from cleveragents.infrastructure.database.unit_of_work import UnitOfWork  # noqa: E402
-from cleveragents.infrastructure.sandbox.merge import (  # noqa: E402
-    GitMergeStrategy,
-    SequentialMergeStrategy,
-)
-
-_ROOT_ULID = "01KHDE6WWS2171PWW3GJEBXZ8R"
-_CHILD_A_ULID = "01KHDE6WWS2171PWW3GJEBXZ8A"
-_CHILD_B_ULID = "01KHDE6WWS2171PWW3GJEBXZ8B"
-_CHILD_C_ULID = "01KHDE6WWS2171PWW3GJEBXZ8C"
-
-_ACTION_YAML = """\
-name: local/m4-verify-action
-description: M4 verification action
-definition_of_done: All M4 criteria pass
-strategy_actor: openai/gpt-4
-execution_actor: openai/gpt-4
-"""
-
-
-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
-
-
-def _mock_parent_plan(
-    phase: PlanPhase = PlanPhase.EXECUTE,
-    state: ProcessingState = ProcessingState.PROCESSING,
-    subplan_statuses: list[SubplanStatus] | None = None,
-    subplan_config: SubplanConfig | None = None,
-) -> Plan:
-    """Create a parent plan with subplan config."""
-    now = datetime.now()
-    return Plan(
-        identity=PlanIdentity(plan_id=_ROOT_ULID),
-        namespaced_name=NamespacedName.parse("local/m4-parent-plan"),
-        description="M4 verification parent plan with subplans",
-        definition_of_done="All subplans complete successfully",
-        action_name="local/m4-verify-action",
-        phase=phase,
-        processing_state=state,
-        project_links=[ProjectLink(project_name="local/m4-project")],
-        arguments={},
-        arguments_order=[],
-        strategy_actor="openai/gpt-4",
-        execution_actor="openai/gpt-4",
-        timestamps=PlanTimestamps(created_at=now, updated_at=now),
-        created_by=None,
-        reusable=True,
-        read_only=False,
-        subplan_config=subplan_config
-        or SubplanConfig(
-            execution_mode=ExecutionMode.PARALLEL,
-            merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY,
-            max_parallel=3,
-            fail_fast=False,
-            retry_failed=True,
-            max_retries=2,
-        ),
-        subplan_statuses=subplan_statuses or [],
-    )
-
-
-def _mock_child_plan(
-    child_id: str,
-    parent_id: str = _ROOT_ULID,
-    root_id: str = _ROOT_ULID,
-    phase: PlanPhase = PlanPhase.EXECUTE,
-    state: ProcessingState = ProcessingState.QUEUED,
-) -> Plan:
-    """Create a child subplan."""
-    now = datetime.now()
-    return Plan(
-        identity=PlanIdentity(
-            plan_id=child_id,
-            parent_plan_id=parent_id,
-            root_plan_id=root_id,
-        ),
-        namespaced_name=NamespacedName.parse(f"local/m4-child-{child_id[-1].lower()}"),
-        description=f"M4 child subplan {child_id[-1]}",
-        definition_of_done="Subplan criteria pass",
-        action_name="local/m4-verify-action",
-        phase=phase,
-        processing_state=state,
-        project_links=[ProjectLink(project_name="local/m4-project")],
-        arguments={},
-        arguments_order=[],
-        strategy_actor="openai/gpt-4",
-        execution_actor="openai/gpt-4",
-        timestamps=PlanTimestamps(created_at=now, updated_at=now),
-        created_by=None,
-        reusable=True,
-        read_only=False,
-    )
-
-
-def _make_subplan_statuses() -> list[SubplanStatus]:
-    """Create SubplanStatus entries for three child plans."""
-    now = datetime.now()
-    return [
-        SubplanStatus(
-            subplan_id=_CHILD_A_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-api"],
-            status=ProcessingState.COMPLETE,
-            started_at=now,
-            completed_at=now,
-            changeset_summary="Updated API endpoints",
-            files_changed=3,
-        ),
-        SubplanStatus(
-            subplan_id=_CHILD_B_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-ui"],
-            status=ProcessingState.COMPLETE,
-            started_at=now,
-            completed_at=now,
-            changeset_summary="Updated UI components",
-            files_changed=5,
-        ),
-        SubplanStatus(
-            subplan_id=_CHILD_C_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-db"],
-            status=ProcessingState.QUEUED,
-            started_at=None,
-            completed_at=None,
-        ),
-    ]
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: spawn-subplans (domain-level, no CLI)
-# ---------------------------------------------------------------------------
-
-
-def spawn_subplans() -> None:
-    """Verify a parent plan can spawn multiple subplans during Execute."""
-    statuses = _make_subplan_statuses()
-    parent = _mock_parent_plan(subplan_statuses=statuses)
-
-    if parent.subplan_config is None:
-        _fail("parent plan missing subplan_config")
-    if parent.subplan_config.execution_mode != ExecutionMode.PARALLEL:
-        _fail(f"expected PARALLEL mode, got {parent.subplan_config.execution_mode}")
-    if not parent.has_subplans:
-        _fail("parent should have subplans")
-    if len(parent.subplan_statuses) != 3:
-        _fail(f"expected 3 subplans, got {len(parent.subplan_statuses)}")
-
-    child_a = _mock_child_plan(_CHILD_A_ULID)
-    child_b = _mock_child_plan(_CHILD_B_ULID)
-    child_c = _mock_child_plan(_CHILD_C_ULID)
-
-    for child in [child_a, child_b, child_c]:
-        if not child.is_subplan:
-            _fail(f"child {child.identity.plan_id} should be a subplan")
-        if child.identity.parent_plan_id != _ROOT_ULID:
-            _fail(f"child parent_plan_id mismatch: {child.identity.parent_plan_id}")
-        if child.identity.root_plan_id != _ROOT_ULID:
-            _fail(f"child root_plan_id mismatch: {child.identity.root_plan_id}")
-
-    if not parent.is_root_plan:
-        _fail("parent should be root plan")
-    if parent.depth != 0:
-        _fail(f"parent depth should be 0, got {parent.depth}")
-    if child_a.depth != -1:
-        _fail(f"child depth should be -1 (placeholder), got {child_a.depth}")
-
-    cli_dict = parent.as_cli_dict()
-    actual = cli_dict.get("subplan_count")
-    if actual != 3:
-        _fail(f"as_cli_dict subplan_count should be 3, got {actual}")
-
-    print("m4-spawn-subplans-ok")
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: plan-tree (domain-level, no CLI)
-# ---------------------------------------------------------------------------
-
-
-def plan_tree() -> None:
-    """Verify subplan tree structure via domain model hierarchy."""
-    statuses = _make_subplan_statuses()
-    parent = _mock_parent_plan(subplan_statuses=statuses)
-
-    child_a = _mock_child_plan(_CHILD_A_ULID)
-    child_b = _mock_child_plan(_CHILD_B_ULID)
-    child_c = _mock_child_plan(_CHILD_C_ULID)
-
-    tree: dict[str, list[str]] = {
-        parent.identity.plan_id: [s.subplan_id for s in parent.subplan_statuses]
-    }
-
-    if _ROOT_ULID not in tree:
-        _fail("root plan not in tree")
-    children_in_tree = tree[_ROOT_ULID]
-    if len(children_in_tree) != 3:
-        _fail(f"expected 3 children in tree, got {len(children_in_tree)}")
-
-    for status in parent.subplan_statuses:
-        if not status.subplan_id:
-            _fail("subplan status missing subplan_id")
-        if not status.action_name:
-            _fail("subplan status missing action_name")
-
-    completed = [
-        s for s in parent.subplan_statuses if s.status == ProcessingState.COMPLETE
-    ]
-    if len(completed) != 2:
-        _fail(f"expected 2 completed subplans, got {len(completed)}")
-    for s in completed:
-        if not s.changeset_summary:
-            _fail(f"completed subplan {s.subplan_id} missing changeset_summary")
-        if s.files_changed <= 0:
-            _fail(f"completed subplan {s.subplan_id} has no files_changed")
-
-    queued = [s for s in parent.subplan_statuses if s.status == ProcessingState.QUEUED]
-    if len(queued) != 1:
-        _fail(f"expected 1 queued subplan, got {len(queued)}")
-    if queued[0].changeset_summary is not None:
-        _fail("queued subplan should not have changeset_summary")
-
-    for child in [child_a, child_b, child_c]:
-        if child.identity.parent_plan_id != parent.identity.plan_id:
-            _fail(f"child {child.identity.plan_id} parent mismatch")
-
-    print("m4-plan-tree-ok")
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: plan-diff (CLI via subprocess)
-# ---------------------------------------------------------------------------
-
-
-def plan_diff() -> None:
-    """Verify ``agents plan diff`` via real CLI subprocess.
-
-    Creates action + plan via subprocess, then runs ``plan diff``.
-    Without AI the plan has no changeset, so diff returns gracefully.
-    """
-    workspace = setup_workspace(prefix="m4_diff_")
-    yaml_path = write_yaml(_ACTION_YAML)
-    try:
-        r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
-        if r1.returncode != 0:
-            _fail(f"action create: {r1.stderr}")
-
-        r2 = run_cli(
-            "plan",
-            "use",
-            "local/m4-verify-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:\n{r2.stdout}")
-
-        r3 = run_cli("plan", "diff", plan_id, workspace=workspace)
-        combined = r3.stdout + r3.stderr
-        if "INTERNAL" in combined or "Traceback" in combined:
-            _fail(f"plan diff crashed:\n{combined}")
-
-        print("m4-plan-diff-ok")
-    finally:
-        os.unlink(yaml_path)
-        cleanup_workspace(workspace)
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: parallel-max (domain-level, no CLI)
-# ---------------------------------------------------------------------------
-
-
-def parallel_max() -> None:
-    """Verify parallel subplan execution respects max_parallel."""
-    config = SubplanConfig(
-        execution_mode=ExecutionMode.PARALLEL,
-        max_parallel=3,
-        fail_fast=False,
-        retry_failed=True,
-        max_retries=2,
-    )
-
-    if config.execution_mode != ExecutionMode.PARALLEL:
-        _fail(f"expected PARALLEL, got {config.execution_mode}")
-    if config.max_parallel != 3:
-        _fail(f"expected max_parallel=3, got {config.max_parallel}")
-
-    try:
-        SubplanConfig(execution_mode=ExecutionMode.PARALLEL, max_parallel=0)
-        _fail("max_parallel=0 should be rejected")
-    except ValueError:
-        pass
-
-    try:
-        SubplanConfig(execution_mode=ExecutionMode.PARALLEL, max_parallel=51)
-        _fail("max_parallel=51 should be rejected")
-    except ValueError:
-        pass
-
-    now = datetime.now()
-    statuses = [
-        SubplanStatus(
-            subplan_id=_CHILD_A_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-1"],
-            status=ProcessingState.PROCESSING,
-            started_at=now,
-        ),
-        SubplanStatus(
-            subplan_id=_CHILD_B_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-2"],
-            status=ProcessingState.PROCESSING,
-            started_at=now,
-        ),
-        SubplanStatus(
-            subplan_id=_CHILD_C_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-3"],
-            status=ProcessingState.QUEUED,
-        ),
-    ]
-    parent = _mock_parent_plan(subplan_config=config, subplan_statuses=statuses)
-    running = [
-        s for s in parent.subplan_statuses if s.status == ProcessingState.PROCESSING
-    ]
-    if len(running) > config.max_parallel:
-        _fail(f"running ({len(running)}) exceeds max_parallel ({config.max_parallel})")
-
-    for mode in ExecutionMode:
-        cfg = SubplanConfig(execution_mode=mode, max_parallel=5)
-        if cfg.execution_mode != mode:
-            _fail(f"execution_mode mismatch: {cfg.execution_mode} != {mode}")
-
-    handler = SubplanFailureHandler()
-    failed_status = SubplanStatus(
-        subplan_id=_CHILD_A_ULID,
-        action_name="local/m4-verify-action",
-        status=ProcessingState.ERRORED,
-        error="ValidationError: tests failed",
-        attempt_number=1,
-    )
-    if handler.should_stop_others(config, failed_status):
-        _fail("parallel + fail_fast=False should not stop others")
-
-    ff_config = SubplanConfig(
-        execution_mode=ExecutionMode.PARALLEL,
-        max_parallel=3,
-        fail_fast=True,
-    )
-    if not handler.should_stop_others(ff_config, failed_status):
-        _fail("fail_fast=True should stop others")
-    if not handler.should_retry(config, failed_status):
-        _fail("ValidationError should be retriable")
-
-    print("m4-parallel-max-ok")
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: merge-clean (domain-level, no CLI)
-# ---------------------------------------------------------------------------
-
-
-def merge_clean() -> None:
-    """Verify three-way merge combines non-conflicting changes."""
-    strategy = GitMergeStrategy()
-    base = "line1\nline2\nline3\nline4\nline5\n"
-    ours = "line1-modified-by-subplan-A\nline2\nline3\nline4\nline5\n"
-    theirs = "line1\nline2\nline3\nline4\nline5-modified-by-subplan-B\n"
-
-    result = strategy.merge(base, ours, theirs)
-    if not result.success:
-        _fail(f"clean merge should succeed, got conflicts: {result.has_conflicts}")
-    if result.has_conflicts:
-        _fail("clean merge should have no conflicts")
-    if "line1-modified-by-subplan-A" not in result.content:
-        _fail("merged content missing ours change")
-    if "line5-modified-by-subplan-B" not in result.content:
-        _fail("merged content missing theirs change")
-
-    seq_strategy = SequentialMergeStrategy()
-    seq_result = seq_strategy.merge(base, ours, theirs)
-    if not seq_result.success:
-        _fail("sequential merge should always succeed")
-    if seq_result.content != theirs:
-        _fail("sequential merge should return theirs")
-
-    print("m4-merge-clean-ok")
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: merge-conflict (domain-level, no CLI)
-# ---------------------------------------------------------------------------
-
-
-def merge_conflict() -> None:
-    """Verify merge conflicts are surfaced correctly."""
-    strategy = GitMergeStrategy()
-    base = "line1\nline2\nline3\n"
-    ours = "line1\nline2-ours\nline3\n"
-    theirs = "line1\nline2-theirs\nline3\n"
-
-    result = strategy.merge(base, ours, theirs)
-    if result.success:
-        _fail("conflicting merge should not succeed")
-    if not result.has_conflicts:
-        _fail("conflicting merge should have conflicts")
-    if "<<<<<<<" not in result.content:
-        _fail("merged content should contain <<<<<<< marker")
-    if ">>>>>>>" not in result.content:
-        _fail("merged content should contain >>>>>>> marker")
-
-    if SubplanMergeStrategy.FAIL_ON_CONFLICT != "fail_on_conflict":
-        _fail("FAIL_ON_CONFLICT value mismatch")
-
-    expected_strategies = {
-        "git_three_way",
-        "sequential_apply",
-        "fail_on_conflict",
-        "last_wins",
-    }
-    actual_strategies = {s.value for s in SubplanMergeStrategy}
-    if actual_strategies != expected_strategies:
-        _fail(
-            f"merge strategies mismatch: {actual_strategies} != {expected_strategies}"
-        )
-
-    print("m4-merge-conflict-ok")
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: parent-tracking (domain-level, no CLI)
-# ---------------------------------------------------------------------------
-
-
-def parent_tracking() -> None:
-    """Verify parent plan tracks all subplan statuses correctly."""
-    now = datetime.now()
-    statuses = [
-        SubplanStatus(
-            subplan_id=_CHILD_A_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-api"],
-            status=ProcessingState.COMPLETE,
-            started_at=now,
-            completed_at=now,
-            changeset_summary="Updated API",
-            files_changed=3,
-            attempt_number=1,
-        ),
-        SubplanStatus(
-            subplan_id=_CHILD_B_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-ui"],
-            status=ProcessingState.ERRORED,
-            started_at=now,
-            completed_at=now,
-            error="TimeoutError: execution timed out",
-            attempt_number=1,
-            previous_attempts=[],
-        ),
-        SubplanStatus(
-            subplan_id=_CHILD_C_ULID,
-            action_name="local/m4-verify-action",
-            target_resources=["res-db"],
-            status=ProcessingState.PROCESSING,
-            started_at=now,
-            attempt_number=2,
-            previous_attempts=[
-                SubplanAttempt(
-                    attempt_number=1,
-                    started_at=now,
-                    completed_at=now,
-                    error="ValidationError: schema mismatch",
-                    was_retried=True,
-                ),
-            ],
-        ),
-    ]
-    parent = _mock_parent_plan(subplan_statuses=statuses)
-
-    if len(parent.subplan_statuses) != 3:
-        _fail(f"expected 3 statuses, got {len(parent.subplan_statuses)}")
-
-    completed = [
-        s for s in parent.subplan_statuses if s.status == ProcessingState.COMPLETE
-    ]
-    if len(completed) != 1:
-        _fail(f"expected 1 completed, got {len(completed)}")
-    errored = [
-        s for s in parent.subplan_statuses if s.status == ProcessingState.ERRORED
-    ]
-    if len(errored) != 1:
-        _fail(f"expected 1 errored, got {len(errored)}")
-    if "TimeoutError" not in (errored[0].error or ""):
-        _fail(f"error should contain TimeoutError, got {errored[0].error}")
-
-    retried = [
-        s for s in parent.subplan_statuses if s.status == ProcessingState.PROCESSING
-    ]
-    if len(retried) != 1:
-        _fail(f"expected 1 processing, got {len(retried)}")
-    if retried[0].attempt_number != 2:
-        _fail(f"retried should be attempt 2, got {retried[0].attempt_number}")
-
-    handler = SubplanFailureHandler()
-    config = parent.subplan_config
-    if config is None:
-        _fail("parent should have subplan_config")
-    if not handler.should_retry(config, errored[0]):
-        _fail("TimeoutError should be retriable")
-
-    non_retriable_status = SubplanStatus(
-        subplan_id=_CHILD_A_ULID,
-        action_name="local/m4-verify-action",
-        status=ProcessingState.ERRORED,
-        error="ConfigurationError: invalid config",
-        attempt_number=1,
-    )
-    if handler.should_retry(config, non_retriable_status):
-        _fail("ConfigurationError should NOT be retriable")
-
-    cli_dict = parent.as_cli_dict()
-    if cli_dict.get("subplan_count") != 3:
-        _fail(f"subplan_count should be 3, got {cli_dict.get('subplan_count')}")
-
-    print("m4-parent-tracking-ok")
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: cli-plan-use (CLI via subprocess)
-# ---------------------------------------------------------------------------
-
-
-def cli_plan_use() -> None:
-    """Verify ``agents plan use`` via real CLI subprocess."""
-    workspace = setup_workspace(prefix="m4_use_")
-    yaml_path = write_yaml(_ACTION_YAML)
-    try:
-        r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
-        if r1.returncode != 0:
-            _fail(f"action create: {r1.stderr}")
-
-        r2 = run_cli(
-            "plan",
-            "use",
-            "local/m4-verify-action",
-            "local/monorepo",
-            "--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:\n{r2.stdout}")
-
-        # Verify plan persisted
-        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 status output:\n{r3.stdout}")
-
-        print("m4-cli-plan-use-ok")
-    finally:
-        os.unlink(yaml_path)
-        cleanup_workspace(workspace)
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: cli-plan-execute (CLI via subprocess)
-# ---------------------------------------------------------------------------
-
-
-def cli_plan_execute() -> None:
-    """Verify ``agents plan execute`` via real CLI subprocess.
-
-    Plan is in strategize/queued so execute correctly rejects
-    with a controlled "not ready" message.
-    """
-    workspace = setup_workspace(prefix="m4_exec_")
-    yaml_path = write_yaml(_ACTION_YAML)
-    try:
-        r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
-        if r1.returncode != 0:
-            _fail(f"action create: {r1.stderr}")
-
-        r2 = run_cli(
-            "plan",
-            "use",
-            "local/m4-verify-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:\n{r2.stdout}")
-
-        r3 = run_cli("plan", "execute", plan_id, workspace=workspace)
-        combined = r3.stdout + r3.stderr
-        if "INTERNAL" in combined or "Traceback" in combined:
-            _fail(f"plan execute crashed:\n{combined}")
-        # Verify the output references the plan
-        if plan_id not in combined:
-            _fail(f"plan execute output missing plan_id {plan_id}")
-
-        print("m4-cli-plan-execute-ok")
-    finally:
-        os.unlink(yaml_path)
-        cleanup_workspace(workspace)
-
-
-# ---------------------------------------------------------------------------
-# Subcommand: cli-plan-tree (CLI via subprocess)
-# ---------------------------------------------------------------------------
-
-
-def cli_plan_tree() -> None:
-    """Verify ``agents plan tree`` displays decision tree via subprocess.
-
-    Seeds decisions into the workspace DB, then runs ``plan tree``
-    via subprocess.
-    """
-    workspace = setup_workspace(prefix="m4_tree_")
-    yaml_path = write_yaml(_ACTION_YAML)
-    try:
-        r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
-        if r1.returncode != 0:
-            _fail(f"action create: {r1.stderr}")
-
-        r2 = run_cli(
-            "plan",
-            "use",
-            "local/m4-verify-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:\n{r2.stdout}")
-
-        # Seed decisions into the same DB
-        db_url = os.environ["CLEVERAGENTS_DATABASE_URL"]
-        uow = UnitOfWork(db_url)
-        prev_db = os.environ.get("CLEVERAGENTS_DATABASE_URL")
-        os.environ["CLEVERAGENTS_DATABASE_URL"] = db_url
-        try:
-            settings = Settings()
-        finally:
-            if prev_db is None:
-                os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
-            else:
-                os.environ["CLEVERAGENTS_DATABASE_URL"] = prev_db
-
-        decision_service = DecisionService(settings=settings, unit_of_work=uow)
-
-        decision_service.record_decision(
-            plan_id=plan_id,
-            decision_type=DecisionType.PROMPT_DEFINITION,
-            question="What is the plan prompt?",
-            chosen_option="Refactor code across monorepo modules",
-            confidence_score=1.0,
-            context_snapshot=ContextSnapshot(
-                hot_context_hash="sha256:test",
-                hot_context_ref="store://test",
-                relevant_resources=[
-                    ResourceRef(resource_id="res-1", path="src/main.py")
-                ],
-                actor_state_ref="checkpoint://test",
-            ),
-        )
-        decision_service.record_decision(
-            plan_id=plan_id,
-            decision_type=DecisionType.STRATEGY_CHOICE,
-            question="How to decompose the refactoring?",
-            chosen_option="Parallel subplans per module",
-            confidence_score=0.85,
-            context_snapshot=ContextSnapshot(
-                hot_context_hash="sha256:test2",
-                hot_context_ref="store://test2",
-                relevant_resources=[
-                    ResourceRef(resource_id="res-2", path="src/api.py")
-                ],
-                actor_state_ref="checkpoint://test2",
-            ),
-        )
-
-        r3 = run_cli(
-            "plan",
-            "tree",
-            plan_id,
-            "--format",
-            "json",
-            workspace=workspace,
-        )
-        if r3.returncode != 0:
-            _fail(f"plan tree rc={r3.returncode}\n{r3.stderr}")
-
-        # Verify JSON output contains decision types
-        import json
-
-        tree_data = json.loads(r3.stdout) if r3.stdout.strip().startswith("[") else None
-        if tree_data is None:
-            # Output may have log lines before JSON — scan for it
-            for line in r3.stdout.split("\n"):
-                line = line.strip()
-                if line.startswith("[") or line.startswith("{"):
-                    try:
-                        tree_data = json.loads(line)
-                        break
-                    except json.JSONDecodeError:
-                        continue
-        if tree_data is not None:
-            tree_str = json.dumps(tree_data)
-            if "prompt_definition" not in tree_str:
-                _fail(f"tree missing prompt_definition:\n{r3.stdout}")
-
-        print("m4-cli-plan-tree-ok")
-    finally:
-        os.unlink(yaml_path)
-        cleanup_workspace(workspace)
-
-
-# ---------------------------------------------------------------------------
-# Main dispatcher
-# ---------------------------------------------------------------------------
-
-_COMMANDS: dict[str, Callable[[], None]] = {
-    "spawn-subplans": spawn_subplans,
-    "plan-tree": plan_tree,
-    "plan-diff": plan_diff,
-    "parallel-max": parallel_max,
-    "merge-clean": merge_clean,
-    "merge-conflict": merge_conflict,
-    "parent-tracking": parent_tracking,
-    "cli-plan-use": cli_plan_use,
-    "cli-plan-execute": cli_plan_execute,
-    "cli-plan-tree": cli_plan_tree,
+# 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(_COMMANDS)}>",
+            f"Usage: helper_m4_e2e_verification.py <{'|'.join(_COMMAND_MAP)}>",
         )
         return 1
     command = sys.argv[1]
-    handler = _COMMANDS.get(command)
+    handler = _resolve_handler(command)
     if handler is None:
         print(f"Unknown command: {command}")
         return 1

robot/helpers_common.py

@@ -1,7 +1,9 @@
 """Shared utilities for Robot Framework helper scripts.
 
 This module centralises the ``reset_global_state()`` function used by
-``helper_m4_e2e_verification.py``,
+``helper_m4_e2e_verification.py`` (and its sub-modules
+``helper_m4_e2e_common.py``, ``helper_m4_e2e_domain.py``,
+``helper_m4_e2e_cli.py``),
 ``helper_m4_correction_subplan_smoke.py``, and
 ``helper_m3_decision_validation_smoke.py``.
 

robot/m4_e2e_verification.robot

@@ -1,7 +1,7 @@
 *** Settings ***
 Documentation    M4 end-to-end verification: subplan spawning, parallel execution,
-...              plan tree viewing, three-way merge, conflict surfacing, and
-...              parent plan subplan status tracking.
+...              plan tree viewing, three-way merge, conflict surfacing,
+...              parent plan subplan status tracking, and CLI error paths.
 Resource         ${CURDIR}/common.resource
 Suite Setup      Setup Test Environment With Database Isolation
 Suite Teardown   Cleanup Test Environment
@@ -21,9 +21,8 @@ Plan Spawns Multiple Subplans During Execute
     Should Contain    ${result.stdout}    m4-spawn-subplans-ok
 
 View Subplan Tree Via Plan Tree
-    [Documentation]    View the subplan tree for a parent plan.
-    ...                Verifies parent-child hierarchy is correctly represented
-    ...                including depth, root_plan_id, and subplan_statuses.
+    [Documentation]    Verify SubplanStatus field completeness and parent-child
+    ...                linkage for the plan tree data model.
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    plan-tree    cwd=${WORKSPACE}    timeout=30s
     Log    ${result.stdout}
     Log    ${result.stderr}
@@ -108,3 +107,43 @@ CLI Plan Tree Displays Subplan Hierarchy
     Log    ${result.stderr}
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    m4-cli-plan-tree-ok
+
+CLI Plan Execute Aborts On Read Only Plan
+    [Documentation]    Invoke agents plan execute on a read-only plan.
+    ...                Verifies the CLI aborts with an appropriate error message
+    ...                and does not call execute_plan.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cli-plan-execute-readonly    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    m4-cli-plan-execute-readonly-ok
+
+CLI Plan Use Aborts On Unavailable Action
+    [Documentation]    Invoke agents plan use with a nonexistent/unavailable action.
+    ...                Verifies the CLI aborts with an appropriate error message
+    ...                and does not call use_action.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cli-plan-use-not-found    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    m4-cli-plan-use-not-found-ok
+
+CLI Plan Diff Aborts On Missing Changeset
+    [Documentation]    Invoke agents plan diff when the plan has no changeset.
+    ...                Verifies the CLI catches PlanError and aborts with a
+    ...                user-friendly error message.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cli-plan-diff-no-changeset    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    m4-cli-plan-diff-no-changeset-ok
+
+CLI Plan Tree Handles Zero Decisions Gracefully
+    [Documentation]    Invoke agents plan tree when no decisions exist.
+    ...                Verifies the CLI prints an informational message and
+    ...                exits cleanly (exit code 0).
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cli-plan-tree-empty    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    m4-cli-plan-tree-empty-ok