"""定时任务守护循环 + 单 job 执行引擎(§8.5;从 app.py lifespan 析出,2026-07-23)。 职责边界:core/scheduler.py 是服务层(job CRUD / claim / next_run 计算 / 投递), 本模块是 web 侧执行引擎 —— 认领到点 job → 解析目标 task → 抢 run 锁 → 复用 run_agent_bg 起 run → 超时协作 cancel → 确定性兜底投递 + 回写 last_*。 住 web/ 不下沉 core:执行依赖 broker / runs / app.state(inflight/draining), 方向同 runs.py。 仿 _disk_scanner 的 plain-asyncio 范式,不引 APScheduler/Celery。tick 间隔只决定 最坏延迟(≤1 tick),不决定会不会漏(claim 取 next_run<=now 的全部)。 ZCBOT_DISABLE_SCHEDULER=1 整体关掉(对照 Claude Code CLAUDE_CODE_DISABLE_CRON)。 """ from __future__ import annotations import asyncio import os from typing import Optional from uuid import UUID, uuid4 from sqlalchemy import select, update from core.paths import from_db_path, to_db_path from core.storage import session_scope from core.storage.models import ScheduledJob, Task from .broker import broker from .common import INSTANCE from .model_gate import resolve_model_profile from .runs import run_agent_bg def start_scheduler(app, cfg: dict) -> Optional[asyncio.Task]: """起定时任务守护循环;ZCBOT_DISABLE_SCHEDULER 在则返 None。""" enabled = os.getenv("ZCBOT_DISABLE_SCHEDULER", "").strip() not in ("1", "true", "yes") if not enabled: return None sched_tick = int(os.getenv("ZCBOT_SCHEDULER_TICK_SECONDS", "10") or "10") sched_sema = asyncio.Semaphore(int(os.getenv("ZCBOT_SCHEDULER_CONCURRENCY", "4") or "4")) async def _execute_scheduled_job(snap: dict) -> None: """认领后跑一个 job:解析目标 task → 抢 run 锁 → run_agent_bg → 投递 + 记账。""" from core.agent_builder import ( resolve_workspace, working_dir_from_name, validate_task_name, InvalidTaskName, ) from core.scheduler import build_run_message, deliver_notify, record_result from core.storage.utils import ensure_local_task_row job_id = snap["job_id"] uid = snap["user_id"] async with sched_sema: try: profile, model_id = resolve_model_profile(snap.get("model_profile") or "") ws = resolve_workspace(None, cfg) # 目标 task:persistent 用绑定 task(缺则新建并回填);isolated 用稳定 per-job 目录 tid: Optional[UUID] = None if snap["mode"] == "persistent" and snap.get("bound_task_id"): tid = snap["bound_task_id"] # 绑定 task 可能已被删(SET NULL 已处理 None;这里再查实在性) with session_scope() as s: exists = s.execute( select(Task.task_id).where( Task.task_id == tid, Task.deleted_at.is_(None) ) ).first() if exists is None: tid = None if tid is None: tid = uuid4() wd_name = f"scheduled-{str(job_id)[:8]}" fs_dir = working_dir_from_name(ws, uid, wd_name) fs_dir.mkdir(parents=True, exist_ok=True) disp = f"{snap['name']}" try: disp = validate_task_name(disp) except InvalidTaskName: disp = wd_name # 名字含非法字符 → 退到安全名 ensure_local_task_row( task_id=tid, name=disp, working_dir=to_db_path(fs_dir), skill=snap.get("skill") or "", user_id=uid, model=model_id, model_profile=profile, description="(定时任务自动创建)", scheduled_job_id=job_id, ) if snap["mode"] == "persistent": with session_scope() as s: s.execute(update(ScheduledJob).where( ScheduledJob.job_id == job_id ).values(bound_task_id=tid)) # 抢 run 锁(同 post_message):busy → 本次跳过,下个 cron 点再来 with session_scope() as s: row = s.execute( select(Task.run_status).where(Task.task_id == tid).with_for_update() ).first() if row is None: record_result(job_id, status="error", task_id=tid, error="目标 task 不存在") return if row.run_status in ("running", "cancelling"): record_result(job_id, status="skipped", task_id=tid, error="目标 task 正忙,本次跳过") print(f"[scheduler] job {str(job_id)[:8]} skipped (task busy)") return s.execute(update(Task).where(Task.task_id == tid).values( run_status="running", run_error=None, run_owner=INSTANCE or None)) message = build_run_message(snap) broker.start(tid) runner = asyncio.create_task(asyncio.to_thread( run_agent_bg, tid, uid, message, "", "", True, )) app.state.inflight[runner] = tid runner.add_done_callback(lambda t: app.state.inflight.pop(t, None)) timeout = int(snap.get("timeout_seconds") or 0) timed_out = False if timeout > 0: done, _pending = await asyncio.wait({runner}, timeout=timeout) if not done: timed_out = True broker.request_cancel(tid) # 协作式停;loop 在 chunk 间 poll 到即退 print(f"[scheduler] job {str(job_id)[:8]} timed out ({timeout}s), cancelling") await runner else: await runner # 超时被掐断:run_agent_bg 对 ok/cancelled 都把 run_status 收回 idle # (二者在 DB 里不可区分),只有这里知道本次是 timeout 中断的。必须记为 # error —— 否则会误判成 ok(掩盖"跑到一半没推送"),且不计入连续失败/不触发 # 兜底。半成品不投递 notify,直接收尾返回。 if timed_out: record_result(job_id, status="error", task_id=tid, error=f"运行超过超时上限 {timeout}s 未完成,已中断(本次未推送)") print(f"[scheduler] job {str(job_id)[:8]} recorded as timeout-error") return # run 终态:run_agent_bg 收尾把 run_status 写回 idle(ok)/error with session_scope() as s: st = s.execute( select(Task.run_status, Task.run_error).where(Task.task_id == tid) ).first() if st is not None and st.run_status == "error": record_result(job_id, status="error", task_id=tid, error=st.run_error) print(f"[scheduler] job {str(job_id)[:8]} run error: {st.run_error}") return # 第 3 层确定性兜底投递(notify);失败不影响 run 已成功这一事实 if snap.get("notify"): try: with session_scope() as s: wd_db = s.execute( select(Task.working_dir).where(Task.task_id == tid) ).scalar_one_or_none() fs_dir = from_db_path(wd_db) if wd_db else ws await asyncio.get_running_loop().run_in_executor( None, lambda: deliver_notify( snap["notify"], job_name=snap["name"], working_dir=fs_dir, tz=snap["tz"], user_id=snap["user_id"], ) ) except Exception as e: print(f"[scheduler] job {str(job_id)[:8]} notify failed: {type(e).__name__}: {e}") record_result(job_id, status="ok", task_id=tid) print(f"[scheduler] job {str(job_id)[:8]} '{snap['name']}' done") except Exception as e: print(f"[scheduler] job {str(job_id)[:8]} crashed: {type(e).__name__}: {e}") try: record_result(job_id, status="error", task_id=None, error=f"{type(e).__name__}: {e}") except Exception: pass async def _scheduler_loop() -> None: from core.scheduler import claim_due_jobs loop = asyncio.get_running_loop() while True: try: await asyncio.sleep(sched_tick) if getattr(app.state, "draining", None) is not None and app.state.draining.is_set(): continue # 关停 drain 期不起新 job due = await loop.run_in_executor(None, claim_due_jobs) for snap in due: asyncio.create_task(_execute_scheduled_job(snap)) if due: print(f"[scheduler] fired {len(due)} job(s)") except asyncio.CancelledError: raise except Exception as e: print(f"[scheduler] loop error: {type(e).__name__}: {e}") task = asyncio.create_task(_scheduler_loop(), name="scheduler") print(f"[scheduler] enabled (tick={sched_tick}s)") return task