fix(cron): preserve fixed scheduling intervals

astrbot/core/cron/manager.py

@@ -2,13 +2,15 @@
 import json
 import re
 from collections.abc import Awaitable, Callable
-from datetime import datetime, timezone
+from datetime import datetime, timedelta
+from datetime import timezone as dt_timezone
 from typing import TYPE_CHECKING, Any
 from zoneinfo import ZoneInfo
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
 from apscheduler.triggers.date import DateTrigger
+from apscheduler.triggers.interval import IntervalTrigger
 
 from astrbot import logger
 from astrbot.core.agent.runners.base import AgentState
@@ -192,10 +194,25 @@ async def add_active_job(
         persistent: bool = True,
         run_once: bool = False,
         run_at: datetime | None = None,
+        interval_seconds: int | None = None,
     ) -> CronJob:
         # If run_once with run_at, store run_at in payload for later reference.
         if run_once and run_at:
             payload = {**payload, "run_at": run_at.isoformat()}
+        interval_anchor_at = None
+        if interval_seconds is not None:
+            interval_anchor_at = datetime.now(dt_timezone.utc)
+            self._build_trigger(
+                CronJob(
+                    name=name,
+                    job_type="active_agent",
+                    cron_expression=cron_expression,
+                    run_once=run_once,
+                    payload=payload,
+                    interval_seconds=interval_seconds,
+                    interval_anchor_at=interval_anchor_at,
+                )
+            )
         job = await self.db.create_cron_job(
             name=name,
             job_type="active_agent",
@@ -206,6 +223,8 @@ async def add_active_job(
             enabled=enabled,
             persistent=persistent,
             run_once=run_once,
+            interval_seconds=interval_seconds,
+            interval_anchor_at=interval_anchor_at,
         )
         if enabled:
             self._schedule_job(job)
@@ -215,8 +234,17 @@ async def update_job(self, job_id: str, **kwargs) -> CronJob | None:
         current_job = await self.db.get_cron_job(job_id)
         if not current_job:
             return None
+        if (
+            "interval_seconds" in kwargs
+            and kwargs["interval_seconds"] != current_job.interval_seconds
+        ):
+            kwargs["interval_anchor_at"] = (
+                datetime.now(dt_timezone.utc)
+                if kwargs["interval_seconds"] is not None
+                else None
+            )
         candidate = current_job.model_copy(update=kwargs)
-        if candidate.enabled:
+        if candidate.enabled or candidate.interval_seconds is not None:
             # Invalid edits must not overwrite the durable job or remove its
             # working schedule. Disabled legacy jobs can still be corrected.
             self._build_trigger(candidate)
@@ -264,7 +292,9 @@ def _remove_scheduled(self, job_id: str) -> None:
         if self.scheduler.get_job(job_id):
             self.scheduler.remove_job(job_id)
 
-    def _build_trigger(self, job: CronJob) -> CronTrigger | DateTrigger:
+    def _build_trigger(
+        self, job: CronJob
+    ) -> CronTrigger | DateTrigger | IntervalTrigger:
         """Validate a job's timing without modifying stored or scheduled jobs.
 
         Args:
@@ -287,7 +317,33 @@ def _build_trigger(self, job: CronJob) -> CronTrigger | DateTrigger:
                         job.timezone,
                         job.job_id,
                     )
-            if job.run_once:
+            if job.interval_seconds is not None:
+                seconds = job.interval_seconds
+                if (
+                    type(seconds) is not int
+                    or not 60 <= seconds <= 2_147_483_647
+                    or seconds % 60
+                ):
+                    raise ValueError(
+                        "interval_seconds must be an integer multiple of 60 "
+                        "between 60 and 2147483647"
+                    )
+                if job.run_once or job.cron_expression or job.payload.get("run_at"):
+                    raise ValueError("interval cannot be combined with cron or run_at")
+                anchor = job.interval_anchor_at
+                if anchor is None:
+                    raise ValueError("interval job missing interval_anchor_at")
+                # SQLite returns naive UTC timestamps. Keep elapsed durations
+                # independent of the display timezone and daylight saving time.
+                if anchor.tzinfo is None:
+                    anchor = anchor.replace(tzinfo=dt_timezone.utc)
+                anchor = anchor.astimezone(dt_timezone.utc)
+                trigger = IntervalTrigger(
+                    seconds=seconds,
+                    start_date=anchor + timedelta(seconds=seconds),
+                    timezone=dt_timezone.utc,
+                )
+            elif job.run_once:
                 run_at_str = None
                 if isinstance(job.payload, dict):
                     run_at_str = job.payload.get("run_at")
@@ -346,7 +402,7 @@ def _get_next_run_time(self, job_id: str):
         aps_job = self.scheduler.get_job(job_id)
         if not aps_job or aps_job.next_run_time is None:
             return None
-        return aps_job.next_run_time.astimezone(timezone.utc)
+        return aps_job.next_run_time.astimezone(dt_timezone.utc)
 
     def get_next_run_time(self, job_id: str) -> datetime | None:
         """Read the live next-run time straight from the scheduler.
@@ -386,7 +442,7 @@ async def _run_job(
             job = await self.db.get_cron_job(job_id)
             if not job or (not job.enabled and not ignore_enabled):
                 return
-            start_time = datetime.now(timezone.utc)
+            start_time = datetime.now(dt_timezone.utc)
             await self.db.update_cron_job(
                 job_id, status="running", last_run_at=start_time, last_error=None
             )

astrbot/core/db/__init__.py

@@ -731,6 +731,8 @@ async def create_cron_job(
         job_type: str,
         cron_expression: str | None,
         *,
+        interval_seconds: int | None = None,
+        interval_anchor_at: datetime.datetime | None = None,
         timezone: str | None = None,
         payload: dict | None = None,
         description: str | None = None,
@@ -750,6 +752,8 @@ async def update_cron_job(
         *,
         name: str | None = None,
         cron_expression: str | None = None,
+        interval_seconds: int | None = None,
+        interval_anchor_at: datetime.datetime | None = None,
         timezone: str | None = None,
         payload: dict | None = None,
         description: str | None = None,

astrbot/core/db/po.py

@@ -203,6 +203,10 @@ class CronJob(TimestampMixin, SQLModel, table=True):
     description: str | None = Field(default=None, sa_type=Text)
     job_type: str = Field(max_length=32, nullable=False)  # basic | active_agent
     cron_expression: str | None = Field(default=None, max_length=255)
+    interval_seconds: int | None = Field(default=None)
+    interval_anchor_at: datetime | None = Field(
+        default=None, sa_type=DateTime(timezone=False)
+    )
     timezone: str | None = Field(default=None, max_length=64)
     payload: dict = Field(default_factory=dict, sa_type=JSON)
     enabled: bool = Field(default=True)

astrbot/core/db/sqlite.py

@@ -99,6 +99,16 @@ async def initialize(self) -> None:
             await self._ensure_platform_message_history_checkpoint_column(conn)
             await self._ensure_chatui_project_workspace_columns(conn)
             await self._ensure_conversation_indexes(conn)
+            result = await conn.execute(text("PRAGMA table_info(cron_jobs)"))
+            cron_columns = {row[1] for row in result.fetchall()}
+            for column, sql_type in (
+                ("interval_seconds", "INTEGER"),
+                ("interval_anchor_at", "DATETIME"),
+            ):
+                if column not in cron_columns:
+                    await conn.execute(
+                        text(f"ALTER TABLE cron_jobs ADD COLUMN {column} {sql_type}")
+                    )
             # The table-level unique constraint already provides an index for UMO
             # lookups. Older schemas also created this redundant explicit index.
             await conn.execute(text("DROP INDEX IF EXISTS ix_umo_aliases_umo"))
@@ -2637,6 +2647,8 @@ async def create_cron_job(
         job_type: str,
         cron_expression: str | None,
         *,
+        interval_seconds: int | None = None,
+        interval_anchor_at: datetime | None = None,
         timezone: str | None = None,
         payload: dict | None = None,
         description: str | None = None,
@@ -2653,6 +2665,8 @@ async def create_cron_job(
                     name=name,
                     job_type=job_type,
                     cron_expression=cron_expression,
+                    interval_seconds=interval_seconds,
+                    interval_anchor_at=interval_anchor_at,
                     timezone=timezone,
                     payload=payload or {},
                     description=description,
@@ -2674,6 +2688,8 @@ async def update_cron_job(
         *,
         name: str | None | object = CRON_FIELD_NOT_SET,
         cron_expression: str | None | object = CRON_FIELD_NOT_SET,
+        interval_seconds: int | None | object = CRON_FIELD_NOT_SET,
+        interval_anchor_at: datetime | None | object = CRON_FIELD_NOT_SET,
         timezone: str | None | object = CRON_FIELD_NOT_SET,
         payload: dict | None | object = CRON_FIELD_NOT_SET,
         description: str | None | object = CRON_FIELD_NOT_SET,
@@ -2692,6 +2708,8 @@ async def update_cron_job(
                 for key, val in {
                     "name": name,
                     "cron_expression": cron_expression,
+                    "interval_seconds": interval_seconds,
+                    "interval_anchor_at": interval_anchor_at,
                     "timezone": timezone,
                     "payload": payload,
                     "description": description,

astrbot/core/tools/cron_tools.py

@@ -276,12 +276,16 @@ async def call(
                 cron_expression = None
                 payload["run_at"] = run_at_dt.isoformat()
             else:
-                if not cron_expression:
+                # Interval jobs have no Cron expression; keep their interval
+                # unless this edit switches them to Cron.
+                if not cron_expression and job.interval_seconds is None:
                     return "error: cron_expression is required when run_once=false."
                 payload.pop("run_at", None)
 
             updates["run_once"] = run_once
             updates["cron_expression"] = cron_expression
+            if job.interval_seconds is not None and (run_once or cron_expression):
+                updates["interval_seconds"] = None
             updates["payload"] = payload
 
             try:

astrbot/dashboard/api/cron.py

@@ -25,7 +25,10 @@ def get_service(request: Request) -> CronService:
 
 
 def _payload_dict(payload: CronJobRequest) -> dict:
-    return payload.model_dump(exclude_none=True)
+    data = payload.model_dump(exclude_none=True)
+    if "interval_seconds" in payload.model_fields_set:
+        data["interval_seconds"] = payload.interval_seconds
+    return data
 
 
 def _raise_cron_error(exc: CronServiceError) -> None:

astrbot/dashboard/schemas.py

@@ -177,7 +177,9 @@ class ChatThreadMessageRequest(OpenModel):
 
 
 class CronJobRequest(OpenModel):
-    pass
+    interval_seconds: int | None = Field(
+        default=None, strict=True, ge=60, le=2_147_483_647, multiple_of=60
+    )
 
 
 class CommandUpdateRequest(BaseModel):

astrbot/dashboard/services/cron_service.py

@@ -26,7 +26,13 @@ def _get_cron_manager(self):
     @staticmethod
     def serialize_job(job) -> dict:
         data = job.model_dump() if hasattr(job, "model_dump") else job.__dict__
-        for key in ["created_at", "updated_at", "last_run_at", "next_run_time"]:
+        for key in [
+            "created_at",
+            "updated_at",
+            "last_run_at",
+            "next_run_time",
+            "interval_anchor_at",
+        ]:
             value = data.get(key)
             if isinstance(value, datetime):
                 if value.tzinfo is None:
@@ -58,7 +64,8 @@ async def create_job(self, payload: object) -> dict:
                 raise CronServiceError("Invalid payload")
 
             name = payload.get("name") or "active_agent_task"
-            cron_expression = payload.get("cron_expression")
+            cron_expression = str(payload.get("cron_expression") or "").strip() or None
+            interval_seconds = payload.get("interval_seconds")
             note = payload.get("note") or payload.get("description") or name
             session = str(payload.get("session") or "").strip()
             persona_id = payload.get("persona_id")
@@ -75,11 +82,15 @@ async def create_job(self, payload: object) -> dict:
             run_once = bool(payload.get("run_once", False))
             run_at = payload.get("run_at")
 
+            if interval_seconds is not None and (run_once or cron_expression or run_at):
+                raise CronServiceError(
+                    "interval cannot be combined with cron or run_at"
+                )
             if run_once and not run_at:
                 raise CronServiceError("run_at is required when run_once=true")
-            if (not run_once) and not cron_expression:
+            if (not run_once) and not cron_expression and interval_seconds is None:
                 raise CronServiceError(
-                    "cron_expression is required when run_once=false"
+                    "cron_expression or interval_seconds is required when run_once=false"
                 )
             if run_once and cron_expression:
                 cron_expression = None
@@ -103,6 +114,7 @@ async def create_job(self, payload: object) -> dict:
                 enabled=enabled,
                 run_once=run_once,
                 run_at=run_at_dt,
+                interval_seconds=interval_seconds,
             )
             return self.serialize_job(job)
         except CronServiceError:
@@ -226,21 +238,33 @@ def _merge_active_agent_updates(self, job, payload: dict, updates: dict) -> None
             else merged_payload.get("run_at")
         )
         run_at_iso = self._normalize_run_at_iso(run_at_raw)
+        next_interval_seconds = (
+            payload["interval_seconds"]
+            if "interval_seconds" in payload
+            else job.interval_seconds
+        )
 
-        if next_run_once:
+        if next_interval_seconds is not None:
+            if next_run_once or next_cron_expression or run_at_iso:
+                raise CronServiceError(
+                    "interval cannot be combined with cron or run_at"
+                )
+            merged_payload.pop("run_at", None)
+        elif next_run_once:
             if not run_at_iso:
                 raise CronServiceError("run_at is required when run_once=true")
             next_cron_expression = None
             merged_payload["run_at"] = run_at_iso
         else:
             if not next_cron_expression:
                 raise CronServiceError(
-                    "cron_expression is required when run_once=false"
+                    "cron_expression or interval_seconds is required when run_once=false"
                 )
             merged_payload.pop("run_at", None)
 
         updates["run_once"] = next_run_once
         updates["cron_expression"] = next_cron_expression
+        updates["interval_seconds"] = next_interval_seconds
         updates["payload"] = merged_payload
 
     @staticmethod

dashboard/src/api/generated/openapi-v1/types.gen.ts

@@ -288,7 +288,14 @@ export type CronJobPatchRequest = CronJobRequest;
 
 export type CronJobRequest = {
     name?: string;
+    /**
+     * Five-field Cron expression. Empty for interval or one-shot jobs.
+     */
     cron_expression?: string;
+    /**
+     * Fixed elapsed interval in whole minutes, expressed in seconds. Mutually exclusive with cron_expression and run_once=true/run_at. The server persists a UTC interval_anchor_at and first runs one interval after it. Restart and enable/disable preserve this anchor; changing the interval resets it. Send null to clear an interval when switching to Cron or a one-shot schedule. Omitting this field in a patch preserves the existing interval.
+     */
+    interval_seconds?: (number) | null;
     timezone?: string;
     session?: string;
     note?: string;
@@ -301,7 +308,7 @@ export type CronJobRequest = {
     payload?: {
         [key: string]: unknown;
     };
-    [key: string]: unknown | string | boolean;
+    [key: string]: unknown | string | number | boolean;
 };
 
 export type DynamicConfig = {