zcbot/web/app.py

"""FastAPI app: 纯 /v1 JSON API(2026-05-15 切换 — 详见 DESIGN §7.9)。

设计要点:
- 所有路由 `/v1/*` 前缀,响应 JSON;模板 / HTMX / 服务端 markdown 渲染全删
- SSE 事件 payload 是 JSON dict 而非 HTML 片段(`event: <type>` + `data: <json>`)
- Auth: PLATFORM_KEY → JWT 兑换(§7 D' 过渡形态,见 web/auth.py);OIDC 替换时只动 /v1/auth/login 内部
- 所有 /v1/tasks* 路由 Depends(require_user),按 user_id 隔离数据
- 豁免:/healthz、/docs、/openapi.json、/、/v1/auth/login、/static/*
- CORS allow_origins=["*"] 本地宽松;真发布按 platform 域名收紧
- `GET /` 302 → /static/dev.html(本地 dev SPA)
"""
from __future__ import annotations

import asyncio
import mimetypes
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,
    TOKEN_EXPIRES_HEADER,
    AuthConfig,
    make_require_admin,
    make_require_user,
)
from .admin import register_admin_routes
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
from .routers.kb import register_kb_routes
from .routers.messages import register_message_routes
from .routers.misc import register_misc_routes
from .routers.models import register_model_routes
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 .static_files import NoCacheStaticFiles

# ────────────────────── App 工厂 ──────────────────────

# web/static 目录路径 — /static 静态挂载用,dev.html 也放这
_STATIC_DIR = Path(__file__).parent / "static"


def create_app() -> FastAPI:
    # fail-fast:env 缺失直接抛,不裸跑无密
    auth_cfg = AuthConfig.from_env()
    require_user = make_require_user(auth_cfg)
    require_admin = make_require_admin(auth_cfg)

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        loop = asyncio.get_running_loop()
        broker.bind_loop(loop)

        # ── 接管默认线程池 executor(§8.4)──────────────────────────────
        # run 走 asyncio.to_thread(用 loop 默认 executor);默认是匿名的,读不到大小、
        # 不可调。显式建一个同尺寸(复刻 Python 默认 min(32, cpu+4))接管,好处:① 监控
        # 能读 max_workers 判断有没有排队 ② 并发不够时改 ZCBOT_RUN_MAX_WORKERS 调大不改码。
        # 注:run 与 disk scan / pptx 转换 / reaper 共享此池(同原默认行为);真要隔离
        # 长任务再另开 run 专用池,那是后话。
        run_max_workers = int(
            os.getenv("ZCBOT_RUN_MAX_WORKERS") or min(32, (os.cpu_count() or 1) + 4)
        )
        run_executor = ThreadPoolExecutor(
            max_workers=run_max_workers, thread_name_prefix="run"
        )
        loop.set_default_executor(run_executor)
        app.state.run_executor = run_executor
        app.state.run_max_workers = run_max_workers
        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
        _cfg = load_config()

        # 优雅 drain 状态(SIGTERM / systemctl restart 兜底,见下方 finally):
        # draining 置位后 POST /messages 返 503;inflight 登记在跑的 BG run task,
        # 关停时 await 它们收尾。inflight 同时给 create_task 持强引用,防被 GC 中途回收。
        app.state.draining = asyncio.Event()
        app.state.inflight = {}  # dict[asyncio.Task, UUID(task_id)]
        _shutdown_cfg = _cfg.get("shutdown") or {}
        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)")

        # 磁盘配额后台扫描(§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)。
            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")

            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
            if wechat_task is not None:
                wechat_stop.set()
                wechat_task.cancel()
                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}")

            # broker 收尾(redis 版停 pubsub reader + 断连;local no-op)。放在 drain
            # 之后 —— drain 期间 run 还要 emit/close。
            try:
                await broker.shutdown()
            except Exception as e:
                print(f"[shutdown] broker shutdown error: {type(e).__name__}: {e}")

            # drain 已 await inflight 收尾、run 线程退完;非阻塞关池(进程在退出,保守清理)
            run_executor.shutdown(wait=False)

    app = FastAPI(
        title="zcbot api",
        version=__version__,
        description=(
            "zcbot 后端 — /v1 JSON API + SSE。Auth: PLATFORM_KEY → JWT(§7 D' 过渡)。"
            "本地 dev SPA: /static/dev.html。"
        ),
        lifespan=lifespan,
    )
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],         # 本地宽松,部署 platform 时按域名收紧
        allow_credentials=False,
        allow_methods=["*"],
        allow_headers=["*"],
        # 滑动续签的新 token 走响应头带回;expose 才能让浏览器 fetch 读到(默认不暴露自定义头)
        expose_headers=[REFRESHED_TOKEN_HEADER, TOKEN_EXPIRES_HEADER],
    )

    if _STATIC_DIR.is_dir():
        # Windows 上 mimetypes 偶尔把 .js 判成 text/plain,会令 <script type="module"> 被浏览器拒执行;
        # 显式兜底,保证静态 ES module 以正确 MIME 下发。
        mimetypes.add_type("text/javascript", ".js")
        app.mount("/static", NoCacheStaticFiles(directory=str(_STATIC_DIR)), name="static")

    # ───────────── 拆分出的路由模块(与 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)
    register_auth_routes(app, require_user=require_user, auth_cfg=auth_cfg)
    register_skill_memory_routes(app, require_user=require_user)
    register_kb_routes(app, require_user=require_user)
    register_schedule_routes(app, require_user=require_user)
    register_file_routes(app, require_user=require_user)
    register_asr_routes(app, require_user=require_user, auth_cfg=auth_cfg)
    register_task_routes(app, require_user=require_user)
    register_message_routes(app, require_user=require_user)

    # ───────────── 管理后台(admin-only)─────────────
    register_admin_routes(app, require_admin)

    return app