diff --git a/astrbot/core/cron/manager.py b/astrbot/core/cron/manager.py index e73a0ffa44..72dfce0210 100644 --- a/astrbot/core/cron/manager.py +++ b/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 ) diff --git a/astrbot/core/db/__init__.py b/astrbot/core/db/__init__.py index 399b1dcc09..68444980a1 100644 --- a/astrbot/core/db/__init__.py +++ b/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, diff --git a/astrbot/core/db/po.py b/astrbot/core/db/po.py index 42138c0aed..e3d039726a 100644 --- a/astrbot/core/db/po.py +++ b/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) diff --git a/astrbot/core/db/sqlite.py b/astrbot/core/db/sqlite.py index c8f7326da3..099961159b 100644 --- a/astrbot/core/db/sqlite.py +++ b/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, diff --git a/astrbot/core/tools/cron_tools.py b/astrbot/core/tools/cron_tools.py index ff6272e4d7..10aad8addf 100644 --- a/astrbot/core/tools/cron_tools.py +++ b/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: diff --git a/astrbot/dashboard/api/cron.py b/astrbot/dashboard/api/cron.py index 1b1f9cc432..6e58a9dd82 100644 --- a/astrbot/dashboard/api/cron.py +++ b/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: diff --git a/astrbot/dashboard/schemas.py b/astrbot/dashboard/schemas.py index 2f2b2e9180..60dacd57be 100644 --- a/astrbot/dashboard/schemas.py +++ b/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): diff --git a/astrbot/dashboard/services/cron_service.py b/astrbot/dashboard/services/cron_service.py index ad08cbdf43..5acdd0403c 100644 --- a/astrbot/dashboard/services/cron_service.py +++ b/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,8 +238,19 @@ 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 @@ -235,12 +258,13 @@ def _merge_active_agent_updates(self, job, payload: dict, updates: dict) -> None 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 diff --git a/dashboard/src/api/generated/openapi-v1/types.gen.ts b/dashboard/src/api/generated/openapi-v1/types.gen.ts index 0ff1ad2af0..190221c8c3 100644 --- a/dashboard/src/api/generated/openapi-v1/types.gen.ts +++ b/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 = { diff --git a/dashboard/src/views/CronJobPage.vue b/dashboard/src/views/CronJobPage.vue index f39a6efb3b..fe682e2e89 100644 --- a/dashboard/src/views/CronJobPage.vue +++ b/dashboard/src/views/CronJobPage.vue @@ -581,6 +581,16 @@ function scheduleProductLabel(item: any): string { if (item.run_once) { return tm("card.onceAt", { time: formatTime(item.run_at) }); } + if (item.interval_seconds != null) { + const seconds = Number(item.interval_seconds); + if (seconds % 86400 === 0) { + return tm("card.everyDays", { count: seconds / 86400 }); + } + if (seconds % 3600 === 0) { + return tm("card.everyHours", { count: seconds / 3600 }); + } + return tm("card.everyMinutes", { count: seconds / 60 }); + } const cron = String(item.cron_expression || "").trim(); const parts = cron.split(/\s+/); @@ -589,39 +599,6 @@ function scheduleProductLabel(item: any): string { } const [minute, hour, dayOfMonth, month, dayOfWeek] = parts; - const minuteInterval = /^\*\/(\d+)$/.exec(minute); - if ( - minuteInterval && - hour === "*" && - dayOfMonth === "*" && - month === "*" && - dayOfWeek === "*" - ) { - return tm("card.everyMinutes", { count: Number(minuteInterval[1]) }); - } - - const hourInterval = /^\*\/(\d+)$/.exec(hour); - if ( - minute === "0" && - hourInterval && - dayOfMonth === "*" && - month === "*" && - dayOfWeek === "*" - ) { - return tm("card.everyHours", { count: Number(hourInterval[1]) }); - } - - const dayInterval = /^\*\/(\d+)$/.exec(dayOfMonth); - if ( - minute === "0" && - hour === "0" && - dayInterval && - month === "*" && - dayOfWeek === "*" - ) { - return tm("card.everyDays", { count: Number(dayInterval[1]) }); - } - const minuteNumber = Number(minute); const hourNumber = Number(hour); const dayOfMonthNumber = Number(dayOfMonth); @@ -849,16 +826,6 @@ function isCronTime(minute: number, hour: number): boolean { function buildCronExpression(): string { const mode = newJob.value.schedule_mode; - if (mode === "interval") { - const value = Math.max(1, Number(newJob.value.interval_value || 1)); - if (newJob.value.interval_unit === "minutes") { - return `*/${Math.min(value, 59)} * * * *`; - } - if (newJob.value.interval_unit === "hours") { - return `0 */${Math.min(value, 23)} * * *`; - } - return `0 0 */${Math.min(value, 31)} * *`; - } if (mode === "daily") { const time = parseTimeParts(newJob.value.daily_time); if (!time) return ""; @@ -897,6 +864,19 @@ function readScheduleFromJob(job: any) { if (job.run_once) { return { ...fallback, schedule_mode: "once" as ScheduleMode }; } + if (job.interval_seconds != null) { + const seconds = Number(job.interval_seconds); + const unit: IntervalUnit = seconds % 86400 === 0 + ? "days" + : seconds % 3600 === 0 ? "hours" : "minutes"; + const divisor = { minutes: 60, hours: 3600, days: 86400 }[unit]; + return { + ...fallback, + schedule_mode: "interval" as ScheduleMode, + interval_value: seconds / divisor, + interval_unit: unit, + }; + } const cron = String(job.cron_expression || "").trim(); const parts = cron.split(/\s+/); @@ -914,54 +894,6 @@ function readScheduleFromJob(job: any) { ? `${padTimePart(hourNumber)}:${padTimePart(minuteNumber)}` : "09:00"; - const minuteInterval = /^\*\/(\d+)$/.exec(minute); - if ( - minuteInterval && - hour === "*" && - dayOfMonth === "*" && - month === "*" && - dayOfWeek === "*" - ) { - return { - ...fallback, - schedule_mode: "interval" as ScheduleMode, - interval_value: Number(minuteInterval[1]), - interval_unit: "minutes" as IntervalUnit, - }; - } - - const hourInterval = /^\*\/(\d+)$/.exec(hour); - if ( - minute === "0" && - hourInterval && - dayOfMonth === "*" && - month === "*" && - dayOfWeek === "*" - ) { - return { - ...fallback, - schedule_mode: "interval" as ScheduleMode, - interval_value: Number(hourInterval[1]), - interval_unit: "hours" as IntervalUnit, - }; - } - - const dayInterval = /^\*\/(\d+)$/.exec(dayOfMonth); - if ( - minute === "0" && - hour === "0" && - dayInterval && - month === "*" && - dayOfWeek === "*" - ) { - return { - ...fallback, - schedule_mode: "interval" as ScheduleMode, - interval_value: Number(dayInterval[1]), - interval_unit: "days" as IntervalUnit, - }; - } - if (hasCronTime && dayOfMonth === "*" && month === "*" && dayOfWeek === "*") { return { ...fallback, @@ -1007,12 +939,18 @@ function readScheduleFromJob(job: any) { function buildPayload() { const runOnce = newJob.value.schedule_mode === "once"; - const cronExpression = runOnce ? "" : buildCronExpression(); + const isInterval = newJob.value.schedule_mode === "interval"; + const cronExpression = runOnce || isInterval ? "" : buildCronExpression(); + const intervalSeconds = isInterval + ? Number(newJob.value.interval_value) * + { minutes: 60, hours: 3600, days: 86400 }[newJob.value.interval_unit] + : null; return { run_once: runOnce, name: newJob.value.name.trim(), note: newJob.value.note.trim(), cron_expression: cronExpression, + interval_seconds: intervalSeconds, run_at: runOnce ? toIsoDatetime(newJob.value.run_at) : "", session: newJob.value.session, timezone: newJob.value.timezone, @@ -1047,7 +985,10 @@ function validateScheduleFields(): boolean { const validUnit = ["minutes", "hours", "days"].includes( newJob.value.interval_unit, ); - if (!Number.isInteger(value) || value < 1 || !validUnit) { + const seconds = value * + { minutes: 60, hours: 3600, days: 86400 }[newJob.value.interval_unit]; + if (!Number.isInteger(value) || value < 1 || !validUnit || + !Number.isSafeInteger(seconds) || seconds > 2147483647) { toast(tm("messages.intervalRequired"), "warning"); return false; } diff --git a/dashboard/tests/cronSchedule.test.mjs b/dashboard/tests/cronSchedule.test.mjs new file mode 100644 index 0000000000..36bb427250 --- /dev/null +++ b/dashboard/tests/cronSchedule.test.mjs @@ -0,0 +1,101 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import test from "node:test"; +import vm from "node:vm"; +import ts from "typescript"; + +function setup() { + const source = readFileSync( + new URL("../src/views/CronJobPage.vue", import.meta.url), "utf8", + ).split('")[0]; + const ast = ts.createSourceFile("cron.ts", source, ts.ScriptTarget.Latest, true); + const names = new Set([ + "buildPayload", "buildCronExpression", "readScheduleFromJob", + "validateScheduleFields", "scheduleProductLabel", "parseTimeParts", + "padTimePart", "isCronTime", + ]); + const definitions = ast.statements.filter( + (node) => ts.isFunctionDeclaration(node) && names.has(node.name.text), + ); + assert.equal(definitions.length, names.size); + const warnings = []; + const context = vm.createContext({ + newJob: { value: { + schedule_mode: "interval", interval_value: 40, interval_unit: "minutes", + name: "Reminder", note: "Reminder note", cron_expression: "", + run_at: "2030-01-01T09:00:00Z", timezone: "UTC", session: "session", enabled: true, + daily_time: "09:00", weekly_time: "09:00", weekly_day: 1, + monthly_time: "09:00", monthly_day: 15, + } }, + tm: (key, params) => ({ key, ...params }), + toast: (message) => warnings.push(message), + formatTime: (value) => value, + toIsoDatetime: (value) => value, + }); + vm.runInContext(ts.transpileModule( + definitions.map((node) => node.getText(ast)).join("\n"), + { compilerOptions: { target: ts.ScriptTarget.ES2022 } }, + ).outputText, context); + return { context, form: context.newJob.value, warnings }; +} + +for (const [value, unit, seconds] of [ + [40, "minutes", 2400], [60, "minutes", 3600], [24, "hours", 86400], + [48, "hours", 172800], [31, "days", 2678400], [40, "days", 3456000], +]) { + test(`${value} ${unit} round-trips as a fixed interval without clipping`, () => { + const { context, form } = setup(); + Object.assign(form, { interval_value: value, interval_unit: unit }); + assert.equal(context.validateScheduleFields(), true); + const payload = context.buildPayload(); + assert.equal(payload.interval_seconds, seconds); + assert.equal(payload.cron_expression, ""); + assert.equal(payload.run_once, false); + assert.equal(payload.run_at, ""); + Object.assign(form, context.readScheduleFromJob(payload)); + assert.equal(form.schedule_mode, "interval"); + assert.equal(context.buildPayload().interval_seconds, seconds); + assert.match(context.scheduleProductLabel(payload).key, /^card.every/); + }); +} + +for (const expression of ["*/40 * * * *", "*/59 * * * *", "0 */23 * * *", "0 0 */31 * *"]) { + test(`legacy ${expression} remains a custom Cron schedule when edited`, () => { + const { context, form } = setup(); + const job = { cron_expression: expression, interval_seconds: null, run_once: false }; + Object.assign(form, context.readScheduleFromJob(job)); + assert.equal(form.schedule_mode, "cron"); + assert.equal(context.scheduleProductLabel(job).key, "card.customCron"); + assert.equal(context.buildPayload().cron_expression, expression); + assert.equal(context.buildPayload().interval_seconds, null); + }); +} + +test("switching away from intervals explicitly clears the backend interval", () => { + const { context, form } = setup(); + for (const [mode, expression] of [ + ["daily", "0 9 * * *"], ["weekly", "0 9 * * 1"], ["monthly", "0 9 15 * *"], + ]) { + form.schedule_mode = mode; + assert.equal(context.validateScheduleFields(), true); + assert.equal(context.buildPayload().interval_seconds, null); + assert.equal(context.buildPayload().cron_expression, expression); + assert.equal(context.readScheduleFromJob({ cron_expression: expression }).schedule_mode, mode); + } + form.schedule_mode = "once"; + assert.equal(context.buildPayload().interval_seconds, null); + assert.equal(context.buildPayload().run_once, true); + assert.equal(context.buildPayload().run_at, form.run_at); +}); + +test("invalid intervals fail validation instead of being coerced or clamped", () => { + const { context, form, warnings } = setup(); + for (const value of [0, -1, 1.5, NaN, Infinity, 2 ** 31, Number.MAX_SAFE_INTEGER]) { + form.interval_value = value; + assert.equal(context.validateScheduleFields(), false); + } + form.interval_value = 1; + form.interval_unit = "invalid"; + assert.equal(context.validateScheduleFields(), false); + assert.equal(warnings.length, 8); +}); diff --git a/docs/en/use/proactive-agent.md b/docs/en/use/proactive-agent.md index 6f2cebaf65..3e80d06ff8 100644 --- a/docs/en/use/proactive-agent.md +++ b/docs/en/use/proactive-agent.md @@ -81,10 +81,21 @@ minute hour day-of-month month day-of-week Recurring schedules use the task's time zone. New WebUI tasks default to the target conversation's configuration time zone, or the system time zone if none is specified. The one-off date/time picker uses the browser's local time and converts it to an absolute timestamp. Verify the resulting schedule when browser, server, and container time zones differ. -The interval selector currently creates Cron calendar steps, rather than a timer measured from creation. For example, every two hours runs at even-numbered hours. Minute, hour, and day steps are capped at 59, 23, and 31 respectively; day steps across months are not a fixed elapsed duration. Monthly tasks on the 29th, 30th, or 31st do not run in months without that date. +Monthly tasks on the 29th, 30th, or 31st do not run in months without that date. AstrBot must be online to execute tasks. Saved schedules are reloaded after a restart, but do not depend on it replaying every missed reminder. Start with a one-off task a few minutes ahead to check timing and delivery. +### Fixed-interval tasks + +Select **Interval** for **Execution time**, enter a positive integer and a unit (minutes, hours, or days), and save. Every 40 minutes always means 40 elapsed minutes; every 24 hours always means 24 elapsed hours, including across hour and day boundaries. + +- The first run occurs one full interval after saving. Changing the interval starts a new timing period at the time of saving. +- Restarting, disabling and re-enabling, or editing other fields such as the name or task requirements preserves the original timing anchor. Missed periods during downtime are skipped; execution resumes at the next time on the original schedule. +- A day means 24 elapsed hours. The local clock time can change across daylight saving transitions. Choose **Daily** for a fixed local time each day. +- Older interval tasks only contain a Cron expression, so their original input cannot be recovered. After upgrading, they appear as **Custom Cron** and retain their existing Cron schedule. To use a true fixed interval, edit the task, select **Interval**, and enter the intended value again; saving starts a new timing period. + +The API represents intervals as `interval_seconds`, accepting positive multiples of 60 up to 2147483647 seconds. The server manages `interval_anchor_at`. An interval cannot be combined with Cron or a one-shot execution time. + ## What a scheduled task can use Tasks use the target conversation's configuration and conversation context. Later changes to models, Personas, and tools can affect subsequent runs. Put required details in the task instructions instead of writing only “do what we discussed.” Tasks without a delivery target use a separate scheduling conversation and the default configuration. diff --git a/docs/zh/use/proactive-agent.md b/docs/zh/use/proactive-agent.md index 641c7d629d..18547452b3 100644 --- a/docs/zh/use/proactive-agent.md +++ b/docs/zh/use/proactive-agent.md @@ -81,10 +81,21 @@ 循环计划使用任务的时区；WebUI 新建时默认从投递会话对应的配置文件取得时区，未指定时使用系统时区。一次性日期时间选择器使用浏览器本地时间并转换为绝对时间。服务器、容器和浏览器时区不同时，务必检查创建后的执行时间。 -“间隔”目前转换为 Cron 日历字段步进，并不是“从创建时刻起每隔 N 小时”的计时器。例如，每隔 2 小时会在偶数整点执行。分钟、小时、天的步进上限分别为 59、23、31；跨月的天数步进不等于固定时长。每月选择 29、30、31 日时，没有该日期的月份不会执行。 +每月选择 29、30、31 日时，没有该日期的月份不会执行。 AstrBot 必须在线才能执行任务。任务会保存，重启后重新加载计划，但不要依赖它补发停机期间错过的所有提醒。首次配置建议先用几分钟后的一次性任务检查时间和投递。 +### 固定间隔任务 + +将 **执行时间** 设为 **间隔**，填写正整数和分钟、小时或天单位，再保存。例如，每 40 分钟会始终相隔 40 分钟，每 24 小时会始终相隔 24 小时，不会在跨小时或跨天时缩短间隔。 + +- 首次执行在保存后经过一个完整间隔时发生。修改间隔会从保存时重新计时。 +- 重启、停用后重新启用、修改名称或任务需求等其他字段，都会保留原来的时间起点。停机期间错过的周期不会逐次补发，恢复后在下一个符合原有节奏的时刻执行。 +- 一天表示固定 24 小时；在实行夏令时的时区，本地钟点可能变化。需要每天固定当地时间执行时，请选择 **每天**。 +- 旧版本保存的间隔任务只有 Cron 表达式，无法恢复最初输入。升级后这些任务显示为 **自定义** 并保持原来的 Cron 调度。若需要真正的固定间隔，请编辑任务，重新选择 **间隔** 并填写期望值；保存后开始新的计时。 + +API 使用 `interval_seconds` 表示间隔，只接受 60 的正整数倍，最大值为 2147483647 秒；起点 `interval_anchor_at` 由服务器管理。间隔与 Cron、单次执行时间不能同时设置。 + ## 执行时会用到哪些能力 定时任务会使用目标会话对应的配置和对话上下文。修改模型、人设、工具配置后，后续任务可能受这些变更影响；任务需求中应写明必要信息，不要只写“按刚才说的做”。未选择投递地的任务使用独立的定时任务会话和默认配置。 diff --git a/openspec/openapi-v1.yaml b/openspec/openapi-v1.yaml index 4b77f1a763..6749efbc0f 100644 --- a/openspec/openapi-v1.yaml +++ b/openspec/openapi-v1.yaml @@ -6628,6 +6628,21 @@ components: type: string cron_expression: type: string + description: Five-field Cron expression. Empty for interval or one-shot jobs. + interval_seconds: + type: integer + nullable: true + minimum: 60 + maximum: 2147483647 + multipleOf: 60 + description: >- + 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. timezone: type: string session: diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 36870e6178..33df34cd2c 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -104,6 +104,7 @@ def test_pipeline_import_is_stable_with_mocked_apscheduler() -> None: "mock_apscheduler.triggers = MagicMock();" "mock_apscheduler.triggers.cron = MagicMock();" "mock_apscheduler.triggers.date = MagicMock();" + "mock_apscheduler.triggers.interval = MagicMock();" "sys.modules['apscheduler'] = mock_apscheduler;" "sys.modules['apscheduler.schedulers'] = mock_apscheduler.schedulers;" "sys.modules['apscheduler.schedulers.asyncio'] = mock_apscheduler.schedulers.asyncio;" @@ -111,6 +112,7 @@ def test_pipeline_import_is_stable_with_mocked_apscheduler() -> None: "sys.modules['apscheduler.triggers'] = mock_apscheduler.triggers;" "sys.modules['apscheduler.triggers.cron'] = mock_apscheduler.triggers.cron;" "sys.modules['apscheduler.triggers.date'] = mock_apscheduler.triggers.date;" + "sys.modules['apscheduler.triggers.interval'] = mock_apscheduler.triggers.interval;" "import astrbot.core.pipeline as pipeline;" "assert pipeline.ProcessStage is not None;" "assert pipeline.RespondStage is not None" diff --git a/tests/unit/test_cron_intervals.py b/tests/unit/test_cron_intervals.py new file mode 100644 index 0000000000..5e8c20a9e8 --- /dev/null +++ b/tests/unit/test_cron_intervals.py @@ -0,0 +1,406 @@ +"""Regression coverage for fixed-duration schedules and legacy Cron records.""" + +import asyncio +import sqlite3 +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from unittest.mock import MagicMock +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +import httpx +import pytest +import pytest_asyncio +from apscheduler.triggers.cron import CronTrigger +from apscheduler.triggers.date import DateTrigger +from apscheduler.triggers.interval import IntervalTrigger +from fastapi import FastAPI +from fastapi.responses import JSONResponse + +from astrbot.core.cron.manager import CronJobManager +from astrbot.core.db.po import CronJob +from astrbot.core.db.sqlite import SQLiteDatabase +from astrbot.core.tools.cron_tools import FutureTaskTool +from astrbot.dashboard.api import cron +from astrbot.dashboard.responses import ApiError +from astrbot.dashboard.services.cron_service import CronService, CronServiceError + + +@pytest_asyncio.fixture +async def interval_store(tmp_path): + """Provide the production service, SQLite store, and a paused scheduler. + + Args: + tmp_path: Directory for an isolated database. + + Yields: + Database, scheduler, and dashboard service. + """ + db = SQLiteDatabase(str(tmp_path / "cron.db")) + await db.initialize() + manager = CronJobManager(db) + await manager.start(SimpleNamespace()) + manager.scheduler.pause() + service = CronService( + SimpleNamespace( + cron_manager=manager, + astrbot_config_mgr=SimpleNamespace(get_conf=lambda _: {"timezone": "UTC"}), + ) + ) + try: + yield db, manager, service + finally: + await asyncio.sleep(0) + await manager.shutdown() + await db.engine.dispose() + + +@pytest.mark.parametrize("seconds", [40 * 60, 60 * 60, 24 * 3600, 31 * 86400]) +@pytest.mark.parametrize( + "anchor", + [ + datetime(2026, 3, 7, 14, tzinfo=timezone.utc), + datetime(2026, 10, 31, 13, tzinfo=timezone.utc), + ], +) +def test_interval_fire_times_and_restart_phase(seconds, anchor): + """Fixed intervals survive calendar boundaries, DST, and missed periods. + + Args: + seconds: Elapsed duration requested by the user. + anchor: UTC origin immediately before a DST transition. + """ + manager = CronJobManager(MagicMock()) + job = CronJob( + name="Reminder", + job_type="active_agent", + timezone="America/New_York", + interval_seconds=seconds, + interval_anchor_at=anchor, + ) + trigger = manager._build_trigger(job) + assert isinstance(trigger, IntervalTrigger) + first = trigger.get_next_fire_time(None, anchor) + assert first == anchor + timedelta(seconds=seconds) + previous = first + for _ in range(5): + following = trigger.get_next_fire_time(previous, previous) + assert (following - previous).total_seconds() == seconds + previous = following + + # A fresh trigger must skip downtime, using the persisted phase rather + # than starting another full interval at the time of restart. + restarted = manager._build_trigger(job) + now = anchor + timedelta(seconds=seconds * 10 + 1) + assert restarted.get_next_fire_time(None, now) == anchor + timedelta( + seconds=seconds * 11 + ) + if seconds == 86400: + local_first = first.astimezone(ZoneInfo("America/New_York")) + assert local_first.hour != anchor.astimezone(ZoneInfo("America/New_York")).hour + + +@pytest.mark.asyncio +async def test_interval_creation_without_system_timezone_data( + interval_store, monkeypatch +): + """UTC anchoring also works on hosts without an IANA timezone database. + + Args: + interval_store: Real database, scheduler, and service. + monkeypatch: Fixture used to simulate missing system timezone data. + """ + _, manager, service = interval_store + monkeypatch.setattr( + "astrbot.core.cron.manager.ZoneInfo", + MagicMock(side_effect=ZoneInfoNotFoundError("No timezone data")), + ) + created = await service.create_job({"interval_seconds": 3600}) + anchor = datetime.fromisoformat(created["interval_anchor_at"]) + assert manager.get_next_run_time(created["job_id"]) == anchor + timedelta(hours=1) + + +@pytest.mark.asyncio +async def test_interval_persistence_edits_and_restart(interval_store): + """Metadata and enable/disable edits keep the phase; duration edits reset it. + + Args: + interval_store: Real database, scheduler, and service. + """ + db, manager, service = interval_store + created = await service.create_job({"name": "Reminder", "interval_seconds": 2400}) + job_id = created["job_id"] + anchor = datetime.fromisoformat(created["interval_anchor_at"]) + assert anchor.utcoffset() == timedelta(0) + assert manager.get_next_run_time(job_id) == anchor + timedelta(seconds=2400) + stored = await db.get_cron_job(job_id) + assert stored.interval_anchor_at.replace(tzinfo=timezone.utc) == anchor + assert stored.cron_expression is None + + for patch in ( + {"name": "Renamed"}, + {"note": "Edited note"}, + {"session": "test:FriendMessage:other"}, + {"timezone": "Asia/Shanghai"}, + {"enabled": False}, + {"enabled": True}, + {"interval_seconds": 2400}, + ): + result = await service.update_job(job_id, patch) + assert result["interval_anchor_at"] == created["interval_anchor_at"] + assert manager.get_next_run_time(job_id) == anchor + timedelta(seconds=2400) + + await manager.shutdown() + await manager.start(SimpleNamespace()) + manager.scheduler.pause() + assert manager.get_next_run_time(job_id) == anchor + timedelta(seconds=2400) + + changed = await service.update_job(job_id, {"interval_seconds": 3600}) + new_anchor = datetime.fromisoformat(changed["interval_anchor_at"]) + assert new_anchor > anchor + assert manager.get_next_run_time(job_id) == new_anchor + timedelta(hours=1) + + changed = await service.update_job( + job_id, + { + "interval_seconds": None, + "cron_expression": "0 9 * * *", + }, + ) + assert changed["interval_anchor_at"] is None + assert isinstance(manager.scheduler.get_job(job_id).trigger, CronTrigger) + + +@pytest.mark.parametrize( + ("switch", "trigger_type"), + [ + ({"cron_expression": "0 9 * * *"}, CronTrigger), + ({"run_once": True, "run_at": "2030-01-01T09:00:00+00:00"}, DateTrigger), + ], +) +@pytest.mark.asyncio +async def test_future_task_tool_edits_interval_job( + interval_store, switch, trigger_type +): + """The agent tool keeps an interval on metadata edits and clears it on switch. + + Args: + interval_store: Real database, scheduler, and service. + switch: Tool arguments that replace the interval schedule. + trigger_type: Scheduler trigger expected after the switch. + """ + db, manager, _ = interval_store + job = await manager.add_active_job( + name="Reminder", + cron_expression=None, + payload={"session": "test:private:user", "sender_id": "user-1", "note": "Hi"}, + interval_seconds=2400, + ) + job_id = job.job_id + anchor = (await db.get_cron_job(job_id)).interval_anchor_at + next_run = manager.get_next_run_time(job_id) + context = SimpleNamespace( + context=SimpleNamespace( + context=SimpleNamespace(cron_manager=manager), + event=SimpleNamespace( + unified_msg_origin="test:private:user", + get_sender_id=lambda: "user-1", + ), + ) + ) + tool = FutureTaskTool() + + result = await tool.call(context, action="edit", job_id=job_id, name="Renamed") + assert result == f"Updated future task {job_id} (Renamed)." + stored = await db.get_cron_job(job_id) + assert stored.interval_seconds == 2400 + assert stored.interval_anchor_at == anchor + assert manager.get_next_run_time(job_id) == next_run + + result = await tool.call(context, action="edit", job_id=job_id, **switch) + assert result == f"Updated future task {job_id} (Renamed)." + stored = await db.get_cron_job(job_id) + assert stored.interval_seconds is None + assert stored.interval_anchor_at is None + assert isinstance(manager.scheduler.get_job(job_id).trigger, trigger_type) + + +@pytest.mark.parametrize("invalid", [0, -60, True, 60.0, 90.5, "60", 61, 2**31, 10**30]) +@pytest.mark.asyncio +async def test_invalid_interval_never_changes_database_or_scheduler( + interval_store, invalid +): + """Reject invalid durations even when a job is created disabled. + + Args: + interval_store: Real database, scheduler, and service. + invalid: Duration with an invalid type, resolution, or range. + """ + db, manager, service = interval_store + with pytest.raises(CronServiceError): + await service.create_job({"interval_seconds": invalid, "enabled": False}) + assert await db.list_cron_jobs() == [] + + created = await service.create_job({"interval_seconds": 2400}) + job_id = created["job_id"] + scheduled = manager.scheduler.get_job(job_id) + with pytest.raises(CronServiceError): + await service.update_job( + job_id, {"interval_seconds": invalid, "name": "Rejected"} + ) + stored = await db.get_cron_job(job_id) + assert stored.interval_seconds == 2400 + assert stored.name == created["name"] + assert manager.scheduler.get_job(job_id) is scheduled + + +@pytest.mark.parametrize( + "conflict", + [ + {"cron_expression": "*/40 * * * *"}, + {"run_once": True, "run_at": "2030-01-01T00:00:00+00:00"}, + {"run_at": "2030-01-01T00:00:00+00:00"}, + ], +) +@pytest.mark.asyncio +async def test_interval_rejects_conflicting_schedules(interval_store, conflict): + """Mixed schedule types cannot silently select one trigger. + + Args: + interval_store: Real database, scheduler, and service. + conflict: An incompatible Cron or one-shot schedule. + """ + db, _, service = interval_store + with pytest.raises(CronServiceError): + await service.create_job({"interval_seconds": 2400, **conflict}) + assert await db.list_cron_jobs() == [] + created = await service.create_job({"interval_seconds": 2400}) + with pytest.raises(CronServiceError): + await service.update_job(created["job_id"], conflict) + assert (await db.get_cron_job(created["job_id"])).interval_seconds == 2400 + + +@pytest.mark.parametrize("prefix", ["/api/v1/cron", "/api/cron"]) +@pytest.mark.asyncio +async def test_http_interval_round_trip_and_schedule_switches(interval_store, prefix): + """Both API families preserve explicit nulls when switching schedule types. + + Args: + interval_store: Real database, scheduler, and service. + prefix: Versioned or legacy dashboard API path. + """ + _, _, service = interval_store + app = FastAPI() + app.state.services = SimpleNamespace(cron=service) + app.include_router(cron.router, prefix="/api/v1") + app.include_router(cron.legacy_router) + app.dependency_overrides[cron.require_system_scope] = lambda: None + app.dependency_overrides[cron.require_dashboard_user] = lambda: "tester" + app.add_exception_handler( + ApiError, + lambda _, exc: JSONResponse( + {"message": exc.message}, + status_code=exc.status_code, + ), + ) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + for invalid in (True, 60.0, "60", 61, 0, 2**31): + response = await client.post( + f"{prefix}/jobs", json={"interval_seconds": invalid} + ) + assert response.status_code == 422 + response = await client.post(f"{prefix}/jobs", json={"interval_seconds": 2400}) + assert response.status_code == 200 + created = response.json()["data"] + url = f"{prefix}/jobs/{created['job_id']}" + response = await client.patch(url, json={"name": "Renamed"}) + assert ( + response.json()["data"]["interval_anchor_at"] + == created["interval_anchor_at"] + ) + response = await client.patch( + url, + json={ + "interval_seconds": None, + "run_once": True, + "run_at": "2030-01-01T00:00:00+00:00", + }, + ) + assert response.status_code == 200 + assert response.json()["data"]["interval_seconds"] is None + response = await client.patch( + url, + json={ + "run_once": False, + "run_at": "", + "interval_seconds": 86400, + }, + ) + assert response.status_code == 200 + assert response.json()["data"]["interval_seconds"] == 86400 + response = await client.patch( + url, + json={ + "interval_seconds": None, + "cron_expression": "*/40 * * * *", + }, + ) + assert response.status_code == 200 + assert response.json()["data"]["interval_anchor_at"] is None + response = await client.patch( + url, + json={ + "cron_expression": "", + "interval_seconds": 2400, + }, + ) + assert response.status_code == 200 + listed = (await client.get(f"{prefix}/jobs")).json()["data"] + assert listed[0]["interval_seconds"] == 2400 + + +@pytest.mark.asyncio +async def test_legacy_database_migration_preserves_cron(tmp_path): + """Add nullable fields idempotently without guessing historical intervals. + + Args: + tmp_path: Directory for a database using the previous schema. + """ + path = tmp_path / "legacy.db" + with sqlite3.connect(path) as connection: + connection.executescript(""" + CREATE TABLE cron_jobs ( + id INTEGER PRIMARY KEY, job_id VARCHAR(64) NOT NULL UNIQUE, + name VARCHAR(255) NOT NULL, description TEXT, + job_type VARCHAR(32) NOT NULL, cron_expression VARCHAR(255), + timezone VARCHAR(64), payload JSON, enabled BOOLEAN, + persistent BOOLEAN, run_once BOOLEAN, status VARCHAR(32), + last_run_at DATETIME, next_run_time DATETIME, last_error TEXT, + created_at DATETIME, updated_at DATETIME + ); + INSERT INTO cron_jobs VALUES ( + 1, 'legacy', 'Reminder', NULL, 'active_agent', '*/40 * * * *', + 'UTC', '{}', 1, 1, 0, 'scheduled', NULL, NULL, NULL, + '2026-10-01 00:00:00', '2026-10-01 00:00:00' + ); + """) + db = SQLiteDatabase(str(path)) + try: + await db.initialize() + await db.initialize() + job = await db.get_cron_job("legacy") + assert job.cron_expression == "*/40 * * * *" + assert job.interval_seconds is None + assert job.interval_anchor_at is None + trigger = CronJobManager(db)._build_trigger(job) + assert isinstance(trigger, CronTrigger) + first = trigger.get_next_fire_time( + None, datetime(2026, 10, 3, tzinfo=timezone.utc) + ) + second = trigger.get_next_fire_time(first, first) + third = trigger.get_next_fire_time(second, second) + assert (second - first).total_seconds() == 2400 + assert (third - second).total_seconds() == 1200 + finally: + await db.engine.dispose() diff --git a/tests/unit/test_cron_tools.py b/tests/unit/test_cron_tools.py index fe95703c27..ef0d64c7aa 100644 --- a/tests/unit/test_cron_tools.py +++ b/tests/unit/test_cron_tools.py @@ -116,6 +116,7 @@ async def test_future_task_edit_updates_existing_job(): job_type="active_agent", run_once=False, cron_expression="0 8 * * *", + interval_seconds=None, payload={ "session": "test:private:session", "sender_id": "user-1",