Coverage for agentos/tools/scheduler.py: 0%
238 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 23:53 +0800
1"""
2Scheduler — lightweight interval + cron-style job scheduler.
4Supports:
5 - Fixed interval scheduling (every N seconds)
6 - Delayed one-shot scheduling
7 - Cron-style scheduling (minute, hour, day of month, month, day of week)
8 - Job lifecycle (start, stop, pause, resume)
9 - Execution history / stats
10 - Thread-safe
11"""
13from __future__ import annotations
15import threading
16import time
17from collections.abc import Callable
18from dataclasses import dataclass, field
19from typing import Any
21# ============================================================================
22# Job
23# ============================================================================
26class JobState:
27 PENDING = "pending"
28 RUNNING = "running"
29 PAUSED = "paused"
30 STOPPED = "stopped"
33@dataclass
34class Job:
35 id: str
36 func: Callable[..., Any]
37 args: tuple = ()
38 kwargs: dict[str, Any] = field(default_factory=dict)
39 # Scheduling
40 interval: float | None = None # Fixed interval in seconds
41 cron: dict[str, str] | None = None # {"minute": "*", "hour": "*", ...}
42 delay: float | None = None # One-shot delay
43 # State
44 state: str = JobState.PENDING
45 next_run: float = 0.0
46 last_run: float | None = None
47 run_count: int = 0
48 error_count: int = 0
49 last_error: str | None = None
50 _timer: threading.Timer | None = field(default=None, repr=False)
53# ============================================================================
54# Cron Parser
55# ============================================================================
58def _cron_next(cron: dict[str, str], now: float) -> float:
59 """Compute next run time from cron expression. Returns timestamp."""
60 import calendar
62 t = time.localtime(now)
63 # Parse fields
64 minutes = _parse_cron_field(cron.get("minute", "*"), 0, 59)
65 hours = _parse_cron_field(cron.get("hour", "*"), 0, 23)
66 doms = _parse_cron_field(cron.get("day", "*"), 1, 31)
67 months = _parse_cron_field(cron.get("month", "*"), 1, 12)
68 dows = _parse_cron_field(cron.get("day_of_week", "*"), 0, 6)
70 # Search forward
71 year = t.tm_year
72 month = t.tm_mon
73 day = t.tm_mday
74 hour = t.tm_hour
75 minute = t.tm_min
77 for _ in range(366 * 5): # max 5 years search
78 # Check month
79 if month not in months:
80 month += 1
81 if month > 12:
82 month = 1
83 year += 1
84 day = 1
85 hour = 0
86 minute = 0
87 continue
89 # Check day (DOM and DOW)
90 if day > calendar.monthrange(year, month)[1]:
91 day = 1
92 month += 1
93 if month > 12:
94 month = 1
95 year += 1
96 hour = 0
97 minute = 0
98 continue
100 dow = calendar.weekday(year, month, day)
101 # Convert Monday=0..Sunday=6 to Sunday=0..Saturday=6
102 dow = (dow + 1) % 7
104 if day not in doms and dow not in dows:
105 day += 1
106 hour = 0
107 minute = 0
108 continue
110 # Check hour
111 if hour not in hours:
112 hour += 1
113 if hour > 23:
114 hour = 0
115 day += 1
116 minute = 0
117 continue
119 # Check minute
120 if minute not in minutes:
121 minute += 1
122 if minute > 59:
123 minute = 0
124 hour += 1
125 continue
127 # Found
128 result = time.mktime((year, month, day, hour, minute, 0, 0, 0, 0))
129 if result > now:
130 return result
131 # Advance past current instant
132 minute += 1
134 raise RuntimeError("No cron match within 5 years")
137def _parse_cron_field(field: str, lo: int, hi: int) -> set[int]:
138 """Parse cron field like '*' or '1,3,5' or '*/15'."""
139 if field == "*":
140 return set(range(lo, hi + 1))
141 result = set()
142 for part in field.split(","):
143 part = part.strip()
144 if "/" in part:
145 base, step = part.split("/")
146 base = lo if base == "*" else int(base)
147 step = int(step)
148 for v in range(base, hi + 1, step):
149 result.add(v)
150 elif "-" in part:
151 a, b = part.split("-")
152 result.update(range(int(a), int(b) + 1))
153 else:
154 result.add(int(part))
155 return result
158# ============================================================================
159# Scheduler
160# ============================================================================
163class Scheduler:
164 """Lightweight job scheduler.
166 Usage:
167 sched = Scheduler()
169 # Every 5 seconds
170 sched.every(5).do(my_func, arg1, arg2)
172 # Every minute
173 sched.cron("*/1 * * * *").do(my_func)
175 # One-shot delay
176 sched.delay(10).do(my_func)
178 sched.start()
179 """
181 def __init__(self):
182 self._jobs: dict[str, Job] = {}
183 self._lock = threading.RLock()
184 self._running = False
185 self._id_counter = 0
187 # ---------- Fluent API ----------
189 def every(self, seconds: float) -> Scheduler:
190 self._last_interval = seconds
191 self._last_cron = None
192 self._last_delay = None
193 return self
195 def cron(self, expression: str) -> Scheduler:
196 """Cron expression: 'minute hour day month day_of_week'"""
197 parts = expression.strip().split()
198 if len(parts) != 5:
199 raise ValueError(
200 "Cron expression must have 5 fields: 'minute hour day month day_of_week'"
201 )
202 self._last_cron = {
203 "minute": parts[0],
204 "hour": parts[1],
205 "day": parts[2],
206 "month": parts[3],
207 "day_of_week": parts[4],
208 }
209 self._last_interval = None
210 self._last_delay = None
211 return self
213 def delay(self, seconds: float) -> Scheduler:
214 self._last_delay = seconds
215 self._last_interval = None
216 self._last_cron = None
217 return self
219 def do(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Job:
220 """Register the job and return it."""
221 self._id_counter += 1
222 job_id = f"job_{self._id_counter}"
224 interval = getattr(self, "_last_interval", None)
225 cron = getattr(self, "_last_cron", None)
226 delay = getattr(self, "_last_delay", None)
228 now = time.time()
229 if interval:
230 next_run = now + interval
231 elif cron:
232 next_run = _cron_next(cron, now)
233 elif delay:
234 next_run = now + delay
235 else:
236 raise ValueError("Must call every(), cron(), or delay() before do()")
238 job = Job(
239 id=job_id,
240 func=func,
241 args=args,
242 kwargs=kwargs,
243 interval=interval,
244 cron=cron,
245 delay=delay,
246 next_run=next_run,
247 )
249 with self._lock:
250 self._jobs[job_id] = job
252 # Schedule if running
253 if self._running:
254 self._schedule_job(job)
256 return job
258 # ---------- Lifecycle ----------
260 def start(self) -> None:
261 with self._lock:
262 if self._running:
263 return
264 self._running = True
265 for job in self._jobs.values():
266 if job.state in (JobState.PENDING,):
267 self._schedule_job(job)
269 def stop(self) -> None:
270 with self._lock:
271 self._running = False
272 for job in self._jobs.values():
273 if job._timer:
274 job._timer.cancel()
275 job._timer = None
276 if job.state == JobState.RUNNING:
277 job.state = JobState.PAUSED
279 def pause(self, job_id: str) -> bool:
280 with self._lock:
281 job = self._jobs.get(job_id)
282 if not job:
283 return False
284 if job._timer:
285 job._timer.cancel()
286 job._timer = None
287 job.state = JobState.PAUSED
288 return True
290 def resume(self, job_id: str) -> bool:
291 with self._lock:
292 job = self._jobs.get(job_id)
293 if not job:
294 return False
295 if job.state != JobState.PAUSED:
296 return False
297 job.state = JobState.PENDING
298 job.next_run = time.time() # run immediately
299 if self._running:
300 self._schedule_job(job)
301 return True
303 def remove(self, job_id: str) -> bool:
304 with self._lock:
305 job = self._jobs.pop(job_id, None)
306 if job and job._timer:
307 job._timer.cancel()
308 return job is not None
310 # ---------- Internal ----------
312 def _schedule_job(self, job: Job) -> None:
313 if job.state == JobState.PAUSED:
314 return
315 job.state = JobState.PENDING
316 delay = max(0, job.next_run - time.time())
317 timer = threading.Timer(delay, self._run_job, args=[job.id])
318 timer.daemon = True
319 job._timer = timer
320 timer.start()
322 def _run_job(self, job_id: str) -> None:
323 with self._lock:
324 job = self._jobs.get(job_id)
325 if not job or not self._running:
326 return
327 job.state = JobState.RUNNING
329 try:
330 job.func(*job.args, **job.kwargs)
331 job.last_error = None
332 except Exception as e:
333 job.error_count += 1
334 job.last_error = str(e)
335 finally:
336 with self._lock:
337 job.last_run = time.time()
338 job.run_count += 1
340 if job.delay:
341 # One-shot — mark done
342 job.state = JobState.STOPPED
343 return
345 if job.state != JobState.PAUSED:
346 # Compute next run
347 if job.interval:
348 job.next_run = time.time() + job.interval
349 elif job.cron:
350 job.next_run = _cron_next(job.cron, time.time())
351 else:
352 job.state = JobState.STOPPED
353 return
355 if self._running:
356 self._schedule_job(job)
358 # ---------- Query ----------
360 def get_job(self, job_id: str) -> dict[str, Any] | None:
361 with self._lock:
362 job = self._jobs.get(job_id)
363 if not job:
364 return None
365 return self._job_info(job)
367 def list_jobs(self) -> list[dict[str, Any]]:
368 with self._lock:
369 return [self._job_info(j) for j in self._jobs.values()]
371 @staticmethod
372 def _job_info(job: Job) -> dict[str, Any]:
373 return {
374 "id": job.id,
375 "state": job.state,
376 "interval": job.interval,
377 "cron": job.cron,
378 "delay": job.delay,
379 "next_run": job.next_run,
380 "last_run": job.last_run,
381 "run_count": job.run_count,
382 "error_count": job.error_count,
383 "last_error": job.last_error,
384 }