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