refactor(web): lifespan 后台协程按域析出——app.py 收敛为纯编排(732→200 行)
- web/scheduler_runner.py:定时任务执行引擎(claim→抢锁→run_agent_bg→ 超时协作 cancel→投递记账),start_scheduler 一个入口 - web/wechat_runner.py:ClawBot 入站长轮询 + PG advisory lock 选主 - web/background.py:磁盘扫描/线程池监控/工具失败巡检/sandbox 池初始化+ reaper/proc 清扫/孤儿 run 收割/优雅 drain,统一 start_*/cancel_and_wait - app.py lifespan 只剩:executor 接管、state 装配、协程起停编排、收尾顺序 行为零变化(启动钩子与关停顺序逐一保持);286 测试全过,TestClient lifespan 起停冒烟干净。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
25d8a6572f
commit
546cb34d94
621
web/app.py
621
web/app.py
|
|
@ -8,6 +8,11 @@
|
|||
- 豁免:/healthz、/docs、/openapi.json、/、/v1/auth/login、/static/*
|
||||
- CORS allow_origins=["*"] 本地宽松;真发布按 platform 域名收紧
|
||||
- `GET /` 302 → /static/dev.html(本地 dev SPA)
|
||||
|
||||
本文件只剩两件事(2026-07-23 拆分后):
|
||||
- `create_app` 工厂:装配 auth 依赖 → 注册 11 个 router 模块(web/routers/)+ admin
|
||||
- `lifespan` 编排:线程池接管 / 孤儿收割 / 6 类后台协程起停 / 优雅 drain
|
||||
(协程本体在 web/background.py + scheduler_runner.py + wechat_runner.py)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -17,22 +22,11 @@ import os
|
|||
from concurrent.futures import ThreadPoolExecutor
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
import resource # Unix only;Windows dev 无此模块,RSS 监控自动降级跳过
|
||||
except ImportError: # pragma: no cover - Windows
|
||||
resource = None
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from sqlalchemy import or_, select, update
|
||||
|
||||
from core import __version__
|
||||
from core.paths import ROOT, from_db_path, to_db_path
|
||||
from core.storage import session_scope
|
||||
from core.storage.models import ScheduledJob, Task
|
||||
|
||||
from .auth import (
|
||||
REFRESHED_TOKEN_HEADER,
|
||||
|
|
@ -42,29 +36,17 @@ from .auth import (
|
|||
make_require_user,
|
||||
)
|
||||
from .admin import register_admin_routes
|
||||
from .background import (
|
||||
cancel_and_wait,
|
||||
drain_inflight,
|
||||
init_sandbox,
|
||||
reap_stale_runs,
|
||||
start_disk_scanner,
|
||||
start_proc_sweeper,
|
||||
start_stats_logger,
|
||||
start_toolfail_scanner,
|
||||
)
|
||||
from .broker import broker
|
||||
from .common import (
|
||||
CHANNEL_MIRROR_KINDS,
|
||||
INSTANCE,
|
||||
STATUS_FILTERS,
|
||||
STATUS_WRITABLE,
|
||||
iso as _iso,
|
||||
outline_snippet as _outline_snippet,
|
||||
parse_ordering as _parse_ordering,
|
||||
sse_event as _sse_event,
|
||||
task_dict as _task_dict,
|
||||
usage_aggregates as _usage_aggregates,
|
||||
)
|
||||
from .model_gate import (
|
||||
FALLBACK_MODEL_PROFILE,
|
||||
list_image_variants as _list_image_variants,
|
||||
list_video_variants as _list_video_variants,
|
||||
model_allowed_for_user as _model_allowed_for_user,
|
||||
resolve_image_model as _resolve_image_model,
|
||||
resolve_model_profile as _resolve_model_profile,
|
||||
resolve_video_model as _resolve_video_model,
|
||||
skill_pinned_profiles as _skill_pinned_profiles,
|
||||
)
|
||||
from .routers.asr import register_asr_routes
|
||||
from .routers.authroutes import register_auth_routes
|
||||
from .routers.files import register_file_routes
|
||||
|
|
@ -76,11 +58,9 @@ from .routers.schedules import register_schedule_routes
|
|||
from .routers.skills_memory import register_skill_memory_routes
|
||||
from .routers.tasks import register_task_routes
|
||||
from .routers.wechat import register_wechat_routes
|
||||
from .runs import (
|
||||
run_agent_bg as _run_agent_bg,
|
||||
run_channel_conversation as _run_channel_conversation,
|
||||
)
|
||||
from .scheduler_runner import start_scheduler
|
||||
from .static_files import NoCacheStaticFiles
|
||||
from .wechat_runner import start_wechat_inbound
|
||||
|
||||
# ────────────────────── App 工厂 ──────────────────────
|
||||
|
||||
|
|
@ -117,7 +97,7 @@ def create_app() -> FastAPI:
|
|||
print(f"[startup] run executor: max_workers={run_max_workers} "
|
||||
f"(override via ZCBOT_RUN_MAX_WORKERS)")
|
||||
|
||||
from core.agent_builder import load_config, resolve_workspace
|
||||
from core.agent_builder import load_config
|
||||
_cfg = load_config()
|
||||
|
||||
# 优雅 drain 状态(SIGTERM / systemctl restart 兜底,见下方 finally):
|
||||
|
|
@ -129,554 +109,39 @@ def create_app() -> FastAPI:
|
|||
drain_timeout = int(_shutdown_cfg.get("drain_timeout_seconds") or 90)
|
||||
cancel_grace = int(_shutdown_cfg.get("cancel_grace_seconds") or 15)
|
||||
|
||||
# Stale-run reaper:上次进程 crash 留下的 "running" / "cancelling" 已无 BG 线程
|
||||
# 继续,启动时标 error,让对应 task 重新可发消息(否则 gate 永挂)。
|
||||
# 蓝绿双实例(RUN.md B 档):ZCBOT_INSTANCE 在时只收 run_owner=自己(上次本色
|
||||
# 实例的孤儿)或 NULL(0020 前的遗留行)的,不动另一实例正在跑的 run;单实例
|
||||
# 部署 INSTANCE="" → 全量收,行为不变。
|
||||
with session_scope() as s:
|
||||
stmt = update(Task).where(Task.run_status.in_(("running", "cancelling")))
|
||||
if INSTANCE:
|
||||
stmt = stmt.where(
|
||||
or_(Task.run_owner == INSTANCE, Task.run_owner.is_(None))
|
||||
)
|
||||
result = s.execute(
|
||||
stmt.values(
|
||||
run_status="error",
|
||||
run_error="server restarted before run finished",
|
||||
)
|
||||
)
|
||||
if result.rowcount:
|
||||
print(f"[startup] reaped {result.rowcount} stale active run(s)")
|
||||
# 启动钩子 + 后台协程群(本体见 web/background.py 等;None=该项未启用)
|
||||
reap_stale_runs()
|
||||
disk_scanner_task = start_disk_scanner(_cfg)
|
||||
stats_logger_task = start_stats_logger(app, run_max_workers)
|
||||
toolfail_task = start_toolfail_scanner()
|
||||
scheduler_task = start_scheduler(app, _cfg)
|
||||
wechat_task, wechat_stop = start_wechat_inbound(app)
|
||||
sandbox_reaper_task = init_sandbox(app, _cfg)
|
||||
proc_sweeper_task = start_proc_sweeper(_cfg)
|
||||
|
||||
# 磁盘配额后台扫描(§7.5 #4 应用层 gate)── 不依赖 docker backend,host
|
||||
# backend 也跑(/v1/files/upload 也走配额 gate)。yaml `quotas.disk_scan_interval_seconds`
|
||||
# 默 900s = 15min;limit_bytes ≤ 0 视为不限,scan 仍跑(用量统计有用),check 短路放行。
|
||||
from core.agent_builder import resolve_workspace
|
||||
from core.storage.disk_quota import parse_bytes, scan_all_users
|
||||
workspace = resolve_workspace(None, _cfg)
|
||||
disk_user_root = workspace / "users"
|
||||
quotas_cfg = _cfg.get("quotas") or {}
|
||||
disk_scan_interval = int(quotas_cfg.get("disk_scan_interval_seconds") or 900)
|
||||
|
||||
async def _disk_scanner() -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
# 启动时跑一次,后续按 interval。首次扫完 check 才能命中。
|
||||
try:
|
||||
n = await loop.run_in_executor(None, scan_all_users, disk_user_root)
|
||||
if n:
|
||||
print(f"[disk_scanner] initial scan: {n} user(s)")
|
||||
except Exception as e:
|
||||
print(f"[disk_scanner] initial scan error: {type(e).__name__}: {e}")
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(disk_scan_interval)
|
||||
n = await loop.run_in_executor(None, scan_all_users, disk_user_root)
|
||||
if n:
|
||||
print(f"[disk_scanner] scanned {n} user(s)")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[disk_scanner] error: {type(e).__name__}: {e}")
|
||||
|
||||
disk_scanner_task = asyncio.create_task(_disk_scanner(), name="disk-scanner")
|
||||
|
||||
# ── 并发/线程池监控(§8.4):周期采样,只在有负载/刷新峰值时打,空闲不刷屏 ──
|
||||
# active_runs 来自 inflight(已提交未完成的 run,含排队中);逼近 max_workers 即
|
||||
# 线程池排队,新 run 的 SSE 会卡着不吐 token。查看:journalctl -u zcbot | grep '\[stats\]'
|
||||
def _rss_peak_mb() -> Optional[float]:
|
||||
if resource is None:
|
||||
return None # Windows dev:降级,不打 rss
|
||||
# Linux ru_maxrss 单位 KB,是峰值/high-water(单调不降 —— 看内存涨势够用)
|
||||
return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
|
||||
|
||||
async def _stats_logger() -> None:
|
||||
peak = 0
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(60)
|
||||
active = len(app.state.inflight)
|
||||
if active > peak:
|
||||
peak = active
|
||||
warn = " [WARN >= max_workers,已在排队]" if active >= run_max_workers else ""
|
||||
print(f"[stats] new peak active_runs={active} "
|
||||
f"max_workers={run_max_workers}{warn}")
|
||||
if active > 0:
|
||||
rss = _rss_peak_mb()
|
||||
rss_s = f" rss_peak={rss:.0f}MB" if rss is not None else ""
|
||||
print(f"[stats] active_runs={active} "
|
||||
f"max_workers={run_max_workers} "
|
||||
f"sse_subs={broker.total_subscribers()}{rss_s}")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[stats] error: {type(e).__name__}: {e}")
|
||||
|
||||
stats_logger_task = asyncio.create_task(_stats_logger(), name="stats-logger")
|
||||
|
||||
# ── 工具失败聚集巡检(0.58.11)── 反 9dcae061 教训(mmdc 在生产挂 90 天
|
||||
# 0 成功无人知,靠人工扫 DB 才发现):每天扫近 7 天 tool 错误消息,同签名
|
||||
# >=5 次且跨 >=2 task 判聚集,发 ZCBOT_DEVELOPER_EMAIL(复用 SMTP_*;未配
|
||||
# 邮箱/SMTP 则只打日志)。签名进程内去重,且只发近 24h 仍活跃的聚集
|
||||
# (count_24h>0)—— 去重集是内存态,重启/蓝绿清零,高频部署期若不加活跃
|
||||
# 过滤,每次部署都会把 7 天窗口内早已安静的存量聚集重发一遍(0.58.19 实改;
|
||||
# 还在烧的重启后再提醒一次是刻意保留的)。ZCBOT_TOOLFAIL_SCAN_INTERVAL
|
||||
# 秒,默 86400,<=0 整体关掉。
|
||||
dev_email = os.getenv("ZCBOT_DEVELOPER_EMAIL", "").strip()
|
||||
toolfail_interval = int(os.getenv("ZCBOT_TOOLFAIL_SCAN_INTERVAL", "").strip() or 86400)
|
||||
|
||||
async def _toolfail_scanner() -> None:
|
||||
from core.toolfail import format_alert, scan_tool_failures
|
||||
from tools.send_email import send_email_smtp, smtp_configured
|
||||
loop = asyncio.get_running_loop()
|
||||
alerted: set = set()
|
||||
while True:
|
||||
try:
|
||||
clusters = await loop.run_in_executor(None, scan_tool_failures)
|
||||
fresh = [c for c in clusters
|
||||
if c.get("count_24h")
|
||||
and (c["tool"], c["signature"]) not in alerted]
|
||||
if fresh:
|
||||
alerted.update((c["tool"], c["signature"]) for c in fresh)
|
||||
for c in fresh:
|
||||
print(f"[toolfail] {c['tool']}/{c['kind']} x{c['count']} "
|
||||
f"tasks={c['task_count']}: {c['signature'][:80]}")
|
||||
if dev_email and smtp_configured():
|
||||
await loop.run_in_executor(
|
||||
None, send_email_smtp, dev_email,
|
||||
f"[zcbot] 工具失败聚集 {len(fresh)} 类",
|
||||
format_alert(fresh, 7),
|
||||
)
|
||||
else:
|
||||
print("[toolfail] ZCBOT_DEVELOPER_EMAIL/SMTP 未配,仅日志")
|
||||
await asyncio.sleep(toolfail_interval)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[toolfail] error: {type(e).__name__}: {e}")
|
||||
await asyncio.sleep(toolfail_interval)
|
||||
|
||||
toolfail_task = (
|
||||
asyncio.create_task(_toolfail_scanner(), name="toolfail-scanner")
|
||||
if toolfail_interval > 0 else None
|
||||
)
|
||||
|
||||
# ── 定时任务守护循环(§8.5)── 仿 _disk_scanner 的 plain-asyncio 范式,不引
|
||||
# APScheduler/Celery。每 ~10s 认领到点 job(claim+advance next_run 防重复触发),
|
||||
# 复用 _run_agent_bg 起 run,跑完确定性兜底投递 + 回写 last_*。间隔只决定最坏延迟
|
||||
# (≤1 tick),不决定会不会漏(claim 取 next_run<=now 的全部)。ZCBOT_DISABLE_SCHEDULER=1
|
||||
# 整体关掉(对照 Claude Code CLAUDE_CODE_DISABLE_CRON)。
|
||||
scheduler_enabled = os.getenv("ZCBOT_DISABLE_SCHEDULER", "").strip() not in ("1", "true", "yes")
|
||||
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}")
|
||||
|
||||
scheduler_task = asyncio.create_task(_scheduler_loop(), name="scheduler") if scheduler_enabled else None
|
||||
if scheduler_enabled:
|
||||
print(f"[scheduler] enabled (tick={sched_tick}s)")
|
||||
|
||||
# ── 微信(ClawBot)入站长轮询管理器(§8.7)── 仅当 ZCBOT_WECHAT_BOT_ENABLED 在。
|
||||
# 每个 active 绑定一条 getupdates 长轮询;收到消息 → 跑用户常驻「微信」task → 回复发回。
|
||||
from core.wechat.service import clawbot_enabled
|
||||
wechat_stop = asyncio.Event()
|
||||
wechat_task = None
|
||||
|
||||
async def _run_wechat_message(uid: UUID, text: str, attachments=None) -> str:
|
||||
"""微信(ClawBot)入站一条消息 → 跑用户常驻「微信」task → 取回复。
|
||||
|
||||
attachments:已下载解密的入站附件(core.wechat.ilink.InboundAttachment,att.data 已回填)。
|
||||
建/复用 task、落盘附件、抢 run 锁、跑 agent 全在渠道无关核心
|
||||
`_run_channel_conversation` 里(企业微信回调走同一核心,channel='wecom')。
|
||||
"""
|
||||
return await _run_channel_conversation(app, uid, text, attachments, channel="wechat")
|
||||
|
||||
if clawbot_enabled():
|
||||
from core.wechat.inbound import run_inbound_manager
|
||||
|
||||
async def _wechat_leader() -> None:
|
||||
"""ClawBot 入站长轮询单实例互斥(蓝绿 B 档,RUN.md)。
|
||||
|
||||
getupdates 长轮询是消费型的:蓝绿并存期两实例同时拉会抢消息 / 同一条
|
||||
入站双跑 agent。用 PG advisory lock 选主:持锁实例才跑 inbound manager,
|
||||
另一实例每 15s 重试;持锁连接断(DB 重启 / 网络抖 = 锁已自动释放)→
|
||||
停管理器重新竞争。单实例部署恒拿到锁,行为与之前完全一致。
|
||||
"""
|
||||
from core.storage import get_engine
|
||||
engine = get_engine()
|
||||
# pg_try_advisory_lock(int4, int4) 的固定 key:('zc'=0x7A63, 1=wechat-inbound)
|
||||
lock_sql = "SELECT pg_try_advisory_lock(31331, 1)"
|
||||
|
||||
def _acquire():
|
||||
conn = engine.connect()
|
||||
# 从连接池摘出独占:advisory lock 是 DBAPI 连接级的,不 detach 的话
|
||||
# close() 只是还池,锁挂在池化连接上既不释放也没人持有
|
||||
conn.detach()
|
||||
try:
|
||||
if conn.exec_driver_sql(lock_sql).scalar():
|
||||
return conn
|
||||
conn.close()
|
||||
return None
|
||||
except Exception:
|
||||
conn.close()
|
||||
raise
|
||||
|
||||
async def _wait(seconds: float) -> None:
|
||||
try:
|
||||
await asyncio.wait_for(wechat_stop.wait(), timeout=seconds)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
while not wechat_stop.is_set():
|
||||
conn = None
|
||||
mgr = None
|
||||
try:
|
||||
conn = await asyncio.to_thread(_acquire)
|
||||
if conn is None:
|
||||
await _wait(15)
|
||||
continue
|
||||
print("[wechat] inbound leader lock acquired, manager starting")
|
||||
mgr = asyncio.create_task(
|
||||
run_inbound_manager(_run_wechat_message, wechat_stop),
|
||||
name="wechat-inbound",
|
||||
)
|
||||
# 30s keepalive ping 锁连接;ping 挂 = 连接断 = 锁已丢 → 重选
|
||||
while not wechat_stop.is_set() and not mgr.done():
|
||||
await _wait(30)
|
||||
if wechat_stop.is_set() or mgr.done():
|
||||
break
|
||||
try:
|
||||
await asyncio.to_thread(
|
||||
lambda: conn.exec_driver_sql("SELECT 1").scalar()
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"[wechat] leader lock lost "
|
||||
f"({type(e).__name__}), re-electing")
|
||||
break
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[wechat] leader loop error: {type(e).__name__}: {e}")
|
||||
await _wait(15)
|
||||
finally:
|
||||
if mgr is not None and not mgr.done():
|
||||
mgr.cancel()
|
||||
try:
|
||||
await mgr
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
if conn is not None:
|
||||
try:
|
||||
conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
wechat_task = asyncio.create_task(_wechat_leader(), name="wechat-leader")
|
||||
print("[wechat] ClawBot inbound enabled (PG advisory lock leader election)")
|
||||
|
||||
# Sandbox pool(§7.5):仅当 ZCBOT_SANDBOX_BACKEND=docker 时启用。
|
||||
# 启动钩子:① init_pool(创建 docker network + pool 实例)② shutdown_all 清
|
||||
# 前驱孤儿(上次进程留下的 zcbot-sandbox-* 容器,内存 _last_active 为空,
|
||||
# 全清重启)③ 后台 reaper task,每 60s 跑 reap_idle。
|
||||
sandbox_backend = os.getenv("ZCBOT_SANDBOX_BACKEND", "host").lower()
|
||||
sandbox_reaper_task = None
|
||||
if sandbox_backend == "docker":
|
||||
from core.paths import ROOT
|
||||
from core.sandbox import init_pool
|
||||
from core.sandbox.check import detect_fs_quota
|
||||
workspace = resolve_workspace(None, _cfg)
|
||||
user_root_base = workspace / "users"
|
||||
# §7.5 #4 fs quota 探测:不阻塞启动(应用层周期扫描已有),仅打 WARN
|
||||
# 提醒外部用户开放前必须升级到 xfs prjquota / ext4 project / zfs。
|
||||
try:
|
||||
level, msg = detect_fs_quota(user_root_base.resolve())
|
||||
print(f"[startup] {'[ok]' if level == 'ok' else '[warn]'} {msg}")
|
||||
except Exception as e:
|
||||
print(f"[startup] [warn] fs quota detect failed: {type(e).__name__}: {e}")
|
||||
try:
|
||||
# repo_root=ROOT 让 SandboxPool 把 <repo>/skills 只读 mount 进容器
|
||||
# (fs 工具进容器后 read SKILL references 需要)
|
||||
# sandbox_cfg=yaml `sandbox` 段(memory/cpus/pids_limit 可调)
|
||||
pool = init_pool(
|
||||
user_root_base, repo_root=ROOT,
|
||||
sandbox_cfg=_cfg.get("sandbox") or {},
|
||||
)
|
||||
removed = pool.shutdown_all()
|
||||
if removed:
|
||||
print(f"[startup] swept {len(removed)} stale sandbox container(s)")
|
||||
|
||||
async def _reaper() -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(60)
|
||||
removed = await loop.run_in_executor(None, pool.reap_idle)
|
||||
if removed:
|
||||
print(f"[reaper] reaped {len(removed)} idle sandbox container(s)")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[reaper] error: {type(e).__name__}: {e}")
|
||||
|
||||
sandbox_reaper_task = asyncio.create_task(_reaper(), name="sandbox-reaper")
|
||||
app.state.sandbox_pool = pool
|
||||
except Exception as e:
|
||||
# ensure_network / docker CLI 不可用 → fail-fast。Stage C 协议:任一
|
||||
# hardening 缺失视为部署未完成,不退化到 host(否则误以为有沙盒实则在裸跑)。
|
||||
raise RuntimeError(
|
||||
f"ZCBOT_SANDBOX_BACKEND=docker but sandbox init failed: {e}"
|
||||
)
|
||||
|
||||
# bg proc 清扫(DESIGN §8.12):终态 proc 目录过 TTL 删除 + 已结束的
|
||||
# zcbot-proc-* 容器回收。host / docker 两种 backend 都要跑(host 模式只做
|
||||
# 文件清扫,docker CLI 不在时 sweep 内部静默跳过容器部分)。每小时一次;
|
||||
# 幂等,蓝绿双实例同时跑无害。启动后先跑一轮,把上个进程周期留下的
|
||||
# 已结束容器/过期目录收掉。
|
||||
from core.procs import sweep as _procs_sweep
|
||||
_procs_users_base = resolve_workspace(None, _cfg) / "users"
|
||||
|
||||
async def _proc_sweeper() -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
while True:
|
||||
try:
|
||||
stats = await loop.run_in_executor(
|
||||
None, _procs_sweep, _procs_users_base
|
||||
)
|
||||
if stats["removed_dirs"] or stats["reaped_containers"]:
|
||||
print(f"[proc-sweep] dirs={stats['removed_dirs']} "
|
||||
f"containers={stats['reaped_containers']}")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[proc-sweep] error: {type(e).__name__}: {e}")
|
||||
await asyncio.sleep(3600)
|
||||
|
||||
proc_sweeper_task = asyncio.create_task(_proc_sweeper(), name="proc-sweeper")
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
# ── 优雅 drain:先拒新 run,等在跑的 run 收尾,超时转协作式 cancel ──
|
||||
# 单实例形态下消除"restart 误杀 in-flight run 标 error"。新 POST /messages
|
||||
# 期间返 503(客户端退避重试覆盖)。drain_timeout 内自然跑完 → idle 零 error;
|
||||
# 超时的 broker.request_cancel → 下个 chunk 间隙退(标 idle);cancel_grace 后仍
|
||||
# 没退的留给 systemd SIGKILL,下次启动 reaper 标 error(最坏退化 = 改前行为)。
|
||||
# ★ systemd TimeoutStopSec 必须 > drain_timeout + cancel_grace + 余量(见 RUN.md)。
|
||||
# 先拒新 run + drain in-flight(细节见 background.drain_inflight)
|
||||
app.state.draining.set()
|
||||
inflight = app.state.inflight
|
||||
if inflight:
|
||||
print(f"[shutdown] draining {len(inflight)} in-flight run(s), "
|
||||
f"timeout={drain_timeout}s")
|
||||
_, pending = await asyncio.wait(
|
||||
list(inflight.keys()), timeout=drain_timeout
|
||||
)
|
||||
if pending:
|
||||
print(f"[shutdown] {len(pending)} run(s) over drain timeout; "
|
||||
f"signalling cooperative cancel")
|
||||
for t in pending:
|
||||
cid = inflight.get(t)
|
||||
if cid is not None:
|
||||
broker.request_cancel(cid)
|
||||
_, still = await asyncio.wait(pending, timeout=cancel_grace)
|
||||
if still:
|
||||
print(f"[shutdown] {len(still)} run(s) still active after "
|
||||
f"cancel grace; SIGKILL takes over, next start reaps them")
|
||||
await drain_inflight(app, drain_timeout, cancel_grace)
|
||||
|
||||
disk_scanner_task.cancel()
|
||||
try:
|
||||
await disk_scanner_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
stats_logger_task.cancel()
|
||||
try:
|
||||
await stats_logger_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
if toolfail_task is not None:
|
||||
toolfail_task.cancel()
|
||||
try:
|
||||
await toolfail_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
if scheduler_task is not None:
|
||||
scheduler_task.cancel()
|
||||
try:
|
||||
await scheduler_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
await cancel_and_wait(disk_scanner_task)
|
||||
await cancel_and_wait(stats_logger_task)
|
||||
await cancel_and_wait(toolfail_task)
|
||||
await cancel_and_wait(scheduler_task)
|
||||
if wechat_task is not None:
|
||||
wechat_stop.set()
|
||||
wechat_task.cancel()
|
||||
await cancel_and_wait(wechat_task)
|
||||
await cancel_and_wait(sandbox_reaper_task)
|
||||
await cancel_and_wait(proc_sweeper_task)
|
||||
|
||||
pool = getattr(app.state, "sandbox_pool", None)
|
||||
if pool is not None:
|
||||
try:
|
||||
await wechat_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
if sandbox_reaper_task is not None:
|
||||
sandbox_reaper_task.cancel()
|
||||
try:
|
||||
await sandbox_reaper_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
proc_sweeper_task.cancel()
|
||||
try:
|
||||
await proc_sweeper_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
if sandbox_backend == "docker":
|
||||
pool = getattr(app.state, "sandbox_pool", None)
|
||||
if pool is not None:
|
||||
try:
|
||||
pool.shutdown_all()
|
||||
except Exception as e:
|
||||
print(f"[shutdown] sandbox shutdown_all error: {type(e).__name__}: {e}")
|
||||
pool.shutdown_all()
|
||||
except Exception as e:
|
||||
print(f"[shutdown] sandbox shutdown_all error: {type(e).__name__}: {e}")
|
||||
|
||||
# broker 收尾(redis 版停 pubsub reader + 断连;local no-op)。放在 drain
|
||||
# 之后 —— drain 期间 run 还要 emit/close。
|
||||
|
|
@ -713,7 +178,7 @@ def create_app() -> FastAPI:
|
|||
mimetypes.add_type("text/javascript", ".js")
|
||||
app.mount("/static", NoCacheStaticFiles(directory=str(_STATIC_DIR)), name="static")
|
||||
|
||||
# ───────────── 拆分出的路由模块(与 register_admin_routes 同范式)─────────────
|
||||
# ───────────── 路由模块(与 register_admin_routes 同范式)─────────────
|
||||
register_misc_routes(app, require_user=require_user)
|
||||
register_wechat_routes(app, require_user=require_user, auth_cfg=auth_cfg)
|
||||
register_model_routes(app, require_user=require_user)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,296 @@
|
|||
"""web 层维护性后台协程(从 app.py lifespan 析出,2026-07-23)。
|
||||
|
||||
磁盘配额扫描 / 线程池监控 / 工具失败巡检 / sandbox 池初始化+reaper /
|
||||
bg proc 清扫 / 孤儿 run 收割 / 优雅 drain。每个 start_* 返回 asyncio.Task
|
||||
(或 None=未启用),关停统一走 `cancel_and_wait`。
|
||||
定时任务引擎另见 scheduler_runner.py,微信入站另见 wechat_runner.py。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy import or_, update
|
||||
|
||||
from core.storage import session_scope
|
||||
from core.storage.models import Task
|
||||
|
||||
from .broker import broker
|
||||
from .common import INSTANCE
|
||||
|
||||
try:
|
||||
import resource # Unix only;Windows dev 无此模块,RSS 监控自动降级跳过
|
||||
except ImportError: # pragma: no cover - Windows
|
||||
resource = None
|
||||
|
||||
|
||||
async def cancel_and_wait(task: Optional[asyncio.Task]) -> None:
|
||||
"""取消一个后台 task 并吞掉收尾异常(None 直接跳过)。"""
|
||||
if task is None:
|
||||
return
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
|
||||
|
||||
def reap_stale_runs() -> None:
|
||||
"""Stale-run reaper:上次进程 crash 留下的 "running" / "cancelling" 已无 BG 线程
|
||||
继续,启动时标 error,让对应 task 重新可发消息(否则 gate 永挂)。
|
||||
蓝绿双实例(RUN.md B 档):ZCBOT_INSTANCE 在时只收 run_owner=自己(上次本色
|
||||
实例的孤儿)或 NULL(0020 前的遗留行)的,不动另一实例正在跑的 run;单实例
|
||||
部署 INSTANCE="" → 全量收,行为不变。
|
||||
"""
|
||||
with session_scope() as s:
|
||||
stmt = update(Task).where(Task.run_status.in_(("running", "cancelling")))
|
||||
if INSTANCE:
|
||||
stmt = stmt.where(
|
||||
or_(Task.run_owner == INSTANCE, Task.run_owner.is_(None))
|
||||
)
|
||||
result = s.execute(
|
||||
stmt.values(
|
||||
run_status="error",
|
||||
run_error="server restarted before run finished",
|
||||
)
|
||||
)
|
||||
if result.rowcount:
|
||||
print(f"[startup] reaped {result.rowcount} stale active run(s)")
|
||||
|
||||
|
||||
def start_disk_scanner(cfg: dict) -> asyncio.Task:
|
||||
"""磁盘配额后台扫描(§7.5 #4 应用层 gate)── 不依赖 docker backend,host
|
||||
backend 也跑(/v1/files/upload 也走配额 gate)。yaml `quotas.disk_scan_interval_seconds`
|
||||
默 900s = 15min;limit_bytes ≤ 0 视为不限,scan 仍跑(用量统计有用),check 短路放行。
|
||||
"""
|
||||
from core.agent_builder import resolve_workspace
|
||||
from core.storage.disk_quota import scan_all_users
|
||||
disk_user_root = resolve_workspace(None, cfg) / "users"
|
||||
quotas_cfg = cfg.get("quotas") or {}
|
||||
interval = int(quotas_cfg.get("disk_scan_interval_seconds") or 900)
|
||||
|
||||
async def _disk_scanner() -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
# 启动时跑一次,后续按 interval。首次扫完 check 才能命中。
|
||||
try:
|
||||
n = await loop.run_in_executor(None, scan_all_users, disk_user_root)
|
||||
if n:
|
||||
print(f"[disk_scanner] initial scan: {n} user(s)")
|
||||
except Exception as e:
|
||||
print(f"[disk_scanner] initial scan error: {type(e).__name__}: {e}")
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(interval)
|
||||
n = await loop.run_in_executor(None, scan_all_users, disk_user_root)
|
||||
if n:
|
||||
print(f"[disk_scanner] scanned {n} user(s)")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[disk_scanner] error: {type(e).__name__}: {e}")
|
||||
|
||||
return asyncio.create_task(_disk_scanner(), name="disk-scanner")
|
||||
|
||||
|
||||
def start_stats_logger(app, run_max_workers: int) -> asyncio.Task:
|
||||
"""并发/线程池监控(§8.4):周期采样,只在有负载/刷新峰值时打,空闲不刷屏。
|
||||
|
||||
active_runs 来自 inflight(已提交未完成的 run,含排队中);逼近 max_workers 即
|
||||
线程池排队,新 run 的 SSE 会卡着不吐 token。查看:journalctl -u zcbot | grep '\\[stats\\]'
|
||||
"""
|
||||
def _rss_peak_mb() -> Optional[float]:
|
||||
if resource is None:
|
||||
return None # Windows dev:降级,不打 rss
|
||||
# Linux ru_maxrss 单位 KB,是峰值/high-water(单调不降 —— 看内存涨势够用)
|
||||
return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
|
||||
|
||||
async def _stats_logger() -> None:
|
||||
peak = 0
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(60)
|
||||
active = len(app.state.inflight)
|
||||
if active > peak:
|
||||
peak = active
|
||||
warn = " [WARN >= max_workers,已在排队]" if active >= run_max_workers else ""
|
||||
print(f"[stats] new peak active_runs={active} "
|
||||
f"max_workers={run_max_workers}{warn}")
|
||||
if active > 0:
|
||||
rss = _rss_peak_mb()
|
||||
rss_s = f" rss_peak={rss:.0f}MB" if rss is not None else ""
|
||||
print(f"[stats] active_runs={active} "
|
||||
f"max_workers={run_max_workers} "
|
||||
f"sse_subs={broker.total_subscribers()}{rss_s}")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[stats] error: {type(e).__name__}: {e}")
|
||||
|
||||
return asyncio.create_task(_stats_logger(), name="stats-logger")
|
||||
|
||||
|
||||
def start_toolfail_scanner() -> Optional[asyncio.Task]:
|
||||
"""工具失败聚集巡检(0.58.11)── 反 9dcae061 教训(mmdc 在生产挂 90 天
|
||||
0 成功无人知,靠人工扫 DB 才发现):每天扫近 7 天 tool 错误消息,同签名
|
||||
>=5 次且跨 >=2 task 判聚集,发 ZCBOT_DEVELOPER_EMAIL(复用 SMTP_*;未配
|
||||
邮箱/SMTP 则只打日志)。签名进程内去重,且只发近 24h 仍活跃的聚集
|
||||
(count_24h>0)—— 去重集是内存态,重启/蓝绿清零,高频部署期若不加活跃
|
||||
过滤,每次部署都会把 7 天窗口内早已安静的存量聚集重发一遍(0.58.19 实改;
|
||||
还在烧的重启后再提醒一次是刻意保留的)。ZCBOT_TOOLFAIL_SCAN_INTERVAL
|
||||
秒,默 86400,<=0 整体关掉。
|
||||
"""
|
||||
dev_email = os.getenv("ZCBOT_DEVELOPER_EMAIL", "").strip()
|
||||
interval = int(os.getenv("ZCBOT_TOOLFAIL_SCAN_INTERVAL", "").strip() or 86400)
|
||||
if interval <= 0:
|
||||
return None
|
||||
|
||||
async def _toolfail_scanner() -> None:
|
||||
from core.toolfail import format_alert, scan_tool_failures
|
||||
from tools.send_email import send_email_smtp, smtp_configured
|
||||
loop = asyncio.get_running_loop()
|
||||
alerted: set = set()
|
||||
while True:
|
||||
try:
|
||||
clusters = await loop.run_in_executor(None, scan_tool_failures)
|
||||
fresh = [c for c in clusters
|
||||
if c.get("count_24h")
|
||||
and (c["tool"], c["signature"]) not in alerted]
|
||||
if fresh:
|
||||
alerted.update((c["tool"], c["signature"]) for c in fresh)
|
||||
for c in fresh:
|
||||
print(f"[toolfail] {c['tool']}/{c['kind']} x{c['count']} "
|
||||
f"tasks={c['task_count']}: {c['signature'][:80]}")
|
||||
if dev_email and smtp_configured():
|
||||
await loop.run_in_executor(
|
||||
None, send_email_smtp, dev_email,
|
||||
f"[zcbot] 工具失败聚集 {len(fresh)} 类",
|
||||
format_alert(fresh, 7),
|
||||
)
|
||||
else:
|
||||
print("[toolfail] ZCBOT_DEVELOPER_EMAIL/SMTP 未配,仅日志")
|
||||
await asyncio.sleep(interval)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[toolfail] error: {type(e).__name__}: {e}")
|
||||
await asyncio.sleep(interval)
|
||||
|
||||
return asyncio.create_task(_toolfail_scanner(), name="toolfail-scanner")
|
||||
|
||||
|
||||
def init_sandbox(app, cfg: dict) -> Optional[asyncio.Task]:
|
||||
"""Sandbox pool(§7.5):仅当 ZCBOT_SANDBOX_BACKEND=docker 时启用。
|
||||
|
||||
启动钩子:① init_pool(创建 docker network + pool 实例)② shutdown_all 清
|
||||
前驱孤儿(上次进程留下的 zcbot-sandbox-* 容器,内存 _last_active 为空,
|
||||
全清重启)③ 后台 reaper task,每 60s 跑 reap_idle。返回 reaper task;
|
||||
host backend 返 None。init 失败 fail-fast(Stage C 协议:hardening 缺失
|
||||
视为部署未完成,不退化到 host —— 否则误以为有沙盒实则在裸跑)。
|
||||
"""
|
||||
backend = os.getenv("ZCBOT_SANDBOX_BACKEND", "host").lower()
|
||||
if backend != "docker":
|
||||
return None
|
||||
from core.agent_builder import resolve_workspace
|
||||
from core.paths import ROOT
|
||||
from core.sandbox import init_pool
|
||||
from core.sandbox.check import detect_fs_quota
|
||||
user_root_base = resolve_workspace(None, cfg) / "users"
|
||||
# §7.5 #4 fs quota 探测:不阻塞启动(应用层周期扫描已有),仅打 WARN
|
||||
# 提醒外部用户开放前必须升级到 xfs prjquota / ext4 project / zfs。
|
||||
try:
|
||||
level, msg = detect_fs_quota(user_root_base.resolve())
|
||||
print(f"[startup] {'[ok]' if level == 'ok' else '[warn]'} {msg}")
|
||||
except Exception as e:
|
||||
print(f"[startup] [warn] fs quota detect failed: {type(e).__name__}: {e}")
|
||||
try:
|
||||
# repo_root=ROOT 让 SandboxPool 把 <repo>/skills 只读 mount 进容器
|
||||
# (fs 工具进容器后 read SKILL references 需要)
|
||||
# sandbox_cfg=yaml `sandbox` 段(memory/cpus/pids_limit 可调)
|
||||
pool = init_pool(
|
||||
user_root_base, repo_root=ROOT,
|
||||
sandbox_cfg=cfg.get("sandbox") or {},
|
||||
)
|
||||
removed = pool.shutdown_all()
|
||||
if removed:
|
||||
print(f"[startup] swept {len(removed)} stale sandbox container(s)")
|
||||
|
||||
async def _reaper() -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(60)
|
||||
removed = await loop.run_in_executor(None, pool.reap_idle)
|
||||
if removed:
|
||||
print(f"[reaper] reaped {len(removed)} idle sandbox container(s)")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[reaper] error: {type(e).__name__}: {e}")
|
||||
|
||||
app.state.sandbox_pool = pool
|
||||
return asyncio.create_task(_reaper(), name="sandbox-reaper")
|
||||
except Exception as e:
|
||||
raise RuntimeError(
|
||||
f"ZCBOT_SANDBOX_BACKEND=docker but sandbox init failed: {e}"
|
||||
)
|
||||
|
||||
|
||||
def start_proc_sweeper(cfg: dict) -> asyncio.Task:
|
||||
"""bg proc 清扫(DESIGN §8.12):终态 proc 目录过 TTL 删除 + 已结束的
|
||||
zcbot-proc-* 容器回收。host / docker 两种 backend 都要跑(host 模式只做
|
||||
文件清扫,docker CLI 不在时 sweep 内部静默跳过容器部分)。每小时一次;
|
||||
幂等,蓝绿双实例同时跑无害。启动后先跑一轮,把上个进程周期留下的
|
||||
已结束容器/过期目录收掉。
|
||||
"""
|
||||
from core.agent_builder import resolve_workspace
|
||||
from core.procs import sweep as _procs_sweep
|
||||
users_base = resolve_workspace(None, cfg) / "users"
|
||||
|
||||
async def _proc_sweeper() -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
while True:
|
||||
try:
|
||||
stats = await loop.run_in_executor(
|
||||
None, _procs_sweep, users_base
|
||||
)
|
||||
if stats["removed_dirs"] or stats["reaped_containers"]:
|
||||
print(f"[proc-sweep] dirs={stats['removed_dirs']} "
|
||||
f"containers={stats['reaped_containers']}")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[proc-sweep] error: {type(e).__name__}: {e}")
|
||||
await asyncio.sleep(3600)
|
||||
|
||||
return asyncio.create_task(_proc_sweeper(), name="proc-sweeper")
|
||||
|
||||
|
||||
async def drain_inflight(app, drain_timeout: int, cancel_grace: int) -> None:
|
||||
"""优雅 drain:先拒新 run(draining 已置位),等在跑的 run 收尾,超时转协作式 cancel。
|
||||
|
||||
单实例形态下消除"restart 误杀 in-flight run 标 error"。新 POST /messages
|
||||
期间返 503(客户端退避重试覆盖)。drain_timeout 内自然跑完 → idle 零 error;
|
||||
超时的 broker.request_cancel → 下个 chunk 间隙退(标 idle);cancel_grace 后仍
|
||||
没退的留给 systemd SIGKILL,下次启动 reaper 标 error(最坏退化 = 改前行为)。
|
||||
★ systemd TimeoutStopSec 必须 > drain_timeout + cancel_grace + 余量(见 RUN.md)。
|
||||
"""
|
||||
inflight = app.state.inflight
|
||||
if not inflight:
|
||||
return
|
||||
print(f"[shutdown] draining {len(inflight)} in-flight run(s), "
|
||||
f"timeout={drain_timeout}s")
|
||||
_, pending = await asyncio.wait(
|
||||
list(inflight.keys()), timeout=drain_timeout
|
||||
)
|
||||
if pending:
|
||||
print(f"[shutdown] {len(pending)} run(s) over drain timeout; "
|
||||
f"signalling cooperative cancel")
|
||||
for t in pending:
|
||||
cid = inflight.get(t)
|
||||
if cid is not None:
|
||||
broker.request_cancel(cid)
|
||||
_, still = await asyncio.wait(pending, timeout=cancel_grace)
|
||||
if still:
|
||||
print(f"[shutdown] {len(still)} run(s) still active after "
|
||||
f"cancel grace; SIGKILL takes over, next start reaps them")
|
||||
|
|
@ -0,0 +1,193 @@
|
|||
"""定时任务守护循环 + 单 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
|
||||
|
|
@ -0,0 +1,117 @@
|
|||
"""微信(ClawBot)入站长轮询启动器(§8.7;从 app.py lifespan 析出,2026-07-23)。
|
||||
|
||||
每个 active 绑定一条 getupdates 长轮询;收到消息 → 跑用户常驻「微信」task →
|
||||
回复发回(渠道无关核心在 web/runs.py::run_channel_conversation)。
|
||||
蓝绿双实例用 PG advisory lock 选主,单实例恒拿锁行为不变。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
from uuid import UUID
|
||||
|
||||
from .runs import run_channel_conversation
|
||||
|
||||
|
||||
def start_wechat_inbound(app) -> tuple[Optional[asyncio.Task], asyncio.Event]:
|
||||
"""ClawBot 入站启用时起 leader 选举 task;未启用返 (None, stop_event)。
|
||||
|
||||
stop_event:关停时先 set 再 cancel task(leader 循环与 keepalive 都在等它)。
|
||||
"""
|
||||
from core.wechat.service import clawbot_enabled
|
||||
|
||||
stop = asyncio.Event()
|
||||
if not clawbot_enabled():
|
||||
return None, stop
|
||||
|
||||
from core.wechat.inbound import run_inbound_manager
|
||||
|
||||
async def _run_wechat_message(uid: UUID, text: str, attachments=None) -> str:
|
||||
"""微信(ClawBot)入站一条消息 → 跑用户常驻「微信」task → 取回复。
|
||||
|
||||
attachments:已下载解密的入站附件(core.wechat.ilink.InboundAttachment,att.data 已回填)。
|
||||
建/复用 task、落盘附件、抢 run 锁、跑 agent 全在渠道无关核心
|
||||
`run_channel_conversation` 里(企业微信回调走同一核心,channel='wecom')。
|
||||
"""
|
||||
return await run_channel_conversation(app, uid, text, attachments, channel="wechat")
|
||||
|
||||
async def _wechat_leader() -> None:
|
||||
"""ClawBot 入站长轮询单实例互斥(蓝绿 B 档,RUN.md)。
|
||||
|
||||
getupdates 长轮询是消费型的:蓝绿并存期两实例同时拉会抢消息 / 同一条
|
||||
入站双跑 agent。用 PG advisory lock 选主:持锁实例才跑 inbound manager,
|
||||
另一实例每 15s 重试;持锁连接断(DB 重启 / 网络抖 = 锁已自动释放)→
|
||||
停管理器重新竞争。单实例部署恒拿到锁,行为与之前完全一致。
|
||||
"""
|
||||
from core.storage import get_engine
|
||||
engine = get_engine()
|
||||
# pg_try_advisory_lock(int4, int4) 的固定 key:('zc'=0x7A63, 1=wechat-inbound)
|
||||
lock_sql = "SELECT pg_try_advisory_lock(31331, 1)"
|
||||
|
||||
def _acquire():
|
||||
conn = engine.connect()
|
||||
# 从连接池摘出独占:advisory lock 是 DBAPI 连接级的,不 detach 的话
|
||||
# close() 只是还池,锁挂在池化连接上既不释放也没人持有
|
||||
conn.detach()
|
||||
try:
|
||||
if conn.exec_driver_sql(lock_sql).scalar():
|
||||
return conn
|
||||
conn.close()
|
||||
return None
|
||||
except Exception:
|
||||
conn.close()
|
||||
raise
|
||||
|
||||
async def _wait(seconds: float) -> None:
|
||||
try:
|
||||
await asyncio.wait_for(stop.wait(), timeout=seconds)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
while not stop.is_set():
|
||||
conn = None
|
||||
mgr = None
|
||||
try:
|
||||
conn = await asyncio.to_thread(_acquire)
|
||||
if conn is None:
|
||||
await _wait(15)
|
||||
continue
|
||||
print("[wechat] inbound leader lock acquired, manager starting")
|
||||
mgr = asyncio.create_task(
|
||||
run_inbound_manager(_run_wechat_message, stop),
|
||||
name="wechat-inbound",
|
||||
)
|
||||
# 30s keepalive ping 锁连接;ping 挂 = 连接断 = 锁已丢 → 重选
|
||||
while not stop.is_set() and not mgr.done():
|
||||
await _wait(30)
|
||||
if stop.is_set() or mgr.done():
|
||||
break
|
||||
try:
|
||||
await asyncio.to_thread(
|
||||
lambda: conn.exec_driver_sql("SELECT 1").scalar()
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"[wechat] leader lock lost "
|
||||
f"({type(e).__name__}), re-electing")
|
||||
break
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[wechat] leader loop error: {type(e).__name__}: {e}")
|
||||
await _wait(15)
|
||||
finally:
|
||||
if mgr is not None and not mgr.done():
|
||||
mgr.cancel()
|
||||
try:
|
||||
await mgr
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
if conn is not None:
|
||||
try:
|
||||
conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
task = asyncio.create_task(_wechat_leader(), name="wechat-leader")
|
||||
print("[wechat] ClawBot inbound enabled (PG advisory lock leader election)")
|
||||
return task, stop
|
||||
Loading…
Reference in New Issue