AgentEvalTool/tests/unit/test_cron_pool_boundaries.py

"""Cron pool boundary tests (Gitea issue #4 / P0).

T4 — auto-scale oscillation across the busy/threshold boundary.
T9 — detect_stuck_crons threshold-edge behaviour.
"""
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock

import pytest
from sqlmodel import Session, select

from agenteval.intelligent_eval import cron_pool
from agenteval.intelligent_eval.cron_pool import (
    MAX_POOL_SIZE,
    MIN_POOL_SIZE,
    SCALE_DOWN_THRESHOLD,
    SCALE_UP_THRESHOLD,
    STUCK_THRESHOLD_MINUTES,
)
from agenteval.storage.db import OpenClawCronPoolDB, utc_now


@pytest.fixture()
def mock_openclaw_client():
    """Mock OpenClaw client (mirrors test_cron_pool.py)."""
    client = MagicMock(spec=__import__("agenteval.intelligent_eval.openclaw_client", fromlist=["OpenClawClient"]).OpenClawClient)
    counter = {"value": 0}

    async def create_cron_impl(**kwargs):
        counter["value"] += 1
        return f"cron-{counter['value']}"

    client.create_cron = AsyncMock(side_effect=create_cron_impl)
    client.delete_cron = AsyncMock()
    client.list_crons = AsyncMock(return_value=[])
    return client


# ---------------------------------------------------------------------------
# T4 — auto-scale threshold boundary + oscillation
# ---------------------------------------------------------------------------


def test_auto_scale_does_not_fire_at_exact_threshold(db_session: Session, mock_openclaw_client):
    """busy/total == SCALE_UP_THRESHOLD (0.8) must NOT trigger scale_up.

    Guard against the off-by-one of using >= vs >; today the implementation
    uses strict > so the boundary is inert.
    """
    import asyncio

    # 10 crons, 8 busy, 2 idle → 0.8 exactly, should NOT scale up.
    for i in range(10):
        status = "busy" if i < 8 else "idle"
        cron = OpenClawCronPoolDB(
            openclaw_cron_id=f"c-{i}", status=status, last_active_at=utc_now()
        )
        db_session.add(cron)
    db_session.commit()

    scaled_up, scaled_down = asyncio.run(
        cron_pool.auto_scale(db_session, mock_openclaw_client)
    )
    assert scaled_up == 0
    assert scaled_down == 0
    assert len(db_session.exec(select(OpenClawCronPoolDB)).all()) == 10


def test_auto_scale_oscillation_within_pool_bounds(db_session: Session, mock_openclaw_client):
    """Oscillating state around the threshold must not breach min/max.

    Asserts the invariants: under repeated auto_scale with busy/idle flipping
    across the 0.8 boundary, total never drops below MIN_POOL_SIZE and never
    exceeds MAX_POOL_SIZE. Per-call direction may flip (no hysteresis) — that
    itself is a finding recorded in .scratch/v111-architecture-scan.md §6.
    """
    import asyncio

    n = MIN_POOL_SIZE + 2  # start safely above min
    for i in range(n):
        db_session.add(
            OpenClawCronPoolDB(
                openclaw_cron_id=f"c-{i}", status="busy", last_active_at=utc_now()
            )
        )
    db_session.commit()

    # Flip busy count across the threshold 8 times and auto-scale after each flip.
    flips = [0.6, 0.9, 0.79, 0.95, 0.8, 0.85, 0.9, 0.7]  # ratios
    for ratio in flips:
        crons = db_session.exec(select(OpenClawCronPoolDB)).all()
        target_busy = int(round(len(crons) * ratio))
        for i, c in enumerate(crons):
            c.status = "busy" if i < target_busy else "idle"
        db_session.commit()
        asyncio.run(cron_pool.auto_scale(db_session, mock_openclaw_client))

    final = db_session.exec(select(OpenClawCronPoolDB)).all()
    total = len(final)
    assert total >= MIN_POOL_SIZE, f"pool shrank below min: {total}"
    assert total <= MAX_POOL_SIZE, f"pool exceeded max: {total}"


# ---------------------------------------------------------------------------
# T9 — detect_stuck_crons critical threshold edges
# ---------------------------------------------------------------------------


def test_detect_stuck_crons_critical_threshold_edges(db_session: Session, monkeypatch: pytest.MonkeyPatch):
    """Stuck-detection boundary at STUCK_THRESHOLD_MINUTES.

    Implementation uses strict `last_active_at < threshold`, so:
      - last_active_at == threshold → NOT stuck (boundary inert)
      - last_active_at = threshold - epsilon → stuck
      - last_active_at = threshold + epsilon (recent) → NOT stuck
    """
    # Anchor a single instant; all times derive from it so == is exact.
    now = utc_now()
    ten_min = timedelta(minutes=STUCK_THRESHOLD_MINUTES)

    # Exactly at threshold (10 min ago) — must NOT be detected as stuck.
    boundary = OpenClawCronPoolDB(
        openclaw_cron_id="boundary",
        status="busy",
        last_active_at=now - ten_min,
    )
    # Just past threshold (1s older) — must be stuck.
    just_stuck = OpenClawCronPoolDB(
        openclaw_cron_id="just-stuck",
        status="busy",
        last_active_at=now - ten_min - timedelta(seconds=1),
    )
    # Just inside window (1s newer) — must NOT be stuck.
    just_active = OpenClawCronPoolDB(
        openclaw_cron_id="just-active",
        status="busy",
        last_active_at=now - ten_min + timedelta(seconds=1),
    )
    db_session.add_all([boundary, just_stuck, just_active])
    db_session.commit()

    # Freeze clock to the same anchor so boundary == threshold is exact.
    monkeypatch.setattr(cron_pool, "utc_now", lambda: now)

    stuck = cron_pool.detect_stuck_crons(db_session)
    stuck_ids = {c.openclaw_cron_id for c in stuck}

    assert "just-stuck" in stuck_ids
    assert "just-active" not in stuck_ids
    # boundary: strict < means == is NOT stuck.
    assert "boundary" not in stuck_ids


def test_detect_stuck_crons_ignores_non_busy(db_session: Session):
    """Idle/stuck crons (even if stale) are not candidates for stuck-detection.

    The detection query filters `status == "busy"`; an idle cron with old
    last_active_at should not be picked up.
    """
    now = utc_now()
    db_session.add_all([
        OpenClawCronPoolDB(
            openclaw_cron_id="idle-stale",
            status="idle",
            last_active_at=now - timedelta(hours=1),
        ),
        OpenClawCronPoolDB(
            openclaw_cron_id="busy-stale",
            status="busy",
            last_active_at=now - timedelta(hours=1),
        ),
    ])
    db_session.commit()

    stuck = cron_pool.detect_stuck_crons(db_session)
    assert {c.openclaw_cron_id for c in stuck} == {"busy-stale"}