cleveragents-core/robot/helper_m4_e2e_domain.py

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