feat: add cron trigger

Author: naaive

openagent/agent/config.py

@@ -35,11 +35,12 @@ def validate_celery_urls(cls, v, values):
 
 
 class TaskConfig(BaseModel):
-    interval: int = Field(description="Interval in seconds between task executions")
-    delay_variation: int = Field(
+    interval: Optional[int] = Field(default=None, description="Interval in seconds between task executions")
+    delay_variation: Optional[int] = Field(
         default=0, description="Maximum random delay in seconds to add to the interval"
     )
     query: str
+    cron: Optional[str] = Field(default=None, description="Cron expression for scheduling tasks")
     schedule: SchedulerConfig = Field(
         default_factory=lambda: SchedulerConfig(type="local"),
         description="Scheduler configuration for this task",
@@ -48,14 +49,14 @@ class TaskConfig(BaseModel):
     @classmethod
     @field_validator("interval")
     def validate_interval(cls, v):
-        if v < 1:
+        if v is not None and v < 1:
             raise ValueError("Interval must be greater than 1 second")
         return v
 
     @classmethod
     @field_validator("delay_variation")
     def validate_delay_variation(cls, v):
-        if v < 0:
+        if v is not None and v < 0:
             raise ValueError("Delay variation must be non-negative")
         return v
 

openagent/agent/scheduler.py

@@ -8,6 +8,7 @@
 from celery.beat import PersistentScheduler
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.interval import IntervalTrigger
+from apscheduler.triggers.cron import CronTrigger
 
 from openagent.agent.config import TaskConfig
 
@@ -32,24 +33,33 @@ def init_scheduled_tasks(self, tasks_config: dict, task_runner):
             if task_config.schedule.type == "queue":
                 self._init_celery_task(task_id, task_config, task_runner)
             else:
-                # Default to local scheduler
                 async def task_wrapper(query, delay_variation):
                     if delay_variation > 0:
                         delay = random.uniform(0, delay_variation)
                         logger.debug(f"Task '{task_id}' sleeping for {delay} seconds")
                         await asyncio.sleep(delay)
                     await task_runner(query)
 
-                self.scheduler.add_job(
-                    func=task_wrapper,
-                    trigger=IntervalTrigger(seconds=task_config.interval),
-                    args=[task_config.query, task_config.delay_variation],
-                    id=task_id,
-                    name=f"Task_{task_id}",
-                )
-                logger.info(
-                    f"Scheduled local task '{task_id}' with interval: {task_config.interval} seconds and delay variation: {task_config.delay_variation} seconds"
-                )
+                if task_config.cron:
+                    self.scheduler.add_job(
+                        func=task_wrapper,
+                        trigger=CronTrigger.from_crontab(task_config.cron),
+                        args=[task_config.query, task_config.delay_variation],
+                        id=task_id,
+                        name=f"Task_{task_id}",
+                    )
+                    logger.info(f"Scheduled cron task '{task_id}' with cron expression: {task_config.cron}")
+                else:
+                    self.scheduler.add_job(
+                        func=task_wrapper,
+                        trigger=IntervalTrigger(seconds=task_config.interval),
+                        args=[task_config.query, task_config.delay_variation],
+                        id=task_id,
+                        name=f"Task_{task_id}",
+                    )
+                    logger.info(
+                        f"Scheduled local task '{task_id}' with interval: {task_config.interval} seconds and delay variation: {task_config.delay_variation} seconds"
+                    )
 
         # Start the local scheduler if we have any local tasks
         if any(task.schedule.type == "local" for task in tasks_config.values()):