diff --git a/backend/package/yuxi/channel/cron/__init__.py b/backend/package/yuxi/channel/cron/__init__.py new file mode 100644 index 00000000..a8683503 --- /dev/null +++ b/backend/package/yuxi/channel/cron/__init__.py @@ -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", +] diff --git a/backend/package/yuxi/channel/cron/alert.py b/backend/package/yuxi/channel/cron/alert.py new file mode 100644 index 00000000..6e3f7c86 --- /dev/null +++ b/backend/package/yuxi/channel/cron/alert.py @@ -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} diff --git a/backend/package/yuxi/channel/cron/delivery.py b/backend/package/yuxi/channel/cron/delivery.py new file mode 100644 index 00000000..a3a5df88 --- /dev/null +++ b/backend/package/yuxi/channel/cron/delivery.py @@ -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) diff --git a/backend/package/yuxi/channel/cron/engine.py b/backend/package/yuxi/channel/cron/engine.py new file mode 100644 index 00000000..04eadd17 --- /dev/null +++ b/backend/package/yuxi/channel/cron/engine.py @@ -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 diff --git a/backend/package/yuxi/channel/cron/handler_registry.py b/backend/package/yuxi/channel/cron/handler_registry.py new file mode 100644 index 00000000..2a1f4ed8 --- /dev/null +++ b/backend/package/yuxi/channel/cron/handler_registry.py @@ -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"} \ No newline at end of file diff --git a/backend/package/yuxi/channel/cron/normalize.py b/backend/package/yuxi/channel/cron/normalize.py new file mode 100644 index 00000000..e816999e --- /dev/null +++ b/backend/package/yuxi/channel/cron/normalize.py @@ -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 diff --git a/backend/package/yuxi/channel/cron/phase_tracker.py b/backend/package/yuxi/channel/cron/phase_tracker.py new file mode 100644 index 00000000..c0229d7d --- /dev/null +++ b/backend/package/yuxi/channel/cron/phase_tracker.py @@ -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 diff --git a/backend/package/yuxi/channel/cron/schedule.py b/backend/package/yuxi/channel/cron/schedule.py new file mode 100644 index 00000000..d9b050bf --- /dev/null +++ b/backend/package/yuxi/channel/cron/schedule.py @@ -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 diff --git a/backend/package/yuxi/channel/cron/types.py b/backend/package/yuxi/channel/cron/types.py new file mode 100644 index 00000000..b607da9a --- /dev/null +++ b/backend/package/yuxi/channel/cron/types.py @@ -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, + }