devplacepy/tests/unit/services/devii/agents/policy.py

# retoor <retoor@molodetz.nl>

from devplacepy.services.devii.agents.policy import (
    CATEGORIES,
    CHECK_INTERVAL_MAX_SECONDS,
    EFFORT_SCALE,
    MAX_ERROR_STREAK,
    NEEDS_USER_TIMEOUT_SECONDS,
    ORCHESTRATOR_ONLY_TOOLS,
    TERMINAL_STATUSES,
    decide,
    next_check_in,
)
from devplacepy.services.devii.registry import CATALOG


def _snapshot(**overrides):
    snap = {
        "status": "running",
        "age_seconds": 10,
        "idle_seconds": 5,
        "iterations": 2,
        "tokens": 1000,
        "checks": 1,
        "nudges_used": 0,
        "retries_used": 0,
        "error_streak": 0,
        "quota_ok": True,
        "failed": False,
        "answered": False,
        "progress": False,
        "effort": "standard",
    }
    snap.update(overrides)
    return snap


def test_terminal_statuses_always_settle():
    for category in CATEGORIES:
        for status in TERMINAL_STATUSES:
            decision, _ = decide(_snapshot(status=status), category)
            assert decision == "settle"


def test_spent_quota_cancels_first():
    for category in CATEGORIES:
        decision, reason = decide(_snapshot(quota_ok=False, progress=True), category)
        assert decision == "cancel"
        assert "quota" in reason


def test_budgets_cancel_at_every_scale():
    for category in CATEGORIES:
        entry = CATEGORIES[category]
        for effort, scale in EFFORT_SCALE.items():
            over_time = _snapshot(
                age_seconds=int(entry.time_budget_seconds * scale) + 1,
                effort=effort,
                progress=True,
            )
            assert decide(over_time, category)[0] == "cancel"
            over_iter = _snapshot(
                iterations=int(entry.iteration_budget * scale) + 1,
                effort=effort,
                progress=True,
            )
            assert decide(over_iter, category)[0] == "cancel"
            over_tokens = _snapshot(
                tokens=int(entry.token_budget * scale) + 1,
                effort=effort,
                progress=True,
            )
            assert decide(over_tokens, category)[0] == "cancel"
            at_edge = _snapshot(
                age_seconds=int(entry.time_budget_seconds * scale),
                iterations=int(entry.iteration_budget * scale),
                tokens=int(entry.token_budget * scale),
                effort=effort,
                progress=True,
            )
            assert decide(at_edge, category)[0] == "continue"


def test_error_streak_cancels():
    for category in CATEGORIES:
        snap = _snapshot(error_streak=MAX_ERROR_STREAK, progress=True)
        assert decide(snap, category)[0] == "cancel"
        snap = _snapshot(error_streak=MAX_ERROR_STREAK - 1, progress=True)
        assert decide(snap, category)[0] == "continue"


def test_failed_run_retries_then_cancels():
    for category in CATEGORIES:
        entry = CATEGORIES[category]
        snap = _snapshot(failed=True, retries_used=0)
        assert decide(snap, category)[0] == "retry_else"
        snap = _snapshot(failed=True, retries_used=entry.max_retries)
        decision, _ = decide(snap, category)
        assert decision == "cancel"


def test_needs_user_waits_answers_and_times_out():
    for category in CATEGORIES:
        waiting = _snapshot(status="needs_user")
        assert decide(waiting, category)[0] == "wait"
        answered = _snapshot(status="needs_user", answered=True)
        assert decide(answered, category)[0] == "continue"
        stale = _snapshot(
            status="needs_user", age_seconds=NEEDS_USER_TIMEOUT_SECONDS + 1
        )
        assert decide(stale, category)[0] == "cancel"


def test_progress_continues_and_grace_waits():
    for category in CATEGORIES:
        entry = CATEGORIES[category]
        assert decide(_snapshot(progress=True), category)[0] == "continue"
        assert (
            decide(_snapshot(idle_seconds=entry.grace_seconds), category)[0]
            == "wait"
        )


def test_stalled_runs_retry_then_nudge_then_cancel():
    for category in CATEGORIES:
        entry = CATEGORIES[category]
        stalled = {"idle_seconds": entry.grace_seconds + 1}
        first, _ = decide(_snapshot(**stalled), category)
        assert first == "retry_else"
        spent_retry = _snapshot(**stalled, retries_used=entry.max_retries)
        second, _ = decide(spent_retry, category)
        assert second == "nudge"
        spent_both = _snapshot(
            **stalled,
            retries_used=entry.max_retries,
            nudges_used=entry.max_nudges,
        )
        assert decide(spent_both, category)[0] == "cancel"


def test_next_check_in_grows_and_caps():
    assert next_check_in(0, 5) == 5
    assert next_check_in(1, 5) == 7
    assert next_check_in(2, 5) == 11
    assert next_check_in(100, 5) == CHECK_INTERVAL_MAX_SECONDS
    assert next_check_in(100, 600) == CHECK_INTERVAL_MAX_SECONDS


def test_denied_set_covers_user_contact_handlers():
    by_name = CATALOG.by_name()
    contact_handlers = {"client", "avatar", "interaction"}
    for name, action in by_name.items():
        if action.handler in contact_handlers:
            assert name in ORCHESTRATOR_ONLY_TOOLS
    for name in (
        "agent_start",
        "agent_status",
        "agent_cancel",
        "agent_result",
        "agent_list",
    ):
        assert name in ORCHESTRATOR_ONLY_TOOLS


def test_category_extras_name_real_tools():
    known = set(CATALOG.by_name())
    for entry in CATEGORIES.values():
        assert entry.fragment.strip()
        assert entry.time_budget_seconds > 0
        assert entry.iteration_budget > 0
        assert entry.token_budget > 0
        assert entry.check_interval_seconds > 0
        assert entry.grace_seconds > 0
        for tool in entry.extras:
            assert tool in known
            assert tool not in ORCHESTRATOR_ONLY_TOOLS