Blame for server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
server.py — 网页通道 → OpenAI 兼容中转站 (v4: 工厂页三模式 + 请求级重试)

v4 变更(经 hf_reco/poc_mock_factory.py 三臂对照实验验证):
  token 工厂页不再必须真实加载目标站。签发只依赖三件事:
    ① 目标站 origin(Turnstile sitekey 域名校验)  ② 真实浏览器引擎
    ③ challenges.cloudflare.com 可达
  "页面"只是容器, 因此提供三种承载方式(FACTORY_MODE, 面板可热切):
    intercept(默认/推荐): Playwright route 拦截, 空壳页纯内存承载,
      零基础设施(无证书/无端口/无 DNS 改动), 签发与目标站页面完全解耦
    mock: --host-resolver-rules 域名映射 + 本地 HTTPS 空壳页(自签证书),
      不经 CDP 拦截层(intercept 被针对时的 Plan B)
    real: 真实加载工厂页(旧行为, sitekey 轮换等场景的保底)

设计原则(2026-09-27 定稿):
  中继不判断、不记忆、不替上游做主 —— 所有机制只朝"让这个请求成功"使劲:
    - 遇到上游错误就换新 token 重试(状态码集合与次数可配, 默认 5 次)
    - token 货架预热(供给的"缓存"), 请求零等待
    - 模型列表 = 原样拉上游, 不探测不筛选
  没有任何基于历史记录的拦截(探测/否定缓存已移除), 失败都是上游当下的真实回答。

特性:
  - Token 预热货架: 常备现货, 请求零等待签发
  - 多工人页并行签发, 运行时可扩缩容 (POST /api/pool/scale)
  - 全格式透传: /v1/chat/completions(流式/非流式边缘聚合, 含 tool_calls) + /v1/completions + /v1/embeddings
  - 请求级重试: 上游错误换新 token 重试(max_retry / retry_statuses 可热改), 仍失败则错误字节级透传
  - /v1/models: 直接拉上游全量列表(60s 微缓存, 只为快不为筛选)
  - Web 管理面板: 实时池状态/签发统计/工人页明细/最近请求/测试对话

依赖: pip install fastapi uvicorn httpx playwright cryptography
运行: uvicorn server:app --host 0.0.0.0 --port 8377
面板: http://127.0.0.1:8377/  (用任一 relay key 登录)
"""
import asyncio
import collections
import json
import math
import ssl
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import (FileResponse, HTMLResponse, JSONResponse,
                               Response, StreamingResponse)
from playwright.async_api import async_playwright

from config import cfg                         # 配置统一来自环境变量

# ----------------------------- 配置(来自 config.py / 环境变量) -----------------------------
UPSTREAM         = cfg.UPSTREAM                # 静态: 换上游需重启
SITEKEY          = cfg.SITEKEY                  # 静态: Cloudflare sitekey
DEFAULT_FACTORY_PAGE = cfg.FACTORY_PAGE         # 工厂页初值(运行时可在面板热切换)
UA               = cfg.UA

DEFAULT_WORKER_PAGES     = cfg.WORKER_PAGES                 # 动态初值(面板可热改)
DEFAULT_MAX_WORKER_PAGES = cfg.MAX_WORKER_PAGES
DEFAULT_SHELF_SIZE       = cfg.SHELF_SIZE
DEFAULT_TOKEN_MAX_AGE    = cfg.TOKEN_MAX_AGE
DEFAULT_IDLE_PAUSE       = cfg.IDLE_PAUSE
DEFAULT_MAX_RETRY        = cfg.MAX_RETRY                    # 上游错误重试次数(0=不重试)
DEFAULT_RETRY_STATUSES   = cfg.RETRY_STATUSES               # 触发重试的上游状态码集合
# /v1/models 微缓存(只为快, 不为筛选)
MODELS_CACHE_TTL = 60
_models_cache: dict = {}     # {"key": path, "ts": monotonic, "status": int, "body": bytes}


def _load_saturation() -> float:
    """标准自报块: 饱和度(0..1) = 忙工人页/总工人页。渠道自定义语义, 平台只看数。"""
    fs = list(pool.factories.values())
    if not fs:
        return 0.0
    return round(sum(1 for f in fs if f.busy) / len(fs), 3)


async def _stats_models(timeout: float = 8.0) -> list:
    """标准自报块: 模型目录(hub 聚合 /v1/models 的数据源)。
    有缓存直接给(容忍过期); 无缓存现拉一次(失败给空, 路由不受影响 —— deepinfra 是默认目录)。"""
    if not _models_cache.get("entry"):
        try:
            r = await upstream_client.get(f"{UPSTREAM}/v1/openai/models",
                                          timeout=timeout)
            _models_cache["entry"] = {"key": "/v1/openai/models",
                                      "ts": time.monotonic(),
                                      "status": r.status_code,
                                      "body": r.content}
        except Exception:
            return []
    try:
        entry = _models_cache.get("entry") or {}
        data = json.loads(entry.get("body", b"{}")).get("data", [])
        return [m.get("id") for m in data if m.get("id")]
    except Exception:
        return []

SIGN_FAIL_RESTART = 5        # 工人页连续签发失败 N 次则回收, 看门狗负责补齐
ADAPTIVE_WINDOW   = 300      # 自适应货架的需求统计窗口(秒)
RATE_LIMIT_RPM    = cfg.RATE_LIMIT_RPM   # 每 key 每分钟限次(0=关)

GET_TOKEN_JS = """
async (sitekey) => {
  if (!window.turnstile) throw new Error('window.turnstile 未加载');
  const el = document.createElement('div');
  el.style.display = 'none';
  document.body.appendChild(el);
  return await new Promise((resolve, reject) => {
    const wid = window.turnstile.render(el, {
      sitekey,
      callback: (t) => { window.turnstile.remove(wid); el.remove(); resolve(t); },
      'error-callback': (e) => reject(new Error('turnstile error: ' + e)),
      'expired-callback': () => reject(new Error('expired')),
    });
  });
}
"""

BASE_HEADERS = {
    "User-Agent": UA,
    "Origin": "https://deepinfra.com",
    "Referer": "https://deepinfra.com/",
    "X-Deepinfra-Source": "model-embed",
    "Accept": "text/event-stream",
    "Content-Type": "application/json",
}

# ------------------------- 空壳工厂页(intercept/mock 模式) -------------------------
FACTORY_MODES = ("intercept", "mock", "real")

SHELL_HTML = """<!DOCTYPE html>
<html lang="en"><head><meta charset="utf-8">
<title>DeepInfra</title>
<script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>
</head>
<body>
<div style="font:14px/1.6 sans-serif;padding:24px;color:#333">
<h1 style="font-size:16px">DeepInfra</h1>
<p>Chat with AI models.</p>
<textarea aria-label="chatbot input prompt" rows="1" style="width:320px"></textarea>
</div>
</body></html>"""


def shell_page_url(factory_page: str) -> str:
    """空壳页导航 URL: 保持目标站 origin(路径随意, 仅供人看)。"""
    host = urlparse(factory_page).netloc or "deepinfra.com"
    return f"https://{host}/__shell__"


# ---- mock 模式: 本地 HTTPS 空壳页服务(自签证书, 容器内 loopback) ----
_mock_server_started = False


def _ensure_mock_cert() -> tuple[str, str]:
    """自签一张 <host> 证书, 缓存在 PROFILE_DIR 旁; 幂等。"""
    cert_dir = Path(cfg.PROFILE_DIR).parent / "relay-mock-cert"
    cert_pem, key_pem = cert_dir / "cert.pem", cert_dir / "key.pem"
    if cert_pem.exists() and key_pem.exists():
        return str(cert_pem), str(key_pem)
    cert_dir.mkdir(parents=True, exist_ok=True)
    host = urlparse(DEFAULT_FACTORY_PAGE).netloc or "deepinfra.com"
    from cryptography import x509
    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import rsa
    from cryptography.x509.oid import NameOID
    import datetime
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, host)])
    now = datetime.datetime.now(datetime.timezone.utc)
    cert = (x509.CertificateBuilder()
            .subject_name(name).issuer_name(name)
            .public_key(key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(now - datetime.timedelta(days=1))
            .not_valid_after(now + datetime.timedelta(days=3650))
            .add_extension(x509.SubjectAlternativeName(
                [x509.DNSName(host)]), critical=False)
            .sign(key, hashes.SHA256()))
    cert_pem.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
    key_pem.write_bytes(key.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.TraditionalOpenSSL,
        serialization.NoEncryption()))
    return str(cert_pem), str(key_pem)


def ensure_mock_server():
    """启动 mock 模式的本地 HTTPS 服务(幂等, 线程化)。"""
    global _mock_server_started
    if _mock_server_started:
        return
    cert_pem, key_pem = _ensure_mock_cert()

    class _Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            body = SHELL_HTML.encode("utf-8")
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def log_message(self, *a):
            pass

    httpd = ThreadingHTTPServer(("127.0.0.1", cfg.MOCK_PORT), _Handler)
    sctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    sctx.load_cert_chain(cert_pem, key_pem)
    httpd.socket = sctx.wrap_socket(httpd.socket, server_side=True)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    _mock_server_started = True
    log(f"[mock] local HTTPS shell page on 127.0.0.1:{cfg.MOCK_PORT}")

_started_at = time.time()


def log(*a):
    print(time.strftime("[%H:%M:%S]"), *a, flush=True)


# ------------------------- 工人页(签发器) -------------------------
class TokenFactory:
    """一个实例 = 一个常驻浏览器标签页 (同一工厂模型页)。
    身份只有签发并发意义; 签出的 token 对全部可路由模型通用。"""
    def __init__(self, wid: int):
        self.wid = wid
        self.page = None
        self.busy = False          # 正在签发中
        self.retired = False       # 缩容标记: 不再派活, 闲下来即关闭
        self.consecutive_fails = 0 # 连续签发失败计数(健康度; 成功即清零)
        self.stats = {"signed": 0, "sign_failed": 0, "last_sign_latency": None}
        self.latencies = collections.deque(maxlen=20)   # 最近签发耗时(秒)

    async def start(self, ctx, page_url: str):
        """按 pool.factory_mode 建页:
        - real:      goto 真实工厂页(page_url), 等 turnstile 懒加载(带兜底)
        - intercept: route 拦截目标站导航, 空壳页纯内存承载(零基础设施)
        - mock:      goto 空壳页 URL(DNS 由启动参数映射到本地 HTTPS 服务)
        """
        if self.page:
            try:
                await self.page.close()
            except Exception:
                pass
        self.page = await ctx.new_page()
        mode = pool.factory_mode
        if mode in ("intercept", "mock"):
            shell_url = shell_page_url(pool.factory_page)
            if mode == "intercept":
                async def _fulfill(route):
                    await route.fulfill(
                        status=200, content_type="text/html; charset=utf-8",
                        body=SHELL_HTML)
                host = urlparse(pool.factory_page).netloc or "deepinfra.com"
                await self.page.route(f"https://{host}/**", _fulfill)
            else:
                ensure_mock_server()
            await self.page.goto(shell_url, wait_until="domcontentloaded",
                                 timeout=60000)
            # 冷 profile 首次加载 api.js 有 302 跳转链, 可能超 15s;
            # warm 后(同 profile 后续工人页)亚秒就绪。60s 是安全上限。
            await self.page.wait_for_function("() => !!window.turnstile",
                                               timeout=60000)
        else:
            await self.page.goto(page_url, wait_until="domcontentloaded",
                                 timeout=180000)
            try:
                await self.page.wait_for_function("() => !!window.turnstile",
                                                  timeout=15000)
            except Exception:
                try:
                    box = self.page.locator(
                        'textarea[aria-label="chatbot input prompt"]')
                    await box.wait_for(state="visible", timeout=20000)
                    await box.click()
                except Exception:
                    pass
                await self.page.wait_for_function("() => !!window.turnstile",
                                                  timeout=60000)
        await self.page.wait_for_timeout(2500)
        log(f"[worker {self.wid}] ready (mode={mode})")

    async def sign(self):
        """在本页执行一次 Turnstile 验证并返回 token; 失败返回 None。"""
        self.busy = True
        t0 = time.monotonic()
        try:
            token = await self.page.evaluate(GET_TOKEN_JS, SITEKEY)
            dt = round(time.monotonic() - t0, 2)
            self.stats["signed"] += 1
            self.stats["last_sign_latency"] = dt
            self.latencies.append(dt)
            self.consecutive_fails = 0
            return token
        except Exception as e:
            self.stats["sign_failed"] += 1
            self.consecutive_fails += 1
            log(f"[worker {self.wid}] sign failed: {str(e).splitlines()[0]}")
            return None
        finally:
            self.busy = False

    async def stop(self):
        self.retired = True
        if self.page:
            try:
                await self.page.close()
            except Exception:
                pass
            self.page = None

    @property
    def ready(self):
        return self.page is not None and not self.retired


# --------------------------- 共享 Token 池 ---------------------------
class TokenPool:
    def __init__(self):
        self.factories: dict[int, TokenFactory] = {}
        self._next_wid = 0
        # 运行时可热改的参数(从 config 拷贝; setter 会唤醒补货循环立即生效)
        self.relay_keys: set[str] = set(cfg.RELAY_KEYS)   # 来自 RELAY_KEYS 环境变量
        self.worker_pages = DEFAULT_WORKER_PAGES
        self.max_worker_pages = DEFAULT_MAX_WORKER_PAGES
        self.shelf_size = DEFAULT_SHELF_SIZE
        self.token_max_age = DEFAULT_TOKEN_MAX_AGE
        self.idle_pause = DEFAULT_IDLE_PAUSE
        self.max_retry = DEFAULT_MAX_RETRY              # 上游错误重试次数(0=不重试)
        self.retry_statuses: set[int] = set()
        self.set_retry_statuses(DEFAULT_RETRY_STATUSES)   # 字符串 "403,429,..." → int 集合
        self.desired_pages = self.worker_pages
        self.factory_page: str = DEFAULT_FACTORY_PAGE   # 工厂页(面板可热切换)
        self.factory_mode: str = (cfg.FACTORY_MODE if cfg.FACTORY_MODE
                                  in FACTORY_MODES else "intercept")
        self.shelf = collections.deque()        # [(token, issued_at)] 全局共享
        self.cond = asyncio.Condition()
        self.inflight = 0                       # 进行中的签发数
        self.last_use = time.monotonic()
        self.ready = False
        self.channel = None                     # 实际使用的浏览器通道
        self._stopping = False
        self._scale_lock = asyncio.Lock()       # 防 keeper 与 /api/pool/scale 并发扩缩容
        self._keeper_task = None                # 工人页看门狗任务
        # 可靠性/配额治理
        self.adaptive_shelf = cfg.ADAPTIVE_SHELF
        self.browser_max_age = cfg.BROWSER_MAX_AGE
        self.issue_timeout = cfg.ISSUE_TIMEOUT
        self.last_error: str | None = None      # 最近一次池级故障摘要(状态页可见)
        self.served_ts = collections.deque()    # 最近取货时刻(自适应货架需求统计)
        self._browser_started: float | None = None
        self._recycling = False
        # 运行时观测
        self.served = 0
        self.surplus_dropped = 0
        self.expired_dropped = 0
        self.requests = collections.deque(maxlen=50)
        # token 用量计数(2026-10-02 对账体系): 空间是唯一寿命 ≥ 流的观测者,
        # 这里累计的总量是 hub 侧 usage 抓取(长流会被 Workers 掐死)的完整兜底。
        # 重启清零, hub 侧用快照差值(delta)取窗口量, 回绕按 0 处理(与 signed_total 同法)。
        self.usage_in = 0      # prompt_tokens 累计
        self.usage_out = 0     # completion_tokens 累计

    def add_usage(self, usage: dict | None):
        """把一笔请求的 usage 记入累计器(缺字段/异常静默, 绝不影响转发)。"""
        try:
            if usage:
                self.usage_in += int(usage.get("prompt_tokens") or 0)
                self.usage_out += int(usage.get("completion_tokens") or 0)
        except Exception:
            pass

    # ---------- 运行时热改参数(setter 即时生效) ----------
    def _wake(self):
        """唤醒补货循环。注意两个坑: asyncio.create_task 只收 coroutine;
        Condition.notify_all 必须持锁调用。旧实现 create_task(self.cond.notify_all())
        原地抛 TypeError 且 except 只接 RuntimeError —— 面板热改
        token_max_age/idle_pause 会 500, 已修复为包一层协程。"""
        async def _notify():
            async with self.cond:
                self.cond.notify_all()
        try:
            asyncio.create_task(_notify())
        except RuntimeError:
            pass   # 无事件循环时(初始化期)忽略

    def set_shelf_size(self, n: int):
        self.shelf_size = max(0, int(n))
        # 收紧时立即丢弃超出部分(避免慢慢等过期), 算入 surplus_dropped
        async def _trim():
            async with self.cond:
                while len(self.shelf) > self.shelf_size:
                    self.shelf.popleft()
                    self.surplus_dropped += 1
                self.cond.notify_all()
        try:
            asyncio.create_task(_trim())
        except RuntimeError:
            pass

    def set_token_max_age(self, n: int):
        self.token_max_age = max(1, int(n))
        self._wake()

    def set_idle_pause(self, n: int):
        self.idle_pause = max(0, int(n))
        self.last_use = time.monotonic()      # 改静默参数视同"有活动", 立即解静默
        self._wake()

    def set_max_retry(self, n: int):
        """重试次数: 任意非负整数, 不设上限(0=不重试)。"""
        self.max_retry = max(0, int(n))

    def set_retry_statuses(self, value):
        """重试状态码集合: 逗号分隔字符串或列表, 如 "403,429,502"。空/非法项忽略。"""
        if isinstance(value, str):
            parts = value.replace(" ", "").split(",")
        elif isinstance(value, (list, tuple)):
            parts = [str(v) for v in value]
        else:
            return False
        codes = set()
        for p in parts:
            try:
                codes.add(int(p))
            except ValueError:
                pass
        if not codes:
            return False
        self.retry_statuses = codes
        return True

    def set_max_worker_pages(self, n: int):
        self.max_worker_pages = max(1, int(n))
        # 若新的硬上限低于当前期望, 顺势下调
        if self.desired_pages > self.max_worker_pages:
            asyncio.create_task(self.scale(self.max_worker_pages))

    # ---------- 生命周期 ----------
    async def _launch_browser(self):
        """启动浏览器上下文。容器/本地自适应:
        - 容器(HF_SPACE / 无显示器): headless 由 cfg.HEADLESS 决定; Xvfb 在 Dockerfile
          外层提供虚拟显示, 这里仅加 --no-sandbox 等容器化参数。mdedge 通道在容器里
          不会有, 直接试 bundled chromium, 失败再退一次。
        - 本机(有显示器): headless=False 保指纹, 复用 POC 验证过的 msedge→chromium 降级。
        """
        import shutil
        last_err = None

        # 容器化参数: 非 root 跑必须 --no-sandbox; /dev/shm 太小会崩, 改用 /tmp。
        extra_args = ["--disable-blink-features=AutomationControlled"]
        if cfg.HF_SPACE:
            extra_args += ["--no-sandbox",
                           "--disable-dev-shm-usage",
                           "--disable-gpu"]
        if cfg.PROXY:
            extra_args.append(f"--proxy-server={cfg.PROXY}")
        if self.factory_mode == "mock":
            # DNS 映射: 目标站 -> 本地空壳页服务(真实站流量从浏览器消失;
            # httpx 走的 api 子域不受影响)
            extra_args.append(
                f"--host-resolver-rules=MAP "
                f"{urlparse(self.factory_page).netloc or 'deepinfra.com'} "
                f"127.0.0.1:{cfg.MOCK_PORT}")

        # 本机可走 msedge 降级; 容器里没装 edge, 只试 bundled chromium。
        channels = ("msedge", None) if not cfg.HF_SPACE else (None,)
        profile = cfg.PROFILE_DIR

        for channel in channels:
            for attempt in range(2):
                try:
                    kwargs = dict(headless=cfg.HEADLESS,
                                  args=extra_args,
                                  viewport={"width": 1280, "height": 900})
                    if self.factory_mode == "mock":
                        kwargs["ignore_https_errors"] = True  # 本地自签证书
                    if channel:
                        kwargs["channel"] = channel
                    self.ctx = await self._pw.chromium.launch_persistent_context(
                        profile, **kwargs)
                    self.channel = channel or "bundled"
                    self._browser_started = time.monotonic()
                    log(f"[pool] browser launched "
                        f"(channel={self.channel} headless={cfg.HEADLESS} "
                        f"hf={cfg.HF_SPACE})")
                    return
                except Exception as e:
                    last_err = e
                    log(f"[pool] launch failed (channel={channel or 'bundled'} "
                        f"attempt {attempt+1}): {str(e).splitlines()[0]}")
                    await asyncio.sleep(4)
            shutil.rmtree(profile, ignore_errors=True)
        raise RuntimeError(f"browser launch failed, last error: {last_err}")

    async def supervisor(self):
        """守护循环: 浏览器上线 → 拉起工人页 + 补货协程 + 看门狗。"""
        while not self._stopping:
            self.ready = False
            try:
                self._pw = await async_playwright().start()
                await self._launch_browser()
                self._refiller = asyncio.create_task(self._refill_loop())
                await self._apply_scale()        # 建工人页(可后台继续, 就绪不依赖)
                self.ready = True
                if self.factories:
                    self.last_error = None       # 本轮至少有工人页建成, 清除旧故障标记
                self._start_keeper()             # 看门狗: 工人页失败/丢失后自动补齐
                log(f"[pool] READY ({len(self.factories)} worker pages)")
                return
            except Exception as e:
                self.last_error = f"supervisor cycle failed: {str(e).splitlines()[0]}"
                log(f"[pool] {self.last_error}")
            await self._teardown()
            if not self._stopping:
                await asyncio.sleep(15)

    async def _teardown(self):
        for f in self.factories.values():
            try:
                await f.stop()
            except Exception:
                pass
        self.factories.clear()
        if getattr(self, "ctx", None):
            try:
                await self.ctx.close()
            except Exception:
                pass
        self.ctx = None
        if getattr(self, "_pw", None):
            try:
                await self._pw.stop()
            except Exception:
                pass
        self._pw = None

    async def stop(self):
        self._stopping = True
        await self._teardown()

    # ---------- 工人页扩缩容 ----------
    async def _apply_scale(self):
        """把工人页数量调整到 desired_pages。(加锁: 看门狗/API 触发的扩缩容可能并发)"""
        async with self._scale_lock:
            n = self.desired_pages
            # 需要新增
            while len(self.factories) < n:
                self._next_wid += 1
                f = TokenFactory(self._next_wid)
                self.factories[f.wid] = f
                try:
                    await f.start(self.ctx, self.factory_page)
                except Exception as e:
                    self.last_error = (f"worker {f.wid} start failed: "
                                       f"{str(e).splitlines()[0]}")
                    log(f"[pool] {self.last_error}")
                    del self.factories[f.wid]
                    break                # 本次放弃, 由看门狗稍后自动重试
                async with self.cond:
                    self.cond.notify_all()
            # 需要缩减: 优先退役闲置页; 忙页打标记, 由补货循环收尾
            actives = [f for f in self.factories.values() if not f.retired]
            for f in actives[:max(0, len(actives) - n)]:
                if not f.busy:
                    await f.stop()
                    self.factories.pop(f.wid, None)
                    log(f"[pool] worker {f.wid} retired (idle)")
                else:
                    f.retired = True
                    log(f"[pool] worker {f.wid} will retire after current sign")

    def _start_keeper(self):
        """启动(或复用)工人页看门狗任务。"""
        if self._keeper_task and not self._keeper_task.done():
            return
        self._keeper_task = asyncio.create_task(self._keeper_loop())

    async def _keeper_loop(self):
        """看门狗: ①存活工人页少于期望时周期性重试扩容; ②浏览器超龄时趁空闲回收。

        修复的缺陷: v2 里 _apply_scale 首个工人页启动失败即 break,
        supervisor 随即置 ready=True 返回——此后没有任何代码再尝试建页,
        池子进入"ready 但 0 工人"的僵尸态, 所有请求等 token 60s 后 503。
        (冷启动时工厂页偶发一次加载失败即可触发, 只能重启或手动 scale 解除。)
        """
        while not self._stopping:
            await asyncio.sleep(30)
            if self._stopping:
                continue
            if not getattr(self, "ctx", None):
                # 浏览器丢了(如回收失败): 直接重建浏览器 + 工人页, 不能干等
                if self.ready and not self._recycling:
                    log("[keeper] browser context missing, relaunching")
                    asyncio.create_task(self._recycle_browser())
                continue
            # 浏览器定期回收: 只在无在签/无忙页的空闲窗口执行
            if (self._browser_started and not self._recycling
                    and self.inflight == 0
                    and all(not f.busy for f in self.factories.values())
                    and time.monotonic() - self._browser_started
                        > self.browser_max_age):
                asyncio.create_task(self._recycle_browser())
                continue
            alive = sum(1 for f in self.factories.values() if not f.retired)
            if alive < self.desired_pages:
                log(f"[keeper] workers {alive}/{self.desired_pages}, "
                    f"retrying scale-up")
                try:
                    await self._apply_scale()
                    alive = sum(1 for f in self.factories.values()
                                if not f.retired)
                    if alive >= self.desired_pages:
                        self.last_error = None    # 补齐成功, 清除故障标记
                except Exception as e:
                    log(f"[keeper] rescale failed: {str(e).splitlines()[0]}")

    async def _recycle_browser(self):
        """定期回收: 长跑 Chromium 会累积内存和 profile 膨胀。
        货架上的 token 是纯字符串, 回收浏览器不清现货, 请求面基本无感。"""
        self._recycling = True
        try:
            age = int(time.monotonic() - (self._browser_started or 0))
            log(f"[pool] periodic browser recycle (age={age}s)")
            await self._teardown()
            self._pw = await async_playwright().start()
            await self._launch_browser()
            await self._apply_scale()
            if self.factories:
                self.last_error = None
            log(f"[pool] browser recycled "
                f"({len(self.factories)} worker pages, shelf kept "
                f"{len(self.shelf)})")
        except Exception as e:
            self.last_error = f"browser recycle failed: {str(e).splitlines()[0]}"
            log(f"[pool] {self.last_error}")
        finally:
            self._recycling = False

    async def scale(self, pages: int) -> dict:
        self.desired_pages = max(1, min(self.max_worker_pages, int(pages)))
        if getattr(self, "ctx", None):
            asyncio.create_task(self._apply_scale())
        return {"desired": self.desired_pages,
                "active": sum(1 for f in self.factories.values()
                              if not f.retired)}

    # ---------- 工厂页热切换 ----------
    def _validate_factory_page(self, url) -> str:
        """工厂页 URL 校验。接受三种写法:
        ① 完整 URL https://deepinfra.com/<owner>/<model>
        ② 站内路径 /<owner>/<model>
        ③ 裸模型 id <owner>/<model>"""
        url = str(url or "").strip()
        if not url:
            raise ValueError("empty factory_page")
        if "/" in url and not url.startswith(("http://", "https://", "/")) \
                and " " not in url:
            url = f"/{url}"                       # 裸 owner/model → 路径
        if url.startswith("/"):
            url = f"https://deepinfra.com{url}"
        if not (url.startswith("https://") or url.startswith("http://")):
            raise ValueError("factory_page must be url or /owner/model")
        return url

    async def switch_factory_page(self, url) -> dict:
        """热切换工厂页: 现有工人页全部退役, 立即按新页重建到 desired_pages。
        货架现货 token 不受影响(签发的 token 对所有可路由模型通用)。
        intercept/mock 模式下仅取其 host 作为空壳页 origin, 路径无实际意义。"""
        url = self._validate_factory_page(url)
        if url == self.factory_page:
            return {"factory_page": url, "changed": False,
                    "active": sum(1 for f in self.factories.values()
                                  if not f.retired)}
        self.factory_page = url
        busy_retired = 0
        # 退役全部现有工人页(闲页立即关, 忙页打标记由签发收尾/看门狗处理)
        for f in list(self.factories.values()):
            f.retired = True
            if not f.busy:
                try:
                    await f.stop()
                except Exception:
                    pass
                self.factories.pop(f.wid, None)
            else:
                busy_retired += 1
        if getattr(self, "ctx", None):
            await self._apply_scale()     # 立即按新 factory_page 重建
        log(f"[pool] factory page switched -> {url} "
            f"(rebuilt {sum(1 for f in self.factories.values() if not f.retired)}"
            f"/{self.desired_pages}, {busy_retired} busy retiring)")
        return {"factory_page": url, "changed": True,
                "active": sum(1 for f in self.factories.values()
                              if not f.retired)}

    async def switch_factory_mode(self, mode) -> dict:
        """热切工厂页承载模式: intercept / mock / real。
        退役全部工人页(货架现货保留, token 通用)。
        mock <-> 其他 的切换会改变浏览器启动参数(host 映射 flag 与
        证书豁免), 需整浏览器回收重建 —— _recycle_browser 里货架不清,
        请求面基本无感; intercept <-> real 只需按新模式重建工人页。"""
        mode = str(mode or "").strip().lower()
        if mode not in FACTORY_MODES:
            raise ValueError(f"factory_mode must be one of {FACTORY_MODES}")
        if mode == self.factory_mode:
            return {"factory_mode": mode, "changed": False,
                    "active": sum(1 for f in self.factories.values()
                                  if not f.retired)}
        old = self.factory_mode
        self.factory_mode = mode
        for f in list(self.factories.values()):
            f.retired = True
            if not f.busy:
                try:
                    await f.stop()
                except Exception:
                    pass
                self.factories.pop(f.wid, None)
        needs_relaunch = (old == "mock") != (mode == "mock")
        if needs_relaunch and getattr(self, "ctx", None):
            if not self._recycling:
                log(f"[pool] factory mode {old} -> {mode}: "
                    f"launch args changed, recycling browser")
                asyncio.create_task(self._recycle_browser())
        elif getattr(self, "ctx", None):
            await self._apply_scale()          # 按新模式重建工人页
        log(f"[pool] factory mode switched: {old} -> {mode}")
        return {"factory_mode": mode, "changed": True,
                "active": sum(1 for f in self.factories.values()
                              if not f.retired)}

    # ---------- 签发与取货 ----------
    def _purge_expired(self):
        now = time.monotonic()
        while self.shelf and now - self.shelf[0][1] > self.token_max_age:
            self.shelf.popleft()
            self.expired_dropped += 1

    def _target_stock(self) -> int:
        """自适应货架目标: 按近 ADAPTIVE_WINDOW 秒的取货速率 × 平均签发时延
        估算"一轮签发周期内会来多少需求", +1 兜底, 封顶 SHELF_SIZE。
        关掉(adaptive_shelf=False)则退回固定 SHELF_SIZE 行为。"""
        if not self.adaptive_shelf:
            return self.shelf_size
        now = time.monotonic()
        while self.served_ts and now - self.served_ts[0] > ADAPTIVE_WINDOW:
            self.served_ts.popleft()
        lat = [x for f in self.factories.values() for x in f.latencies]
        avg_lat = (sum(lat) / len(lat)) if lat else 8.0      # 无样本按 8s 估
        qps = len(self.served_ts) / ADAPTIVE_WINDOW
        return max(1, min(self.shelf_size, math.ceil(qps * avg_lat) + 1))

    async def _sign_and_stock(self, f: TokenFactory):
        token = await f.sign()
        async with self.cond:
            self.inflight -= 1
            if (token and not self._stopping
                    and len(self.shelf) < self._target_stock()):
                self.shelf.append((token, time.monotonic()))
            elif token:
                self.surplus_dropped += 1       # 需求尖峰造成的超产
            self.cond.notify_all()
        if f.retired and not f.busy:            # 缩容收尾
            await f.stop()
            self.factories.pop(f.wid, None)
            log(f"[pool] worker {f.wid} retired (after sign)")
        if not token:
            await asyncio.sleep(3)              # 签发失败退避
            # 连续失败视为死页: 回收掉, 由看门狗建新页顶替
            if f.consecutive_fails >= SIGN_FAIL_RESTART and f.wid in self.factories:
                self.last_error = (f"worker {f.wid}: {f.consecutive_fails} "
                                   f"consecutive sign failures, page recycled")
                log(f"[pool] {self.last_error}")
                try:
                    await f.stop()
                finally:
                    self.factories.pop(f.wid, None)

    async def _refill_loop(self):
        """货架不足且非闲置期时, 把签发任务派给空闲工人页(并发度=空闲页数)。"""
        while not self._stopping:
            async with self.cond:
                self._purge_expired()
                idle = (time.monotonic() - self.last_use > self.idle_pause)
                deficit = self._target_stock() - len(self.shelf) - self.inflight
                while not idle and deficit > 0:
                    free = next((f for f in self.factories.values()
                                 if f.ready and not f.busy), None)
                    if free is None:
                        break
                    free.busy = True              # 占位, 防重复派单
                    self.inflight += 1
                    deficit -= 1
                    asyncio.create_task(self._sign_and_stock(free))
                try:
                    await asyncio.wait_for(self.cond.wait(), timeout=30)
                except asyncio.TimeoutError:
                    pass                          # 周期性自醒: 过期清理/闲置判断

    async def issue(self, timeout=None) -> str:
        """取一个新鲜 token: 有现货秒回, 无现货等补货(上限 ISSUE_TIMEOUT 可配)。"""
        # 空池快速失败: 无存活工人页且非浏览器回收中 → 继续等无意义
        # (看门狗 ≥30s 才可能扩容), 立即超时让 CF failover 马上切走, 而不是挂满 ISSUE_TIMEOUT
        if not self._recycling and not any(
                not f.retired for f in self.factories.values()):
            raise TimeoutError("无存活工人页(池恢复中), 快速失败")
        timeout = self.issue_timeout if timeout is None else timeout
        deadline = time.monotonic() + timeout
        async with self.cond:
            self.last_use = time.monotonic()    # 有需求, 解除闲置静默
            while True:
                self._purge_expired()
                if self.shelf:
                    token, _ = self.shelf.popleft()
                    self.served += 1
                    self.served_ts.append(time.monotonic())   # 自适应货架统计
                    self.cond.notify_all()
                    return token
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise TimeoutError("等待 token 补货超时")
                self.cond.notify_all()
                try:
                    await asyncio.wait_for(self.cond.wait(), timeout=remaining)
                except asyncio.TimeoutError:
                    raise TimeoutError("等待 token 补货超时")

    # ---------- 观测 ----------
    def log_request(self, entry: dict):
        entry["ts"] = time.strftime("%H:%M:%S")
        self.requests.appendleft(entry)

    def stats(self) -> dict:
        now = time.monotonic()
        lat = [x for f in self.factories.values() for x in f.latencies]
        return {
            "ready": self.ready,
            "uptime_s": int(time.time() - _started_at),
            "browser_channel": self.channel,
            "browser_age_s": (int(now - self._browser_started)
                              if self._browser_started else None),
            "last_error": self.last_error,
            "recycling": self._recycling,
            "shelf": {
                "stock": len(self.shelf), "capacity": self.shelf_size,
                "target": self._target_stock(),
                "adaptive": self.adaptive_shelf,
                "oldest_age_s": (round(now - self.shelf[0][1], 1)
                                 if self.shelf else None),
                "inflight_signs": self.inflight,
                "served_total": self.served,
                "expired_dropped": self.expired_dropped,
                "surplus_dropped": self.surplus_dropped,
            },
            "signing": {
                "signed_total": sum(f.stats["signed"]
                                    for f in self.factories.values()),
                "failed_total": sum(f.stats["sign_failed"]
                                    for f in self.factories.values()),
                "avg_latency": (round(sum(lat) / len(lat), 2) if lat else None),
                "recent_latencies": lat[-30:],
            },
            "usage": {
                "prompt_tokens_total": self.usage_in,
                "completion_tokens_total": self.usage_out,
            },
            "workers": [{
                "id": f.wid,
                "state": ("retiring" if f.retired else
                          "busy" if f.busy else
                          "ready" if f.ready else "starting"),
                "busy": f.busy,
                "signed": f.stats["signed"],
                "sign_failed": f.stats["sign_failed"],
                "consecutive_fails": f.consecutive_fails,
                "last_sign_latency": f.stats["last_sign_latency"],
                "latencies": list(f.latencies),
            } for f in sorted(self.factories.values(), key=lambda x: x.wid)],
            "workers_desired": self.desired_pages,
            "workers_max": self.max_worker_pages,
            "config": {
                "shelf_size": self.shelf_size,
                "token_max_age": self.token_max_age,
                "idle_pause": self.idle_pause,
                "max_retry": self.max_retry,
                "retry_statuses": ",".join(map(str, sorted(self.retry_statuses))),
                "worker_pages_init": DEFAULT_WORKER_PAGES,
                "max_worker_pages": self.max_worker_pages,
                "factory_page": self.factory_page,
                "factory_mode": self.factory_mode,
            },
        }

    def stats_lite(self) -> dict:
        """快照字段子集(与 /api/stats 同名同构) —— /health 附带它,
        CF keeper 直接从 /health 落快照, 每空间省一次 /api/stats 子请求"""
        return {
            "last_error": self.last_error,
            "shelf": {"stock": len(self.shelf), "capacity": self.shelf_size,
                      "target": self._target_stock(), "served_total": self.served},
            "signing": {"signed_total": sum(f.stats["signed"] for f in self.factories.values()),
                        "failed_total": sum(f.stats["sign_failed"] for f in self.factories.values())},
            "workers": [{"state": ("retiring" if f.retired else "busy" if f.busy
                                   else "ready" if f.ready else "starting")}
                        for f in self.factories.values()],
            "workers_desired": self.desired_pages,
            "usage": {"prompt_tokens_total": self.usage_in,
                      "completion_tokens_total": self.usage_out},
        }


pool = TokenPool()
# openapi_url=None: 不暴露 /docs /redoc /openapi.json, 避免公开 API 结构
app = FastAPI(title="app",
              docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"],
                   allow_headers=["*"])

# 共享上游连接池(连接复用; 每个响应由调用方负责关闭)
upstream_client: httpx.AsyncClient | None = None


@app.on_event("startup")
async def _startup():
    global upstream_client
    upstream_client = httpx.AsyncClient(
        timeout=300, limits=httpx.Limits(max_connections=64,
                                         max_keepalive_connections=16))
    asyncio.create_task(pool.supervisor())


@app.on_event("shutdown")
async def _shutdown():
    await pool.stop()
    if upstream_client:
        await upstream_client.aclose()


def ensure_ready():
    if not pool.ready:
        raise HTTPException(503, "token factories warming up, retry shortly")


# 每 key 滑动窗口限速窗口(RATE_LIMIT_RPM=0 时关闭)
_rate_windows: dict[str, collections.deque] = {}


def check_key(authorization: str = Header(default="")) -> str:
    """客户端鉴权: OpenAI 兼容端点共用。key 来自 cfg.RELAY_KEYS (环境变量 RELAY_KEYS);
    ADMIN_KEY 亦放行 —— 单 key 部署时只设一个 key 即可全线通行。
    附带每 key 滑动窗口限速: key 泄露时可防止匿名配额被打爆。"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "missing bearer key")
    key = authorization[7:].strip()
    if (key not in pool.relay_keys
            and not (cfg.ADMIN_KEY and key == cfg.ADMIN_KEY)):
        raise HTTPException(401, "invalid key")
    if RATE_LIMIT_RPM > 0:
        now = time.monotonic()
        w = _rate_windows.setdefault(key, collections.deque())
        while w and now - w[0] > 60:
            w.popleft()
        if len(w) >= RATE_LIMIT_RPM:
            raise HTTPException(
                429, "relay rate limit exceeded, slow down or contact admin")
        w.append(now)
    return key


def check_admin(authorization: str = Header(default="")) -> None:
    """管理面板鉴权: 额外要求 ADMIN_KEY。若未配 ADMIN_KEY 或未启 ADMIN_ENABLED,
    所有 /admin 与写操作型 /api/* 路由直接 404/401, 防公网误触。

    HF 部署默认 ADMIN_ENABLED=false —— 即控制台完全收起。需要在线运维时:
    1) 在 Space > Settings > Repository secrets 加 ADMIN_KEY=sk-admin-xxx;
    2) 在 Variables 里设 ADMIN_ENABLED=true(或直接私有也行)。"""
    if not cfg.ADMIN_ENABLED or not cfg.ADMIN_KEY:
        raise HTTPException(404, "admin disabled")
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "missing admin bearer key")
    if authorization[7:].strip() != cfg.ADMIN_KEY:
        raise HTTPException(401, "invalid admin key")
    return None


# --------------------------- 上游调用 ---------------------------
def _retry_backoff(attempt: int) -> float:
    """第 attempt 次重试前的等待秒数: 0.5s 起步翻倍, 封顶 8s。"""
    return min(8.0, 0.5 * (2 ** attempt))


async def call_with_token(open_fn, retries=None):
    """open_fn(token) -> httpx.Response(流式开启)。
    上游响应状态码落在 retry_statuses 里 → 换新 token 重试(短退避),
    次数 pool.max_retry(0=不重试, 支持运行时热改)。最终原样返回最后一次响应。"""
    if retries is None:
        retries = pool.max_retry
    resp = None
    for attempt in range(1 + retries):
        token = await pool.issue()
        resp = await open_fn(token)
        if resp.status_code not in pool.retry_statuses:
            return resp
        if attempt < retries:        # 只有还要重试才关闭+退避, 兜底返回时留给调用方读
            await resp.aclose()
            await asyncio.sleep(_retry_backoff(attempt))
    return resp


def auth_headers(token):
    return dict(BASE_HEADERS, **{"X-DeepInfra-Turnstile": token})


# --------------------------- OpenAI 兼容路由 ---------------------------
@app.post("/v1/chat/completions")
async def chat_completions(request: Request, key: str = Depends(check_key)):
    ensure_ready()
    payload = await request.json()
    model = payload.get("model")
    want_stream = bool(payload.get("stream", False))
    payload["stream"] = True                         # 上游永远流式, 边缘适配
    payload.setdefault("stream_options", {"include_usage": True})
    t0 = time.monotonic()

    async def open_fn(token):
        return await upstream_client.send(
            upstream_client.build_request(
                "POST", f"{UPSTREAM}/v1/openai/chat/completions",
                headers=auth_headers(token), json=payload),
            stream=True)

    try:
        resp = await call_with_token(open_fn)
    except Exception as e:
        pool.log_request({"model": model, "stream": want_stream,
                          "status": 503, "err": str(e)[:120],
                          "latency_ms": int((time.monotonic() - t0) * 1000)})
        raise HTTPException(503, f"token factory unavailable: {e}")

    if resp.status_code != 200:
        body = await resp.aread()
        ctype = resp.headers.get("content-type", "application/json")
        await resp.aclose()
        pool.log_request({"model": model, "stream": want_stream,
                          "status": resp.status_code, "err": body[:120].decode(
                              "utf-8", "replace"),
                          "latency_ms": int((time.monotonic() - t0) * 1000)})
        return Response(content=body, status_code=resp.status_code,
                        media_type=ctype)              # 错误体字节级透传

    if want_stream:
        # 尾部缓冲: usage 只在 SSE 结尾统计块里(stream_options.include_usage),
        # 原始块 O(1) 入窗, 流结束只解析最后 ~64KB —— 计数器是 hub 长流 usage
        # 丢失的对账兜底, 提取失败也绝不影响转发。
        tail = collections.deque()
        tail_len = 0

        def _drain_usage():
            try:
                text = b"".join(tail).decode("utf-8", "replace")
                for line in reversed(text.split("\n")):
                    line = line.strip()
                    if not line.startswith("data:") or '"usage"' not in line:
                        continue
                    try:
                        pool.add_usage(json.loads(line[5:].strip()).get("usage"))
                    except Exception:
                        pass
                    break
            finally:
                tail.clear()
        async def sse_passthrough():
            nonlocal tail_len
            try:
                async for chunk in resp.aiter_raw():    # 原样转发, 全保真
                    tail.append(chunk)
                    tail_len += len(chunk)
                    while tail_len - len(tail[0]) >= 65536 and len(tail) > 1:
                        tail_len -= len(tail.popleft())
                    yield chunk
            finally:
                await resp.aclose()
                _drain_usage()
                pool.log_request({"model": model, "stream": True,
                                  "status": 200, "err": None,
                                  "latency_ms": int(
                                      (time.monotonic() - t0) * 1000)})
        return StreamingResponse(sse_passthrough(),
                                 media_type="text/event-stream")

    # 非流式: 聚合 delta.content / reasoning_content / tool_calls, 透传真实 finish_reason
    text, reasoning, usage = [], [], None
    tool_calls: dict[int, dict] = {}          # index -> 累积中的 tool call 片段
    finish_reason = None
    async for line in resp.aiter_lines():
        if not line.startswith("data:"):
            continue
        data = line[5:].strip()
        if data == "[DONE]":
            break
        try:
            c = json.loads(data)
            for ch in c.get("choices", []):
                d = ch.get("delta", {})
                if d.get("reasoning_content"):
                    reasoning.append(d["reasoning_content"])
                if d.get("content"):
                    text.append(d["content"])
                for tc in d.get("tool_calls") or []:
                    slot = tool_calls.setdefault(tc.get("index", 0), {
                        "id": "", "type": "function",
                        "function": {"name": "", "arguments": ""}})
                    if tc.get("id"):
                        slot["id"] = tc["id"]
                    fn = tc.get("function") or {}
                    if fn.get("name"):
                        slot["function"]["name"] += fn["name"]
                    if fn.get("arguments"):
                        slot["function"]["arguments"] += fn["arguments"]
                if ch.get("finish_reason"):
                    finish_reason = ch["finish_reason"]
            if c.get("usage"):
                usage = c["usage"]
        except Exception:
            pass
    await resp.aclose()
    pool.add_usage(usage)
    pool.log_request({"model": model, "stream": False, "status": 200,
                      "err": None,
                      "latency_ms": int((time.monotonic() - t0) * 1000)})
    msg = {"role": "assistant",
           "content": "".join(text) if (text or not tool_calls) else None}
    if reasoning:
        msg["reasoning_content"] = "".join(reasoning)
    if tool_calls:
        msg["tool_calls"] = [
            {"id": v["id"] or f"call_{idx}", "type": "function",
             "function": v["function"]}
            for idx, v in sorted(tool_calls.items())
        ]
    if not finish_reason:
        finish_reason = "tool_calls" if tool_calls else "stop"
    return {
        "id": "chatcmpl-relay", "object": "chat.completion",
        "created": int(time.time()),
        "model": model,
        "choices": [{"index": 0, "message": msg, "finish_reason": finish_reason}],
        "usage": usage,
    }


async def passthrough_json(request: Request, upstream_path: str):
    """embeddings / completions 等: 原样转发, 不搞边缘聚合。"""
    ensure_ready()
    payload = await request.json()

    async def open_fn(token):
        return await upstream_client.send(
            upstream_client.build_request(
                "POST", f"{UPSTREAM}{upstream_path}",
                headers=auth_headers(token), json=payload),
            stream=True)
    try:
        resp = await call_with_token(open_fn)
    except Exception as e:
        raise HTTPException(503, f"token factory unavailable: {e}")

    if payload.get("stream"):
        async def gen():
            try:
                async for chunk in resp.aiter_raw():
                    yield chunk
            finally:
                await resp.aclose()
        return StreamingResponse(gen(), media_type="text/event-stream",
                                 status_code=resp.status_code)
    body = await resp.aread()
    ctype = resp.headers.get("content-type", "application/json")
    await resp.aclose()
    return Response(content=body, status_code=resp.status_code,
                    media_type=ctype)


@app.post("/v1/embeddings")
async def embeddings(request: Request, key: str = Depends(check_key)):
    return await passthrough_json(request, "/v1/openai/embeddings")


@app.post("/v1/completions")
async def completions(request: Request, key: str = Depends(check_key)):
    return await passthrough_json(request, "/v1/openai/completions")


@app.get("/v1/models")
async def models(key: str = Depends(check_key), full: bool = False):
    """上游全量模型列表原样透传(60s 微缓存, 只为快不为筛选)。
    默认 OpenAI 格式, ?full=true 返回官网富元数据(不缓存)。"""
    path = "/models/list" if full else "/v1/openai/models"
    now = time.monotonic()
    hit = _models_cache.get("entry")
    if not full and hit and hit["key"] == path and now - hit["ts"] < MODELS_CACHE_TTL:
        return Response(content=hit["body"], status_code=hit["status"],
                        media_type="application/json")
    r = await upstream_client.get(f"{UPSTREAM}{path}", timeout=60)
    body = r.content
    if not full:
        _models_cache["entry"] = {"key": path, "ts": now,
                                  "status": r.status_code, "body": body}
    return JSONResponse(status_code=r.status_code,
                        content=json.loads(body.decode("utf-8", "replace")))


# --------------------------- 管理面板与 API ---------------------------
ADMIN_HTML = Path(__file__).with_name("admin.html")

ROOT_HTML = """<!DOCTYPE html>
<html lang="zh-CN"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Service</title>
<style>body{background:#f6f8fa;color:#57606a;font:14px/1.6 -apple-system,
"Segoe UI","Microsoft YaHei",sans-serif;padding:48px;max-width:640px;margin:0 auto}</style></head>
<body>
<h1 style="font-size:20px;color:#24292f">🟢 Service is running</h1>
<p>Nothing to see here.</p>
</body></html>"""


@app.get("/", response_class=HTMLResponse)
async def status_page():
    """根路径: 极简占位页, 不含任何运行数据/端点提示。"""
    return HTMLResponse(ROOT_HTML)


@app.get("/health")
async def health():
    """HF Space 探活端点。带快照字段子集(与 /api/stats 同名同构, 见 stats_lite)
    —— CF keeper 从这里直接落快照, 每空间省一次 /api/stats 子请求;
    models 也自报(hub 目录路由的动态成员 + /v1/models 聚合数据源)。
    ⚠ 曾有一次编辑事故: 函数上方残留了重复的 @app.get("/health") 装饰器(首个
    定义体只有 docstring), FastAPI 按注册顺序命中前者 → /health 返回 null,
    hub 的健康捷径静默失效回退全量拉取, ready 字段一直为空。改动此处时
    保持只有一个 /health 路由。"""
    return {"ok": pool.ready, "ready": pool.ready,
            "uptime_s": int(time.time() - _started_at),
            "load": {"saturation": _load_saturation()},
            "models": await _stats_models(),
            **pool.stats_lite()}


@app.get("/admin", response_class=FileResponse)
async def admin_page(request: Request):
    """完整管理控制台。页面骨架本身不含数据, 数据/写操作 API 各自独立鉴权。

    低成本伪装: 无 ?ui=1 查询参数时返回 404(与不存在的路由表现一致),
    避免被扫描器/随手访问者发现控制台入口; 自己访问用 /admin?ui=1。"""
    if not cfg.ADMIN_ENABLED or not cfg.ADMIN_KEY:
        raise HTTPException(404, "admin disabled")
    if request.query_params.get("ui") != "1":
        raise HTTPException(404, "Not Found")
    return FileResponse(ADMIN_HTML, media_type="text/html")


@app.get("/api/stats")
async def api_stats(key: str = Depends(check_key)):
    s = pool.stats()
    # 标准自报块(hub 平台层契约): 通用健康 + 饱和度 + 目录
    s["load"] = {"saturation": _load_saturation()}
    s["models"] = await _stats_models()
    return s


@app.get("/stats")                 # 兼容旧测试套件
async def stats_compat(key: str = Depends(check_key)):
    return pool.stats()


@app.get("/api/requests")
async def api_requests(key: str = Depends(check_key)):
    return {"requests": list(pool.requests)}


@app.post("/api/pool/scale")
async def api_scale(request: Request, _admin=Depends(check_admin)):
    body = await request.json()
    pages = int(body.get("pages", pool.desired_pages))
    return await pool.scale(pages)


# ---------- 参数热改 ----------
# 数值型参数的校验范围 + 对应 setter; hi=None 表示只设下限不封顶(如重试次数)。
CONFIG_FIELDS = {
    "shelf_size":        (0, 100000, pool.set_shelf_size),
    "token_max_age":     (1, 300,    pool.set_token_max_age),
    "idle_pause":        (0, 86400,  pool.set_idle_pause),
    "max_retry":         (0, None,   pool.set_max_retry),
    "max_worker_pages":  (1, 256,    pool.set_max_worker_pages),
}


def config_snapshot() -> dict:
    return {
        "shelf_size": pool.shelf_size,
        "token_max_age": pool.token_max_age,
        "idle_pause": pool.idle_pause,
        "max_retry": pool.max_retry,
        "retry_statuses": ",".join(map(str, sorted(pool.retry_statuses))),
        "worker_pages": pool.desired_pages,        # 调它请用 /api/pool/scale
        "max_worker_pages": pool.max_worker_pages,
        "factory_page": pool.factory_page,         # 热切换: 工人页按新页重建
        "factory_mode": pool.factory_mode,         # 热切换: intercept/mock/real
    }


@app.get("/api/config")
async def api_config_get(key: str = Depends(check_key)):
    return config_snapshot()


@app.put("/api/config")
async def api_config_put(request: Request, _admin=Depends(check_admin)):
    """批量热改参数(需 ADMIN_KEY), 只接受 CONFIG_FIELDS 里的字段; 越界/非法静默跳过并报告。"""
    body = await request.json()
    applied, skipped = {}, {}
    for name, value in body.items():
        # worker_pages / factory_page / factory_mode / retry_statuses 走下方
        # 的特殊处理, 不在数值表里
        if name in ("worker_pages", "factory_page", "factory_mode",
                    "retry_statuses"):
            continue
        spec = CONFIG_FIELDS.get(name)
        if not spec:
            skipped[name] = "unknown field"; continue
        lo, hi, setter = spec
        try:
            v = int(value)
        except (TypeError, ValueError):
            skipped[name] = f"not int: {value!r}"; continue
        if v < lo or (hi is not None and v > hi):
            skipped[name] = f"out of range [lo={lo}, hi={'∞' if hi is None else hi}]"; continue
        setter(v)
        applied[name] = v
    # retry_statuses 是唯一的集合型热改字段
    if "retry_statuses" in body:
        if pool.set_retry_statuses(body["retry_statuses"]):
            applied["retry_statuses"] = ",".join(map(str, sorted(pool.retry_statuses)))
        else:
            skipped["retry_statuses"] = "no valid status codes"
    # worker_pages 改了则触发一次扩缩容(经 /api/pool/scale 同一路径)
    if "worker_pages" in body:
        try:
            await pool.scale(int(body["worker_pages"]))
            applied["worker_pages"] = pool.desired_pages
        except Exception as e:
            skipped["worker_pages"] = str(e)
    # factory_page 是唯一的字符串型热改字段: 切换后工人页按新页重建
    if "factory_page" in body:
        try:
            r = await pool.switch_factory_page(body["factory_page"])
            applied["factory_page"] = r["factory_page"]
        except Exception as e:
            skipped["factory_page"] = str(e)
    # factory_mode: 工厂页承载方式热切(intercept/mock/real)
    if "factory_mode" in body:
        try:
            r = await pool.switch_factory_mode(body["factory_mode"])
            applied["factory_mode"] = r["factory_mode"]
        except Exception as e:
            skipped["factory_mode"] = str(e)
    return {"applied": applied, "skipped": skipped, "now": config_snapshot()}


# ---------- 鉴权 key 管理(热加/删, 需 ADMIN_KEY) ----------
@app.get("/api/keys")
async def api_keys_get(_admin=Depends(check_admin)):
    return {"keys": sorted(pool.relay_keys)}


@app.put("/api/keys")
async def api_keys_put(request: Request, _admin=Depends(check_admin)):
    """body: {"add":[...],"remove":[...]} 热改 relay key 集合。需 ADMIN_KEY。
    ⚠️ 删掉当前正在用的 key 会立即让本次会话之后的请求失效——别把自己锁门外。"""
    body = await request.json()
    add = set(body.get("add") or [])
    rm = set(body.get("remove") or [])
    if not add and not rm:
        return {"keys": sorted(pool.relay_keys), "note": "no changes"}
    before = set(pool.relay_keys)
    pool.relay_keys = (before | add) - rm
    return {"keys": sorted(pool.relay_keys),
            "added": sorted(add & pool.relay_keys),
            "removed": sorted(rm & before)}


@app.post("/api/test/chat")
async def api_test_chat(request: Request, _admin=Depends(check_admin)):
    """面板测试对话(需 ADMIN_KEY): 走与 /v1/chat/completions 相同的路径, 返回耗时拆解。"""
    body = await request.json()
    model = body.get("model") or "moonshotai/Kimi-K3"
    prompt = body.get("prompt") or "回一个 pong"
    ensure_ready()
    payload = {"model": model,
               "messages": [{"role": "user", "content": prompt}],
               "stream": True,
               "stream_options": {"include_usage": True},
               "max_tokens": int(body.get("max_tokens", 200))}
    t0 = time.monotonic()
    t_token, resp, status = None, None, 503
    try:
        for attempt in range(1 + pool.max_retry):
            token = await pool.issue()
            t_token = round(time.monotonic() - t0, 2)   # 等 token 的耗时
            resp = await upstream_client.send(
                upstream_client.build_request(
                    "POST", f"{UPSTREAM}/v1/openai/chat/completions",
                    headers=auth_headers(token), json=payload),
                stream=True)
            status = resp.status_code
            if status not in pool.retry_statuses:
                break
            await resp.aclose()
            await asyncio.sleep(_retry_backoff(attempt))
    except Exception as e:
        return {"ok": False, "stage": "token", "error": str(e)[:200]}

    t_up0 = time.monotonic()
    if status != 200:
        err = (await resp.aread()).decode("utf-8", "replace")[:300]
        await resp.aclose()
        return {"ok": False, "stage": "upstream", "status": status,
                "error": err, "t_token_s": t_token}
    text = []
    usage = None
    async for line in resp.aiter_lines():
        if line.startswith("data:"):
            data = line[5:].strip()
            if data == "[DONE]":
                break
            try:
                c = json.loads(data)
                for ch in c.get("choices", []):
                    d = ch.get("delta", {})
                    if d.get("content"):
                        text.append(d["content"])
                usage = c.get("usage") or usage
            except Exception:
                pass
    await resp.aclose()
    t_up = round(time.monotonic() - t_up0, 2)
    return {"ok": True, "model": model, "content": "".join(text)[:500],
            "usage": usage, "t_token_s": t_token, "t_upstream_s": t_up,
            "t_total_s": round(time.monotonic() - t0, 2)}