cleveragents-core/robot/helper_error_recovery.py

"""Robot Framework helper for error recovery smoke tests.

Provides a CLI-style interface for Robot to invoke error recovery
operations. Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_error_recovery.py classify
    python robot/helper_error_recovery.py record-error
    python robot/helper_error_recovery.py retry-policy
    python robot/helper_error_recovery.py format-output
    python robot/helper_error_recovery.py model-checks
"""

from __future__ import annotations

import sys
from pathlib import Path
from unittest.mock import MagicMock

# 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.insert(0, _SRC)

from cleveragents.application.services.error_recovery_service import (  # noqa: E402
    ErrorRecoveryService,
)
from cleveragents.domain.models.core.error_recovery import (  # noqa: E402
    ErrorCategory,
    ErrorHistory,
    ErrorRecord,
    RecoveryAction,
    RecoveryHint,
    classify_error,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
)

_PLAN_ID = "01ERRTEST000000000000001"


def _make_plan() -> MagicMock:
    plan = MagicMock()
    plan.identity.plan_id = _PLAN_ID
    plan.phase = PlanPhase.EXECUTE
    plan.processing_state = ProcessingState.ERRORED
    plan.is_terminal = False
    plan.error_message = None
    plan.error_details = None
    plan.timestamps = PlanTimestamps()
    return plan


def _make_service(
    plan: MagicMock,
    threshold: float = 0.0,
    max_retries: int = 3,
) -> ErrorRecoveryService:
    lifecycle = MagicMock()
    lifecycle.get_plan.return_value = plan
    lifecycle.commit_plan = MagicMock()
    return ErrorRecoveryService(
        lifecycle_service=lifecycle,
        auto_retry_threshold=threshold,
        max_retries=max_retries,
    )


def _cmd_classify() -> int:
    """Test error classification."""
    assert classify_error("Connection timed out") == ErrorCategory.TRANSIENT
    assert classify_error("Schema validation failed") == ErrorCategory.VALIDATION
    assert classify_error("Missing configuration") == ErrorCategory.CONFIGURATION
    assert classify_error("Merge conflict in file") == ErrorCategory.MERGE_CONFLICT
    assert classify_error("Authentication failed") == ErrorCategory.AUTHENTICATION
    assert classify_error("Model not available") == ErrorCategory.PROVIDER
    assert classify_error("Unexpected internal error") == ErrorCategory.INTERNAL
    assert classify_error("Something novel") == ErrorCategory.UNKNOWN
    # Exception type classification
    assert classify_error("fail", "RateLimitError") == ErrorCategory.TRANSIENT
    print("classify-ok")
    return 0


def _cmd_record_error() -> int:
    """Test error recording."""
    plan = _make_plan()
    svc = _make_service(plan)
    record = svc.record_error(
        plan_id=_PLAN_ID,
        phase="execute",
        message="Connection timed out after 30s",
    )
    assert record.category == ErrorCategory.TRANSIENT
    assert record.phase == "execute"
    assert len(record.recovery_hints) > 0

    history = svc.get_error_history(_PLAN_ID)
    assert history.total_errors == 1

    print("record-error-ok")
    return 0


def _cmd_retry_policy() -> int:
    """Test retry policy decisions."""
    plan = _make_plan()
    # Auto-retry enabled
    svc = _make_service(plan, threshold=0.0)
    svc.record_error(_PLAN_ID, "execute", "Connection timed out")
    assert svc.should_retry(_PLAN_ID), "Expected should_retry=True"

    # Auto-retry disabled (human threshold)
    svc2 = _make_service(plan, threshold=1.0)
    svc2.record_error(_PLAN_ID, "execute", "Connection timed out")
    assert svc2.should_escalate(_PLAN_ID), "Expected should_escalate=True"

    print("retry-policy-ok")
    return 0


def _cmd_format_output() -> int:
    """Test error output formatting."""
    plan = _make_plan()
    svc = _make_service(plan)
    svc.record_error(_PLAN_ID, "execute", "Provider timed out")

    plain = svc.format_error_output(_PLAN_ID, fmt="plain")
    assert "transient" in plain
    assert "Provider timed out" in plain

    json_out = svc.format_error_output(_PLAN_ID, fmt="json")
    assert "plan_id" in json_out

    print("format-output-ok")
    return 0


def _cmd_model_checks() -> int:
    """Test model construction and enums."""
    # ErrorCategory values
    assert ErrorCategory.TRANSIENT == "transient"
    assert ErrorCategory.VALIDATION == "validation"
    assert ErrorCategory.UNKNOWN == "unknown"

    # RecoveryAction values
    assert RecoveryAction.RETRY == "retry"
    assert RecoveryAction.CANCEL == "cancel"

    # RecoveryHint construction
    hint = RecoveryHint(
        action=RecoveryAction.RETRY,
        message="Retry the operation",
        cli_command="agents plan prompt TEST",
        priority=0,
    )
    assert hint.action == RecoveryAction.RETRY

    # ErrorRecord construction and properties
    record = ErrorRecord(
        error_id="ERR-001",
        phase="execute",
        category=ErrorCategory.TRANSIENT,
        message="test",
        retry_count=0,
        max_retries=3,
    )
    assert record.is_retriable
    assert not record.retries_exhausted

    # ErrorHistory construction
    history = ErrorHistory(plan_id="TEST")
    history.add(record)
    assert history.total_errors == 1
    assert history.latest == record

    # to_cli_dict
    cli_dict = record.to_cli_dict()
    assert "error_id" in cli_dict
    assert "category" in cli_dict

    print("model-checks-ok")
    return 0


def main() -> int:
    if len(sys.argv) < 2:
        print(
            "Usage: helper_error_recovery.py "
            "<classify|record-error|retry-policy|format-output|model-checks>"
        )
        return 1

    cmd = sys.argv[1]
    commands = {
        "classify": _cmd_classify,
        "record-error": _cmd_record_error,
        "retry-policy": _cmd_retry_policy,
        "format-output": _cmd_format_output,
        "model-checks": _cmd_model_checks,
    }
    func = commands.get(cmd)
    if func is None:
        print(f"Unknown command: {cmd}")
        return 1
    return func()


if __name__ == "__main__":
    sys.exit(main())