kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(cron): 实现完整的定时任务调度模块

Browse Source 
新增了从调度引擎、任务注册、状态追踪到结果投递的全套cron任务系统，支持cron表达式、一次性任务和周期任务，包含失败告警、任务持久化和运行日志功能
This commit is contained in:
Kris 2026-05-21 10:24:54 +08:00
parent 4962d1f6f8
commit 0dcc370819
9 changed files with 2046 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
backend/package/yuxi/channel/cron/__init__.py Normal file
View File

@ -0,0 +1,47 @@
from yuxi.channel.cron.engine import CronEngine, get_cron_engine
from yuxi.channel.cron.handler_registry import (
list_registered,
register_handler,
resolve_handler,
)
from yuxi.channel.cron.normalize import normalize_job
from yuxi.channel.cron.phase_tracker import PhaseTracker, create_run_id
from yuxi.channel.cron.types import (
CronAgentExecutionPhase,
CronDeliveryReceipt,
CronDeliveryTarget,
CronFailureAlert,
CronJob,
CronJobState,
CronJobStatus,
CronPhaseRecord,
CronRunDiagnostic,
CronRunOutcome,
CronRunStatus,
DeliveryHandler,
ScheduleKind,
)
__all__ = [
"CronAgentExecutionPhase",
"CronDeliveryReceipt",
"CronDeliveryTarget",
"CronEngine",
"CronFailureAlert",
"CronJob",
"CronJobState",
"CronJobStatus",
"CronPhaseRecord",
"CronRunDiagnostic",
"CronRunOutcome",
"CronRunStatus",
"DeliveryHandler",
"PhaseTracker",
"ScheduleKind",
"create_run_id",
"get_cron_engine",
"list_registered",
"normalize_job",
"register_handler",
"resolve_handler",
]

122
backend/package/yuxi/channel/cron/alert.py Normal file
View File

@ -0,0 +1,122 @@
import asyncio
import logging
import time
from yuxi.channel.cron.types import CronFailureAlert, CronJob, DeliveryHandler
logger = logging.getLogger(__name__)
ALERT_COOLDOWN_MS = 600_000
RECOVERY_COOLDOWN_MS = 300_000
_GLOBAL_ALERT_MAX_PER_MINUTE = 10
_global_alert_timestamps: list[int] = []
def _global_rate_check(now_ms: int) -> bool:
cutoff = now_ms - 60_000
global _global_alert_timestamps
_global_alert_timestamps = [t for t in _global_alert_timestamps if t > cutoff]
if len(_global_alert_timestamps) >= _GLOBAL_ALERT_MAX_PER_MINUTE:
return False
_global_alert_timestamps.append(now_ms)
return True
def _format_cooldown_ms(ms: int) -> str:
if ms < 60_000:
return f"{ms // 1000}s"
if ms < 3_600_000:
return f"{ms // 60_000}min"
return f"{ms // 3_600_000}h"
def format_alert_message(job: CronJob, consecutive_errors: int, last_error: str | None) -> str:
job_label = job.name or job.id
return (
f"🚨 Cron 任务连续失败告警\n\n"
f"任务：{job_label}\n"
f"连续失败次数：{consecutive_errors}\n"
f"告警触发阈值：{job.failure_alert.threshold}\n"
f"最近错误：{last_error or '未知'}\n\n"
f"任务已自动标记为 FAILED，冷却期 {_format_cooldown_ms(job.failure_alert.cooldown_ms)} 内不再告警"
)
def format_recovery_message(job: CronJob) -> str:
job_label = job.name or job.id
return f"✅ Cron 任务已恢复正常\n\n任务：{job_label}\n任务状态已恢复为 ACTIVE"
async def send_alert(
handler: DeliveryHandler | None,
delivery_targets: list,
job: CronJob,
content: str,
) -> None:
if handler is None:
logger.warning("Cron alert for '%s' cannot be delivered: no delivery handler configured", job.id)
return
if not delivery_targets:
logger.warning("Cron alert for '%s': no delivery targets configured", job.id)
return
for target in delivery_targets:
if not target.enabled or not target.channel or not target.target_id:
continue
try:
await handler(target.channel, target.target_id, content)
except Exception:
logger.exception("Cron alert delivery to %s:%s failed", target.channel, target.target_id)
async def check_and_alert(
handler: DeliveryHandler | None,
job: CronJob,
) -> bool:
alert = job.failure_alert
now_ms = int(time.time() * 1000)
delivery_targets = job.delivery
if alert.should_alert(job.state.consecutive_errors, now_ms):
if not _global_rate_check(now_ms):
logger.warning("Cron alert for '%s' suppressed by global rate limit", job.id)
return False
content = format_alert_message(job, job.state.consecutive_errors, job.state.last_error)
await send_alert(handler, delivery_targets, job, content)
alert.mark_alerted(now_ms)
logger.warning(
"Cron failure alert sent for '%s' (consecutive errors: %d)",
job.id,
job.state.consecutive_errors,
)
return True
return False
async def check_and_recover(
handler: DeliveryHandler | None,
job: CronJob,
) -> bool:
alert = job.failure_alert
now_ms = int(time.time() * 1000)
delivery_targets = job.delivery
if alert.should_recover(job.state.consecutive_errors, now_ms):
content = format_recovery_message(job)
await send_alert(handler, delivery_targets, job, content)
alert.mark_recovered(now_ms)
logger.info("Cron recovery notification sent for '%s'", job.id)
return True
return False
async def evaluate_alerts(
handler: DeliveryHandler | None,
job: CronJob,
) -> dict:
alert_sent = await check_and_alert(handler, job)
recovery_sent = await check_and_recover(handler, job)
return {"alert_sent": alert_sent, "recovery_sent": recovery_sent}

203
backend/package/yuxi/channel/cron/delivery.py Normal file
View File

@ -0,0 +1,203 @@
import asyncio
import logging
import time
from dataclasses import dataclass, field
from yuxi.channel.cron.types import (
CronAgentExecutionPhase,
CronDeliveryReceipt,
CronDeliveryTarget,
CronJob,
CronRunDiagnostic,
DeliveryHandler,
)
logger = logging.getLogger(__name__)
DEFAULT_MAX_RETRIES = 3
DEFAULT_RETRY_BASE_MS = 500
DEFAULT_RETRY_MAX_MS = 10_000
DEFAULT_RETRY_BACKOFF = 2.0
@dataclass
class DeliveryConfig:
max_retries: int = DEFAULT_MAX_RETRIES
retry_base_ms: int = DEFAULT_RETRY_BASE_MS
retry_max_ms: int = DEFAULT_RETRY_MAX_MS
retry_backoff: float = DEFAULT_RETRY_BACKOFF
concurrency: int = 5
request_timeout_ms: int = 30_000
total_timeout_ms: int = 120_000
def resolve_delivery_targets(job: CronJob) -> list[CronDeliveryTarget]:
return [t for t in job.delivery if t.enabled and t.channel and t.target_id]
def format_delivery_content(
job: CronJob,
result: dict | None,
phase: CronAgentExecutionPhase | None,
error: str | None,
) -> str:
job_label = job.name or job.id
if error:
return f"⛔ Cron 任务执行失败\n\n任务：{job_label}\n阶段：{phase.value if phase else '未知'}\n错误：{error}"
if result is None:
return f"✅ Cron 任务执行完成\n\n任务：{job_label}\n阶段：{phase.value if phase else '未知'}\n结果：无返回值"
result_text = _format_result_value(result)
return f"✅ Cron 任务执行完成\n\n任务：{job_label}\n阶段：{phase.value if phase else '未知'}\n结果：\n{result_text}"
def _format_result_value(result: dict) -> str:
if not result:
return "(空)"
lines = []
for k, v in result.items():
if isinstance(v, (list, dict)):
v = str(v)[:500]
elif isinstance(v, str) and len(v) > 500:
v = v[:500] + "..."
lines.append(f" {k}: {v}")
return "\n".join(lines)
def _backoff_delay(attempt: int, config: DeliveryConfig) -> int:
base = config.retry_base_ms * (config.retry_backoff ** (attempt - 1))
return min(int(base), config.retry_max_ms)
async def deliver_to_target(
handler: DeliveryHandler,
target: CronDeliveryTarget,
content: str,
config: DeliveryConfig,
) -> CronDeliveryReceipt:
receipt = CronDeliveryReceipt(target=target)
started_mono = time.monotonic()
receipt.sent_at_ms = int(time.time() * 1000)
total_started_mono = time.monotonic()
for attempt in range(1, config.max_retries + 1):
try:
success = await asyncio.wait_for(
handler(target.channel, target.target_id, content),
timeout=config.request_timeout_ms / 1000.0,
)
if success:
receipt.success = True
receipt.retries = attempt - 1
receipt.sent_at_ms = int(time.time() * 1000)
receipt.duration_ms = int((time.monotonic() - started_mono) * 1000)
return receipt
receipt.error = "handler returned False"
except asyncio.TimeoutError:
receipt.error = f"delivery timeout ({config.request_timeout_ms}ms)"
logger.warning(
"Cron delivery attempt %d/%d to %s:%s timed out after %dms",
attempt,
config.max_retries,
target.channel,
target.target_id,
config.request_timeout_ms,
)
except Exception as e:
receipt.error = str(e)
logger.warning(
"Cron delivery attempt %d/%d to %s:%s failed: %s",
attempt,
config.max_retries,
target.channel,
target.target_id,
e,
)
total_elapsed_ms = int((time.monotonic() - total_started_mono) * 1000)
if total_elapsed_ms >= config.total_timeout_ms:
receipt.error = f"total delivery timeout ({config.total_timeout_ms}ms)"
logger.warning(
"Cron delivery to %s:%s exceeded total timeout %dms",
target.channel,
target.target_id,
config.total_timeout_ms,
)
break
if attempt < config.max_retries:
delay = _backoff_delay(attempt, config)
await asyncio.sleep(delay / 1000.0)
receipt.success = False
receipt.retries = config.max_retries
receipt.duration_ms = int((time.monotonic() - started_mono) * 1000)
return receipt
async def batch_deliver(
handler: DeliveryHandler,
job: CronJob,
result: dict | None,
error: str | None,
diagnostic: CronRunDiagnostic | None,
config: DeliveryConfig | None = None,
) -> list[CronDeliveryReceipt]:
targets = resolve_delivery_targets(job)
if not targets:
return []
config = config or DeliveryConfig()
content = format_delivery_content(job, result, CronAgentExecutionPhase.DELIVERING, error)
semaphore = asyncio.Semaphore(config.concurrency)
async def _deliver_one(target: CronDeliveryTarget) -> CronDeliveryReceipt:
async with semaphore:
return await deliver_to_target(handler, target, content, config)
receipts = await asyncio.gather(*[_deliver_one(t) for t in targets], return_exceptions=True)
results: list[CronDeliveryReceipt] = []
for i, r in enumerate(receipts):
if isinstance(r, CronDeliveryReceipt):
results.append(r)
elif isinstance(r, Exception):
logger.exception("Cron delivery task crashed for target [%d]", i)
target = targets[i] if i < len(targets) else CronDeliveryTarget(channel="unknown", target_id="unknown")
results.append(
CronDeliveryReceipt(
target=target,
success=False,
error=str(r),
)
)
if diagnostic is not None:
diagnostic.delivery_receipts = results
succeeded = sum(1 for r in results if r.success)
failed = len(results) - succeeded
if failed > 0:
logger.warning(
"Cron batch delivery for '%s': %d/%d succeeded, %d failed",
job.id,
succeeded,
len(results),
failed,
)
return results
async def deliver_result(
handler: DeliveryHandler,
job: CronJob,
result: dict | None,
error: str | None,
diagnostic: CronRunDiagnostic | None,
config: DeliveryConfig | None = None,
) -> list[CronDeliveryReceipt]:
return await batch_deliver(handler, job, result, error, diagnostic, config)

755
backend/package/yuxi/channel/cron/engine.py Normal file
View File

@ -0,0 +1,755 @@
import asyncio
import json
import logging
import os
import time
from pathlib import Path
from yuxi.channel.cron.alert import evaluate_alerts
from yuxi.channel.config.defaults import TIMEOUT
from yuxi.channel.cron.delivery import DeliveryConfig, deliver_result
from yuxi.channel.cron.handler_registry import resolve_handler
from yuxi.channel.cron.normalize import normalize_job
from yuxi.channel.cron.phase_tracker import PhaseTracker
from yuxi.channel.cron.schedule import (
compute_next_run_at_ms,
compute_next_with_error_tracking,
)
from yuxi.channel.cron.types import (
CronAgentExecutionPhase,
CronDeliveryReceipt,
CronDeliveryTarget,
CronFailureAlert,
CronJob,
CronJobState,
CronJobStatus,
CronPhaseRecord,
CronRunDiagnostic,
CronRunOutcome,
CronRunStatus,
DeliveryHandler,
ScheduleKind,
)
logger = logging.getLogger(__name__)
DEFAULT_MAX_CONCURRENT = 3
DEFAULT_TICK_INTERVAL = 1.0
DEFAULT_STORE_DIR = "cron"
DEFAULT_STORE_FILE = "jobs.json"
DEFAULT_STATE_FILE = "jobs-state.json"
DEFAULT_RUN_LOG_MAX_BYTES = 2_000_000
DEFAULT_RUN_LOG_KEEP_LINES = 2_000
DEFAULT_DIAGNOSTIC_KEEP = 200
DEFAULT_DIAGNOSTIC_LOG_MAX_BYTES = 5_000_000
DEFAULT_DIAGNOSTIC_LOG_KEEP_LINES = 5000
TOKEN_USAGE_KEYS = {"prompt_tokens", "completion_tokens", "total_tokens"}
TOKEN_USAGE_NESTED_KEYS = {"usage", "token_usage", "tokens"}
def _cron_dir() -> Path:
from yuxi.channel.config.discovery import resolve_config_dir
return Path(resolve_config_dir()) / DEFAULT_STORE_DIR
def _store_path() -> Path:
return _cron_dir() / DEFAULT_STORE_FILE
def _state_path() -> Path:
return _cron_dir() / DEFAULT_STATE_FILE
def _diagnostic_log_path() -> Path:
return _cron_dir() / "diagnostics.jsonl"
def _run_log_path(job_id: str) -> Path:
runs_dir = _cron_dir() / "runs"
safe_id = job_id.strip().replace("\\", "_").replace("/", "_")
return runs_dir / f"{safe_id}.jsonl"
class CronEngine:
def __init__(
self,
max_concurrent: int = DEFAULT_MAX_CONCURRENT,
tick_interval: float = DEFAULT_TICK_INTERVAL,
store_path: str | None = None,
state_path: str | None = None,
):
self._jobs: dict[str, CronJob] = {}
self._semaphore = asyncio.Semaphore(max_concurrent)
self._tick_interval = tick_interval
self._stop_event = asyncio.Event()
self._task: asyncio.Task | None = None
self._store_path = Path(store_path) if store_path else _store_path()
self._state_path = Path(state_path) if state_path else _state_path()
self._cron_dir = self._store_path.parent
self._loaded = False
self._delivery_handler: DeliveryHandler | None = None
self._delivery_config = DeliveryConfig()
self._diagnostics: list[CronRunDiagnostic] = []
self._running_tasks: set[asyncio.Task] = set()
def _cron_dir_path(self) -> Path:
return self._cron_dir
def _diagnostic_log_path(self) -> Path:
return self._cron_dir / "diagnostics.jsonl"
def _run_log_path(self, job_id: str) -> Path:
runs_dir = self._cron_dir / "runs"
safe_id = job_id.strip().replace("\\", "_").replace("/", "_")
return runs_dir / f"{safe_id}.jsonl"
@property
def is_running(self) -> bool:
return self._task is not None and not self._task.done()
def set_delivery_handler(self, handler: DeliveryHandler) -> None:
self._delivery_handler = handler
def set_delivery_config(self, config: DeliveryConfig) -> None:
self._delivery_config = config
def _ensure_loaded(self):
if not self._loaded:
self._load_store()
def _load_store(self):
self._loaded = True
try:
self._store_path.parent.mkdir(parents=True, exist_ok=True)
except OSError:
pass
self._load_diagnostics()
state_data: dict[str, dict] = {}
if self._state_path.exists():
try:
raw = self._state_path.read_text(encoding="utf-8")
state_raw = json.loads(raw)
if isinstance(state_raw, dict):
state_data = state_raw.get("jobs", {})
if isinstance(state_data, list):
state_data = {item.get("id", ""): item for item in state_data if isinstance(item, dict)}
except (json.JSONDecodeError, OSError):
logger.warning("CronEngine: failed to load state, starting fresh")
if not self._store_path.exists():
return
try:
raw = self._store_path.read_text(encoding="utf-8")
data = json.loads(raw)
jobs_data = data.get("jobs", []) if isinstance(data, dict) else []
except (json.JSONDecodeError, OSError):
logger.warning("CronEngine: failed to load store, starting fresh")
return
for jd in jobs_data:
if not isinstance(jd, dict):
continue
try:
job = self._deserialize_job(jd, state_data.get(jd.get("id", ""), {}))
normalize_job(job)
self._jobs[job.id] = job
except Exception:
logger.exception("CronEngine: failed to deserialize job '%s'", jd.get("id", "?"))
logger.info("CronEngine: loaded %d jobs from %s", len(self._jobs), self._store_path)
def _load_diagnostics(self):
diag_path = self._diagnostic_log_path()
if not diag_path.exists():
return
try:
self._diagnostics = []
with open(diag_path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
self._diagnostics.append(
CronRunDiagnostic(
run_id=data.get("run_id", ""),
job_id=data.get("job_id", ""),
outcome=CronRunOutcome(data.get("outcome", "success")),
phases=[
CronPhaseRecord(
phase=CronAgentExecutionPhase(p["phase"]),
started_at_ms=p["started_at_ms"],
finished_at_ms=p.get("finished_at_ms"),
error=p.get("error"),
)
for p in data.get("phases", [])
],
total_duration_ms=data.get("total_duration_ms", 0),
error=data.get("error"),
token_usage=data.get("token_usage", {}),
delivery_receipts=[
CronDeliveryReceipt(
target=CronDeliveryTarget.from_dict(r["target"]),
success=r.get("success", False),
message_id=r.get("message_id"),
error=r.get("error"),
retries=r.get("retries", 0),
sent_at_ms=r.get("sent_at_ms", 0),
duration_ms=r.get("duration_ms", 0),
)
for r in data.get("delivery_receipts", [])
],
started_at_ms=data.get("started_at_ms", 0),
finished_at_ms=data.get("finished_at_ms", 0),
)
)
except (json.JSONDecodeError, KeyError, TypeError):
logger.warning("CronEngine: failed to parse diagnostic line")
self._diagnostics = self._diagnostics[-DEFAULT_DIAGNOSTIC_KEEP:]
logger.info("CronEngine: loaded %d diagnostics from %s", len(self._diagnostics), diag_path)
except OSError:
logger.warning("CronEngine: failed to load diagnostics, starting fresh")
def _save_store(self):
try:
self._store_path.parent.mkdir(parents=True, exist_ok=True)
data = {
"version": 1,
"jobs": [self._serialize_job(j) for j in self._jobs.values()],
}
tmp_path = self._store_path.with_suffix(".tmp")
tmp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(tmp_path, self._store_path)
except OSError:
logger.exception("CronEngine: failed to save store")
def _save_state(self):
try:
self._state_path.parent.mkdir(parents=True, exist_ok=True)
state_data = {
"version": 1,
"jobs": {j.id: j.state.to_dict() for j in self._jobs.values()},
}
tmp_path = self._state_path.with_suffix(".tmp")
tmp_path.write_text(json.dumps(state_data, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(tmp_path, self._state_path)
except OSError:
logger.exception("CronEngine: failed to save state")
def _save_all(self):
self._save_store()
self._save_state()
def _serialize_job(self, job: CronJob) -> dict:
d = job.to_dict()
d.pop("handler", None)
d.pop("handler_args", None)
return d
def _deserialize_job(self, jd: dict, state_data: dict | None = None) -> CronJob:
state_dict = jd.get("state", {})
if state_data:
state_dict = {**state_dict, **state_data}
job = CronJob(
id=jd["id"],
name=jd.get("name", ""),
handler=resolve_handler(jd.get("handler_name", "")),
handler_name=jd.get("handler_name", ""),
handler_args={},
schedule_kind=ScheduleKind(jd.get("schedule_kind", "cron")),
schedule_value=jd.get("schedule_value", jd.get("cron_expr", "* * * * *")),
schedule_tz=jd.get("schedule_tz"),
stagger_ms=jd.get("stagger_ms", 0),
every_ms=jd.get("every_ms"),
anchor_ms=jd.get("anchor_ms"),
status=CronJobStatus(jd.get("status", "active")),
enabled=jd.get("enabled", True),
max_runs=jd.get("max_runs", 0),
delete_after_run=jd.get("delete_after_run", False),
error_threshold=jd.get("error_threshold", 5),
delivery=[CronDeliveryTarget.from_dict(t) for t in jd.get("delivery", []) if isinstance(t, dict)],
failure_alert=CronFailureAlert.from_dict(jd.get("failure_alert")),
state=CronJobState.from_dict(state_dict),
created_at=jd.get("created_at", time.time()),
updated_at=jd.get("updated_at", time.time()),
)
return job
def add_job(self, job: CronJob) -> None:
self._ensure_loaded()
if job.id in self._jobs:
raise ValueError(f"Job '{job.id}' already registered")
normalize_job(job)
job.state.next_run_at_ms = compute_next_run_at_ms(job)
self._jobs[job.id] = job
self._save_all()
def remove_job(self, job_id: str) -> None:
self._ensure_loaded()
self._jobs.pop(job_id, None)
self._save_all()
def get_job(self, job_id: str) -> CronJob | None:
self._ensure_loaded()
return self._jobs.get(job_id)
def list_jobs(self) -> list[CronJob]:
self._ensure_loaded()
return list(self._jobs.values())
def update_job(self, job_id: str, **kwargs) -> CronJob | None:
self._ensure_loaded()
job = self._jobs.get(job_id)
if job is None:
return None
allowed = frozenset({
"name", "enabled", "schedule_kind", "schedule_value", "schedule_tz",
"stagger_ms", "every_ms", "anchor_ms", "max_runs", "error_threshold",
"delivery", "failure_alert", "delete_after_run",
})
for k, v in kwargs.items():
if k in allowed and hasattr(job, k):
setattr(job, k, v)
job.updated_at = time.time()
normalize_job(job)
self._save_all()
return job
def pause_job(self, job_id: str) -> None:
self._ensure_loaded()
job = self._jobs.get(job_id)
if job:
job.status = CronJobStatus.PAUSED
job.updated_at = time.time()
self._save_all()
def resume_job(self, job_id: str) -> None:
self._ensure_loaded()
job = self._jobs.get(job_id)
if job:
job.status = CronJobStatus.ACTIVE
job.state.next_run_at_ms = compute_next_run_at_ms(job)
job.updated_at = time.time()
self._save_all()
async def start(self) -> None:
self._ensure_loaded()
if self.is_running:
return
self._stop_event.clear()
self._task = asyncio.create_task(self._loop(), name="cron-engine")
logger.info("CronEngine started with %d jobs", len(self._jobs))
async def stop(self) -> None:
if not self.is_running:
return
self._stop_event.set()
if self._task and not self._task.done():
self._task.cancel()
running = [t for t in self._running_tasks if not t.done()]
if running:
for task in running:
task.cancel()
try:
await asyncio.wait_for(asyncio.gather(*running, return_exceptions=True), timeout=TIMEOUT.cron.gather)
except TimeoutError:
logger.warning("CronEngine: some running jobs did not cancel within 15s")
logger.info("CronEngine cancelled %d running jobs", len(running))
if self._task:
try:
await asyncio.wait_for(self._task, timeout=TIMEOUT.cron.task)
except (asyncio.CancelledError, TimeoutError):
pass
finally:
if not self._task.done():
logger.warning("CronEngine: main task did not terminate gracefully")
self._task = None
self._running_tasks.clear()
self._save_all()
logger.info("CronEngine stopped")
async def force_run(self, job_id: str) -> dict:
self._ensure_loaded()
job = self._jobs.get(job_id)
if job is None:
return {"success": False, "error": f"Job '{job_id}' not found"}
if job.state.running_at_ms is not None:
return {"success": False, "error": f"Job '{job_id}' is already running"}
return await self._run_job(job)
async def _loop(self) -> None:
while not self._stop_event.is_set():
try:
await self._tick()
except Exception:
logger.exception("CronEngine tick error")
try:
await asyncio.wait_for(
self._stop_event.wait(),
timeout=self._tick_interval,
)
break
except TimeoutError:
pass
self._save_all()
async def _tick(self) -> None:
now_ms = int(time.time() * 1000)
due_jobs = [
j
for j in self._jobs.values()
if j.status == CronJobStatus.ACTIVE
and j.enabled
and j.state.next_run_at_ms is not None
and now_ms >= j.state.next_run_at_ms
and j.state.running_at_ms is None
]
if not due_jobs:
return
tasks = [asyncio.create_task(self._run_job(j), name=f"cron-job-{j.id}") for j in due_jobs]
self._running_tasks.update(tasks)
try:
await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=TIMEOUT.cron.delivery,
)
except TimeoutError:
logger.warning("CronEngine tick timeout: %d jobs did not complete in 300s", len(due_jobs))
for task in tasks:
if not task.done():
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
finally:
for task in tasks:
if not task.done():
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
self._running_tasks.difference_update(tasks)
async def _run_job(self, job: CronJob) -> dict:
async with self._semaphore:
if job.state.running_at_ms is not None:
return {"success": False, "error": f"Job '{job.id}' is already running"}
if job.max_runs > 0 and job.state.run_count >= job.max_runs:
job.status = CronJobStatus.EXPIRED
self._save_all()
return {"success": False, "error": "max_runs reached"}
tracker = PhaseTracker(job)
diagnostic = tracker.diagnostic
exec_phase = tracker.start_executing()
run_start_mono = time.monotonic()
job.state.running_at_ms = diagnostic.started_at_ms
try:
result = await job.handler(**(job.handler_args or {}))
tracker.finish_executing(exec_phase)
error = None
outcome = CronRunOutcome.SUCCESS
job.state.last_run_status = CronRunStatus.OK
job.state.last_error = None
job.state.run_count += 1
job.state.consecutive_errors = 0
job.state.consecutive_skipped = 0
job.state.next_run_at_ms = compute_next_with_error_tracking(job)
job.status = CronJobStatus.ACTIVE
summ_phase = tracker.start_summarizing()
tracker.finish_summarizing(summ_phase)
deliver_phase = tracker.start_delivering()
delivery_receipts: list = []
if job.delivery and self._delivery_handler:
delivery_receipts = await deliver_result(
self._delivery_handler,
job,
result if isinstance(result, dict) else {"result": str(result)},
None,
diagnostic,
self._delivery_config,
)
tracker.finish_delivering(deliver_phase)
receipt_dicts = [r.to_dict() for r in delivery_receipts]
except Exception as e:
try:
tracker.finish_executing(exec_phase, str(e))
except Exception:
logger.exception("CronEngine: failed to finish executing tracker for job '%s'", job.id)
error = str(e)
outcome = CronRunOutcome.FAILED
job.state.error_count += 1
job.state.consecutive_errors += 1
job.state.last_run_status = CronRunStatus.ERROR
job.state.last_error = error
job.state.next_run_at_ms = compute_next_with_error_tracking(job)
if job.state.consecutive_errors >= job.error_threshold:
job.status = CronJobStatus.FAILED
deliver_phase = tracker.start_delivering()
delivery_receipts: list = []
if job.delivery and self._delivery_handler:
delivery_receipts = await deliver_result(
self._delivery_handler,
job,
None,
error,
diagnostic,
self._delivery_config,
)
tracker.finish_delivering(deliver_phase)
receipt_dicts = [r.to_dict() for r in delivery_receipts]
result = None
logger.exception(
"CronEngine job '%s' failed (consecutive: %d)",
job.id,
job.state.consecutive_errors,
)
duration_ms = int((time.monotonic() - run_start_mono) * 1000)
token_usage = _extract_token_usage(result) if not error else {}
diagnostic = tracker.close(outcome, error, token_usage)
self._append_diagnostic(diagnostic)
job.state.last_run_at_ms = diagnostic.started_at_ms
job.state.last_duration_ms = duration_ms
job.state.running_at_ms = None
job.updated_at = time.time()
self._append_run_log(
job,
{
"ts": diagnostic.started_at_ms,
"job_id": job.id,
"action": "finished",
"status": "ok" if outcome == CronRunOutcome.SUCCESS else "error",
"run_id": diagnostic.run_id,
"run_at_ms": diagnostic.started_at_ms,
"duration_ms": duration_ms,
"next_run_at_ms": job.state.next_run_at_ms,
"consecutive_errors": job.state.consecutive_errors,
**({"error": error} if error else {}),
**(
{"result": result if isinstance(result, dict) else {"result": str(result)}}
if result is not None
else {}
),
"phases": [p.to_dict() for p in diagnostic.phases],
"delivery": receipt_dicts,
},
)
if job.delete_after_run and job.schedule_kind == ScheduleKind.AT:
self._jobs.pop(job.id, None)
logger.info("CronEngine: deleted one-shot job '%s'", job.id)
await evaluate_alerts(self._delivery_handler, job)
self._save_all()
if outcome == CronRunOutcome.SUCCESS:
return {
"success": True,
"job_id": job.id,
"run_id": diagnostic.run_id,
"diagnostic": diagnostic.to_dict(),
}
return {
"success": False,
"error": error,
"run_id": diagnostic.run_id,
"diagnostic": diagnostic.to_dict(),
}
def _append_run_log(self, job: CronJob, entry: dict):
try:
log_path = self._run_log_path(job.id)
log_path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(entry, ensure_ascii=False) + "\n"
with open(log_path, "a", encoding="utf-8") as f:
f.write(line)
self._prune_run_log(log_path)
except OSError:
logger.exception("CronEngine: failed to write run log for '%s'", job.id)
def _prune_run_log(self, log_path: Path):
stat = log_path.stat()
if stat.st_size <= DEFAULT_RUN_LOG_MAX_BYTES:
return
try:
lines = log_path.read_text(encoding="utf-8").strip().split("\n")
kept = lines[-DEFAULT_RUN_LOG_KEEP_LINES:]
log_path.write_text("\n".join(kept) + "\n", encoding="utf-8")
except OSError:
pass
def read_run_log(self, job_id: str, limit: int = 50, offset: int = 0) -> dict:
log_path = self._run_log_path(job_id)
if not log_path.exists():
return {"entries": [], "total": 0, "offset": offset, "limit": limit, "has_more": False}
try:
raw = log_path.read_text(encoding="utf-8")
lines = [json.loads(line) for line in raw.strip().split("\n") if line.strip()]
except (OSError, json.JSONDecodeError):
return {"entries": [], "total": 0, "offset": offset, "limit": limit, "has_more": False}
total = len(lines)
start = max(0, min(total, offset))
end = min(total, start + limit)
entries = lines[start:end]
return {
"entries": entries,
"total": total,
"offset": offset,
"limit": limit,
"has_more": end < total,
}
def _append_diagnostic(self, diagnostic: CronRunDiagnostic) -> None:
self._diagnostics.append(diagnostic)
if len(self._diagnostics) > DEFAULT_DIAGNOSTIC_KEEP:
self._diagnostics = self._diagnostics[-DEFAULT_DIAGNOSTIC_KEEP:]
self._write_diagnostic_log(diagnostic)
def _write_diagnostic_log(self, diagnostic: CronRunDiagnostic) -> None:
try:
diag_path = self._diagnostic_log_path()
diag_path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(diagnostic.to_dict(), ensure_ascii=False) + "\n"
with open(diag_path, "a", encoding="utf-8") as f:
f.write(line)
self._prune_diagnostic_log(diag_path)
except OSError:
logger.exception("CronEngine: failed to write diagnostic log")
def _prune_diagnostic_log(self, diag_path: Path) -> None:
try:
stat = diag_path.stat()
except OSError:
return
if stat.st_size <= DEFAULT_DIAGNOSTIC_LOG_MAX_BYTES:
return
try:
lines = diag_path.read_text(encoding="utf-8").strip().split("\n")
kept = lines[-DEFAULT_DIAGNOSTIC_LOG_KEEP_LINES:]
diag_path.write_text("\n".join(kept) + "\n", encoding="utf-8")
except OSError:
pass
def read_diagnostics(self, job_id: str | None = None, limit: int = 50, offset: int = 0) -> dict:
all_diags = self._read_all_diagnostics()
filtered = all_diags
if job_id:
filtered = [d for d in filtered if d.job_id == job_id]
total = len(filtered)
start = max(0, min(total, offset))
end = min(total, start + limit)
entries = [d.to_dict() for d in filtered[start:end]]
return {
"entries": entries,
"total": total,
"offset": offset,
"limit": limit,
"has_more": end < total,
}
def _read_all_diagnostics(self) -> list[CronRunDiagnostic]:
diag_path = self._diagnostic_log_path()
if not diag_path.exists():
return list(self._diagnostics)
memory_ids = {d.run_id for d in self._diagnostics if d.run_id}
result = list(self._diagnostics)
try:
raw = diag_path.read_text(encoding="utf-8")
for line in raw.strip().split("\n"):
if not line.strip():
continue
try:
data = json.loads(line)
if data.get("run_id") in memory_ids:
continue
result.append(
CronRunDiagnostic(
run_id=data.get("run_id", ""),
job_id=data.get("job_id", ""),
outcome=CronRunOutcome(data.get("outcome", "success")),
phases=[
CronPhaseRecord(
phase=CronAgentExecutionPhase(p["phase"]),
started_at_ms=p["started_at_ms"],
finished_at_ms=p.get("finished_at_ms"),
error=p.get("error"),
)
for p in data.get("phases", [])
],
total_duration_ms=data.get("total_duration_ms", 0),
error=data.get("error"),
token_usage=data.get("token_usage", {}),
delivery_receipts=[
CronDeliveryReceipt(
target=CronDeliveryTarget.from_dict(r["target"]),
success=r.get("success", False),
message_id=r.get("message_id"),
error=r.get("error"),
retries=r.get("retries", 0),
sent_at_ms=r.get("sent_at_ms", 0),
duration_ms=r.get("duration_ms", 0),
)
for r in data.get("delivery_receipts", [])
],
started_at_ms=data.get("started_at_ms", 0),
finished_at_ms=data.get("finished_at_ms", 0),
)
)
except (json.JSONDecodeError, KeyError, TypeError):
pass
except OSError:
pass
return sorted(result, key=lambda d: d.started_at_ms)
def list_diagnostics_for_job(self, job_id: str, limit: int = 10) -> list[dict]:
all_diags = self._read_all_diagnostics()
matching = [d for d in all_diags if d.job_id == job_id]
return [d.to_dict() for d in matching[-limit:]]
def _extract_token_usage(result: dict | None) -> dict:
if not result or not isinstance(result, dict):
return {}
usage = {}
for key in TOKEN_USAGE_KEYS:
if key in result and isinstance(result[key], (int, float)):
usage[key] = result[key]
for nested_key in TOKEN_USAGE_NESTED_KEYS:
if nested_key in result and isinstance(result[nested_key], dict):
for k, v in result[nested_key].items():
if isinstance(v, (int, float)):
usage[k] = v
return usage
_engine: CronEngine | None = None
def get_cron_engine() -> CronEngine:
global _engine
if _engine is None:
_engine = CronEngine()
return _engine

26
backend/package/yuxi/channel/cron/handler_registry.py Normal file
View File

@ -0,0 +1,26 @@
import logging
from collections.abc import Callable, Awaitable
logger = logging.getLogger(__name__)
type CronHandler = Callable[..., Awaitable[dict]]
_handler_registry: dict[str, CronHandler] = {}
def register_handler(name: str, handler: CronHandler) -> None:
if name in _handler_registry:
logger.warning("CronHandler '%s' already registered, overwriting", name)
_handler_registry[name] = handler
def resolve_handler(name: str) -> CronHandler:
return _handler_registry.get(name, _noop_handler)
def list_registered() -> list[str]:
return list(_handler_registry.keys())
async def _noop_handler(**kwargs) -> dict:
return {"message": "noop handler, job needs reassignment after reload"}

169
backend/package/yuxi/channel/cron/normalize.py Normal file
View File

@ -0,0 +1,169 @@
import logging
import time
from datetime import datetime
from yuxi.channel.cron.schedule import (
compute_next_run_at_ms,
is_recurring_top_of_hour_cron_expr,
resolve_default_cron_stagger_ms,
)
from yuxi.channel.cron.types import (
CronDeliveryTarget,
CronFailureAlert,
CronJob,
ScheduleKind,
)
logger = logging.getLogger(__name__)
CRON_FIELD_COUNT = 5
CRON_FIELD_RANGES = [
(0, 59),
(0, 23),
(1, 31),
(1, 12),
(0, 7),
]
def _validate_cron_expr_format(expr: str) -> str | None:
parts = expr.strip().split()
if len(parts) != CRON_FIELD_COUNT:
return f"cron 表达式应包含 {CRON_FIELD_COUNT} 个字段，实际 {len(parts)} 个: {expr}"
for i, (field, (lo, hi)) in enumerate(zip(parts, CRON_FIELD_RANGES)):
for segment in field.split(","):
step = 1
if "/" in segment:
segment, step_str = segment.split("/", 1)
try:
step = int(step_str)
except ValueError:
return f"cron 表达式字段 {i} 步长无效: {step_str}"
if step < 1:
return f"cron 表达式字段 {i} 步长必须 >= 1: {step}"
if segment == "*":
continue
if "-" in segment:
try:
low_str, high_str = segment.split("-", 1)
low = int(low_str)
high = int(high_str)
except ValueError:
return f"cron 表达式字段 {i} 范围格式无效: {segment}"
if low < lo or high > hi or low > high:
return f"cron 表达式字段 {i} 范围 {low}-{high} 超出 [{lo}, {hi}]"
else:
try:
val = int(segment)
except ValueError:
return f"cron 表达式字段 {i} 值无效: {segment}"
if val < lo or val > hi:
return f"cron 表达式字段 {i} 值 {val} 超出 [{lo}, {hi}]"
return None
def _validate_at_value(value: str) -> str | None:
try:
at_ms = int(value)
if at_ms <= 0:
return f"AT 调度值必须为正整数 (毫秒时间戳): {value}"
return None
except (ValueError, TypeError):
pass
try:
datetime.fromisoformat(value)
return None
except (ValueError, TypeError):
return f"AT 调度值格式无效 (需为毫秒时间戳或 ISO 8601): {value}"
def _validate_every_value(every_ms: int | None, anchor_ms: int | None) -> str | None:
if every_ms is None:
return "EVERY 调度必须设置 every_ms"
if every_ms <= 0:
return f"EVERY 调度的 every_ms 必须 > 0: {every_ms}"
if anchor_ms is not None and anchor_ms < 0:
return f"EVERY 调度的 anchor_ms 不能为负数: {anchor_ms}"
return None
def _validate_delivery_target(target: CronDeliveryTarget) -> str | None:
if not target.channel or not target.channel.strip():
return f"投递目标缺少 channel: {target.to_dict()}"
if not target.target_id or not target.target_id.strip():
return f"投递目标缺少 target_id: {target.to_dict()}"
return None
def normalize_job(job: CronJob) -> list[str]:
warnings: list[str] = []
if not job.id or not job.id.strip():
warnings.append("CronJob.id 不能为空")
job.id = job.id or f"unnamed_{int(time.time() * 1000)}"
if not job.name:
job.name = job.id
warnings.append(f"未设置 name，已使用 id 作为名称: {job.name}")
if job.error_threshold < 1:
job.error_threshold = 5
warnings.append(f"error_threshold 必须 >= 1，已重置为 {job.error_threshold}")
if job.max_runs < 0:
job.max_runs = 0
warnings.append(f"max_runs 不能为负数，已重置为 0 (无限制)")
if job.stagger_ms < 0:
job.stagger_ms = 0
warnings.append(f"stagger_ms 不能为负数，已重置为 0")
if job.schedule_kind == ScheduleKind.CRON:
err = _validate_cron_expr_format(job.schedule_value)
if err:
warnings.append(err)
if job.stagger_ms == 0 and job.schedule_value:
default_stagger = resolve_default_cron_stagger_ms(job.schedule_value)
if default_stagger > 0:
job.stagger_ms = default_stagger
warnings.append(f"整点 CRON 表达式未设置 stagger，已自动设置 {default_stagger}ms 错峰")
elif job.schedule_kind == ScheduleKind.AT:
err = _validate_at_value(job.schedule_value)
if err:
warnings.append(err)
if job.stagger_ms > 0:
warnings.append("AT 调度不支持 stagger，已忽略")
job.stagger_ms = 0
elif job.schedule_kind == ScheduleKind.EVERY:
err = _validate_every_value(job.every_ms, job.anchor_ms)
if err:
warnings.append(err)
for i, target in enumerate(job.delivery):
err = _validate_delivery_target(target)
if err:
warnings.append(f"投递目标 [{i}]: {err}")
if job.failure_alert.threshold < 1:
job.failure_alert.threshold = 5
warnings.append(f"告警阈值必须 >= 1，已重置为 {job.failure_alert.threshold}")
if job.failure_alert.cooldown_ms < 0:
job.failure_alert.cooldown_ms = 600_000
warnings.append(f"告警冷却时间不能为负数，已重置为 {job.failure_alert.cooldown_ms}ms")
try:
next_at = compute_next_run_at_ms(job)
if next_at is not None:
job.state.next_run_at_ms = next_at
except Exception as e:
warnings.append(f"无法计算下次执行时间: {e}")
if warnings:
logger.warning("CronJob '%s' 规范化产生 %d 条警告: %s", job.id, len(warnings), warnings)
return warnings

104
backend/package/yuxi/channel/cron/phase_tracker.py Normal file
View File

@ -0,0 +1,104 @@
import logging
import time
import uuid
from yuxi.channel.cron.types import (
CronAgentExecutionPhase,
CronJob,
CronPhaseRecord,
CronRunDiagnostic,
CronRunOutcome,
)
logger = logging.getLogger(__name__)
def create_run_id(job_id: str, started_at_ms: int) -> str:
return f"cron:{job_id}:{started_at_ms}:{uuid.uuid4().hex}"
def create_diagnostic(job: CronJob, started_at_ms: int) -> CronRunDiagnostic:
return CronRunDiagnostic(
run_id=create_run_id(job.id, started_at_ms),
job_id=job.id,
started_at_ms=started_at_ms,
)
def record_phase(
diagnostic: CronRunDiagnostic,
phase: CronAgentExecutionPhase,
) -> CronPhaseRecord:
now_ms = int(time.time() * 1000)
record = CronPhaseRecord(
phase=phase,
started_at_ms=now_ms,
)
diagnostic.phases.append(record)
return record
def finish_phase(record: CronPhaseRecord, error: str | None = None) -> None:
record.finished_at_ms = int(time.time() * 1000)
if error:
record.error = error
def finalize_diagnostic(
diagnostic: CronRunDiagnostic,
outcome: CronRunOutcome,
error: str | None = None,
token_usage: dict | None = None,
started_mono: float = 0.0,
) -> None:
diagnostic.finished_at_ms = int(time.time() * 1000)
diagnostic.total_duration_ms = _mono_duration_ms(started_mono) if started_mono > 0 else diagnostic.finished_at_ms - diagnostic.started_at_ms
diagnostic.outcome = outcome
if error:
diagnostic.error = error
if token_usage:
diagnostic.token_usage = token_usage
def _mono_duration_ms(started_mono: float) -> int:
return int((time.monotonic() - started_mono) * 1000)
class PhaseTracker:
def __init__(self, job: CronJob):
self.job = job
started_at = int(time.time() * 1000)
self._started_mono = time.monotonic()
self.diagnostic = create_diagnostic(job, started_at)
self._init_phase = record_phase(self.diagnostic, CronAgentExecutionPhase.INITIALIZING)
finish_phase(self._init_phase)
def start_planning(self) -> CronPhaseRecord:
return record_phase(self.diagnostic, CronAgentExecutionPhase.PLANNING)
def finish_planning(self, record: CronPhaseRecord, error: str | None = None) -> None:
finish_phase(record, error)
def start_executing(self) -> CronPhaseRecord:
return record_phase(self.diagnostic, CronAgentExecutionPhase.EXECUTING)
def finish_executing(self, record: CronPhaseRecord, error: str | None = None) -> None:
finish_phase(record, error)
def start_summarizing(self) -> CronPhaseRecord:
return record_phase(self.diagnostic, CronAgentExecutionPhase.SUMMARIZING)
def finish_summarizing(self, record: CronPhaseRecord, error: str | None = None) -> None:
finish_phase(record, error)
def start_delivering(self) -> CronPhaseRecord:
return record_phase(self.diagnostic, CronAgentExecutionPhase.DELIVERING)
def finish_delivering(self, record: CronPhaseRecord, error: str | None = None) -> None:
finish_phase(record, error)
def close(
self, outcome: CronRunOutcome, error: str | None = None, token_usage: dict | None = None
) -> CronRunDiagnostic:
finalize_diagnostic(self.diagnostic, outcome, error, token_usage, self._started_mono)
return self.diagnostic

318
backend/package/yuxi/channel/cron/schedule.py Normal file
View File

@ -0,0 +1,318 @@
import calendar
import logging
import random
import time
from collections import OrderedDict
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from yuxi.channel.cron.types import CronJob, CronJobStatus, ScheduleKind
logger = logging.getLogger(__name__)
CRON_EVAL_CACHE_MAX = 512
_cron_eval_cache: OrderedDict[str, tuple[set[int] | None, ...]] = OrderedDict()
DEFAULT_TOP_OF_HOUR_STAGGER_MS = 5 * 60 * 1000
def _clear_cron_cache_for_test():
_cron_eval_cache.clear()
def _get_cron_cache_size_for_test():
return len(_cron_eval_cache)
def _resolve_field(field: str, lo: int, hi: int) -> set[int] | None:
if field == "*":
return None
result: set[int] = set()
for part in field.split(","):
step = 1
if "/" in part:
part, step_str = part.split("/", 1)
step = int(step_str)
low = lo
high = hi
if "-" in part:
low_str, high_str = part.split("-", 1)
low = int(low_str)
high = int(high_str)
elif part != "*":
low = high = int(part)
for v in range(low, high + 1, step):
if lo <= v <= hi:
result.add(v)
return result or None
def _parse_cron_expr(cron_expr: str) -> tuple[set[int] | None, ...]:
cached = _cron_eval_cache.get(cron_expr)
if cached is not None:
_cron_eval_cache.move_to_end(cron_expr)
return cached
parts = cron_expr.strip().split()
if len(parts) != 5:
raise ValueError(f"Invalid cron expression: {cron_expr}")
parsed = (
_resolve_field(parts[0], 0, 59),
_resolve_field(parts[1], 0, 23),
_resolve_field(parts[2], 1, 31),
_resolve_field(parts[3], 1, 12),
_resolve_field(parts[4], 0, 7),
)
if len(_cron_eval_cache) >= CRON_EVAL_CACHE_MAX:
_cron_eval_cache.popitem(last=False)
_cron_eval_cache[cron_expr] = parsed
return parsed
def _resolve_tz(tz_name: str | None) -> ZoneInfo:
if tz_name:
try:
return ZoneInfo(tz_name)
except Exception:
logger.warning("Invalid timezone '%s', using system local time", tz_name)
local_tz = datetime.now().astimezone().tzinfo
if local_tz is not None:
return local_tz
return ZoneInfo("UTC")
def _compute_cron_next(expr: str, tz_name: str | None, now_ms: int) -> int | None:
parsed = _parse_cron_expr(expr)
tz = _resolve_tz(tz_name)
dt = datetime.fromtimestamp(now_ms / 1000.0, tz=tz)
current_min = dt.minute
current_hour = dt.hour
current_day = dt.day
current_month = dt.month
current_year = dt.year
current_wday = (dt.weekday() + 1) % 7
minutes_set, hours_set, days_set, months_set, weekdays_set = parsed
max_iterations = 525600
for _ in range(max_iterations):
if months_set is not None and current_month not in months_set:
current_month += 1
if current_month > 12:
current_month = 1
current_year += 1
current_day = 1
current_hour = 0
current_min = -1
continue
month_days = calendar.monthrange(current_year, current_month)[1]
if days_set is not None and current_day not in days_set:
current_day += 1
if current_day > month_days:
current_day = 1
current_month += 1
if current_month > 12:
current_month = 1
current_year += 1
current_hour = 0
current_min = -1
continue
if hours_set is not None and current_hour not in hours_set:
current_hour += 1
if current_hour >= 24:
current_hour = 0
current_day += 1
current_wday = (current_wday + 1) % 7
if current_day > month_days:
current_day = 1
current_month += 1
if current_month > 12:
current_month = 1
current_year += 1
current_min = -1
continue
current_min += 1
if current_min >= 60:
current_min = 0
current_hour += 1
if current_hour >= 24:
current_hour = 0
current_day += 1
current_wday = (current_wday + 1) % 7
month_days = calendar.monthrange(current_year, current_month)[1]
if current_day > month_days:
current_day = 1
current_month += 1
if current_month > 12:
current_month = 1
current_year += 1
if (
minutes_set is not None
and current_min not in minutes_set
or hours_set is not None
and current_hour not in hours_set
or days_set is not None
and current_day not in days_set
or months_set is not None
and current_month not in months_set
or weekdays_set is not None
and current_wday not in weekdays_set
):
continue
try:
candidate = datetime(
current_year,
current_month,
current_day,
current_hour,
current_min,
0,
tzinfo=tz,
)
candidate_ms = int(candidate.timestamp() * 1000)
except (OverflowError, ValueError):
continue
if candidate_ms > now_ms:
return candidate_ms
logger.warning("Cron next-run computation exceeded %d iterations for expr='%s'", max_iterations, expr)
return now_ms + 60000
def _compute_cron_previous(expr: str, tz_name: str | None, now_ms: int) -> int | None:
parsed = _parse_cron_expr(expr)
tz = _resolve_tz(tz_name)
minutes_set, hours_set, days_set, months_set, weekdays_set = parsed
search_dt = datetime.fromtimestamp(now_ms / 1000.0, tz=tz)
max_iterations = 525600
for _ in range(max_iterations):
search_dt -= timedelta(minutes=1)
sm = search_dt.minute
sh = search_dt.hour
sd = search_dt.day
sM = search_dt.month
sw = (search_dt.weekday() + 1) % 7
if (
minutes_set is not None
and sm not in minutes_set
or hours_set is not None
and sh not in hours_set
or days_set is not None
and sd not in days_set
or months_set is not None
and sM not in months_set
or weekdays_set is not None
and sw not in weekdays_set
):
continue
candidate_ms = int(search_dt.timestamp() * 1000)
if candidate_ms < now_ms:
return candidate_ms
logger.warning("Cron previous-run computation exceeded %d iterations for expr='%s'", max_iterations, expr)
return None
def _compute_every_next(every_ms: int, anchor_ms: int, now_ms: int) -> int:
every = max(1, every_ms)
anchor = max(0, anchor_ms)
if now_ms < anchor:
return anchor
elapsed = now_ms - anchor
steps = max(1, (elapsed + every - 1) // every)
return anchor + steps * every
def is_recurring_top_of_hour_cron_expr(expr: str) -> bool:
fields = expr.strip().split()
if len(fields) == 5:
minute_field, hour_field = fields[0], fields[1]
return minute_field == "0" and "*" in hour_field
return False
def resolve_default_cron_stagger_ms(expr: str) -> int:
return DEFAULT_TOP_OF_HOUR_STAGGER_MS if is_recurring_top_of_hour_cron_expr(expr) else 0
def resolve_cron_stagger_ms(job: CronJob) -> int:
if job.stagger_ms > 0:
return job.stagger_ms
if job.schedule_kind == ScheduleKind.CRON:
return resolve_default_cron_stagger_ms(job.schedule_value)
return 0
def compute_next_run_at_ms(job: CronJob) -> int | None:
now_ms = int(time.time() * 1000)
if job.schedule_kind == ScheduleKind.AT:
try:
at_ms = int(job.schedule_value)
except (ValueError, TypeError):
try:
parsed = datetime.fromisoformat(job.schedule_value)
at_ms = int(parsed.timestamp() * 1000)
except (ValueError, TypeError):
return None
return at_ms if at_ms > now_ms else None
if job.schedule_kind == ScheduleKind.EVERY:
if job.every_ms is None or job.every_ms <= 0:
return None
raw = _compute_every_next(job.every_ms, job.anchor_ms or now_ms, now_ms)
stagger = job.stagger_ms
if stagger > 0:
raw += random.randint(0, stagger)
return raw
if job.schedule_kind == ScheduleKind.CRON:
raw = _compute_cron_next(job.schedule_value, job.schedule_tz, now_ms)
if raw is None:
return None
stagger = resolve_cron_stagger_ms(job)
if stagger > 0:
raw += random.randint(0, stagger)
return raw
return None
def compute_previous_run_at_ms(job: CronJob) -> int | None:
now_ms = int(time.time() * 1000)
if job.schedule_kind == ScheduleKind.CRON:
return _compute_cron_previous(job.schedule_value, job.schedule_tz, now_ms)
return None
def compute_next_with_error_tracking(job: CronJob) -> int | None:
try:
return compute_next_run_at_ms(job)
except Exception:
job.state.schedule_error_count += 1
logger.exception("CronEngine schedule computation error for '%s'", job.id)
if job.state.schedule_error_count >= 10:
job.status = CronJobStatus.FAILED
logger.error(
"CronEngine disabling job '%s' after %d schedule errors",
job.id,
job.state.schedule_error_count,
)
return None

302
backend/package/yuxi/channel/cron/types.py Normal file
View File

@ -0,0 +1,302 @@
import time
import uuid
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from enum import StrEnum
type CronHandler = Callable[..., Awaitable[dict]]
type DeliveryHandler = Callable[[str, str, str], Awaitable[bool]]
class ScheduleKind(StrEnum):
AT = "at"
EVERY = "every"
CRON = "cron"
class CronJobStatus(StrEnum):
ACTIVE = "active"
PAUSED = "paused"
EXPIRED = "expired"
FAILED = "failed"
class CronRunStatus(StrEnum):
OK = "ok"
ERROR = "error"
SKIPPED = "skipped"
class CronAgentExecutionPhase(StrEnum):
INITIALIZING = "initializing"
PLANNING = "planning"
EXECUTING = "executing"
SUMMARIZING = "summarizing"
DELIVERING = "delivering"
class CronRunOutcome(StrEnum):
SUCCESS = "success"
FAILED = "failed"
TIMEOUT = "timeout"
CANCELLED = "cancelled"
@dataclass
class CronPhaseRecord:
phase: CronAgentExecutionPhase
started_at_ms: int
finished_at_ms: int | None = None
error: str | None = None
@property
def duration_ms(self) -> int | None:
if self.finished_at_ms is not None:
return self.finished_at_ms - self.started_at_ms
return None
def to_dict(self) -> dict:
return {
"phase": self.phase.value,
"started_at_ms": self.started_at_ms,
"finished_at_ms": self.finished_at_ms,
"error": self.error,
"duration_ms": self.duration_ms,
}
@dataclass
class CronDeliveryTarget:
channel: str
target_id: str
thread_id: str | None = None
reply_to_id: str | None = None
account_id: str | None = None
enabled: bool = True
def to_dict(self) -> dict:
return {
"channel": self.channel,
"target_id": self.target_id,
"thread_id": self.thread_id,
"reply_to_id": self.reply_to_id,
"account_id": self.account_id,
"enabled": self.enabled,
}
@classmethod
def from_dict(cls, data: dict) -> "CronDeliveryTarget":
return cls(
channel=data.get("channel", ""),
target_id=data.get("target_id", ""),
thread_id=data.get("thread_id"),
reply_to_id=data.get("reply_to_id"),
account_id=data.get("account_id"),
enabled=data.get("enabled", True),
)
@dataclass
class CronDeliveryReceipt:
target: CronDeliveryTarget
success: bool = False
message_id: str | None = None
error: str | None = None
retries: int = 0
sent_at_ms: int = 0
duration_ms: int = 0
def to_dict(self) -> dict:
return {
"target": self.target.to_dict(),
"success": self.success,
"message_id": self.message_id,
"error": self.error,
"retries": self.retries,
"sent_at_ms": self.sent_at_ms,
"duration_ms": self.duration_ms,
}
@dataclass
class CronFailureAlert:
threshold: int = 5
cooldown_ms: int = 600_000
last_alert_at_ms: int = 0
last_recovery_at_ms: int = 0
enabled: bool = True
def should_alert(self, consecutive_errors: int, now_ms: int) -> bool:
if not self.enabled:
return False
if consecutive_errors < self.threshold:
return False
if self.last_alert_at_ms > 0 and (now_ms - self.last_alert_at_ms) < self.cooldown_ms:
return False
return True
def should_recover(self, consecutive_errors: int, now_ms: int) -> bool:
if not self.enabled:
return False
if self.last_alert_at_ms <= 0:
return False
if self.last_recovery_at_ms > 0 and (now_ms - self.last_recovery_at_ms) < self.cooldown_ms:
return False
return consecutive_errors == 0
def mark_alerted(self, now_ms: int) -> None:
self.last_alert_at_ms = now_ms
def mark_recovered(self, now_ms: int) -> None:
self.last_recovery_at_ms = now_ms
def to_dict(self) -> dict:
return {
"threshold": self.threshold,
"cooldown_ms": self.cooldown_ms,
"last_alert_at_ms": self.last_alert_at_ms,
"last_recovery_at_ms": self.last_recovery_at_ms,
"enabled": self.enabled,
}
@classmethod
def from_dict(cls, data: dict | None) -> "CronFailureAlert":
if not data:
return cls()
return cls(
threshold=data.get("threshold", 5),
cooldown_ms=data.get("cooldown_ms", 600_000),
last_alert_at_ms=data.get("last_alert_at_ms", 0),
last_recovery_at_ms=data.get("last_recovery_at_ms", 0),
enabled=data.get("enabled", True),
)
@dataclass
class CronRunDiagnostic:
run_id: str = ""
job_id: str = ""
outcome: CronRunOutcome = CronRunOutcome.SUCCESS
phases: list[CronPhaseRecord] = field(default_factory=list)
total_duration_ms: int = 0
error: str | None = None
token_usage: dict = field(default_factory=dict)
delivery_receipts: list[CronDeliveryReceipt] = field(default_factory=list)
started_at_ms: int = 0
finished_at_ms: int = 0
def to_dict(self) -> dict:
return {
"run_id": self.run_id,
"job_id": self.job_id,
"outcome": self.outcome.value,
"phases": [p.to_dict() for p in self.phases],
"total_duration_ms": self.total_duration_ms,
"error": self.error,
"token_usage": self.token_usage,
"delivery_receipts": [r.to_dict() for r in self.delivery_receipts],
"started_at_ms": self.started_at_ms,
"finished_at_ms": self.finished_at_ms,
}
@dataclass
class CronJobState:
next_run_at_ms: int | None = None
running_at_ms: int | None = None
last_run_at_ms: int | None = None
last_run_status: CronRunStatus | None = None
last_error: str | None = None
last_duration_ms: int | None = None
consecutive_errors: int = 0
consecutive_skipped: int = 0
schedule_error_count: int = 0
run_count: int = 0
error_count: int = 0
def to_dict(self) -> dict:
return {
"next_run_at_ms": self.next_run_at_ms,
"running_at_ms": self.running_at_ms,
"last_run_at_ms": self.last_run_at_ms,
"last_run_status": self.last_run_status.value if self.last_run_status else None,
"last_error": self.last_error,
"last_duration_ms": self.last_duration_ms,
"consecutive_errors": self.consecutive_errors,
"consecutive_skipped": self.consecutive_skipped,
"schedule_error_count": self.schedule_error_count,
"run_count": self.run_count,
"error_count": self.error_count,
}
@classmethod
def from_dict(cls, data: dict | None) -> "CronJobState":
if not data:
return cls()
state = cls()
state.next_run_at_ms = data.get("next_run_at_ms")
state.running_at_ms = data.get("running_at_ms")
state.last_run_at_ms = data.get("last_run_at_ms")
status = data.get("last_run_status")
if status and status in {s.value for s in CronRunStatus}:
state.last_run_status = CronRunStatus(status)
state.last_error = data.get("last_error")
state.last_duration_ms = data.get("last_duration_ms")
state.consecutive_errors = data.get("consecutive_errors", 0)
state.consecutive_skipped = data.get("consecutive_skipped", 0)
state.schedule_error_count = data.get("schedule_error_count", 0)
state.run_count = data.get("run_count", 0)
state.error_count = data.get("error_count", 0)
return state
@dataclass
class CronJob:
id: str
name: str
handler: CronHandler
handler_name: str = ""
handler_args: dict = field(default_factory=dict)
schedule_kind: ScheduleKind = ScheduleKind.CRON
schedule_value: str = "* * * * *"
schedule_tz: str | None = None
stagger_ms: int = 0
every_ms: int | None = None
anchor_ms: int | None = None
status: CronJobStatus = CronJobStatus.ACTIVE
enabled: bool = True
max_runs: int = 0
delete_after_run: bool = False
error_threshold: int = 5
delivery: list[CronDeliveryTarget] = field(default_factory=list)
failure_alert: CronFailureAlert = field(default_factory=CronFailureAlert)
state: CronJobState = field(default_factory=CronJobState)
created_at: float = field(default_factory=time.time)
updated_at: float = field(default_factory=time.time)
def to_dict(self) -> dict:
return {
"id": self.id,
"name": self.name,
"handler_name": self.handler_name,
"schedule_kind": self.schedule_kind.value,
"schedule_value": self.schedule_value,
"schedule_tz": self.schedule_tz,
"stagger_ms": self.stagger_ms,
"every_ms": self.every_ms,
"anchor_ms": self.anchor_ms,
"status": self.status.value,
"enabled": self.enabled,
"max_runs": self.max_runs,
"delete_after_run": self.delete_after_run,
"error_threshold": self.error_threshold,
"delivery": [t.to_dict() for t in self.delivery],
"failure_alert": self.failure_alert.to_dict(),
"state": self.state.to_dict(),
"created_at": self.created_at,
"updated_at": self.updated_at,
}