api_server.py

"""aiduMEM 应用组装层。业务逻辑位于 ducky/ 各模块。"""
from __future__ import annotations

# ── posthog stub（省 ~23M RSS）──────────────────────────
# mem0 SDK 顶层 import posthog 做遥测，但我们不需要。
# 在 mem0 之前注入一个空壳模块，避免加载真正的 posthog 包。
# 不改 mem0 源码，升级安全。
import types as _types, os as _os

# ── .env 早期加载（v20.2.4 · 外审 F-22）────────────────────────────
#
# `ducky.utils.load_env_file()` 一直存在，但**从没有任何生产入口调用过它**
# （全仓只有测试在调）。于是照 README 复制一份 .env 再 `python api_server.py`，
# token / UI 口令可能根本不生效 —— 文档教的那条路和代码走的那条路不是同一条。
#
# 为什么内联在这里、而不是调 ducky.utils.load_env_file：**顺序契约**。
# `ducky.utils` 的 BASE_DIR / DATA_DIR 等一批常量是 **import 期求值**的，
# 一旦 `import ducky.*` 发生，再注入环境变量就晚了。所以这段必须跑在第一次
# import ducky 之前，而它又不能依赖 ducky —— 只好就地写这十几行。
# 解析规则与 ducky.utils.parse_env_file 保持一致（不覆盖已存在的变量）。
def _bootstrap_env_file() -> int:
    _here = _os.path.dirname(_os.path.abspath(__file__))
    _path = _os.environ.get("AIDUMEM_ENV_FILE") or _os.path.join(_here, ".env")
    _n = 0
    try:
        with open(_path, encoding="utf-8") as _fh:
            for _line in _fh:
                _line = _line.strip()
                if not _line or _line.startswith("#") or "=" not in _line:
                    continue
                _k, _v = _line.split("=", 1)
                _k, _v = _k.strip(), _v.strip().strip('"').strip("'")
                # f0.7.1：与 parse_env_file 一致，容忍 `export KEY=VAL`（给 shell source 用的 .env）。
                # 原先不剥前缀 → 键名变成 "export AIDUMEM_API_TOKEN"，服务端门禁静默关闭，
                # 而钩子/脚本照常带 token 调用，一切看似正常。
                if _k.startswith("export ") or _k.startswith("export\t"):
                    _k = _k[len("export"):].strip()
                # f0.8（外审 deepseek F-21）：环境里「存在但为空」不算显式给值——此前
                # `AIDUMEM_API_TOKEN=`（systemd Environment=、compose -e）会把 .env 里的真 token
                # 顶掉，门禁静默关闭。空/纯空白的环境值让位给 .env 的非空值。
                if _k and not _os.environ.get(_k, "").strip() and (_v or _k not in _os.environ):
                    _os.environ[_k] = _v
                    _n += 1
    except OSError:
        return 0          # 文件不存在 / 读不了：静默，绝不拖垮启动
    return _n


_ENV_FILE_INJECTED = _bootstrap_env_file()

_os.environ.setdefault("MEM0_TELEMETRY", "false")
_stub = _types.ModuleType("posthog")
class _NoopPosthog:
    """Lightweight posthog stub — all calls are silent no-ops."""
    def __init__(self, *a, **kw): pass
    def capture(self, *a, **kw): pass
    def shutdown(self, *a, **kw): pass
    def evaluate_flags(self, *a, **kw): return {}
    def feature_enabled(self, *a, **kw): return False
_stub.Posthog = _NoopPosthog
import sys
sys.modules["posthog"] = _stub
del _stub, _NoopPosthog, _types
# ── end posthog stub ──────────────────────────────────

import logging
from contextlib import asynccontextmanager as _asynccontextmanager
import os
import threading
import uuid as _uuid
import time
from ducky.security.auth import safe_equal as _safe_equal
from pathlib import Path

import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles

from ducky.autodream import autodream_background_loop
from ducky.evolve_mem import evolve_background_loop
from ducky.core_memory import init_core_memory
from ducky.extended import _auto_expire_loop, auto_memory_background_loop
from ducky.hot.health import set_version_info
from ducky.hot.legacy import (
    _background_consolidation_loop,
    _background_scene_cluster_loop,
    _extract_entities,
    _extract_key_facts,
    _get_db,
)
from ducky.mem0_runtime import get_memory
from ducky.reflect import reflect_background_loop
from ducky.consolidate import consolidate_background_loop
from ducky.routes_registry import UI_ROUTES, register_all_routes
from ducky.schema_bootstrap import ensure_core_schema
from ducky.text_fts import _init_text_fts
from ducky.utils import LOG_DIR
from ducky.version import SERVICE_VERSION, CODENAME, CODENAME_ZH, DISPLAY_NAME

_os.makedirs(LOG_DIR, exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(_os.path.join(LOG_DIR, "api_server.log")),
        logging.StreamHandler(sys.stderr),
    ],
)
logger = logging.getLogger(f"aiduMEM-v{SERVICE_VERSION}")


@_asynccontextmanager
async def _lifespan(_app: "FastAPI"):
    """把后台能力挂进 FastAPI 生命周期（v20 · P1-6）。

    整改前，全部后台能力（建表、mem0 预热、WAL 启动对账与自愈、FTS 初始化、
    核心记忆初始化、实体词表自检、以及 `_BACKGROUND_LOOPS` 里那一批常驻线程）
    只挂在 `main()` 里。`main()` 是 `python api_server.py` 和控制台入口点走的路 ——
    但**不是唯一的路**：

        uvicorn api_server:app        # 官方文档最常见的起法
        gunicorn -k uvicorn.workers.UvicornWorker api_server:app
        任何把 `app` 当 ASGI 对象导入的进程

    这些起法完全不经过 `main()`。于是服务照样监听、`/health` 照样返回 ok、
    读写接口照样能用 —— **而 WAL 对账没跑、后台线程一个没起**。这不是崩溃，
    是「服务看着是好的，一半的能力静默缺席」（静默失败铁律）。全文此前 0 处
    `lifespan` / `on_event`，也就是说这条路上没有任何东西会告诉你缺了什么。

    `_start_background()` 本身是幂等的（`_background_started` 双检锁），所以
    `main()` 那一次调用保持原样不动 —— 两条路各自都能起，重复调用无副作用。

    v20.2.4（外审 F-01）：安全门禁也挪到了这里，且放在 `_start_background()`
    **之前** —— 拒绝启动就该在建表、预热、起线程之前发生。
    """
    _enforce_public_binding_policy()
    # f0.7.5（5 家外审共识）：配了 agent 钥匙却没开强制 / 主 token 未绑定 → 醒目告警。
    try:
        from ducky.agent_keys import startup_warnings
        for _w in startup_warnings():
            logger.warning("⚠️ [AgentKeys] %s", _w)
    except (ImportError, OSError, ValueError) as _e:  # 观测不许拖垮启动
        logger.debug("agent key startup check skipped: %s", _e)
    # f0.8+ C：启动时读一次 git HEAD，/health 的 git_sha 随代码走，不随 .env 漂移。
    try:
        from ducky.hot.health import _git_sha
        logger.info("build git_sha=%s", _git_sha()[:12])
    except (ImportError, OSError, ValueError) as _e:  # 观测不许拖垮启动
        logger.debug("git_sha warm-up skipped: %s", _e)
    # v20.3.2-beta（外审 P0-3）：单进程契约与公网熔断同级，**同样在起线程之前**。
    _enforce_single_process_policy()
    from ducky.process_lock import acquire_api_process_lock
    from ducky.utils import DATA_DIR
    acquire_api_process_lock(DATA_DIR)
    # Establish durable write debt before any background writer or request can run.
    # A corrupt/unreadable journal prevents startup; uncertain writes stay held
    # for explicit repair rather than being blindly submitted to the SDK again.
    from ducky.mutation_journal import startup_recover
    recovery = startup_recover()
    if recovery.get("error") or recovery.get("integrity") != "ok":
        raise RuntimeError("mutation journal startup integrity check failed")
    if recovery.get("status") != "ok":
        logger.warning("Durable writes require repair: %d", recovery.get("repair_required", 0))
    # v20.3.2 正式版（用户审计 H / GLM F-3）：设了本系统**不读**的 AIDUME?_ 变量，大概率是
    # 拼错了前缀（AIDUMEM_/AIDUMEI_ 双前缀 90+ 个，分界无规律）。此前静默按默认值跑 ——
    # 作者本人同一版本踩了两次。现在启动期扫 os.environ，未知名字 WARNING + 最接近的正确名。
    from ducky.env_registry import warn_unknown_env_vars
    warn_unknown_env_vars()
    # E1（v20.3.2 正式版）：部署方自定义互斥规则（DATA_DIR/conflict_rules.json）—— 逃生阀终于接线
    from ducky.conflict_resolver import load_custom_exclusion_patterns_from_file
    try:
        _n_rules = load_custom_exclusion_patterns_from_file()
        if _n_rules:
            logger.info("🐙 已装载 %d 条部署方自定义互斥规则", _n_rules)
    except Exception as _exc:
        logger.warning("自定义互斥规则装载失败（忽略，使用内置规则）：%s", _exc)
    _start_background()
    yield
    # P2-20（v20.3.2 正式版 · Gemini P1-4）：优雅停机 —— 叫停七个循环，最多等 3s 让
    # 正在跑的一轮收尾。此前守护线程被进程退出硬切，半截巩固/反思留在库里。
    from ducky.shutdown import request_shutdown
    request_shutdown()
    _deadline = time.monotonic() + 3.0
    for _t in list(_BACKGROUND_THREADS):
        _t.join(max(0.0, _deadline - time.monotonic()))
    _alive = [t.name for t in _BACKGROUND_THREADS if t.is_alive()]
    if _alive:
        logger.warning("停机：%d 个后台循环 3s 内未退出（守护线程随进程结束）：%s", len(_alive), _alive)
    else:
        logger.info("停机：后台循环全部收尾退出")


app = FastAPI(
    lifespan=_lifespan,
    # /docs 的封面标题，是 API 侧的「品牌门面」。
    # 注意只改展示文案：logger 名、/health 的 service 字段、线程名、
    # `AIDUMEM_*` 环境变量、各模块 docstring 里的 aiduMEM 都是机器契约或历史内部名，
    # 生产侧的日志采集与监控按它们匹配，v19.4.2 决策 D2 起一律不动。
    # v20 把这条分界写成了可执行的守卫，见 tests/test_v20_brand_policy.py：
    # 露脸的字（打印/注入/--help/systemctl status/用户会打开的文档与样例）必须是
    # 当前品牌名；机器认的键一个都不动；历史记录只新增、不改写。
    # v20 uses a two-part version and intentionally has no current codename.
    # Keep the OpenAPI version machine-readable instead of coupling clients to
    # a display label that changed every mythological generation.
    title=f"aiduMEI API v{SERVICE_VERSION}",
    version=SERVICE_VERSION,
)


# f0.6 W-26 施工实录（cue A11）：chunked 超限最初试过三条信号路径——裸异常、
# 专用 ClientDisconnect 子类 + app 级异常处理器、HTTPException(413)——全部
# 被 fastapi 路由层 body 读取的 except Exception 包成假 400，或在
# BaseHTTPMiddleware 的 wrapped_receive 桥接层变形。终版方案在 middleware
# 内直接分块消费 stream、就地回 413、限额内把 body 塞 request._body 缓存
# 交下游（_CachedRequest state-3 分支原样转交）——异常根本不产生。

# ── 前端 UI 托管（aiduMEM 自带面板）──────────────────────
# UI_DIR 指向 frontend/ 目录；未指定时取本文件同级的 frontend/。
# 访问 / 与 /ui/ 即可打开控制台，页面通过 /api/* 与本服务通信。
_UI_DIR = os.environ.get("UI_DIR", str(Path(__file__).resolve().parent / "frontend"))
if Path(_UI_DIR).is_dir():
    app.mount("/ui", StaticFiles(directory=_UI_DIR, html=True), name="ui")

    @app.get("/", include_in_schema=False)
    def _ui_root():
        return RedirectResponse("/ui/")

    logger.info("🖥️ 前端 UI 已挂载: %s → /ui/", _UI_DIR)
else:
    logger.warning("⚠️ 未找到前端目录 %s（仅 API 模式运行）", _UI_DIR)

# 兼容旧模块仍从 api_server 导入这些符号。
__all__ = [
    "app",
    "get_memory",
    "_extract_entities",
    "_extract_key_facts",
    "_get_db",
]

# 注册所有路由（统一入口）
register_all_routes(app, get_memory, _get_db, _extract_entities)

# ── /api 子应用（控制台 UI）──────────────────────────────
# 前端以 /api/* 为调用根（API.base = '/api'）。f0.6+++ WP2：子应用只注册
# UI 实际调用的端点（UI_ROUTES），不再整表镜像扁平路由。
_api_alias = FastAPI(title="aiduMEI /api (console)")
register_all_routes(_api_alias, get_memory, _get_db, _extract_entities, only=UI_ROUTES)
app.mount("/api", _api_alias)

# ── 统一鉴权门禁（🔴P0-1 v19.4.1 鉴权贯通）──────────────────────────────
#
# v19.4.0 的问题：UI 口令与 API token 是两套互不相通的凭据，实测两种组合
# 都不可用 ——
#   · 只设 AIDUMEM_UI_PASSWORD：中间件因 token 为空整段放行，未登录直接
#     GET /api/facts → 200，全部记忆裸奔（UI 登录只是前端 sessionStorage 障眼法）；
#   · 只设 AIDUMEM_API_TOKEN：/api/login 返回 200，但前端从不发 Authorization 头，
#     登录后所有面板 → 401，控制台报废。
#
# v19.4.1 的解法：**一道门禁，两把钥匙**。
#   钥匙 A：HttpOnly session cookie（/login 成功后服务端签发，浏览器自动携带）
#   钥匙 B：Authorization: Bearer <AIDUMEM_API_TOKEN>（脚本 / MCP / CI 用）
# 任一有效即放行。门禁在「配置了 token **或** 配置了 UI 口令」时启用 ——
# 于是只设 UI 口令的部署方也真的被保护，而不是自以为被保护。
# 两者都没配时保持旧行为（本机回环零配置可用）。
#
# 注意：token 与口令状态都在**请求时**实时读取，不在模块加载时定格。
# 理由有二：① `/config/password` 可在运行时首次设置口令，门禁必须立刻生效；
# ② 模块级常量会迫使测试用 sys.modules 清洗来切换部署形态，那会连带
# 抹掉其它测试文件在 import 期做的 DB 重定向（本版施工中真实踩到）。
def _api_token() -> str:
    return os.environ.get("AIDUMEM_API_TOKEN", "").strip()


# 交互式 API 文档（/docs /redoc /openapi.json）是否免凭据。
#
# 🟡（v19.4.1 用户审计）：门禁启用后这三个路径仍返回 200，等于把 135 个
#     端点的完整清单（含参数与请求体结构）交给未授权访问者。对自托管
#     记忆库而言，这是一份现成的攻击面地图。
#     但它同时也是开发调试与接入排障的主要入口，直接锁死会明显劣化体验。
#     取舍：**门禁启用时默认一并保护**，需要公开时显式设
#     AIDUMEM_PUBLIC_DOCS=1（例如本机开发、或已在反代层另加保护）。
#     门禁未启用时（本机零配置）行为完全不变。
def _public_docs_enabled() -> bool:
    return os.environ.get("AIDUMEM_PUBLIC_DOCS", "0").strip().lower() in {"1", "true", "yes"}


_DOC_PATHS = frozenset({
    "/docs", "/api/docs",
    "/redoc", "/api/redoc",
    "/openapi.json", "/api/openapi.json",
})

# 登录与健康检查必须永久免凭据：前者是拿到凭据的唯一入口，
# 后者是监控探针的依赖，锁死会让服务「看起来挂了」。
# /livez、/readyz 同理（v20.4 P2-17）：负载均衡器/编排器无凭据高频探测，
# 两者响应只含版本与检查名布尔，不泄路径；/diagnostics 是完整深度探针，
# 一律留在门禁后，不进本集合。
_ALWAYS_PUBLIC_PATHS = frozenset({
    "/", "/ui",
    "/api/login", "/api/login/hint",
    "/health", "/api/health",
    "/livez", "/readyz",
})


def _is_public_path(path: str) -> bool:
    """无需凭据即可访问的路径（登录、健康检查、静态 UI，以及可选的文档）。"""
    if path in _ALWAYS_PUBLIC_PATHS:
        return True
    if path in _DOC_PATHS:
        return _public_docs_enabled()
    if path.startswith("/ui/"):
        return True
    return False


def _auth_enabled() -> bool:
    """门禁是否启用：配了 API token 或 UI 口令任一即启用。

    每次请求实时判定 —— `/config/password` 可在运行时首次设置口令，
    门禁必须立刻生效，不能等重启。
    """
    if _api_token():
        return True
    try:
        from ducky import agent_keys
        if agent_keys.configured():
            return True
        from ducky.security.auth import ui_password_configured
        return ui_password_configured()
    except Exception as exc:
        # 判定失败时**保守启用**门禁：宁可要求凭据，也不裸奔。
        logger.debug("门禁启用判定失败，保守启用: %s", exc)
        return True


def _request_authorized(request: Request) -> bool:
    """钥匙 A（session cookie）∨ 钥匙 B（Bearer token），任一有效即放行。"""
    from ducky.security.auth import (
        SESSION_COOKIE_NAME,
        fingerprint_token,
        set_request_auth_kind,
        set_request_token_fingerprint,
        validate_session,
    )

    from ducky import agent_keys
    agent_keys.clear_current()
    session_token = request.cookies.get(SESSION_COOKIE_NAME, "")
    if session_token and validate_session(session_token):
        # v22.0（雷霆审计 A3）：UI 会话 = 主人直连，跨殿授权可免 caller
        set_request_auth_kind("session")
        return True

    token = _api_token()
    supplied = request.headers.get("Authorization", "")
    alt = request.headers.get("X-API-Token", "")
    if token:
        if supplied and _safe_equal(supplied, f"Bearer {token}"):
            # v20.5.1（T-07）：钥匙 B 通过时把 token 指纹挂进请求上下文，
            # 供联邦层 _require_caller 做 caller↔凭据轻量绑定
            # （AIDUMEI_CALLER_BINDINGS 未配置时该指纹无人读取，零行为变化）。
            set_request_token_fingerprint(token)
            # v22.0（雷霆审计 A3）：API token = agent/集成，跨殿授权必须带 caller
            set_request_auth_kind("bearer")
            agent_keys.set_current(None, fingerprint_token(token))
            return True
        # 兼容以 X-API-Token 头传递的调用方
        if alt and _safe_equal(alt, token):
            set_request_token_fingerprint(token)
            set_request_auth_kind("bearer")
            agent_keys.set_current(None, fingerprint_token(token))
            return True
    # f0.7+（外审 S3）：每个 agent 一把钥匙，服务端只存 sha256，钥匙绑定身份（agent_keys.json）。
    candidate = supplied[7:].strip() if supplied[:7].lower() == "bearer " else alt.strip()
    entry = agent_keys.match(candidate) if candidate else None
    if entry is not None:
        set_request_token_fingerprint(candidate)
        set_request_auth_kind("bearer")
        agent_keys.set_current(entry, fingerprint_token(candidate))
        return True
    return False


def _is_test_client(host: str) -> bool:
    """v22.0（雷霆审计 B4 · GLM F-06）：测试客户端身份判定。

    host 是 IP 才有网络语义——非法 IP 才有可能是伪造。`testclient` 是
    ASGI 直连测试客户端的无 TCP 对端占位符，真实网络请求的 client.host
    一定是 IP，所以这条在生产里不可能被利用。
    默认信任测试客户端（测试基座不中断）；`AIDUMEI_TEST_CLIENT_TRUST=0`
    可在加固部署里强制拒绝。
    """
    if host.strip().lower() != "testclient":
        return True
    return _os.environ.get("AIDUMEI_TEST_CLIENT_TRUST", "").strip() != "0"


def _client_is_loopback(request: "Request") -> bool:
    """请求是否来自回环。

    非 IP 形态的 `client.host`（ASGI 直连测试客户端报的是 "testclient"）视为本地：
    真实网络请求的 client.host 一定是 IP，所以这条在生产里不可能被利用；而把它
    判成「非回环」会让所有不配凭据的用例集体 503 —— 那是拿测试基座换一个假的
    严格。取舍写在这里，**别把它放宽成「任意字符串都放行」**。
    拿不到来源时 fail-closed（当作非回环）。
    """
    import ipaddress
    client = getattr(request, "client", None)
    host = (getattr(client, "host", "") or "").strip()
    if not host:
        return False
    if host == "testclient":
        return _is_test_client(host)
    try:
        return ipaddress.ip_address(host).is_loopback
    except ValueError:
        return host in ("localhost", "testclient")


_LOOPBACK_HOSTS = ("127.0.0.1", "localhost", "::1")

# ── 反代痕迹探测（v20.3.2-beta · 外审 C-1）────────────────────────────────
#
# 🔴 C-1：`_client_is_loopback` 判的是 **TCP 对端 IP**，而同机 nginx 反代的
# 对端**恒为 127.0.0.1**。于是「无凭据 → 只服务回环」这道防线，在本项目
# **标准生产形态**（域名 + 同机反代）下等于全部放行 —— 而「无凭据」正是
# 默认安装的终点（自动生成口令 source=auto 刻意不启门禁）。
#
# 判据只看代理头**在不在**，绝不看它的**值**：
#   · 值可伪造（rate_guard.py 早就写明拒绝采信 XFF 的值，那是对的）；
#   · 但**存在性不可否认** —— 这个请求经过了一跳，是客户端无法抹掉的事实。
# 无凭据实例不该服务任何经过跳转的请求，所以「有痕迹 → 拒绝」是 fail-closed。
#
# 逃生阀是 AIDUMEI_TRUST_PROXY：部署方显式声明「我知道我在反代后面」才放行。
# 宁可让反代部署者多配一个变量，也不让默认安装裸奔。
_PROXY_HINT_HEADERS = (
    "x-forwarded-for", "x-real-ip", "forwarded",
    "x-forwarded-host", "x-forwarded-proto", "x-client-ip",
)


def _request_via_proxy(headers) -> bool:
    """请求是否带反代痕迹。**只判存在性，不读值。**"""
    try:
        for name in _PROXY_HINT_HEADERS:
            v = headers.get(name)
            if v is not None and str(v).strip():
                return True
    except Exception:
        # 取不到头时 fail-closed：当作经过了代理（更严的赢）。
        return True
    return False


# ── P1-12（v20.3.2 正式版 · Codex F-04）：无凭据模式的 Host 校验 ─────────────
# 无凭据实例只服务回环 —— 但「对端是回环」拦不住 DNS rebinding：攻击页把
# attacker.example 解析到 127.0.0.1，用户的浏览器就以回环对端、攻击者 Host 打本地 API，
# 无 CORS 预检的简单请求还能直接写。所以对端判据之外再加两道只看请求头的判据：
#   ① Host 必须是回环名 / AIDUMEM_HOST / AIDUMEI_TRUSTED_HOSTS 之一（否则 421）；
#   ② 浏览器发起的写（有 Origin 或 Sec-Fetch-Site: cross-site）Origin 必须同样受信（否则 403）。
# 无 Origin 的 CLI / MCP / cron 请求不受 ② 影响；有凭据的实例由鉴权守住写面，这两道不启用；
# 声明了 AIDUMEI_TRUST_PROXY=1 时 Host 校验交给反代（反代转发的是对外域名）。
# f0.8（外审 deepseek F-22 附带）：0.0.0.0 不是回环名——部分浏览器把它当本机发请求（"0.0.0.0 day"），
# 把它当受信 Host 等于给 DNS rebinding 留了一个口。真绑 0.0.0.0 的部署必须配凭据，不走这条。
_LOOPBACK_HOST_NAMES = frozenset({"localhost", "127.0.0.1", "::1", "[::1]"})


def _trusted_host_names() -> frozenset:
    extra = {h.strip().lower() for h in os.getenv("AIDUMEI_TRUSTED_HOSTS", "").split(",") if h.strip()}
    bound = (os.getenv("AIDUMEM_HOST") or "").strip().lower()
    return frozenset(_LOOPBACK_HOST_NAMES | extra | ({bound} if bound else set()))


def _host_without_port(value: str) -> str:
    h = (value or "").strip().lower()
    if h.startswith("["):            # [::1]:8767
        return h.split("]", 1)[0] + "]"
    if h.count(":") == 1:            # 127.0.0.1:8767 / localhost:8767
        return h.rsplit(":", 1)[0]
    return h                          # 裸 IPv6 或无端口


def _host_header_allowed(request: "Request") -> bool:
    if _trust_proxy_enabled():
        return True
    client_host = (getattr(getattr(request, "client", None), "host", "") or "").strip()
    if client_host == "testclient":
        # ASGI 直连测试客户端：没有 TCP 对端、没有 DNS，rebinding 不成立。
        # v22.0：改显式 env 注入，不再靠字符串字面量硬编码。
        return _is_test_client(client_host)
    return _host_without_port(request.headers.get("host", "")) in _trusted_host_names()


def _browser_cross_site_write(request: "Request") -> bool:
    if request.method.upper() not in ("POST", "PUT", "PATCH", "DELETE"):
        return False
    if (request.headers.get("sec-fetch-site") or "").strip().lower() == "cross-site":
        return True
    origin = (request.headers.get("origin") or "").strip().lower()
    if not origin:
        return False                  # 非浏览器客户端不带 Origin
    if origin == "null":
        return True                   # 不透明来源（file:// 等）
    from urllib.parse import urlsplit
    return _host_without_port(urlsplit(origin).netloc) not in _trusted_host_names()


def _cookie_write_not_same_origin(request: "Request") -> bool:
    """f0.8+（外审 R3-13）：UI 会话（cookie）认证的写请求必须来自同源页面。

    SameSite=Lax 挡得住跨站 POST，挡不住**同站**的其他子域（同一域名下的别的服务）。
    判据：浏览器自带的 Sec-Fetch-Site 有就以它为准（页面伪造不了），只认 same-origin / none；
    没有时看 Origin：null 拒，有值就必须与 Host（可信反代时也认 X-Forwarded-Host）同 host:port。
    两个头都没有的（非浏览器客户端）不受影响。"""
    if request.method.upper() not in ("POST", "PUT", "PATCH", "DELETE"):
        return False
    sfs = (request.headers.get("sec-fetch-site") or "").strip().lower()
    if sfs:
        return sfs not in ("same-origin", "none")
    origin = (request.headers.get("origin") or "").strip().lower()
    if not origin:
        return False
    if origin == "null":
        return True
    from urllib.parse import urlsplit
    parts = urlsplit(origin)
    netloc = parts.netloc
    default_port = {"http": ":80", "https": ":443"}.get(parts.scheme, "")
    if default_port and netloc.endswith(default_port):
        netloc = netloc[: -len(default_port)]
    hosts = {(request.headers.get("host") or "").strip().lower()}
    if _trust_proxy_enabled():
        hosts |= {h.strip().lower() for h in (request.headers.get("x-forwarded-host") or "").split(",") if h.strip()}
    hosts = {h[: -len(default_port)] if default_port and h.endswith(default_port) else h for h in hosts if h}
    return netloc not in hosts


def _is_admin_request() -> bool:
    """主人直连：UI 会话，或主令牌（不是按 agent 发的钥匙）。"""
    from ducky import agent_keys
    from ducky.security.auth import current_request_auth_kind
    kind = current_request_auth_kind()
    if kind == "session":
        return True
    cur = agent_keys._CURRENT.get()
    return kind == "bearer" and bool(cur) and bool(cur.get("main"))


def _post_auth_refusal(request: "Request"):
    """已认证之后的两道附加拒绝（f0.8+）：R3-13 cookie 写的同源校验；R3-14 文档只给主人。"""
    from ducky.security.auth import current_request_auth_kind
    if current_request_auth_kind() == "session" and _cookie_write_not_same_origin(request):
        logger.warning("🛑 [Security] 拒绝非同源的 UI 会话写请求：Origin=%s Sec-Fetch-Site=%s 路径=%s",
                       request.headers.get("origin", "-"), request.headers.get("sec-fetch-site", "-"),
                       request.url.path)
        return JSONResponse(status_code=403, content={
            "detail": "Console-session writes must come from the console's own origin.",
            "code": "cross_origin_session_write_refused"})
    if request.url.path in _DOC_PATHS and not _is_admin_request():
        # R3-14：接口清单（/docs /openapi.json）只给 UI 会话与主令牌；agent 钥匙看不到
        return JSONResponse(status_code=403, content={
            "detail": "API docs are available to the console session and the main token only.",
            "code": "docs_admin_only"})
    return None


def _trust_proxy_enabled() -> bool:
    """部署方是否显式声明「本实例在可信反代之后」。"""
    return os.environ.get("AIDUMEI_TRUST_PROXY", "").strip().lower() in {
        "1", "true", "yes", "on",
    }


def _enforce_single_process_policy() -> None:
    """单进程契约由机器守住（v20.3.2-beta · 外审 P0-3 / 跨进程风险面）。

    会话表（auth._sessions）、限流窗口（rate_guard._WINDOWS）、用量计数
    （mem0_runtime._llm_usage）全是**进程内**状态 —— 这是**显式设计决策**
    （注释与文档都写明单进程部署形态，跨实例属 v21 路线图）。

    问题不在这个决策，而在**没有任何东西阻止别人违反它**：
    本文件的 docstring 就把 `gunicorn -k uvicorn.workers.UvicornWorker` 列为起法之一，
    任何人顺手加 `--workers 4` 就会得到「登录时好时坏 / 限流失灵 / 后台线程 ×4」——
    而这类症状**最难排查**，且服务看起来完全正常。

    所以判据落在**启动期**、拒绝启动、并给出唯一合规路径：与既有公网熔断同一风格。
    两个来源都看（更严的赢）：WEB_CONCURRENCY 环境变量，以及 argv 里的
    `--workers/-w`（gunicorn 的 worker 数是它自己的命令行参数，环境变量看不见它
    —— 与 F-01 的 `--host` 同理）。
    """
    def _fail(source: str, value: str) -> None:
        raise RuntimeError(
            f"🛑 [Security] 拒绝以多进程形态启动（{source}={value}）。"
            "本服务是**单进程契约**：会话表、限流窗口、用量计数均为进程内状态，"
            "多 worker 下会表现为「登录时好时坏、限流失灵」这类难排查症状。"
            "唯一合规路径：把 worker 数设为 1（WEB_CONCURRENCY=1，或去掉 --workers），"
            "后台任务如需独立进程请用 aidumem-worker 形态。"
            "跨实例共享状态属 v21 多副本路线图。"
        )

    raw = os.environ.get("WEB_CONCURRENCY", "").strip()
    if raw:
        try:
            if int(raw) > 1:
                _fail("WEB_CONCURRENCY", raw)
        except ValueError:
            # 非法值不静默：配置面出错必须出声（与 env_config 同纪律）
            logger.warning("WEB_CONCURRENCY 值非法（%r），按单进程处理", raw)

    argv = list(sys.argv[1:])
    for i, tok in enumerate(argv):
        val = None
        if tok in ("--workers", "-w") and i + 1 < len(argv):
            val = argv[i + 1]
        elif tok.startswith("--workers="):
            val = tok.split("=", 1)[1]
        if val is None:
            continue
        try:
            if int(val) > 1:
                _fail("--workers", val)
        except ValueError:
            continue


def _env_bind_host() -> str:
    """AIDUMEM_HOST 的唯一解析（f0.8 · 外审 deepseek F-22）：空/纯空白 → 127.0.0.1。

    此前 `_detect_bind_host()` 把空值当回环放行门禁，`main()` 却把空串原样交给 uvicorn
    （空 host = 绑全部网卡）——门禁以为在回环，服务已对外。两处现在都读这一个函数。"""
    return (os.environ.get("AIDUMEM_HOST") or "").strip() or "127.0.0.1"


def _detect_bind_host() -> str:
    """推断**实际**监听地址（v20.2.4 · 外审 F-01）。

    `AIDUMEM_HOST` 只管我们自己的入口。而外审门槛点名的那一种起法 ——
    `uvicorn api_server:app --host 0.0.0.0` —— 里的 host 是 **uvicorn 自己的
    命令行参数**，环境变量根本看不到它。所以这里也扫一眼 argv：土，但它是
    这条路上唯一能在「开始接收请求之前」拿到的事实，而门槛要的正是这个时点。

    两个来源里**非回环的那个优先**（更严的赢）。argv 也看不到的情况
    （编程式 `uvicorn.Server(config)`）由请求期那道兜底。
    """
    env_host = _env_bind_host()
    argv_host = ""
    argv = list(sys.argv or [])
    for i, a in enumerate(argv):
        if a in ("--host", "-b", "--bind") and i + 1 < len(argv):
            argv_host = argv[i + 1].rsplit(":", 1)[0].strip().strip("[]")
        elif a.startswith("--host="):
            argv_host = a.split("=", 1)[1].strip()
        elif a.startswith("--bind="):
            argv_host = a.split("=", 1)[1].rsplit(":", 1)[0].strip()
    for h in (argv_host, env_host):
        if h and h not in _LOOPBACK_HOSTS:
            return h
    return env_host or argv_host or "127.0.0.1"


def _enforce_public_binding_policy() -> None:
    """公网裸奔门禁 —— **必须在开始接收请求之前执行**（v20.2.4 · 外审 F-01）。

    此前这段只长在 `main()` 里。而上面 `_lifespan` 的 docstring 讲的正是这条路：
    `uvicorn api_server:app` / gunicorn 把 `app` 当 ASGI 对象导入，**完全不经过
    main()**。v20 的 P1-6 认识到这条路会让后台能力静默缺席、把它们挪进了
    lifespan，却漏了同一条路上的**安全门禁** —— 于是无 token + 非回环时整个
    路由表对公网敞开，而鉴权 middleware 按设计是「没配凭据就整体关闭」。

    判据用 `_auth_enabled()`（token 或 UI 口令任一）而不是原来的 `_api_token()`：
    只配了 UI 口令的部署，middleware 其实会鉴权，原判据会误报。
    """
    # v20.4.0（三方审计 P1-10 · 动态审计 🟡-2）：TRUST_PROXY=1 且无凭据时，
    # 三道防线（反代痕迹 503 / Host 校验 / 浏览器跨站写拒绝）被一个开关
    # 同时旁路，而此前只有一行 INFO。动态审计实测该形态下带 XFF 的请求可打通
    # /facts、/config、/delete_all。这里升 WARNING 并把三项让渡逐条点名 ——
    # 部署方必须知道自己让渡的不是一项。
    if _trust_proxy_enabled() and not _auth_enabled():
        logger.warning(
            "⚠️ [Security] AIDUMEI_TRUST_PROXY=1 且未配置任何凭据：以下三道防线"
            "已全部让渡给反代 —— ① 带反代痕迹（X-Forwarded-For 等）的请求不再 503；"
            "② Host 校验（防 DNS rebinding）关闭；③ 浏览器跨站写拒绝关闭。"
            "反代必须自行承担鉴权与 Host 路由，强烈建议改配 AIDUMEM_API_TOKEN。")
    host = _detect_bind_host()
    if host in ("127.0.0.1", "localhost", "::1"):
        return
    if _auth_enabled():
        return
    if os.environ.get("AIDUMEM_ALLOW_INSECURE_PUBLIC", "0").lower() in {"1", "true", "yes"}:
        # v22.0（雷霆审计 A7 · GLM F-04）：逃逸门组合闸。
        # INSECURE_PUBLIC=1 ∧ TRUST_PROXY=1 ∧ 无凭据 = 公网裸奔且三道防线
        # （反代痕迹 503 / Host 校验 / 跨站写拒绝）全部旁路——单个逃生舱是
        # 部署方的知情选择，**组合态**此前只剩一行 WARNING，没有任何机制
        # 阻止「顺手多开一个开关」把实例推上公网。组合态必须二次显式确认：
        # 确认变量的值必须逐字等于实际监听地址（防复制粘贴的 1/true 蒙混）。
        #
        # f0.3：**单开** INSECURE_PUBLIC 也改为同一道二次确认。单开的后果已经是
        # 「任何能连到这台机器的人都能读写全部记忆」，此前却只打一行 WARNING 就
        # 放行 —— 一个 `=1` 就把实例推上公网，比组合态少的只是反代那三道防线。
        # 两种形态共用 AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH，值必须逐字等于监听地址。
        confirm = os.environ.get("AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH", "").strip()
        combo = _trust_proxy_enabled()
        if confirm != host:
            if combo:
                logger.critical(
                    "🛑 [Security Fatal] 拒绝启动：AIDUMEM_ALLOW_INSECURE_PUBLIC=1 与 "
                    "AIDUMEI_TRUST_PROXY=1 同时开启且未配置任何凭据——公网监听 '%s' "
                    "且全部请求期防线让渡给反代。若确属知情部署，请设 "
                    "AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH=%s（值必须逐字等于监听地址）；"
                    "否则请配置 AIDUMEM_API_TOKEN 或关掉其中一个开关。", host, host,
                )
                raise RuntimeError(
                    "Fatal Security Policy: INSECURE_PUBLIC + TRUST_PROXY without any "
                    f"credential on '{host}' requires AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH='{host}'."
                )
            logger.critical(
                "🛑 [Security Fatal] 拒绝启动：AIDUMEM_ALLOW_INSECURE_PUBLIC=1 且未配置任何凭据"
                "——公网监听 '%s'，任何能连到本机的人都能读写全部记忆。若确属知情部署，"
                "请再设 AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH=%s（值必须逐字等于监听地址）；"
                "否则请配置 AIDUMEM_API_TOKEN。", host, host,
            )
            raise RuntimeError(
                "Fatal Security Policy: INSECURE_PUBLIC without any credential on "
                f"'{host}' requires AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH='{host}'."
            )
        if combo:
            logger.critical(
                "🛑 [Security] 组合逃逸门已二次确认（AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH=%s）："
                "公网裸奔 + 三道请求期防线全部让渡给反代，后果自负。", host,
            )
        else:
            logger.critical(
                "🛑 [Security] 无凭据公网监听已二次确认（AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH=%s）："
                "任何能连到本机的人都能读写全部记忆，后果自负。", host,
            )
        return
    logger.critical(
        "🛑 [Security Fatal] 拒绝启动：监听地址为公网/非回环 '%s' 且未配置任何凭据"
        "（AIDUMEM_API_TOKEN 或 UI 口令）。请配置凭据；确属知情的无凭据公网部署，须同时设 "
        "AIDUMEM_ALLOW_INSECURE_PUBLIC=1 与 AIDUMEI_I_CONFIRM_PUBLIC_NO_AUTH=%s。", host, host,
    )
    raise RuntimeError(
        f"Fatal Security Policy: binding to '{host}' without any credential is prohibited."
    )


async def _limited_call(request: Request, call_next):
    """f0.8 ③-6：放行之后、进路由之前——limit 类参数统一钳制 + 读路径按钥匙限流。"""
    from ducky.request_limits import apply as _apply_limits
    retry = _apply_limits(request)
    if retry is not None:
        return JSONResponse(status_code=429, headers={"Retry-After": str(retry)},
                            content={"detail": "read rate limit exceeded (AIDUMEI_RATE_READ_PER_MIN)",
                                     "code": "read_rate_limited", "retry_after": retry})
    return await call_next(request)


@app.middleware("http")
async def _require_credentials(request: Request, call_next):
    if not _auth_enabled():
        # v20.2.4（外审 F-01 第二道）：无凭据的实例**只服务回环**。
        #
        # 配置面的检查（_enforce_public_binding_policy）拦不住一种情况：
        # `uvicorn --host 0.0.0.0` 里的 host 是 **uvicorn 自己的命令行参数**，
        # 环境变量 AIDUMEM_HOST 根本看不到它。所以这一道把判据换成**运行时
        # 事实**——请求是谁发来的。不管服务监听在哪，没配凭据就只招待本机。
        _proxied = _request_via_proxy(request.headers) and not _trust_proxy_enabled()
        if not _client_is_loopback(request) or _proxied:
            logger.critical(
                "🛑 [Security] 拒绝非回环请求：本实例未配置任何凭据"
                "（AIDUMEM_API_TOKEN 或 UI 口令），只服务回环。来源=%s 路径=%s",
                getattr(getattr(request, "client", None), "host", "?"),
                request.url.path,
            )
            return JSONResponse(
                status_code=503,
                content={
                    "detail": (
                        "This instance has no credential configured and therefore "
                        "serves direct loopback only. Set AIDUMEM_API_TOKEN (or a UI "
                        "password) before exposing it beyond localhost. "
                        "If this request arrived through a trusted reverse proxy on "
                        "the same host, the proxy hop is what was refused: configure "
                        "a credential, or set AIDUMEI_TRUST_PROXY=1 to declare the "
                        "proxy trusted."
                    ),
                    "code": "no_credential_public_access_denied",
                },
            )
        if not _host_header_allowed(request):
            logger.warning(
                "🛑 [Security] 拒绝 Host=%s：无凭据实例只接受回环/受信主机名（防 DNS rebinding）。"
                "反代部署请配置凭据或设 AIDUMEI_TRUST_PROXY=1；自定义域名加入 AIDUMEI_TRUSTED_HOSTS。",
                request.headers.get("host", "?"))
            return JSONResponse(
                status_code=421,
                content={
                    "detail": (
                        "Host header is not a trusted name for this credential-less instance "
                        "(DNS rebinding protection). Use 127.0.0.1/localhost, add the name to "
                        "AIDUMEI_TRUSTED_HOSTS, or configure a credential."
                    ),
                    "code": "host_not_allowed",
                },
            )
        if _browser_cross_site_write(request):
            logger.warning("🛑 [Security] 拒绝跨站浏览器写请求：Origin=%s Sec-Fetch-Site=%s 路径=%s",
                           request.headers.get("origin", "-"), request.headers.get("sec-fetch-site", "-"),
                           request.url.path)
            return JSONResponse(
                status_code=403,
                content={"detail": "Cross-site browser write refused on a credential-less instance.",
                         "code": "cross_site_write_refused"},
            )
        return await _limited_call(request, call_next)
    if _is_public_path(request.url.path):
        return await call_next(request)
    if _request_authorized(request):
        refused = _post_auth_refusal(request)
        if refused is not None:
            return refused
        return await _limited_call(request, call_next)
    return JSONResponse(
        {
            "error": "unauthorized",
            "detail": "Missing or invalid credentials: log in to the console "
                      "or send Authorization: Bearer <AIDUMEM_API_TOKEN>",
        },
        status_code=401,
    )


# v20.4.0-alpha（P0-1）：全局 body 硬顶。/add 序列化上限 64 KiB 的 16 倍余量，
# 正常调用远够不着；超过的一定是异常载荷或客户端 bug。
_MAX_BODY_BYTES = 1024 * 1024


@app.middleware("http")
async def _payload_too_large_guard(request: Request, call_next):
    """全局请求体硬顶（v20.4.0-alpha · P0-1；f0.6 W-26 补 chunked）。

    字段级上限（api_models 的 max_length）挡得住「声明了的字段」，挡不住
    「整个请求体就是垃圾」：extra="allow" 的兼容设计意味着未声明字段也能进
    JSON 解析 —— 10 MB 的匿名 JSON 照样吃内存。所以在鉴权之前硬顶 1 MiB。

    f0.6 W-26（cue 审计 A11）：无 Content-Length 的 chunked 请求此前**不拦**
    （旧注释把职责推给反代）——公开入口匿名可发大体积 chunked 流耗资源。
    现在用 ASGI receive 包装器**边转发边累计实际字节**，超限立即 413 且
    不进 JSON 解析；内存占用上限=已收到的字节本身，不先读全再检查。
    Content-Length 声明超限的分支保持原样（不读 body 直接拒）。

    注册位置（源码序）：鉴权 → 本闸 → 安全头 → 计数；运行层次即
    计数（最外）→ 安全头 → 本闸 → 鉴权 → 路由 —— 被挡的 413 照样计数、
    照样带安全头（v20.3.2 中间件顺序教训：自己加的头要盖到自己加的拒绝分支）。
    """
    raw = request.headers.get("content-length", "")
    if raw.isdigit() and int(raw) > _MAX_BODY_BYTES:
        logger.warning("🛑 [Security] 拒绝超大请求体：%s 字节 > %d，路径=%s",
                       raw, _MAX_BODY_BYTES, request.url.path)
        return JSONResponse(
            status_code=413,
            content={
                "detail": f"Request body too large: {raw} bytes exceeds the "
                          f"{_MAX_BODY_BYTES}-byte hard cap. Split the payload "
                          "or send it in multiple calls.",
                "code": "payload_too_large",
            },
        )

    # W-26：chunked/无长度声明 → 分块消费 stream 累计实际字节。限额内把
    # 结果塞 request._body 缓存——BaseHTTPMiddleware 的 _CachedRequest 会
    # 把它原样转交下游（state-3 分支），fastapi 的 body 读取命中缓存不再
    # 碰原始 receive；超限立即 413，不再拉取后续块（已读上限=1MiB+一块，
    # 内存可控），response 先行后连接由服务器层收尾。
    if not raw:
        from starlette.requests import ClientDisconnect as _W26Disconnect
        body_buf = bytearray()
        try:
            async for chunk in request.stream():
                body_buf += chunk
                if len(body_buf) > _MAX_BODY_BYTES:
                    logger.warning(
                        "🛑 [Security] chunked 请求体超限：%d 字节 > %d（累计），路径=%s",
                        len(body_buf), _MAX_BODY_BYTES, request.url.path)
                    return JSONResponse(
                        status_code=413,
                        content={
                            "detail": f"Request body too large (streamed, no "
                                      f"Content-Length): exceeds the "
                                      f"{_MAX_BODY_BYTES}-byte hard cap.",
                            "code": "payload_too_large",
                        },
                    )
        except _W26Disconnect:
            return JSONResponse(
                status_code=400,
                content={"detail": "Client disconnected mid-body", "code": "client_disconnected"},
            )
        request._body = bytes(body_buf)  # 交下游（含空 body——GET 等无体请求也走这）
    return await call_next(request)


@app.middleware("http")
async def _security_headers(request: Request, call_next):
    """浏览器侧兜底（v20.3.2-beta · 外审 F-4 / DeepSeek P1-6）。

    本服务自带 Web 控制台且常被放在反代之后，而全仓此前**一个安全头都没有**
    —— 与它在鉴权、注入、交付硬化上的完成度极不相称，属于低垂果实。

    CSP 演进：v20.3.2-beta 先宽后紧（当时前端有 inline style，`style-src`
    带 'unsafe-inline' 容纳，登记收紧为 v20.4 候选）；v20.4.0-alpha（P2-11，
    外审 Qwen #3）六块蜂窝砖坐标与更新徽章显隐已收进 `frontend/css/style.css`
    （tests/test_v20_4_misc_guards.py 守卫防回流），`style-src` 收紧为 'self'。
    JS 的 `el.style.x = ...` 是 CSSOM 写操作，不受 style-src 管辖，不受影响。

    `script-src` 同理，但这一格是**生产替我们发现的**：2026-09-07 在生产机
    上把它加回 'unsafe-inline'（热修 `3219f8c4e`，未推远端），因为
    `frontend/index.html` 与 `login.html` 各有一段 inline `<script>` 被打死——
    登录页的背景/版本徽章/口令提示与主控台的未登录跳转全不动。P2-11 当时只
    盯着自己刚改的 style 那一格，守卫射程没盖到 script（SOP 铁律 12）。
    v20.4.0-alpha 并回的做法是**收文件而不是放 CSP**：两段脚本逐字搬进
    `frontend/js/auth-guard.js` 与 `frontend/js/login.js`，`script-src` 保持
    'self'。tests/test_v20_4_prod_hotfix_reconcile.py 钉三面：HTML 无 inline、
    实测响应头无 'unsafe-inline'、被引用的 src 真能服务出来。
    """
    response = await call_next(request)
    # 已存在的值不覆盖（反代可能已经加过，部署方的显式设置优先）
    for name, value in (
        ("X-Content-Type-Options", "nosniff"),
        ("X-Frame-Options", "DENY"),
        ("Referrer-Policy", "no-referrer"),
        ("Content-Security-Policy",
         "default-src 'self'; "
         "img-src 'self' data: https:; "
         "style-src 'self'; "
         "script-src 'self'; "
         "connect-src 'self'; "
         "frame-ancestors 'none'; "
         "base-uri 'self'"),
    ):
        if name not in response.headers:
            response.headers[name] = value
    return response


@app.middleware("http")
async def _record_http_outcome(request: Request, call_next):
    """把每个响应的状态码记进进程内计数器（v20 · P1-9）。

    放在鉴权 middleware **外面**是刻意的：被门禁挡掉的 401 也是一次真实结局，
    事故里「凭据链断了导致满屏 401」正是要看见的形态之一。
    未处理异常按 500 计入后再抛，否则「打挂了」这一类会从统计里凭空消失。

    🔴v20.3.2 正式版（外审 GLM F-2 / Codex F-08）：**这段注释曾与事实相反。**
    Starlette 是「后注册者在更外层」，而本函数原先注册在**最前**（第 289 行），
    鉴权注册在最后 —— 于是鉴权成了最外层，它直接 return 的 401/503 **从不经过**
    这里，也不经过安全头中间件：实测 401+200 两个请求只计了 1 个，401 响应无 CSP。
    我上一轮加的四个安全头，没盖到我上一轮加的拒绝分支。
    现在三个中间件按「鉴权 → 安全头 → 计数」的**源码顺序**注册，运行时层次即
    「计数（最外）→ 安全头 → 鉴权（最内）→ 路由」。tests/test_v20_3_2_release_real_asgi.py
    用真 uvicorn + 真 socket 钉行为，用 AST 钉顺序 —— 两条都要。
    """
    # P1-9（收口后修正）：给本次请求盖借用上下文戳。contextvars 随 anyio 线程池传播到同步路由的
    # 工作线程，utils 据此把「同一请求嵌套借连接」与「上一个请求留下悬挂事务」分开 —— 后者才计数。
    from ducky.utils import BORROW_CONTEXT as _BORROW_CONTEXT, rollback_dangling_for_context as _tx_janitor
    _ctx_token = _BORROW_CONTEXT.set(_uuid.uuid4().hex)
    from ducky import http_metrics
    try:
        response = await call_next(request)
    except Exception:
        http_metrics.record(500)
        raise
    finally:
        # f0.6 W-04（hark 审计 P0-2）：请求级事务清道夫。此前只检测不回滚——
        # 请求 X 执行了 INSERT/UPDATE 但 commit 前抛异常，同线程下一个无关
        # 请求 Y 的任何一次 commit 会把 X 的半截写一起提交（AST 扫描 69 处
        # 写路径无回滚）。现在请求收尾时按上下文戳回滚「本请求开启但未提交」
        # 的事务；嵌套借用不受影响（戳相同不回滚）。
        # 注意：本中间件跑在事件循环线程，这里只够得着 async 路由的连接；同步路由
        # （add/search/update/delete 等）在工作线程，由 utils._get_thread_conn 在同线程
        # 下一次借出时回滚。
        _tx_janitor(_BORROW_CONTEXT.get())
        _BORROW_CONTEXT.reset(_ctx_token)
        # f0.8+ C：任何写请求（成败都算，失败也可能半写）作废写入者的 /search 缓存。
        try:
            from ducky import search_cache as _search_cache
            _search_cache.invalidate_for_http(
                request.method, request.url.path, request.query_params.get("user_id"),
                getattr(request, "_body", None))
        except (ImportError, OSError, ValueError, TypeError, RuntimeError) as _sc_exc:
            logger.debug("search cache invalidation fell back to clear: %s", _sc_exc)
            try:
                from ducky import search_cache as _search_cache
                _search_cache.clear()
            except (ImportError, OSError, ValueError, TypeError, RuntimeError):
                pass
    http_metrics.record(response.status_code)
    return response

# ── UI 登录与会话（🔴P0-1 v19.4.1）───────────────────────────────────────
# /login 校验口令后**服务端签发会话**（HttpOnly cookie），不再只在浏览器
# sessionStorage 打个标记 —— 后者对 REST 接口毫无约束力。
# 口令统一走 ducky.security.auth：PBKDF2 存储、旧格式自动升级（P2-2）。


def _ensure_ui_password() -> None:
    """初始化控制台口令：既无环境变量又无哈希文件时，自动生成强随机口令。

    v19.4.1：哈希改用 PBKDF2-HMAC-SHA256（200k 轮），文件权限收紧 0600。
    """
    from ducky.security.auth import (
        hash_password,
        initial_password_path,
        password_hash_path,
        write_initial_password,
        write_password_hash,
    )
    import secrets

    env_pwd = os.environ.get("AIDUMEM_UI_PASSWORD", "").strip()
    hash_file = password_hash_path()
    if env_pwd:
        # f0.7.1：部署方显式设了环境变量口令 → 作废自动生成的口令（哈希 + 明文文件）。
        # 原先两套口令同时有效，用户以为「已换密」，旧的自动口令（明文还躺在
        # data/.ui_initial_password）照样能登录。只动 source=auto 的哈希；控制台改过的
        # （source=user）不碰。
        from ducky.security.auth import env_password_shadowed, retire_auto_password
        retire_auto_password()
        if env_password_shadowed():
            logger.warning("🔐 已设置 AIDUMEM_UI_PASSWORD，但控制台改过口令：环境变量口令不再有效（f0.8）。"
                           "要用它重置，删掉 %s 再重启", hash_file)
        return
    if not os.path.exists(hash_file):
        gen_pwd = secrets.token_urlsafe(12)

        # v20 · P1-7：明文口令**只落 0600 文件，不进日志**。
        # 原先这里是 logger.warning(… 初始口令: %s …, gen_pwd, …)，于是每次
        # 首启都把一条可直接登录的凭据写进 journald 和 logrotate 归档 ——
        # 日志的读者范围永远大于口令的读者范围。
        #
        # 顺序是「先落明文、再落哈希」，不是反过来：
        #   · 明文先落 → 若哈希落盘失败，最坏是留下一份没生效的口令文件（噪声）；
        #   · 哈希先落 → 若明文落盘失败，口令**已经生效但没人知道它是什么**，
        #     控制台当场锁死。两种失败都要处理，但只有后者会锁死用户。
        if not write_initial_password(gen_pwd):
            logger.error(
                "🔐 [安全加固] 自动生成了控制台初始口令，但写 %s 失败 —— "
                "本次**不落哈希**，下次启动重试（避免口令已生效却无人知晓而锁死控制台）。"
                "按纪律这里绝不把明文打进日志兜底：请显式设置 AIDUMEM_UI_PASSWORD 后重启。",
                initial_password_path(),
            )
            return

        # source="auto"：自动生成的口令只守 UI 登录，不启用 API 门禁 ——
        # 否则存量部署（hermes 插件 / MCP / cron 全走回环不带凭据）
        # 会在升级瞬间集体 401。详见 ducky/security/auth.py 的 provenance 说明。
        if write_password_hash(hash_password(gen_pwd), source="auto"):
            logger.warning(
                "🔐 [安全加固] 未配置 AIDUMEM_UI_PASSWORD，已自动生成随机控制台初始口令。"
                "明文只写在 %s（权限 0600，日志中不留明文），PBKDF2 哈希在 %s。"
                "请取用后通过控制台改密，并删除该明文文件。",
                initial_password_path(),
                hash_file,
            )
        else:
            logger.error(
                "🔐 [安全加固] 初始口令明文已落 %s，但哈希落盘失败 —— 该口令并未生效。"
                "请显式设置 AIDUMEM_UI_PASSWORD 后重启，并删除那个明文文件。",
                initial_password_path(),
            )


_ensure_ui_password()


def _register_login(route_app: FastAPI) -> None:
    @route_app.post("/login", include_in_schema=False)
    async def ui_login(request: Request):
        from ducky.security.auth import (
            SESSION_COOKIE_NAME,
            check_ui_password,
            create_session,
        )

        # 🚪 v20.2.3（外审 M-1）：爆破护栏。**先查后验** —— 超限直接 429，
        # 连 PBKDF2 都不跑（省 100ms，也不给攻击者旁路信号）。
        from ducky.rate_guard import login_locked, record_login_failure
        _ip = (request.client.host if request.client else "unknown")
        _retry = login_locked(_ip)
        if _retry is not None:
            logger.warning("🚪 登录失败次数超限，暂时拒绝（%ss 后重试）", _retry)
            return JSONResponse(
                {"success": False,
                 "message": f"登录失败次数过多，请 {_retry} 秒后重试 / "
                            f"Too many failed attempts, retry in {_retry}s"},
                status_code=429, headers={"Retry-After": str(_retry)},
            )

        try:
            payload = await request.json()
        except Exception:
            payload = {}
        if not isinstance(payload, dict):
            # f0.7.6（外审 R2-16）：数组 / 字符串 / null 是合法 JSON 但不是对象——400，并计一次失败
            record_login_failure(_ip)
            return JSONResponse(
                {"success": False, "message": "请求体必须是 JSON 对象 / body must be a JSON object"},
                status_code=400,
            )
        given = payload.get("password")
        if not isinstance(given, str) or not given:
            # 空口令也计数：否则攻击者可以拿它当免费的「探测门是否开着」。
            record_login_failure(_ip)
            return JSONResponse(
                {"success": False, "message": "密码不能为空或格式错误"}, status_code=401
            )

        # v20.3.2（外审 Gemini P1-1，实测 15.4ms）：PBKDF2 200k 轮是纯 CPU，
        # 在 async def 里同步跑会冻住整个事件循环；丢进线程池。
        import asyncio as _asyncio
        if not await _asyncio.to_thread(check_ui_password, given):
            _n = record_login_failure(_ip)
            logger.warning("🚪 UI 登录失败（密码错误，本窗口第 %d 次）", _n)
            return JSONResponse(
                {"success": False, "message": "访问密码错误 / Wrong password"},
                status_code=401,
            )

        # 🔴P0-1：签发服务端会话，浏览器后续请求靠它自证身份。
        # HttpOnly 阻断 JS 读取（防 XSS 窃取）；SameSite=Lax 防基础 CSRF。
        # secure 标志按部署形态决定：HTTPS 反代后设 AIDUMEM_COOKIE_SECURE=1。
        token, ttl = create_session()
        secure_flag = os.environ.get("AIDUMEM_COOKIE_SECURE", "0").strip().lower() in {
            "1", "true", "yes",
        }
        # v20.2.3（外审 L-2）：默认关是对的（回环部署走 HTTP，开了 cookie
        # 根本发不出去）。但**反代之后还没开**就是真漏洞：会话 cookie 会
        # 随一次降级到 HTTP 的请求裸奔出去。请求头里带着 HTTPS 痕迹却没开
        # secure —— 这种「配置与部署形态不匹配」必须出声，不许静默。
        if not secure_flag and request.headers.get("x-forwarded-proto", "").lower() == "https":
            logger.warning(
                "🔓 检测到 HTTPS 反代（X-Forwarded-Proto: https）但 "
                "AIDUMEM_COOKIE_SECURE 未开启 —— 会话 cookie 缺少 secure 标志，"
                "遇到降级到 HTTP 的请求会明文外泄。请设 AIDUMEM_COOKIE_SECURE=1。"
            )
        resp = JSONResponse({"success": True, "expires_in": ttl})
        resp.set_cookie(
            key=SESSION_COOKIE_NAME,
            value=token,
            max_age=ttl,
            httponly=True,
            samesite="lax",
            secure=secure_flag,
            path="/",
        )
        logger.info("🚪 UI 登录成功（已签发会话，有效期 %d 秒）", ttl)
        return resp


    @route_app.get("/login/hint", include_in_schema=False)
    async def ui_login_hint():
        # 安全修复：任何情况下都不把密码明文吐给前端。
        # 未设置 AIDUMEM_UI_PASSWORD 时启动日志会提醒部署方自行设置。
        return {"hint": None}


_register_login(_api_alias)   # UI 登录只走 /api/login

# 注入版本信息到 health 端点（唯一真相源）
set_version_info(SERVICE_VERSION, CODENAME, CODENAME_ZH)

_background_started = False
_background_lock = threading.Lock()
_BACKGROUND_THREADS: list = []   # P2-20（v20.3.2 正式版）：已启动的循环线程，lifespan 停机时 join

_BACKGROUND_LOOPS = {
    "consolidation": _background_consolidation_loop,
    "scene_cluster": _background_scene_cluster_loop,
    "auto_memory": auto_memory_background_loop,
    "auto_expire": _auto_expire_loop,
    "autodream": autodream_background_loop,
    "evolve_mem": evolve_background_loop,
    "reflect": reflect_background_loop,
    "consolidate_nightly": consolidate_background_loop,
}


def _start_background() -> None:
    """幂等启动后台循环并初始化存储。"""
    global _background_started
    with _background_lock:
        if _background_started:
            return
        _background_started = True

    # 核心表建表必须最先做：facts/entities 是所有功能的地基，
    # 全新克隆时它们还不存在（v14 Aegis 起由代码建，不再依赖手工部署）。
    ensure_core_schema()

    try:
        get_memory()
        logger.info("🧠 记忆内核预热完成")
    except Exception as exc:
        logger.warning(f"⚠️ mem0 预热失败（主服务仍会启动）: {exc}")

    _init_text_fts()
    # 📼 v19.4.0 明镜工程 Phase 1: Verbatim Vault 原文保真层建表（幂等，失败降级）
    try:
        from ducky.verbatim_vault import ensure_verbatim_schema
        ensure_verbatim_schema()
    except Exception as _vs:
        logger.warning(f"📼 Verbatim Vault 建表跳过（主服务仍会启动）: {_vs}")
    init_core_memory()
    # 启动 WAL 对账与自愈（v19.2.0 P0-DATA）
    try:
        from ducky.wal_engine import reconcile_startup
        _rec_report = reconcile_startup()
        if _rec_report.get("recovered", 0) > 0:
            logger.info("🔧 [WAL Reconcile] 成功自愈恢复 %d 条挂起事务", _rec_report["recovered"])
        if _rec_report.get("remaining") != 0 or _rec_report.get("reconciliation_paused"):
            logger.warning("WAL recovery needs attention: remaining=%s integrity=%s",
                           _rec_report.get("remaining"), _rec_report.get("wal_integrity", "unknown"))
    except Exception as _re:
        logger.warning(f"⚠️ WAL 启动对账异常: {_re}")

    try:
        from ducky.gear import ensure_half_open_probe_daemon
        ensure_half_open_probe_daemon()
    except Exception as _gear_probe_exc:
        logger.warning("⚠️ Gear 半开主动探测启动跳过: %s", _gear_probe_exc)

    # 启动自检：实体词表漏配是「静默故障」——闸门会把涉及自定义人名/
    # 项目代号的查询判成 no_signal 而零召回，不报错也不留痕。v15 起
    # 在启动日志里显式告警，别再让部署方自己去猜为什么查不到。
    try:
        from ducky.pipeline.memory_gate import entity_keywords_status
        _ek = entity_keywords_status()
        if _ek["configured"]:
            logger.info("🎯 相关性闸门实体词表已加载：%d 个词条", _ek["count"])
        else:
            logger.warning(
                "⚠️ %s 未配置 —— 涉及自定义人名/项目代号的查询会被闸门判为"
                " no_signal 并静默零召回。请参考 .env.example 配置后重启服务。",
                _ek["env_var"],
            )
    except Exception as exc:
        logger.warning("⚠️ 实体词表自检失败: %s", exc)

    for name, loop_fn in _BACKGROUND_LOOPS.items():
        thread = threading.Thread(
            target=loop_fn,
            daemon=True,
            name=f"aiduMEM-{name}",
        )
        thread.start()
        _BACKGROUND_THREADS.append(thread)  # P2-20：停机时逐个 join
        logger.info(f"▶ {name} 后台线程已启动")

    logger.info(
        "✅ aiduMEI v%s %s 后台线程已启动 (%s 个)",
        SERVICE_VERSION,
        DISPLAY_NAME,
        len(_BACKGROUND_LOOPS),
    )


def main():
    # 🔴v20.3.2 正式版（外审 Codex F-09）：**不在这里起后台。**
    # 原先第一行就是 _start_background()，而安全策略检查在 20 行之后 ——
    # 一个最终会因公网绑定/多 worker 被拒绝启动的进程，退出前已经建表、预热、
    # 起了线程。lifespan 是后台生命周期的唯一负责人（且有 _background_started
    # 幂等标志）；main() 只做纯配置校验，然后把进程交给 uvicorn。
    host = _env_bind_host()
    # v20.2.3（外审 M-2）：端口曾是裸 int() —— 写错一个字符服务直接起不来，
    # 且报错是 ValueError 而非「端口配置无效」。回退默认 + 点名出声。
    # or 链保留原语义（新键优先、空串落到旧键），但错误要点名到**真正
    # 提供了这个值的那个 env**，不许张冠李戴。
    from ducky.env_config import int_env as _int_env
    # f0.6+++ WP6：`python api_server.py --port N` 此前被静默忽略（只认 env）。
    # 命令行优先，缺省再走 env 链。
    import argparse
    _cli = argparse.ArgumentParser(add_help=False, allow_abbrev=False)
    _cli.add_argument("--port")
    _port_name, _port_raw = "--port", _cli.parse_known_args(sys.argv[1:])[0].port
    if not _port_raw:
        _port_name, _port_raw = "AIDUMEM_API_PORT", os.environ.get("AIDUMEM_API_PORT")
    if not _port_raw:
        _port_name, _port_raw = "MEM0_API_PORT", os.environ.get("MEM0_API_PORT")
    if not _port_raw:
        # 容器 PaaS（Dockhold / Heroku / Railway / Render / Fly / Cloud Run 等）在**运行时**
        # 才分配端口，且只通过 `PORT` 传入。部署方事先不知道值，写不进
        # AIDUMEM_API_PORT —— 于是服务监听 8767、平台往另一个端口转发，
        # 表现为「进程健康、外面 502」，日志里一句异常都没有。
        # 排在两个专名之后：本地与 compose 部署的行为完全不变。
        _port_name, _port_raw = "PORT", os.environ.get("PORT")
    port = _int_env(_port_name, 8767, minimum=1, maximum=65535, raw=_port_raw)
    if not _api_token():
        logger.warning(
            "⚠️ 未设置 AIDUMEM_API_TOKEN：REST 接口无鉴权。"
            "本机/回环使用可接受；对外部署请务必设置 token。"
        )
    env_pwd = os.environ.get("AIDUMEM_UI_PASSWORD", "").strip()
    if not env_pwd:
        logger.info("🔐 UI 登录使用 data/.ui_password_hash 安全凭据（或通过环境变量 AIDUMEM_UI_PASSWORD 配置）")
    # v20.2.4（F-01）：与 lifespan 共用**同一个**门禁实现 —— 两份拷贝
    # 早晚会长歪一份，而长歪的那份恰好是没人跑的那条路。
    _enforce_public_binding_policy()
    _enforce_single_process_policy()
    # 🔴v20.3.2 正式版（外审 GLM F-1）：**proxy_headers=False。**
    # uvicorn 默认 True → ProxyHeadersMiddleware 用 X-Forwarded-For 的**值**改写
    # request.client.host。于是 _client_is_loopback 在生产路径上读到的是伪造 IP，
    # AIDUMEI_TRUST_PROXY=1 也恒 503 —— 而 TestClient 不经过这层，9 条守卫全绿。
    # 全仓「不采信 XFF 值」的哲学，被一个默认启动参数推翻了。现在 client.host
    # 保持 TCP 对端语义；反代痕迹只看头**在不在**（_request_via_proxy），不读值。
    uvicorn.run(app, host=host, port=port, log_level="info", proxy_headers=False)


if __name__ == "__main__":
    main()