[feat]:[FL-65][新增定时任务管理页面]

Co-authored-by: multica-agent <github@multica.ai>

api/app/api/router.py

@@ -10,6 +10,7 @@ from .v1.fl_analysis import router as fl_analysis_router
 from .v1.flower_monitor import router as flower_monitor_router
 from .v1.lightning import router as lightning_router
 from .v1.lines import router as lines_router
+from .v1.scheduled_tasks import router as scheduled_tasks_router
 from .v1.system_params import router as system_params_router
 from .v1.task_monitor import router as task_monitor_router
 from .v1.tower_models import router as tower_models_router
@@ -25,6 +26,7 @@ v1_router.include_router(admin_router)
 v1_router.include_router(admin_files_router)
 v1_router.include_router(atp_models_router)
 v1_router.include_router(task_monitor_router)
+v1_router.include_router(scheduled_tasks_router)
 v1_router.include_router(system_params_router)
 v1_router.include_router(elevation_router)
 v1_router.include_router(fault_recurrence_router)

api/app/api/v1/scheduled_tasks.py

@@ -0,0 +1,87 @@
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session
+
+from ...core.database import get_db
+from ...core.dependencies import CurrentUser, require_any_permission, require_permission
+from ...schemas.scheduled_task import (
+    ScheduledTaskCreateRequest,
+    ScheduledTaskListResponse,
+    ScheduledTaskRunResponse,
+    ScheduledTaskSummary,
+    ScheduledTaskUpdateRequest,
+)
+from ...services.scheduled_task_service import (
+    create_scheduled_task,
+    get_scheduled_task_by_id,
+    list_scheduled_tasks,
+    run_scheduled_task_now,
+    serialize_scheduled_task,
+    update_scheduled_task,
+)
+
+router = APIRouter(prefix="/admin/scheduled-tasks", tags=["admin-scheduled-tasks"])
+
+
+@router.get("", response_model=ScheduledTaskListResponse)
+def get_scheduled_tasks(
+    keyword: str | None = Query(default=None),
+    status_filter: str | None = Query(default=None, alias="status"),
+    _: CurrentUser = Depends(require_any_permission("celery.read", "celery.manage")),
+    db: Session = Depends(get_db),
+) -> ScheduledTaskListResponse:
+    return list_scheduled_tasks(db, keyword=keyword, status_filter=status_filter)
+
+
+@router.post("", response_model=ScheduledTaskSummary)
+def create_scheduled_task_endpoint(
+    payload: ScheduledTaskCreateRequest,
+    current_user: CurrentUser = Depends(require_permission("celery.manage")),
+    db: Session = Depends(get_db),
+) -> ScheduledTaskSummary:
+    try:
+        created = create_scheduled_task(db, payload, actor_user_id=current_user.user.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+    if not created:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Scheduled task key already exists")
+    return created
+
+
+@router.get("/{task_id}", response_model=ScheduledTaskSummary)
+def get_scheduled_task_detail(
+    task_id: int,
+    _: CurrentUser = Depends(require_any_permission("celery.read", "celery.manage")),
+    db: Session = Depends(get_db),
+) -> ScheduledTaskSummary:
+    item = get_scheduled_task_by_id(db, task_id)
+    if not item:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled task not found")
+    return serialize_scheduled_task(item)
+
+
+@router.patch("/{task_id}", response_model=ScheduledTaskSummary)
+def update_scheduled_task_endpoint(
+    task_id: int,
+    payload: ScheduledTaskUpdateRequest,
+    current_user: CurrentUser = Depends(require_permission("celery.manage")),
+    db: Session = Depends(get_db),
+) -> ScheduledTaskSummary:
+    try:
+        updated = update_scheduled_task(db, task_id, payload, actor_user_id=current_user.user.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+    if not updated:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled task not found")
+    return updated
+
+
+@router.post("/{task_id}/run", response_model=ScheduledTaskRunResponse)
+def run_scheduled_task_endpoint(
+    task_id: int,
+    current_user: CurrentUser = Depends(require_permission("celery.manage")),
+    db: Session = Depends(get_db),
+) -> ScheduledTaskRunResponse:
+    result = run_scheduled_task_now(db, task_id, actor_user_id=current_user.user.id)
+    if not result:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled task not found")
+    return result

api/app/core/celery_app.py

@@ -14,6 +14,7 @@ celery_app = Celery(
         "app.tasks.atp_model_tasks",
         "app.tasks.elevation_tasks",
         "app.tasks.fl_analysis_tasks",
+        "app.tasks.scheduled_task_tasks",
         "app.tasks.wine_tasks",
         "app.tasks.worker_registry_tasks",
     ],
@@ -26,6 +27,10 @@ celery_app.conf.update(
             "task": "app.tasks.worker_registry_tasks.sweep_worker_registry_offline",
             "schedule": 30.0,
         },
+        "dispatch-due-scheduled-tasks": {
+            "task": "app.tasks.scheduled_task_tasks.dispatch_due_scheduled_tasks",
+            "schedule": 60.0,
+        },
     },
     enable_utc=True,
     result_serializer="json",

api/app/core/database.py

@@ -416,6 +416,7 @@ def init_db() -> None:
         menu,
         object_group,
         rbac,
+        scheduled_task,
         system_param,
         tower_model,
         tower_profile,

api/app/main.py

@@ -5,7 +5,8 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from .api.router import api_router
 from .core.config import get_settings
-from .core.database import init_db
+from .core.database import SessionLocal, init_db
+from .services.scheduled_task_service import seed_default_scheduled_tasks
 
 settings = get_settings()
 
@@ -13,6 +14,12 @@ settings = get_settings()
 @asynccontextmanager
 async def lifespan(_: FastAPI):
     init_db()
+    db = SessionLocal()
+    try:
+        seed_default_scheduled_tasks(db)
+        db.commit()
+    finally:
+        db.close()
     yield
 
 

api/app/models/__init__.py

@@ -4,7 +4,7 @@ Import all model modules during package initialization so SQLAlchemy can
 resolve string-based relationships regardless of route/service import order.
 """
 
-from . import atp_model, audit_log, auth_session, elevation, file_storage, fl_analysis, lightning_event, lightning_sample, line, line_tower, menu, object_group, rbac, system_param, tower_model, tower_profile, user, wine, worker_registry
+from . import atp_model, audit_log, auth_session, elevation, file_storage, fl_analysis, lightning_event, lightning_sample, line, line_tower, menu, object_group, rbac, scheduled_task, system_param, tower_model, tower_profile, user, wine, worker_registry
 
 __all__ = [
     "atp_model",
@@ -20,6 +20,7 @@ __all__ = [
     "menu",
     "object_group",
     "rbac",
+    "scheduled_task",
     "system_param",
     "tower_model",
     "tower_profile",

api/app/models/scheduled_task.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import JSON, DateTime, ForeignKey, Index, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from ..core.database import Base
+from .base import utcnow
+
+if TYPE_CHECKING:
+    from .user import User
+
+
+class ScheduledTask(Base):
+    __tablename__ = "scheduled_tasks"
+    __table_args__ = (
+        Index("idx_scheduled_tasks_status", "status"),
+        Index("idx_scheduled_tasks_enabled", "enabled"),
+        Index("idx_scheduled_tasks_next_run", "next_run_at"),
+        Index("idx_scheduled_tasks_task_type", "task_type"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    task_key: Mapped[str] = mapped_column(String(128), unique=True, index=True)
+    name: Mapped[str] = mapped_column(String(128), index=True)
+    task_type: Mapped[str] = mapped_column(String(64), index=True)
+    description: Mapped[str | None] = mapped_column(Text(), default="")
+    cron_expression: Mapped[str] = mapped_column(String(128))
+    timezone: Mapped[str] = mapped_column(String(64), default="Asia/Shanghai")
+    retain_days: Mapped[int] = mapped_column(Integer, default=30)
+    enabled: Mapped[bool] = mapped_column(default=True, index=True)
+    status: Mapped[str] = mapped_column(String(32), default="idle", index=True)
+    last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True)
+    next_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True)
+    last_success_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_error_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_error_message: Mapped[str | None] = mapped_column(Text())
+    last_result_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
+    run_count: Mapped[int] = mapped_column(Integer, default=0)
+    create_user: Mapped[str | None] = mapped_column(
+        String(36),
+        ForeignKey("users.user_id", ondelete="SET NULL"),
+        index=True,
+    )
+    update_user: Mapped[str | None] = mapped_column(
+        String(36),
+        ForeignKey("users.user_id", ondelete="SET NULL"),
+        index=True,
+    )
+    create_date: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True)
+    update_date: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=utcnow,
+        onupdate=utcnow,
+    )
+
+    creator: Mapped[User | None] = relationship(
+        "User",
+        foreign_keys=[create_user],
+        lazy="selectin",
+    )
+    updater: Mapped[User | None] = relationship(
+        "User",
+        foreign_keys=[update_user],
+        lazy="selectin",
+    )

api/app/schemas/scheduled_task.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+from .auth import UserPublic
+
+
+ScheduledTaskStatus = Literal["idle", "queued", "running", "success", "failed", "disabled"]
+ScheduledTaskType = Literal["syslog_cleanup"]
+
+
+class ScheduledTaskSummary(BaseModel):
+    id: int
+    task_key: str
+    name: str
+    task_type: ScheduledTaskType
+    description: str | None = None
+    cron_expression: str
+    timezone: str
+    retain_days: int
+    enabled: bool
+    status: ScheduledTaskStatus
+    last_run_at: datetime | None = None
+    next_run_at: datetime | None = None
+    last_success_at: datetime | None = None
+    last_error_at: datetime | None = None
+    last_error_message: str | None = None
+    last_result_json: dict[str, Any] = Field(default_factory=dict)
+    run_count: int
+    create_user: str | None = None
+    update_user: str | None = None
+    create_date: datetime
+    update_date: datetime
+    creator: UserPublic | None = None
+    updater: UserPublic | None = None
+
+
+class ScheduledTaskListResponse(BaseModel):
+    items: list[ScheduledTaskSummary]
+    total: int
+
+
+class ScheduledTaskCreateRequest(BaseModel):
+    task_key: str = Field(min_length=2, max_length=128)
+    name: str = Field(min_length=2, max_length=128)
+    task_type: ScheduledTaskType
+    description: str | None = Field(default="", max_length=4000)
+    cron_expression: str = Field(min_length=9, max_length=128)
+    timezone: str = Field(default="Asia/Shanghai", min_length=2, max_length=64)
+    retain_days: int = Field(default=30, ge=1, le=3650)
+    enabled: bool = True
+
+
+class ScheduledTaskUpdateRequest(BaseModel):
+    name: str | None = Field(default=None, min_length=2, max_length=128)
+    description: str | None = Field(default=None, max_length=4000)
+    cron_expression: str | None = Field(default=None, min_length=9, max_length=128)
+    timezone: str | None = Field(default=None, min_length=2, max_length=64)
+    retain_days: int | None = Field(default=None, ge=1, le=3650)
+    enabled: bool | None = None
+
+
+class ScheduledTaskRunResponse(BaseModel):
+    success: bool
+    task: ScheduledTaskSummary
+    celery_task_id: str | None = None

api/app/services/admin_service.py

@@ -418,6 +418,7 @@ def delete_menu(db: Session, menu_id: int) -> bool:
         "admin.lightning_distribution",
         "admin.workers",
         "admin.task_monitor",
+        "admin.scheduled_tasks",
         "admin.atp_models",
         "admin.tower_models",
         "admin.files",

api/app/services/legacy_admin_rbac_service.py

@@ -82,6 +82,7 @@ PROTECTED_MENU_CODES = {
     "admin.lightning_distribution",
     "admin.workers",
     "admin.task_monitor",
+    "admin.scheduled_tasks",
     "admin.atp_models",
     "admin.data_query",
     "admin.hot_search",

api/app/services/legacy_authz_service.py

@@ -94,6 +94,7 @@ MENU_CODE_PERMISSION_MAP: dict[str, set[str]] = {
     "admin.tower_models": {"tower_model.read", "tower_model.manage"},
     "admin.workers": {"celery.read", "celery.manage"},
     "admin.task_monitor": {"celery.read", "celery.manage"},
+    "admin.scheduled_tasks": {"celery.read", "celery.manage"},
     "admin.atp_models": {"atp.read", "atp.manage", "atp.run"},
     "admin.fault_recurrence": {"line.read", "line.manage", "tower.read", "tower.manage"},
     "admin.lightning_currents": {"lightning.read", "lightning.manage"},
@@ -157,6 +158,17 @@ SYNTHETIC_LEGACY_MENU_ROWS: list[dict[str, Any]] = [
         "seq": 54,
         "state": "ENABLED",
     },
+    {
+        "menu_id": "admin.scheduled_tasks",
+        "menu_name": "admin.scheduled_tasks",
+        "menu_label": "定时任务管理",
+        "menu_type": "MENU",
+        "parent_id": None,
+        "url": "/admin/scheduled-tasks",
+        "menu_icon": "CalendarClock",
+        "seq": 55,
+        "state": "ENABLED",
+    },
     {
         "menu_id": "admin.atp_models",
         "menu_name": "admin.atp_models",
@@ -165,7 +177,7 @@ SYNTHETIC_LEGACY_MENU_ROWS: list[dict[str, Any]] = [
         "parent_id": None,
         "url": "/admin/atp-models",
         "menu_icon": "Experiment",
-        "seq": 55,
+        "seq": 56,
         "state": "ENABLED",
     },
     {

api/app/services/scheduled_task_service.py

@@ -0,0 +1,506 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from celery.schedules import crontab, crontab_parser
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session, selectinload
+
+from ..core.config import get_settings
+from ..core.database import SessionLocal
+from ..models.audit_log import AuditLog
+from ..models.base import utcnow
+from ..models.scheduled_task import ScheduledTask
+from ..schemas.scheduled_task import (
+    ScheduledTaskCreateRequest,
+    ScheduledTaskListResponse,
+    ScheduledTaskRunResponse,
+    ScheduledTaskSummary,
+    ScheduledTaskUpdateRequest,
+)
+from .push_service import publish_topic
+from .user_service import serialize_user
+
+SCHEDULED_TASK_TOPIC = "admin.scheduled-tasks"
+SYSLOG_CLEANUP_TASK_KEY = "syslog.cleanup.default"
+SUPPORTED_TASK_TYPES = {"syslog_cleanup"}
+CRON_FIELD_LIMITS: tuple[tuple[int, int], ...] = (
+    (60, 0),
+    (24, 0),
+    (31, 1),
+    (12, 1),
+    (7, 0),
+)
+
+
+@dataclass(slots=True)
+class ScheduledTaskExecutionResult:
+    detail: str
+    payload: dict[str, int | str]
+
+
+def _scheduled_task_stmt():
+    return select(ScheduledTask).options(
+        selectinload(ScheduledTask.creator),
+        selectinload(ScheduledTask.updater),
+    )
+
+
+def serialize_scheduled_task(item: ScheduledTask) -> ScheduledTaskSummary:
+    return ScheduledTaskSummary(
+        id=item.id,
+        task_key=item.task_key,
+        name=item.name,
+        task_type=item.task_type,
+        description=item.description,
+        cron_expression=item.cron_expression,
+        timezone=item.timezone,
+        retain_days=item.retain_days,
+        enabled=item.enabled,
+        status=item.status,
+        last_run_at=item.last_run_at,
+        next_run_at=item.next_run_at,
+        last_success_at=item.last_success_at,
+        last_error_at=item.last_error_at,
+        last_error_message=item.last_error_message,
+        last_result_json=item.last_result_json or {},
+        run_count=item.run_count,
+        create_user=item.create_user,
+        update_user=item.update_user,
+        create_date=item.create_date,
+        update_date=item.update_date,
+        creator=serialize_user(item.creator) if item.creator else None,
+        updater=serialize_user(item.updater) if item.updater else None,
+    )
+
+
+def list_scheduled_tasks(
+    db: Session,
+    *,
+    keyword: str | None,
+    status_filter: str | None,
+) -> ScheduledTaskListResponse:
+    stmt = _scheduled_task_stmt()
+    total_stmt = select(func.count()).select_from(ScheduledTask)
+
+    normalized_keyword = (keyword or "").strip()
+    if normalized_keyword:
+        like = f"%{normalized_keyword}%"
+        criteria = or_(
+            ScheduledTask.task_key.ilike(like),
+            ScheduledTask.name.ilike(like),
+            ScheduledTask.description.ilike(like),
+            ScheduledTask.cron_expression.ilike(like),
+        )
+        stmt = stmt.where(criteria)
+        total_stmt = total_stmt.where(criteria)
+
+    normalized_status = (status_filter or "").strip().lower()
+    if normalized_status in {"enabled", "disabled"}:
+        enabled = normalized_status == "enabled"
+        stmt = stmt.where(ScheduledTask.enabled.is_(enabled))
+        total_stmt = total_stmt.where(ScheduledTask.enabled.is_(enabled))
+    elif normalized_status in {"idle", "queued", "running", "success", "failed", "disabled"}:
+        stmt = stmt.where(ScheduledTask.status == normalized_status)
+        total_stmt = total_stmt.where(ScheduledTask.status == normalized_status)
+
+    total = db.scalar(total_stmt) or 0
+    items = db.execute(
+        stmt.order_by(ScheduledTask.enabled.desc(), ScheduledTask.next_run_at.asc(), ScheduledTask.id.asc())
+    ).scalars().all()
+    return ScheduledTaskListResponse(items=[serialize_scheduled_task(item) for item in items], total=total)
+
+
+def get_scheduled_task_by_id(db: Session, task_id: int) -> ScheduledTask | None:
+    return db.execute(_scheduled_task_stmt().where(ScheduledTask.id == task_id)).scalar_one_or_none()
+
+
+def get_scheduled_task_by_key(db: Session, task_key: str) -> ScheduledTask | None:
+    return db.execute(_scheduled_task_stmt().where(ScheduledTask.task_key == task_key)).scalar_one_or_none()
+
+
+def create_scheduled_task(
+    db: Session,
+    payload: ScheduledTaskCreateRequest,
+    *,
+    actor_user_id: str | None,
+) -> ScheduledTaskSummary | None:
+    task_key = payload.task_key.strip()
+    if not task_key:
+        return None
+    if db.scalar(select(ScheduledTask.id).where(ScheduledTask.task_key == task_key)):
+        return None
+
+    normalized_type = payload.task_type.strip()
+    _validate_task_definition(
+        cron_expression=payload.cron_expression,
+        timezone_name=payload.timezone,
+        task_type=normalized_type,
+    )
+
+    task = ScheduledTask(
+        task_key=task_key,
+        name=payload.name.strip(),
+        task_type=normalized_type,
+        description=(payload.description or "").strip(),
+        cron_expression=normalize_cron_expression(payload.cron_expression),
+        timezone=payload.timezone.strip(),
+        retain_days=payload.retain_days,
+        enabled=payload.enabled,
+        status="idle" if payload.enabled else "disabled",
+        next_run_at=compute_next_run_at(
+            normalize_cron_expression(payload.cron_expression),
+            payload.timezone.strip(),
+            from_time=None,
+        ) if payload.enabled else None,
+        create_user=actor_user_id,
+        update_user=actor_user_id,
+    )
+    db.add(task)
+    db.commit()
+    saved = get_scheduled_task_by_id(db, task.id)
+    if not saved:
+        return None
+    _publish_scheduled_task_changed(
+        action="created",
+        task=saved,
+        requires_refetch=["/api/v1/admin/scheduled-tasks"],
+    )
+    return serialize_scheduled_task(saved)
+
+
+def update_scheduled_task(
+    db: Session,
+    task_id: int,
+    payload: ScheduledTaskUpdateRequest,
+    *,
+    actor_user_id: str,
+) -> ScheduledTaskSummary | None:
+    item = get_scheduled_task_by_id(db, task_id)
+    if not item:
+        return None
+
+    update_data = payload.model_dump(exclude_unset=True)
+    next_name = str(update_data.get("name", item.name)).strip()
+    next_description = (
+        (str(update_data["description"]) if update_data["description"] is not None else "")
+        if "description" in update_data else (item.description or "")
+    ).strip()
+    next_cron = normalize_cron_expression(str(update_data.get("cron_expression", item.cron_expression)))
+    next_timezone = str(update_data.get("timezone", item.timezone)).strip()
+    next_retain_days = int(update_data.get("retain_days", item.retain_days))
+    next_enabled = bool(update_data.get("enabled", item.enabled))
+
+    _validate_task_definition(
+        cron_expression=next_cron,
+        timezone_name=next_timezone,
+        task_type=item.task_type,
+    )
+
+    item.name = next_name
+    item.description = next_description
+    item.cron_expression = next_cron
+    item.timezone = next_timezone
+    item.retain_days = next_retain_days
+    item.enabled = next_enabled
+    item.status = "disabled" if not next_enabled else ("idle" if item.status == "disabled" else item.status)
+    item.next_run_at = compute_next_run_at(next_cron, next_timezone, from_time=None) if next_enabled else None
+    item.update_user = actor_user_id
+    db.commit()
+    saved = get_scheduled_task_by_id(db, task_id)
+    if not saved:
+        return None
+    _publish_scheduled_task_changed(
+        action="updated",
+        task=saved,
+        requires_refetch=["/api/v1/admin/scheduled-tasks", f"/api/v1/admin/scheduled-tasks/{saved.id}"],
+    )
+    return serialize_scheduled_task(saved)
+
+
+def run_scheduled_task_now(
+    db: Session,
+    task_id: int,
+    *,
+    actor_user_id: str,
+) -> ScheduledTaskRunResponse | None:
+    item = get_scheduled_task_by_id(db, task_id)
+    if not item:
+        return None
+
+    item.status = "queued"
+    item.update_user = actor_user_id
+    db.commit()
+
+    from ..tasks.scheduled_task_tasks import execute_scheduled_task_job
+
+    result = execute_scheduled_task_job.delay(item.id, actor_user_id)
+    saved = get_scheduled_task_by_id(db, task_id)
+    if not saved:
+        return None
+    _publish_scheduled_task_changed(
+        action="queued",
+        task=saved,
+        requires_refetch=["/api/v1/admin/scheduled-tasks"],
+    )
+    return ScheduledTaskRunResponse(
+        success=True,
+        task=serialize_scheduled_task(saved),
+        celery_task_id=result.id,
+    )
+
+
+def seed_default_scheduled_tasks(db: Session) -> None:
+    if get_scheduled_task_by_key(db, SYSLOG_CLEANUP_TASK_KEY):
+        return
+
+    payload = ScheduledTaskCreateRequest(
+        task_key=SYSLOG_CLEANUP_TASK_KEY,
+        name="系统日志定时清理",
+        task_type="syslog_cleanup",
+        description="按保留天数自动清理历史系统日志，避免审计表持续膨胀。",
+        cron_expression="0 3 * * *",
+        timezone=get_settings().celery_timezone,
+        retain_days=30,
+        enabled=True,
+    )
+    create_scheduled_task(db, payload, actor_user_id=None)
+
+
+def dispatch_due_scheduled_tasks(*, actor_user_id: str = "system") -> dict[str, int]:
+    now = utcnow()
+    queued_count = 0
+    scanned_count = 0
+    db = SessionLocal()
+    try:
+        items = db.execute(
+            _scheduled_task_stmt().where(
+                ScheduledTask.enabled.is_(True),
+                ScheduledTask.next_run_at.is_not(None),
+                ScheduledTask.next_run_at <= now,
+            ).order_by(ScheduledTask.next_run_at.asc(), ScheduledTask.id.asc())
+        ).scalars().all()
+
+        from ..tasks.scheduled_task_tasks import execute_scheduled_task_job
+
+        for item in items:
+            scanned_count += 1
+            item.status = "queued"
+            item.update_user = actor_user_id
+            item.next_run_at = compute_next_run_at(item.cron_expression, item.timezone, from_time=now)
+            db.commit()
+            execute_scheduled_task_job.delay(item.id, actor_user_id)
+            queued_count += 1
+            _publish_scheduled_task_changed(
+                action="queued",
+                task=item,
+                requires_refetch=["/api/v1/admin/scheduled-tasks"],
+            )
+        return {
+            "scanned_count": scanned_count,
+            "queued_count": queued_count,
+        }
+    finally:
+        db.close()
+
+
+def execute_scheduled_task(task_id: int, *, actor_user_id: str = "system") -> dict[str, object]:
+    db = SessionLocal()
+    try:
+        item = get_scheduled_task_by_id(db, task_id)
+        if not item:
+            return {"success": False, "detail": "scheduled task not found", "task_id": task_id}
+        return _execute_scheduled_task_with_session(db, item, actor_user_id=actor_user_id)
+    finally:
+        db.close()
+
+
+def cleanup_audit_logs(db: Session, *, retain_days: int) -> int:
+    threshold = utcnow() - timedelta(days=retain_days)
+    candidate_ids = db.scalars(
+        select(AuditLog.id).where(AuditLog.created_at < threshold).order_by(AuditLog.id.asc()).limit(5000)
+    ).all()
+    if not candidate_ids:
+        return 0
+    deleted = db.query(AuditLog).filter(AuditLog.id.in_(candidate_ids)).delete(synchronize_session=False)
+    return int(deleted or 0)
+
+
+def normalize_cron_expression(value: str) -> str:
+    fields = value.strip().split()
+    if len(fields) != 5:
+        raise ValueError("cron expression must contain exactly 5 fields")
+    return " ".join(field.strip() for field in fields)
+
+
+def compute_next_run_at(
+    cron_expression: str,
+    timezone_name: str,
+    *,
+    from_time: datetime | None,
+) -> datetime:
+    tz = _get_zoneinfo(timezone_name)
+    base = (from_time or utcnow()).astimezone(tz).replace(second=0, microsecond=0) + timedelta(minutes=1)
+    minutes, hours, days, months, weekdays = _parse_cron_expression(cron_expression)
+
+    for offset in range(0, 366 * 24 * 60):
+        candidate = base + timedelta(minutes=offset)
+        if candidate.minute not in minutes:
+            continue
+        if candidate.hour not in hours:
+            continue
+        if candidate.day not in days:
+            continue
+        if candidate.month not in months:
+            continue
+        weekday = candidate.isoweekday() % 7
+        if weekday not in weekdays:
+            continue
+        return candidate.astimezone(ZoneInfo("UTC"))
+    raise ValueError("unable to compute next run time within 1 year")
+
+
+def _execute_scheduled_task_with_session(
+    db: Session,
+    item: ScheduledTask,
+    *,
+    actor_user_id: str,
+) -> dict[str, object]:
+    now = utcnow()
+    item.status = "running"
+    item.last_run_at = now
+    item.update_user = actor_user_id
+    db.commit()
+    _publish_scheduled_task_changed(
+        action="running",
+        task=item,
+        requires_refetch=["/api/v1/admin/scheduled-tasks"],
+    )
+
+    try:
+        result = _run_task_handler(db, item)
+        now = utcnow()
+        item.status = "success"
+        item.last_success_at = now
+        item.last_error_at = None
+        item.last_error_message = None
+        item.last_result_json = result.payload
+        item.run_count += 1
+        item.update_user = actor_user_id
+        if item.enabled:
+            item.next_run_at = compute_next_run_at(item.cron_expression, item.timezone, from_time=now)
+        db.add(
+            AuditLog(
+                user_id=actor_user_id if actor_user_id != "system" else None,
+                action="scheduled_task.run",
+                detail=f"{item.task_key}: {result.detail}",
+            )
+        )
+        db.commit()
+        _publish_scheduled_task_changed(
+            action="success",
+            task=item,
+            requires_refetch=["/api/v1/admin/scheduled-tasks", "/api/v1/admin/audit-logs"],
+        )
+        return {
+            "success": True,
+            "detail": result.detail,
+            "task": serialize_scheduled_task(item).model_dump(mode="json"),
+        }
+    except Exception as exc:
+        now = utcnow()
+        item.status = "failed"
+        item.last_error_at = now
+        item.last_error_message = str(exc)
+        item.update_user = actor_user_id
+        if item.enabled:
+            item.next_run_at = compute_next_run_at(item.cron_expression, item.timezone, from_time=now)
+        db.add(
+            AuditLog(
+                user_id=actor_user_id if actor_user_id != "system" else None,
+                action="scheduled_task.run_failed",
+                detail=f"{item.task_key}: {exc}",
+            )
+        )
+        db.commit()
+        _publish_scheduled_task_changed(
+            action="failed",
+            task=item,
+            requires_refetch=["/api/v1/admin/scheduled-tasks", "/api/v1/admin/audit-logs"],
+        )
+        raise
+
+
+def _run_task_handler(db: Session, item: ScheduledTask) -> ScheduledTaskExecutionResult:
+    if item.task_type == "syslog_cleanup":
+        deleted_count = cleanup_audit_logs(db, retain_days=item.retain_days)
+        return ScheduledTaskExecutionResult(
+            detail=f"已清理 {deleted_count} 条系统日志",
+            payload={"deleted_count": deleted_count, "retain_days": item.retain_days},
+        )
+    raise ValueError(f"unsupported scheduled task type: {item.task_type}")
+
+
+def _parse_cron_expression(value: str) -> tuple[set[int], set[int], set[int], set[int], set[int]]:
+    fields = normalize_cron_expression(value).split()
+    parsed: list[set[int]] = []
+    for field, (max_value, min_value) in zip(fields, CRON_FIELD_LIMITS, strict=True):
+        parsed.append({int(item) for item in crontab_parser(max_value, min_value).parse(field)})
+    return tuple(parsed)  # type: ignore[return-value]
+
+
+def _validate_task_definition(*, cron_expression: str, timezone_name: str, task_type: str) -> None:
+    if task_type not in SUPPORTED_TASK_TYPES:
+        raise ValueError(f"unsupported task type: {task_type}")
+    normalized_cron = normalize_cron_expression(cron_expression)
+    _parse_cron_expression(normalized_cron)
+    _get_zoneinfo(timezone_name)
+    minute, hour, day_of_month, month_of_year, day_of_week = normalized_cron.split()
+    crontab(
+        minute=minute,
+        hour=hour,
+        day_of_month=day_of_month,
+        month_of_year=month_of_year,
+        day_of_week=day_of_week,
+    )
+
+
+def _get_zoneinfo(value: str) -> ZoneInfo:
+    normalized = value.strip()
+    if not normalized:
+        raise ValueError("timezone is required")
+    try:
+        return ZoneInfo(normalized)
+    except ZoneInfoNotFoundError as exc:
+        raise ValueError(f"unknown timezone: {normalized}") from exc
+
+
+def _publish_scheduled_task_changed(
+    *,
+    action: str,
+    task: ScheduledTask,
+    requires_refetch: list[str],
+) -> None:
+    _fire_and_forget(
+        publish_topic(
+            SCHEDULED_TASK_TOPIC,
+            name="scheduled_tasks.changed",
+            payload={"action": action, "task_id": task.id, "task_key": task.task_key},
+            requires_refetch=requires_refetch,
+            dedupe_key=f"scheduled-tasks:{action}:{task.id}",
+        )
+    )
+
+
+def _fire_and_forget(coro: object) -> None:
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        close = getattr(coro, "close", None)
+        if callable(close):
+            close()
+        return
+    loop.create_task(coro)

api/app/services/seed_service.py

@@ -10,6 +10,7 @@ from ..models.menu import Menu
 from ..models.rbac import Permission, Role
 from ..models.tower_model import TowerModel
 from ..models.user import User
+from .scheduled_task_service import seed_default_scheduled_tasks
 from .tower_model_service import seed_tower_models_from_legacy
 
 settings = get_settings()
@@ -296,6 +297,19 @@ DEFAULT_MENUS: list[dict[str, object]] = [
         "cacheable": False,
         "permission_code": "celery.read",
     },
+    {
+        "code": "admin.scheduled_tasks",
+        "name": "定时任务管理",
+        "path": "/admin/scheduled-tasks",
+        "icon": "CalendarClock",
+        "parent_code": None,
+        "type": "menu",
+        "sort_order": 55,
+        "status": "enabled",
+        "visible": True,
+        "cacheable": False,
+        "permission_code": "celery.read",
+    },
     {
         "code": "admin.atp_models",
         "name": "ATP模型管理",
@@ -303,7 +317,7 @@ DEFAULT_MENUS: list[dict[str, object]] = [
         "icon": "Experiment",
         "parent_code": None,
         "type": "menu",
-        "sort_order": 55,
+        "sort_order": 56,
         "status": "enabled",
         "visible": True,
         "cacheable": False,
@@ -316,7 +330,7 @@ DEFAULT_MENUS: list[dict[str, object]] = [
         "icon": "Apartment",
         "parent_code": None,
         "type": "menu",
-        "sort_order": 56,
+        "sort_order": 57,
         "status": "enabled",
         "visible": True,
         "cacheable": False,
@@ -329,7 +343,7 @@ DEFAULT_MENUS: list[dict[str, object]] = [
         "icon": "FolderTree",
         "parent_code": None,
         "type": "menu",
-        "sort_order": 57,
+        "sort_order": 58,
         "status": "enabled",
         "visible": True,
         "cacheable": False,
@@ -342,7 +356,7 @@ DEFAULT_MENUS: list[dict[str, object]] = [
         "icon": "Database",
         "parent_code": None,
         "type": "menu",
-        "sort_order": 58,
+        "sort_order": 59,
         "status": "enabled",
         "visible": True,
         "cacheable": False,
@@ -355,7 +369,7 @@ DEFAULT_MENUS: list[dict[str, object]] = [
         "icon": "FileText",
         "parent_code": None,
         "type": "menu",
-        "sort_order": 59,
+        "sort_order": 60,
         "status": "enabled",
         "visible": True,
         "cacheable": False,
@@ -428,6 +442,7 @@ ROLE_MENU_BINDINGS: dict[str, list[str]] = {
         "admin.lightning_distribution",
         "admin.workers",
         "admin.task_monitor",
+        "admin.scheduled_tasks",
         "admin.atp_models",
         "admin.tower_models",
         "admin.files",
@@ -507,6 +522,9 @@ def seed_defaults(db: Session, *, force: bool = False) -> SeedDefaultsResult:
         legacy_summary.updated += legacy_seed_result.updated_models
         legacy_summary.unchanged += legacy_seed_result.skipped_models
 
+    seed_default_scheduled_tasks(db)
+    db.commit()
+
     return result
 
 

api/app/services/topic_registry.py

@@ -26,6 +26,7 @@ TOPIC_RULES: dict[str, TopicRule] = {
     "admin.elevation": TopicRule(any_permission_codes={"elevation.read", "elevation.manage"}),
     "admin.atp-models": TopicRule(any_permission_codes={"atp.read", "atp.run", "atp.manage"}),
     "admin.audit_logs": TopicRule(any_permission_codes={"menu.read", "menu.manage"}),
+    "admin.scheduled-tasks": TopicRule(any_permission_codes={"celery.read", "celery.manage"}),
 }
 
 

api/app/tasks/scheduled_task_tasks.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from ..core.celery_app import celery_app
+from ..services.scheduled_task_service import dispatch_due_scheduled_tasks, execute_scheduled_task
+
+
+@celery_app.task(name="app.tasks.scheduled_task_tasks.dispatch_due_scheduled_tasks")
+def dispatch_due_scheduled_tasks_job() -> dict[str, int]:
+    return dispatch_due_scheduled_tasks(actor_user_id="system")
+
+
+@celery_app.task(name="app.tasks.scheduled_task_tasks.execute_scheduled_task_job")
+def execute_scheduled_task_job(task_id: int, actor_user_id: str = "system") -> dict[str, object]:
+    return execute_scheduled_task(task_id, actor_user_id=actor_user_id)

api/tests/test_legacy_module_cleanup_contract.py

@@ -63,6 +63,15 @@ class LegacyModuleCleanupContractTest(unittest.TestCase):
         self.assertNotIn("admin.todos", PROTECTED_MENU_CODES)
         self.assertNotIn("admin.jwt_generator", PROTECTED_MENU_CODES)
 
+    def test_new_scheduled_task_dispatcher_is_registered(self) -> None:
+        include = set(celery_app.conf.include or [])
+        self.assertIn("app.tasks.scheduled_task_tasks", include)
+
+        beat_schedule = celery_app.conf.beat_schedule or {}
+        self.assertIn("dispatch-due-scheduled-tasks", beat_schedule)
+        entry = beat_schedule["dispatch-due-scheduled-tasks"]
+        self.assertEqual(entry.get("task"), "app.tasks.scheduled_task_tasks.dispatch_due_scheduled_tasks")
+
 
 if __name__ == "__main__":
     unittest.main()

api/tests/test_scheduled_task_service.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.core.database import Base
+from app.models.audit_log import AuditLog
+from app.models.scheduled_task import ScheduledTask
+from app.schemas.scheduled_task import ScheduledTaskCreateRequest, ScheduledTaskUpdateRequest
+from app.services.scheduled_task_service import (
+    cleanup_audit_logs,
+    compute_next_run_at,
+    create_scheduled_task,
+    get_scheduled_task_by_key,
+    seed_default_scheduled_tasks,
+    update_scheduled_task,
+)
+
+
+def _build_sessionmaker(*tables):
+    engine = create_engine(
+        "sqlite+pysqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine, tables=list(tables))
+    return sessionmaker(bind=engine, autocommit=False, autoflush=False, expire_on_commit=False)
+
+
+def test_seed_default_scheduled_tasks_creates_syslog_cleanup_task() -> None:
+    testing_session = _build_sessionmaker(ScheduledTask.__table__)
+    session: Session = testing_session()
+    try:
+        seed_default_scheduled_tasks(session)
+        created = get_scheduled_task_by_key(session, "syslog.cleanup.default")
+        assert created is not None
+        assert created.task_type == "syslog_cleanup"
+        assert created.enabled is True
+        assert created.retain_days == 30
+        assert created.next_run_at is not None
+    finally:
+        session.close()
+
+
+def test_create_and_update_scheduled_task_recomputes_next_run() -> None:
+    testing_session = _build_sessionmaker(ScheduledTask.__table__)
+    session: Session = testing_session()
+    try:
+        created = create_scheduled_task(
+            session,
+            ScheduledTaskCreateRequest(
+                task_key="syslog.cleanup.weekly",
+                name="每周日志清理",
+                task_type="syslog_cleanup",
+                description="weekly cleanup",
+                cron_expression="0 2 * * 1",
+                timezone="Asia/Shanghai",
+                retain_days=14,
+                enabled=True,
+            ),
+            actor_user_id=None,
+        )
+        assert created is not None
+        original_next_run = created.next_run_at
+        assert original_next_run is not None
+
+        updated = update_scheduled_task(
+            session,
+            created.id,
+            ScheduledTaskUpdateRequest(cron_expression="0 4 * * 1", retain_days=21),
+            actor_user_id=None,
+        )
+        assert updated is not None
+        assert updated.retain_days == 21
+        assert updated.next_run_at is not None
+        assert updated.next_run_at != original_next_run
+    finally:
+        session.close()
+
+
+def test_cleanup_audit_logs_only_removes_expired_rows() -> None:
+    testing_session = _build_sessionmaker(AuditLog.__table__)
+    session: Session = testing_session()
+    try:
+        expired = AuditLog(user_id=None, action="expired", detail="old")
+        fresh = AuditLog(user_id=None, action="fresh", detail="new")
+        session.add_all([expired, fresh])
+        session.flush()
+        expired.created_at = expired.created_at - timedelta(days=45)
+        fresh.created_at = fresh.created_at - timedelta(days=5)
+        session.commit()
+
+        deleted_count = cleanup_audit_logs(session, retain_days=30)
+        session.commit()
+
+        assert deleted_count == 1
+        remaining_actions = session.scalars(select(AuditLog.action).order_by(AuditLog.id.asc())).all()
+        assert remaining_actions == ["fresh"]
+    finally:
+        session.close()
+
+
+def test_compute_next_run_at_returns_future_utc_timestamp() -> None:
+    next_run = compute_next_run_at("0 3 * * *", "Asia/Shanghai", from_time=None)
+    assert next_run.tzinfo is not None
+    assert next_run > datetime.now(timezone.utc)

api/tests/test_seed_defaults_contract.py

@@ -19,6 +19,7 @@ from api.app.api.v1.admin import router as admin_router
 from api.app.core.database import Base, get_db, init_db
 from api.app.core.dependencies import CurrentUser, get_current_user
 from api.app.models.menu import Menu
+from api.app.models.scheduled_task import ScheduledTask
 from api.app.models.user import User
 from api.app.services.legacy_authz_service import DEFAULT_ADMIN_PERMISSION_CODES, SYNTHETIC_LEGACY_MENU_ROWS
 from api.app.services.seed_service import DEFAULT_MENUS, DEFAULT_PERMISSIONS, DEFAULT_ROLES, SeedDefaultsResult, seed_defaults
@@ -165,6 +166,20 @@ class SeedDefaultsServiceTest(DatabaseFixtureTestCase):
         self.assertEqual(str(default_menu["path"]), "/admin/atp-models")
         self.assertEqual(menu.path, "/admin/atp-models")
 
+    def test_seed_defaults_include_scheduled_tasks_menu_and_default_task(self) -> None:
+        self._run_seed()
+
+        menu = self._load_menu("admin.scheduled_tasks")
+        default_menu = DEFAULT_MENU_BY_CODE["admin.scheduled_tasks"]
+        scheduled_task = self.session.scalar(
+            select(ScheduledTask).where(ScheduledTask.task_key == "syslog.cleanup.default")
+        )
+
+        self.assertEqual(str(default_menu["path"]), "/admin/scheduled-tasks")
+        self.assertEqual(menu.path, "/admin/scheduled-tasks")
+        self.assertIsNotNone(scheduled_task)
+        self.assertTrue(scheduled_task.enabled if scheduled_task else False)
+
 
 class SeedDefaultsEndpointTest(DatabaseFixtureTestCase):
     def setUp(self) -> None:

web/src/app/admin/layout.tsx

@@ -6,6 +6,7 @@ import { usePathname } from "next/navigation";
 import Icon, {
   ApartmentOutlined,
   AppstoreOutlined,
+  CalendarOutlined,
   ConsoleSqlOutlined,
   CompressOutlined,
   DatabaseOutlined,
@@ -107,6 +108,7 @@ const MENU_ICON_COMPONENTS = {
   Map: GlobalOutlined,
   DeploymentUnitOutlined,
   RadarChart: RadarChartOutlined,
+  CalendarClock: CalendarOutlined,
   Experiment: ExperimentOutlined,
   Apartment: ApartmentOutlined,
   FolderTree: FolderOpenOutlined,
@@ -121,6 +123,7 @@ const MENU_ICON_COMPONENTS = {
   ThunderboltOutlined,
   GlobalOutlined,
   RadarChartOutlined,
+  CalendarOutlined,
   ExperimentOutlined,
   ApartmentOutlined,
   FolderOpenOutlined,

web/src/app/admin/scheduled-tasks/page.tsx

@@ -0,0 +1,642 @@
+"use client";
+
+import Link from "next/link";
+import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
+import { useCallback, useEffect, useMemo, useRef, useState, type CSSProperties } from "react";
+import {
+  Button,
+  Card,
+  Empty,
+  Form,
+  Input,
+  InputNumber,
+  Modal,
+  Select,
+  Space,
+  Spin,
+  Switch,
+  Table,
+  Tag,
+  Typography,
+  type CardProps,
+  type TableColumnsType,
+} from "antd";
+import type { ComponentType } from "react";
+
+import { useAuth } from "@/components/auth-provider";
+import { useToastFeedback } from "@/hooks/use-toast-feedback";
+import { useTopicSubscription } from "@/hooks/use-topic-subscription";
+import { readApiError } from "@/lib/api";
+import type { ScheduledTaskListResponse, ScheduledTaskRunResponse, ScheduledTaskSummary } from "@/types/auth";
+
+type StatusFilter = "all" | "enabled" | "disabled" | ScheduledTaskSummary["status"];
+
+type FormState = {
+  task_key: string;
+  name: string;
+  description: string;
+  cron_expression: string;
+  timezone: string;
+  retain_days: number;
+  enabled: boolean;
+};
+
+const EMPTY_FORM: FormState = {
+  task_key: "",
+  name: "",
+  description: "",
+  cron_expression: "0 3 * * *",
+  timezone: "Asia/Shanghai",
+  retain_days: 30,
+  enabled: true,
+};
+
+const STATUS_FILTER_OPTIONS = [
+  { label: "全部", value: "all" },
+  { label: "已启用", value: "enabled" },
+  { label: "已禁用", value: "disabled" },
+  { label: "空闲", value: "idle" },
+  { label: "已排队", value: "queued" },
+  { label: "运行中", value: "running" },
+  { label: "成功", value: "success" },
+  { label: "失败", value: "failed" },
+] as const satisfies ReadonlyArray<{ label: string; value: StatusFilter }>;
+
+const TIMEZONE_OPTIONS = [
+  { label: "Asia/Shanghai", value: "Asia/Shanghai" },
+  { label: "UTC", value: "UTC" },
+] as const;
+
+const AntCard = Card as unknown as ComponentType<CardProps>;
+
+const TABLE_MIN_SCROLL_Y = 180;
+const TABLE_VIEWPORT_GAP = 40;
+const TABLE_FALLBACK_RESERVE = 220;
+
+function formatDateTime(value: string | null | undefined): string {
+  if (!value) {
+    return "-";
+  }
+  const date = new Date(value);
+  if (Number.isNaN(date.getTime())) {
+    return "-";
+  }
+  return date.toLocaleString();
+}
+
+function renderStatus(value: ScheduledTaskSummary["status"]) {
+  if (value === "success") {
+    return <Tag color="green">成功</Tag>;
+  }
+  if (value === "failed") {
+    return <Tag color="red">失败</Tag>;
+  }
+  if (value === "running") {
+    return <Tag color="processing">运行中</Tag>;
+  }
+  if (value === "queued") {
+    return <Tag color="blue">已排队</Tag>;
+  }
+  if (value === "disabled") {
+    return <Tag>已禁用</Tag>;
+  }
+  return <Tag color="default">空闲</Tag>;
+}
+
+function renderTaskType(value: ScheduledTaskSummary["task_type"]) {
+  if (value === "syslog_cleanup") {
+    return "系统日志清理";
+  }
+  return value;
+}
+
+export default function AdminScheduledTasksPage() {
+  const { user, initializing, fetchWithAuth, hasPermission } = useAuth();
+  const queryClient = useQueryClient();
+  const [formApi] = Form.useForm<FormState>();
+
+  const [keyword, setKeyword] = useState("");
+  const [statusFilter, setStatusFilter] = useState<StatusFilter>("all");
+  const [editingId, setEditingId] = useState<number | null>(null);
+  const [editorOpen, setEditorOpen] = useState(false);
+  const [error, setError] = useState("");
+  const [success, setSuccess] = useState("");
+  const [tableScrollY, setTableScrollY] = useState(TABLE_MIN_SCROLL_Y);
+  const [runningId, setRunningId] = useState<number | null>(null);
+  const tableScrollAnchorRef = useRef<HTMLDivElement | null>(null);
+
+  const canRead = hasPermission("celery.read") || hasPermission("celery.manage");
+  const canManage = hasPermission("celery.manage");
+
+  const listPath = useMemo(() => {
+    const params = new URLSearchParams();
+    if (keyword.trim()) {
+      params.set("keyword", keyword.trim());
+    }
+    if (statusFilter !== "all") {
+      params.set("status", statusFilter);
+    }
+    const qs = params.toString();
+    return `/api/v1/admin/scheduled-tasks${qs ? `?${qs}` : ""}`;
+  }, [keyword, statusFilter]);
+
+  const listQuery = useQuery({
+    queryKey: [listPath],
+    enabled: !!user && canRead,
+    queryFn: async () => {
+      const response = await fetchWithAuth(listPath);
+      if (!response.ok) {
+        throw new Error(await readApiError(response));
+      }
+      return (await response.json()) as ScheduledTaskListResponse;
+    },
+  });
+
+  const refreshList = useCallback(async () => {
+    await queryClient.invalidateQueries({
+      predicate: (query) =>
+        Array.isArray(query.queryKey)
+        && typeof query.queryKey[0] === "string"
+        && query.queryKey[0].startsWith("/api/v1/admin/scheduled-tasks"),
+    });
+  }, [queryClient]);
+
+  useTopicSubscription("admin.scheduled-tasks", useCallback(() => {
+    void refreshList();
+  }, [refreshList]));
+
+  const resetForm = useCallback(() => {
+    setEditingId(null);
+    formApi.setFieldsValue(EMPTY_FORM);
+  }, [formApi]);
+
+  const closeEditor = useCallback(() => {
+    setEditorOpen(false);
+    resetForm();
+  }, [resetForm]);
+
+  const startCreate = useCallback(() => {
+    setError("");
+    setSuccess("");
+    resetForm();
+    setEditorOpen(true);
+  }, [resetForm]);
+
+  const startEdit = useCallback((item: ScheduledTaskSummary) => {
+    setError("");
+    setSuccess("");
+    setEditingId(item.id);
+    formApi.setFieldsValue({
+      task_key: item.task_key,
+      name: item.name,
+      description: item.description ?? "",
+      cron_expression: item.cron_expression,
+      timezone: item.timezone,
+      retain_days: item.retain_days,
+      enabled: item.enabled,
+    });
+    setEditorOpen(true);
+  }, [formApi]);
+
+  const saveMutation = useMutation({
+    mutationFn: async () => {
+      if (!canManage) {
+        throw new Error("缺少 celery.manage 权限");
+      }
+
+      const values = await formApi.validateFields();
+      if (!values.name.trim() || !values.task_key.trim()) {
+        throw new Error("任务键和任务名称不能为空");
+      }
+
+      if (editingId === null) {
+        const response = await fetchWithAuth("/api/v1/admin/scheduled-tasks", {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({
+            task_key: values.task_key.trim(),
+            name: values.name.trim(),
+            task_type: "syslog_cleanup",
+            description: values.description,
+            cron_expression: values.cron_expression.trim(),
+            timezone: values.timezone,
+            retain_days: values.retain_days,
+            enabled: values.enabled,
+          }),
+        });
+        if (!response.ok) {
+          throw new Error(await readApiError(response));
+        }
+        return "created" as const;
+      }
+
+      const response = await fetchWithAuth(`/api/v1/admin/scheduled-tasks/${editingId}`, {
+        method: "PATCH",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          name: values.name.trim(),
+          description: values.description,
+          cron_expression: values.cron_expression.trim(),
+          timezone: values.timezone,
+          retain_days: values.retain_days,
+          enabled: values.enabled,
+        }),
+      });
+      if (!response.ok) {
+        throw new Error(await readApiError(response));
+      }
+      return "updated" as const;
+    },
+    onSuccess: async (mode) => {
+      setError("");
+      setSuccess(mode === "created" ? "定时任务已创建" : "定时任务已更新");
+      closeEditor();
+      await refreshList();
+    },
+    onError: (candidate) => {
+      setSuccess("");
+      setError(candidate instanceof Error ? candidate.message : "保存失败");
+    },
+  });
+
+  const runMutation = useMutation({
+    mutationFn: async (item: ScheduledTaskSummary) => {
+      const response = await fetchWithAuth(`/api/v1/admin/scheduled-tasks/${item.id}/run`, {
+        method: "POST",
+      });
+      if (!response.ok) {
+        throw new Error(await readApiError(response));
+      }
+      return (await response.json()) as ScheduledTaskRunResponse;
+    },
+    onSuccess: async (payload) => {
+      setError("");
+      setSuccess(payload.celery_task_id ? `任务已触发，Celery ID: ${payload.celery_task_id}` : "任务已触发");
+      await refreshList();
+    },
+    onError: (candidate) => {
+      setSuccess("");
+      setError(candidate instanceof Error ? candidate.message : "执行失败");
+    },
+  });
+
+  const runTask = useCallback(async (item: ScheduledTaskSummary) => {
+    setRunningId(item.id);
+    try {
+      await runMutation.mutateAsync(item);
+    } finally {
+      setRunningId(null);
+    }
+  }, [runMutation]);
+
+  useToastFeedback({
+    errorMessage: error || (listQuery.error instanceof Error ? listQuery.error.message : ""),
+    successMessage: success,
+    clearError: () => setError(""),
+    clearSuccess: () => setSuccess(""),
+  });
+
+  const items = listQuery.data?.items ?? [];
+
+  const columns = useMemo<TableColumnsType<ScheduledTaskSummary>>(() => {
+    const baseColumns: TableColumnsType<ScheduledTaskSummary> = [
+      {
+        title: "任务",
+        key: "name",
+        width: 260,
+        render: (_, record) => (
+          <Space direction="vertical" size={0}>
+            <Typography.Text strong>{record.name}</Typography.Text>
+            <Typography.Text type="secondary" className="font-mono text-xs">
+              {record.task_key}
+            </Typography.Text>
+          </Space>
+        ),
+      },
+      {
+        title: "类型",
+        dataIndex: "task_type",
+        key: "task_type",
+        width: 140,
+        render: (value: ScheduledTaskSummary["task_type"]) => renderTaskType(value),
+      },
+      {
+        title: "Cron / 时区",
+        key: "cron",
+        width: 220,
+        render: (_, record) => (
+          <Space direction="vertical" size={0}>
+            <Typography.Text className="font-mono text-xs">{record.cron_expression}</Typography.Text>
+            <Typography.Text type="secondary">{record.timezone}</Typography.Text>
+          </Space>
+        ),
+      },
+      {
+        title: "保留天数",
+        dataIndex: "retain_days",
+        key: "retain_days",
+        width: 110,
+      },
+      {
+        title: "状态",
+        key: "status",
+        width: 160,
+        render: (_, record) => (
+          <Space direction="vertical" size={4}>
+            {renderStatus(record.status)}
+            <Tag color={record.enabled ? "green" : "default"}>{record.enabled ? "已启用" : "已禁用"}</Tag>
+          </Space>
+        ),
+      },
+      {
+        title: "最近执行 / 下次执行",
+        key: "runtime",
+        width: 260,
+        render: (_, record) => (
+          <Space direction="vertical" size={0}>
+            <Typography.Text>{formatDateTime(record.last_run_at)}</Typography.Text>
+            <Typography.Text type="secondary">{formatDateTime(record.next_run_at)}</Typography.Text>
+          </Space>
+        ),
+      },
+      {
+        title: "结果",
+        key: "result",
+        render: (_, record) => (
+          <Space direction="vertical" size={0}>
+            <Typography.Text type="secondary">
+              最近成功：{formatDateTime(record.last_success_at)}
+            </Typography.Text>
+            <Typography.Text type={record.last_error_message ? "danger" : "secondary"}>
+              {record.last_error_message || `已累计执行 ${record.run_count} 次`}
+            </Typography.Text>
+          </Space>
+        ),
+      },
+    ];
+
+    if (canManage) {
+      baseColumns.push({
+        title: "操作",
+        key: "actions",
+        fixed: "right",
+        width: 180,
+        render: (_, record) => (
+          <Space size="small">
+            <Button size="small" onClick={() => startEdit(record)}>
+              编辑
+            </Button>
+            <Button
+              size="small"
+              type="primary"
+              loading={runningId === record.id}
+              onClick={() => void runTask(record)}
+            >
+              立即执行
+            </Button>
+          </Space>
+        ),
+      });
+    }
+
+    return baseColumns;
+  }, [canManage, runTask, runningId, startEdit]);
+
+  const updateTableScrollY = useCallback(() => {
+    if (typeof window === "undefined") {
+      return;
+    }
+    const anchor = tableScrollAnchorRef.current;
+    if (!anchor) {
+      return;
+    }
+
+    const anchorTop = anchor.getBoundingClientRect().top;
+    const tableWrapper = anchor.querySelector<HTMLElement>(".ant-table-wrapper");
+    const tableBody = anchor.querySelector<HTMLElement>(".ant-table-body");
+
+    let nextHeight = Math.floor(window.innerHeight - anchorTop - TABLE_FALLBACK_RESERVE);
+    if (tableWrapper) {
+      const wrapperRect = tableWrapper.getBoundingClientRect();
+      const bodyHeight = tableBody?.getBoundingClientRect().height ?? TABLE_MIN_SCROLL_Y;
+      const nonBodyHeight = Math.max(0, wrapperRect.height - bodyHeight);
+      const topGap = Math.max(0, wrapperRect.top - anchorTop);
+      nextHeight = Math.floor(window.innerHeight - anchorTop - topGap - nonBodyHeight - TABLE_VIEWPORT_GAP);
+    }
+
+    const clampedHeight = Math.max(TABLE_MIN_SCROLL_Y, nextHeight);
+    setTableScrollY((previous) => (Math.abs(previous - clampedHeight) <= 1 ? previous : clampedHeight));
+  }, []);
+
+  useEffect(() => {
+    updateTableScrollY();
+  }, [items.length, listQuery.isFetching, error, success, updateTableScrollY]);
+
+  useEffect(() => {
+    if (typeof window === "undefined") {
+      return;
+    }
+
+    const onViewportChange = () => {
+      window.requestAnimationFrame(updateTableScrollY);
+    };
+
+    window.addEventListener("resize", onViewportChange);
+    return () => {
+      window.removeEventListener("resize", onViewportChange);
+    };
+  }, [updateTableScrollY]);
+
+  useEffect(() => {
+    if (typeof window === "undefined" || typeof ResizeObserver === "undefined") {
+      return;
+    }
+
+    const anchor = tableScrollAnchorRef.current;
+    if (!anchor) {
+      return;
+    }
+
+    const resizeObserver = new ResizeObserver(() => {
+      window.requestAnimationFrame(updateTableScrollY);
+    });
+    resizeObserver.observe(anchor);
+
+    return () => {
+      resizeObserver.disconnect();
+    };
+  }, [updateTableScrollY]);
+
+  if (initializing || listQuery.isLoading) {
+    return (
+      <div className="flex min-h-[240px] items-center justify-center">
+        <Spin tip="定时任务加载中..." />
+      </div>
+    );
+  }
+
+  if (!user) {
+    return (
+      <main className="mx-auto flex min-h-screen w-full max-w-4xl flex-col justify-center gap-4 px-6 py-20">
+        <p className="text-sm text-[var(--gray-11)]">请先登录后再访问定时任务管理页面。</p>
+        <Link
+          href="/"
+          className="inline-flex w-fit items-center justify-center rounded-md border border-[var(--gray-6)] bg-[var(--gray-a2)] px-4 py-2 text-sm font-medium text-[var(--gray-12)] transition hover:bg-[var(--gray-a3)]"
+        >
+          返回首页
+        </Link>
+      </main>
+    );
+  }
+
+  if (!canRead) {
+    return (
+      <main className="mx-auto flex min-h-screen w-full max-w-4xl flex-col justify-center gap-4 px-6 py-20">
+        <p className="text-sm text-[var(--gray-11)]">你没有访问该页面的权限（需要 `celery.read`）。</p>
+        <Link
+          href="/"
+          className="inline-flex w-fit items-center justify-center rounded-md border border-[var(--gray-6)] bg-[var(--gray-a2)] px-4 py-2 text-sm font-medium text-[var(--gray-12)] transition hover:bg-[var(--gray-a3)]"
+        >
+          返回首页
+        </Link>
+      </main>
+    );
+  }
+
+  return (
+    <div className="space-y-6">
+      <AntCard
+        title="定时任务管理"
+        extra={(
+          <Space>
+            {listQuery.isFetching && <Spin size="small" />}
+            {canManage ? (
+              <Button type="primary" onClick={startCreate}>
+                新建任务
+              </Button>
+            ) : null}
+          </Space>
+        )}
+      >
+        <Form layout="inline" style={{ rowGap: 12 }}>
+          <Form.Item label="关键词" className="min-w-[240px]">
+            <Input
+              value={keyword}
+              allowClear
+              onChange={(event) => setKeyword(event.currentTarget.value)}
+              placeholder="按任务键/名称/Cron 筛选"
+            />
+          </Form.Item>
+
+          <Form.Item label="状态" className="min-w-[180px]">
+            <Select<StatusFilter>
+              value={statusFilter}
+              options={[...STATUS_FILTER_OPTIONS]}
+              onChange={(value) => setStatusFilter(value)}
+            />
+          </Form.Item>
+
+          <Form.Item>
+            <Button
+              onClick={() => {
+                setKeyword("");
+                setStatusFilter("all");
+              }}
+            >
+              重置筛选
+            </Button>
+          </Form.Item>
+        </Form>
+
+        <div
+          ref={tableScrollAnchorRef}
+          className="admin-scheduled-tasks-table-anchor mt-4"
+          style={{ "--admin-scheduled-tasks-table-body-min-height": `${tableScrollY}px` } as CSSProperties}
+        >
+          <Table<ScheduledTaskSummary>
+            rowKey="id"
+            loading={listQuery.isFetching}
+            dataSource={items}
+            columns={columns}
+            scroll={{ x: 1380, y: tableScrollY }}
+            pagination={{
+              pageSize: 20,
+              showSizeChanger: true,
+              pageSizeOptions: [10, 20, 50, 100],
+              showTotal: (total) => `共 ${total} 条`,
+              style: { marginBottom: 0 },
+            }}
+            locale={{
+              emptyText: <Empty description="暂无定时任务。" image={Empty.PRESENTED_IMAGE_SIMPLE} />,
+            }}
+          />
+        </div>
+      </AntCard>
+
+      {canManage ? (
+        <Modal
+          title={editingId === null ? "新建定时任务" : "编辑定时任务"}
+          open={editorOpen}
+          onCancel={closeEditor}
+          width={760}
+          destroyOnClose
+          footer={(
+            <Space>
+              <Button type="primary" loading={saveMutation.isPending} onClick={() => saveMutation.mutate()}>
+                {saveMutation.isPending ? "提交中..." : editingId === null ? "创建" : "保存"}
+              </Button>
+              <Button onClick={resetForm}>重置</Button>
+            </Space>
+          )}
+        >
+          <Form<FormState> form={formApi} layout="vertical" initialValues={EMPTY_FORM}>
+            <div className="grid gap-4 md:grid-cols-2">
+              <Form.Item
+                name="task_key"
+                label="任务键"
+                rules={[{ required: true, message: "请输入任务键" }]}
+                extra="建议使用稳定英文键，如 syslog.cleanup.default。"
+              >
+                <Input disabled={editingId !== null} placeholder="syslog.cleanup.default" />
+              </Form.Item>
+              <Form.Item name="name" label="任务名称" rules={[{ required: true, message: "请输入任务名称" }]}>
+                <Input placeholder="系统日志定时清理" />
+              </Form.Item>
+            </div>
+
+            <div className="grid gap-4 md:grid-cols-2">
+              <Form.Item
+                name="cron_expression"
+                label="Cron 表达式"
+                rules={[{ required: true, message: "请输入 Cron 表达式" }]}
+                extra="格式：分钟 小时 日 月 周，例如 `0 3 * * *`。"
+              >
+                <Input placeholder="0 3 * * *" />
+              </Form.Item>
+              <Form.Item name="timezone" label="时区" rules={[{ required: true, message: "请选择时区" }]}>
+                <Select options={[...TIMEZONE_OPTIONS]} />
+              </Form.Item>
+            </div>
+
+            <div className="grid gap-4 md:grid-cols-2">
+              <Form.Item
+                name="retain_days"
+                label="日志保留天数"
+                rules={[{ required: true, message: "请输入日志保留天数" }]}
+              >
+                <InputNumber min={1} max={3650} className="w-full" />
+              </Form.Item>
+              <Form.Item name="enabled" label="启用状态" valuePropName="checked">
+                <Switch checkedChildren="启用" unCheckedChildren="停用" />
+              </Form.Item>
+            </div>
+
+            <Form.Item name="description" label="说明">
+              <Input.TextArea rows={4} placeholder="说明任务用途、影响范围和运行窗口。" />
+            </Form.Item>
+          </Form>
+        </Modal>
+      ) : null}
+    </div>
+  );
+}

web/src/app/globals.css

@@ -261,6 +261,10 @@ body {
   min-height: var(--admin-system-params-table-body-min-height, 180px);
 }
 
+.admin-scheduled-tasks-table-anchor .ant-table-body {
+  min-height: var(--admin-scheduled-tasks-table-body-min-height, 180px);
+}
+
 .admin-syslog-table-anchor .ant-table-body {
   min-height: var(--admin-syslog-table-body-min-height, 180px);
 }

web/src/types/auth.ts

@@ -103,6 +103,46 @@ export type SystemParamListResponse = {
   total: number;
 };
 
+export type ScheduledTaskStatus = "idle" | "queued" | "running" | "success" | "failed" | "disabled";
+export type ScheduledTaskType = "syslog_cleanup";
+
+export type ScheduledTaskSummary = {
+  id: number;
+  task_key: string;
+  name: string;
+  task_type: ScheduledTaskType;
+  description: string | null;
+  cron_expression: string;
+  timezone: string;
+  retain_days: number;
+  enabled: boolean;
+  status: ScheduledTaskStatus;
+  last_run_at: string | null;
+  next_run_at: string | null;
+  last_success_at: string | null;
+  last_error_at: string | null;
+  last_error_message: string | null;
+  last_result_json: Record<string, unknown>;
+  run_count: number;
+  create_user: string | null;
+  update_user: string | null;
+  create_date: string;
+  update_date: string;
+  creator: UserPublic | null;
+  updater: UserPublic | null;
+};
+
+export type ScheduledTaskListResponse = {
+  items: ScheduledTaskSummary[];
+  total: number;
+};
+
+export type ScheduledTaskRunResponse = {
+  success: boolean;
+  task: ScheduledTaskSummary;
+  celery_task_id: string | null;
+};
+
 export type FileStorageDriverType = "VFS" | "S3";
 
 export type FileStorageBackendSummary = {