cleveragents-core/robot/helper_fix_then_revalidate.py

"""Helper script for Robot Framework fix-then-revalidate integration tests.

Self-contained Python helper that exercises the FixThenRevalidateOrchestrator
and its models, printing sentinel strings on success and exiting
with code 1 on failure.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

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

from cleveragents.application.services.fix_then_revalidate import (
    FixAttemptRecord,
    FixThenRevalidateOrchestrator,
    FixThenRevalidateResult,
)
from cleveragents.application.services.validation_pipeline import (
    ValidationPipeline,
    ValidationResult,
)
from cleveragents.domain.models.core.tool import ValidationMode

# ---------------------------------------------------------------------------
# Mock helpers
# ---------------------------------------------------------------------------


def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"passed": True, "message": f"{validation_name} ok"}


def _make_failed_result(
    name: str,
    mode: ValidationMode = ValidationMode.REQUIRED,
) -> ValidationResult:
    return ValidationResult(
        validation_name=name,
        resource_id="RES0001",
        resource_name="test-resource",
        mode=mode,
        passed=False,
        message=f"{name} failed",
        data={"hint": "check config"},
        duration_ms=10.0,
    )


def _always_fix(result: ValidationResult) -> str:
    return f"Fixed {result.validation_name}"


class _CountingRevalidate:
    def __init__(self, pass_after: int) -> None:
        self._pass_after = pass_after
        self._count = 0

    def __call__(self, result: ValidationResult) -> ValidationResult:
        self._count += 1
        passed = self._count >= self._pass_after
        return ValidationResult(
            validation_name=result.validation_name,
            resource_id=result.resource_id,
            resource_name=result.resource_name,
            mode=result.mode,
            passed=passed,
            message="passed" if passed else "still failing",
            data=result.data,
            duration_ms=5.0,
        )


class _NeverPassRevalidate:
    def __call__(self, result: ValidationResult) -> ValidationResult:
        return ValidationResult(
            validation_name=result.validation_name,
            resource_id=result.resource_id,
            resource_name=result.resource_name,
            mode=result.mode,
            passed=False,
            message="still failing",
            data=result.data,
            duration_ms=5.0,
        )


def _check(condition: bool, message: str) -> None:
    """Explicit check that works even with Python -O optimization."""
    if not condition:
        raise AssertionError(message)


# ---------------------------------------------------------------------------
# Sub-commands
# ---------------------------------------------------------------------------


def test_models() -> None:
    """Verify FixAttemptRecord and FixThenRevalidateResult models."""
    record = FixAttemptRecord(
        attempt_number=1,
        validation_name="check-syntax",
        fix_description="Adjusted indentation",
        success=True,
    )
    _check(record.attempt_number == 1, "attempt_number should be 1")
    _check(record.success is True, "success should be True")

    result = FixThenRevalidateResult(
        validation_attempts=2,
        final_passed=True,
        validation_fix_history=[record],
        escalated=False,
        terminal_failure=False,
    )
    _check(result.validation_attempts == 2, "validation_attempts should be 2")
    _check(result.final_passed is True, "final_passed should be True")
    _check(
        len(result.validation_fix_history) == 1,
        "validation_fix_history should have 1 record",
    )
    print("models-ok")


def test_fix_succeeds_first_attempt() -> None:
    """Fix succeeds on first attempt."""
    pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
    orchestrator = FixThenRevalidateOrchestrator(
        validation_pipeline=pipeline,
        max_retries=3,
    )
    failed = [_make_failed_result("check-syntax")]
    result = orchestrator.run_fix_loop(
        plan_id="PLAN001",
        failed_results=failed,
        fix_callback=_always_fix,
        revalidate_callback=_CountingRevalidate(pass_after=1),
    )
    _check(result.final_passed is True, "final_passed should be True")
    _check(result.validation_attempts == 1, "validation_attempts should be 1")
    _check(
        len(result.validation_fix_history) == 1,
        "validation_fix_history should have 1 record",
    )
    _check(
        result.validation_fix_history[0].success is True,
        "first fix history record should be success",
    )
    print("fix-first-ok")


def test_fix_succeeds_after_multiple() -> None:
    """Fix succeeds after multiple attempts."""
    pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
    orchestrator = FixThenRevalidateOrchestrator(
        validation_pipeline=pipeline,
        max_retries=3,
    )
    failed = [_make_failed_result("check-lint")]
    result = orchestrator.run_fix_loop(
        plan_id="PLAN002",
        failed_results=failed,
        fix_callback=_always_fix,
        revalidate_callback=_CountingRevalidate(pass_after=2),
    )
    _check(result.final_passed is True, "final_passed should be True")
    _check(result.validation_attempts == 2, "validation_attempts should be 2")
    print("fix-multiple-ok")


def test_terminal_failure() -> None:
    """Terminal failure when all attempts exhausted and no strategy revision."""
    pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
    orchestrator = FixThenRevalidateOrchestrator(
        validation_pipeline=pipeline,
        max_retries=3,
        auto_strategy_revision=1.0,
    )
    failed = [_make_failed_result("check-deploy")]
    result = orchestrator.run_fix_loop(
        plan_id="PLAN003",
        failed_results=failed,
        fix_callback=_always_fix,
        revalidate_callback=_NeverPassRevalidate(),
    )
    _check(result.final_passed is False, "final_passed should be False")
    _check(result.needs_user_escalation is True, "needs_user_escalation should be True")
    _check(result.escalated is False, "escalated should be False")
    print("terminal-failure-ok")


def test_escalation() -> None:
    """Escalation to strategy revision when auto_strategy_revision is 0.0."""
    pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
    orchestrator = FixThenRevalidateOrchestrator(
        validation_pipeline=pipeline,
        max_retries=3,
        auto_strategy_revision=0.0,
    )
    failed = [_make_failed_result("check-security")]
    result = orchestrator.run_fix_loop(
        plan_id="PLAN004",
        failed_results=failed,
        fix_callback=_always_fix,
        revalidate_callback=_NeverPassRevalidate(),
    )
    _check(result.final_passed is False, "final_passed should be False")
    _check(result.escalated is True, "escalated should be True")
    _check(result.terminal_failure is False, "terminal_failure should be False")
    print("escalation-ok")


def test_informational_skipped() -> None:
    """Informational validations are skipped by the fix loop."""
    pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
    orchestrator = FixThenRevalidateOrchestrator(
        validation_pipeline=pipeline,
        max_retries=3,
    )
    failed = [_make_failed_result("check-style", ValidationMode.INFORMATIONAL)]
    result = orchestrator.run_fix_loop(
        plan_id="PLAN005",
        failed_results=failed,
        fix_callback=_always_fix,
        revalidate_callback=_CountingRevalidate(pass_after=1),
    )
    _check(result.final_passed is True, "final_passed should be True")
    _check(result.validation_attempts == 0, "validation_attempts should be 0")
    _check(
        len(result.validation_fix_history) == 0,
        "validation_fix_history should be empty",
    )
    print("informational-skipped-ok")


def test_retry_counting() -> None:
    """Verify retry counting per validation per plan."""
    pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
    orchestrator = FixThenRevalidateOrchestrator(
        validation_pipeline=pipeline,
        max_retries=3,
    )
    _check(
        orchestrator.get_retry_count("P1", "check-a") == 0,
        "initial retry count should be 0",
    )
    failed = [_make_failed_result("check-a")]
    orchestrator.run_fix_loop(
        plan_id="P1",
        failed_results=failed,
        fix_callback=_always_fix,
        revalidate_callback=_CountingRevalidate(pass_after=2),
    )
    _check(
        orchestrator.get_retry_count("P1", "check-a") == 0,
        "retry count should be 0 after run (auto-reset)",
    )
    orchestrator.reset_retry_counts("P1")
    _check(
        orchestrator.get_retry_count("P1", "check-a") == 0,
        "retry count should be 0 after explicit reset (no-op)",
    )
    print("retry-counting-ok")


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

COMMANDS = {
    "models": test_models,
    "fix_first": test_fix_succeeds_first_attempt,
    "fix_multiple": test_fix_succeeds_after_multiple,
    "terminal_failure": test_terminal_failure,
    "escalation": test_escalation,
    "informational_skipped": test_informational_skipped,
    "retry_counting": test_retry_counting,
}


def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: helper_fix_then_revalidate.py <command>", file=sys.stderr)
        sys.exit(1)

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

    try:
        func()
    except Exception as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()