diff --git a/web/app.py b/web/app.py index bd23619..075c40c 100644 --- a/web/app.py +++ b/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 把 /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) diff --git a/web/background.py b/web/background.py new file mode 100644 index 0000000..7798993 --- /dev/null +++ b/web/background.py @@ -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 把 /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") diff --git a/web/scheduler_runner.py b/web/scheduler_runner.py new file mode 100644 index 0000000..93216fa --- /dev/null +++ b/web/scheduler_runner.py @@ -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 diff --git a/web/wechat_runner.py b/web/wechat_runner.py new file mode 100644 index 0000000..3c50cde --- /dev/null +++ b/web/wechat_runner.py @@ -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