handler/tests/test_schedules.py

"""Recurring agent spawns: the schedules API and the worker sweep that turns due
schedules into queued ``spawn`` commands with timestamped agent names."""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

from handler.control import worker
from handler.db import repository as repo
from handler.db.engine import get_engine


def _project(client, auth, tmp_path, project_id="p"):
    root = tmp_path / project_id
    root.mkdir(exist_ok=True)
    r = client.post(
        "/projects", json={"id": project_id, "root_dir": str(root)}, headers=auth
    )
    assert r.status_code == 201, r.text


CONTINUE_TASK = "Read @notes.md, continue from there; before finishing, overwrite that file."


def test_schedule_crud(client, auth, env, tmp_path):
    _project(client, auth, tmp_path)

    r = client.post(
        "/projects/p/schedules",
        json={"name_prefix": "nightly", "task": CONTINUE_TASK, "interval_seconds": 3600},
        headers=auth,
    )
    assert r.status_code == 201, r.text
    sched = r.json()
    assert sched["enabled"] is True
    assert sched["task"] == CONTINUE_TASK
    # First run fires on the worker's next pass.
    next_run = datetime.fromisoformat(sched["next_run_at"])
    assert next_run <= datetime.now(UTC) + timedelta(seconds=5)

    assert client.get("/schedules", headers=auth).json()[0]["id"] == sched["id"]
    assert client.get("/projects/p/schedules", headers=auth).json()[0]["id"] == sched["id"]

    r = client.patch(
        f"/schedules/{sched['id']}",
        json={"interval_seconds": 60, "enabled": False},
        headers=auth,
    )
    assert r.json()["interval_seconds"] == 60
    assert r.json()["enabled"] is False

    r = client.delete(f"/schedules/{sched['id']}", headers=auth)
    assert r.status_code == 200
    assert client.get("/schedules", headers=auth).json() == []


def test_schedule_unknown_project_404(client, auth, env):
    r = client.post(
        "/projects/ghost/schedules",
        json={"name_prefix": "x", "task": "t", "interval_seconds": 60},
        headers=auth,
    )
    assert r.status_code == 404


def test_schedule_validation_422(client, auth, env, tmp_path):
    _project(client, auth, tmp_path)
    r = client.post(
        "/projects/p/schedules",
        json={"name_prefix": "", "task": "t", "interval_seconds": 60},
        headers=auth,
    )
    assert r.status_code == 422
    r = client.post(
        "/projects/p/schedules",
        json={"name_prefix": "x", "task": "t", "interval_seconds": 1},
        headers=auth,
    )
    assert r.status_code == 422


def test_fire_due_schedules_enqueues_spawn(env, tmp_path):
    now = datetime.now(UTC)
    with get_engine().begin() as conn:
        repo.create_project(conn, "p", root_dir=str(tmp_path))
        sched = repo.create_schedule(
            conn,
            project_id="p",
            name_prefix="nightly",
            task=CONTINUE_TASK,
            interval_seconds=3600,
            next_run_at=now - timedelta(seconds=1),
            role="junior",
        )

    assert worker.fire_due_schedules(now) == 1

    with get_engine().begin() as conn:
        commands = repo.list_commands(conn)
        advanced = repo.get_schedule(conn, sched["id"])

    assert len(commands) == 1
    cmd = commands[0]
    assert cmd["type"] == "spawn"
    assert cmd["project_id"] == "p"
    assert cmd["agent_name"].startswith("nightly-")
    assert cmd["payload"]["task"] == CONTINUE_TASK
    assert cmd["payload"]["role"] == "junior"
    assert cmd["requested_by"] == f"schedule:{sched['id']}"

    # The schedule advanced: it will not re-fire until the next interval.
    assert advanced["last_run_at"] is not None
    assert advanced["next_run_at"] > now
    assert advanced["last_command_id"] == cmd["id"]
    assert worker.fire_due_schedules(now) == 0


def test_fire_skips_disabled_and_future(env, tmp_path):
    now = datetime.now(UTC)
    with get_engine().begin() as conn:
        repo.create_project(conn, "p", root_dir=str(tmp_path))
        repo.create_schedule(
            conn, project_id="p", name_prefix="off", task="t",
            interval_seconds=60, next_run_at=now - timedelta(seconds=1), enabled=False,
        )
        repo.create_schedule(
            conn, project_id="p", name_prefix="later", task="t",
            interval_seconds=60, next_run_at=now + timedelta(hours=1),
        )
    assert worker.fire_due_schedules(now) == 0


def test_missed_intervals_collapse_into_one_run(env, tmp_path):
    """A worker that was down for hours fires once, not once per missed interval."""
    now = datetime.now(UTC)
    with get_engine().begin() as conn:
        repo.create_project(conn, "p", root_dir=str(tmp_path))
        sched = repo.create_schedule(
            conn, project_id="p", name_prefix="hourly", task="t",
            interval_seconds=3600, next_run_at=now - timedelta(hours=10),
        )
    assert worker.fire_due_schedules(now) == 1
    with get_engine().begin() as conn:
        advanced = repo.get_schedule(conn, sched["id"])
    assert advanced["next_run_at"] == now + timedelta(hours=1)
    assert worker.fire_due_schedules(now) == 0