cleveragents-core/robot/helper_plan_lifecycle_model_validation.py

"""Robot Framework helper for plan lifecycle model validation tests.

Each function is invoked by the corresponding Robot test case via
``Run Process    python    <this-file>    <command>``.
"""

from __future__ import annotations

import sys

from pydantic import ValidationError

from cleveragents.domain.models.core.plan import (
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    ProcessingState,
)

_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"


def _make_plan(
    phase: PlanPhase,
    processing_state: ProcessingState,
) -> Plan:
    """Build a Plan with the given phase and processing state."""
    return Plan(
        identity=PlanIdentity(plan_id=_ULID),
        namespaced_name=NamespacedName(
            server=None, namespace="local", name="plmv-robot"
        ),
        description="Plan lifecycle model validation robot test",
        action_name="local/plmv-robot-action",
        phase=phase,
        processing_state=processing_state,
    )


def _errored_is_terminal() -> None:
    """Verify ERRORED state is terminal in all applicable phases."""
    for phase in (
        PlanPhase.ACTION,
        PlanPhase.STRATEGIZE,
        PlanPhase.EXECUTE,
        PlanPhase.APPLY,
    ):
        plan = _make_plan(phase, ProcessingState.ERRORED)
        assert plan.is_terminal, f"ERRORED should be terminal in {phase.value} phase"
        assert plan.is_errored, (
            f"ERRORED plan should report is_errored in {phase.value} phase"
        )
    print("errored-is-terminal-ok")


def _processing_any_phase() -> None:
    """Verify PROCESSING is accepted in all phases."""
    for phase in PlanPhase:
        plan = _make_plan(phase, ProcessingState.PROCESSING)
        assert plan.processing_state == ProcessingState.PROCESSING, (
            f"PROCESSING not set in {phase.value}"
        )
    print("processing-any-phase-ok")


def _applied_only_apply() -> None:
    """Verify APPLIED is rejected outside APPLY phase."""
    # Valid in APPLY
    plan = _make_plan(PlanPhase.APPLY, ProcessingState.APPLIED)
    assert plan.is_terminal

    # Invalid in STRATEGIZE, EXECUTE, and ACTION
    for phase in (PlanPhase.STRATEGIZE, PlanPhase.EXECUTE, PlanPhase.ACTION):
        try:
            _make_plan(phase, ProcessingState.APPLIED)
            raise AssertionError(
                f"Expected ValidationError for APPLIED in {phase.value}"
            )
        except ValidationError as exc:
            assert "APPLIED is only valid in the APPLY phase" in str(exc)

    print("applied-only-apply-ok")


def _constrained_only_apply() -> None:
    """Verify CONSTRAINED is rejected outside APPLY phase."""
    # Valid in APPLY
    plan = _make_plan(PlanPhase.APPLY, ProcessingState.CONSTRAINED)
    assert plan.is_terminal

    # Invalid in STRATEGIZE, EXECUTE, and ACTION
    for phase in (PlanPhase.STRATEGIZE, PlanPhase.EXECUTE, PlanPhase.ACTION):
        try:
            _make_plan(phase, ProcessingState.CONSTRAINED)
            raise AssertionError(
                f"Expected ValidationError for CONSTRAINED in {phase.value}"
            )
        except ValidationError as exc:
            assert "CONSTRAINED is only valid in the APPLY phase" in str(exc)

    print("constrained-only-apply-ok")


def _complete_strategize_execute() -> None:
    """Verify COMPLETE is only valid in STRATEGIZE and EXECUTE phases."""
    # Valid
    for phase in (PlanPhase.STRATEGIZE, PlanPhase.EXECUTE):
        plan = _make_plan(phase, ProcessingState.COMPLETE)
        assert not plan.is_terminal, f"COMPLETE should not be terminal in {phase.value}"

    # Invalid in APPLY
    try:
        _make_plan(PlanPhase.APPLY, ProcessingState.COMPLETE)
        raise AssertionError("Expected ValidationError for COMPLETE in APPLY")
    except ValidationError as exc:
        assert "COMPLETE is only valid in STRATEGIZE or EXECUTE" in str(exc)

    # Invalid in ACTION
    try:
        _make_plan(PlanPhase.ACTION, ProcessingState.COMPLETE)
        raise AssertionError("Expected ValidationError for COMPLETE in ACTION")
    except ValidationError as exc:
        assert "COMPLETE is only valid in STRATEGIZE or EXECUTE" in str(exc)

    print("complete-strategize-execute-ok")


def _queued_any_phase() -> None:
    """Verify QUEUED is accepted in all phases."""
    for phase in PlanPhase:
        plan = _make_plan(phase, ProcessingState.QUEUED)
        assert plan.processing_state == ProcessingState.QUEUED, (
            f"QUEUED not set in {phase.value}"
        )
    print("queued-any-phase-ok")


def _assignment_order_safety() -> None:
    """Verify that state-first assignment order avoids validator errors.

    With ``validate_assignment=True`` on the Plan model, each field
    set triggers the phase-state validator.  Setting ``processing_state``
    (to a universally valid value like QUEUED) before ``phase`` avoids
    transient invalid intermediate states.
    """
    plan = _make_plan(PlanPhase.STRATEGIZE, ProcessingState.COMPLETE)
    # State-first transition: COMPLETE → QUEUED then STRATEGIZE → EXECUTE
    plan.processing_state = ProcessingState.QUEUED
    plan.phase = PlanPhase.EXECUTE
    assert plan.phase == PlanPhase.EXECUTE
    assert plan.processing_state == ProcessingState.QUEUED
    print("assignment-order-safety-ok")


def _deserialization_coercion() -> None:
    """Verify that to_domain() coerces legacy invalid phase/state combos.

    The defensive coercion in LifecyclePlanModel.to_domain() should
    map APPLY/COMPLETE → APPLY/APPLIED without crashing.
    """
    from cleveragents.infrastructure.database.models import (
        LifecyclePlanModel,
    )

    model = LifecyclePlanModel()
    model.plan_id = _ULID
    model.namespaced_name = "local/plmv-deser"
    model.description = "deserialization coercion test"
    model.action_name = "local/plmv-action"
    model.phase = "apply"
    model.processing_state = "complete"  # legacy invalid combo
    model.attempt = 1
    model.created_at = "2025-01-01T00:00:00"
    model.updated_at = "2025-01-01T00:00:00"
    model.reusable = True
    model.read_only = False
    model.sandbox_refs_json = "[]"
    model.tags_json = "[]"
    model.reversion_count = 0
    model.last_completed_step = -1

    plan = model.to_domain()
    # Should have coerced COMPLETE → APPLIED in APPLY phase
    assert plan.processing_state.value == "applied", (
        f"Expected 'applied', got '{plan.processing_state.value}'"
    )
    assert plan.phase.value == "apply"
    print("deserialization-coercion-ok")


def _deserialization_coercion_all_branches() -> None:
    """Verify all three coercion branches in to_domain().

    Branch 1: APPLY/COMPLETE → APPLY/APPLIED
    Branch 2: non-APPLY/APPLIED → non-APPLY/ERRORED
    Branch 3: ACTION/COMPLETE → ACTION/QUEUED
    """
    from cleveragents.infrastructure.database.models import (
        LifecyclePlanModel,
    )

    def _make_legacy_model(phase: str, state: str, suffix: str) -> LifecyclePlanModel:
        model = LifecyclePlanModel()
        model.plan_id = _ULID
        model.namespaced_name = f"local/plmv-deser-{suffix}"
        model.description = "deserialization coercion all-branches test"
        model.action_name = "local/plmv-action"
        model.phase = phase
        model.processing_state = state
        model.attempt = 1
        model.created_at = "2025-01-01T00:00:00"
        model.updated_at = "2025-01-01T00:00:00"
        model.reusable = True
        model.read_only = False
        model.sandbox_refs_json = "[]"
        model.tags_json = "[]"
        model.reversion_count = 0
        model.last_completed_step = -1
        return model

    # Branch 1: APPLY/COMPLETE → APPLY/APPLIED
    plan1 = _make_legacy_model("apply", "complete", "b1").to_domain()
    assert plan1.processing_state.value == "applied", (
        f"Branch 1: expected 'applied', got '{plan1.processing_state.value}'"
    )

    # Branch 2: STRATEGIZE/APPLIED → STRATEGIZE/ERRORED
    plan2 = _make_legacy_model("strategize", "applied", "b2").to_domain()
    assert plan2.processing_state.value == "errored", (
        f"Branch 2: expected 'errored', got '{plan2.processing_state.value}'"
    )

    # Branch 2b: EXECUTE/CONSTRAINED → EXECUTE/ERRORED
    plan2b = _make_legacy_model("execute", "constrained", "b2b").to_domain()
    assert plan2b.processing_state.value == "errored", (
        f"Branch 2b: expected 'errored', got '{plan2b.processing_state.value}'"
    )

    # Branch 3: ACTION/COMPLETE → ACTION/QUEUED
    plan3 = _make_legacy_model("action", "complete", "b3").to_domain()
    assert plan3.processing_state.value == "queued", (
        f"Branch 3: expected 'queued', got '{plan3.processing_state.value}'"
    )

    print("deserialization-coercion-all-branches-ok")


_COMMANDS: dict[str, object] = {
    "errored-is-terminal": _errored_is_terminal,
    "processing-any-phase": _processing_any_phase,
    "applied-only-apply": _applied_only_apply,
    "constrained-only-apply": _constrained_only_apply,
    "complete-strategize-execute": _complete_strategize_execute,
    "queued-any-phase": _queued_any_phase,
    "assignment-order-safety": _assignment_order_safety,
    "deserialization-coercion": _deserialization_coercion,
    "deserialization-coercion-all-branches": _deserialization_coercion_all_branches,
}


def main() -> None:
    """Dispatch to the requested test function."""
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
        print(f"Commands: {', '.join(_COMMANDS)}", file=sys.stderr)
        sys.exit(1)

    cmd = sys.argv[1]
    fn = _COMMANDS.get(cmd)
    if fn is None:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)

    assert callable(fn)
    fn()


if __name__ == "__main__":
    main()