cleveragents-core/robot/helper_plan_preflight_guardrails.py

"""Helper script for Robot Framework plan preflight guardrails tests."""

from __future__ import annotations

import os
import sys
import tempfile
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from cleveragents.application.services.plan_preflight_guardrail import (
    PlanPreflightGuardrail,
    PreflightCheckName,
    PreflightRejection,
)


def _test_all_pass() -> None:
    """All 7 pre-flight checks pass with fully populated registries."""
    guardrail = PlanPreflightGuardrail()
    with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as tmp:
        pass
    try:
        report = guardrail.run_all_checks(
            action_name="test-action",
            action_registry={"test-action": {"name": "test-action"}},
            actor_registry={
                "strategy": "s",
                "execution": "e",
                "estimation": "est",
                "invariant_reconciliation": "ir",
            },
            tool_names=("tool-a",),
            tool_registry={"tool-a": {}},
            automation_profile={"name": "auto"},
            require_checkpoints=False,
            resource_paths=(tmp.name,),
            validation_names=("v1",),
            validation_registry={"v1": {}},
        )
        assert report.all_passed
        assert len(report.results) == 7
        print("all-pass-ok")
    finally:
        os.unlink(tmp.name)


def _test_action_missing() -> None:
    """PreflightRejection raised when action is missing."""
    guardrail = PlanPreflightGuardrail()
    try:
        guardrail.run_all_checks(
            action_name="no-such-action",
            action_registry={},
            actor_registry={
                "strategy": "s",
                "execution": "e",
                "estimation": "est",
                "invariant_reconciliation": "ir",
            },
            automation_profile={"name": "auto"},
        )
        raise AssertionError("Expected PreflightRejection")
    except PreflightRejection as exc:
        assert exc.check == PreflightCheckName.ACTION_SCHEMA
        assert "not found" in str(exc)
        print("action-missing-ok")


def _test_rollback_check() -> None:
    """Rollback feasibility check detects non-checkpointable tools."""
    guardrail = PlanPreflightGuardrail()
    result = guardrail.check_rollback_feasibility(
        require_checkpoints=True,
        tool_capabilities={"safe": True, "unsafe": False},
    )
    assert not result.passed
    assert "unsafe" in result.message

    # Skipped when not required
    result2 = guardrail.check_rollback_feasibility(
        require_checkpoints=False,
        tool_capabilities={"unsafe": False},
    )
    assert result2.passed
    assert "skipped" in result2.message
    print("rollback-check-ok")


def _test_estimation_warning() -> None:
    """Preflight emits warning for estimation actor without response_format."""
    guardrail = PlanPreflightGuardrail()
    report = guardrail.run_all_checks(
        action_name="test-action",
        action_registry={"test-action": {"name": "test-action"}},
        actor_registry={
            "strategy": {"name": "local/strategist"},
            "execution": {"name": "local/executor"},
            "estimation": {
                "name": "local/estimator",
                "context_view": "strategist",
            },
            "invariant_reconciliation": {"name": "local/reconciler"},
        },
        tool_names=("tool-a",),
        tool_registry={"tool-a": {}},
        automation_profile={"name": "auto"},
        require_checkpoints=False,
        validation_names=("v1",),
        validation_registry={"v1": {}},
    )
    assert report.all_passed
    assert any("response_format" in warning.message for warning in report.warnings)
    print("estimation-warning-ok")


def _test_estimation_context_view_warning() -> None:
    """Preflight emits warning for estimation actor with wrong context_view."""
    guardrail = PlanPreflightGuardrail()
    report = guardrail.run_all_checks(
        action_name="test-action",
        action_registry={"test-action": {"name": "test-action"}},
        actor_registry={
            "strategy": {"name": "local/strategist"},
            "execution": {"name": "local/executor"},
            "estimation": {
                "name": "local/estimator",
                "context_view": "executor",
                "response_format": {"type": "object"},
            },
            "invariant_reconciliation": {"name": "local/reconciler"},
        },
        tool_names=("tool-a",),
        tool_registry={"tool-a": {}},
        automation_profile={"name": "auto"},
        require_checkpoints=False,
        validation_names=("v1",),
        validation_registry={"v1": {}},
    )
    assert report.all_passed
    assert any("context_view" in warning.message for warning in report.warnings)
    print("estimation-context-view-warning-ok")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "all-pass"
    dispatch: dict[str, Any] = {
        "all-pass": _test_all_pass,
        "action-missing": _test_action_missing,
        "rollback-check": _test_rollback_check,
        "estimation-warning": _test_estimation_warning,
        "estimation-context-view-warning": _test_estimation_context_view_warning,
    }
    fn = dispatch.get(cmd)
    if fn:
        fn()
    else:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)