194 lines
9.7 KiB
Python
194 lines
9.7 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 .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
|