plugin.py

"""MaiBot 文件读取插件（File Reader）—— 入口文件

参考 astrbot_plugin_file_reader_pro 移植，架构对齐：
  文件 → 解析(file_parser) → 递归分块(chunker) → 向量化(llm.embed) → 会话级向量库(vector_store)
  提问时 → 语义检索 Top-K → 注入 LLM 上下文(before_model_request hook)

与 AstrBot 版的差异（MaiBot 无内置 FaissVecDB / RerankProvider）：
  - 向量库用 numpy 余弦相似度自实现（vector_store.py）
  - 注入走 `maisaka.replyer.before_model_request`，传 items 而非 messages
  - 文件内容从入站消息的 file/attachment 段提取（chat.receive.after_process）

铁律遵循：入口文件不写 `from __future__ import annotations`（见 skill runtime-gotchas 5.1）。
"""
import asyncio
import base64
import ipaddress
import json
import logging
import os
import re
import socket
import ssl
import sys
import tempfile
import time
import urllib.error
import urllib.request
from collections import deque
from pathlib import Path
from typing import Any, Callable, ClassVar, Iterable, Optional
from urllib.parse import urljoin, urlsplit, urlunsplit

# 补 sys.path，让同目录辅助模块可导入（runtime-gotchas 5.2）
_PLUGIN_DIR = str(Path(__file__).resolve().parent)
if _PLUGIN_DIR not in sys.path:
    sys.path.insert(0, _PLUGIN_DIR)

from maibot_sdk import (  # noqa: E402
    Command,
    EventHandler,
    Field,
    HookHandler,
    MaiBotPlugin,
    PluginConfigBase,
    Tool,
)
from maibot_sdk.types import (  # noqa: E402
    ErrorPolicy,
    EventType,
    HookMode,
    HookOrder,
    ToolParameterInfo,
    ToolParamType,
)

from chunker import RecursiveCharacterChunker  # noqa: E402
from file_parser import describe_supported_types, get_extension, read_any_file_to_text  # noqa: E402
from lexical_index import apply_relevance_gate, rrf_fuse  # noqa: E402
from vector_store import SessionStore  # noqa: E402

# v1.1.0：降级文本解析正则提到模块级预编译（每次 hook 现编译是无谓开销）
_RE_HINT_URL = re.compile(r"链接[:：]\s*(https?://\S+)")
# v1.3.1：不再用 `\s*[，,]?\s*$` 收尾。Host 1.3.16 起 `[文件]` 纯文本尾部会追加
# `类型: mime` / `file_ref: xxx`，旧式 `$` 锚定会让 `.+?` 一路吃到行尾，把
# `附件，大小: 15397，file_ref: 5a1b…` 整串当成文件名（真机实测：扩展名丢失 ⇒
# 解析器按扩展名分派直接拒收）。改为「名字不含逗号 + 可选大小组 + 其余尾巴一律忽略」。
# 代价：文件名自身含中文/英文逗号时会被截断（历史版本同样处理不了，非回归）。
_RE_HINT_NAME = re.compile(
    r"\[文件\]\s*(?P<name>[^，,]+?)(?:\s*[，,]\s*大小[:：]\s*(?P<size>\d+))?(?:\s*[，,].*)?$"
)


class _DownloadTooLargeError(Exception):
    """下载前/下载中即判定超限（v1.1.0）：不再把超限文件全量拉进内存。"""

    def __init__(self, size_bytes: float) -> None:
        super().__init__(f"download exceeds limit: {size_bytes:.0f} bytes")
        self.size_mb = size_bytes / 1024 / 1024


# ─── URL 下载校验（v1.2.0 SSRF 防护） ────────────────────────────
# 威胁模型：待下载的 URL 来自**消息正文**——适配器把 file 段降级成
# `[文件] x.docx，大小: 9547，链接: <URL>` 纯文本，`_parse_file_hints` 再把 URL 抠出来。
# 也就是说 URL 完全由发消息的人控制：任何群成员发一句带内网地址的"文件消息"，
# 旧版就会直接 GET 它并把响应体解析进 LLM 上下文。
#
# 关键设计点：**必须校验"解析后的 IP"，不能只校验域名**。
#   只挡域名会被这些写法绕过（都能表达 127.0.0.1）：
#     http://2130706433/         十进制 IP
#     http://0x7f000001/         十六进制 IP
#     http://127.1/              短写
#     http://127.0.0.1.nip.io/   通配 DNS 解析到回环
#     http://attacker.com/       自己的 A 记录指向 10.0.0.1
#   所以流程是「域名白名单 → 解析 → 逐个 IP 判公网」，两层都过才放行。
DOWNLOAD_POLICY_WHITELIST = "whitelist"
DOWNLOAD_POLICY_PUBLIC = "public"
DOWNLOAD_POLICY_OFF = "off"
DOWNLOAD_POLICIES: tuple[str, ...] = (
    DOWNLOAD_POLICY_WHITELIST,
    DOWNLOAD_POLICY_PUBLIC,
    DOWNLOAD_POLICY_OFF,
)

# 默认白名单：QQ 侧提供文件下载的域名（子域一并放行）。
# ftn.qq.com 覆盖 tjc-download / gzc-download / hzfile 等实际出现过的 CDN 主机。
DEFAULT_DOWNLOAD_HOSTS = "ftn.qq.com,qfile.qq.com,nt.qq.com.cn,qq.com,weixin.qq.com"


class _UrlRejectedError(Exception):
    """URL 未通过下载前校验。reason 是稳定标识，便于日志检索与测试断言。"""

    def __init__(self, reason: str, detail: str = "") -> None:
        super().__init__(f"{reason}: {detail}" if detail else reason)
        self.reason = reason
        self.detail = detail


def normalize_download_policy(raw: Any) -> str:
    """归一化策略名；无法识别时**退回最严格的 whitelist**（配置写错不应导致校验被绕过）。"""
    value = str(raw or "").strip().lower()
    return value if value in DOWNLOAD_POLICIES else DOWNLOAD_POLICY_WHITELIST


def parse_allowed_hosts(raw: Any) -> list[str]:
    """白名单解析：既接受逗号/分号/空白分隔的字符串（TOML 友好），也接受列表形态。"""
    if raw is None:
        return []
    if isinstance(raw, str):
        items: list[str] = re.split(r"[,;\s]+", raw)
    elif isinstance(raw, (list, tuple, set)):
        items = [str(x) for x in raw]
    else:
        items = [str(raw)]
    out: list[str] = []
    for item in items:
        item = item.strip()
        if item and item not in out:
            out.append(item)
    return out


def host_in_allowlist(host: str, allowed: Iterable[str]) -> bool:
    """后缀白名单匹配（含子域），且**边界安全**。

    边界安全指：`eviltfn.qq.com` 不能因为以 `ftn.qq.com` 结尾就放行——
    必须匹配 `host == suffix` 或 `host` 以 `.suffix` 结尾。
    顺手容忍用户把条目填成 `https://ftn.qq.com:443/` 这类形态。
    """
    target = str(host or "").strip().lower().rstrip(".")
    if not target:
        return False
    for entry in allowed:
        suffix = str(entry or "").strip().lower()
        suffix = suffix.split("://")[-1].split("/")[0]  # 容忍整条 URL
        suffix = suffix.split("@")[-1]  # 容忍内嵌凭据
        suffix = suffix.split(":")[0]  # 容忍带端口
        suffix = suffix.lstrip(".").rstrip(".")
        if not suffix:
            continue
        if target == suffix or target.endswith("." + suffix):
            return True
    return False


def resolve_host_ips(host: str, port: int) -> list[str]:
    """主机名 → 去重后的 IP 列表（含 IPv4/IPv6）。解析失败抛 OSError。"""
    ips: list[str] = []
    for info in socket.getaddrinfo(host, port or None, proto=socket.IPPROTO_TCP):
        addr = info[4][0]
        if addr and addr not in ips:
            ips.append(addr)
    return ips


def is_public_ip(ip_text: str) -> bool:
    """是否公网单播地址。回环/私网/链路本地/组播/保留/未指定一律 False。

    注意：**不能只信 `is_global`**。CPython 的 IPv4 `is_global` 只排除
    `_private_networks` 里那一段（0/8、10/8、127/8、169.254/16、172.16/12、
    192.0.0/29、192.0.2/24、192.168/16、198.18/15、198.51.100/24、
    203.0.113/24、240/4、255.255.255.255/32），**组播 224.0.0.0/4 不在其中**——
    实测 `is_global('224.0.0.1')` 返回 True。所以这里把 is_multicast /
    is_reserved / is_link_local / is_loopback / is_private / is_unspecified
    一并显式排掉，再叠加 is_global（本用例由测试抓出）。

    `169.254.169.254`（云元数据）落在 IPv4 链路本地段；`::1` 是 IPv6 回环；
    IPv4-mapped IPv6（`::ffff:127.0.0.1`）要拆成 IPv4 再判，否则会被当成公网。
    """
    try:
        ip = ipaddress.ip_address(str(ip_text).strip())
    except ValueError:
        return False
    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
        ip = ip.ipv4_mapped
    if (
        ip.is_multicast
        or ip.is_reserved
        or ip.is_link_local
        or ip.is_loopback
        or ip.is_private
        or ip.is_unspecified
    ):
        return False
    return bool(ip.is_global)


def validate_download_target(
    url: str,
    *,
    policy: str = DOWNLOAD_POLICY_WHITELIST,
    allowed_hosts: Iterable[str] = (),
    allow_private_hosts: bool = False,
    resolver: Optional[Callable[[str, int], list[str]]] = None,
    collect_ips: Optional[list[str]] = None,
) -> tuple[str, str]:
    """下载前校验目标 URL，返回 (原始 URL, 主机名)；不合规抛 _UrlRejectedError。

    校验顺序（任一不过即拒绝）：
      1. scheme 只能是 http/https（挡 `file://`、`gopher://`、`ftp://`）
      2. 禁止 URL 内嵌凭据（`http://user:pass@host/`）——容易被用来混淆主机
      3. policy=whitelist：主机必须命中白名单（白名单为空则拒绝一切）
      4. 解析主机 → **每个** IP 都必须是公网地址

    resolver 可注入（默认 system DNS），便于脱机单测模拟"域名合法但解析到内网"。

    collect_ips（v1.2.1）：传入 list 时，校验通过后把本次依据的解析结果填进去，
    供调用方把实际连接**钉定**在这些 IP 上——否则 httpx 连接时会再解析一次 DNS，
    两次解析之间存在 DNS 重绑定窗口（校验看到公网 IP、连接时被换到内网 IP）。
    """
    raw = str(url or "").strip()
    if not raw:
        raise _UrlRejectedError("empty_url")
    try:
        parts = urlsplit(raw)
    except ValueError as e:
        raise _UrlRejectedError("bad_url", str(e)) from e

    scheme = (parts.scheme or "").lower()
    if scheme not in ("http", "https"):
        raise _UrlRejectedError("bad_scheme", scheme or "(none)")
    if parts.username or parts.password:
        raise _UrlRejectedError("userinfo_not_allowed")

    host = (parts.hostname or "").strip().lower().rstrip(".")
    if not host:
        raise _UrlRejectedError("no_host")
    try:
        port = parts.port or (443 if scheme == "https" else 80)
    except ValueError as e:
        raise _UrlRejectedError("bad_port", str(e)) from e

    policy_norm = normalize_download_policy(policy)
    if policy_norm == DOWNLOAD_POLICY_OFF:
        # 显式关闭校验（排障用）。仍保留 scheme/凭据的前三条检查——这两项永不放开。
        return (raw, host)

    allowed = parse_allowed_hosts(allowed_hosts)
    if policy_norm == DOWNLOAD_POLICY_WHITELIST:
        if not allowed:
            raise _UrlRejectedError("allowlist_empty", "download_allowed_hosts 为空，默认拒绝")
        if not host_in_allowlist(host, allowed):
            raise _UrlRejectedError("host_not_allowed", host)

    resolve = resolver or resolve_host_ips
    try:
        ips = resolve(host, port)
    except OSError as e:
        raise _UrlRejectedError("dns_failed", f"{host}: {e}") from e
    if not ips:
        raise _UrlRejectedError("dns_empty", host)

    if not allow_private_hosts:
        blocked = [ip for ip in ips if not is_public_ip(ip)]
        if blocked:
            raise _UrlRejectedError("non_public_address", f"{host} → {','.join(blocked)}")

    if collect_ips is not None:
        collect_ips.extend(ips)

    return (raw, host)


def redact_url(url: str) -> str:
    """日志脱敏：QQ 文件链接的 query 里带 `fname`/signature 等凭据，只留 scheme+host+path。"""
    try:
        parts = urlsplit(str(url or ""))
    except ValueError:
        return "(unparsable url)"
    tail = "?<已隐去>" if parts.query else ""
    return urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")) + tail


def build_pinned_request(url: str, host: str, ip: Optional[str]) -> tuple[str, dict[str, Any]]:
    """把请求钉到已校验的 IP 上（v1.2.1 防 DNS 重绑定）。

    validate_download_target 校验通过的 IP 经 collect_ips 带出后，用本函数生成
    实际请求参数：URL 主机替换为 IP 字面量，连接不再触发第二次 DNS 解析；
    同时：
      - Host 头固定为原主机名（带非默认端口），虚拟主机路由不丢；
      - https 附加 `sni_hostname` 扩展（httpcore 连 IP、按原主机名做 TLS），
        证书校验仍针对原主机名，**不降级**；
      - ip 为 None（policy=off 未解析，无从钉定）时原样返回，维持旧行为。

    返回 (实际请求 URL, 传给 client.stream 的额外 kwargs)。
    """
    if not ip:
        return url, {}
    try:
        parts = urlsplit(url)
        port = parts.port
    except ValueError:
        return url, {}
    scheme = (parts.scheme or "").lower()
    netloc = f"[{ip}]" if ":" in ip else ip
    if port:
        netloc = f"{netloc}:{port}"
    pinned = urlunsplit((scheme, netloc, parts.path, parts.query, ""))
    req_extra: dict[str, Any] = {"headers": {"host": f"{host}:{port}" if port else host}}
    if scheme == "https":
        req_extra["extensions"] = {"sni_hostname": host}
    return pinned, req_extra


# ─── 配置模型 ────────────────────────────────────────────────────
class PluginSectionConfig(PluginConfigBase):
    """插件基础配置。"""

    __ui_label__ = "插件"
    __ui_icon__ = "file"
    __ui_order__ = 0

    enabled: bool = Field(default=True, description="是否启用插件")
    config_version: str = Field(default="1.1.0", description="配置版本（v1.3.0 从 1.0.0 升到 1.1.0：新增混合检索/生命周期/回执三类配置项，不升版这些键不会出现在已部署实例的 config.toml 里）")


class ReaderConfig(PluginConfigBase):
    """文件读取与检索配置。"""

    __ui_label__ = "文件读取"
    __ui_icon__ = "folder"
    __ui_order__ = 1

    max_file_size: int = Field(default=30, description="单文件大小上限（MB）。注意内存占用约为文件大小的 2~3 倍（字节+解析文本+向量同时驻留），大内存机器可调高")
    chunk_size: int = Field(default=512, description="分块大小（字符数）")
    chunk_overlap: int = Field(default=100, description="分块重叠（字符数）")
    chunk_merge_enabled: bool = Field(default=True, description="短碎片合并（v1.0.18 性能优化）：递归切分出的相邻短碎片合并到接近 chunk_size 再成块，块数降至约 1/3，入库向量化耗时同步下降；关闭则回退旧的逐碎片成块行为")
    retrieve_top_k: int = Field(default=6, description="检索返回的相关块数量")
    retrieve_use_matrix: bool = Field(default=True, description="矩阵化检索（v1.0.18 性能优化）：缓存 L2 归一化向量矩阵，查询时一次点积算全库相似度（入库/清理/删文件自动置脏重建）；块数多时检索开销显著下降，结果与逐条余弦一致；异常自动回退旧路径")
    file_retention_time: int = Field(default=60, description="文件有效时间（分钟，自**最后一次被注入**起算，v1.3.0 起为滑动续期）")
    file_max_rounds: int = Field(default=20, description="文件最大参与轮数（v1.3.0：5→20；只给实际被注入的文件计轮）")
    file_absolute_ttl: int = Field(default=720, description="文件绝对存活上限（分钟，自入库起算）：无论是否被问到，超过即清理，防文件永久驻留；0 = 不限制")
    cleanup_interval: int = Field(default=5, description="后台清理间隔（分钟）：清理过期文件、回收空会话向量库、裁剪概要冷却表")
    enable_group: bool = Field(default=True, description="是否处理群聊文件")
    download_url_policy: str = Field(
        default=DOWNLOAD_POLICY_WHITELIST,
        description=(
            "URL 下载校验策略（v1.2.0 防 SSRF）：whitelist = 只允许 download_allowed_hosts 内的域名（默认）；"
            "public = 允许任意主机但仍必须是公网 IP；off = 不校验（仅排障用，危险）。"
            "无论哪种策略都始终拒绝回环/私网/链路本地地址，并逐跳校验重定向。"
            "写错的值会退回最严格的 whitelist"
        ),
    )
    download_allowed_hosts: str = Field(
        default=DEFAULT_DOWNLOAD_HOSTS,
        description=(
            "URL 下载域名白名单，逗号分隔，后缀匹配（含子域）。"
            f"默认：{DEFAULT_DOWNLOAD_HOSTS}。留空 = whitelist 策略下拒绝一切 URL 下载"
        ),
    )
    download_max_redirects: int = Field(
        default=3,
        description="URL 下载允许的最大重定向跳数：每一跳重新做域名+地址校验（防白名单域名 302 到内网）；0 = 不跟随重定向",
    )
    download_allow_private_hosts: bool = Field(
        default=False,
        description=(
            "允许下载目标解析到内网/回环地址（v1.2.0）。"
            "仅当你的适配器用本机 HTTP 服务提供文件时才需要开启，且必须把该主机加进白名单；"
            "开启等于重新放开对内网的访问能力，默认保持关闭"
        ),
    )
    download_ca_bundle: str = Field(
        default="",
        description=(
            "自定义 CA 证书路径（PEM）：运行机器有 TLS MITM 代理时，指向代理根证书即可**保持证书校验开启**，"
            "比 insecure_download 安全得多，建议优先用这个"
        ),
    )
    insecure_download: bool = Field(
        default=False,
        description=(
            "下载文件时跳过 SSL 证书校验（运行机器存在 TLS MITM 代理、报 CERTIFICATE_VERIFY_FAILED 时开启）。"
            "优先改用 download_ca_bundle 指定代理根证书；开启后白名单域名之外仍是拒绝的，但对该域名的证书不再把关"
        ),
    )
    injection_marker: str = Field(default="【文件检索】", description="注入文本的幂等标记")
    inject_memo_enabled: bool = Field(default=True, description="注入去重缓存（v1.0.18）：同一会话同一问题在 TTL 内直接复用上次注入文本（零 embedding、零检索），覆盖 hook 每次尝试重跑与 Planner/回复双触发；文件入库、清文件、会话清理时自动失效")
    inject_memo_ttl: int = Field(default=90, description="注入去重缓存的存活秒数")
    silent_success: bool = Field(default=True, description="文件入库成功后保持静默（不发回执，日志仍记录）；关闭后每次入库都回复「已解析 N 块」")
    silent_errors: bool = Field(default=True, description="文件处理失败（不支持的类型/下载失败/超限/嵌入耗尽等）时保持静默（不发回执，日志仍记录，可用 /file_status 排查）；关闭后失败会回复 ⚠️ 提示")
    embed_retry_interval: float = Field(default=2.0, description="嵌入失败后台重试间隔（分钟）；入库时 embedding 超时会先在后台队列排队，定时重试")
    embed_max_retries: int = Field(default=30, description="后台重试最大次数（超过后放弃并提示重发文件；默认 30 次 × 2 分钟 ≈ 覆盖 1 小时拥塞）")
    embed_batch_size: int = Field(default=64, description="单次 llm.embed RPC 调用的最大文本条数；v1.0.18 配合分块合并把默认从 16 提到 64（合并后单批约 8K 字仍远低于 30s RPC 上限），批数下降进一步缩短入库耗时；若 host 限流可调回 16")
    embed_concurrency: int = Field(default=2, description="入库拆批后的并发批数：多批同时调用 llm.embed（1 = 旧串行行为）；host 出现限流/批量报错时调回 1")
    query_embed_timeout: float = Field(default=8.0, description="提问时查询 embedding 的超时秒数：仅单次调用、超时立即放弃并降级到概要兜底，不重试（避免拥塞期把用户卡在请求模型之前）；0 = 关闭超时（旧行为，最坏可卡 ~96s）")
    direct_inject_max_chars: int = Field(default=6000, description="全文直注阈值（解析后总字符数）：会话内所有文件全文合计不超过该值时，跳过 RAG 检索、直接把文件全文注入上下文（小文件无需检索即可全量可见，还省一次查询 embedding 调用）；设为 0 关闭该行为，始终走检索")
    summary_enabled: bool = Field(default=True, description="大文件 LLM 概要：全文超 direct_inject_max_chars 的文件在入库后用 llm.generate 生成一段概要，注入上下文时附在检索片段前，让 LLM 对大文件先有整体认识；生成失败静默降级（不影响检索）")
    summary_source_chars: int = Field(default=12000, description="生成概要时截取的源文本长度（字符数）：从文件头、中、尾三段均匀取样拼接，控制摘要调用的 token 开销")
    summary_max_chars: int = Field(default=500, description="概要文本的最大长度（字符数）")
    summary_await_on_inject: bool = Field(default=False, description="提问注入时是否同步等待概要生成（v1.0.17 旧行为）：概要缺失时同步调用 llm.generate 会把 BLOCKING hook 卡住数秒；默认 false = 缺失时立即用已有内容检索注入，概要在后台生成、下次提问自然带上")
    summary_retry_interval: float = Field(default=600.0, description="概要生成失败后的冷却秒数：冷却期内不再尝试生成（避免每次提问都撞一次超时）；0 = 不冷却（每次注入都重试）")
    # ── v1.3.0 混合检索（向量 + BM25，RRF 融合） ──
    hybrid_retrieve_enabled: bool = Field(default=True, description="BM25 词法检索总开关（v1.3.0）：开启后检索走「向量 + 词法 RRF 融合」，对函数名/报错码/编号这类字面查询显著更准；体感变差可关闭退回纯向量，也可用 /file_bm25 off 运行期关闭")
    hybrid_candidate_k: int = Field(default=20, description="每路（向量/词法）参与融合的候选数")
    rrf_k: int = Field(default=60, description="RRF 融合常数（越大排名差异越平缓）")
    lexical_max_tokens: int = Field(default=2000000, description="单会话词法索引 token 上限：超过即停用该会话的词法检索并退回纯向量（防大文件会话内存膨胀）；0 = 不限制")
    retrieve_min_score: float = Field(default=0.15, description="向量相关度绝对下限：余弦低于此值的候选在融合前丢弃（只作用于有向量分的候选，纯词法命中不受影响）；0 = 关闭")
    retrieve_min_ratio: float = Field(default=0.5, description="融合后相对下限：低于 Top1 分数该比例的结果丢弃；0 = 关闭")
    query_rewrite_enabled: bool = Field(default=True, description="追问改写（v1.3.0）：「那第三章呢」这类指代/短问补回上一问再检索；纯规则实现，不调 LLM（BLOCKING hook 只有 11s 预算）")
    # ── v1.3.0 生命周期与回执 ──
    ack_min_chars: int = Field(default=20000, description="大文件入库回执阈值（原文字符数）：达到该值（或处理耗时 ≥ 10s）时发一条中性回执「已读取《X》，共 N 段」，避免用户以为插件没反应；0 = 不发回执")
    parse_notice_ttl: int = Field(default=600, description="解析失败通知的存活秒数（v1.3.0）：失败原因注入给模型（不发群消息），TTL 内只注入一次")


class NapcatConfig(PluginConfigBase):
    """NapCat HTTP API 兜底配置。

    真机实测（v1.0.6）：napcat-adapter 会把 file 段降级成纯文本
    `[文件] xxx.docx，大小: 9547`，**连 chat.receive.before_process（SessionMessage.process()
    之前）拿到的也已经是 text 段** —— hook 层拿不到文件本体，只能回头找 NapCat 要原始事件。
    """

    __ui_label__ = "NapCat 兜底"
    __ui_icon__ = "cloud-download"
    __ui_order__ = 2

    enabled: bool = Field(
        default=False,
        description="启用 NapCat HTTP API 兜底取文件（适配器会把文件段转成纯文本，不启用则读不到内容）",
    )
    http_url: str = Field(default="http://127.0.0.1:3001", description="NapCat OneBot HTTP 服务地址")
    access_token: str = Field(default="", description="NapCat HTTP access_token（未设置则留空）")
    timeout: float = Field(default=15.0, description="HTTP 请求超时（秒）")
    verify_ssl: bool = Field(
        default=True,
        description=(
            "HTTPS 时校验证书与主机名（v1.2.1 起默认开启）。本地 http 地址不受影响；"
            "自签证书请优先让系统信任该证书（或加入信任库），关闭此开关等于"
            "允许中间人窃听/篡改与 NapCat 的全部通信（含 access_token），需自担风险"
        ),
    )
    cache_dir: str = Field(
        default="",
        description="可选：NapCat 文件缓存目录，get_file 失败时按文件名+大小在此目录搜索",
    )


class FileReaderConfig(PluginConfigBase):
    """插件配置。"""

    plugin: PluginSectionConfig = Field(default_factory=PluginSectionConfig)
    reader: ReaderConfig = Field(default_factory=ReaderConfig)
    napcat: NapcatConfig = Field(default_factory=NapcatConfig)


class FileReaderPlugin(MaiBotPlugin):
    """文件读取插件。"""

    config_model = FileReaderConfig
    # 订阅全局模型配置热重载（embedding 模型变化时重建）
    config_reload_subscriptions: ClassVar[Iterable[str]] = ("model",)

    def get_webui_config_schema(self, **kwargs) -> dict:
        """覆写 SDK 的 WebUI 配置 Schema：做可视化模式的显示层补丁。

        Runner 调这个方法拿配置页 Schema 且异常会被吞掉（变成空 Schema、
        配置页整页空白），所以这里自己兜底：补丁失败就原样返回 SDK 输出。
        """

        schema = super().get_webui_config_schema(**kwargs)
        try:
            return _apply_webui_display_polish(schema)
        except Exception:  # noqa: BLE001 —— 显示补丁失败绝不能让配置页变空白
            logging.getLogger(__name__).exception("修正 WebUI 配置 Schema 失败，回退 SDK 原样输出")
            return schema

    # ─── 生命周期 ────────────────────────────────────────────────
    async def on_load(self) -> None:
        """插件加载时初始化。"""
        self._store = SessionStore(Path(self.ctx.paths.data_dir) / "file_reader")
        self._cleanup_task: Optional[asyncio.Task] = None
        self._embedding_ok = True
        self._embedding_err = ""
        # 双 hook（before/after）去重：key = f"{session_id}:{file_name}" -> 上次处理时刻
        self._recent_file_keys: dict[str, float] = {}
        # 最近一次"被降级成纯文本"的文件消息留档（供 /file_status 排查）
        self._last_file_hint: dict[str, Any] = {}
        # embedding 失败重试队列（v1.0.11）：[{session_id, conversation_id, stream_id, name, tmp_path, attempts, next_ts}]
        # v1.1.0：队列条目存临时文件路径而非解析全文——全文（可达数 MB）不再驻留队列最长 1 小时
        self._embed_retry_queue: list[dict[str, Any]] = []
        self._retry_task: Optional[asyncio.Task] = None
        # 注入去重缓存（v1.0.18 C3）：(session_id, query_key) -> (ts, inject_text)
        # 覆盖 hook 每次尝试重跑与 Planner/回复双触发；入库/清文件/会话清理时必须失效
        self._inject_memo: dict[tuple[str, str], tuple[float, str]] = {}
        # 概要异步化（v1.0.18 C4）：in-flight 去重（同一文件的概要任务不重复起）
        # 与失败冷却（file_name -> 上次失败时刻，冷却期内不再撞 llm.generate 超时）
        # v1.1.0：pending 键从 id(entry) 改为 file_name——id() 在 entry 释放后可能被新对象复用，
        # 理论上会误判 in-flight 导致该文件概要永远不生成
        self._summary_pending: set[str] = set()
        self._summary_failed_ts: dict[str, float] = {}
        # v1.1.0：后台任务句柄注册表——create_task 裸调只靠事件循环弱引用持有，
        # 有 GC 提前回收风险；且 on_unload 需要统一取消，防任务泄漏到重载后的旧 store
        self._bg_tasks: set[asyncio.Task] = set()
        # v1.1.0：下载用 httpx 单例（连接复用，免每次新建 client + SSL 上下文）
        self._http_client: Optional[Any] = None
        # v1.2.0：缓存的是「实际传给 httpx 的 verify 值」——可能是 True / False / CA 证书路径
        self._http_client_verify: Any = True
        # v1.1.0：NapCat urllib SSL 上下文缓存（create_default_context 是 ms 级 CPU 调用）
        self._napcat_ssl_ctx: Optional[ssl.SSLContext] = None
        self._napcat_ssl_verify: bool = False
        # v1.3.0：检索诊断环形缓冲（每会话最近 20 条）——没有 recall@k 评测集时，
        # 这是唯一能回答「为什么没检索到」的可观测窗口
        self._retrieval_diag: deque = deque(maxlen=20)
        # v1.3.0：解析失败通知队列 session_id -> [{name, reason, ts, delivered}]
        # 失败只告诉模型、不刷群；注入一次即标记 delivered
        self._file_notices: dict[str, list[dict]] = {}
        # v1.3.0：BM25 运行期开关（None = 跟随配置；True/False = /file_bm25 强制覆盖）
        # 为什么需要在代码层再留一道：已部署实例的 config.toml 不会自动补新键，
        # 只改 Field(default) 对老实例完全无效，体感变差时必须有立刻可用的退路
        self._hybrid_override: Optional[bool] = None

        # 行为自检：确认 chunker / 解析器在进程内可用（runtime-gotchas 5.1 的对策）
        self._run_self_check()

        self.ctx.logger.info(
            "文件读取插件已加载：data_dir=%s，chunk_size=%d，top_k=%d，有效时间=%dmin，最大轮数=%d",
            self._store.data_dir,
            self.config.reader.chunk_size,
            self.config.reader.retrieve_top_k,
            self.config.reader.file_retention_time,
            self.config.reader.file_max_rounds,
        )
        # NapCat 兜底开关是"能否读到文件"的关键，加载时显式打出来
        self.ctx.logger.info(
            "NapCat 兜底: %s (url=%s, token=%s, cache_dir=%s)",
            "启用" if self.config.napcat.enabled else "停用（适配器降级的文件读不到内容！）",
            self.config.napcat.http_url,
            "已设置" if (self.config.napcat.access_token or "").strip() else "空",
            self.config.napcat.cache_dir or "未设置",
        )
        await self._start_cleanup_loop()
        await self._start_retry_loop()

    async def on_unload(self) -> None:
        """插件卸载时停止后台任务。"""
        if self._cleanup_task:
            self._cleanup_task.cancel()
            try:
                await self._cleanup_task
            except (asyncio.CancelledError, Exception):
                pass
            self._cleanup_task = None
        if self._retry_task:
            self._retry_task.cancel()
            try:
                await self._retry_task
            except (asyncio.CancelledError, Exception):
                pass
            self._retry_task = None
        # v1.1.0：统一取消进行中的下载/解析/概要任务（旧版泄漏到重载后仍写旧 store）
        for task in list(self._bg_tasks):
            task.cancel()
        if self._bg_tasks:
            await asyncio.gather(*self._bg_tasks, return_exceptions=True)
            self._bg_tasks.clear()
        # v1.1.0：关闭 httpx 单例连接池
        if self._http_client is not None:
            try:
                await self._http_client.aclose()
            except Exception:  # noqa: BLE001
                pass
            self._http_client = None
        # v1.1.0：清掉重试队列残留的临时文件（条目已改存路径）
        for item in self._embed_retry_queue:
            tmp = item.get("tmp_path")
            if tmp:
                try:
                    os.remove(str(tmp))
                except OSError:
                    pass
        self._embed_retry_queue = []
        self.ctx.logger.info("文件读取插件已卸载")

    def _spawn_bg(self, coro: Any) -> None:
        """注册式 create_task（v1.1.0）：持句柄防 GC 提前回收，on_unload 统一取消。"""
        task = asyncio.create_task(coro)
        self._bg_tasks.add(task)
        task.add_done_callback(self._bg_tasks.discard)

    async def on_config_update(self, scope: str, config_data: dict[str, Any], version: str) -> None:
        """配置热重载。"""
        del scope, config_data, version
        # v1.0.18 C5：热重载同步矩阵检索开关到已存在的会话库
        use_matrix = bool(self.config.reader.retrieve_use_matrix)
        for vs in self._store._sessions.values():
            vs.retrieve_use_matrix = use_matrix
        self.ctx.logger.info(
            "配置已热重载：chunk_size=%d, top_k=%d, retention=%dmin, max_rounds=%d, matrix_retrieve=%s",
            self.config.reader.chunk_size,
            self.config.reader.retrieve_top_k,
            self.config.reader.file_retention_time,
            self.config.reader.file_max_rounds,
            use_matrix,
        )

    # ─── 行为自检 ────────────────────────────────────────────────
    def _run_self_check(self) -> None:
        """拿固定样例喂进程内函数对象，验证行为而非读源码文本（runtime-gotchas）。"""
        try:
            c = RecursiveCharacterChunker(10, 2)
            out = c.chunk("一二三四五六七八九十十一十二十三十四")
            assert isinstance(out, list) and out and all(isinstance(x, str) for x in out), "分块结果异常"
            self.ctx.logger.info("[自检] 分块器: PASS (%d 块)", len(out))
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("[自检] 分块器: FAIL (%s)", e)

        try:
            from file_parser import get_extension, is_supported

            assert get_extension("a.PDF") == "pdf"
            assert is_supported("report.xlsx")
            assert not is_supported("photo.mp4")
            self.ctx.logger.info("[自检] 解析器类型表: PASS")
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("[自检] 解析器类型表: FAIL (%s)", e)

        # v1.2.0：下载安全策略自检——把风险配置在启动时喊出来，而不是等出事
        try:
            raw_policy = str(self.config.reader.download_url_policy or "")
            if raw_policy.strip().lower() not in DOWNLOAD_POLICIES:
                self.ctx.logger.error(
                    "[自检] download_url_policy=%r 不是合法值，已退回最严格的 %s（可选：%s）",
                    raw_policy,
                    DOWNLOAD_POLICY_WHITELIST,
                    "/".join(DOWNLOAD_POLICIES),
                )
            policy = normalize_download_policy(raw_policy)
            hosts = parse_allowed_hosts(self.config.reader.download_allowed_hosts)
            risky: list[str] = []
            if policy == DOWNLOAD_POLICY_OFF:
                risky.append("download_url_policy=off（URL 下载不做任何域名/IP 校验）")
            if bool(self.config.reader.download_allow_private_hosts):
                risky.append("download_allow_private_hosts=true（允许下载目标指向内网/回环）")
            if bool(self.config.reader.insecure_download):
                risky.append(
                    "insecure_download=true（关闭 TLS 证书校验；建议改用 download_ca_bundle 指定代理根证书）"
                )
            # v1.2.1（review 项）：verify_ssl=false 在 http 场景无影响，但配了 https
            # 地址时等于把证书与主机名校验一起关掉（CERT_NONE）。默认值已翻转为
            # true，但存量实例的 config.toml 不会跟随新默认值——靠这条启动告警兜住。
            napcat_url = str(self.config.napcat.http_url or "").strip()
            if napcat_url.lower().startswith("https") and not bool(self.config.napcat.verify_ssl):
                risky.append(
                    "napcat.verify_ssl=false 且 napcat.http_url 为 HTTPS 地址"
                    "（证书与主机名校验全部关闭，与 NapCat 的通信可被中间人窃听/篡改；"
                    "如因自签证书报错，应让系统信任该证书而不是关校验）"
                )
            if policy == DOWNLOAD_POLICY_WHITELIST and not hosts:
                self.ctx.logger.warning(
                    "[自检] 下载白名单为空且策略为 whitelist —— 所有 URL 来源的文件都会被拒绝下载"
                )
            if risky:
                for line in risky:
                    self.ctx.logger.warning("[自检][安全] 风险配置：%s", line)
            else:
                self.ctx.logger.info(
                    "[自检] 下载安全策略: PASS (policy=%s, 白名单 %d 项, 重定向上限 %d)",
                    policy,
                    len(hosts),
                    int(self.config.reader.download_max_redirects),
                )
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("[自检] 下载安全策略: FAIL (%s)", e)

    # ─── embedding 封装 ──────────────────────────────────────────
    async def _embed(self, texts: list[str], *, query_mode: bool = False) -> Any:
        """调用 MaiBot llm.embed，带自动重试（指数退避），最终失败返回空（上层抛可读错误）。

        v1.0.18：拆批后多批并发（embed_concurrency，默认 2）+ 批大小提到 64；
        max_concurrent 一并透传给 host 侧调度。任一批失败仍整单返回 {}（防残缺向量）。
        query_mode=True 时走查询分层：单次调用 + wait_for 超时，不重试（拥塞期不卡用户）。
        """
        batch_size = max(1, int(self.config.reader.embed_batch_size))
        if len(texts) <= batch_size:
            return await self._embed_once(texts, query_mode=query_mode)

        # 拆批
        batches = [texts[i : i + batch_size] for i in range(0, len(texts), batch_size)]
        total_batches = len(batches)
        concurrency = max(1, int(self.config.reader.embed_concurrency)) if not query_mode else 1

        if concurrency <= 1:
            # 串行路径（embed_concurrency=1 等价旧行为）
            results: list[Any] = []
            for idx, part in enumerate(batches):
                part_result = await self._embed_once(part, query_mode=query_mode)
                if part_result == {} or part_result is None:
                    self.ctx.logger.warning(
                        "embedding 拆批调用第 %d/%d 批失败，整单失败待重试", idx + 1, total_batches
                    )
                    return {}
                results.append(part_result)
                self.ctx.logger.info("embedding 拆批进度: %d/%d 批（%d 块）", idx + 1, total_batches, len(part))
            return self._merge_embed_results(results)

        # 并发路径：Semaphore 限流，return_exceptions 收集失败批
        sem = asyncio.Semaphore(concurrency)
        done_log: dict[int, str] = {}

        async def _run_one(idx: int, part: list[str]) -> Any:
            async with sem:
                return await self._embed_once(part, query_mode=query_mode)

        async def _guarded(idx: int, part: list[str]) -> tuple[int, Any]:
            try:
                r = await _run_one(idx, part)
                done_log[idx] = "ok"
                return (idx, r)
            except Exception as e:  # noqa: BLE001
                done_log[idx] = f"{type(e).__name__}: {e}"
                return (idx, {})

        gathered = await asyncio.gather(*(_guarded(i, b) for i, b in enumerate(batches)))
        gathered.sort(key=lambda x: x[0])  # 按原批序拼回，保证向量与文本对齐

        failed = [i for i, r in gathered if r == {} or r is None]
        if failed:
            self.ctx.logger.warning(
                "embedding 并发拆批 %d/%d 批失败（%s），整单失败待重试",
                len(failed), total_batches, ",".join(str(i + 1) for i in failed),
            )
            return {}
        for i in range(total_batches):
            self.ctx.logger.info("embedding 拆批进度: %d/%d 批（%d 块）", i + 1, total_batches, len(batches[i]))
        return self._merge_embed_results([r for _, r in gathered])

    def _merge_embed_results(self, results: list[Any]) -> Any:
        """把多批 embed 结果拼成一个与单批同构的结果。"""
        # 纯 list 形态：直接拼接
        if all(isinstance(r, list) for r in results):
            merged: list[Any] = []
            for r in results:
                merged.extend(r)
            return merged
        # dict 形态：兼容 results / embeddings 两种键
        if all(isinstance(r, dict) for r in results):
            merged_dict: dict[str, Any] = {}
            for key in ("results", "embeddings"):
                if all(key in r for r in results):
                    items: list[Any] = []
                    for r in results:
                        items.extend(r[key])
                    merged_dict[key] = items
            if merged_dict:
                return merged_dict
            # 单条形态（embedding 键）：每批只有 1 条时退化为列表语义不可行，直接取第一键拼接不了，
            # 保守返回第一批（调用方 _embed_query 只用于单条查询，不会走拆批路径）
            return results[0]
        return results[0]

    async def _embed_once(self, texts: list[str], *, query_mode: bool = False) -> Any:
        """单批 embed 调用 + 自动重试。

        v1.0.18 重试分层：
        - 入库（query_mode=False）：2 次尝试、退避 2s——入库走后台任务，可等；
        - 查询（query_mode=True）：单次调用 + wait_for(query_embed_timeout)——
          BLOCKING hook 卡在请求模型之前是用户感知延迟主因（旧行为最坏 ~96s），
          超时立即返回 {}，上层注入路径会降级到概要兜底。
        """
        if query_mode:
            timeout = float(self.config.reader.query_embed_timeout or 0)
            try:
                if timeout > 0:
                    return await asyncio.wait_for(self.ctx.llm.embed(texts=texts), timeout=timeout)
                return await self.ctx.llm.embed(texts=texts)
            except (asyncio.TimeoutError, Exception) as e:  # noqa: BLE001
                self._embedding_ok = False
                self._embedding_err = f"{type(e).__name__}: {e}"
                self.ctx.logger.warning("查询 embedding 失败（不重试，降级概要兜底）: %s", e)
                return {}

        last_err: Optional[Exception] = None
        for attempt in range(2):
            try:
                result = await self.ctx.llm.embed(texts=texts)
                self._embedding_ok = True
                self._embedding_err = ""
                return result
            except Exception as e:  # noqa: BLE001
                last_err = e
                self._embedding_ok = False
                self._embedding_err = f"{type(e).__name__}: {e}"
                if attempt < 1:  # 还有重试机会
                    delay = 2.0
                    self.ctx.logger.warning(
                        "embedding 调用失败（第 %d 次），%.0fs 后重试: %s",
                        attempt + 1,
                        delay,
                        e,
                    )
                    await asyncio.sleep(delay)
        self.ctx.logger.error("embedding 调用失败（已重试 1 次）: %s", last_err)
        return {}

    async def _embed_query(self, query: str) -> Optional[list[float]]:
        result = await self._embed([query], query_mode=True)
        if isinstance(result, dict):
            if isinstance(result.get("results"), list) and result["results"]:
                item = result["results"][0]
                if isinstance(item, dict) and "embedding" in item:
                    return list(item["embedding"])
            if result.get("embedding") is not None:
                return list(result["embedding"])
        if isinstance(result, list) and result:
            return list(result[0])
        return None

    # ─── 文件消息检测（双阶段 hook：before_process 拿原始段，after_process 兜底） ───
    @HookHandler("chat.receive.before_process", name="detect_file_early", mode=HookMode.OBSERVE, error_policy=ErrorPolicy.SKIP)
    async def on_file_message_early(self, **kwargs: Any) -> dict[str, Any]:
        """预处理阶段检测文件。

        官方事件管线：before_process 在 SessionMessage.process() 轻量转换**之前**触发，
        raw_message 可能仍是原始消息段（含 file 段的 url/base64/file_id）——
        after_process 阶段实测文件已被转换成 text 描述（raw=list[text(-)]），故加此 hook。
        """
        return await self._detect_file("before", **kwargs)

    @HookHandler("chat.receive.after_process", name="detect_file", mode=HookMode.OBSERVE, error_policy=ErrorPolicy.SKIP)
    async def on_file_message(self, **kwargs: Any) -> dict[str, Any]:
        """处理完成阶段检测文件（兜底：before_process 未提取到时再试一次）。"""
        return await self._detect_file("after", **kwargs)

    async def _detect_file(self, stage: str, **kwargs: Any) -> dict[str, Any]:
        """双阶段共用的文件检测逻辑；同文件 10 秒窗口去重防重复入库。"""
        message = kwargs.get("message")

        # 诊断：疑似文件消息必打（看清载荷），普通消息每阶段只打一次
        self._log_file_hook_diag(stage, kwargs, message)

        if not isinstance(message, dict) or not message:
            return {"action": "continue"}

        if not self.config.plugin.enabled:
            self.ctx.logger.info("[诊断] 插件已停用（plugin.enabled=false），跳过文件检测")
            return {"action": "continue"}

        # 群聊开关（兼容 message_info.group_id 平铺与 group_info 嵌套两种形态）
        if not self.config.reader.enable_group:
            mi = message.get("message_info") or {}
            gi = mi.get("group_info") if isinstance(mi, dict) else None
            is_group = bool(
                (isinstance(mi, dict) and mi.get("group_id"))
                or (isinstance(gi, dict) and (gi.get("group_id") or gi.get("group_name") or gi.get("group_all")))
                or kwargs.get("is_group")
            )
            if is_group:
                return {"action": "continue"}

        # 提取文件段
        files = self._extract_files(message)
        if not files and self._looks_like_file(message):
            # 真机形态：适配器已把 file 段降级成 `[文件] xxx.docx，大小: 9547` 纯文本，
            # 本体只能拿 message_id 回头找 NapCat 要。
            # v1.0.19：防抖插件会把连续文件消息合并成一条，需解析全部 [文件] 行
            hints = self._parse_file_hints(message)
            if hints:
                mid = message.get("message_id")
                for hint in hints:
                    hint["napcat_message_id"] = mid
                # 留档：/file_status 里能看到最近一次文件消息的 message_id，便于核对 NapCat 回溯
                self._last_file_hint = {
                    "name": "、".join(h["name"] for h in hints[:5]) + ("…" if len(hints) > 5 else ""),
                    "size": hints[0]["size_hint"],
                    "mid": repr(mid),
                    "stage": stage,
                }
                files = hints
        if not files:
            return {"action": "continue"}

        session_id = str(self._pick_session_id(message, kwargs) or "default")
        conversation_id = str(kwargs.get("session_id") or session_id)
        stream_id = str(self._pick_stream_id(message, kwargs) or "")

        # 去重：同一 session 同名文件 10 秒内只处理一次（before/after 双 hook 各看到一次）
        now = time.time()
        to_handle: list[dict[str, Any]] = []
        for fdata in files:
            key = f"{session_id}:{fdata.get('name', '')}"
            last = self._recent_file_keys.get(key, 0.0)
            if now - last < 10.0:
                continue
            self._recent_file_keys[key] = now
            to_handle.append(fdata)
        # 顺带清理过期键
        if len(self._recent_file_keys) > 64:
            self._recent_file_keys = {k: v for k, v in self._recent_file_keys.items() if now - v < 60.0}

        # 后台处理，不阻塞聊天主流程（runtime-gotchas 5.3）
        for fdata in to_handle:
            self._spawn_bg(self._handle_file(session_id, conversation_id, stream_id, fdata))
        return {"action": "continue"}

    def _looks_like_file(self, message: dict[str, Any]) -> bool:
        """轻量判断：该消息是否疑似文件消息（不做内容提取，只判标记与段类型）。"""
        plain = str(message.get("processed_plain_text") or "")
        if "[文件]" in plain or "[文件]" in str(message.get("text") or ""):
            return True
        segs = message.get("raw_message")
        if isinstance(segs, list):
            for s in segs:
                if isinstance(s, dict) and str(s.get("type", "")).lower() in ("file", "attachment"):
                    return True
        return False

    def _parse_file_hints(self, message: dict[str, Any]) -> list[dict[str, Any]]:
        """从降级文本里解析全部文件（v1.0.19）：防抖插件会把连续多条文件消息合并成一条，
        文本里出现多行 `[文件] xxx，大小: N，链接: https://...`——必须逐行解析，漏一行丢一个文件。

        单行形态（v1.0.8 既有）：
        - 私聊：`[文件] 文章.docx，大小: 9547`
        - 群聊：`[文件] 文章.docx，大小: 9547，链接: https://tjc-download.ftn.qq.com/...`

        返回 [{"name", "size_hint", "url"?}, ...]；一行解析不出则跳过该行。
        """
        plain = str(message.get("processed_plain_text") or "") or str(message.get("text") or "")
        if "[文件]" not in plain:
            return []
        hints: list[dict[str, Any]] = []
        # 逐行解析：链接可能含逗号，先按行切分再在行内摘 URL
        for line in plain.splitlines():
            line = line.strip()
            if "[文件]" not in line:
                continue
            url_m = _RE_HINT_URL.search(line)
            url = url_m.group(1).rstrip("，,。") if url_m else ""
            # 去掉链接尾巴再匹配文件名，避免长 URL 被吞进 name
            name_part = line[: url_m.start()] if url_m else line
            m = _RE_HINT_NAME.search(name_part)
            if not m:
                continue
            name = m.group("name").strip()
            if not name:
                continue
            size_raw = m.group("size")
            hint: dict[str, Any] = {"name": name, "size_hint": int(size_raw) if size_raw else 0}
            if url:
                hint["url"] = url
            hints.append(hint)
        return hints

    def _parse_file_hint(self, message: dict[str, Any]) -> Optional[dict[str, Any]]:
        """单文件形态兼容入口（v1.0.8 版语义，内部走 _parse_file_hints）。"""
        hints = self._parse_file_hints(message)
        return hints[0] if hints else None

    def _log_file_hook_diag(self, stage: str, kwargs: dict[str, Any], message: Any) -> None:
        """诊断载荷结构。

        策略（v1.0.5）：
        - 疑似文件消息：每阶段最多打 5 条完整段结构——排障靠的就是这些行，
          不能让第一条普通文本消息把一次性标志消耗掉（v1.0.4 踩过）。
        - 普通消息：每阶段只打一行，防刷屏。
        """
        if not isinstance(message, dict) or not message:
            flag = f"_file_diag_logged_{stage}"
            if getattr(self, flag, False):
                return
            setattr(self, flag, True)
            self.ctx.logger.info(
                "[诊断] file hook[%s] 载荷非 dict: message_type=%s kwargs_keys=%s",
                stage,
                type(message).__name__,
                sorted(kwargs.keys()),
            )
            return
        looks_file = self._looks_like_file(message)
        if looks_file:
            cnt_attr = f"_file_diag_file_count_{stage}"
            cnt = int(getattr(self, cnt_attr, 0))
            if cnt >= 5:
                return
            setattr(self, cnt_attr, cnt + 1)
            tag = "FILE"
        else:
            flag = f"_file_diag_logged_{stage}"
            if getattr(self, flag, False):
                return
            setattr(self, flag, True)
            tag = "msg"
        self._log_diag_detail(stage, tag, kwargs, message)

    def _log_diag_detail(self, stage: str, tag: str, kwargs: dict[str, Any], message: dict[str, Any]) -> None:
        """打印单条诊断：载荷键结构 + raw_message 段细节 + 纯文本预览。"""
        try:
            mi = message.get("message_info")
            raw = message.get("raw_message")
            raw_desc = self._describe_raw(raw)
            plain = str(message.get("processed_plain_text") or "")
            self.ctx.logger.info(
                "[诊断] file hook[%s][%s] 触发: kwargs_keys=%s message_keys=%s message_info_keys=%s mid=%s raw=%s plain=%s",
                stage,
                tag,
                sorted(kwargs.keys()),
                sorted(message.keys()),
                sorted(mi.keys()) if isinstance(mi, dict) else "-",
                repr(message.get("message_id")),
                raw_desc,
                repr(plain[:80]),
            )
        except Exception:  # noqa: BLE001
            pass

    def _describe_raw(self, raw: Any) -> str:
        """描述 raw_message 结构：list → 各段 type + data 摘要（data 非 dict 时打印值与类型）。"""
        if isinstance(raw, list):
            descs = []
            for s in raw[:8]:
                if not isinstance(s, dict):
                    descs.append(f"{type(s).__name__}={str(s)[:40]!r}")
                    continue
                data = s.get("data")
                if isinstance(data, dict):
                    detail = "|".join(sorted(data.keys()))
                elif data is None:
                    detail = "-"
                else:
                    # data 为非 dict（如纯字符串）：打印类型与值，便于看清 text 段到底存了什么
                    detail = f"{type(data).__name__}={str(data)[:60]!r}"
                descs.append(f"{s.get('type')}({detail})")
            return f"list[{', '.join(descs)}]"
        if isinstance(raw, dict):
            return f"dict keys={sorted(raw.keys())}"
        return type(raw).__name__

    def _extract_files(self, message: dict[str, Any]) -> list[dict[str, Any]]:
        """从消息里找出文件段，返回 [{name, bytes|path|url}, ...]。

        兼容多种结构：
        - segments / message_chain / content / message 列表里的 {"type": "file", ...}
        - OneBot 风格 {"type": "file", "data": {...}}（napcat 等适配器）
        - raw_message 里嵌套的 file / attachment / files
        - raw_message.message 数组（OneBot 原始事件）
        """
        found: list[dict[str, Any]] = []

        def _scan_segments(segs: Any) -> None:
            if not isinstance(segs, list):
                return
            for seg in segs:
                if not isinstance(seg, dict):
                    continue
                stype = str(seg.get("type", "")).lower()
                if stype in ("file", "attachment"):
                    # OneBot 风格：实际字段嵌在 data 里
                    payload = seg.get("data") if isinstance(seg.get("data"), dict) else seg
                    fdata = self._seg_to_file(payload)
                    if fdata:
                        found.append(fdata)
                    else:
                        self.ctx.logger.warning(
                            "文件段无法提取内容: payload_keys=%s", sorted(payload.keys())
                        )

        # 1) 消息级段列表
        for key in ("segments", "message_chain", "content", "message"):
            _scan_segments(message.get(key))

        # 2) raw_message：list（napcat 真机形态：消息段数组）或 dict（嵌套容器）
        raw = message.get("raw_message")
        if isinstance(raw, list):
            # 真机确认：raw_message 本身就是 [{"type": "file", "data": {...}}, ...] 段数组
            _scan_segments(raw)
        elif isinstance(raw, dict):
            for key in ("file", "attachment", "files"):
                val = raw.get(key)
                if isinstance(val, dict):
                    fdata = self._seg_to_file(val)
                    if fdata:
                        found.append(fdata)
                else:
                    _scan_segments(val)
            # OneBot 原始事件：message 数组
            if isinstance(raw.get("message"), list):
                _scan_segments(raw["message"])
        return found

    def _seg_to_file(self, seg: dict[str, Any]) -> Optional[dict[str, Any]]:
        """从单个文件段提取 name + 内容（base64 / 本地路径 / URL）。

        兼容字段名：name / file_name / filename / file / file_id / title；
        内容：base64 / data / content_base64 / url / file_url / path / file_path / local_path。
        OneBot 的 data.file 可能是 URL、本地缓存路径或纯文件名，按形态分发。
        """
        raw_file = str(seg.get("file") or "")
        name = (
            seg.get("name")
            or seg.get("file_name")
            or seg.get("filename")
            or seg.get("title")
            or Path(raw_file).name  # OneBot data.file 常带路径，取文件名
            or seg.get("file_id")
            or "unnamed_file"
        )
        content_b64 = seg.get("base64") or seg.get("content_base64")
        url = seg.get("url") or seg.get("file_url")
        path = seg.get("path") or seg.get("file_path") or seg.get("local_path") or seg.get("tmp_path")

        # OneBot data.file 形态分发：URL / 本地路径 / 纯文件名
        if raw_file:
            if raw_file.startswith(("http://", "https://")) and not url:
                url = raw_file
            elif not path and Path(raw_file).exists():
                path = raw_file

        if content_b64:
            try:
                raw_bytes = base64.b64decode(content_b64)
                return {"name": str(name), "bytes": raw_bytes}
            except Exception:  # noqa: BLE001
                pass
        if path and Path(str(path)).exists():
            return {"name": str(name), "path": str(path)}
        if url:
            return {"name": str(name), "url": str(url)}
        # v1.3.2：结构化 file 段（SnowLuma Adapter v1.1.10+）只有 file_id/name/size/busid，
        # **没有任何内容字段**。旧版在这里返回 None，整段被丢弃并打 warning，只能退到
        # `[文件]` 文本 → `get_msg` 回溯：多一次 RPC，且定位依赖 size 消歧（同名同大小就串档）。
        # 直接把 file_id 交出去，`_handle_file` 用 `get_file` 一步就能取到本地路径。
        file_id = seg.get("file_id") or seg.get("id")
        if file_id:
            try:
                size_hint = int(str(seg.get("size") or seg.get("file_size") or 0).strip())
            except (TypeError, ValueError):
                size_hint = 0
            try:
                busid = int(str(seg.get("busid") or 0).strip())
            except (TypeError, ValueError):
                busid = 0
            return {"name": str(name), "file_id": str(file_id), "busid": busid, "size_hint": size_hint}
        # 无可用内容
        return None

    def _pick_session_id(self, message: dict[str, Any], kwargs: dict[str, Any]) -> str:
        for key in ("session_id", "user_id", "sender_id", "user_openid"):
            if kwargs.get(key):
                return str(kwargs[key])
        # 真机形态：message 顶层 session_id（napcat，与 before_model_request 的 session 一致）
        if message.get("session_id"):
            return str(message["session_id"])
        mi = message.get("message_info") or {}
        if not isinstance(mi, dict):
            return ""
        # 真机形态：user_info / group_info 嵌套
        ui = mi.get("user_info")
        if isinstance(ui, dict):
            for key in ("user_id", "id", "qq", "user_openid"):
                if ui.get(key):
                    return str(ui[key])
        gi = mi.get("group_info")
        if isinstance(gi, dict):
            for key in ("group_id", "id", "group_all"):
                if gi.get(key):
                    return str(gi[key])
        for key in ("user_id", "sender_id", "user_openid", "group_id"):
            if mi.get(key):
                return str(mi[key])
        return ""

    def _pick_stream_id(self, message: dict[str, Any], kwargs: dict[str, Any]) -> str:
        for key in ("stream_id", "chat_id", "session_id", "stream"):
            if kwargs.get(key):
                return str(kwargs[key])
        # 真机形态：message 顶层 session_id（napcat，回执发送目标）
        if message.get("session_id"):
            return str(message["session_id"])
        mi = message.get("message_info") or {}
        if isinstance(mi, dict):
            for key in ("stream_id", "chat_id", "group_id"):
                if mi.get(key):
                    return str(mi[key])
            gi = mi.get("group_info")
            if isinstance(gi, dict):
                for key in ("group_id", "id"):
                    if gi.get(key):
                        return str(gi[key])
        return ""

    async def _handle_file(self, session_id: str, conversation_id: str, stream_id: str, fdata: dict[str, Any]) -> None:
        """后台：写临时文件 → 解析 → 向量化 → 落库 → 回执。

        embedding 失败（真机 14:27 日志：服务拥塞持续超 6s 退避窗口）时：
        已解析文本放入后台重试队列，由 _retry_loop 定时重试入库，对用户透明。
        """
        name = str(fdata.get("name", "unnamed_file"))
        vs = self._store.get_or_create(
            session_id,
            conversation_id,
            self._embed,
            self.config.reader.chunk_size,
            self.config.reader.chunk_overlap,
            self.config.reader.retrieve_top_k,
            chunk_merge=bool(self.config.reader.chunk_merge_enabled),
            retrieve_use_matrix=bool(self.config.reader.retrieve_use_matrix),
            lexical_max_tokens=int(self.config.reader.lexical_max_tokens or 0),
        )
        tmp_path: Optional[str] = None
        queued_for_retry = False  # v1.1.0：入队后临时文件归重试队列持有，finally 不再删
        started_at = time.time()  # v1.3.0：大文件回执要用到处理耗时
        napcat_path: Optional[str] = None  # v1.3.1：NapCat 命中的真实路径，用于回填文件名
        # 大小上限提前算好：下载路径在下载前/下载中就要用它做预检（v1.1.0）
        max_bytes = self.config.reader.max_file_size * 1024 * 1024
        try:
            # 取文件字节
            # v1.1.0：本地路径读文件包 to_thread——100MB 级同步 read_bytes 会冻结整个事件循环
            raw_bytes = fdata.get("bytes")
            if raw_bytes is None and fdata.get("path"):
                raw_bytes = await asyncio.to_thread(Path(str(fdata["path"])).read_bytes)
            if raw_bytes is None and fdata.get("url"):
                try:
                    raw_bytes = await self._download(str(fdata["url"]), max_bytes)
                except _DownloadTooLargeError as e:
                    await self._reply_error(
                        stream_id,
                        f"⚠️ 文件「{name}」大小约 {e.size_mb:.1f}MB 超过上限 {self.config.reader.max_file_size}MB，已跳过。",
                        reason="too_large", session_id=session_id, file_name=name,
                    )
                    return
                except _UrlRejectedError as e:
                    # v1.2.0：URL 来源的文件被安全策略拦下——回执里说清"是没有下载权限"，
                    # 而不是笼统的"读不到内容"（后者会让人以为插件坏了）
                    await self._reply_error(
                        stream_id,
                        self._reject_reply_text(name, e),
                        reason=f"url_rejected:{e.reason}", session_id=session_id, file_name=name,
                    )
                    return
            if raw_bytes is None and fdata.get("file_id") and self.config.napcat.enabled:
                # v1.3.2：结构化 file 段直通（SnowLuma v1.1.10+）。段里只有 file_id，
                # 直接用 `get_file` 取路径，省掉 `get_msg` 回溯，也不依赖 size 消歧。
                napcat_path = await asyncio.to_thread(self._napcat_resolve_path, fdata)
                if napcat_path:
                    raw_bytes = await asyncio.to_thread(Path(str(napcat_path)).read_bytes)
            if raw_bytes is None and fdata.get("napcat_message_id"):
                # 适配器降级形态：拿 message_id 回头找 NapCat 要原始文件
                if not self.config.napcat.enabled:
                    await self._reply_error(
                        stream_id,
                        f"⚠️ 文件「{name}」读不到内容：适配器把文件降级成了纯文本，"
                        "请在插件配置里启用「NapCat 兜底」并填好 HTTP 地址与 token。",
                        reason="napcat_disabled", session_id=session_id, file_name=name,
                    )
                    return
                raw_bytes, napcat_path = await asyncio.to_thread(self._resolve_via_napcat, fdata)
                if raw_bytes is None and napcat_path:
                    raw_bytes = await asyncio.to_thread(Path(str(napcat_path)).read_bytes)
            if raw_bytes is None:
                if fdata.get("napcat_message_id") or fdata.get("file_id"):
                    await self._reply_error(
                        stream_id,
                        f"⚠️ 文件「{name}」无法读取：NapCat 未返回文件内容"
                        "（检查 HTTP 服务/token/message_id/file_id 是否对得上，或配置 cache_dir）。",
                        reason="napcat_no_content", session_id=session_id, file_name=name,
                    )
                else:
                    await self._reply_error(
                        stream_id,
                        f"⚠️ 文件「{name}」无法读取：消息里没有文件内容，也没有可用于回溯的 message_id。",
                        reason="no_content", session_id=session_id, file_name=name,
                    )
                return

            # 大小检查（base64/本地路径来源在此兜底；URL 来源已在下载中预检）
            if len(raw_bytes) > max_bytes:
                await self._reply_error(
                    stream_id,
                    f"⚠️ 文件「{name}」大小 {len(raw_bytes) / 1024 / 1024:.1f}MB 超过上限 {self.config.reader.max_file_size}MB，已跳过。",
                    reason="too_large", session_id=session_id, file_name=name,
                )
                return

            # v1.3.1：文件名回填（真机 11-02 21:26 故障）。SnowLuma 适配器 v1.1.10 起把
            # file 段转成结构化 {name,size,file_id,busid}，而 NapCat 私聊 file 段只有
            # `file`（真实文件名）没有 `name` ⇒ 适配器落到占位符「附件」；再叠加
            # Host 1.3.16+ 文本尾部带 `file_ref`，降级解析还会整行吞名。两条路径都会让
            # 扩展名为空，解析器按扩展名分派直接拒收（"不支持的文件类型: .(无扩展名)"）。
            # 但 NapCat `get_file` 已返回带真实文件名的路径，用它的 basename 回填最准，
            # 同时让入库索引 / 回执 / 后续检索都显示真名。
            if napcat_path:
                real_name = Path(str(napcat_path)).name
                if real_name and not get_extension(name) and get_extension(real_name):
                    self.ctx.logger.info("[napcat] 文件名回填: %r → %r", name, real_name)
                    name = real_name

            # 写临时文件
            suffix = Path(name).suffix or ""
            fd, tmp_path = tempfile.mkstemp(suffix=suffix)
            with os.fdopen(fd, "wb") as f:
                f.write(raw_bytes)
            # 解析前释放原始字节引用，降低瞬时内存峰值（raw_bytes + 文本 + 向量曾同时驻留）
            raw_bytes = None

            # 解析
            text = await asyncio.to_thread(read_any_file_to_text, tmp_path, name)

            # 向量化 + 入库
            try:
                entry = await vs.add_file(name, text)
            except RuntimeError as e:
                # embedding 为空/超时：放入后台重试队列（v1.0.11）
                # 注意：ValueError（内容为空）不在此列，走下方通用分支
                # v1.1.0：队列条目存临时文件路径而非全文，临时文件转交队列持有
                await self._enqueue_embed_retry(session_id, conversation_id, stream_id, name, str(tmp_path))
                queued_for_retry = True
                self.ctx.logger.warning("文件 %s 入库失败（embedding），已入后台重试队列: %s", name, e)
                return
            self.ctx.logger.info(
                "已入库文件 %s：%d 块（session=%s）",
                name,
                len(entry.chunks),
                session_id,
            )
            # v1.0.17：大文件后台预热概要（不阻塞回执；注入时 _ensure_summary 也会懒生成，这里只是让首次提问即命中）
            # v1.0.18 C4：改走 _ensure_summary_async（in-flight 去重 + 失败冷却），注入触发时不会重复起任务
            # v1.0.18 C1：阈值判定用 entry.total_chars()（原文长度），overlap 合并会让 join(chunks) 膨胀
            if (
                bool(self.config.reader.summary_enabled)
                and entry.total_chars() > max(1, int(self.config.reader.direct_inject_max_chars))
                and not entry.summary
            ):
                self._ensure_summary_async(entry)
            # v1.0.18 C3：文件变动必须失效注入缓存，否则同一问题会复用旧文件的注入内容
            self._memo_invalidate_session(session_id)
            # v1.3.0：同名文件这次读成功了，清掉之前留下的失败通知
            self._clear_notices(session_id, name)
            # v1.3.0 大文件回执：入库耗时动辄几十秒，全程静默会让用户以为插件没反应。
            # 只在「真的大」或「真的慢」时发一条中性回执，小文件依旧完全静默。
            try:
                ack_min = int(self.config.reader.ack_min_chars or 0)
            except (TypeError, ValueError):
                ack_min = 0
            elapsed = time.time() - started_at
            if ack_min > 0 and stream_id and (entry.total_chars() >= ack_min or elapsed >= 10.0):
                await self._reply(stream_id, f"已读取《{name}》，共 {len(entry.chunks)} 段")
            if not self.config.reader.silent_success:
                await self._reply(
                    stream_id,
                    f"📄 已解析「{name}」，切成 {len(entry.chunks)} 块并向量化。现在可以直接问我文件内容了。",
                )
        except ValueError as e:
            await self._reply_error(stream_id, f"⚠️ {e}", reason="value_error", session_id=session_id, file_name=name)
        except RuntimeError as e:
            await self._reply_error(stream_id, f"⚠️ {e}", reason="runtime_error", session_id=session_id, file_name=name)
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("处理文件 %s 失败: %s", name, e, exc_info=True)
            await self._reply_error(
                stream_id,
                f"⚠️ 处理文件「{name}」失败：{type(e).__name__}",
                reason="exception",
                session_id=session_id,
                file_name=name,
            )
        finally:
            if tmp_path and not queued_for_retry:
                try:
                    os.remove(tmp_path)
                except OSError:
                    pass

    def _napcat_ssl_context(self) -> ssl.SSLContext:
        """NapCat urllib SSL 上下文缓存（v1.1.0）：create_default_context 是 ms 级 CPU 调用，
        旧版每次 _napcat_post 新建（每文件 2~3 次请求）；verify_ssl 配置变化时重建。"""
        verify = bool(self.config.napcat.verify_ssl)
        if self._napcat_ssl_ctx is None or self._napcat_ssl_verify != verify:
            ctx = ssl.create_default_context()
            if not verify:
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
            self._napcat_ssl_ctx = ctx
            self._napcat_ssl_verify = verify
        return self._napcat_ssl_ctx

    # ─── NapCat HTTP 兜底（同步实现，调用处包 asyncio.to_thread） ───
    def _napcat_post(self, endpoint: str, payload: dict[str, Any]) -> Optional[Any]:
        """调 NapCat OneBot HTTP API，返回 data 字段；失败返回 None。"""
        cfg = self.config.napcat
        url = f"{str(cfg.http_url).rstrip('/')}/{endpoint}"
        body = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            url, data=body, headers={"Content-Type": "application/json"}, method="POST"
        )
        token = (cfg.access_token or "").strip()
        if token:
            req.add_header("Authorization", f"Bearer {token}")
        ctx = self._napcat_ssl_context()
        try:
            with urllib.request.urlopen(req, timeout=cfg.timeout, context=ctx) as resp:
                data = json.loads(resp.read().decode("utf-8", errors="replace"))
        except urllib.error.HTTPError as e:
            self.ctx.logger.error("[napcat] %s HTTP %s: %s", endpoint, e.code, e.read()[:200])
            return None
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("[napcat] %s 请求失败: %s: %s", endpoint, type(e).__name__, e)
            return None
        if not isinstance(data, dict):
            return None
        status = str(data.get("status", ""))
        retcode = data.get("retcode")
        if status == "ok" or retcode == 0 or (status == "" and retcode is None and "data" in data):
            return data.get("data")
        self.ctx.logger.error(
            "[napcat] %s 返回异常: status=%s retcode=%s wording=%s",
            endpoint,
            status,
            retcode,
            data.get("wording") or data.get("message"),
        )
        return None

    def _napcat_get_msg(self, message_id: Any) -> Optional[dict[str, Any]]:
        """拿原始 OneBot 事件。message_id 形态未知，先原样试，失败再转 int 重试。"""
        if message_id in (None, ""):
            self.ctx.logger.error("[napcat] 消息没有 message_id，无法回溯原始事件")
            return None
        candidates: list[Any] = [message_id]
        try:
            as_int = int(str(message_id))
            if as_int != message_id:
                candidates.append(as_int)
        except (TypeError, ValueError):
            pass
        for cand in candidates:
            data = self._napcat_post("get_msg", {"message_id": cand})
            if isinstance(data, dict):
                self.ctx.logger.info(
                    "[napcat] get_msg 成功: message_id=%r raw_keys=%s",
                    cand,
                    sorted(data.keys()) if isinstance(data, dict) else "-",
                )
                return data
        return None

    @staticmethod
    def _pick_onebot_file_segs(msg_data: dict[str, Any]) -> list[dict[str, Any]]:
        """从原始事件的 message 段数组里取出**全部** file 段（OneBot：{type, data:{...}}）。

        v1.3.0：旧版 `_pick_onebot_file_seg` 只返回第一个 file 段。防抖合并后 N 条
        `[文件]` 提示共用同一个 message_id（消息体里却是 N 个 file 段），于是 N 个
        文件全被读成第一个文件的内容、再按各自文件名分别入库——静默串档，且污染检索
        （库里会出现 N 份「同一份内容 + N 个不同文件名」）。
        """
        segs = msg_data.get("message")
        if isinstance(msg_data.get("data"), dict) and "message" in msg_data["data"]:
            segs = msg_data["data"]["message"]
        if not isinstance(segs, list):
            return []
        out: list[dict[str, Any]] = []
        for seg in segs:
            if not isinstance(seg, dict):
                continue
            if str(seg.get("type", "")).lower() != "file":
                continue
            data = seg.get("data")
            out.append(data if isinstance(data, dict) else seg)
        return out

    @staticmethod
    def _match_file_seg(
        segs: list[dict[str, Any]], name: str, size_hint: int = 0
    ) -> tuple[Optional[dict[str, Any]], str]:
        """在多个 file 段里挑出与 (文件名, 大小) 对应的那一个。

        返回 `(段, 匹配方式)`，匹配方式 ∈ {name, ambiguous, fallback, none}，
        由调用方打日志。串档是**静默故障**——不把匹配方式打出来，真机上根本无从排查。

        匹配规则：
          1. `size_hint` 与段的 `file_size` 都存在且不等 → 排除该段（同名文件靠大小消歧）；
          2. 文件名按 `Path(x).name.lower()` 精确比较（段里可能是带路径的名字）；
             候选字段依次取 `file_name` / `name` / `file` / `title`（适配器命名不统一）；
          3. 唯一命中 → `name`；多个命中 → 取第一个并标 `ambiguous`；
          4. 名字全未命中但 `size_hint` 唯一命中一个段 → 采用并标 `size`
             （v1.3.1：适配器占位名「附件」的场景靠这条消歧）；
          5. 仍无解 → 回退第一个并标 `fallback`（保留旧行为，但可观测）。
        """
        if not segs:
            return None, "none"
        target = Path(str(name or "")).name.lower()
        try:
            hint = int(size_hint or 0)
        except (TypeError, ValueError):
            hint = 0

        hits: list[dict[str, Any]] = []
        sized: list[dict[str, Any]] = []  # 通过 size 过滤的候选（v1.3.1：占位名的消歧依据）
        for seg in segs:
            if not isinstance(seg, dict):
                continue
            try:
                seg_size = int(seg.get("file_size") or seg.get("size") or 0)
            except (TypeError, ValueError):
                seg_size = 0
            if hint > 0 and seg_size > 0 and seg_size != hint:
                continue
            sized.append(seg)
            if not target:
                hits.append(seg)
                continue
            for key in ("file_name", "name", "file", "title"):
                cand = seg.get(key)
                if not cand:
                    continue
                if Path(str(cand)).name.lower() == target:
                    hits.append(seg)
                    break
        if not hits:
            # v1.3.1：占位名（SnowLuma 把没有 name 的 file 段兜底成「附件」）永远匹配不上
            # 段内真实文件名，但 size_hint 通常能唯一锁定那一个段 —— 直接采用并标 `size`。
            # 不这么做的话：单文件消息每次文件都打 fallback WARNING（噪音），
            # 多文件消息则退回「取第一个」而可能串档。
            if hint > 0 and len(sized) == 1:
                return sized[0], "size"
            return segs[0], "fallback"
        if len(hits) > 1:
            return hits[0], "ambiguous"
        return hits[0], "name"

    def _napcat_resolve_path(self, file_seg: dict[str, Any]) -> Optional[str]:
        """file 段 → 本地路径：优先 get_file(file_id)，退化到段内路径。"""
        file_id = file_seg.get("file_id") or file_seg.get("id")
        if file_id:
            data = self._napcat_post("get_file", {"file_id": str(file_id)})
            if isinstance(data, dict):
                p = data.get("file") or data.get("path")
                if p and Path(str(p)).exists():
                    self.ctx.logger.info("[napcat] get_file 命中: %s", p)
                    return str(p)
                self.ctx.logger.warning("[napcat] get_file 返回路径不可用: %r", p)
            elif isinstance(data, str) and Path(data).exists():
                return data
        # 退路：段里的 file 字段本身就是本地路径
        for key in ("path", "file_path", "file"):
            val = file_seg.get(key)
            if val and Path(str(val)).exists():
                return str(val)
        return None

    def _search_cache_dir(self, name: str, size_hint: int) -> Optional[str]:
        """最后兜底：在 NapCat 缓存目录按 文件名+大小 精确匹配。"""
        root = (self.config.napcat.cache_dir or "").strip()
        if not root or not size_hint:
            return None
        base = Path(root)
        if not base.is_dir():
            self.ctx.logger.warning("[napcat] cache_dir 不存在: %s", root)
            return None
        try:
            for f in base.rglob(Path(name).name):
                if f.is_file() and f.stat().st_size == size_hint:
                    self.ctx.logger.info("[napcat] cache_dir 命中: %s", f)
                    return str(f)
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("[napcat] cache_dir 扫描失败: %s", e)
        return None

    def _resolve_via_napcat(self, fdata: dict[str, Any]) -> tuple[Optional[bytes], Optional[str]]:
        """同步：拿 NapCat 原始事件 → file 段 → 文件内容。返回 (bytes, path)。"""
        name = str(fdata.get("name", ""))
        size_hint = int(fdata.get("size_hint") or 0)
        msg_data = self._napcat_get_msg(fdata.get("napcat_message_id"))
        segs = self._pick_onebot_file_segs(msg_data) if isinstance(msg_data, dict) else []
        file_seg, match_mode = self._match_file_seg(segs, name, size_hint)
        if file_seg is None:
            self.ctx.logger.error(
                "[napcat] 原始事件里没有 file 段: message_id=%r data_keys=%s",
                fdata.get("napcat_message_id"),
                sorted(msg_data.keys()) if isinstance(msg_data, dict) else None,
            )
            # 原始事件都翻不到时，退到缓存目录
            path = self._search_cache_dir(name, size_hint)
            return (None, path)
        # v1.3.0：匹配方式必须可见。取错段是静默串档——只在 fallback/ambiguous 时报 WARNING，
        # 正常命中（name / v1.3.1 新增的 size 消歧）走 INFO，避免日志噪音淹没真正的异常。
        log_fn = self.ctx.logger.info if match_mode in ("name", "size") else self.ctx.logger.warning
        log_fn(
            "[napcat] 文件段匹配: name=%s mode=%s segs=%d",
            name,
            match_mode,
            len(segs),
        )
        b64 = file_seg.get("base64")
        if b64:
            try:
                return (base64.b64decode(b64), None)
            except Exception as e:  # noqa: BLE001
                self.ctx.logger.error("[napcat] base64 解码失败: %s", e)
        path = self._napcat_resolve_path(file_seg)
        if not path:
            path = self._search_cache_dir(name, size_hint)
        return (None, path)

    def _download_verify_setting(self) -> Any:
        """下载用的 TLS 校验设置（v1.2.0）：默认系统 CA；配了 CA 证书就用该路径；
        只有显式开 insecure_download 才彻底关闭校验。

        顺序刻意如此——运行机器有 TLS MITM 代理时，正确做法是把代理根证书路径填进
        download_ca_bundle（校验照常生效），而不是关掉校验。
        """
        if bool(self.config.reader.insecure_download):
            return False
        bundle = str(self.config.reader.download_ca_bundle or "").strip()
        if bundle:
            if Path(bundle).is_file():
                return bundle
            self.ctx.logger.error(
                "download_ca_bundle 不是有效文件，已忽略并回到系统默认证书: %s", bundle
            )
        return True

    async def _get_http_client(self) -> Any:
        """下载用 httpx 单例（v1.1.0）：连接池复用，免每次下载新建 client 与 SSL 上下文。
        insecure_download / download_ca_bundle 变化时关闭旧 client 重建。

        v1.2.0：显式 `follow_redirects=False`——重定向改由 _download 手动逐跳处理，
        每一跳都要重新过域名+地址校验（否则白名单域名可以 302 到 169.254.169.254）。
        """
        import httpx  # type: ignore

        verify = self._download_verify_setting()
        if self._http_client is not None and self._http_client_verify != verify:
            try:
                await self._http_client.aclose()
            except Exception:  # noqa: BLE001
                pass
            self._http_client = None
        if self._http_client is None:
            self._http_client = httpx.AsyncClient(
                timeout=30, verify=verify, follow_redirects=False
            )
            self._http_client_verify = verify
        return self._http_client

    async def _download(self, url: str, max_bytes: int) -> Optional[bytes]:
        """流式下载文件（v1.1.0 重写，v1.2.0 加 SSRF 防护 + 逐跳重定向，v1.2.1 连接钉定）。

        v1.1.0 已解决的内存问题：Content-Length 预检 + 边下边累计，超限立即中断。

        v1.2.0 新增的安全边界（URL 由消息正文控制，见文件顶部 URL 校验段说明）：
          - 请求前：validate_download_target 校验 scheme/凭据/域名白名单/解析后 IP 必须公网
          - 重定向：关闭 httpx 自动跟随，改为手动逐跳——**每一跳都重新校验**，
            并限制跳数（download_max_redirects）。这样即使白名单域名被 302 到
            169.254.169.254 也会在下一跳被拦下。
          - 失败原因全部落日志（含具体 host），但 URL 的 query 部分打码——
            QQ 文件链接的 query 里带下载凭据。

        v1.2.1 新增（review SSRF 项）：**连接钉定，关闭 DNS 重绑定窗口**。
        旧版校验时解析一次 DNS，把 URL 交给 httpx 后连接时会**再解析一次**——
        两次解析之间域名可以换 IP：download_url_policy=public 时，攻击者自持
        DNS 的域名能先解析到公网 IP 过检、连接时再解析回内网 IP。现在校验通过
        的解析结果经 collect_ips 带出，实际请求直接连这些 IP（build_pinned_request：
        Host 头/SNI/证书校验仍按原主机名），重定向每一跳独立「校验 → 钉定」。
        policy=off 不解析也就无从钉定，维持旧行为按主机名连接。

        连接细节：
          - 多个已校验 IP 依次尝试，只有连接类错误（ConnectError/ConnectTimeout）
            才换下一个，绝不重新解析 DNS；
          - 阻塞的 getaddrinfo 挪进 asyncio.to_thread，不再卡事件循环。

        运行机器若存在 TLS MITM 代理（proxy-root-ca.cer），HTTPS 会报
        CERTIFICATE_VERIFY_FAILED；建议配 reader.download_ca_bundle 指向代理根证书
        （校验保持开启），万不得已再用 reader.insecure_download 跳过校验。
        """
        try:
            import httpx  # type: ignore  # noqa: F401
        except ImportError:
            self.ctx.logger.error("下载文件需要 httpx，请安装：pip install httpx")
            return None

        policy = normalize_download_policy(self.config.reader.download_url_policy)
        allowed_hosts = parse_allowed_hosts(self.config.reader.download_allowed_hosts)
        allow_private = bool(self.config.reader.download_allow_private_hosts)
        try:
            max_hops = max(0, int(self.config.reader.download_max_redirects))
        except (TypeError, ValueError):
            max_hops = 3

        current = str(url or "")
        try:
            client = await self._get_http_client()
            for hop in range(max_hops + 1):
                # ① 每一跳都先校验目标（首跳=原始 URL，后续=上一跳的 Location），
                #    校验依据的解析结果经 collect_ips 带出来，供 ② 钉定连接（v1.2.1）。
                #    顺带：getaddrinfo 是阻塞调用，挪进线程池不再卡事件循环。
                resolved: list[str] = []
                target, host = await asyncio.to_thread(
                    validate_download_target,
                    current,
                    policy=policy,
                    allowed_hosts=allowed_hosts,
                    allow_private_hosts=allow_private,
                    collect_ips=resolved,
                )
                # ② 连接钉定（v1.2.1 防 DNS 重绑定）：只连上面已通过校验的 IP，
                #    httpx 连接时不再二次解析。多个已校验 IP 依次尝试，**只有连接类
                #    错误才换下一个，绝不重新解析 DNS**。resolved 为空（policy=off
                #    不解析，无从钉定）时保持旧行为按主机名连接。
                candidate_ips: list[Optional[str]] = list(resolved) or [None]
                next_url: Optional[str] = None
                result: Optional[bytes] = None
                connect_err: Optional[Exception] = None
                for ip in candidate_ips:
                    req_url, req_extra = build_pinned_request(target, host, ip)
                    try:
                        async with client.stream("GET", req_url, **req_extra) as resp:
                            # ③ 重定向：手动接管，绝不自动跟随
                            if resp.is_redirect:
                                location = resp.headers.get("location")
                                if not location:
                                    self.ctx.logger.warning(
                                        "[下载拒绝] %s 返回 %s 但没有 Location 头", host, resp.status_code
                                    )
                                    return None
                                if hop >= max_hops:
                                    self.ctx.logger.warning(
                                        "[下载拒绝] 重定向超过 %d 跳（最后一跳 %s → %s）",
                                        max_hops,
                                        host,
                                        redact_url(location),
                                    )
                                    return None
                                # 相对 Location 按当前 URL 解析为绝对地址，供下一轮重新校验
                                next_url = urljoin(target, location)
                                self.ctx.logger.info(
                                    "[下载] 第 %d 跳重定向：%s → %s", hop + 1, host, redact_url(next_url)
                                )
                                break
                            resp.raise_for_status()
                            # ④ 预检：Content-Length 已超限直接中断（不必开始下载）
                            cl = resp.headers.get("content-length")
                            if cl and cl.isdigit() and int(cl) > max_bytes:
                                raise _DownloadTooLargeError(float(cl))
                            chunks: list[bytes] = []
                            total = 0
                            async for chunk in resp.aiter_bytes(65536):
                                total += len(chunk)
                                if total > max_bytes:
                                    raise _DownloadTooLargeError(float(total))
                                chunks.append(chunk)
                            result = b"".join(chunks)
                            break
                    except (httpx.ConnectError, httpx.ConnectTimeout) as e:
                        # 连接类错误才换下一个已校验 IP；读流/状态码/体积异常不在此列
                        connect_err = e
                        continue
                else:
                    self.ctx.logger.error(
                        "下载 %s 失败：%d 个已校验地址均连接失败（%s: %s）",
                        redact_url(target),
                        len(candidate_ips),
                        type(connect_err).__name__ if connect_err else "-",
                        connect_err or "",
                    )
                    return None
                if next_url is not None:
                    current = next_url
                    continue
                return result
            return None
        except _DownloadTooLargeError:
            raise
        except _UrlRejectedError as e:
            # 安全拦截：向上抛，让 _handle_file 回一句可读的拒绝原因（而不是"读不到内容"）
            self.ctx.logger.warning(
                "[下载拒绝] URL 未通过安全校验（%s）: %s", e.reason, redact_url(current)
            )
            raise
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("下载 %s 失败: %s", redact_url(current), e)
            return None

    def _reject_reply_text(self, name: str, err: _UrlRejectedError) -> str:
        """把 URL 拒绝原因翻译成一句人话（不透露解析到的内网地址等细节）。"""
        head = f"⚠️ 文件「{name}」没有下载："
        reason = err.reason
        if reason == "host_not_allowed":
            return head + "文件链接不在允许下载的域名白名单内（默认只允许 QQ 文件域）。"
        if reason == "allowlist_empty":
            return head + "插件未配置任何允许下载的域名（download_allowed_hosts 为空）。"
        if reason == "non_public_address":
            return head + "文件链接指向内网/本机地址，出于安全已拒绝下载。"
        if reason == "bad_scheme":
            return head + "文件链接的协议不受支持（仅支持 http/https）。"
        if reason == "userinfo_not_allowed":
            return head + "文件链接格式异常（含内嵌凭据），已拒绝下载。"
        if reason in ("dns_failed", "dns_empty"):
            return head + "文件链接的域名解析失败。"
        if reason == "bad_url":
            return head + "文件链接无法解析。"
        return head + f"文件链接未通过安全校验（{reason}）。"

    async def _reply(self, stream_id: str, text: str) -> None:
        if not stream_id:
            self.ctx.logger.warning("无 stream_id，跳过回执: %s", text[:50])
            return
        try:
            await self.ctx.send.text(text, stream_id)
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("回执发送失败: %s", e)

    #: reason → 面向模型的失败原因（v1.3.0：失败要能被模型说清楚，而不是只进日志）
    _REASON_TEXT = {
        "too_large": "文件超过大小上限",
        "napcat_disabled": "适配器把文件降级成了纯文本，且未启用 NapCat 兜底",
        "napcat_no_content": "NapCat 未返回文件内容",
        "no_content": "消息里没有文件内容",
        "embed_failed": "向量化失败",
        "exception": "处理过程异常",
    }

    async def _reply_error(
        self,
        stream_id: str,
        text: str,
        *,
        reason: str,
        session_id: str = "",
        file_name: str = "",
    ) -> None:
        """错误回执收口（v1.0.20）：silent_errors 开启时只写日志不发群，
        避免批量发文件时 ⚠️ 刷屏；日志带 reason 便于 /file_status 排查。

        v1.3.0：静默模式下失败**仍然要登记模型通知**——旧版失败只写日志，
        用户和模型都不知道发生过什么，用户追问时模型只能凭空猜。
        """
        self.ctx.logger.warning("文件处理失败（silent_errors=%s）: %s | %s",
                                bool(self.config.reader.silent_errors), reason, text)
        if session_id and file_name and bool(self.config.reader.silent_errors):
            detail = self._REASON_TEXT.get(reason, "")
            if reason.startswith("url_rejected"):
                detail = f"下载被安全策略拒绝（{reason.split(':', 1)[-1]}）"
            elif not detail:
                # value_error / runtime_error 的 text 本身就是「⚠️ 具体原因」
                detail = str(text).lstrip("⚠️ ").strip() or "读取失败"
            self._add_notice(session_id, file_name, detail)
        if not bool(self.config.reader.silent_errors):
            await self._reply(stream_id, text)

    # ─── LLM 上下文注入（before_model_request hook） ─────────────
    # v1.0.16 教训：辅助方法必须放在 @HookHandler/@Tool 装饰器之前——
    # 装饰器绑定的是紧随其后的函数，曾把 _full_text_within_limit 注册成
    # inject_file_context hook 本体（本地直调测试发现不了，真机必炸）。

    def _memo_get(self, session_id: str, query_key: str) -> Optional[str]:
        """查注入去重缓存：TTL 内命中返回上次注入文本，否则 None。"""
        if not bool(self.config.reader.inject_memo_enabled):
            return None
        item = self._inject_memo.get((session_id, query_key))
        if item is None:
            return None
        ts, text = item
        ttl = max(1, int(self.config.reader.inject_memo_ttl))
        if time.time() - ts > ttl:
            self._inject_memo.pop((session_id, query_key), None)
            return None
        return text

    def _memo_put(self, session_id: str, query_key: str, text: str) -> None:
        """写注入去重缓存：LRU 上限 128 条，超限先淘汰最旧。"""
        if not bool(self.config.reader.inject_memo_enabled) or not text:
            return
        if len(self._inject_memo) >= 128:
            oldest_key = min(self._inject_memo, key=lambda k: self._inject_memo[k][0])
            self._inject_memo.pop(oldest_key, None)
        self._inject_memo[(session_id, query_key)] = (time.time(), text)

    def _memo_invalidate_session(self, session_id: str) -> None:
        """文件变动后失效该会话的全部注入缓存（防止注入陈旧内容）。"""
        for key in [k for k in self._inject_memo if k[0] == session_id]:
            self._inject_memo.pop(key, None)

    def _tick_rounds_once(
        self,
        vs: Any,
        query_key: str = "",
        session_id: str = "",
        file_names: Optional[list] = None,
    ) -> bool:
        """轮数去重（v1.0.18 C3，v1.3.0 修订）：同一会话同一问题在 120s 窗口内只计一轮。

        返回是否真的计了一轮。

        v1.3.0 两处修正：
          - key 加入 `session_id`：旧 key 不含会话，A 会话问过「总结一下」之后 B
            会话问同样的话会被误判成"本轮已计数"，轮数不再增长（表现为文件永不过期）；
          - 只给 `file_names` 里的文件计轮：旧版给会话内**所有**文件 +1，没被问到
            的文件也跟着提前过期。

        旧逻辑的问题（v1.0.18 已修一半）：hook 与 tool 各 +1、且每次 attempt 都 +1，
        5 轮配额实际 2-3 轮就用光 → 文件提前过期 → 用户重发 → 重新解析 + 重新 embedding。
        """
        key = ("__round__", session_id, query_key.strip() or str(time.time() // 120))
        item = self._inject_memo.get(key)
        now = time.time()
        if item is not None and now - item[0] < 120:
            return False  # 同一问题本轮已计数
        if len(self._inject_memo) >= 128:
            oldest_key = min(self._inject_memo, key=lambda k: self._inject_memo[k][0])
            self._inject_memo.pop(oldest_key, None)
        self._inject_memo[key] = (now, "")
        if file_names:
            for name in file_names:
                vs.increment_rounds(name)
        else:
            vs.increment_rounds()
        return True

    def _full_text_within_limit(self, vs: Any) -> Optional[str]:
        """直注判定：会话内所有文件全文合计 ≤ direct_inject_max_chars 时返回拼接全文，否则 None。

        小文件全文直注（v1.0.14）：检索只取 Top-K 块，小文件场景反而丢内容——
        全文直接可见比「相关度排序」更符合预期；同时省一次查询 embedding 调用
        （拥塞期查询也会超时，直注路径完全绕开）。
        """
        limit = int(self.config.reader.direct_inject_max_chars)
        if limit <= 0 or not vs.files:
            return None
        # v1.0.18：阈值判定用原文长度（source_chars），overlap 合并会让 join(chunks) 膨胀约 20%
        total = sum(entry.total_chars() for entry in vs.files.values())
        if total > limit:
            return None
        parts: list[str] = []
        for fname, entry in vs.files.items():
            parts.append(f"《{fname}》全文：\n{entry.full_text()}")
        return "\n\n".join(parts)

    @staticmethod
    def _sample_text_for_summary(entry: Any, max_chars: int) -> str:
        """为概要生成取样源文本：头 / 中 / 尾三段均匀取样拼接。

        大文件全文直接喂摘要 LLM token 开销大且易撞上下文限制；
        首尾段保留开头与结尾信息，中段保留主体内容。
        """
        full = entry.full_text()
        if len(full) <= max_chars:
            return full
        seg = max_chars // 3
        head = full[:seg]
        mid_start = max(0, (len(full) - seg) // 2)
        mid = full[mid_start : mid_start + seg]
        tail = full[-seg:]
        return f"{head}\n……（中略）……\n{mid}\n……（中略）……\n{tail}"

    async def _generate_file_summary(self, entry: Any) -> str:
        """为单个大文件生成 LLM 概要；失败返回空串（静默降级，不影响检索）。"""
        source_chars = max(1000, int(self.config.reader.summary_source_chars))
        max_len = max(100, int(self.config.reader.summary_max_chars))
        source = self._sample_text_for_summary(entry, source_chars)
        if not source.strip():
            return ""
        prompt = (
            "请为以下文档内容写一段简洁的中文概要，概括文档的主题、结构与主要内容，"
            f"不超过 {max_len} 字，不要逐句罗列，直接输出概要正文：\n\n"
            f"文档名：{entry.file_name}\n\n{source}"
        )
        try:
            result = await self.ctx.llm.generate(prompt=prompt, max_tokens=max_len)
            if not isinstance(result, dict) or not result.get("success"):
                self.ctx.logger.warning(
                    "文件 %s 概要生成失败: %s", entry.file_name, result if not isinstance(result, dict) else result.get("reasoning", "unknown")
                )
                return ""
            text = str(result.get("response", "")).strip()
            if len(text) > max_len:
                text = text[:max_len]
            self.ctx.logger.info(
                "已生成文件 %s 的概要（%d 字）", entry.file_name, len(text)
            )
            return text
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.warning("文件 %s 概要生成异常（静默降级）: %s", entry.file_name, e)
            return ""

    def _summary_cooling(self, entry: Any) -> bool:
        """概要失败冷却判断（v1.0.18 C4）：冷却期内不再尝试生成。"""
        interval = float(self.config.reader.summary_retry_interval)
        if interval <= 0:
            return False
        last_fail = self._summary_failed_ts.get(entry.file_name)
        return last_fail is not None and (time.time() - last_fail) < interval

    def _ensure_summary_async(self, entry: Any) -> bool:
        """为单个大文件启动后台概要任务（v1.0.18 C4）。

        in-flight 去重（与入库预热共用）：同一 entry 不重复起任务；
        失败冷却期内跳过。返回是否实际启动了任务。
        """
        if entry.summary or entry.file_name in self._summary_pending or self._summary_cooling(entry):
            return False
        self._summary_pending.add(entry.file_name)

        async def _run() -> None:
            try:
                summary = await self._generate_file_summary(entry)
                if summary:
                    entry.summary = summary
                    self._summary_failed_ts.pop(entry.file_name, None)
                else:
                    self._summary_failed_ts[entry.file_name] = time.time()
            except Exception as e:  # noqa: BLE001
                self._summary_failed_ts[entry.file_name] = time.time()
                self.ctx.logger.warning("文件 %s 概要后台生成异常（已进入冷却）: %s", entry.file_name, e)
            finally:
                self._summary_pending.discard(entry.file_name)

        self._spawn_bg(_run())
        return True

    async def _ensure_summary(self, vs: Any) -> None:
        """注入时的概要保障（v1.0.18 C4 异步化）。

        summary_await_on_inject=true 时保持 v1.0.17 同步行为（测试/排障用）；
        默认 false：只负责触发后台生成（in-flight + 冷却去重），不阻塞注入，
        概要缺失的本次注入仍可正常检索，下次提问自然带上概要。
        """
        if not bool(self.config.reader.summary_enabled):
            return
        limit = max(1, int(self.config.reader.direct_inject_max_chars))
        await_now = bool(self.config.reader.summary_await_on_inject)
        for entry in vs.files.values():
            # v1.0.18：阈值判定用原文长度
            if entry.total_chars() <= limit or entry.summary:
                continue
            if await_now:
                if self._summary_cooling(entry):
                    continue
                summary = await self._generate_file_summary(entry)
                if summary:
                    entry.summary = summary
                else:
                    self._summary_failed_ts[entry.file_name] = time.time()
            else:
                self._ensure_summary_async(entry)

    # ─── 混合检索（v1.3.0 批次 2） ──────────────────────────────
    # 以下都是普通辅助方法：必须排在 inject_file_context 的 @HookHandler 之前
    # （装饰器只认紧贴其后的 def，插队会让组件静默绑错人）。
    _RE_REFERENCE = re.compile(r"这|那|它|上面|刚才|前面|该|此|第[一二三四五六七八九十百\d]{1,3}[章节页条]")

    def _hybrid_enabled(self) -> bool:
        """BM25 总开关：运行期命令（/file_bm25）覆盖优先于配置文件。

        为什么要有运行期覆盖：已部署实例的 config.toml 不会自动补新键，只改
        Field(default) 对老实例完全无效——体感变差时必须有一条**立刻可用**的退路。
        """
        if self._hybrid_override is not None:
            return bool(self._hybrid_override)
        return bool(self.config.reader.hybrid_retrieve_enabled)

    @staticmethod
    def _collect_user_queries(items: list, limit: int = 2) -> list[str]:
        """按时间顺序取最近 N 条用户文本（用于追问改写）。"""
        out: list[str] = []
        for item in reversed(items or []):
            if not isinstance(item, dict):
                continue
            if item.get("item_type") in ("UserMessageItem", "user"):
                for p in item.get("parts") or []:
                    if isinstance(p, dict) and p.get("type") == "text":
                        t = str(p.get("text", "")).strip()
                        if t:
                            out.append(t)
                        break
            if len(out) >= limit:
                break
        out.reverse()
        return out

    def _rewrite_query(self, items: list, current: str) -> tuple[str, str]:
        """追问改写：「那第三章呢」→「上一问 + 当前问」。返回 (改写后, 原始)。

        刻意**不调 LLM**：注入 hook 是 BLOCKING 且只有 11s 预算，为改写再搭一次
        模型调用会把超时风险直接顶满。规则触发的收益已经覆盖了绝大部分追问形态。
        """
        cur = str(current or "").strip()
        if not cur or not bool(self.config.reader.query_rewrite_enabled):
            return cur, cur
        prev_queries = self._collect_user_queries(items, limit=2)
        prev = prev_queries[0] if len(prev_queries) >= 2 else ""
        if not prev or prev == cur or cur in prev:
            return cur, cur
        need = (
            len(cur) <= 12
            or bool(self._RE_REFERENCE.search(cur))
            or (len(cur) <= 20 and cur.endswith(("呢", "吗")))
        )
        if not need:
            return cur, cur
        return f"{prev} {cur}", cur

    def _record_diag(self, diag: dict) -> None:
        """记录一条检索诊断（环形缓冲 20 条）。"""
        try:
            self._retrieval_diag.append(diag)
        except Exception:  # noqa: BLE001  诊断绝不能影响主流程
            pass

    async def _retrieve_hybrid(self, vs: Any, query: str, session_id: str = "") -> tuple[list, dict]:
        """向量 + 词法 RRF 融合检索。返回 `(hits, diag)`。

        降级矩阵（向量 / 词法）：
          | 可用 | 可用 | → RRF 融合
          | 可用 | 停用 | → 纯向量 + 阈值
          | 失败 | 可用 | → **纯词法**（v1.3.0 新增：embedding 拥塞时代码/报错码类问题仍能答）
          | 失败 | 停用 | → 空（调用方走概要兜底）
        """
        started = time.time()
        try:
            cand_k = max(1, int(self.config.reader.hybrid_candidate_k or 20))
        except (TypeError, ValueError):
            cand_k = 20
        try:
            top_k = max(1, int(self.config.reader.retrieve_top_k or 6))
        except (TypeError, ValueError):
            top_k = 6

        use_lexical = self._hybrid_enabled()
        lexical_hits = vs.search_lexical(query, cand_k) if use_lexical else []
        qvec = await self._embed_query(query)
        vector_hits = vs.retrieve(qvec, cand_k) if qvec is not None else []

        if qvec is None:
            mode = "lexical" if lexical_hits else "unavailable"
        elif not lexical_hits:
            mode = "vector"
        else:
            mode = "hybrid"

        fused = rrf_fuse(vector_hits, lexical_hits, rrf_k=int(self.config.reader.rrf_k or 60))
        gated = apply_relevance_gate(
            fused,
            min_score=float(self.config.reader.retrieve_min_score or 0.0),
            min_ratio=float(self.config.reader.retrieve_min_ratio or 0.0),
        )
        hits = gated[:top_k]

        sources: dict[str, int] = {}
        for h in hits:
            key = str(h.get("source", "?"))
            sources[key] = sources.get(key, 0) + 1
        diag = {
            "ts": time.time(),
            "session_id": session_id,
            "query": query,
            "mode": mode,
            "vector_candidates": len(vector_hits),
            "lexical_candidates": len(lexical_hits),
            "fused": len(fused),
            "after_gate": len(gated),
            "kept": len(hits),
            "top1_score": round(gated[0]["score"], 4) if gated else 0.0,
            "elapsed_ms": int((time.time() - started) * 1000),
            "sources": sources,
        }
        return hits, diag

    @HookHandler(
        "maisaka.planner.before_request",
        name="announce_file_context",
        mode=HookMode.BLOCKING,
        order=HookOrder.EARLY,
        error_policy=ErrorPolicy.SKIP,
        timeout_ms=2000,
    )
    async def announce_file_context(self, **kwargs: Any) -> dict[str, Any]:
        """向 Planner 公告「本会话已读入哪些文件」（v1.3.2）。

        真机 11-02 21:39 实测：文件全文已直注 replyer（2264 字）且答得正确，
        但 Planner 上下文里没有这份内容 —— 它据此判定「模型没读到却编了标题」，
        触发 `recall_last_message` 撤回重答。**只注入 replyer 会造成两侧信息不对称，
        Planner 的自我校对反而制造故障。**

        这里只注入一句公告（几十字），**不注入正文**：Planner 只需要知道
        「内容已有供给」，逐轮 attempt 灌全文会把 token 打爆。
        """
        try:
            session_id = str(kwargs.get("session_id") or "")
            items = kwargs.get("items")
            if not session_id or not isinstance(items, list):
                return {"action": "continue"}
            if self._already_injected(items):
                return {"action": "continue"}

            vs = None
            for (sid, _cid), candidate in self._store._sessions.items():
                if sid == session_id:
                    vs = candidate
                    break
            if vs is None or not vs.files:
                return {"action": "continue"}

            names = [str(n) for n in vs.files.keys()]
            total = sum(int(e.total_chars()) for e in vs.files.values())
            listed = "、".join(f"《{n}》" for n in names[:3])
            if len(names) > 3:
                listed += f" 等 {len(names)} 个"
            supply = (
                "全文会在生成回复时自动附上"
                if self._full_text_within_limit(vs) is not None
                else "相关片段会在生成回复时自动附上"
            )
            text = (
                f"本会话已读入文件：{listed}（合计约 {total} 字）。{supply}，"
                "规划阶段不需要再调用文件读取工具；需要精确定位某段时用 search_file。"
            )
            return self._inject_items(items, kwargs, text, kind="planner-notice")
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("向 Planner 公告文件状态失败: %s", e)
            return {"action": "continue"}

    @HookHandler(
        "maisaka.replyer.before_model_request",
        name="inject_file_context",
        mode=HookMode.BLOCKING,
        order=HookOrder.EARLY,
        error_policy=ErrorPolicy.SKIP,
        # v1.0.18：显式超时（查询 embedding 单次 8s + 检索/拼装余量），防止 BLOCKING hook 卡死请求
        timeout_ms=11000,
    )
    async def inject_file_context(self, **kwargs: Any) -> dict[str, Any]:
        """检索相关文件内容，注入 LLM 上下文。

        参考 runtime-gotchas 2：传的是 items 而非 messages；
        只回 modified_kwargs["items"]；带幂等标记避免重试叠加。
        """
        try:
            session_id = str(kwargs.get("session_id") or "")
            items = kwargs.get("items")
            if not session_id or not isinstance(items, list):
                return {"action": "continue"}

            # v1.3.0：解析失败通知优先处理——即使本会话已经没有文件库，也要让模型
            # 知道「用户传过文件但没读进来」，否则用户追问时模型只能凭空猜。
            # silent_errors=false 时失败已经发过群消息了，这里不再重复注入。
            if bool(self.config.reader.silent_errors):
                notice_text = self._take_notice(session_id)
                if notice_text:
                    return self._inject_items(items, kwargs, notice_text, kind="notice")

            # 找当前会话的向量库
            vs = None
            for (sid, _cid), candidate in self._store._sessions.items():
                if sid == session_id:
                    vs = candidate
                    break
            if vs is None or not vs.files:
                return {"action": "continue"}

            # 幂等（v1.3.0）：先看 item_id 前缀——与正文内容解耦，改文案不会破坏幂等
            # （旧版靠正文里的 marker 判重，marker 被配成空串就变成永不注入）。
            if self._already_injected(items):
                return {"action": "continue"}

            # 直注路径（v1.0.14）：全文合计 ≤ 阈值 → 不做检索、不调查询 embedding，
            # 直接把文件全文注入。注意 query 为空（如纯表情消息）时也落这里兜底。
            full_text = self._full_text_within_limit(vs)
            if full_text:
                # v1.3.0：事实式文案，不带分数/编号/「供参考」——模型会复述这些元信息
                inject_text = "以下是本会话已读过的文件内容：\n\n" + full_text
                self.ctx.logger.info(
                    "文件全文直注（%.0f 字 ≤ 阈值 %d），跳过检索",
                    sum(e.total_chars() for e in vs.files.values()),
                    int(self.config.reader.direct_inject_max_chars),
                )
                self._touch_injected(vs, list(vs.files.keys()), session_id, "direct-inject")
                return self._inject_items(items, kwargs, inject_text, kind="direct")

            # 用最近一条用户消息做查询（简化：取最后一条 user item 的文本）
            raw_query = self._extract_last_user_query(items)
            if not raw_query:
                return {"action": "continue"}
            query, _original = self._rewrite_query(items, raw_query)

            # 注入去重（v1.0.18 C3）：同一会话同一问题 TTL 内直接复用上次注入文本。
            # hook 每次 attempt 都会重跑、Planner 与回复各触发一次——旧幂等只挡 items 内的
            # marker，挡不住跨触发的重复 embedding 与重复检索。
            # memo key 用改写后的查询：同一句话在不同上下文里检索结果不该共用缓存
            memo_key = query.strip()
            memo_text = self._memo_get(session_id, memo_key)
            if memo_text:
                self.ctx.logger.info("文件注入命中去重缓存（同会话同问题 TTL 内），跳过检索")
                return self._inject_items(items, kwargs, memo_text, kind="context")

            # 大文件概要（v1.0.17）：检索前先确保大文件有概要（异步生成、失败静默），
            # 有概要时即使本次检索无命中也能给 LLM 一个整体认识
            await self._ensure_summary(vs)
            summaries = [
                (fname, entry.summary)
                for fname, entry in vs.files.items()
                if entry.summary
            ]

            hits, diag = await self._retrieve_hybrid(vs, query, session_id)
            diag["raw_query"] = _original
            self._record_diag(diag)

            if not hits:
                # v1.3.0：**无命中就不再注入**。旧版会用概要顶上，等于给模型一份
                # 它无法定位、无法引用的内容，反而诱导它编。只有两路检索都不可用
                # （embedding 失败且词法也不可用）才用概要兜底。
                if diag["mode"] == "unavailable" and summaries:
                    parts = ["以下是本会话已读过的文件概要："]
                    for fname, s in summaries:
                        parts.append(f"《{fname}》概要：\n{s}")
                    inject_text = "\n\n".join(parts)
                    self._memo_put(session_id, memo_key, inject_text)
                    self._touch_injected(vs, [f for f, _ in summaries], session_id, memo_key)
                    return self._inject_items(items, kwargs, inject_text, kind="summary")
                # 未命中分支的日志级别不能低于命中分支（否则"没检索到"永远查不出来）
                self.ctx.logger.warning(
                    "检索无命中，本次不注入（mode=%s，向量候选=%d，词法候选=%d，融合后=%d）",
                    diag["mode"],
                    diag["vector_candidates"],
                    diag["lexical_candidates"],
                    diag["after_gate"],
                )
                return {"action": "continue"}

            # 拼接注入文本（v1.3.0 去机器味）：不带分数、不带编号、不带「供参考」，
            # 同一文件的多块合并到同一标题下。
            parts = ["以下是本会话已读过的文件内容："]
            for fname, s in summaries:
                parts.append(f"《{fname}》概要：\n{s}")
            by_file: dict[str, list[str]] = {}
            for h in hits:
                by_file.setdefault(str(h.get("file_name")), []).append(str(h.get("text", "")))
            for fname, texts in by_file.items():
                parts.append(f"《{fname}》\n" + "\n".join(texts))
            inject_text = "\n\n".join(parts)
            self._memo_put(session_id, memo_key, inject_text)
            self._touch_injected(vs, list(by_file.keys()), session_id, memo_key)
            return self._inject_items(items, kwargs, inject_text, kind="context")
        except Exception as e:  # noqa: BLE001
            self.ctx.logger.error("注入文件上下文失败: %s", e, exc_info=True)
            return {"action": "continue"}

    @staticmethod
    def _item_text(item: Any) -> str:
        if not isinstance(item, dict):
            return ""
        parts = item.get("parts") or []
        texts = []
        for p in parts:
            if isinstance(p, dict) and p.get("type") == "text":
                texts.append(str(p.get("text", "")))
        return " ".join(texts)

    @staticmethod
    def _extract_last_user_query(items: list[Any]) -> str:
        for item in reversed(items):
            if not isinstance(item, dict):
                continue
            if item.get("item_type") in ("UserMessageItem", "user"):
                parts = item.get("parts") or []
                for p in parts:
                    if isinstance(p, dict) and p.get("type") == "text":
                        t = str(p.get("text", "")).strip()
                        if t:
                            return t
        return ""

    @staticmethod
    def _make_system_item(text: str, kind: str = "context") -> dict[str, Any]:
        """构造注入用的 SystemMessageItem。

        v1.3.0：`item_id` 统一带 `file-reader:{kind}:` 前缀。幂等判定改看这个前缀，
        与正文内容彻底解耦——旧版靠正文里的 marker 判重，marker 被配成空串就变成
        「永不注入」，改文案也会跟着失效。
        """
        import datetime
        import uuid

        return {
            "item_type": "SystemMessageItem",
            "meta": {
                "item_id": f"file-reader:{kind}:{uuid.uuid4().hex}",
                "logical_turn_id": None,
                "timestamp": datetime.datetime.now().isoformat(),
            },
            "parts": [{"type": "text", "text": text}],
        }

    # v1.3.0：search_file 工具返回内容时带的标记（代码层常量，不可配置）。
    # 工具结果不是 SystemMessageItem，拿不到我们的 item_id 前缀，只能靠它识别
    # 「模型已经拿到文件内容了」，否则 hook 会紧接着再注入一遍。
    _TOOL_TEXT_MARK = "【文件检索】"

    @classmethod
    def _already_injected(cls, items: list) -> bool:
        """幂等判定：item_id 前缀（本插件注过的）+ 正文标记（工具已给过内容）。"""
        for item in items or []:
            if not isinstance(item, dict):
                continue
            meta = item.get("meta")
            if isinstance(meta, dict) and str(meta.get("item_id") or "").startswith("file-reader:"):
                return True
            if cls._TOOL_TEXT_MARK in cls._item_text(item):
                return True
        return False

    @classmethod
    def _inject_items(cls, items: list, kwargs: dict, text: str, kind: str = "context") -> dict[str, Any]:
        """统一构造 modified_kwargs（完整回传，不能只回传改动键）。"""
        modified = dict(kwargs)
        modified["items"] = list(items) + [cls._make_system_item(text, kind)]
        if "item_schema_version" in kwargs:
            modified["item_schema_version"] = kwargs["item_schema_version"]
        return {"action": "continue", "modified_kwargs": modified}

    def _touch_injected(self, vs: Any, file_names: list, session_id: str, query_key: str) -> None:
        """只对**实际被注入**的文件滑动续期并计一轮。

        v1.3.0 修正两点：
          - 旧版 `increment_rounds()` 给会话内**所有**文件加轮，没被问到的文件
            照样过期；
          - 轮数去重 key 不含 session_id，两个会话问同一句话会互相抑制。
        """
        vs.touch_files(list(file_names))
        if not self._tick_rounds_once(vs, query_key, session_id=session_id, file_names=file_names):
            return

    # ─── 解析失败通知（v1.3.0） ──────────────────────────────────
    def _add_notice(self, session_id: str, name: str, reason: str) -> None:
        """登记一条解析失败通知：只告诉模型，不发群消息。

        为什么不是发群消息：语气归 persona 管，插件刷群等于替 bot 决定说什么。
        模型知道「用户传了《x.epub》但读不了」就够了，说不说由它判断。
        """
        if not session_id:
            return
        notices = self._file_notices.setdefault(session_id, [])
        # 同名文件重新失败 → 覆盖旧条目（保持 delivered=False，让模型重新知道）
        for notice in notices:
            if notice.get("name") == name:
                notice.update({"reason": reason, "ts": time.time(), "delivered": False})
                return
        notices.append({"name": name, "reason": reason, "ts": time.time(), "delivered": False})
        if len(notices) > 10:  # 单会话上限，防无界增长
            del notices[0]

    def _take_notice(self, session_id: str) -> str:
        """取出待投递的失败通知（投递过的不重复），返回注入文案；无则空串。"""
        notices = self._file_notices.get(session_id) or []
        if not notices:
            return ""
        ttl = max(1, int(self.config.reader.parse_notice_ttl or 600))
        now = time.time()
        lines: list[str] = []
        for notice in notices:
            if notice.get("delivered") or (now - float(notice.get("ts", 0)) > ttl):
                continue
            notice["delivered"] = True
            lines.append(f"[文件状态] 用户上传了《{notice.get('name')}》，读取失败：{notice.get('reason')}")
        return "\n".join(lines)

    def _clear_notices(self, session_id: str, name: str = "") -> None:
        """同名文件成功入库、或 /clear_file 时清掉对应通知。"""
        if not session_id:
            return
        if not name:
            self._file_notices.pop(session_id, None)
            return
        notices = self._file_notices.get(session_id) or []
        self._file_notices[session_id] = [n for n in notices if n.get("name") != name]

    # ─── 命令 ────────────────────────────────────────────────────
    @Command("clear_file", description="清除当前会话已入库的文件", pattern=r"^\s*[/／]\s*(?:clear_file|clean_file|清文件|清理文件)\s*$")
    async def cmd_clear_file(self, **kwargs: Any) -> tuple[bool, str, int]:
        """清除当前会话所有文件。"""
        session_id = ""
        for key in ("session_id", "user_id", "chat_id", "stream_id"):
            if kwargs.get(key):
                session_id = str(kwargs[key])
                break
        if not session_id:
            return True, "无法定位当前会话，清除失败。", 0

        removed = self._store.drop(session_id)
        self._memo_invalidate_session(session_id)  # v1.0.18 C3：清文件后失效注入缓存
        self._clear_notices(session_id)  # v1.3.0：连失败通知一起清掉
        text = f"🗑️ 已清除当前会话 {removed} 个文件块。" if removed else "当前会话没有已入库的文件。"
        await self._reply(str(kwargs.get("stream_id", "") or ""), text)
        return True, text, 2 if removed else 0

    @Command("file_status", description="查看文件读取插件状态", pattern=r"^\s*[/／]\s*(?:file_status|文件状态|文件读取状态)\s*$")
    async def cmd_status(self, **kwargs: Any) -> tuple[bool, str, int]:
        """状态命令：库规模 / 嵌入模型可用性 / 关键配置。"""
        total_files = sum(len(vs.files) for vs in self._store._sessions.values())
        total_chunks = sum(vs.total_chunks() for vs in self._store._sessions.values())
        sessions = len(self._store._sessions)
        embed_state = "正常" if self._embedding_ok else f"异常({self._embedding_err})"
        lines = [
            "📊 文件读取插件状态：",
            f"- 会话数: {sessions}",
            f"- 已入库文件: {total_files}",
            f"- 向量块总数: {total_chunks}",
            f"- 嵌入模型: {embed_state}",
            f"- 分块: {self.config.reader.chunk_size}/{self.config.reader.chunk_overlap}",
            f"- 检索 Top-K: {self.config.reader.retrieve_top_k}",
            f"- 全文直注阈值: {self.config.reader.direct_inject_max_chars} 字" + ("（关闭）" if int(self.config.reader.direct_inject_max_chars) <= 0 else ""),
            f"- 有效时间: {self.config.reader.file_retention_time}min / 最大 {self.config.reader.file_max_rounds} 轮",
            f"- 支持类型: {describe_supported_types()}",
            f"- NapCat 兜底: {'启用' if self.config.napcat.enabled else '停用'} ({self.config.napcat.http_url})",
        ]
        if self._embed_retry_queue:
            names = "、".join(f"{item['name']}(第{item['attempts']}次)" for item in self._embed_retry_queue[:5])
            lines.append(f"- ⏳ 待重试入库: {len(self._embed_retry_queue)} 个 — {names}")
        hint = getattr(self, "_last_file_hint", {}) or {}
        if hint:
            lines.append(
                f"- 最近文件消息: {hint.get('name')} 大小={hint.get('size')} mid={hint.get('mid')} 阶段={hint.get('stage')}"
            )
        # v1.3.0 检索诊断：没有 recall@k 评测集时，这是唯一能回答「为什么没检索到」的窗口。
        # 只输出模式/候选数/分数/耗时，不输出正文。
        override_note = "（运行期覆盖）" if self._hybrid_override is not None else ""
        lines.append(
            f"- 检索模式: {'向量 + 词法(BM25)' if self._hybrid_enabled() else '纯向量'}{override_note}"
        )
        total_tokens = 0
        disabled_reason = ""
        for candidate in self._store._sessions.values():
            st = candidate.lexical_state
            total_tokens += int(st.get("tokens") or 0)
            if st.get("disabled") and not disabled_reason:
                disabled_reason = str(st.get("reason") or "")
        if disabled_reason:
            lines.append(f"- 词法索引: {total_tokens} tokens（已停用：{disabled_reason}）")
        else:
            lines.append(f"- 词法索引: {total_tokens} tokens")
        if self._retrieval_diag:
            d = self._retrieval_diag[-1]
            lines.append(
                f"- 最近检索: 模式={d['mode']} 向量候选={d['vector_candidates']} "
                f"词法候选={d['lexical_candidates']} 融合={d['fused']}→过阈={d['after_gate']} "
                f"Top1={d['top1_score']} 耗时={d['elapsed_ms']}ms 来源={d['sources']}"
            )
            raw_q = str(d.get("raw_query") or "")
            if raw_q and raw_q != str(d.get("query") or ""):
                lines.append(f"  └ 原始提问: {raw_q[:40]} → 改写: {str(d.get('query'))[:60]}")
            elif str(d.get("query") or ""):
                lines.append(f"  └ 查询: {str(d.get('query'))[:60]}")
        text = "\n".join(lines)
        stream_id = str(kwargs.get("stream_id", "") or "")
        if stream_id:
            await self._reply(stream_id, text)
        return True, text, 2

    @Command(
        "file_bm25",
        description="运行期开关 BM25 词法检索（v1.3.0 回滚开关）：体感变差时可立即退回纯向量，无需改配置文件",
        pattern=r"^\s*[/／]\s*(?:file_bm25|词法检索|混合检索)\s*(?P<state>on|off|开|关|status|状态)?\s*$",
    )
    async def cmd_bm25(self, **kwargs: Any) -> tuple[bool, str, int]:
        """BM25 运行期开关。

        为什么命令与配置并存：已部署实例的 config.toml **不会自动补新键**，只改
        Field(default) 对老实例完全无效（宿主只在 config.toml 不存在时才初始化，
        config_version 相同时干脆不写回）。真机上发现混合检索变差时，靠配置关不掉，
        必须有一条立刻生效的退路。这里改的是内存里的覆盖值，重启后回到配置值。
        """
        raw = str(self._command_arg(kwargs, "state") or "").strip().lower()
        if raw in ("off", "关", "false", "0"):
            self._hybrid_override = False
            self._inject_memo.clear()  # 检索通路变了，缓存的注入文本必须作废
            text = "✅ 已关闭 BM25 词法检索，退回纯向量检索（运行期生效，重启后回到配置文件的值）"
        elif raw in ("on", "开", "true", "1"):
            self._hybrid_override = True
            self._inject_memo.clear()
            text = "✅ 已开启 BM25 词法检索（运行期生效，重启后回到配置文件的值）"
        else:
            cfg = bool(self.config.reader.hybrid_retrieve_enabled)
            effective = self._hybrid_enabled()
            text = (
                f"BM25 词法检索：配置={'开' if cfg else '关'}，当前生效={'开' if effective else '关'}"
                + ("（已被运行期命令覆盖）" if self._hybrid_override is not None else "")
                + "。用法：/file_bm25 on|off"
            )
        stream_id = str(kwargs.get("stream_id", "") or "")
        if stream_id:
            await self._reply(stream_id, text)
        return True, text, 2

    @staticmethod
    def _command_arg(kwargs: dict, name: str) -> str:
        """取命令参数：优先用正则命名组（宿主支持时直接传进来），
        拿不到再从常见消息字段里抠一遍——不同宿主版本传参形态并不统一，
        命令开关这种「必须能用」的入口不能只赌一种。"""
        val = kwargs.get(name)
        if isinstance(val, str) and val.strip():
            return val.strip()
        for key in ("raw_message", "text", "full_text", "content", "message"):
            raw = kwargs.get(key)
            if isinstance(raw, dict):
                for sub in ("text", "raw_message", "processed_plain_text"):
                    v = raw.get(sub)
                    if isinstance(v, str) and v.strip():
                        raw = v
                        break
            if isinstance(raw, str) and raw.strip():
                m = re.search(r"(?:file_bm25|词法检索|混合检索)\s*([A-Za-z\u4e00-\u9fff]+)?", raw)
                if m:
                    return str(m.group(1) or "").strip()
        return ""

    # ─── Tool：手动查询文件内容（供 LLM 主动调用） ────────────────
    def _has_any_file_session(self) -> bool:
        """是否存在任何存有文件的会话（仅用于区分报错文案，不暴露会话信息）。"""
        return any(candidate.files for candidate in self._store._sessions.values())

    def _resolve_tool_session(self, session_id: str) -> tuple[Optional[Any], str]:
        """工具调用的会话兜底解析。

        真机 18:51 日志教训：Planner 调 search_file 时首次传 session_id=''、
        之后干脆不传——文件 18:51:30 已入库，工具却三次误报「没有文件」。
        不能指望 LLM 每次都传对 session_id。规则（v1.2.1 收紧，v1.3.3 接住宿主注入）：
        1) 显式 session_id（或宿主注入的 stream_id/chat_id）且该会话有文件 → 直接用；
        2) 否则**仅在全局只有一个会话存有文件时**回退到该会话（单会话场景必命中，
           v1.0.15 的真机问题不回退）；
        3) 多个会话都有文件、而本次调用未指明或指错了 session_id → **拒绝兜底**。

        v1.3.3：入参可能来自 `search_file` 接住的 `kwargs["stream_id"]`——那是 Host
        `_build_tool_context_payload` 按当前执行上下文注入的权威值（= context.stream_id
        or context.session_id），与 `_store` 的 session_id 同口径，属**已知归属**而非猜测。
        v1.2.1 收紧时以为「工具调用链路拿不到当前会话」，那个前提在 Host 1.3.x 已不成立。

        v1.2.1 收紧原因（市场 review 隐私项）：旧版兜底是「最近有文件入库的会话」，
        多人群场景下 Planner 不传 session_id 时，会把 A 会话上传的文件正文塞进
        B 会话的 LLM 上下文——真机日志证明 Planner 确实经常不传该参数，不是纸上
        风险。工具调用链路拿不到当前消息所在的 stream/chat，无法做会话归属校验，
        所以歧义时宁可如实报告并提示补传 session_id（fail-closed），不猜。
        hook 侧 inject_file_context 做的是精确匹配注入，不受本问题影响。

        返回 (VectorStore 或 None, 实际采用的 session_id)；None = 没有可用的会话。

        v1.0.16 教训：本方法曾插在 @Tool 装饰器与 search_file 定义之间，
        导致 @Tool 把 _resolve_tool_session 注册成了工具本体（真机报
        unexpected keyword argument 'query'）。普通辅助方法必须放在
        @Tool 装饰器之前定义。
        """
        if session_id:
            for (sid, _cid), candidate in self._store._sessions.items():
                if sid == session_id and candidate.files:
                    return candidate, session_id
        # 兜底候选：按 sid 聚合「存有文件」的会话
        sessions_with_files: dict[str, list[Any]] = {}
        for (sid, _cid), candidate in self._store._sessions.items():
            if candidate.files:
                sessions_with_files.setdefault(sid, []).append(candidate)
        if not sessions_with_files:
            return None, ""
        if len(sessions_with_files) > 1:
            # 隐私护栏：没有可靠依据判断调用方在哪个会话，回退「最近入库」
            # 等于把别的会话的文件喂给当前调用方——拒绝并留痕（审计可查）。
            self.ctx.logger.warning(
                "search_file 会话兜底被拒：%d 个会话存有已入库文件且本次调用未指明有效 "
                "session_id，按会话隔离原则不做跨会话回退",
                len(sessions_with_files),
            )
            return None, ""
        the_sid, stores = next(iter(sessions_with_files.items()))
        best = max(stores, key=lambda s: max(e.upload_time for e in s.files.values()))
        if session_id and session_id != the_sid:
            self.ctx.logger.info(
                "search_file 传入 session_id=%s 无文件，回退到唯一存有文件的会话 %s",
                session_id,
                the_sid,
            )
        else:
            self.ctx.logger.info(
                "search_file 未收到（有效）session_id，回退到唯一存有文件的会话 %s", the_sid
            )
        return best, the_sid

    @Tool(
        "search_file",
        description=(
            "在当前会话已上传的文件中检索与问题相关的片段。"
            "当用户上传过文件（PDF/Word/Excel/PPT/代码等）或询问文件内容时调用。"
            "会话归属由宿主自动带入，正常情况下**不需要**传 session_id。"
        ),
        parameters=[
            ToolParameterInfo(
                name="query",
                param_type=ToolParamType.STRING,
                description="要在文件中检索的问题或关键词",
                required=True,
            ),
            ToolParameterInfo(
                name="session_id",
                param_type=ToolParamType.STRING,
                description="通常留空（宿主会自动带入当前会话）。仅在需要显式指定其他会话时填写",
                required=False,
            ),
        ],
    )
    async def search_file(self, query: str, session_id: str = "", **kwargs: Any) -> dict[str, str]:
        """检索并返回相关片段。"""
        # v1.3.3：接住 Host 注入的权威会话标识。
        # Host 1.3.x 的 `_build_tool_context_payload`（src/plugin_runtime/component_query.py）
        # 会往工具参数里补 `stream_id`/`chat_id`（= context.stream_id or context.session_id），
        # 且只在模型没传同名字段时才补。旧版开头 `del kwargs` 把它一起丢掉，
        # 于是工具永远拿不到会话 ⇒ 只能靠「全局唯一有文件的会话」兜底；
        # 一旦群聊与私聊都传过文件，search_file 直接 fail-closed 拒绝（真机 21:53 日志即此状态）。
        # 这不是猜：值由 Host 按当前执行上下文注入，口径与 `_store` 的 session_id 一致。
        runtime_sid = str(kwargs.get("stream_id") or kwargs.get("chat_id") or "").strip()
        del kwargs
        if not query.strip():
            return {"content": "请提供要检索的问题。"}
        explicit_sid = str(session_id or "").strip()
        vs, resolved_sid = self._resolve_tool_session(explicit_sid or runtime_sid)
        if vs is None or not vs.files:
            # v1.2.1：区分「确实没有文件」与「有文件但拒绝跨会话兜底」，
            # 给 LLM 可执行的指引（补传 session_id），而不是含糊的"没有文件"。
            if self._has_any_file_session():
                if explicit_sid or runtime_sid:
                    return {
                        "content": (
                            f"会话 {explicit_sid or runtime_sid} 下没有已入库的文件。"
                            "其他会话存有的文件出于会话隔离不会返回；"
                            "如需检索当前会话的文件，请确认 session_id 是否正确后重试。"
                        )
                    }
                return {
                    "content": (
                        "无法确定要检索哪个会话：本插件有多个会话存有已入库文件，"
                        "而本次调用未提供 session_id。为避免把其他会话的文件内容错误地"
                        "带入当前对话，本次不做自动定位；请带上当前会话的 session_id "
                        "重新调用 search_file。"
                    )
                }
            return {"content": "当前会话还没有已入库的文件，请先上传文件。"}

        # 直注路径（v1.0.14）：小文件直接返回全文，不检索
        full_text = self._full_text_within_limit(vs)
        if full_text:
            self._touch_injected(vs, list(vs.files.keys()), resolved_sid, query)
            names = "、".join(vs.files.keys())
            return {
                "content": (
                    # 标记用于让 hook 识别「工具已经把内容给过模型」，不再重复注入
                    f"{self._TOOL_TEXT_MARK} 当前会话文件较小（≤ {int(self.config.reader.direct_inject_max_chars)} 字），"
                    # 会话 ID 是给 Planner 的定位信息（多会话场景它要能说清查的是哪个会话），
                    # 「去机器味」只针对注入给用户对话的 hook 文案，工具返回保留
                    f"直接给出全文（会话 {resolved_sid}，文件：{names}）：\n\n{full_text}"
                )
            }

        # v1.3.0：工具侧与 hook 侧走同一套混合检索，避免"工具查得到、注入查不到"
        hits, diag = await self._retrieve_hybrid(vs, query, resolved_sid)
        self._record_diag(diag)
        if not hits:
            return {"content": "未在文件中检索到相关内容。"}
        parts = [
            f"{self._TOOL_TEXT_MARK} 检索到以下相关片段"
            f"（会话 {resolved_sid}，已入库文件：{'、'.join(vs.files.keys())}）："
        ]
        by_file: dict[str, list[str]] = {}
        for h in hits:
            by_file.setdefault(str(h.get("file_name")), []).append(str(h.get("text", "")))
        for fname, texts in by_file.items():
            parts.append(f"《{fname}》\n" + "\n".join(texts))
        self._touch_injected(vs, list(by_file.keys()), resolved_sid, query)
        return {"content": "\n\n".join(parts)}

    # ─── 后台清理循环 ────────────────────────────────────────────
    async def _start_cleanup_loop(self) -> None:
        async def loop() -> None:
            while True:
                try:
                    await asyncio.sleep(self.config.reader.cleanup_interval * 60)
                    retention = self.config.reader.file_retention_time * 60
                    max_rounds = self.config.reader.file_max_rounds
                    # v1.3.0：绝对上限（分钟→秒），防「一直被问到」的文件永久驻留
                    try:
                        absolute_ttl = float(self.config.reader.file_absolute_ttl or 0) * 60
                    except (TypeError, ValueError):
                        absolute_ttl = 0.0
                    total_removed = 0
                    for (sid, cid), vs in list(self._store._sessions.items()):
                        removed = vs.cleanup_expired(retention, max_rounds, absolute_ttl)
                        if removed:
                            self._memo_invalidate_session(sid)  # v1.0.18 C3：清理后失效该会话注入缓存
                        total_removed += len(removed)
                        # v1.1.0：文件清空后回收空会话壳——旧版 (sid,cid)→空 VectorStore 永久驻留，
                        # 长期运行的 bot 会话数无界增长
                        if not vs.files:
                            self._store.drop(sid, cid)
                    if total_removed:
                        self.ctx.logger.info("后台清理：移除 %d 个过期文件", total_removed)
                    # v1.1.0：概要失败冷却表随清理周期裁剪——旧版按文件名无界增长
                    self._prune_summary_failed()
                except asyncio.CancelledError:
                    raise
                except Exception as e:  # noqa: BLE001
                    self.ctx.logger.error("后台清理异常: %s", e)

        self._cleanup_task = asyncio.create_task(loop())

    def _prune_summary_failed(self) -> None:
        """裁剪概要失败冷却表（v1.1.0）：删 24 小时前的记录，且总量封顶 256 条（超了删最旧）。"""
        if not self._summary_failed_ts:
            return
        now = time.time()
        self._summary_failed_ts = {
            k: v for k, v in self._summary_failed_ts.items() if now - v < 86400.0
        }
        while len(self._summary_failed_ts) > 256:
            oldest = min(self._summary_failed_ts, key=lambda k: self._summary_failed_ts[k])
            self._summary_failed_ts.pop(oldest, None)

    # ─── embedding 失败后台重试队列（v1.0.11） ───────────────────
    async def _enqueue_embed_retry(
        self,
        session_id: str,
        conversation_id: str,
        stream_id: str,
        name: str,
        tmp_path: str,
    ) -> None:
        """把解析好但 embedding 失败的文件放进后台重试队列，并回复准确措辞。

        措辞设计：明确"文件已收到、内容已读出"，只是嵌入模型暂时无响应——
        避免 bot 之后从空检索结果瞎猜成「链接拉取超时」（真机 14:28 日志踩过）。

        v1.1.0：条目存临时文件路径而非解析全文——旧版全文（可达数 MB）会在队列里
        驻留最长约 1 小时（30 次 × 2 分钟），多文件并发失败时内存叠加；
        现由重试循环在到期时重新读取解析，成功/放弃/on_unload 时删除临时文件。
        """
        # 同名文件已在队列里（如 10s 去重窗口外的重复发送）：覆盖旧条目即可
        for item in self._embed_retry_queue:
            if item["session_id"] == session_id and item["name"] == name:
                old_tmp = item.get("tmp_path")
                if old_tmp and old_tmp != tmp_path:
                    try:
                        os.remove(str(old_tmp))
                    except OSError:
                        pass
        self._embed_retry_queue = [
            item for item in self._embed_retry_queue if not (item["session_id"] == session_id and item["name"] == name)
        ]
        max_retries = max(1, int(self.config.reader.embed_max_retries))
        self._embed_retry_queue.append(
            {
                "session_id": session_id,
                "conversation_id": conversation_id,
                "stream_id": stream_id,
                "name": name,
                "tmp_path": tmp_path,
                "attempts": 0,
                "next_ts": time.time() + 30.0,  # 首次重试等 30s（短期抖动大概率自愈）
            }
        )
        await self._reply_error(
            stream_id,
            f"⚠️ 文件「{name}」已收到、内容也读出来了，但嵌入模型暂时无响应。"
            f"我会每隔 {self.config.reader.embed_retry_interval:.0f} 分钟自动重试入库（最多 {max_retries} 次），"
            "成功后无需重新发送。",
            reason="embed_queued",
        )

    @staticmethod
    def _discard_retry_tmp(item: dict[str, Any]) -> None:
        """删除重试条目持有的临时文件（成功入库 / 放弃 / 文件丢失时调用）。"""
        tmp = item.get("tmp_path")
        if tmp:
            try:
                os.remove(str(tmp))
            except OSError:
                pass
            item["tmp_path"] = ""

    async def _start_retry_loop(self) -> None:
        """后台定时扫描重试队列：到期的条目重新尝试入库。"""

        async def loop() -> None:
            while True:
                try:
                    await asyncio.sleep(20.0)  # 扫描频率 20s；条目自身带 next_ts 控制实际节奏
                    if not self._embed_retry_queue:
                        continue
                    now = time.time()
                    due = [item for item in self._embed_retry_queue if now >= item["next_ts"]]
                    if not due:
                        continue
                    interval = max(0.5, float(self.config.reader.embed_retry_interval)) * 60.0
                    still_queued: list[dict[str, Any]] = [
                        item for item in self._embed_retry_queue if now < item["next_ts"]
                    ]
                    for item in due:
                        item["attempts"] += 1
                        # max_retries 重试时实时读配置（真机 15:20 日志教训：入队时写死会
                        # 在长拥塞里耗尽；热改配置立即生效，正在排队的文件也能受益）
                        max_retries = max(1, int(self.config.reader.embed_max_retries))
                        # v1.1.0：条目只存临时文件路径，重试前重新读取解析；
                        # 文件已丢失（如系统清了 temp 目录）直接放弃，不再无谓重试
                        tmp_path = str(item.get("tmp_path") or "")
                        if not tmp_path or not os.path.exists(tmp_path):
                            self.ctx.logger.error(
                                "文件 %s 的临时文件已丢失，放弃重试（请重新发送文件）", item["name"]
                            )
                            await self._reply_error(
                                item["stream_id"],
                                f"⚠️ 文件「{item['name']}」的缓存已被系统清理，无法自动重试，麻烦重新发送一次文件。",
                                reason="embed_retry_tmp_lost",
                            )
                            continue
                        try:
                            text = await asyncio.to_thread(read_any_file_to_text, tmp_path, item["name"])
                            vs = self._store.get_or_create(
                                item["session_id"],
                                item["conversation_id"],
                                self._embed,
                                self.config.reader.chunk_size,
                                self.config.reader.chunk_overlap,
                                self.config.reader.retrieve_top_k,
                                chunk_merge=bool(self.config.reader.chunk_merge_enabled),
                                retrieve_use_matrix=bool(self.config.reader.retrieve_use_matrix),
                            )
                            entry = await vs.add_file(item["name"], text)
                        except Exception as e:  # noqa: BLE001
                            if item["attempts"] >= max_retries:
                                self.ctx.logger.error(
                                    "文件 %s 重试 %d 次仍失败，放弃: %s", item["name"], item["attempts"], e
                                )
                                self._discard_retry_tmp(item)
                                await self._reply_error(
                                    item["stream_id"],
                                    f"⚠️ 文件「{item['name']}」自动重试 {item['attempts']} 次仍未入库"
                                    "（嵌入模型持续无响应），麻烦重新发送一次文件。",
                                    reason="embed_retries_exhausted",
                                )
                                continue
                            item["next_ts"] = time.time() + interval
                            still_queued.append(item)
                            self.ctx.logger.warning(
                                "文件 %s 重试第 %d/%d 次失败，%.0f 分钟后再试: %s",
                                item["name"],
                                item["attempts"],
                                max_retries,
                                interval / 60.0,
                                e,
                            )
                            continue
                        # 成功入库：释放临时文件
                        self._discard_retry_tmp(item)
                        # 成功入库
                        self.ctx.logger.info(
                            "重试入库成功 %s：%d 块（第 %d 次尝试，session=%s）",
                            item["name"],
                            len(entry.chunks),
                            item["attempts"],
                            item["session_id"],
                        )
                        if not self.config.reader.silent_success:
                            await self._reply(
                                item["stream_id"],
                                f"📄 文件「{item['name']}」已重试入库成功，切成 {len(entry.chunks)} 块。现在可以直接问我内容了。",
                            )
                    self._embed_retry_queue = still_queued
                except asyncio.CancelledError:
                    raise
                except Exception as e:  # noqa: BLE001
                    self.ctx.logger.error("embedding 重试循环异常: %s", e)

        self._retry_task = asyncio.create_task(loop())


# ================================================================ WebUI 显示层补丁
#
# 可视化模式的 FieldRenderer（dashboard/src/routes/plugin-config.tsx:163-361，
# 1.3.1 布局）按 ui_type 渲染控件时只输出 label / hint / placeholder，
# **从不渲染 description**；本插件字段的填法说明都写在 description 里，
# 不搬进 hint，用户在配置页上一个字都看不到（源代码模式才见得到）。
# 这里只改展示元数据，不碰任何配置键与校验语义。

#: 默认收起的 section：标题自带「可选 / 默认关闭」的功能节。收起后标题
#: 与说明仍可见，点开即可配置，避免整页卡片全开淹没常用配置。
_WEBUI_COLLAPSED_SECTIONS: frozenset = frozenset({"napcat"})

#: 手工指定的 section 标题（键 = section 名）；仅在 SDK 输出标题等于
#: 节名（未配置 __ui_label__）时采用。
_WEBUI_SECTION_TITLES: dict = {}

#: 手工指定的字段 label（键 = 字段名）；仅在自动推导不可用时采用。
_WEBUI_LABEL_OVERRIDES: dict = {
    "http_url": "HTTP 服务地址",
    "access_token": "HTTP Token",
    "cache_dir": "文件缓存目录",
}


#: 推导 label 用的中文分隔符（取最靠左的一个）
_WEBUI_CJK_SEPS = "。！？；，：（(、"


def _webui_label_from_description(description: str) -> str:
    """从中文 description 里取第一小句当显示标题（取不到返回空串）。

    在中英文标点里找**最靠左**的分隔符，取它前面的短语——通常是字段
    本身的名字；超长时截到 12 字符并避免把英文单词截一半。
    """
    text = (description or "").strip()
    if not text:
        return ""
    cut = len(text)
    for sep in _WEBUI_CJK_SEPS:
        idx = text.find(sep)
        if 0 < idx < cut:
            cut = idx
    text = text[:cut].strip()
    if len(text) > 12:
        text = text[:12]
        if " " in text[4:]:
            text = text[: text.rfind(" ")].rstrip() or text
    while text and text[-1] in "（(\"“：:，,；;、":
        text = text[:-1].rstrip()
    return text if len(text) >= 2 else ""


def _apply_webui_display_polish(schema: dict) -> dict:
    """把 description 抄进 hint、补中文 label / 节标题、收起可选功能节。"""
    if not isinstance(schema, dict):
        return schema
    sections = schema.get("sections")
    if not isinstance(sections, dict):
        return schema
    for name, section in sections.items():
        if not isinstance(section, dict):
            continue
        if name in _WEBUI_COLLAPSED_SECTIONS:
            section["collapsed"] = True
        title = section.get("title")
        if not title or title == name:
            new_title = _WEBUI_SECTION_TITLES.get(name) or _webui_label_from_description(
                section.get("description") or ""
            )
            if new_title and new_title != name:
                section["title"] = new_title
        for fname, field in (section.get("fields") or {}).items():
            if not isinstance(field, dict):
                continue
            if fname == "config_version":
                field["hidden"] = True  # 插件自维护字段：可视化模式不渲染，源代码模式仍可见
            if not field.get("hint") and field.get("description"):
                field["hint"] = field["description"]
            label = field.get("label")
            if (not label or label == fname) and field.get("description"):
                new_label = _WEBUI_LABEL_OVERRIDES.get(fname) or _webui_label_from_description(
                    field["description"]
                )
                if new_label and new_label != fname:
                    field["label"] = new_label
    return schema


def create_plugin() -> FileReaderPlugin:
    """创建插件实例。"""
    return FileReaderPlugin()