cleveragents-core/robot/helper_phase_reversion.py

"""Helper script for Robot Framework phase reversion tests."""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from cleveragents.application.services.plan_lifecycle_service import (
    PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.core.exceptions import PlanError
from cleveragents.domain.models.core.action import (
    ActionArgument,
    ArgumentRequirement,
    ArgumentType,
)
from cleveragents.domain.models.core.plan import (
    AutomationProfileProvenance,
    AutomationProfileRef,
    InvariantSource,
    PlanInvariant,
    PlanPhase,
    ProcessingState,
    ProjectLink,
)


def _make_service() -> PlanLifecycleService:
    """Create a fresh PlanLifecycleService."""
    return PlanLifecycleService(settings=Settings())


def _create_plan(
    service: PlanLifecycleService,
    action_name: str = "local/test-action",
    profile: str = "manual",
) -> str:
    """Create an action and plan, return plan_id."""
    service.create_action(
        name=action_name,
        description=f"Test {action_name}",
        definition_of_done="Tests pass",
        strategy_actor="local/strategist",
        execution_actor="local/executor",
    )
    plan = service.use_action(action_name)
    plan.automation_profile = AutomationProfileRef(
        profile_name=profile,
        provenance=AutomationProfileProvenance.PLAN,
    )
    return plan.identity.plan_id


def _advance_to_execute(service: PlanLifecycleService, plan_id: str) -> None:
    """Advance plan to Execute phase."""
    service.start_strategize(plan_id)
    service.complete_strategize(plan_id)
    plan = service.get_plan(plan_id)
    if plan.phase != PlanPhase.EXECUTE:
        service.execute_plan(plan_id)


def _advance_to_apply_constrained(
    service: PlanLifecycleService,
    plan_id: str,
) -> None:
    """Advance plan to Apply/constrained."""
    _advance_to_execute(service, plan_id)
    service.start_execute(plan_id)
    service.complete_execute(plan_id)
    plan = service.get_plan(plan_id)
    if plan.phase != PlanPhase.APPLY:
        service.apply_plan(plan_id)
    plan = service.get_plan(plan_id)
    if plan.processing_state == ProcessingState.QUEUED:
        service.start_apply(plan_id)
    plan = service.get_plan(plan_id)
    plan.processing_state = ProcessingState.CONSTRAINED
    plan.error_message = "Constraints too restrictive"


def _advance_to_applied(service: PlanLifecycleService, plan_id: str) -> None:
    """Advance plan to Applied terminal state."""
    _advance_to_execute(service, plan_id)
    service.start_execute(plan_id)
    service.complete_execute(plan_id)
    plan = service.get_plan(plan_id)
    if plan.phase != PlanPhase.APPLY:
        service.apply_plan(plan_id)
    plan = service.get_plan(plan_id)
    if plan.processing_state == ProcessingState.QUEUED:
        service.start_apply(plan_id)
    service.complete_apply(plan_id)


def test_manual_revert() -> None:
    """Test manual reversion from Execute to Strategize."""
    service = _make_service()
    plan_id = _create_plan(service)
    _advance_to_execute(service, plan_id)

    plan = service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="test revert")
    assert plan.phase == PlanPhase.STRATEGIZE
    assert plan.processing_state == ProcessingState.QUEUED
    assert plan.reversion_count == 1
    assert len(plan.decisions) == 1
    assert plan.decisions[0]["type"] == "reversion"
    print("reversion-manual-ok")


def test_auto_revert_apply() -> None:
    """Test auto-reversion from constrained Apply."""
    service = _make_service()
    plan_id = _create_plan(service, profile="ci")
    _advance_to_apply_constrained(service, plan_id)

    plan = service.try_auto_revert_from_apply(plan_id, "constraints too strict")
    assert plan.phase == PlanPhase.STRATEGIZE
    assert plan.reversion_count == 1
    print("reversion-auto-apply-ok")


def test_loop_guard() -> None:
    """Test loop guard prevents more than 3 reversions."""
    service = _make_service()
    plan_id = _create_plan(service)
    _advance_to_execute(service, plan_id)

    plan = service.get_plan(plan_id)
    plan.reversion_count = 3

    try:
        service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="should fail")
        print("ERROR: should have raised PlanError")
        sys.exit(1)
    except PlanError as e:
        assert "maximum reversion count" in str(e).lower() or "max" in str(e).lower()
        print("loop-guard-ok")


def test_profile_gate() -> None:
    """Test that manual profile blocks auto-reversion."""
    service = _make_service()
    plan_id = _create_plan(service, profile="manual")
    _advance_to_apply_constrained(service, plan_id)

    plan = service.try_auto_revert_from_apply(plan_id, "should stay")
    assert plan.phase == PlanPhase.APPLY
    assert plan.processing_state == ProcessingState.CONSTRAINED
    print("profile-gate-ok")


def test_terminal_block() -> None:
    """Test that applied terminal plan cannot be reverted."""
    service = _make_service()
    plan_id = _create_plan(service)
    _advance_to_applied(service, plan_id)

    try:
        service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="should fail")
        print("ERROR: should have raised PlanError")
        sys.exit(1)
    except PlanError as e:
        assert "terminal" in str(e).lower()
        print("terminal-block-ok")


def test_context_preserved() -> None:
    """Test that plan context is preserved through reversion."""
    service = _make_service()
    service.create_action(
        name="local/ctx-test",
        description="Context test",
        definition_of_done="Tests pass",
        strategy_actor="local/strategist",
        execution_actor="local/executor",
        arguments=[
            ActionArgument(
                name="coverage",
                arg_type=ArgumentType.INTEGER,
                requirement=ArgumentRequirement.OPTIONAL,
                description="Coverage threshold",
            ),
        ],
    )
    plan = service.use_action(
        "local/ctx-test",
        project_links=[ProjectLink(project_name="local/my-project")],
        arguments={"coverage": 95},
        invariants=[PlanInvariant(text="No regressions", source=InvariantSource.PLAN)],
    )
    plan.automation_profile = AutomationProfileRef(
        profile_name="manual",
        provenance=AutomationProfileProvenance.PLAN,
    )
    plan_id = plan.identity.plan_id

    service.start_strategize(plan_id)
    service.complete_strategize(plan_id)
    service.execute_plan(plan_id)
    plan = service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="context check")

    assert plan.arguments == {"coverage": 95}
    assert len(plan.project_links) == 1
    assert plan.project_links[0].project_name == "local/my-project"
    inv_texts = [inv.text for inv in plan.invariants]
    assert "No regressions" in inv_texts
    print("context-preserved-ok")


def test_decision_recording() -> None:
    """Test that each reversion is recorded as a decision."""
    service = _make_service()
    plan_id = _create_plan(service)
    _advance_to_execute(service, plan_id)

    plan = service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="first")
    assert len(plan.decisions) == 1
    assert plan.decisions[0]["type"] == "reversion"
    assert plan.decisions[0]["source_phase"] == "execute"
    assert plan.decisions[0]["target_phase"] == "strategize"
    assert plan.decisions[0]["reversion_number"] == 1
    assert "timestamp" in plan.decisions[0]
    print("decision-recording-ok")


def test_can_revert_to() -> None:
    """Test can_revert_to validation."""
    service = _make_service()
    plan_id = _create_plan(service)
    plan = service.get_plan(plan_id)

    # Strategize cannot revert to Apply or Action
    plan.phase = PlanPhase.STRATEGIZE
    plan.processing_state = ProcessingState.QUEUED
    assert not plan.can_revert_to(PlanPhase.APPLY)
    assert not plan.can_revert_to(PlanPhase.ACTION)

    # Execute can revert to Strategize
    plan.phase = PlanPhase.EXECUTE
    assert plan.can_revert_to(PlanPhase.STRATEGIZE)

    # Apply can revert to Strategize
    plan.phase = PlanPhase.APPLY
    plan.processing_state = ProcessingState.CONSTRAINED
    assert plan.can_revert_to(PlanPhase.STRATEGIZE)

    # Applied cannot revert
    plan.processing_state = ProcessingState.APPLIED
    assert not plan.can_revert_to(PlanPhase.STRATEGIZE)

    # Max reversions reached blocks reversion — set processing_state
    # first so the phase-state validator sees QUEUED (valid in any phase).
    plan.processing_state = ProcessingState.QUEUED
    plan.phase = PlanPhase.EXECUTE
    plan.reversion_count = 3
    assert not plan.can_revert_to(PlanPhase.STRATEGIZE)

    print("can-revert-to-ok")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "manual_revert"
    dispatch = {
        "manual_revert": test_manual_revert,
        "auto_revert_apply": test_auto_revert_apply,
        "loop_guard": test_loop_guard,
        "profile_gate": test_profile_gate,
        "terminal_block": test_terminal_block,
        "context_preserved": test_context_preserved,
        "decision_recording": test_decision_recording,
        "can_revert_to": test_can_revert_to,
    }
    fn = dispatch.get(cmd)
    if fn:
        fn()
    else:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)