"""Scheduled prompts: cron parsing and offsets, ScheduleWakeup's clamp and stop, cron jobs that fire and re-arm, ``/loop`` expansion at schedule time, and the one fallback wake-up of a loop iteration that forgot to re-arm.""" from __future__ import annotations import time from collections.abc import AsyncIterator, Callable, Sequence from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import pytest from noeta.backends.stores.sqlite import SqliteStore from noeta.core.agent import Agent from noeta.core.client import Client from noeta.core.events import Event, ExtensionRecorded, InputTaken, TimerFired, ToolFinished from noeta.core.loop.state import CONFIG_SIGNAL, PLUGINS_OFF from noeta.core.runner import Runtime from noeta.core.store import Appended, Claim, NewTask, Patch, Post from noeta.core.tool import ToolContext, ToolOutput, text_output from noeta.plugins.schedule import JOB_TIMER, WAKEUP_TIMER, Schedule, cron, texts from noeta.testing import ScriptedProvider, reply # ------------------------------------------------------------ cron def test_parse_accepts_lists_ranges_steps_and_sunday_as_seven() -> None: c = cron.parse("*/15 9-17 * * 1-5") assert c is not None and c.minute == frozenset({0, 15, 30, 45}) assert c.hour == frozenset(range(9, 18)) and c.weekday == frozenset(range(1, 6)) sunday = cron.parse("0 9 * * 7") assert sunday is not None and sunday.weekday == frozenset({0}) for bad in ("* * * *", "60 * * * *", "* * 0 * *", "5-1 * * * *", "*/0 * * * *", "a * * * *"): assert cron.parse(bad) is None, bad def test_next_match_is_the_next_local_minute_that_matches() -> None: start = datetime(2026, 10, 28, 9, 7, 30).timestamp() nxt = cron.next_match("30 14 * * *", start) assert nxt is not None and datetime.fromtimestamp(nxt) == datetime(2026, 10, 28, 14, 30) nxt = cron.next_match("0 9 * * 1", start) # 2026-10-28 is a Wednesday assert nxt is not None and datetime.fromtimestamp(nxt) == datetime(2026, 11, 2, 9, 0) assert cron.next_match("0 0 31 2 *", start) is None # Feb 31 never comes def test_offsets_late_for_recurring_early_for_one_shot_on_the_half_hour() -> None: start = datetime(2026, 10, 28, 9, 7, 30).timestamp() hourly = cron.next_fire("7 * * * *", start, "80000000", recurring=True) # half the job's fraction (0.5) of the one-hour period: 15 minutes late assert hourly == datetime(2026, 10, 28, 10, 7).timestamp() + 900 capped = cron.next_fire("7 9 * * *", start, "ffffffff", recurring=True) assert capped is not None and capped - datetime(2026, 10, 29, 9, 7).timestamp() == 1800 once = cron.next_fire("30 14 28 10 *", start, "80000000", recurring=False) assert once == datetime(2026, 10, 28, 14, 30).timestamp() - 45 off = cron.next_fire("31 14 28 10 *", start, "80000000", recurring=False) assert off == datetime(2026, 10, 28, 14, 31).timestamp() five = cron.next_fire("*/5 * * * *", start, "00000000", recurring=True) assert five == start + 300 - cron.CACHE_LEAD_SECONDS def test_describe_reads_the_common_shapes() -> None: assert cron.describe("*/5 * * * *") == "Every 5 minutes" assert cron.describe("7 * * * *") == "Every hour at :07" assert cron.describe("0 */2 * * *") == "Every 2 hours" assert cron.describe("57 8 * * *") == "Every day at 8:57 AM" assert cron.describe("0 9 * * 1-5") == "Weekdays at 9:00 AM" assert cron.describe("0 9 * * 0") == "Every Sunday at 9:00 AM" assert cron.describe("0 9 1 * *") == "0 9 1 * *" # ------------------------------------------------------------ in a task @pytest.fixture async def store(tmp_path: Path) -> AsyncIterator[SqliteStore]: s = SqliteStore(str(tmp_path / "t.db")) yield s await s.close() async def _no_sleep(_: float) -> None: return None def runtime(store: SqliteStore, provider: ScriptedProvider, ext: Schedule) -> Runtime: agent = Agent(name="a", model="m", extensions=[ext]) return Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider, env_for=lambda _s: None, sleep=_no_sleep) # type: ignore[arg-type,return-value] def _results(events: list[object]) -> list[str]: return [e.result.text for e in events if isinstance(e, ToolFinished)] async def test_wakeup_clamps_expands_the_loop_and_fires_when_due(store: SqliteStore) -> None: past = time.time() - 10_000 # every wake-up computed from it is already due provider = ScriptedProvider([ reply("", (texts.WAKEUP_TOOL, {"delaySeconds": 5, "reason": "watching CI", "prompt": "/loop check the deploy", "noop": True})), reply("waiting"), reply("", (texts.WAKEUP_TOOL, {"stop": True})), reply("stopped"), ]) rt = runtime(store, provider, Schedule(clock=lambda: past)) client = Client(store) tid = await client.create("a", "/loop check the deploy") await rt.run_until_idle() events = [r.event for r in await client.events(tid)] first = _results(events)[0] assert "(in " in first and "clamped to 60s" in first # The fired prompt is the /loop command's text, expanded when scheduled. fired = provider.requests[2].messages[-1].text assert fired.startswith("# /loop") and fired.endswith("check the deploy") stopped = _results(events)[-1] assert stopped.startswith("Loop stopped: no wake-up was pending") assert not any(isinstance(e, ExtensionRecorded) and e.kind == "timer" and e.data["name"] == WAKEUP_TIMER and e.data["at"] > time.time() for e in events) async def test_wakeup_validates_the_fields_it_needs(store: SqliteStore) -> None: provider = ScriptedProvider([ reply("", (texts.WAKEUP_TOOL, {"reason": "x", "prompt": "p", "noop": True}), (texts.WAKEUP_TOOL, {"delaySeconds": 60, "reason": "x", "noop": True}), (texts.WAKEUP_TOOL, {"delaySeconds": 60, "reason": "x", "prompt": "p"})), reply("done"), ]) rt = runtime(store, provider, Schedule()) client = Client(store) tid = await client.create("a", "go") await rt.run_until_idle() assert _results([r.event for r in await client.events(tid)]) == [ texts.WAKEUP_NEEDS_DELAY, texts.WAKEUP_NEEDS_PROMPT, texts.WAKEUP_NEEDS_NOOP] async def test_an_autonomous_tick_that_forgets_to_rearm_gets_one_fallback( store: SqliteStore) -> None: past = time.time() - 10_000 provider = ScriptedProvider([ reply("", (texts.WAKEUP_TOOL, {"delaySeconds": 600, "reason": "idle", "prompt": texts.DYNAMIC_SENTINEL, "noop": True})), reply("armed"), reply("tick without re-arming"), # the wake-up fires; no ScheduleWakeup reply("fallback tick"), # the keepalive fires once ]) rt = runtime(store, provider, Schedule(clock=lambda: past)) client = Client(store) tid = await client.create("a", "/loop") await rt.run_until_idle() assert provider.requests[2].messages[-1].text == texts.DYNAMIC_TICK assert provider.requests[3].messages[-1].text == texts.DYNAMIC_TICK assert not provider.steps # and no third tick: the fallback is armed once row = await client.get(tid) assert row is not None and row.ready_at is None async def test_cron_jobs_create_list_fire_rearm_and_delete(store: SqliteStore) -> None: provider = ScriptedProvider([ reply("", ("ToolSearch", {"query": "select:CronCreate,CronList,CronDelete"})), reply("", (texts.CRON_CREATE_TOOL, {"cron": "*/5 * * * *", "prompt": "check CI"}), (texts.CRON_CREATE_TOOL, {"cron": "31 14 28 2 *", "prompt": "once", "recurring": False}), (texts.CRON_CREATE_TOOL, {"cron": "bad", "prompt": "x"})), reply("", (texts.CRON_LIST_TOOL, {})), reply("scheduled"), ]) ext = Schedule() rt = runtime(store, provider, ext) client = Client(store) tid = await client.create("a", "set it up") await rt.run_until_idle() events = [r.event for r in await client.events(tid)] results = _results(events) assert results[1].startswith("Scheduled recurring job ") assert results[2].startswith("Scheduled one-shot job ") and "deletes itself" in results[2] assert results[3] == texts.CRON_INVALID.format(cron="bad") listing = results[4].splitlines() assert len(listing) == 2 and "Every 5 minutes (recurring)" in listing[0] assert "(one-shot)" in listing[1] and listing[1].endswith(": once") row = await client.get(tid) assert row is not None and row.ready_at is not None and row.ready_at > time.time() jobs = (await _state(client, tid, ext))["jobs"] recurring = next(j for j in jobs.values() if j["recurring"]) # Firing (as the runner would on the due timer) re-arms a recurring job. provider.add(reply("checked"), reply("", (texts.CRON_DELETE_TOOL, {"id": recurring["id"]}), (texts.CRON_DELETE_TOOL, {"id": "nope"})), reply("gone")) await store.post(Post(task_id=tid, body=TimerFired(name=JOB_TIMER + recurring["id"]))) await rt.run_until_idle() assert provider.requests[4].messages[-1].text == "check CI" after = (await _state(client, tid, ext))["jobs"][recurring["id"]] assert after["next_at"] > recurring["next_at"] - 1 await client.send(tid, "cancel it") await rt.run_until_idle() results = _results([r.event for r in await client.events(tid)]) assert results[-2:] == [texts.CRON_DELETED.format(id=recurring["id"]), texts.CRON_UNKNOWN.format(id="nope")] assert recurring["id"] not in (await _state(client, tid, ext))["jobs"] def test_next_match_reads_the_expression_in_a_given_zone() -> None: zone = ZoneInfo("Pacific/Kiritimati") # UTC+14, rarely a machine's own start = datetime(2026, 10, 28, 9, 7, 30, tzinfo=zone).timestamp() nxt = cron.next_match("30 14 * * *", start, zone) assert nxt is not None assert datetime.fromtimestamp(nxt, zone) == datetime(2026, 10, 28, 14, 30, tzinfo=zone) async def test_a_cron_job_runs_in_the_tasks_time_zone(store: SqliteStore) -> None: """The task config's ``timezone`` (the person's), not the worker's.""" provider = ScriptedProvider([ reply("", ("ToolSearch", {"query": "select:CronCreate"})), reply("", (texts.CRON_CREATE_TOOL, {"cron": "1 9 * * *", "prompt": "standup", "recurring": False})), reply("scheduled"), ]) ext = Schedule() rt = runtime(store, provider, ext) client = Client(store) tid = await client.create("a", "remind me", config={"timezone": "Pacific/Kiritimati"}) await rt.run_until_idle() jobs = (await _state(client, tid, ext))["jobs"] assert isinstance(jobs, dict) [job] = jobs.values() at = datetime.fromtimestamp(job["next_at"], ZoneInfo("Pacific/Kiritimati")) assert (at.hour, at.minute, at.second) == (9, 1, 0) assert "+14" in _results([r.event for r in await client.events(tid)])[1] async def test_a_continued_task_does_not_inherit_its_sources_jobs(store: SqliteStore) -> None: provider = ScriptedProvider([ reply("", ("ToolSearch", {"query": "select:CronCreate,CronList"})), reply("", (texts.CRON_CREATE_TOOL, {"cron": "*/5 * * * *", "prompt": "check CI"})), reply("scheduled"), reply("", (texts.CRON_LIST_TOOL, {})), reply("listed"), ]) ext = Schedule() rt = runtime(store, provider, ext) client = Client(store) src = await client.create("a", "set it up") await rt.run_until_idle() assert (await _state(client, src, ext))["jobs"] fork = await client.create("a", "what is scheduled here?", history_from=src) await rt.run_until_idle() assert _results([r.event for r in await client.events(fork)]) == [texts.CRON_NONE] async def _state(client: Client, tid: str, ext: Schedule) -> dict[str, object]: st: object = ext.initial_state() for r in await client.events(tid): e = r.event if isinstance(e, ExtensionRecorded) and e.extension == "schedule": st = ext.reduce(st, e.kind, e.data) assert isinstance(st, dict) return st async def test_loop_is_a_command_front_ends_list() -> None: from noeta.core.commands import expand_command, list_commands from noeta.core.extension import CommandContext agent = Agent(name="a", model="m", extensions=[Schedule()]) ctx = CommandContext(cwd="/tmp") assert [c.name for c in await list_commands(agent, ctx)] == ["loop"] text = await expand_command(agent, ctx, "/loop 5m /babysit") assert text is not None and text.startswith("# /loop") and text.endswith("5m /babysit") # ------------------------------------------------------------ a crash mid-firing class Crash(BaseException): """Stands in for the process dying.""" class CrashAfterFiring(SqliteStore): """Dies right after committing the first append that takes a fired timer (once ``armed``): the firing is in the log, its follow-up is not.""" armed = False async def append(self, claim: Claim, events: Sequence[Event], *, expected_seq: int, patch: Patch = Patch(), posts: Sequence[Post] = (), creates: Sequence[NewTask] = ()) -> Appended: done = await super().append(claim, events, expected_seq=expected_seq, patch=patch, posts=posts, creates=creates) if self.armed and any(isinstance(e, InputTaken) and any( isinstance(i.body, TimerFired) for i in e.items) for e in events): self.armed = False raise Crash() return done async def _crash_then_recover(path: str, crashing: CrashAfterFiring, tid: str, make: Callable[[SqliteStore], Runtime]) -> SqliteStore: """Run until the armed crash, then pick the task up in a new process.""" with pytest.raises(Crash): await make(crashing).run_until_idle() await crashing.close() store = SqliteStore(path) def tx(conn: Any) -> None: conn.execute("UPDATE tasks SET lease_until = 0 WHERE task_id = ?", (tid,)) await store._run(tx) await make(store).run_until_idle() return store async def test_a_crash_after_a_cron_job_fires_does_not_drop_it(tmp_path: Path) -> None: path = str(tmp_path / "t.db") crashing = CrashAfterFiring(path) provider = ScriptedProvider([ reply("", ("ToolSearch", {"query": "select:CronCreate"})), reply("", (texts.CRON_CREATE_TOOL, {"cron": "*/5 * * * *", "prompt": "check CI"}), (texts.CRON_CREATE_TOOL, {"cron": "31 14 28 2 *", "prompt": "once", "recurring": False})), reply("scheduled"), reply("checked"), ]) ext = Schedule() client = Client(crashing) tid = await client.create("a", "set it up") await runtime(crashing, provider, ext).run_until_idle() jobs = (await _state(client, tid, ext))["jobs"] assert isinstance(jobs, dict) and len(jobs) == 2 crashing.armed = True for job_id in jobs: await crashing.post(Post(task_id=tid, body=TimerFired(name=JOB_TIMER + job_id))) store = await _crash_then_recover(path, crashing, tid, lambda s: runtime(s, provider, ext)) after = (await _state(Client(store), tid, ext))["jobs"] assert isinstance(after, dict) recurring = next(j for j in jobs.values() if j["recurring"]) assert list(after) == [recurring["id"]] # the one-shot job is done row = await store.get(tid) assert row is not None and row.ready_at is not None and row.ready_at > time.time() await store.close() async def test_a_crash_after_a_loop_wakeup_fires_keeps_its_fallback(tmp_path: Path) -> None: path = str(tmp_path / "t.db") crashing = CrashAfterFiring(path) crashing.armed = True past = time.time() - 10_000 provider = ScriptedProvider([ reply("", (texts.WAKEUP_TOOL, {"delaySeconds": 600, "reason": "idle", "prompt": texts.DYNAMIC_SENTINEL, "noop": True})), reply("armed"), reply("tick without re-arming"), # the wake-up fired before the crash reply("fallback tick"), # the keepalive still fires once ]) ext = Schedule(clock=lambda: past) tid = await Client(crashing).create("a", "/loop") store = await _crash_then_recover(path, crashing, tid, lambda s: runtime(s, provider, ext)) assert provider.requests[3].messages[-1].text == texts.DYNAMIC_TICK assert not provider.steps await store.close() class _SwitchScheduleOff: name = "SwitchOff" description = "Switches the schedule plugin off for the task." input_schema = {"type": "object"} read_only = True clearable = False def __init__(self, client: Client) -> None: self.client = client async def run(self, input: dict[str, Any], ctx: ToolContext) -> ToolOutput: await self.client.signal(ctx.task_id, CONFIG_SIGNAL, {PLUGINS_OFF: ["schedule"]}) return text_output("off") async def test_a_job_due_while_schedule_is_off_waits_and_rearms_once_back( store: SqliteStore) -> None: """A switched-off schedule's timers neither run their prompt nor wake the task, also one already posted; switched back on, the due job runs once and is re-armed as a late firing would be.""" clock = [time.time() - 10_000] # the job's first firing is long due client = Client(store) provider = ScriptedProvider([ reply("", ("ToolSearch", {"query": "select:CronCreate"})), reply("", (texts.CRON_CREATE_TOOL, {"cron": "*/5 * * * *", "prompt": "check CI"}), ("SwitchOff", {})), reply("scheduled"), ]) ext = Schedule(clock=lambda: clock[0]) agent = Agent(name="a", model="m", extensions=[ext], tools=[_SwitchScheduleOff(client)]) rt = Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider, # type: ignore[arg-type,return-value] env_for=lambda _s: None, sleep=_no_sleep) # type: ignore[arg-type,return-value] tid = await client.create("a", "set it up") await rt.run_until_idle() jobs = (await _state(client, tid, ext))["jobs"] assert isinstance(jobs, dict) [job] = jobs.values() assert job["next_at"] < time.time() and len(provider.requests) == 3 row = await client.get(tid) assert row.status == "waiting" and row.ready_at is None # A firing posted before the switch is taken while it is off: held too. await store.post(Post(task_id=tid, body=TimerFired(name=JOB_TIMER + job["id"]))) await rt.run_until_idle() assert len(provider.requests) == 3 clock[0] = time.time() provider.add(reply("checked")) await client.signal(tid, CONFIG_SIGNAL, {PLUGINS_OFF: None}) await rt.run_until_idle() assert len(provider.requests) == 4 assert provider.requests[3].messages[-1].text == "check CI" jobs = (await _state(client, tid, ext))["jobs"] assert isinstance(jobs, dict) and jobs[job["id"]]["next_at"] > time.time() row = await client.get(tid) assert row.ready_at is not None and row.ready_at > time.time()