"""Helper script for subplan model Robot Framework smoke tests. Exercises SubplanConfig, SubplanStatus, SubplanFailureHandler, and Plan hierarchy properties without requiring the service layer or persistence. """ from __future__ import annotations import sys from cleveragents.domain.models.core.plan import ( ExecutionMode, NamespacedName, Plan, PlanIdentity, PlanPhase, ProcessingState, SubplanConfig, SubplanFailureHandler, SubplanMergeStrategy, SubplanStatus, ) # --------------------------------------------------------------------------- # Valid ULID constants # --------------------------------------------------------------------------- _ROOT_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV" _PARENT_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAW" _CHILD_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAX" _GRANDCHILD_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAY" _SUB_STATUS_ULID = "01KH72MDGQMPRW3R5WPZVB8KC1" def _make_plan( *, plan_id: str = _ROOT_ULID, parent_plan_id: str | None = None, root_plan_id: str | None = None, phase: PlanPhase = PlanPhase.STRATEGIZE, processing_state: ProcessingState = ProcessingState.QUEUED, subplan_config: SubplanConfig | None = None, subplan_statuses: list[SubplanStatus] | None = None, ) -> Plan: """Create a Plan for testing with minimal required fields.""" return Plan( identity=PlanIdentity( plan_id=plan_id, parent_plan_id=parent_plan_id, root_plan_id=root_plan_id, ), namespaced_name=NamespacedName( server=None, namespace="local", name="robot-subplan-test" ), description="Robot subplan test plan", action_name="local/robot-subplan-action", phase=phase, processing_state=processing_state, subplan_config=subplan_config, subplan_statuses=subplan_statuses or [], ) # --------------------------------------------------------------------------- # Command handlers # --------------------------------------------------------------------------- def _root_identity() -> None: """Verify root plan identity flags.""" plan = _make_plan() assert plan.is_root_plan, "Expected root plan" assert not plan.is_subplan, "Expected not subplan" assert plan.depth == 0, f"Expected depth 0, got {plan.depth}" print("root-identity-ok") def _child_identity() -> None: """Verify child plan identity flags.""" plan = _make_plan( plan_id=_CHILD_ULID, parent_plan_id=_PARENT_ULID, root_plan_id=_ROOT_ULID, ) assert plan.is_subplan, "Expected subplan" assert not plan.is_root_plan, "Expected not root plan" assert plan.depth == -1, f"Expected depth -1, got {plan.depth}" print("child-identity-ok") def _hierarchy_propagation() -> None: """Verify root_plan_id propagates through three levels.""" root = _make_plan(plan_id=_ROOT_ULID, root_plan_id=_ROOT_ULID) parent = _make_plan( plan_id=_PARENT_ULID, parent_plan_id=_ROOT_ULID, root_plan_id=_ROOT_ULID, ) grandchild = _make_plan( plan_id=_GRANDCHILD_ULID, parent_plan_id=_PARENT_ULID, root_plan_id=_ROOT_ULID, ) assert root.is_root_plan assert parent.is_subplan assert grandchild.is_subplan assert grandchild.identity.root_plan_id == root.identity.plan_id print("hierarchy-propagation-ok") def _config_defaults() -> None: """Verify SubplanConfig defaults.""" cfg = SubplanConfig() assert cfg.execution_mode == ExecutionMode.SEQUENTIAL assert cfg.merge_strategy == SubplanMergeStrategy.GIT_THREE_WAY assert cfg.max_parallel == 5 assert cfg.fail_fast is False assert cfg.retry_failed is True assert cfg.max_retries == 2 assert cfg.timeout_per_subplan_seconds is None print("config-defaults-ok") def _config_custom() -> None: """Verify SubplanConfig accepts custom settings.""" cfg = SubplanConfig( execution_mode=ExecutionMode.PARALLEL, merge_strategy=SubplanMergeStrategy.FAIL_ON_CONFLICT, max_parallel=10, fail_fast=True, retry_failed=True, max_retries=4, timeout_per_subplan_seconds=300, ) assert cfg.execution_mode == ExecutionMode.PARALLEL assert cfg.merge_strategy == SubplanMergeStrategy.FAIL_ON_CONFLICT assert cfg.max_parallel == 10 assert cfg.fail_fast is True assert cfg.max_retries == 4 assert cfg.timeout_per_subplan_seconds == 300 print("config-custom-ok") def _failure_stop_others() -> None: """Verify SubplanFailureHandler stops others when fail_fast=True.""" config = SubplanConfig( execution_mode=ExecutionMode.PARALLEL, fail_fast=True, ) status = SubplanStatus( subplan_id=_SUB_STATUS_ULID, action_name="local/sub-task", status=ProcessingState.ERRORED, error="TimeoutError: timed out", ) handler = SubplanFailureHandler() assert handler.should_stop_others(config, status) is True print("failure-stop-others-ok") def _failure_retry_retriable() -> None: """Verify SubplanFailureHandler retries retriable errors.""" config = SubplanConfig(retry_failed=True, max_retries=2) status = SubplanStatus( subplan_id=_SUB_STATUS_ULID, action_name="local/sub-task", status=ProcessingState.ERRORED, error="TimeoutError: timed out", attempt_number=1, ) handler = SubplanFailureHandler() assert handler.should_retry(config, status) is True print("failure-retry-retriable-ok") def _failure_skip_non_retriable() -> None: """Verify SubplanFailureHandler does not retry non-retriable errors.""" config = SubplanConfig(retry_failed=True, max_retries=2) status = SubplanStatus( subplan_id=_SUB_STATUS_ULID, action_name="local/sub-task", status=ProcessingState.ERRORED, error="ConfigurationError: bad config", attempt_number=1, ) handler = SubplanFailureHandler() assert handler.should_retry(config, status) is False print("failure-skip-non-retriable-ok") def _cli_dict_child() -> None: """Verify as_cli_dict includes parent_plan_id for child plan.""" plan = _make_plan( plan_id=_CHILD_ULID, parent_plan_id=_PARENT_ULID, root_plan_id=_ROOT_ULID, ) cli = plan.as_cli_dict() assert "parent_plan_id" in cli, "Expected parent_plan_id in CLI dict" assert cli["parent_plan_id"] == _PARENT_ULID print("cli-dict-child-ok") def _cli_dict_subplan_count() -> None: """Verify as_cli_dict includes subplan_count when subplans exist.""" statuses = [ SubplanStatus( subplan_id=_SUB_STATUS_ULID, action_name="local/sub-task", status=ProcessingState.QUEUED, ), ] plan = _make_plan(subplan_statuses=statuses) cli = plan.as_cli_dict() assert "subplan_count" in cli, "Expected subplan_count in CLI dict" assert cli["subplan_count"] == 1 print("cli-dict-subplan-count-ok") def _plan_with_config() -> None: """Verify Plan model accepts subplan config and statuses.""" config = SubplanConfig( execution_mode=ExecutionMode.SEQUENTIAL, merge_strategy=SubplanMergeStrategy.SEQUENTIAL_APPLY, ) statuses = [ SubplanStatus( subplan_id=_SUB_STATUS_ULID, action_name="local/sub-task", target_resources=["src/main.py"], status=ProcessingState.QUEUED, ), ] plan = _make_plan(subplan_config=config, subplan_statuses=statuses) assert plan.has_subplans assert plan.subplan_config is not None assert plan.subplan_config.execution_mode == ExecutionMode.SEQUENTIAL assert len(plan.subplan_statuses) == 1 print("plan-with-config-ok") def main() -> None: """Dispatch command from sys.argv.""" if len(sys.argv) < 2: raise SystemExit("Expected command argument") command = sys.argv[1] commands: dict[str, object] = { "root-identity": _root_identity, "child-identity": _child_identity, "hierarchy-propagation": _hierarchy_propagation, "config-defaults": _config_defaults, "config-custom": _config_custom, "failure-stop-others": _failure_stop_others, "failure-retry-retriable": _failure_retry_retriable, "failure-skip-non-retriable": _failure_skip_non_retriable, "cli-dict-child": _cli_dict_child, "cli-dict-subplan-count": _cli_dict_subplan_count, "plan-with-config": _plan_with_config, } if command not in commands: raise SystemExit(f"Unknown command: {command}") func = commands[command] if callable(func): func() if __name__ == "__main__": main()