203 lines
9.6 KiB
Python
203 lines
9.6 KiB
Python
"""定时任务守护循环 + 单 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 .model_gate import resolve_model_profile
|
||
from .run_lifecycle import (
|
||
RunScheduleError,
|
||
RunTaskBusy,
|
||
RunTaskNotFound,
|
||
claim_run_with_message,
|
||
schedule_claimed_run,
|
||
)
|
||
|
||
|
||
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))
|
||
|
||
message = build_run_message(snap)
|
||
try:
|
||
claim_run_with_message(tid, uid, message)
|
||
except RunTaskNotFound:
|
||
record_result(
|
||
job_id, status="error", task_id=tid,
|
||
error="目标 task 不存在",
|
||
)
|
||
return
|
||
except RunTaskBusy:
|
||
record_result(
|
||
job_id, status="skipped", task_id=tid,
|
||
error="目标 task 正忙,本次跳过",
|
||
)
|
||
print(f"[scheduler] job {str(job_id)[:8]} skipped (task busy)")
|
||
return
|
||
try:
|
||
runner = schedule_claimed_run(
|
||
app, tid, uid, message, scheduled=True,
|
||
)
|
||
except RunScheduleError as e:
|
||
record_result(
|
||
job_id, status="error", task_id=tid, error=str(e),
|
||
)
|
||
print(
|
||
f"[scheduler] job {str(job_id)[:8]} scheduling failed: {e}"
|
||
)
|
||
return
|
||
|
||
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 把 task 标 cancelled，但只有这里知道它是
|
||
# scheduler timeout 而非用户停止。job 必须另记 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
|