fix: independent persist executor + reset pending_push on failure

- backend/app/services/persist.py: add dedicated ThreadPoolExecutor
(so upload_folder can't starve business run_in_executor like
ChromaDB query / BGE-M3 encode)
- push_to_hf(): reset pending_push flag on failure too
(was getting stuck at True after first error, blocking all future
schedule_push calls)

app/services/llm_cache.py

@@ -0,0 +1,94 @@
+"""LLM 响应缓存 (内存 LRU).
+
+为什么需要:
+- 个人使用场景下, 同一问题反复问很常见
+- 命中时直接跳过 LLM 调用 + 流式回放 token
+- 节省 API 费用 + 缩短延迟
+
+设计:
+- key = sha256( (query + top_doc_ids + temperature) )  -- 只缓存"标准问答", 不缓存工具调用
+- value = 完整回答内容 + 引用 + 工具调用结果
+- LRU 容量可配 (默认 200)
+- 进程内, 重启清空 (避免引入 Redis)
+"""
+from __future__ import annotations
+
+import hashlib
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from cachetools import LRUCache
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CachedAnswer:
+    content: str
+    citations: list[dict[str, Any]]
+    tool_calls: list[dict[str, Any]]
+    tokens: list[str]  # 预切好的 token 序列, 流式回放
+
+
+_cache: LRUCache | None = None
+_hits = 0
+_misses = 0
+
+
+def _make_key(query: str, top_doc_ids: list[str], temperature: float) -> str:
+    """缓存 key. 包含 query + 命中文档 id + 温度, 避免不同上下文错命中."""
+    payload = f"{query.strip()}|{','.join(sorted(top_doc_ids))}|{temperature:.2f}"
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def get_cache() -> LRUCache:
+    global _cache
+    if _cache is None:
+        size = settings.llm_cache_size if settings.llm_cache_enabled else 0
+        _cache = LRUCache(maxsize=size)
+        logger.info("LLM cache initialized: enabled=%s size=%d", settings.llm_cache_enabled, size)
+    return _cache
+
+
+def lookup(query: str, top_doc_ids: list[str], temperature: float) -> CachedAnswer | None:
+    if not settings.llm_cache_enabled:
+        return None
+    key = _make_key(query, top_doc_ids, temperature)
+    hit = get_cache().get(key)
+    global _hits, _misses
+    if hit is not None:
+        _hits += 1
+        logger.debug("LLM cache HIT key=%s", key[:12])
+    else:
+        _misses += 1
+    return hit
+
+
+def store(query: str, top_doc_ids: list[str], temperature: float, answer: CachedAnswer) -> None:
+    if not settings.llm_cache_enabled:
+        return
+    key = _make_key(query, top_doc_ids, temperature)
+    get_cache()[key] = answer
+    logger.debug("LLM cache STORE key=%s tokens=%d", key[:12], len(answer.tokens))
+
+
+def stats() -> dict[str, Any]:
+    return {
+        "enabled": settings.llm_cache_enabled,
+        "size": len(get_cache()),
+        "max_size": get_cache().maxsize,
+        "hits": _hits,
+        "misses": _misses,
+        "hit_rate": (_hits / max(_hits + _misses, 1)),
+    }
+
+
+def clear() -> None:
+    global _cache, _hits, _misses
+    _cache = None
+    _hits = 0
+    _misses = 0
+    logger.info("LLM cache cleared")

app/services/parsers/__init__.py

@@ -0,0 +1,123 @@
+"""Parser 工厂: 智能路由 + 降级链.
+
+智能路由逻辑:
+1. 查 settings.parser_primary (默认 docling)
+2. 主 parser 失败 -> 降级到 settings.parser_fallback
+3. 全部失败 -> 抛 IngestionFailedError
+"""
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from app.config import settings
+from app.core.errors import IngestionFailedError
+from app.services.parsers.base_parser import BaseParser, ParsedDocument
+
+logger = logging.getLogger(__name__)
+
+
+# 延迟注册: 实际 import 在 _build_parser 内
+_REGISTRY: dict[str, type[BaseParser]] = {}
+
+
+def _register_default() -> None:
+    """懒加载各 parser. 失败的 (未装) 仅记 warning, 不抛."""
+    if _REGISTRY:
+        return
+    try:
+        from app.services.parsers.docling_parser import DoclingParser
+        _REGISTRY["docling"] = DoclingParser
+    except ImportError as e:
+        logger.warning("docling not installed: %s", e)
+    try:
+        from app.services.parsers.simple_parser import SimpleParser
+        _REGISTRY["simple"] = SimpleParser
+    except ImportError as e:
+        logger.warning("simple parser not installed: %s", e)
+    # mineru / vlm 留 hook (按需装)
+
+
+def _build_parser(name: str) -> BaseParser:
+    _register_default()
+    cls = _REGISTRY.get(name)
+    if cls is None:
+        raise IngestionFailedError(
+            f"Parser '{name}' is not installed. pip install docling / marker-pdf.",
+            code="parser_unavailable",
+        )
+    return cls()
+
+
+def get_parser(name: str | None = None) -> BaseParser:
+    """获取单个 parser 实例 (按名字)."""
+    return _build_parser(name or settings.parser_primary)
+
+
+def get_parser_chain() -> list[BaseParser]:
+    """按 settings 配置返回 [primary, fallback] 链."""
+    chain: list[BaseParser] = []
+    for name in (settings.parser_primary, settings.parser_fallback):
+        if name and name not in {p.name for p in chain}:
+            try:
+                chain.append(_build_parser(name))
+            except IngestionFailedError:
+                # 跳过未装的, 继续
+                continue
+    return chain
+
+
+async def parse_with_fallback(file_path: Path) -> ParsedDocument:
+    """按链逐个尝试, 全部失败抛 IngestionFailedError."""
+    chain = get_parser_chain()
+    if not chain:
+        raise IngestionFailedError(
+            "No parser available. Install at least one of: docling, marker-pdf.",
+            code="no_parser_available",
+        )
+
+    # 选能处理该扩展名的 parser
+    candidates = [p for p in chain if p.can_handle(file_path)]
+    if not candidates:
+        raise IngestionFailedError(
+            f"No parser in chain supports {file_path.suffix}",
+            code="unsupported_format",
+            detail={"suffix": file_path.suffix, "chain": [p.name for p in chain]},
+        )
+
+    last_err: Exception | None = None
+    last_traceback: str | None = None
+    for parser in candidates:
+        try:
+            return await parser.parse(file_path)
+        except Exception as e:  # noqa: BLE001
+            import traceback
+            tb = traceback.format_exc()
+            logger.warning("Parser %s failed for %s: %s\n%s", parser.name, file_path.name, e, tb)
+            last_err = e
+            last_traceback = tb
+
+    # 把最后一个 parser 的具体异常信息暴露给前端, 方便诊断
+    err_msg = f"All parsers failed for {file_path.name}"
+    if last_err:
+        err_msg += f" (last: {type(last_err).__name__}: {last_err})"
+    raise IngestionFailedError(
+        err_msg,
+        code="all_parsers_failed",
+        detail={
+            "chain": [p.name for p in candidates],
+            "last_error_type": type(last_err).__name__ if last_err else None,
+            "last_error_msg": str(last_err)[:500] if last_err else None,
+            "last_traceback": (last_traceback or "")[-1500:],  # 末 1.5KB, 防爆
+        },
+    ) from last_err
+
+
+__all__ = [
+    "BaseParser",
+    "ParsedDocument",
+    "PageContent",
+    "get_parser",
+    "get_parser_chain",
+    "parse_with_fallback",
+]

app/services/parsers/base_parser.py

@@ -0,0 +1,56 @@
+"""文档解析抽象 + 数据结构.
+
+支持的输入: PDF, DOCX, PNG/JPG (含扫描件).
+
+设计:
+- BaseParser 抽象 parse() -> ParsedDocument
+- ParsedDocument 同时携带 markdown 全文 + 页面级结构 (含表格 / 图片)
+- 各具体 parser (Docling / Marker / MinerU) 实现同一接口, 可热替换
+"""
+from __future__ import annotations
+
+import abc
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass
+class PageContent:
+    """单页内容. 至少要有 text, 可选带 tables / images."""
+
+    page_no: int
+    text: str
+    tables: list[dict] = field(default_factory=list)   # {html, bbox, rows}
+    images: list[dict] = field(default_factory=list)   # {bbox, caption, b64_thumb}
+    headings: list[str] = field(default_factory=list)  # 当前页的标题
+
+
+@dataclass
+class ParsedDocument:
+    """解析后的统一文档结构.
+
+    业务侧只用 markdown (全文) + pages (带页码引用) 两个核心字段.
+    """
+
+    markdown: str                                    # 全文 markdown (LLM 友好)
+    pages: list[PageContent] = field(default_factory=list)
+    tables: list[dict] = field(default_factory=list)  # 全部表格 (跨页表已合并)
+    images: list[dict] = field(default_factory=list)
+    meta: dict = field(default_factory=dict)          # page_count, parser, elapsed_ms, ...
+
+
+class BaseParser(abc.ABC):
+    """文档解析器抽象基类."""
+
+    name: str = "abstract"
+
+    @abc.abstractmethod
+    def supported_extensions(self) -> set[str]:
+        """形如 {'.pdf', '.docx'}."""
+
+    @abc.abstractmethod
+    async def parse(self, file_path: Path) -> ParsedDocument:
+        """同步 IO + 异步 wrapper. 重 CPU 解析可放线程池."""
+
+    def can_handle(self, file_path: Path) -> bool:
+        return file_path.suffix.lower() in self.supported_extensions()

app/services/parsers/docling_parser.py

@@ -0,0 +1,179 @@
+"""Docling parser - 主力.
+
+特性:
+- 结构感知 (reading order / headings)
+- 跨页表合并 (TableFormer)
+- 内置 OCR (PaddleOCR 支持中英)
+- DOCX / PPTX / 图片 / HTML 全支持
+
+Docling API 在 2.x 期间变动较多, 本实现基于 2.30+ 兼容.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from pathlib import Path
+
+from app.config import settings
+from app.services.parsers.base_parser import BaseParser, PageContent, ParsedDocument
+
+logger = logging.getLogger(__name__)
+
+
+class DoclingParser(BaseParser):
+    name = "docling"
+
+    def supported_extensions(self) -> set[str]:
+        return {".pdf", ".docx", ".pptx", ".png", ".jpg", ".jpeg", ".tiff", ".html", ".xlsx"}
+
+    async def parse(self, file_path: Path) -> ParsedDocument:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._parse_sync, file_path)
+
+    def _parse_sync(self, file_path: Path) -> ParsedDocument:
+        """实际解析. CPU 密集, 在线程池跑."""
+        # Force CPU device for Docling to prevent MPS NotImplementedError on Intel Mac
+        try:
+            import docling.utils.accelerator_utils
+            docling.utils.accelerator_utils.decide_device = lambda *args, **kwargs: "cpu"
+        except ImportError:
+            pass
+
+        # 延迟 import, 避免启动期未装 docling 时崩溃
+        from docling.document_converter import DocumentConverter, PdfFormatOption
+        from docling.datamodel.base_models import InputFormat
+        from docling.datamodel.pipeline_options import PdfPipelineOptions
+
+        started = time.time()
+        logger.info("Docling parsing: %s", file_path.name)
+
+        from docling.datamodel.accelerator_options import AcceleratorOptions, AcceleratorDevice
+        opts = PdfPipelineOptions()
+        opts.do_ocr = settings.parser_enable_ocr
+        opts.do_table_structure = settings.parser_table_structure
+        opts.images_scale = 2.0
+        # ⚠️ 不要设 artifacts_path: 设了但目录为空会被 Docling 拒绝 (报 "is not valid")
+        # 不设时, Docling 通过 huggingface_hub 走 HF_HOME 自动下载 + 缓存
+        # 我们的 Dockerfile 设了 HF_HOME=/data/.cache/huggingface (持久卷), 所以重启后还在
+        opts.accelerator_options = AcceleratorOptions(device=AcceleratorDevice.CPU)
+
+        converter = DocumentConverter(
+            format_options={
+                InputFormat.PDF: PdfFormatOption(pipeline_options=opts),
+            }
+        )
+
+        try:
+            result = converter.convert(str(file_path))
+        except Exception as e:  # noqa: BLE001
+            logger.exception("Docling parse failed: %s", e)
+            raise
+
+        doc = result.document
+
+        # 全文 markdown
+        markdown = doc.export_to_markdown()
+
+        # 页面级 (Docling 用 iterate_items / pages 属性, 视版本略有差异)
+        pages: list[PageContent] = []
+        try:
+            page_count = len(doc.pages) if hasattr(doc, "pages") else 0
+            for idx in range(page_count):
+                page = doc.pages[idx]
+                # 提取该页文本 (Docling 2.x 没有现成 API, 用 page-level export 近似)
+                page_md = ""
+                if hasattr(page, "export_to_markdown"):
+                    try:
+                        page_md = page.export_to_markdown()
+                    except Exception:  # noqa: BLE001
+                        page_md = ""
+                pages.append(PageContent(
+                    page_no=idx + 1,
+                    text=page_md,
+                    headings=[],  # Docling 不直接给页级 headings, 留空
+                ))
+        except Exception as e:  # noqa: BLE001
+            logger.warning("Docling page extraction partial: %s", e)
+
+        # 表格 (简化提取)
+        tables: list[dict] = []
+        try:
+            for t in (doc.tables or []):
+                tables.append({
+                    "html": t.export_to_html() if hasattr(t, "export_to_html") else "",
+                    "caption": getattr(t, "caption", None),
+                })
+        except Exception:  # noqa: BLE001
+            pass
+
+        elapsed_ms = int((time.time() - started) * 1000)
+        logger.info("Docling done: %s pages, %s tables, %dms", len(pages), len(tables), elapsed_ms)
+
+        return ParsedDocument(
+            markdown=markdown,
+            pages=pages,
+            tables=tables,
+            images=[],
+            meta={
+                "parser": self.name,
+                "page_count": len(pages),
+                "elapsed_ms": elapsed_ms,
+            },
+        )
+
+
+def _prewarm_docling_models() -> None:
+    """预下载 Docling 需要的模型 (layout/heron, tableformer, paddleocr 等, 共 ~2GB).
+
+    在 Space 启动 lifespan 阶段跑一次, 避免首次上传时下载超时或下载失败.
+    模型会缓存到 settings.hf_cache_dir, 后续启动跳过.
+    """
+    import logging
+    from pathlib import Path
+    from app.config import settings
+
+    logger_local = logging.getLogger(__name__)
+    logger_local.info("Docling model prewarm: pulling layout/table/ocr models...")
+
+    from huggingface_hub import snapshot_download
+
+    # Docling 模型都在 ds4sd 命名空间下
+    repos = [
+        "ds4sd/docling-models",  # 主模型集 (layout, tableformer)
+    ]
+
+    cache_dir = Path(settings.hf_cache_dir) if hasattr(settings, "hf_cache_dir") else None
+    if cache_dir is None:
+        from app.core.paths import data_dir
+        cache_dir = data_dir() / ".cache" / "huggingface"
+
+    for repo in repos:
+        try:
+            p = snapshot_download(
+                repo_id=repo,
+                cache_dir=str(cache_dir),
+                # 避免下载所有 variants, 只下必需的
+                allow_patterns=[
+                    "*.json",
+                    "*.txt",
+                    "*.safetensors",
+                    "tokenizer*",
+                ],
+            )
+            logger_local.info("Docling model %s cached at %s", repo, p)
+        except Exception as e:  # noqa: BLE001
+            logger_local.warning("Docling model prewarm %s failed: %s", repo, e)
+
+    # PaddleOCR 模型 (Docling 内置 OCR 用). 单独下载.
+    try:
+        from paddleocr import PaddleOCR  # type: ignore
+        # 实例化会触发模型下载到 ~/.paddleocr
+        PaddleOCR(use_angle_cls=False, lang="ch", show_log=False)
+        logger_local.info("PaddleOCR (ch) model cached")
+    except Exception as e:  # noqa: BLE001
+        # PaddleOCR 可能没装 (e.g. arm64 平台), 不阻塞
+        logger_local.warning("PaddleOCR prewarm skipped: %s", e)
+
+    logger_local.info("Docling model prewarm done")
+

app/services/parsers/marker_parser.py

@@ -0,0 +1,62 @@
+"""Marker parser - 兜底 (Docling 失败时启用).
+
+特性:
+- PDF -> Markdown 转换, 速度快
+- 内置 surya-ocr, 支持 90+ 语言
+- 不擅长复杂表格 / 跨页表 (不如 Docling), 但胜在稳定
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from pathlib import Path
+
+from app.services.parsers.base_parser import BaseParser, ParsedDocument
+
+logger = logging.getLogger(__name__)
+
+
+class MarkerParser(BaseParser):
+    name = "marker"
+
+    def supported_extensions(self) -> set[str]:
+        # Marker 强项是 PDF; 其它类型建议直接 Docling
+        return {".pdf"}
+
+    async def parse(self, file_path: Path) -> ParsedDocument:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._parse_sync, file_path)
+
+    def _parse_sync(self, file_path: Path) -> ParsedDocument:
+        # Marker 1.x 推荐用 marker's Python API 而非 CLI subprocess
+        from marker.converters.pdf import PdfConverter
+        from marker.models import create_model_dict
+        from marker.output import text_from_rendered
+
+        started = time.time()
+        logger.info("Marker parsing: %s", file_path.name)
+
+        converter = PdfConverter(artifact_dict=create_model_dict())
+        rendered = converter(str(file_path))
+        markdown, _, _ = text_from_rendered(rendered)
+
+        # Marker 不直接给 page-level 拆分, 走全文 markdown
+        elapsed_ms = int((time.time() - started) * 1000)
+        logger.info("Marker done: %dms", elapsed_ms)
+
+        # 尝试从 rendered 拿 metadata
+        meta = rendered.metadata if hasattr(rendered, "metadata") else {}
+        page_count = meta.get("page_stats", {}).get("total_pages") if isinstance(meta, dict) else None
+
+        return ParsedDocument(
+            markdown=markdown,
+            pages=[],  # Marker 不分页
+            tables=[],
+            images=[],
+            meta={
+                "parser": self.name,
+                "page_count": page_count,
+                "elapsed_ms": elapsed_ms,
+            },
+        )

app/services/parsers/simple_parser.py

@@ -0,0 +1,88 @@
+"""SimpleParser - 零依赖的 PDF 文本提取 fallback.
+
+适用场景:
+- Docling 失败 / 模型下载卡住 / Docling 不可用
+- 文本型 PDF (非扫描件) - 用 pypdf 直接抽文字
+
+特性:
+- 零 ML 依赖 (只有 pypdf)
+- 快速, 内存友好
+- 拿不到结构 (无表格识别), 只做 "能分页 + 拿文本"
+
+这是 marker-pdf 的替代方案, 因为 marker-pdf 1.10+ 与 pydantic-ai 的 anthropic 版本冲突.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from pathlib import Path
+
+from app.services.parsers.base_parser import BaseParser, PageContent, ParsedDocument
+
+logger = logging.getLogger(__name__)
+
+
+class SimpleParser(BaseParser):
+    """pypdf-based 轻量级 PDF 解析器. 兜底中的兜底."""
+
+    name = "simple"
+
+    def supported_extensions(self) -> set[str]:
+        # 只支持 PDF; 其它格式 (docx/png) 还是走 Docling
+        return {".pdf"}
+
+    async def parse(self, file_path: Path) -> ParsedDocument:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._parse_sync, file_path)
+
+    def _parse_sync(self, file_path: Path) -> ParsedDocument:
+        started = time.time()
+        logger.info("SimpleParser (pypdf) parsing: %s", file_path.name)
+
+        try:
+            from pypdf import PdfReader
+        except ImportError as e:
+            raise RuntimeError(
+                "pypdf not installed. Add 'pypdf>=4.0' to requirements.txt"
+            ) from e
+
+        reader = PdfReader(str(file_path))
+        pages: list[PageContent] = []
+        page_texts: list[str] = []
+
+        for idx, page in enumerate(reader.pages):
+            try:
+                text = page.extract_text() or ""
+            except Exception as e:  # noqa: BLE001
+                logger.warning("pypdf extract_text failed on page %d: %s", idx, e)
+                text = ""
+            pages.append(PageContent(
+                page_no=idx + 1,
+                text=text,
+                tables=[],  # simple parser 不识别表格
+                images=[],
+            ))
+            page_texts.append(text)
+
+        # 全文 markdown (无结构, 直接拼)
+        markdown = "\n\n".join(page_texts)
+
+        elapsed_ms = int((time.time() - started) * 1000)
+        logger.info(
+            "SimpleParser done: %s pages, %dms (using pypdf)",
+            len(pages), elapsed_ms,
+        )
+
+        return ParsedDocument(
+            markdown=markdown,
+            pages=pages,
+            tables=[],
+            images=[],
+            meta={
+                "parser": self.name,
+                "page_count": len(pages),
+                "elapsed_ms": elapsed_ms,
+                "backend": "pypdf",
+            },
+        )

app/services/persist.py

@@ -0,0 +1,204 @@
+"""HF Dataset repo 持久化同步.
+
+为什么需要: HF Spaces 免费版磁盘是临时的 (容器重启后 /data 内的非持久卷数据会丢).
+唯一免费的持久化方案是把 /data 同步到 HF Dataset repo (Git LFS).
+
+调用模式:
+- lifespan startup: restore_from_hf()
+- 每次写操作后: schedule_push() (异步, 不阻塞用户)
+- lifespan shutdown: push_to_hf() (同步, 尽量)
+
+健壮性:
+- 首次部署 (repo 不存在) → 捕获 RepositoryNotFoundError → 标记 "fresh_start"
+- HF_TOKEN 缺失 → 跳过持久化, 降级为纯本地
+- 网络错误 → 重试 3 次后放弃, 不阻塞业务
+- upload_folder 跑在**独立 executor** 上, 不会占业务池 (否则 HF Space → HF Dataset
+  一旦卡/慢, ChromaDB query / BGE-M3 encode 等业务的 run_in_executor 全排不上, chat 直接挂)
+- push 失败时**也重置** pending_push flag, 不然 schedule_push 永远 short-circuit
+"""
+from __future__ import annotations
+
+import asyncio
+import concurrent.futures
+import logging
+import shutil
+from pathlib import Path
+from typing import Literal
+
+from huggingface_hub import (
+    create_repo,
+    snapshot_download,
+    upload_folder,
+)
+from huggingface_hub.errors import RepositoryNotFoundError
+
+from app.config import settings
+from app.core.paths import data_dir, sqlite_dir, chroma_dir, upload_dir
+
+logger = logging.getLogger(__name__)
+
+
+# ✅ 独立 ThreadPoolExecutor, 不跟业务 (Chroma / BGE-M3 / run_in_executor) 抢线程
+# 2 个 worker 够用: push 是单飞, 第二个留给 restore (启动期偶发重入)
+_persist_executor = concurrent.futures.ThreadPoolExecutor(
+    max_workers=2,
+    thread_name_prefix="persist",
+)
+
+# 状态机: 持久化是否启用 / 启动模式
+_state: dict[str, str | bool] = {
+    "mode": "disabled",  # disabled | cold_restore | fresh_start
+    "last_push_at": 0.0,
+    "pending_push": False,
+}
+
+
+def persist_mode() -> Literal["disabled", "cold_restore", "fresh_start"]:
+    return _state["mode"]  # type: ignore[return-value]
+
+
+async def restore_from_hf() -> None:
+    """从 HF Dataset repo 拉取数据到本地.
+
+    调用时机: FastAPI lifespan 启动.
+    """
+    if not settings.is_persist_enabled():
+        logger.info("Persistence disabled (HF_PERSIST_REPO or HF_TOKEN not set)")
+        _state["mode"] = "disabled"
+        return
+
+    repo_id = settings.hf_persist_repo
+    token = settings.hf_token.get_secret_value()
+    local_root = data_dir()
+    target_subdirs = ["chroma", "sqlite", "uploads"]
+
+    loop = asyncio.get_running_loop()
+    try:
+        await loop.run_in_executor(
+            _persist_executor,  # ✅ 独立池
+            lambda: snapshot_download(
+                repo_id=repo_id,
+                repo_type="dataset",
+                local_dir=str(local_root),
+                token=token,
+                allow_patterns=[f"{d}/*" for d in target_subdirs] + target_subdirs,
+            ),
+        )
+        _state["mode"] = "cold_restore"
+        logger.info("Persisted data restored from %s", repo_id)
+    except RepositoryNotFoundError:
+        # 首次部署: repo 还没创建, 属正常情况
+        _state["mode"] = "fresh_start"
+        logger.info(
+            "Persist repo %s not found (first deploy?). "
+            "Will create on first push.",
+            repo_id,
+        )
+    except Exception as e:  # noqa: BLE001
+        logger.error("Persist restore failed (will start fresh): %s", e)
+        _state["mode"] = "fresh_start"
+        # 不阻塞启动, 提示用户在 /readyz 看到降级状态
+
+
+async def push_to_hf() -> None:
+    """同步推送本地数据到 HF Dataset repo. 阻塞."""
+    if not settings.is_persist_enabled():
+        return
+    if _state["mode"] == "fresh_start":
+        # 首次需要先 create_repo
+        await _ensure_repo_exists()
+
+    repo_id = settings.hf_persist_repo
+    token = settings.hf_token.get_secret_value()
+    local_root = data_dir()
+
+    loop = asyncio.get_running_loop()
+    try:
+        await loop.run_in_executor(
+            _persist_executor,  # ✅ 独立池, 不阻塞业务 (Chroma / BGE-M3) 的 run_in_executor
+            lambda: upload_folder(
+                folder_path=str(local_root),
+                repo_id=repo_id,
+                repo_type="dataset",
+                token=token,
+                commit_message=f"sync {asyncio.get_running_loop().time():.0f}",
+                ignore_patterns=[".cache/*", "*.tmp", "*.lock"],
+            ),
+        )
+        _state["last_push_at"] = asyncio.get_running_loop().time()
+        _state["pending_push"] = False
+        logger.info("Persisted data pushed to %s", repo_id)
+    except Exception as e:  # noqa: BLE001
+        # ✅ 失败也重置 flag, 否则 schedule_push 永远 short-circuit, 数据再也不推
+        _state["pending_push"] = False
+        logger.error("Persist push failed: %s", e)
+
+
+async def schedule_push() -> None:
+    """异步推送, 不阻塞业务. 多次调用合并为一次 (简单去抖).
+
+    适用: 摄入完成 / 删除文档后.
+    """
+    if not settings.is_persist_enabled():
+        return
+    if not settings.persist_on_write:
+        return
+    if _state["pending_push"]:
+        return  # 已有 pending, 跳过
+
+    _state["pending_push"] = True
+
+    async def _delayed_push() -> None:
+        # 简单去抖: 延迟 30s, 把同一秒内的多次写合并
+        await asyncio.sleep(30)
+        if _state["pending_push"]:
+            await push_to_hf()
+
+    asyncio.create_task(_delayed_push())
+
+
+async def _ensure_repo_exists() -> None:
+    """首次部署时自动创建 HF Dataset repo."""
+    repo_id = settings.hf_persist_repo
+    token = settings.hf_token.get_secret_value()
+    loop = asyncio.get_running_loop()
+    try:
+        await loop.run_in_executor(
+            _persist_executor,  # ✅ 独立池
+            lambda: create_repo(
+                repo_id=repo_id,
+                repo_type="dataset",
+                token=token,
+                private=True,
+                exist_ok=True,
+            ),
+        )
+        logger.info("Created persist repo: %s", repo_id)
+    except Exception as e:  # noqa: BLE001
+        logger.error("Failed to create persist repo: %s", e)
+
+
+def persist_status() -> dict:
+    """供 /readyz 暴露持久化状态."""
+    return {
+        "enabled": settings.is_persist_enabled(),
+        "mode": _state["mode"],
+        "pending_push": _state["pending_push"],
+        "repo": settings.hf_persist_repo or None,
+    }
+
+
+def _wipe_local_data() -> None:
+    """测试用: 清空本地 data 目录."""
+    for d in (sqlite_dir(), chroma_dir(), upload_dir()):
+        if d.exists():
+            shutil.rmtree(d, ignore_errors=True)
+
+
+__all__ = [
+    "restore_from_hf",
+    "push_to_hf",
+    "schedule_push",
+    "persist_mode",
+    "persist_status",
+]