cleveragents-core/robot/helper_decision_phase_gating.py

"""Helper script for Robot Framework decision phase-gating smoke tests.

Usage:
    python robot/helper_decision_phase_gating.py <subcommand>

Subcommands:
    valid-strategize     Record valid strategize-phase types
    valid-execute        Record valid execute-phase types
    invalid-strategize   Reject execute-only types in strategize
    invalid-execute      Reject strategize-only types in execute
    phase-agnostic       Phase-agnostic types work in both phases
    error-attributes     DecisionPhaseViolationError attributes
"""

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.decision_service import DecisionService
from cleveragents.core.exceptions import DecisionPhaseViolationError
from cleveragents.domain.models.core.decision import (
    EXECUTE_TYPES,
    STRATEGIZE_TYPES,
    DecisionType,
)

_PLAN_ID = "01HV000000000000000000PG01"


def _record(
    svc: DecisionService,
    dtype: DecisionType,
    phase: str,
    *,
    parent: str | None = "01HV000000000000000000PP01",
) -> None:
    """Record a decision with phase-gating."""
    if dtype == DecisionType.PROMPT_DEFINITION:
        parent = None
    svc.record_decision(
        plan_id=_PLAN_ID,
        decision_type=dtype,
        question=f"Q ({dtype})",
        chosen_option=f"A ({dtype})",
        parent_decision_id=parent,
        plan_phase=phase,
    )


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def _valid_strategize() -> None:
    svc = DecisionService()
    for dt in STRATEGIZE_TYPES:
        _record(svc, dt, "strategize")
    print("valid-strategize-ok")


def _valid_execute() -> None:
    svc = DecisionService()
    for dt in EXECUTE_TYPES:
        _record(svc, dt, "execute")
    print("valid-execute-ok")


def _invalid_strategize() -> None:
    svc = DecisionService()
    execute_only = EXECUTE_TYPES - STRATEGIZE_TYPES
    for dt in execute_only:
        try:
            _record(svc, dt, "strategize")
            raise AssertionError(
                f"Expected DecisionPhaseViolationError for {dt} in strategize"
            )
        except DecisionPhaseViolationError as exc:
            assert exc.decision_type == str(dt)
            assert exc.plan_phase == "strategize"
    print("invalid-strategize-ok")


def _invalid_execute() -> None:
    svc = DecisionService()
    strategize_only = STRATEGIZE_TYPES - EXECUTE_TYPES
    for dt in strategize_only:
        try:
            _record(svc, dt, "execute")
            raise AssertionError(
                f"Expected DecisionPhaseViolationError for {dt} in execute"
            )
        except DecisionPhaseViolationError as exc:
            assert exc.decision_type == str(dt)
            assert exc.plan_phase == "execute"
    print("invalid-execute-ok")


def _phase_agnostic() -> None:
    svc = DecisionService()
    agnostic = STRATEGIZE_TYPES & EXECUTE_TYPES
    assert len(agnostic) > 0, "No phase-agnostic types found"
    for dt in agnostic:
        _record(svc, dt, "strategize")
        _record(svc, dt, "execute")
    print("phase-agnostic-ok")


def _error_attributes() -> None:
    svc = DecisionService()
    try:
        _record(svc, DecisionType.TOOL_INVOCATION, "strategize")
        raise AssertionError("Expected DecisionPhaseViolationError")
    except DecisionPhaseViolationError as exc:
        assert exc.decision_type == "tool_invocation"
        assert exc.plan_phase == "strategize"
        assert isinstance(exc.allowed_types, frozenset)
        assert len(exc.allowed_types) > 0
        assert "tool_invocation" not in exc.allowed_types
        msg = str(exc)
        assert "tool_invocation" in msg
        assert "strategize" in msg
        assert "Allowed types" in msg
    print("error-attributes-ok")


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS = {
    "valid-strategize": _valid_strategize,
    "valid-execute": _valid_execute,
    "invalid-strategize": _invalid_strategize,
    "invalid-execute": _invalid_execute,
    "phase-agnostic": _phase_agnostic,
    "error-attributes": _error_attributes,
}


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
    command = sys.argv[1]
    handler = _COMMANDS.get(command)
    if handler is None:
        raise SystemExit(f"Unknown command: {command}")
    handler()


if __name__ == "__main__":
    main()