diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 450a779..b5daea9 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -84,3 +84,23 @@ jobs:
           tests/unit/test_platform.py
           tests/unit/test_mcp_market_service.py
           tests/unit/test_skill_market_service.py
+
+  frozen-rag-smoke:
+    runs-on: windows-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install build dependencies
+        run: pip install -e ".[dev,build]"
+
+      - name: Build frozen application
+        run: pyinstaller misaka.spec
+
+      - name: Run frozen RAG smoke test
+        run: .\dist\Misaka\Misaka.exe --rag-smoke
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 556f3be..1b2dbee 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -55,6 +55,9 @@ jobs:
       - name: Build with PyInstaller
         run: pyinstaller misaka.spec
 
+      - name: Run frozen RAG smoke test
+        run: .\dist\Misaka\Misaka.exe --rag-smoke
+
       - name: Install Inno Setup
         run: choco install innosetup -y
 
diff --git a/docs/reviews/knowledge-base-audit-2026-08-11.md b/docs/reviews/knowledge-base-audit-2026-08-11.md
new file mode 100644
index 0000000..1b610a4
--- /dev/null
+++ b/docs/reviews/knowledge-base-audit-2026-08-11.md
@@ -0,0 +1,436 @@
+# Misaka 知识库（RAG）功能审查报告
+
+## 1. 审查摘要
+
+审查日期：2026-08-11
+审查分支：`codex/fix-issue-2-marketplace`
+基线提交：`940132eadc2a707886c78f0e30d7e564d55ac493`（`fix: make environment setup reliable (#15)`）
+运行环境：Windows / Python 3.13.9 / Flet 0.80.5
+审查方式：源码走查、架构与需求对照、静态检查、现有测试、故障注入、并发复现、真实文件解析、现有 PyInstaller 产物检查。
+
+### 总体结论
+
+知识库模块的分层和基础主流程已经成形：知识库 CRUD、文档上传、解析、分块、嵌入、SQLite/SeekDB 向量后端、混合检索、聊天上下文注入和管理 UI 均已接通。在当前源码开发环境中，已有自动化测试和本次构造的正常路径均可通过。
+
+但是，当前版本不能判定为“可稳定正常使用”或“发布就绪”。失败重建会先销毁旧索引且保留虚假统计；处理中的文档可被删除并留下孤儿向量；远程向量清理失败仍会删除本地记录；检索配置有部分完全不生效；RRF 会把同一分块作为两个结果；打包配置又排除了默认 BM25 必需的 NumPy，现有冻结产物中也没有 sqlite-vec 的 `vec0.dll`。这些问题会导致数据丢失、索引与界面状态不一致、检索失效或发布版知识库不可用。
+
+本次共记录 20 项问题：
+
+| 等级 | 数量 | 含义 |
+|---|---:|---|
+| P0 | 2 | 发布阻断：可造成已建索引丢失，或冻结发布版核心能力不可用 |
+| P1 | 7 | 严重：数据一致性、后端切换、检索正确性或失败可见性问题 |
+| P2 | 8 | 中等：格式兼容、性能、安全边界、验证和确定性问题 |
+| P3 | 3 | 较低：规模化、资源管理和长期维护风险 |
+
+建议在修复 P0 与 P1、增加对应回归测试，并完成冻结版 smoke test 和真实远程 SeekDB/模型端点 E2E 后，再将该功能视为可发布。
+
+## 2. 审查范围与架构
+
+### 2.1 覆盖范围
+
+- `misaka/services/knowledge/`：知识库、文档、RAG 编排器及 LangChain/SeekDB 适配器。
+- `misaka/ui/knowledge/`：知识库列表、详情、创建编辑、文档上传/查看/重处理、聊天选择器。
+- `misaka/services/chat/preprocessors.py`：聊天发送前的 RAG 检索和上下文注入。
+- `misaka/ui/chat/components/message_input.py`：知识库选择入口。
+- `misaka/ui/settings/components/vector_backend_panel.py` 与 `misaka/main.py`：向量后端配置及重建状态。
+- `misaka/db/`：知识库、文档、分块、设置的模型、迁移和 CRUD。
+- `misaka.spec`、`pyproject.toml`：打包与运行依赖。
+- `tests/integration/test_knowledge_backend_flow.py`、知识库相关单元测试和全量测试。
+- `docs/demand/KNOWLEDGE_BASE_DESIGN.md`、`docs/demand/RAG_BEST_PRACTICES.md`、`docs/plans/KNOWLEDGE_BASE_SEEKDB_TODO.md`：需求与实现对照。
+
+审查期间工作树存在由其他任务产生的 Marketplace/MCP/README/CI 等未提交修改；本报告未修改这些文件。知识库目录和数据库相关目录在审查期间未发现并发改动。`misaka/main.py` 的并发修改仅涉及 MCP Marketplace 服务注入，不改变本报告引用的向量后端判断逻辑。
+
+### 2.2 实际数据流
+
+```text
+UI 上传文件
+  -> DocumentService 校验、计算 SHA-256、复制到 ~/.misaka/knowledge_bases/<kb>/
+  -> RAGOrchestrator 解析 -> 分块 -> 嵌入 -> 写向量后端
+  -> DocumentService 写 kb_chunks、更新 kb_documents 和知识库统计
+
+聊天选择知识库
+  -> RAGPreprocessor 构造各知识库嵌入配置
+  -> RAGOrchestrator 对每个知识库嵌入查询
+  -> 向量检索 + BM25 -> RRF -> 可选 reranker
+  -> format_context() 拼接 XML 风格上下文 -> 发送给模型
+```
+
+主要架构优点：
+
+- 解析器、分块器、嵌入器、向量存储、检索器、重排器有抽象接口，SQLite 与 SeekDB 后端的切换边界清晰。
+- 数据库迁移包含知识库、文档、分块和后端配置，基础 CRUD 较完整。
+- 聊天预处理器与 UI 选择器解耦，未选择知识库时不会增加 RAG 开销。
+- 文档 hash 去重、文件大小限制、嵌入批处理、模型维度记录、索引 stale/pending 状态等基础机制已实现。
+
+## 3. 验证结果
+
+### 3.1 自动化与静态检查
+
+| 检查 | 结果 | 说明 |
+|---|---|---|
+| 全量测试 `python -m pytest -q` | 通过 | `602 passed in 12.86s` |
+| 知识库定向测试 | 通过 | 17 项：SQLite、fake SeekDB、重建状态、创建对话框、后端面板、服务容器 |
+| Ruff `ruff check misaka/` | 通过 | 无 lint 错误 |
+| Mypy（知识库相关 30 个源文件，禁用增量缓存） | 通过 | 无类型错误 |
+| `pip check` | 通过 | 当前开发环境无破损依赖 |
+| 核心知识库测试覆盖率 | 61% | 1,055 statements / 410 missed；关键失败路径覆盖不足 |
+
+覆盖率较低的核心文件包括：`kb_service.py` 40%、`document_service.py` 52%、解析器 31%、分块器 22%、embedding 0%、reranker 0%。现有集成测试中的解析、嵌入和 SeekDB 多数使用 fake，因此“602 项通过”只证明已有断言满足，不能覆盖生产端点、冻结打包和本报告中的故障路径。
+
+### 3.2 正常路径验证
+
+| 功能 | 结果 | 备注 |
+|---|---|---|
+| 创建/读取/更新/删除知识库 | 基本通过 | 数据库与 UI 路径已接通；异常清理问题见后文 |
+| 文档上传、去重、分块、持久化 | 开发环境通过 | fake embedding + SQLite/SeekDB 路径可完成 |
+| TXT/Markdown/DOCX/XLSX/PDF 解析 | 部分通过 | 文本可读取；页/工作表 metadata 错误；`.xls` 不可用 |
+| SQLite sqlite-vec 写入与查询 | 开发环境通过 | 真实 sqlite-vec 扩展在当前 Python 环境可加载 |
+| SeekDB 适配器 API 兼容性 | 合约通过 | 与本机 pyseekdb 1.4.0 签名匹配；未连接真实服务器 |
+| 聊天 RAG 注入 | 基础路径通过 | fake embedding/检索可注入上下文；配置、融合和错误反馈有缺陷 |
+| 后端切换与重建提示 | 部分通过 | 后端类型变化可标 stale；同类型远程目标变化不能识别 |
+
+### 3.3 故障注入与边界复现
+
+以下是本次独立构造、可稳定复现且现有测试未覆盖的关键结果：
+
+1. 重处理时模拟新摄取失败：数据库真实分块变为 0，但文档和知识库仍显示 `chunk_count=1`，且知识库仍出现在聊天可选列表。
+2. 全量重建时构造 `storage_path` 为空：返回 `success_count=1, error_count=0`，旧分块已删除，知识库却被标为 `active` 和已重建。
+3. 嵌入进行中删除文档：删除按钮可用；最终文档行和文件被删除，但 fake 向量后端保留已写入向量，上传任务因外键错误返回 `error`。
+4. 模拟远程向量删除失败：删除 API 仍返回成功，本地文档和文件消失，远程向量仍存在。
+5. `seekdb_remote` 从一组 host/database 改到另一组：`changed=false`、索引不 stale、原知识库仍可选择，但新目标没有原索引。
+6. 同一分块同时命中向量与 BM25：RRF 输出两个内容完全相同、ID 不同的结果。
+7. 配置 `top_k=1`、`similarity_threshold=0.9`，检索仍保留 3 条（分数 0.95/0.5/0.1），说明两个用户可见配置未传入/未应用。
+8. 多工作表 XLSX 的第二张表内容被分块，但所有分块 metadata 都标成第一张表；两页 PDF 的所有分块都标成第一页；旧 `.xls` 被 openpyxl 直接拒绝。
+
+## 4. 详细问题
+
+### KB-001（P0，已修复）：重处理/重建先删除旧索引，失败后造成数据丢失和虚假可用状态
+
+证据：
+
+- `misaka/services/knowledge/document_service.py:246-271` 在新摄取前删除旧向量和 `kb_chunks`；摄取失败时仅写 `status=error`，没有恢复旧索引，也没有把 `content_text/chunk_count` 清零。
+- `misaka/services/knowledge/kb_service.py:170-187` 全量重建先 drop 整个知识库向量集合；每个文档处理完成后依据异常计数决定 `active/error`。
+- `misaka/services/knowledge/kb_service.py:243-247` 已删除旧分块后，`storage_path` 为空会直接 `return`，调用方仍增加成功数。
+- `misaka/services/knowledge/kb_service.py:99-108` 用文档上的反规范化 `chunk_count` 汇总，而不是统计真实 `kb_chunks`。
+- `misaka/services/knowledge/kb_service.py:318-339` 聊天可选性依赖上述缓存统计和 KB 状态。
+
+影响：一次临时模型错误、文件丢失或网络异常就能销毁原本可用的索引；随后 UI 仍可能显示有分块并允许聊天选择，实际检索为空。这既是数据丢失，也是可用性状态错误。
+
+建议：采用“新索引版本构建 -> 校验分块/向量数量 -> 数据库事务切换 active version -> 最后删除旧版本”的 copy-on-write 流程。缺失源文件必须计为失败。所有异常路径从真实 chunk 行重算统计，禁止在未成功切换时调用 `mark_index_rebuilt()`。
+
+### KB-002（P0，已修复）：PyInstaller 发布版缺少默认 RAG 所需运行依赖
+
+证据：
+
+- `misaka.spec:165-170` 明确排除 `numpy`。
+- `misaka/services/knowledge/rag/langchain/retriever.py:65-77` 默认 SQLite 混合检索在请求时导入 `rank_bm25.BM25Okapi`；`rank_bm25` 顶层依赖 NumPy。
+- `misaka/services/knowledge/rag/langchain/vector_store.py:140-147` 运行时动态导入 `sqlite_vec` 并加载原生扩展。
+- 现有 `build/Misaka/Analysis-00.toc` 能看到 `rank_bm25` 和 `sqlite_vec` Python 模块，但 `dist/Misaka/_internal` 中没有 NumPy，也没有 `sqlite_vec/vec0.dll`；所有 PyInstaller TOC 中均未找到 `vec0.dll`。
+
+影响：冻结发布版默认 BM25 路径会因 NumPy 被排除而失败；sqlite-vec 原生扩展也很可能无法加载。检索层会捕获部分异常并退化成空结果，因此用户可能只感知到“知识库没有效果”，而不是明确崩溃。SeekDB/pyseekdb 同样依赖 NumPy，排除规则还会影响替代后端。
+
+建议：移除不兼容的 NumPy排除项，显式收集 sqlite-vec 原生库（使用 `collect_dynamic_libs/collect_data_files` 或自定义 hook），并在 CI 中对冻结目录执行：启动、创建临时 KB、加载 sqlite-vec、写入、查询、BM25 融合的 smoke test。旧产物日期早于本次审查，修复后必须重新构建验证，不能只依赖静态分析。
+
+### KB-003（P1，已修复）：处理中的文档可被删除，导致孤儿向量和外键失败
+
+证据：
+
+- `misaka/services/knowledge/document_service.py:82-108` 先创建文档记录，再等待完整摄取；向量由编排器先写入。
+- `misaka/ui/knowledge/components/document_list.py:121-140` 查看、重处理、删除按钮不根据 `pending/parsing/embedding` 状态禁用。
+- `misaka/services/knowledge/document_service.py:204-227` 删除不持有文档级锁，也没有取消正在运行的摄取任务。
+
+影响：删除与上传并发时，向量可以在删除之后写入，随后 chunk 行因文档外键已不存在而失败，形成无法由普通 UI 定位/清理的孤儿向量。在远程后端中还构成数据留存风险。
+
+建议：每个文档/知识库引入异步互斥与取消令牌；处理期间禁用删除/重处理，或删除操作先取消并等待任务完成。无论文档行是否仍存在，异常清理都必须以 `document_id` 删除已写向量。
+
+### KB-004（P1，已修复）：向量清理异常被吞掉，本地删除仍报告成功
+
+证据：
+
+- `misaka/services/knowledge/document_service.py:211-227` 捕获向量删除异常后继续删除数据库和文件并返回成功。
+- `misaka/services/knowledge/kb_service.py:82-95` 删除整个 KB 时采用相同策略。
+
+影响：网络中断或 SeekDB 故障时，本地元数据被永久删除，远程向量仍保存；之后无法从正常业务记录中得知清理范围，也可能在复用表/过滤错误时产生陈旧检索和隐私风险。
+
+建议：删除应有可恢复状态（`deleting/delete_failed`）和持久化清理队列。只有远程清理确认完成后才最终删除元数据；若业务要求本地先删除，也至少保留 tombstone、backend fingerprint、表名和待清理 document IDs，并向用户明确报告部分失败。
+
+### KB-005（P1，已复现）：SeekDB 远程目标变化不会把索引标记为 stale
+
+证据：
+
+- `misaka/main.py:239-264` 仅比较 `previous != vector_backend`。当后端类型始终是 `seekdb_remote`，修改 host、port、database 或连接身份不会触发 `mark_all_kb_indexes_stale()`。
+
+影响：编排器开始连接新数据库，但 UI 仍把基于旧数据库构建的 KB 视为可用；查询新目标时得到空结果或错误。若不同环境存在同名表，还可能检索到不属于该 KB 的数据。
+
+建议：存储并比较后端身份 fingerprint（backend type + host + port + database/tenant；密码变化通常不需要重建）。目标身份变化时全部标 stale，并阻止选择直到重建成功。
+
+### KB-006（P1，已复现）：RRF 使用不一致的分块 ID，重复返回同一内容
+
+证据：
+
+- `misaka/services/knowledge/rag/langchain/retriever.py:85-94` BM25 结果 ID 被构造成 `chunk_<index>`。
+- 向量写入使用 `metadata["chunk_db_id"]` 的 UUID（`misaka/services/knowledge/rag_orchestrator.py:118-137`）。
+- `misaka/services/knowledge/rag/langchain/retriever.py:98-121` RRF 按 `chunk_id` 合并，因此同一分块被当成两个结果。
+
+影响：真正同时满足语义和关键词的分块无法获得融合加权，反而占用两个 top-k 名额，重复上下文浪费 token，并可能挤掉其他相关内容。
+
+建议：BM25 结果优先使用 `chunk.metadata["chunk_db_id"]`，缺失时才使用稳定、带文档 ID 的 fallback。融合后增加按真实 chunk ID 的唯一性断言和回归测试。
+
+### KB-007（P1，已复现）：用户配置的 `top_k` 和 `similarity_threshold` 不生效
+
+证据：
+
+- 两项配置可在 `misaka/ui/knowledge/components/kb_create_dialog.py:88-94` 编辑并保存。
+- `misaka/services/chat/preprocessors.py:99-108` 调用 `retrieve()` 时没有传入每个 KB 的 `top_k` 或 threshold。
+- `misaka/services/knowledge/rag_orchestrator.py:163-219` 只接受单个全局 `top_k=5`，没有相似度阈值参数。
+- `reranker_top_k` 会生效，但在多 KB 场景中只取某一个 KB 的配置，见 KB-017。
+
+影响：界面提供了看似有效的检索精度控制，但实际结果数和低分过滤不受其影响。用户可能错误地认为已经提高了检索阈值或限制了上下文大小。
+
+建议：明确多 KB 合并语义。推荐每个 KB 使用自身的 candidate top-k/threshold 先过滤，再以会话级 final top-k 合并；阈值需根据后端统一后的相似度定义应用。UI 保存后增加配置传播测试。
+
+### KB-008（P1，源码确认）：修改分块参数不会重建已有索引
+
+证据：
+
+- `misaka/ui/knowledge/components/kb_create_dialog.py:168-181` 只有 embedding router/model 变化才进入确认和重建流程。
+- `chunk_size`、`chunk_overlap` 改动直接保存，但现有 `kb_chunks` 和向量保持旧切分。
+
+影响：配置页面显示的是新参数，实际检索索引仍使用旧参数，且没有 stale 标记，后续排障无法信任 KB 配置。
+
+建议：将 chunk size/overlap、解析策略和 embedding 模型/维度都纳入 `index_fingerprint`。任何影响索引内容的字段变化都标 stale 并要求重建；也可先保存为 pending config，成功切换后再替换 active config。
+
+### KB-009（P1，源码确认）：没有 10 秒总检索超时，且单 KB 失败会静默降级
+
+证据：
+
+- 需求文档要求检索超过 10 秒取消 RAG 并通知用户。
+- `misaka/services/knowledge/rag_orchestrator.py:163-219` 没有包围整个检索的 timeout；embedding 客户端自身 60 秒、reranker 30 秒，SeekDB 同步调用没有统一查询超时。
+- 同文件 `:190-216` 捕获各 KB 的 embedding、retrieval、reranker 异常并继续，最终可能返回空列表。
+- `misaka/services/chat/preprocessors.py:103-115` 空结果被视为正常，失败通知仅覆盖抛到外层的异常。
+
+影响：聊天发送可能长时间卡住；服务故障时用户收到普通模型回答，却不知道知识库根本没有参与，产生高风险的错误信任。
+
+建议：用 `asyncio.timeout(10)` 或可配置截止时间包围整个 RAG；返回结构化的 `results + per_kb_errors + timed_out`，在保留部分结果时提示部分失败，全部失败时必须通知并在消息上标注 RAG 未生效。
+
+### KB-010（P2，已复现）：PDF 页码、Excel 工作表 metadata 丢失，且声明支持的 `.xls` 实际不可读
+
+证据：
+
+- `misaka/services/knowledge/rag/langchain/parser.py:34-45` 把 loader 返回的所有 Document 文本拼成一个字符串，只保留第一项 metadata。
+- `_OpenpyxlLoader` 本来会在 `:106-145` 为每个 worksheet 创建 Document，但随后被上述逻辑压平。
+- `misaka/services/knowledge/document_service.py:27-36` 声明 `.xls` 和 `.xlsx` 都支持，而 parser 对两者都使用 openpyxl；openpyxl 明确拒绝旧 `.xls`。
+
+影响：引用页码/工作表错误，文档查看和未来引用展示不可信；用户可以选择 `.xls`，但上传必然失败。
+
+建议：让 parser 返回带各自 metadata 的逻辑文档列表，分块器逐 Document 分块并继承 `page/sheet_name`；去掉 `.xls` 声明，或引入 xlrd 等真正支持旧格式的 loader 和测试样本。
+
+### KB-011（P2，已复现）：模型可用性检查忽略 `is_selected`
+
+证据：
+
+- `misaka/services/knowledge/kb_service.py:112-140` 只判断 router 配置中是否存在同 model ID，不检查模型是否被用户选中。
+- 实际上传/重处理的模型选择来自 `misaka/services/settings/router_config_service.py:274-278` 的 selected-only 列表。
+
+影响：UI 可显示 embedding 模型可用，但上传对话框找不到该模型并提前返回，用户没有得到一致的不可用提示。
+
+建议：统一“可用”的定义和查询入口，同时校验 router 启用状态、模型 `is_selected`、base URL/API key 基本完整性；上传按钮应显示明确错误而不是静默返回。
+
+### KB-012（P2，源码确认）：多个重操作仍在 UI 事件循环同步执行
+
+证据：
+
+- `misaka/services/knowledge/document_service.py:59-86` 在 async 上传方法中同步执行 stat、SHA-256 全文件读取和最多 100 MB 文件复制。
+- `misaka/services/knowledge/rag_orchestrator.py:106-113` 同步分块；SQLite BM25 会读取该 KB 全部分块并同步构建/计算语料。
+- `misaka/services/knowledge/rag/langchain/retriever.py:51-59` 同步执行向量查询与 BM25。
+- SeekDB upsert/search/delete/refresh 是同步 SDK 调用，但从 async UI/RAG 路径直接调用。
+
+影响：大文件、大知识库或远程网络抖动时会卡住 Flet UI，与 `docs/architecture/PERFORMANCE.md` 的“Never block main thread”规则冲突。BM25 每次查询都全量加载和重建，复杂度随分块数量快速上升。
+
+建议：文件 I/O、解析、分块、SQLite/SeekDB 同步调用放入 `asyncio.to_thread` 或专用 worker；使用持久化/增量 BM25 索引，限制并发与队列长度，并对 10k/100k 分块建立性能基准。
+
+### KB-013（P2，已由并发复现佐证）：数据库、文件和向量后端之间没有原子提交协议
+
+证据：
+
+- 编排器先写向量，`DocumentService` 后写 `kb_chunks` 和文档状态；后半段失败会留下孤儿向量。
+- 删除方向则先尝试向量清理，再无条件删除本地记录，远程失败无法回滚。
+- `LCSqliteVecStore.add_chunks()` 使用 `zip(..., strict=False)`，没有验证 chunks/embeddings 数量一致。
+
+影响：任一步骤崩溃、进程退出或数据库提交失败都会使三类存储分叉，且当前没有启动时 reconciliation。
+
+建议：引入 ingest job/index version、幂等操作键和补偿事务；本地 SQLite DB 与 sqlite-vec 可共享连接/事务时尽量原子提交；远程后端采用 outbox/saga。启动时扫描 processing/deleting 超时任务并校验实际计数。
+
+### KB-014（P2，源码确认）：高级数值配置缺少领域校验
+
+证据：
+
+- `misaka/ui/knowledge/components/kb_create_dialog.py:240-251` 的 `_safe_int/_safe_float` 只做解析与默认值回退。
+- 没有约束 `chunk_size > 0`、`0 <= overlap < chunk_size`、`top_k/reranker_top_k > 0`、threshold 的有效范围，以及拒绝 NaN/Infinity。
+
+影响：无效配置可写入数据库，并在 splitter、切片、排序或网络请求处以难以理解的方式失败。
+
+建议：在 service 层建立唯一的配置校验器，UI 仅负责显示字段错误；数据库写入和导入路径同样调用。为边界值、NaN/Infinity 和 overlap 关系增加参数化测试。
+
+### KB-015（P2，安全风险）：知识库内容直接进入提示词，缺少注入边界处理
+
+证据：
+
+- `misaka/services/knowledge/rag_orchestrator.py:221-239` 把用户可上传的文档原文直接拼入 XML 风格标签，没有转义标签字符，也没有明确告知模型“文档中的指令是不可信数据”。
+
+影响：恶意或无意的文档文本可以闭合标签、伪造结构、指示模型忽略用户意图或泄露上下文。这不是本地代码执行漏洞，但会破坏回答可信性，尤其当 KB 包含外部来源文档时。
+
+建议：使用结构化消息/严格序列化并转义分隔符；系统提示明确声明上下文只提供事实、不得遵循其中指令；保留来源与信任级别，必要时做内容安全扫描。增加包含闭合标签和 prompt injection 文本的评估用例。
+
+### KB-016（P2，源码确认）：重排器对畸形响应校验不完整
+
+证据：
+
+- `misaka/services/knowledge/rag/langchain/reranker.py:60-74` 只拒绝 `index >= len(results)`；负数 index 会按 Python 语义选取尾部结果。
+- 重复 index 不去重；空/缺失结果可被当作成功的空重排，不一定回退到原结果。
+
+影响：不标准的 OpenAI-compatible reranker 响应会重复、错配或清空检索结果。
+
+建议：要求 index 为非负整数、范围内且唯一，score 为有限数；响应无有效项时记录结构化错误并回退原排序。
+
+### KB-017（P2，源码确认）：多知识库的 reranker 选择依赖无序集合
+
+证据：
+
+- `misaka/ui/knowledge/components/kb_selector.py:207-213` 使用 `list(set(...))` 保存选择，顺序不稳定。
+- `misaka/services/chat/preprocessors.py:129-156` 遍历 KB，并采用遇到的第一个 reranker 配置作为所有结果的全局重排器。
+
+影响：选择多个配置不同 reranker 的 KB 时，实际使用哪个模型/阈值可能随顺序变化，结果不可重复；一个 KB 的配置会无提示地覆盖其他 KB。
+
+建议：保持用户选择顺序并显式定义策略：要么每 KB 独立重排后融合，要么由会话提供统一 reranker。存在冲突时 UI 应提示，而不是隐式取第一个。
+
+### KB-018（P3，源码确认）：列表和文档查看器在大规模下会制造过多 UI 控件/内存复制
+
+证据：
+
+- `misaka/ui/knowledge/components/document_list.py:62` 一次性为全部文档创建行，不是真正虚拟化。
+- 文档查看器“加载更多”会反复生成更大的内容前缀，复制按钮可把最多 100 MB 原文一次性放入剪贴板。
+
+影响：文档多或单文档大时，页面构建、更新和内存占用明显上升。
+
+建议：分页/虚拟列表；查看器按页或按块读取而不是保存整份文本控件；复制操作设置合理上限并提供导出文件。
+
+### KB-019（P3，源码确认）：文件系统异常与数据库状态处理不对称
+
+证据：
+
+- 创建 KB 时数据库 commit 早于存储目录创建，mkdir 失败会留下不可用 KB 记录。
+- 删除 KB 使用 `shutil.rmtree(..., ignore_errors=True)`，目录删除失败仍返回成功。
+
+影响：磁盘权限、文件占用或空间错误后出现幽灵记录/残留文件，用户无法从 UI 得知。
+
+建议：创建采用补偿删除或先准备临时目录再提交；删除记录并展示文件清理失败，保留重试任务。
+
+### KB-020（P3，设计风险）：表名截断和旧后端资源缺少生命周期管理
+
+证据：
+
+- 每 KB 的向量表名只使用 UUID 前 8 个十六进制字符，约 32 bit 命名空间；规模增大后碰撞概率不可忽略。
+- 后端切换只标 stale，不清理旧后端集合；删除 KB 后 pending rebuild IDs 也没有统一清除。
+
+影响：长期运行可能积累远程/本地残留资源；极大规模下表名碰撞可能让两个 KB 共享/覆盖表。
+
+建议：使用完整 UUID 或至少 128-bit 编码；为后端资源记录 backend fingerprint、collection ID 和生命周期状态，提供迁移/清理命令及 orphan audit。
+
+## 5. 与设计文档的主要偏差
+
+| 设计要求 | 当前实现 | 结论 |
+|---|---|---|
+| RAG 超过 10 秒取消并通知 | 无总 timeout；分支异常被吞并退化为空 | 未实现 |
+| 文档后台处理不阻塞 UI | 解析使用 `to_thread`，但 hash/copy/chunk/BM25/SeekDB 等仍同步 | 部分实现 |
+| 嵌入批次并发度 3 | 批次顺序执行 | 未实现 |
+| 上传显示 parse/chunk/embed 进度 | 对话框只显示每个文件的粗状态，未传 `on_progress` | 部分实现 |
+| 上传路径安全检查 | 存在相关设计，但上传链路未调用统一 `is_path_safe` | 未实现 |
+| 删除处理中文档前等待或中止任务 | 操作不按状态禁用，也无任务取消/锁 | 未实现 |
+| KB 在处理期使用 building/error 状态机 | 普通上传/重处理通常不改变 KB 状态 | 部分实现 |
+| PDF 按页、Excel 按表保留 metadata | 全部文本压平并只保留首项 metadata | 未实现 |
+| 聊天选择依据真实可用嵌入分块 | 依赖反规范化的 KB `chunk_count` | 不可靠 |
+| KB 的 top-k/threshold 配置生效 | 两者未传入检索 | 未实现 |
+
+## 6. 修复优先级与实施建议
+
+### 第一阶段：阻止数据丢失和发布版失效
+
+1. 重构 ingest/rebuild 为版本化、copy-on-write 索引切换；任何失败保留旧 active 索引。
+2. 为上传、重处理、重建、删除增加 job 状态、文档/KB 互斥和取消等待。
+3. 向量删除失败进入持久化清理队列，禁止无痕吞错。
+4. 修复 PyInstaller：NumPy、rank-bm25、sqlite-vec 原生 DLL，并新增冻结版 RAG smoke test。
+5. 从真实 `kb_chunks`/向量统计做 reconciliation，修复已有虚假计数和孤儿数据。
+
+### 第二阶段：恢复检索语义正确性
+
+1. 统一 BM25 与向量结果的 `chunk_db_id`，增加融合去重测试。
+2. 传播并应用每 KB 的 top-k/threshold；定义多 KB final top-k 和 reranker 策略。
+3. 后端 fingerprint 覆盖远程 host/port/database，目标变化即 stale。
+4. 加入 10 秒总 deadline、部分失败结构化结果和用户可见通知。
+5. chunk/embedding/parser 配置纳入 index fingerprint，变更必须重建。
+
+### 第三阶段：兼容性、性能与可信性
+
+1. 保留 PDF page、Excel sheet metadata；删除虚假的 `.xls` 支持或加入真正 loader。
+2. 把文件 I/O、chunk、BM25、SeekDB 同步调用移出 UI 事件循环；构建增量 BM25。
+3. service 层统一数值验证、模型可用性定义和 reranker 响应验证。
+4. 对知识库文本做结构化转义和 prompt-injection 防护提示。
+5. 引入大规模基准、分页/虚拟化和后端 orphan audit 工具。
+
+## 7. 建议补充的测试
+
+- 重处理失败后旧索引仍可查询，统计不变。
+- 缺失 storage file 的全量重建必须失败且不能 `mark_index_rebuilt`。
+- 上传与删除/重处理并发，确保取消、无 FK 错误、无孤儿向量。
+- 远程删除失败的 tombstone/outbox 重试和最终一致性。
+- remote host/database 变化触发 stale；仅密码轮换不触发重建。
+- 同一分块同时命中 vector/BM25，只输出一次且融合分数增加。
+- top-k、threshold、chunk 配置从 UI -> DB -> preprocessor -> retriever 的传播测试。
+- PDF 多页、XLSX 多 sheet metadata；`.xls` 的明确支持或拒绝测试。
+- 负 index、重复 index、NaN score、空 reranker 响应。
+- 10 秒 timeout、单 KB 部分失败、全部失败及用户通知。
+- 10k/100k chunks 的上传、BM25、切换页面和聊天延迟基准。
+- PyInstaller 构建后 sqlite-vec load/write/search + BM25 的 smoke test。
+
+## 8. 第一阶段实施记录（2026-08-11）
+
+### 8.1 已完成范围
+
+本次仅落实第 6 节“第一阶段”的五项内容，对应 KB-001、KB-002、KB-003、KB-004 及统计 reconciliation。第二、三阶段问题仍保持原审查结论，未在本次修复中标记完成。
+
+### 8.2 实施方案与变更
+
+1. 新增数据库 migration v7。`knowledge_bases.active_index_version` 指向聊天正在使用的版本；`kb_chunks.index_version` 将分块与索引版本绑定；历史未版本化数据继续使用空版本对应的旧表名，保证升级兼容。
+2. 新增 copy-on-write 索引流程。上传、重处理、全量重建和删除都先构建完整的新版本向量表/collection；全部文件解析、分块、嵌入和向量写入成功后，才在一个 SQLite 事务中写入分块、更新文档元数据/真实统计并切换 active version。失败或取消会删除 staging 版本，旧 active version、旧统计和聊天可用性保持不变。
+3. 新增 `kb_jobs` 持久化作业状态及 KB 级异步互斥协调器。上传、重处理、重建、文档删除和 KB 删除均串行化；删除操作先取消并等待同一 KB 的活动任务。处理中的文档在 UI 中禁用重处理和删除按钮。
+4. 新增 `kb_cleanup_jobs` 持久化清理队列。远端/本地向量版本删除失败不再被吞掉：会记录待重试任务，应用启动时重试；成功后再删除退役版本的 `kb_chunks` 行。SeekDB adapter 不再吞掉 `delete_collection` 异常。
+5. 统计和聊天选择改为读取 active index version 的真实 `kb_chunks`。这会修复历史反规范化 `document_count/chunk_count` 与实际分块不一致时的虚假可选状态。
+6. PyInstaller spec 显式保留 NumPy、`rank_bm25`、`sqlite_vec`，并通过 `collect_dynamic_libs` / `collect_data_files` 收集 sqlite-vec 原生库。新增 `--rag-smoke`：在冻结产物中验证 sqlite-vec 加载、写入、检索和 NumPy 支持的 BM25 融合；CI Windows job 和 Windows release job 均会执行该 smoke test。
+
+### 8.3 新增回归验证
+
+- 重处理解析失败后，active version、真实分块统计和聊天可选性保持不变。
+- 源文件缺失的全量重建返回失败，且不会执行 `mark_index_rebuilt()`。
+- 退役向量索引删除失败会进入持久化队列；恢复后可重试并清理历史分块。
+- 删除会取消并等待活动上传，确保不会留下孤儿向量或外键失败。
+- 已执行针对性回归、静态检查和源代码 RAG smoke；冻结产物 smoke 由新增 CI job 验证。
+- 可选的真实 SeekDB 远程集成测试与真实 OpenAI-compatible embedding/reranker contract test（凭据由 CI secret 提供）。
+
+## 8. 最终判定
+
+| 判定项 | 结论 |
+|---|---|
+| 架构完整度 | 良好，抽象和模块边界清楚 |
+| 开发环境基础流程 | 可运行 |
+| 失败恢复与数据一致性 | 不合格，存在可复现数据丢失和孤儿数据 |
+| 检索逻辑正确性 | 部分正确，RRF、top-k、threshold、多 KB reranker 有实质错误 |
+| 文件格式语义 | 部分正确，内容可读但 page/sheet metadata 错，`.xls` 虚假支持 |
+| UI 响应与规模化 | 小规模可用，大文件/大 KB 有阻塞风险 |
+| 错误可观察性 | 不足，检索失败可能静默退化 |
+| 安全与隐私边界 | 需加强，存在孤儿远程向量和 prompt injection 风险 |
+| 冻结发布版 | 当前不可判定为可用，已有证据显示关键依赖缺失 |
+| 生产发布建议 | 暂缓；先完成 P0/P1 修复与真实/冻结 E2E |
+
+因此，本次审查的最终结论是：知识库功能已经达到“功能原型和开发环境主流程可演示”的程度，但尚未达到“可靠、逻辑一致、可安全发布”的标准。
diff --git a/misaka.spec b/misaka.spec
index c2ab681..08882b4 100644
--- a/misaka.spec
+++ b/misaka.spec
@@ -19,6 +19,7 @@ from pathlib import Path
 import certifi
 import flet
 import flet_desktop
+from PyInstaller.utils.hooks import collect_data_files, collect_dynamic_libs
 
 block_cipher = None
 
@@ -62,6 +63,12 @@ _certifi_datas = [
     (str(_certifi_pem), "certifi"),
 ]
 
+# sqlite-vec loads vec0 dynamically at runtime, so PyInstaller cannot infer
+# the native library from the import graph.  Collect its package data and
+# dynamic libraries explicitly; rank-bm25 also requires NumPy at runtime.
+_sqlite_vec_datas = collect_data_files("sqlite_vec")
+_sqlite_vec_binaries = collect_dynamic_libs("sqlite_vec")
+
 _datas = [
     (str(_i18n_dir / "en.json"), "misaka/i18n"),
     (str(_i18n_dir / "zh_CN.json"), "misaka/i18n"),
@@ -70,12 +77,13 @@ _datas = [
     *_flet_datas,
     *_flet_desktop_datas,
     *_certifi_datas,
+    *_sqlite_vec_datas,
 ]
 
 a = Analysis(
     [str(project_root / "misaka" / "main.py")],
     pathex=[str(project_root)],
-    binaries=[],
+    binaries=_sqlite_vec_binaries,
     datas=_datas,
     hiddenimports=[
         "misaka",
@@ -99,6 +107,22 @@ a = Analysis(
         "misaka.services.file",
         "misaka.services.file.file_service",
         "misaka.services.file.update_check_service",
+        "misaka.services.knowledge",
+        "misaka.services.knowledge.document_service",
+        "misaka.services.knowledge.frozen_smoke",
+        "misaka.services.knowledge.index_manager",
+        "misaka.services.knowledge.job_coordinator",
+        "misaka.services.knowledge.kb_service",
+        "misaka.services.knowledge.rag",
+        "misaka.services.knowledge.rag.abstractions",
+        "misaka.services.knowledge.rag.factory",
+        "misaka.services.knowledge.rag.langchain.chunker",
+        "misaka.services.knowledge.rag.langchain.embedding",
+        "misaka.services.knowledge.rag.langchain.parser",
+        "misaka.services.knowledge.rag.langchain.reranker",
+        "misaka.services.knowledge.rag.langchain.retriever",
+        "misaka.services.knowledge.rag.langchain.vector_store",
+        "misaka.services.knowledge.rag_orchestrator",
         "misaka.services.mcp",
         "misaka.services.mcp.mcp_service",
         "misaka.services.session",
@@ -154,6 +178,9 @@ a = Analysis(
         "watchdog",
         "watchdog.observers",
         "sqlite3",
+        "sqlite_vec",
+        "rank_bm25",
+        "numpy",
         "certifi",
     ],
     hookspath=[],
@@ -165,7 +192,6 @@ a = Analysis(
     excludes=[
         "tkinter",
         "matplotlib",
-        "numpy",
         "pandas",
         "scipy",
         "IPython",
diff --git a/misaka/db/database.py b/misaka/db/database.py
index d1796b1..53d8f70 100644
--- a/misaka/db/database.py
+++ b/misaka/db/database.py
@@ -15,6 +15,7 @@
 from misaka.db.models import (
     ChatSession,
     KBChunk,
+    KBCleanupJob,
     KBDocument,
     KnowledgeBase,
     Message,
@@ -362,6 +363,27 @@ def get_kb_chunks_by_document(self, doc_id: str) -> list[KBChunk]:
     def get_kb_chunks_by_kb(self, kb_id: str) -> list[KBChunk]:
         """Return all chunks for a knowledge base, ordered by chunk_index."""
 
+    @abstractmethod
+    def get_kb_chunks_by_index(
+        self, kb_id: str, index_version: str,
+    ) -> list[KBChunk]:
+        """Return chunks belonging to one immutable KB index version."""
+
+    @abstractmethod
+    def activate_kb_index(
+        self,
+        kb_id: str,
+        index_version: str,
+        chunks: list[KBChunk],
+        document_updates: dict[str, dict[str, Any]],
+        dimensions: int,
+    ) -> None:
+        """Atomically publish staged chunks and their corresponding document state."""
+
+    @abstractmethod
+    def delete_kb_chunks_by_index(self, kb_id: str, index_version: str) -> None:
+        """Remove persisted chunks belonging to a retired index version."""
+
     @abstractmethod
     def delete_kb_chunks_by_document(self, doc_id: str) -> None:
         """Delete all chunks belonging to a specific document."""
@@ -370,6 +392,32 @@ def delete_kb_chunks_by_document(self, doc_id: str) -> None:
     def update_kb_chunk_embedded(self, chunk_ids: list[str]) -> None:
         """Mark chunks as embedded (``is_embedded = 1``)."""
 
+    # ----- KB background jobs and durable cleanup -----
+
+    @abstractmethod
+    def create_kb_job(self, kb_id: str, document_id: str, operation: str) -> str:
+        """Create and return a durable KB operation record."""
+
+    @abstractmethod
+    def update_kb_job(self, job_id: str, status: str, error_message: str = "") -> None:
+        """Update a KB operation record."""
+
+    @abstractmethod
+    def create_kb_cleanup_job(
+        self, kb_id: str, index_version: str, operation: str, error_message: str,
+    ) -> str:
+        """Persist vector cleanup that must be retried."""
+
+    @abstractmethod
+    def get_pending_kb_cleanup_jobs(self) -> list[KBCleanupJob]:
+        """Return all cleanup jobs still awaiting successful vector deletion."""
+
+    @abstractmethod
+    def update_kb_cleanup_job(
+        self, job_id: str, status: str, error_message: str = "",
+    ) -> None:
+        """Record a cleanup attempt or completion."""
+
     # ----- Dashboard aggregation -----
 
     @abstractmethod
diff --git a/misaka/db/migrations.py b/misaka/db/migrations.py
index b1a33c8..c019809 100644
--- a/misaka/db/migrations.py
+++ b/misaka/db/migrations.py
@@ -12,7 +12,7 @@
 logger = logging.getLogger(__name__)
 
 # Current schema version. Increment when adding new migrations.
-SCHEMA_VERSION = 6
+SCHEMA_VERSION = 7
 
 
 def run_migrations(conn: sqlite3.Connection) -> None:
@@ -43,6 +43,9 @@ def run_migrations(conn: sqlite3.Connection) -> None:
     if current < 6:
         _migrate_v6(conn)
 
+    if current < 7:
+        _migrate_v7(conn)
+
     _set_version(conn, SCHEMA_VERSION)
     conn.commit()
 
@@ -295,3 +298,71 @@ def _migrate_v6(conn: sqlite3.Connection) -> None:
             updated_at TEXT NOT NULL DEFAULT (datetime('now'))
         )
     """)
+
+
+def _migrate_v7(conn: sqlite3.Connection) -> None:
+    """Migration v7: versioned KB indexes and durable cleanup jobs.
+
+    Existing rows continue to address the legacy vector-table name through
+    the empty version string.  New writes always receive an opaque version
+    and are made visible only after a complete index has been built.
+    """
+    logger.info("Running migration v7")
+    existing_tables = {
+        row[0]
+        for row in conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table'"
+        ).fetchall()
+    }
+    if "knowledge_bases" in existing_tables:
+        kb_columns = _get_column_names(conn, "knowledge_bases")
+        if "active_index_version" not in kb_columns:
+            conn.execute(
+                "ALTER TABLE knowledge_bases "
+                "ADD COLUMN active_index_version TEXT NOT NULL DEFAULT ''"
+            )
+
+    if "kb_chunks" in existing_tables:
+        chunk_columns = _get_column_names(conn, "kb_chunks")
+        if "index_version" not in chunk_columns:
+            conn.execute(
+                "ALTER TABLE kb_chunks "
+                "ADD COLUMN index_version TEXT NOT NULL DEFAULT ''"
+            )
+        conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_kb_chunks_index_version
+                ON kb_chunks(knowledge_base_id, index_version)
+        """)
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS kb_cleanup_jobs (
+            id TEXT PRIMARY KEY,
+            knowledge_base_id TEXT NOT NULL,
+            index_version TEXT NOT NULL DEFAULT '',
+            operation TEXT NOT NULL,
+            status TEXT NOT NULL DEFAULT 'pending',
+            attempts INTEGER NOT NULL DEFAULT 0,
+            error_message TEXT NOT NULL DEFAULT '',
+            created_at TEXT NOT NULL DEFAULT (datetime('now')),
+            updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+        )
+    """)
+    conn.execute("""
+        CREATE INDEX IF NOT EXISTS idx_kb_cleanup_jobs_pending
+            ON kb_cleanup_jobs(status, created_at)
+    """)
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS kb_jobs (
+            id TEXT PRIMARY KEY,
+            knowledge_base_id TEXT NOT NULL,
+            document_id TEXT NOT NULL DEFAULT '',
+            operation TEXT NOT NULL,
+            status TEXT NOT NULL DEFAULT 'queued',
+            error_message TEXT NOT NULL DEFAULT '',
+            created_at TEXT NOT NULL DEFAULT (datetime('now')),
+            updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+        )
+    """)
+    conn.execute("""
+        CREATE INDEX IF NOT EXISTS idx_kb_jobs_active
+            ON kb_jobs(knowledge_base_id, status, updated_at)
+    """)
diff --git a/misaka/db/models.py b/misaka/db/models.py
index 5208b1c..25a82f9 100644
--- a/misaka/db/models.py
+++ b/misaka/db/models.py
@@ -383,6 +383,11 @@ class KnowledgeBase:
 
     status: Literal["active", "building", "error"] = "active"
 
+    # The version suffix of the vector index currently served to chat.  An
+    # empty value refers to the pre-versioning table name used by v6 and
+    # earlier databases.
+    active_index_version: str = ""
+
     created_at: str = ""
     updated_at: str = ""
 
@@ -428,9 +433,28 @@ class KBChunk:
 
     is_embedded: int = 0
 
+    # Chunks are staged under a new index version and become visible only
+    # once that version has been atomically activated.
+    index_version: str = ""
+
     created_at: str = ""
 
 
+@dataclass
+class KBCleanupJob:
+    """A durable retry record for vector-index cleanup."""
+
+    id: str
+    knowledge_base_id: str
+    index_version: str
+    operation: str
+    status: str = "pending"
+    attempts: int = 0
+    error_message: str = ""
+    created_at: str = ""
+    updated_at: str = ""
+
+
 @dataclass
 class KBSearchResult:
     """A single RAG retrieval result (runtime model, not persisted)."""
diff --git a/misaka/db/row_mappers.py b/misaka/db/row_mappers.py
index d792609..d51e2c7 100644
--- a/misaka/db/row_mappers.py
+++ b/misaka/db/row_mappers.py
@@ -108,6 +108,7 @@ def row_to_knowledge_base(row: sqlite3.Row) -> KnowledgeBase:
         document_count=row["document_count"],
         chunk_count=row["chunk_count"],
         status=row["status"],
+        active_index_version=row["active_index_version"],
         created_at=row["created_at"],
         updated_at=row["updated_at"],
     )
@@ -143,5 +144,6 @@ def row_to_kb_chunk(row: sqlite3.Row) -> KBChunk:
         end_char=row["end_char"],
         metadata_json=row["metadata_json"],
         is_embedded=row["is_embedded"],
+        index_version=row["index_version"],
         created_at=row["created_at"],
     )
diff --git a/misaka/db/sqlite_backend.py b/misaka/db/sqlite_backend.py
index 4961d25..b573d45 100644
--- a/misaka/db/sqlite_backend.py
+++ b/misaka/db/sqlite_backend.py
@@ -20,6 +20,7 @@
 from misaka.db.models import (
     ChatSession,
     KBChunk,
+    KBCleanupJob,
     KBDocument,
     KnowledgeBase,
     Message,
@@ -731,8 +732,8 @@ def create_knowledge_base(self, kb: KnowledgeBase) -> None:
                 chunk_size, chunk_overlap,
                 top_k, similarity_threshold, reranker_top_k,
                 document_count, chunk_count,
-                status, created_at, updated_at)
-               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                 status, active_index_version, created_at, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
             (
                 kb.id, kb.name, kb.description,
                 kb.embedding_model_id, kb.embedding_router_config_id, kb.embedding_dimensions,
@@ -740,7 +741,8 @@ def create_knowledge_base(self, kb: KnowledgeBase) -> None:
                 kb.chunk_size, kb.chunk_overlap,
                 kb.top_k, kb.similarity_threshold, kb.reranker_top_k,
                 kb.document_count, kb.chunk_count,
-                kb.status, kb.created_at or _now(), kb.updated_at or _now(),
+                kb.status, kb.active_index_version,
+                kb.created_at or _now(), kb.updated_at or _now(),
             ),
         )
         self._maybe_commit()
@@ -769,7 +771,7 @@ def update_knowledge_base(self, kb_id: str, **kwargs: Any) -> None:
             "reranker_model_id", "reranker_router_config_id",
             "chunk_size", "chunk_overlap",
             "top_k", "similarity_threshold", "reranker_top_k",
-            "document_count", "chunk_count", "status",
+            "document_count", "chunk_count", "status", "active_index_version",
         }
         sets: list[str] = ["updated_at = ?"]
         params: list[Any] = [_now()]
@@ -874,7 +876,7 @@ def create_kb_chunks_batch(self, chunks: list[KBChunk]) -> None:
                 c.id, c.document_id, c.knowledge_base_id,
                 c.content, c.chunk_index,
                 c.start_char, c.end_char, c.metadata_json,
-                c.is_embedded, c.created_at or now,
+                c.is_embedded, c.index_version, c.created_at or now,
             )
             for c in chunks
         ]
@@ -883,8 +885,8 @@ def create_kb_chunks_batch(self, chunks: list[KBChunk]) -> None:
                (id, document_id, knowledge_base_id,
                 content, chunk_index,
                 start_char, end_char, metadata_json,
-                is_embedded, created_at)
-               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                 is_embedded, index_version, created_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
             rows,
         )
         self._maybe_commit()
@@ -905,6 +907,93 @@ def get_kb_chunks_by_kb(self, kb_id: str) -> list[KBChunk]:
         ).fetchall()
         return [row_to_kb_chunk(r) for r in rows]
 
+    def get_kb_chunks_by_index(
+        self, kb_id: str, index_version: str,
+    ) -> list[KBChunk]:
+        conn = self._get_conn()
+        rows = conn.execute(
+            """SELECT * FROM kb_chunks
+               WHERE knowledge_base_id = ? AND index_version = ?
+               ORDER BY document_id ASC, chunk_index ASC""",
+            (kb_id, index_version),
+        ).fetchall()
+        return [row_to_kb_chunk(r) for r in rows]
+
+    def activate_kb_index(
+        self,
+        kb_id: str,
+        index_version: str,
+        chunks: list[KBChunk],
+        document_updates: dict[str, dict[str, Any]],
+        dimensions: int,
+    ) -> None:
+        """Publish a complete staged index with its matching DB metadata.
+
+        The vector store is built before this method runs.  Keeping all DB
+        mutations in one SQLite transaction guarantees readers observe
+        either the complete old index or the complete new index, never an
+        empty/mixed set of chunk rows.
+        """
+        conn = self._get_conn()
+        now = _now()
+        try:
+            conn.execute("BEGIN")
+            if chunks:
+                conn.executemany(
+                    """INSERT INTO kb_chunks
+                       (id, document_id, knowledge_base_id,
+                        content, chunk_index,
+                        start_char, end_char, metadata_json,
+                        is_embedded, index_version, created_at)
+                       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                    [
+                        (
+                            c.id, c.document_id, c.knowledge_base_id,
+                            c.content, c.chunk_index,
+                            c.start_char, c.end_char, c.metadata_json,
+                            c.is_embedded, c.index_version, c.created_at or now,
+                        )
+                        for c in chunks
+                    ],
+                )
+            for doc_id, values in document_updates.items():
+                conn.execute(
+                    """UPDATE kb_documents
+                       SET content_text = ?, content_length = ?, chunk_count = ?,
+                           status = ?, error_message = ?, updated_at = ?
+                       WHERE id = ?""",
+                    (
+                        str(values.get("content_text", "")),
+                        int(values.get("content_length", 0)),
+                        int(values.get("chunk_count", 0)),
+                        str(values.get("status", "ready")),
+                        str(values.get("error_message", "")),
+                        now,
+                        doc_id,
+                    ),
+                )
+            document_count = len(document_updates)
+            conn.execute(
+                """UPDATE knowledge_bases
+                   SET active_index_version = ?, embedding_dimensions = ?,
+                       document_count = ?, chunk_count = ?, status = 'active',
+                       updated_at = ?
+                   WHERE id = ?""",
+                (index_version, dimensions, document_count, len(chunks), now, kb_id),
+            )
+            conn.commit()
+        except Exception:
+            conn.rollback()
+            raise
+
+    def delete_kb_chunks_by_index(self, kb_id: str, index_version: str) -> None:
+        conn = self._get_conn()
+        conn.execute(
+            "DELETE FROM kb_chunks WHERE knowledge_base_id = ? AND index_version = ?",
+            (kb_id, index_version),
+        )
+        self._maybe_commit()
+
     def delete_kb_chunks_by_document(self, doc_id: str) -> None:
         conn = self._get_conn()
         conn.execute(
@@ -923,6 +1012,80 @@ def update_kb_chunk_embedded(self, chunk_ids: list[str]) -> None:
         )
         self._maybe_commit()
 
+    # ----- KB background jobs and durable cleanup -----
+
+    def create_kb_job(self, kb_id: str, document_id: str, operation: str) -> str:
+        job_id = _generate_id()
+        conn = self._get_conn()
+        now = _now()
+        conn.execute(
+            """INSERT INTO kb_jobs
+               (id, knowledge_base_id, document_id, operation, status, created_at, updated_at)
+               VALUES (?, ?, ?, ?, 'queued', ?, ?)""",
+            (job_id, kb_id, document_id, operation, now, now),
+        )
+        self._maybe_commit()
+        return job_id
+
+    def update_kb_job(self, job_id: str, status: str, error_message: str = "") -> None:
+        conn = self._get_conn()
+        conn.execute(
+            """UPDATE kb_jobs
+               SET status = ?, error_message = ?, updated_at = ?
+               WHERE id = ?""",
+            (status, error_message, _now(), job_id),
+        )
+        self._maybe_commit()
+
+    def create_kb_cleanup_job(
+        self, kb_id: str, index_version: str, operation: str, error_message: str,
+    ) -> str:
+        job_id = _generate_id()
+        conn = self._get_conn()
+        now = _now()
+        conn.execute(
+            """INSERT INTO kb_cleanup_jobs
+               (id, knowledge_base_id, index_version, operation, status,
+                attempts, error_message, created_at, updated_at)
+               VALUES (?, ?, ?, ?, 'pending', 0, ?, ?, ?)""",
+            (job_id, kb_id, index_version, operation, error_message, now, now),
+        )
+        self._maybe_commit()
+        return job_id
+
+    def get_pending_kb_cleanup_jobs(self) -> list[KBCleanupJob]:
+        conn = self._get_conn()
+        rows = conn.execute(
+            """SELECT * FROM kb_cleanup_jobs
+               WHERE status = 'pending' ORDER BY created_at ASC"""
+        ).fetchall()
+        return [
+            KBCleanupJob(
+                id=row["id"],
+                knowledge_base_id=row["knowledge_base_id"],
+                index_version=row["index_version"],
+                operation=row["operation"],
+                status=row["status"],
+                attempts=row["attempts"],
+                error_message=row["error_message"],
+                created_at=row["created_at"],
+                updated_at=row["updated_at"],
+            )
+            for row in rows
+        ]
+
+    def update_kb_cleanup_job(
+        self, job_id: str, status: str, error_message: str = "",
+    ) -> None:
+        conn = self._get_conn()
+        conn.execute(
+            """UPDATE kb_cleanup_jobs
+               SET status = ?, attempts = attempts + 1, error_message = ?, updated_at = ?
+               WHERE id = ?""",
+            (status, error_message, _now(), job_id),
+        )
+        self._maybe_commit()
+
     # ----- Dashboard aggregation -----
 
     def get_session_counts(self) -> dict[str, int]:
diff --git a/misaka/main.py b/misaka/main.py
index 7c9a58b..d2b1653 100644
--- a/misaka/main.py
+++ b/misaka/main.py
@@ -179,11 +179,17 @@ def __init__(self, db: DatabaseBackend) -> None:
 
         # Knowledge Base / RAG services
         from misaka.services.knowledge.document_service import DocumentService
+        from misaka.services.knowledge.job_coordinator import KnowledgeBaseJobCoordinator
         from misaka.services.knowledge.kb_service import KnowledgeBaseService
 
         self.rag_orchestrator = self._create_rag_orchestrator()
-        self.kb_service = KnowledgeBaseService(db, self.rag_orchestrator)
-        self.document_service = DocumentService(db, self.rag_orchestrator)
+        self.kb_job_coordinator = KnowledgeBaseJobCoordinator(db)
+        self.kb_service = KnowledgeBaseService(
+            db, self.rag_orchestrator, self.kb_job_coordinator,
+        )
+        self.document_service = DocumentService(
+            db, self.rag_orchestrator, self.kb_job_coordinator,
+        )
 
     def _create_rag_orchestrator(self):
         """Build the RAG pipeline for the currently persisted vector backend."""
@@ -228,8 +234,11 @@ def rebuild_rag(self) -> None:
         self.rag_orchestrator = self._create_rag_orchestrator()
         if hasattr(self, "kb_service"):
             self.kb_service._orchestrator = self.rag_orchestrator
+            if self.kb_service._index_manager is not None:
+                self.kb_service._index_manager._orchestrator = self.rag_orchestrator
         if hasattr(self, "document_service"):
             self.document_service._orchestrator = self.rag_orchestrator
+            self.document_service._index_manager._orchestrator = self.rag_orchestrator
 
     def configure_vector_backend(
         self,
@@ -445,6 +454,17 @@ def _asyncio_exception_handler(loop: asyncio.AbstractEventLoop, ctx: dict) -> No
 
     page.run_task(_run_env_check)
 
+    # --- Retry durable vector cleanup from interrupted/failed operations ---
+    async def _retry_kb_cleanup() -> None:
+        try:
+            completed = await services.kb_service.retry_pending_cleanup()
+            if completed:
+                logger.info("Completed %d deferred knowledge-base vector cleanups", completed)
+        except Exception:
+            logger.exception("Deferred knowledge-base vector cleanup retry failed")
+
+    page.run_task(_retry_kb_cleanup)
+
     # --- Check for Claude Code updates ---
     async def _run_update_check() -> None:
         result = await services.update_check_service.check_for_update()
@@ -478,6 +498,12 @@ def main() -> None:
     """Application entry point."""
     multiprocessing.freeze_support()
 
+    if "--rag-smoke" in sys.argv:
+        from misaka.services.knowledge.frozen_smoke import run_frozen_rag_smoke
+
+        run_frozen_rag_smoke()
+        return
+
     if _maybe_delegate_hot_reload():
         return
 
diff --git a/misaka/services/knowledge/document_service.py b/misaka/services/knowledge/document_service.py
index ca603ea..3e7ae7f 100644
--- a/misaka/services/knowledge/document_service.py
+++ b/misaka/services/knowledge/document_service.py
@@ -12,7 +12,9 @@
 from typing import TYPE_CHECKING, Any
 
 from misaka.config import get_kb_storage_dir
-from misaka.db.models import KBChunk, KBDocument
+from misaka.db.models import KBDocument
+from misaka.services.knowledge.index_manager import KBIndexManager
+from misaka.services.knowledge.job_coordinator import KnowledgeBaseJobCoordinator
 from misaka.services.knowledge.rag.abstractions import EmbeddingConfig
 
 if TYPE_CHECKING:
@@ -42,9 +44,12 @@ def __init__(
         self,
         db: DatabaseBackend,
         orchestrator: RAGOrchestrator,
+        coordinator: KnowledgeBaseJobCoordinator | None = None,
     ) -> None:
         self._db = db
         self._orchestrator = orchestrator
+        self._index_manager = KBIndexManager(db, orchestrator)
+        self._coordinator = coordinator or KnowledgeBaseJobCoordinator(db)
 
     # ── Upload ────────────────────────────────────────────────────────
 
@@ -98,39 +103,15 @@ async def upload_document(
         self._db.create_kb_document(doc)
 
         try:
-            result = await self._orchestrator.ingest_document(
-                file_path=str(dest),
-                file_type=file_type,
-                kb=kb,
-                embedding_config=embedding_config,
-                on_progress=on_progress,
-                document_id=doc_id,
-            )
-
-            if result.error:
-                self._db.update_kb_document(doc_id, status="error", error_message=result.error)
-                doc.status = "error"
-                doc.error_message = result.error
-                return doc
-
-            self._persist_chunks(doc_id, kb_id, result.chunks)
-            self._db.update_kb_document(
-                doc_id,
-                content_text=result.content_text,
-                content_length=result.content_length,
-                chunk_count=result.chunk_count,
-                status="ready",
-            )
-            if result.dimensions and kb.embedding_dimensions == 0:
-                self._db.update_knowledge_base(
-                    kb_id, embedding_dimensions=result.dimensions,
+            async with self._coordinator.job(kb_id, "upload", doc_id):
+                self._db.update_kb_document(doc_id, status="parsing", error_message="")
+                current_docs = self._documents_for_next_index(kb_id, {doc_id})
+                await self._index_manager.build_and_activate(
+                    kb, current_docs, embedding_config, on_progress,
                 )
-
-            doc.status = "ready"
-            doc.content_text = result.content_text
-            doc.content_length = result.content_length
-            doc.chunk_count = result.chunk_count
-
+                updated = self._db.get_kb_document(doc_id)
+                if updated is not None:
+                    return updated
         except asyncio.CancelledError:
             logger.warning("Upload cancelled for document %s", src.name)
             self._db.update_kb_document(doc_id, status="error", error_message="Upload cancelled")
@@ -142,7 +123,7 @@ async def upload_document(
             doc.status = "error"
             doc.error_message = str(exc)
 
-        return doc
+        return self._db.get_kb_document(doc_id) or doc
 
     async def upload_documents_batch(
         self,
@@ -201,27 +182,32 @@ def get_document_content(self, doc_id: str) -> str:
 
     # ── Delete ────────────────────────────────────────────────────────
 
-    def delete_document(self, doc_id: str) -> bool:
+    async def delete_document(
+        self, doc_id: str, embedding_config: EmbeddingConfig,
+    ) -> bool:
+        """Cancel in-flight work, publish an index without this document, then delete it."""
         doc = self._db.get_kb_document(doc_id)
         if not doc:
             return False
 
-        chunks = self._db.get_kb_chunks_by_document(doc_id)
-        chunk_ids = [c.id for c in chunks]
-        if chunk_ids:
-            try:
-                self._orchestrator.delete_chunks_from_vector_store(
-                    doc.knowledge_base_id, chunk_ids,
-                )
-            except Exception:
-                logger.exception("Failed to remove vectors for doc %s", doc_id)
-
-        self._db.delete_kb_document(doc_id)
+        await self._coordinator.cancel_and_wait(doc.knowledge_base_id)
+        async with self._coordinator.job(doc.knowledge_base_id, "delete_document", doc_id):
+            kb = self._db.get_knowledge_base(doc.knowledge_base_id)
+            if kb is None:
+                return False
+            remaining_docs = [
+                item for item in self._documents_for_next_index(doc.knowledge_base_id, set())
+                if item.id != doc_id
+            ]
+            await self._index_manager.build_and_activate(
+                kb, remaining_docs, embedding_config,
+            )
+            self._db.delete_kb_document(doc_id)
 
-        if doc.storage_path:
-            p = Path(doc.storage_path)
-            if p.exists():
-                p.unlink(missing_ok=True)
+            if doc.storage_path:
+                path = Path(doc.storage_path)
+                if path.exists():
+                    path.unlink(missing_ok=True)
 
         logger.info("Deleted document '%s' (id=%s)", doc.file_name, doc_id)
         return True
@@ -243,42 +229,29 @@ async def reprocess_document(
         if not kb:
             return None
 
-        old_chunks = self._db.get_kb_chunks_by_document(doc_id)
-        old_ids = [c.id for c in old_chunks]
-        if old_ids:
-            try:
-                self._orchestrator.delete_chunks_from_vector_store(
-                    doc.knowledge_base_id, old_ids,
+        try:
+            async with self._coordinator.job(doc.knowledge_base_id, "reprocess", doc_id):
+                # This only signals UI activity; the active chunk rows and
+                # document statistics remain untouched until activation.
+                self._db.update_kb_document(doc_id, status="embedding", error_message="")
+                current_docs = self._documents_for_next_index(
+                    doc.knowledge_base_id, {doc_id},
                 )
-            except Exception:
-                logger.exception("Failed to remove old vectors for doc %s", doc_id)
-        self._db.delete_kb_chunks_by_document(doc_id)
-
-        self._db.update_kb_document(doc_id, status="parsing", error_message="")
-
-        result = await self._orchestrator.ingest_document(
-            file_path=doc.storage_path,
-            file_type=doc.file_type,
-            kb=kb,
-            embedding_config=embedding_config,
-            on_progress=on_progress,
-            document_id=doc.id,
-        )
-
-        if result.error:
+                await self._index_manager.build_and_activate(
+                    kb, current_docs, embedding_config, on_progress,
+                )
+        except asyncio.CancelledError:
             self._db.update_kb_document(
-                doc_id, status="error", error_message=result.error,
+                doc_id, status=doc.status, error_message=doc.error_message,
             )
-        else:
-            self._persist_chunks(doc_id, doc.knowledge_base_id, result.chunks)
+            raise
+        except Exception:
+            # Reprocessing an existing document must not make its known-good
+            # active version appear failed or unavailable.
             self._db.update_kb_document(
-                doc_id,
-                content_text=result.content_text,
-                content_length=result.content_length,
-                chunk_count=result.chunk_count,
-                status="ready",
+                doc_id, status=doc.status, error_message=doc.error_message,
             )
-
+            raise
         return self._db.get_kb_document(doc_id)
 
     # ── Dedup helper ──────────────────────────────────────────────────
@@ -290,35 +263,16 @@ def check_duplicate(self, kb_id: str, file_path: str) -> KBDocument | None:
 
     # ── Private helpers ───────────────────────────────────────────────
 
-    def _persist_chunks(
-        self,
-        doc_id: str,
-        kb_id: str,
-        chunks: list,
-    ) -> None:
-        """Write ChunkData objects to the kb_chunks table.
-
-        The chunk ID is derived from the same logic used by the vector
-        store, so that ``delete_document`` can reliably remove vectors by
-        ID later.
-        """
-        db_chunks: list[KBChunk] = []
-        for c in chunks:
-            cid = str(c.metadata.get("chunk_db_id", f"chunk_{c.index}"))
-            db_chunks.append(KBChunk(
-                id=cid,
-                document_id=doc_id,
-                knowledge_base_id=kb_id,
-                content=c.content,
-                chunk_index=c.index,
-                start_char=c.start_char,
-                end_char=c.end_char,
-                metadata_json=self._safe_json(c.metadata),
-                is_embedded=1,
-            ))
-        if db_chunks:
-            self._db.create_kb_chunks_batch(db_chunks)
-            self._db.update_kb_chunk_embedded([c.id for c in db_chunks])
+    def _documents_for_next_index(
+        self, kb_id: str, include_ids: set[str],
+    ) -> list[KBDocument]:
+        """Return documents represented by the next complete index snapshot."""
+        return [
+            doc
+            for doc in self._db.get_kb_documents_by_kb(kb_id)
+            if doc.status == "ready" or doc.id in include_ids
+        ]
+
 
     @staticmethod
     def _resolve_file_type(path: Path) -> str:
@@ -342,14 +296,5 @@ def _compute_hash(path: Path) -> str:
                 sha.update(data)
         return sha.hexdigest()
 
-    @staticmethod
-    def _safe_json(data: dict) -> str:
-        import json
-        try:
-            return json.dumps(data, ensure_ascii=False)
-        except (TypeError, ValueError):
-            return "{}"
-
-
 class DuplicateDocumentError(Exception):
     """Raised when attempting to upload a document that already exists."""
diff --git a/misaka/services/knowledge/frozen_smoke.py b/misaka/services/knowledge/frozen_smoke.py
new file mode 100644
index 0000000..14b2fbb
--- /dev/null
+++ b/misaka/services/knowledge/frozen_smoke.py
@@ -0,0 +1,47 @@
+"""Self-contained RAG smoke test for a PyInstaller distribution."""
+
+from __future__ import annotations
+
+import asyncio
+import tempfile
+from pathlib import Path
+
+from misaka.services.knowledge.rag.abstractions import ChunkData
+from misaka.services.knowledge.rag.langchain.retriever import LCHybridRetriever
+from misaka.services.knowledge.rag.langchain.vector_store import LCSqliteVecStore
+
+
+def run_frozen_rag_smoke() -> None:
+    """Exercise sqlite-vec load/write/search and NumPy-backed BM25 fusion."""
+    with tempfile.TemporaryDirectory(prefix="misaka-rag-smoke-") as temp_dir:
+        db_path = str(Path(temp_dir) / "vectors.sqlite3")
+        store = LCSqliteVecStore(db_path)
+        table_name = "smoke_vectors"
+        chunks = [
+            ChunkData(
+                content="Misaka knowledge base vector retrieval",
+                index=0,
+                metadata={"chunk_db_id": "chunk-a", "document_id": "smoke"},
+            ),
+            ChunkData(
+                content="Unrelated calendar appointment notes",
+                index=1,
+                metadata={"chunk_db_id": "chunk-b", "document_id": "smoke"},
+            ),
+        ]
+        try:
+            store.ensure_table(table_name, 2)
+            store.add_chunks(table_name, chunks, [[1.0, 0.0], [0.0, 1.0]])
+            results = asyncio.run(
+                LCHybridRetriever(store).retrieve(
+                    "knowledge retrieval",
+                    [1.0, 0.0],
+                    table_name,
+                    top_k=1,
+                    chunks_for_bm25=chunks,
+                )
+            )
+            if not results or results[0].content != chunks[0].content:
+                raise RuntimeError("Frozen RAG smoke test returned no expected result")
+        finally:
+            store.close()
diff --git a/misaka/services/knowledge/index_manager.py b/misaka/services/knowledge/index_manager.py
new file mode 100644
index 0000000..220aef2
--- /dev/null
+++ b/misaka/services/knowledge/index_manager.py
@@ -0,0 +1,197 @@
+"""Copy-on-write construction and cleanup of knowledge-base vector indexes."""
+
+from __future__ import annotations
+
+import json
+import uuid
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from misaka.db.models import KBChunk
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from misaka.db.database import DatabaseBackend
+    from misaka.db.models import KBDocument, KnowledgeBase
+    from misaka.services.knowledge.rag.abstractions import EmbeddingConfig
+    from misaka.services.knowledge.rag_orchestrator import RAGOrchestrator
+
+
+@dataclass
+class IndexBuildResult:
+    """The outcome of a successfully activated index build."""
+
+    index_version: str
+    document_count: int
+    chunk_count: int
+    dimensions: int
+
+
+class KBIndexManager:
+    """Build complete immutable index versions, then atomically activate them."""
+
+    def __init__(self, db: DatabaseBackend, orchestrator: RAGOrchestrator) -> None:
+        self._db = db
+        self._orchestrator = orchestrator
+
+    async def build_and_activate(
+        self,
+        kb: KnowledgeBase,
+        documents: list[KBDocument],
+        embedding_config: EmbeddingConfig,
+        on_progress: Callable[[str], None] | None = None,
+    ) -> IndexBuildResult:
+        """Build a complete replacement index without touching the active one.
+
+        The vector store receives an isolated versioned table/collection.  DB
+        chunk rows and document statistics are published together only after
+        every source document has been parsed, embedded and written.  Any
+        error removes the staging version and leaves the active version intact.
+        """
+        self._validate_sources(documents)
+        new_version = uuid.uuid4().hex
+        staged_chunks: list[KBChunk] = []
+        document_updates: dict[str, dict[str, Any]] = {}
+        dimensions = 0
+
+        try:
+            for document in documents:
+                result = await self._orchestrator.ingest_document(
+                    file_path=document.storage_path,
+                    file_type=document.file_type,
+                    kb=kb,
+                    embedding_config=embedding_config,
+                    on_progress=on_progress,
+                    document_id=document.id,
+                    index_version=new_version,
+                )
+                if result.error:
+                    raise RuntimeError(f"{document.file_name}: {result.error}")
+                if not result.chunks:
+                    raise RuntimeError(f"{document.file_name}: no chunks were produced")
+                if dimensions and result.dimensions != dimensions:
+                    raise RuntimeError(
+                        f"{document.file_name}: embedding dimensions changed within one index build"
+                    )
+                dimensions = result.dimensions or dimensions
+                staged_chunks.extend(
+                    self._to_db_chunks(document.id, kb.id, new_version, result.chunks)
+                )
+                document_updates[document.id] = {
+                    "content_text": result.content_text,
+                    "content_length": result.content_length,
+                    "chunk_count": result.chunk_count,
+                    "status": "ready",
+                    "error_message": "",
+                }
+
+            if documents and not dimensions:
+                raise RuntimeError("Embedding provider returned no vector dimensions")
+
+            old_version = kb.active_index_version
+            self._db.activate_kb_index(
+                kb.id,
+                new_version,
+                staged_chunks,
+                document_updates,
+                dimensions or kb.embedding_dimensions,
+            )
+        except BaseException as exc:
+            await self._discard_staging(kb.id, new_version, str(exc))
+            raise
+
+        await self._retire_index(kb.id, old_version, operation="retire_previous_index")
+        return IndexBuildResult(
+            index_version=new_version,
+            document_count=len(document_updates),
+            chunk_count=len(staged_chunks),
+            dimensions=dimensions or kb.embedding_dimensions,
+        )
+
+    async def retry_pending_cleanup(self) -> int:
+        """Retry every persisted vector-table deletion and return successes."""
+        succeeded = 0
+        for job in self._db.get_pending_kb_cleanup_jobs():
+            try:
+                self._orchestrator.drop_kb_vectors(
+                    job.knowledge_base_id, job.index_version,
+                )
+                self._db.delete_kb_chunks_by_index(
+                    job.knowledge_base_id, job.index_version,
+                )
+                self._db.update_kb_cleanup_job(job.id, "completed")
+                succeeded += 1
+            except Exception as exc:
+                self._db.update_kb_cleanup_job(job.id, "pending", str(exc))
+        return succeeded
+
+    async def delete_index_or_enqueue(
+        self, kb_id: str, index_version: str, operation: str,
+    ) -> bool:
+        """Delete one index version, persisting retry work on failure."""
+        try:
+            self._orchestrator.drop_kb_vectors(kb_id, index_version)
+        except Exception as exc:
+            self._db.create_kb_cleanup_job(kb_id, index_version, operation, str(exc))
+            return False
+        self._db.delete_kb_chunks_by_index(kb_id, index_version)
+        return True
+
+    async def _retire_index(
+        self, kb_id: str, index_version: str, operation: str,
+    ) -> None:
+        # There is nothing to retire for a brand-new KB.  Retiring only when
+        # DB rows exist avoids treating a missing legacy table as an error.
+        if not self._db.get_kb_chunks_by_index(kb_id, index_version):
+            return
+        await self.delete_index_or_enqueue(kb_id, index_version, operation)
+
+    async def _discard_staging(
+        self, kb_id: str, index_version: str, reason: str,
+    ) -> None:
+        try:
+            self._orchestrator.drop_kb_vectors(kb_id, index_version)
+        except Exception as cleanup_error:
+            self._db.create_kb_cleanup_job(
+                kb_id,
+                index_version,
+                "discard_staged_index",
+                f"build failed: {reason}; cleanup failed: {cleanup_error}",
+            )
+
+    @staticmethod
+    def _validate_sources(documents: list[KBDocument]) -> None:
+        for document in documents:
+            if not document.storage_path or not Path(document.storage_path).is_file():
+                raise FileNotFoundError(
+                    f"Source file for '{document.file_name}' is missing; active index was preserved"
+                )
+
+    @staticmethod
+    def _to_db_chunks(
+        document_id: str,
+        kb_id: str,
+        index_version: str,
+        chunks: list,
+    ) -> list[KBChunk]:
+        db_chunks: list[KBChunk] = []
+        for chunk in chunks:
+            metadata = "{}"
+            with suppress(TypeError, ValueError):
+                metadata = json.dumps(chunk.metadata, ensure_ascii=False)
+            db_chunks.append(KBChunk(
+                id=str(chunk.metadata["chunk_db_id"]),
+                document_id=document_id,
+                knowledge_base_id=kb_id,
+                content=chunk.content,
+                chunk_index=chunk.index,
+                start_char=chunk.start_char,
+                end_char=chunk.end_char,
+                metadata_json=metadata,
+                is_embedded=1,
+                index_version=index_version,
+            ))
+        return db_chunks
diff --git a/misaka/services/knowledge/job_coordinator.py b/misaka/services/knowledge/job_coordinator.py
new file mode 100644
index 0000000..5c9aeae
--- /dev/null
+++ b/misaka/services/knowledge/job_coordinator.py
@@ -0,0 +1,72 @@
+"""Mutual exclusion and cancellation for knowledge-base operations."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager, suppress
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from misaka.db.database import DatabaseBackend
+
+
+class KnowledgeBaseJobCoordinator:
+    """Serialise mutating operations per KB and persist their lifecycle.
+
+    A single KB index is the consistency boundary: permitting concurrent
+    upload/reprocess/delete operations would let one operation publish a
+    stale snapshot over another.  The coordinator therefore uses a KB-wide
+    lock and lets destructive operations cancel and await the active job.
+    """
+
+    def __init__(self, db: DatabaseBackend) -> None:
+        self._db = db
+        self._locks: dict[str, asyncio.Lock] = {}
+        self._active: dict[str, tuple[asyncio.Task[object], str]] = {}
+
+    @asynccontextmanager
+    async def job(
+        self, kb_id: str, operation: str, document_id: str = "",
+    ) -> AsyncIterator[str]:
+        """Run one KB mutation, recording queued/running/final states."""
+        job_id = self._db.create_kb_job(kb_id, document_id, operation)
+        lock = self._locks.setdefault(kb_id, asyncio.Lock())
+        async with lock:
+            task = asyncio.current_task()
+            if task is None:  # pragma: no cover - asyncio always supplies one
+                raise RuntimeError("Knowledge-base operations require an asyncio task")
+            self._active[kb_id] = (task, job_id)
+            self._db.update_kb_job(job_id, "running")
+            try:
+                yield job_id
+            except asyncio.CancelledError:
+                self._db.update_kb_job(job_id, "cancelled", "Operation cancelled")
+                raise
+            except Exception as exc:
+                self._db.update_kb_job(job_id, "failed", str(exc))
+                raise
+            else:
+                self._db.update_kb_job(job_id, "completed")
+            finally:
+                current = self._active.get(kb_id)
+                if current and current[1] == job_id:
+                    self._active.pop(kb_id, None)
+
+    async def cancel_and_wait(self, kb_id: str) -> None:
+        """Cancel the active KB job, if any, and wait for it to finish."""
+        active = self._active.get(kb_id)
+        if not active:
+            return
+        task, job_id = active
+        if task is asyncio.current_task() or task.done():
+            return
+        self._db.update_kb_job(job_id, "cancelling", "Cancelled by a destructive operation")
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+    def is_busy(self, kb_id: str) -> bool:
+        """Return whether an in-process mutating operation owns this KB."""
+        active = self._active.get(kb_id)
+        return bool(active and not active[0].done())
diff --git a/misaka/services/knowledge/kb_service.py b/misaka/services/knowledge/kb_service.py
index 1fd0b6f..ff77863 100644
--- a/misaka/services/knowledge/kb_service.py
+++ b/misaka/services/knowledge/kb_service.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import contextlib
 import json
 import logging
 import shutil
@@ -11,6 +10,8 @@
 
 from misaka.config import SettingKeys, get_kb_storage_dir
 from misaka.db.models import KnowledgeBase
+from misaka.services.knowledge.index_manager import KBIndexManager
+from misaka.services.knowledge.job_coordinator import KnowledgeBaseJobCoordinator
 
 if TYPE_CHECKING:
     from misaka.db.database import DatabaseBackend
@@ -26,9 +27,14 @@ def __init__(
         self,
         db: DatabaseBackend,
         orchestrator: RAGOrchestrator | None = None,
+        coordinator: KnowledgeBaseJobCoordinator | None = None,
     ) -> None:
         self._db = db
         self._orchestrator = orchestrator
+        self._coordinator = coordinator or KnowledgeBaseJobCoordinator(db)
+        self._index_manager = (
+            KBIndexManager(db, orchestrator) if orchestrator is not None else None
+        )
 
     # ── Queries ───────────────────────────────────────────────────────
 
@@ -74,22 +80,28 @@ def update(self, kb_id: str, **kwargs: Any) -> KnowledgeBase | None:
         self._db.update_knowledge_base(kb_id, **kwargs)
         return self._db.get_knowledge_base(kb_id)
 
-    def delete(self, kb_id: str) -> bool:
+    async def delete(self, kb_id: str) -> bool:
+        """Delete a KB after cancelling work; retain failed vector cleanup durably."""
         kb = self._db.get_knowledge_base(kb_id)
         if not kb:
             return False
 
-        if self._orchestrator:
-            try:
-                self._orchestrator.drop_kb_vectors(kb_id)
-            except Exception:
-                logger.exception("Failed to drop vectors for kb %s", kb_id)
+        await self._coordinator.cancel_and_wait(kb_id)
+        async with self._coordinator.job(kb_id, "delete_knowledge_base"):
+            if self._index_manager:
+                versions = {chunk.index_version for chunk in self._db.get_kb_chunks_by_kb(kb_id)}
+                if kb.active_index_version:
+                    versions.add(kb.active_index_version)
+                for version in versions:
+                    await self._index_manager.delete_index_or_enqueue(
+                        kb_id, version, "delete_knowledge_base",
+                    )
 
-        self._db.delete_knowledge_base(kb_id)
+            self._db.delete_knowledge_base(kb_id)
 
-        storage_dir = get_kb_storage_dir(kb_id)
-        if storage_dir.exists():
-            shutil.rmtree(storage_dir, ignore_errors=True)
+            storage_dir = get_kb_storage_dir(kb_id)
+            if storage_dir.exists():
+                shutil.rmtree(storage_dir, ignore_errors=True)
 
         logger.info("Deleted knowledge base '%s' (id=%s)", kb.name, kb_id)
         return True
@@ -98,9 +110,12 @@ def delete(self, kb_id: str) -> bool:
 
     def update_statistics(self, kb_id: str) -> None:
         """Recompute document_count and chunk_count from related rows."""
-        docs = self._db.get_kb_documents_by_kb(kb_id)
-        doc_count = len(docs)
-        chunk_count = sum(d.chunk_count for d in docs)
+        kb = self._db.get_knowledge_base(kb_id)
+        if kb is None:
+            return
+        chunks = self._db.get_kb_chunks_by_index(kb_id, kb.active_index_version)
+        doc_count = len({chunk.document_id for chunk in chunks})
+        chunk_count = len(chunks)
         self._db.update_knowledge_base(
             kb_id,
             document_count=doc_count,
@@ -153,7 +168,7 @@ async def rebuild_embeddings(
         on_progress: Any = None,
     ) -> dict[str, Any]:
         """Re-embed all documents after an embedding model change."""
-        if not self._orchestrator:
+        if not self._orchestrator or not self._index_manager:
             return {"success_count": 0, "error_count": 0, "errors": ["No orchestrator"]}
 
         kb = self._db.get_knowledge_base(kb_id)
@@ -161,36 +176,19 @@ async def rebuild_embeddings(
             return {"success_count": 0, "error_count": 0, "errors": ["KB not found"]}
 
         docs = self._db.get_kb_documents_by_kb(kb_id)
-        success_count = 0
-        error_count = 0
-        errors: list[str] = []
-
-        self._db.update_knowledge_base(kb_id, status="building")
         try:
-            self._orchestrator.drop_kb_vectors(kb_id)
-        except Exception:
-            logger.info("No existing vector collection to drop for kb %s", kb_id)
-
-        for doc in docs:
-            try:
-                await self._rebuild_single_doc(kb, doc, embedding_config, on_progress)
-                success_count += 1
-            except Exception as exc:
-                logger.exception("Rebuild failed for doc %s", doc.id)
-                error_count += 1
-                errors.append(f"{doc.file_name}: {exc}")
-
-        new_status = "active" if error_count == 0 else "error"
-        self._db.update_knowledge_base(kb_id, status=new_status)
-        self.update_statistics(kb_id)
-        if error_count == 0:
-            self.mark_index_rebuilt(kb_id)
+            async with self._coordinator.job(kb_id, "rebuild"):
+                await self._index_manager.build_and_activate(
+                    kb, docs, embedding_config, on_progress,
+                )
+        except Exception as exc:
+            logger.exception("Rebuild failed for KB %s", kb_id)
+            # Keep the old active version and its reconciled statistics intact.
+            self.update_statistics(kb_id)
+            return {"success_count": 0, "error_count": 1, "errors": [str(exc)]}
 
-        return {
-            "success_count": success_count,
-            "error_count": error_count,
-            "errors": errors,
-        }
+        self.mark_index_rebuilt(kb_id)
+        return {"success_count": len(docs), "error_count": 0, "errors": []}
 
     # ----- Vector backend rebuild state -----
 
@@ -233,85 +231,11 @@ def _save_pending_kb_ids(self, kb_ids: list[str]) -> None:
             json.dumps(kb_ids),
         )
 
-    async def _rebuild_single_doc(
-        self,
-        kb: KnowledgeBase,
-        doc: Any,
-        embedding_config: Any,
-        on_progress: Any,
-    ) -> None:
-        """Delete old chunks/vectors and re-ingest a single document."""
-        self._remove_old_chunks(doc)
-
-        if not doc.storage_path:
-            return
-
-        result = await self._orchestrator.ingest_document(
-            file_path=doc.storage_path,
-            file_type=doc.file_type,
-            kb=kb,
-            embedding_config=embedding_config,
-            on_progress=on_progress,
-            document_id=doc.id,
-        )
-        if result.error:
-            self._db.update_kb_document(doc.id, status="error", error_message=result.error)
-            raise RuntimeError(result.error)
-
-        self._persist_rebuilt_chunks(doc.id, doc.knowledge_base_id, result.chunks)
-        self._db.update_kb_document(
-            doc.id,
-            content_text=result.content_text,
-            content_length=result.content_length,
-            chunk_count=result.chunk_count,
-            status="ready",
-        )
-        if result.dimensions and kb.embedding_dimensions != result.dimensions:
-            self._db.update_knowledge_base(
-                kb.id, embedding_dimensions=result.dimensions,
-            )
-
-    def _remove_old_chunks(self, doc: Any) -> None:
-        """Remove existing chunk rows and their vectors."""
-        old_chunks = self._db.get_kb_chunks_by_document(doc.id)
-        old_ids = [c.id for c in old_chunks]
-        if old_ids:
-            try:
-                self._orchestrator.delete_chunks_from_vector_store(
-                    doc.knowledge_base_id, old_ids,
-                )
-            except Exception:
-                logger.warning("Failed to remove old vectors for doc %s", doc.id)
-        self._db.delete_kb_chunks_by_document(doc.id)
-
-    def _persist_rebuilt_chunks(
-        self, doc_id: str, kb_id: str, chunks: list,
-    ) -> None:
-        """Write ChunkData objects to the kb_chunks table."""
-        import json
-
-        from misaka.db.models import KBChunk
-
-        db_chunks: list[KBChunk] = []
-        for c in chunks:
-            cid = str(c.metadata.get("chunk_db_id", f"chunk_{c.index}"))
-            meta = "{}"
-            with contextlib.suppress(TypeError, ValueError):
-                meta = json.dumps(c.metadata, ensure_ascii=False)
-            db_chunks.append(KBChunk(
-                id=cid,
-                document_id=doc_id,
-                knowledge_base_id=kb_id,
-                content=c.content,
-                chunk_index=c.index,
-                start_char=c.start_char,
-                end_char=c.end_char,
-                metadata_json=meta,
-                is_embedded=1,
-            ))
-        if db_chunks:
-            self._db.create_kb_chunks_batch(db_chunks)
-            self._db.update_kb_chunk_embedded([c.id for c in db_chunks])
+    async def retry_pending_cleanup(self) -> int:
+        """Retry failed remote/local vector cleanup recorded in the outbox."""
+        if not self._index_manager:
+            return 0
+        return await self._index_manager.retry_pending_cleanup()
 
     # ── Chat selection ────────────────────────────────────────────────
 
@@ -328,13 +252,14 @@ def get_kb_for_chat_selection(self) -> list[dict[str, Any]]:
                 continue
             if self.is_index_stale(kb.id):
                 continue
-            if kb.chunk_count <= 0:
+            chunks = self._db.get_kb_chunks_by_index(kb.id, kb.active_index_version)
+            if not chunks:
                 continue
             result.append({
                 "id": kb.id,
                 "name": kb.name,
                 "description": kb.description,
-                "document_count": kb.document_count,
-                "chunk_count": kb.chunk_count,
+                "document_count": len({chunk.document_id for chunk in chunks}),
+                "chunk_count": len(chunks),
             })
         return result
diff --git a/misaka/services/knowledge/rag/seekdb/vector_store.py b/misaka/services/knowledge/rag/seekdb/vector_store.py
index a1d5a17..0be1ed0 100644
--- a/misaka/services/knowledge/rag/seekdb/vector_store.py
+++ b/misaka/services/knowledge/rag/seekdb/vector_store.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import contextlib
 from pathlib import Path
 from typing import Any
 
@@ -81,8 +80,10 @@ def delete_by_ids(self, table_name: str, chunk_ids: list[str]) -> None:
 
     def drop_table(self, table_name: str) -> None:
         client = self._get_client()
-        with contextlib.suppress(Exception):
-            client.delete_collection(table_name)
+        # Callers use failures to enqueue durable cleanup work.  Suppressing
+        # remote deletion errors here previously made orphaned vectors look
+        # successfully deleted.
+        client.delete_collection(table_name)
 
     def close(self) -> None:
         client = self._client
diff --git a/misaka/services/knowledge/rag_orchestrator.py b/misaka/services/knowledge/rag_orchestrator.py
index d1917a7..a413938 100644
--- a/misaka/services/knowledge/rag_orchestrator.py
+++ b/misaka/services/knowledge/rag_orchestrator.py
@@ -73,6 +73,7 @@ async def ingest_document(
         embedding_config: EmbeddingConfig,
         on_progress: Callable[[str], None] | None = None,
         document_id: str = "",
+        index_version: str = "",
     ) -> IngestResult:
         """Full ingestion pipeline: parse → chunk → embed → store.
 
@@ -130,7 +131,7 @@ async def ingest_document(
             embeddings = await self._embedding.embed_texts(texts, embedding_config)
 
             dimensions = self._embedding.get_dimensions(embeddings[0])
-            table_name = self._get_table_name(kb.id)
+            table_name = self._get_table_name(kb.id, index_version)
 
             _notify(on_progress, "storing")
             self._vector_store.ensure_table(table_name, dimensions)
@@ -240,21 +241,22 @@ def format_context(self, results: list[KBSearchResult]) -> str:
 
     # ── Resource management ───────────────────────────────────────────
 
-    def drop_kb_vectors(self, kb_id: str) -> None:
-        """Delete the vector table for a knowledge base."""
+    def drop_kb_vectors(self, kb_id: str, index_version: str = "") -> None:
+        """Delete a specific version of a knowledge-base vector table."""
         self._ensure_components()
-        self._vector_store.drop_table(self._get_table_name(kb_id))
+        self._vector_store.drop_table(self._get_table_name(kb_id, index_version))
 
     def delete_chunks_from_vector_store(
         self,
         kb_id: str,
         chunk_ids: list[str],
+        index_version: str = "",
     ) -> None:
         """Remove specific chunk vectors from the store."""
         if chunk_ids:
             self._ensure_components()
             self._vector_store.delete_by_ids(
-                self._get_table_name(kb_id), chunk_ids,
+                self._get_table_name(kb_id, index_version), chunk_ids,
             )
 
     def close(self) -> None:
@@ -294,10 +296,14 @@ async def _retrieve_single_kb(
         top_k: int,
     ) -> list[RetrievalResult]:
         """Retrieve from a single knowledge base and tag results."""
-        table_name = self._get_table_name(kb_id)
+        kb = self._db.get_knowledge_base(kb_id)
+        if kb is None:
+            return []
+        index_version = kb.active_index_version
+        table_name = self._get_table_name(kb_id, index_version)
         chunks_for_bm25 = []
         if getattr(self._retriever, "requires_bm25_chunks", False):
-            for chunk in self._db.get_kb_chunks_by_kb(kb_id):
+            for chunk in self._db.get_kb_chunks_by_index(kb_id, index_version):
                 metadata = {}
                 try:
                     value = json.loads(chunk.metadata_json)
@@ -359,8 +365,15 @@ def _normalize_scores(
         return normalised
 
     @staticmethod
-    def _get_table_name(kb_id: str) -> str:
-        return f"kb_vec_{kb_id.replace('-', '')[:8]}"
+    def _get_table_name(kb_id: str, index_version: str = "") -> str:
+        """Return the stable legacy or immutable-version vector table name."""
+        base = f"kb_vec_{kb_id.replace('-', '')[:8]}"
+        if not index_version:
+            return base
+        suffix = "".join(char for char in index_version if char.isalnum())[:16]
+        if not suffix:
+            raise ValueError("Index version must contain at least one alphanumeric character")
+        return f"{base}_{suffix}"
 
     def _to_search_results(
         self,
diff --git a/misaka/ui/knowledge/components/document_list.py b/misaka/ui/knowledge/components/document_list.py
index 71d8938..1252a27 100644
--- a/misaka/ui/knowledge/components/document_list.py
+++ b/misaka/ui/knowledge/components/document_list.py
@@ -117,6 +117,7 @@ def _build_doc_row(
     status_color = _STATUS_COLORS.get(doc.status, ft.Colors.GREY)
     status_icon = _STATUS_ICONS.get(doc.status, ft.Icons.HELP_OUTLINE)
     type_icon = _TYPE_ICONS.get(doc.file_type, ft.Icons.INSERT_DRIVE_FILE_OUTLINED)
+    is_processing = doc.status in {"pending", "parsing", "embedding"}
 
     actions = ft.Row(
         controls=[
@@ -131,12 +132,14 @@ def _build_doc_row(
                 tooltip=t("kb.doc_reprocess"),
                 on_click=lambda _, did=doc.id: on_reprocess(did) if on_reprocess else None,
                 icon_size=15,
+                disabled=is_processing,
             ),
             make_icon_button(
                 ft.Icons.DELETE_OUTLINE,
                 tooltip=t("kb.doc_delete"),
                 on_click=lambda _, did=doc.id: on_delete(did) if on_delete else None,
                 icon_size=15,
+                disabled=is_processing,
             ),
         ],
         spacing=0,
diff --git a/misaka/ui/knowledge/pages/kb_detail_page.py b/misaka/ui/knowledge/pages/kb_detail_page.py
index 7ef5aac..35ac0ff 100644
--- a/misaka/ui/knowledge/pages/kb_detail_page.py
+++ b/misaka/ui/knowledge/pages/kb_detail_page.py
@@ -351,12 +351,31 @@ def _on_delete_doc(self, doc_id: str) -> None:
         page = self.state.page
 
         def _do_delete(_: ft.ControlEvent) -> None:
-            doc_svc.delete_document(doc_id)
             kb_svc = self.state.get_service("kb_service")
-            if kb_svc:
-                kb_svc.update_statistics(self._kb_id)
-            page.pop_dialog()
-            self.refresh()
+            router_svc = self.state.get_service("router_config_service")
+            kb = kb_svc.get(self._kb_id) if kb_svc else None
+            embed_config = _find_embed_config(
+                router_svc.get_available_embedding_models(), kb,
+            ) if router_svc and kb else None
+            if not embed_config:
+                show_snackbar(
+                    page, t("kb.embedding_config_not_found"), bgcolor=ft.Colors.ERROR,
+                )
+                return
+
+            async def _run_delete() -> None:
+                from misaka.services.knowledge.rag.abstractions import EmbeddingConfig
+
+                try:
+                    await doc_svc.delete_document(doc_id, EmbeddingConfig(**embed_config))
+                except Exception as exc:
+                    logger.exception("Delete failed for doc %s", doc_id)
+                    show_snackbar(page, f"Delete failed: {exc}", bgcolor=ft.Colors.ERROR)
+                else:
+                    page.pop_dialog()
+                    self.refresh()
+
+            page.run_task(_run_delete)
 
         page.show_dialog(
             ft.AlertDialog(
diff --git a/misaka/ui/knowledge/pages/knowledge_page.py b/misaka/ui/knowledge/pages/knowledge_page.py
index eeba1c5..049ccf6 100644
--- a/misaka/ui/knowledge/pages/knowledge_page.py
+++ b/misaka/ui/knowledge/pages/knowledge_page.py
@@ -211,7 +211,7 @@ def _do_delete(_: ft.ControlEvent) -> None:
     async def _async_delete_kb(self, kb_id: str) -> None:
         svc = self.state.get_service("kb_service")
         if svc:
-            svc.delete(kb_id)
+            await svc.delete(kb_id)
         self._refresh_and_update()
 
     def _back_to_list(self) -> None:
diff --git a/pyproject.toml b/pyproject.toml
index f2fcf99..8588d42 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -29,6 +29,7 @@ dependencies = [
     "docx2txt>=0.8",
     "openpyxl>=3.1.0",
     # Hybrid retrieval
+    "numpy>=1.24.0",
     "rank-bm25>=0.2.2",
     # HTTP client for model detection & reranker API
     "httpx>=0.27.0",
diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py
index d130002..1543484 100644
--- a/tests/unit/test_database.py
+++ b/tests/unit/test_database.py
@@ -190,7 +190,7 @@ def test_migration_drops_api_providers_table(self, tmp_path) -> None:
 
         assert row is None
         assert version is not None
-        assert version[0] == 6
+        assert version[0] == 7
 
     def test_migration_creates_seekdb_config_table(self, tmp_path) -> None:
         conn = sqlite3.connect(tmp_path / "seekdb-migrate.db")
diff --git a/tests/unit/test_kb_backend_rebuild_state.py b/tests/unit/test_kb_backend_rebuild_state.py
index 799650d..703a1f8 100644
--- a/tests/unit/test_kb_backend_rebuild_state.py
+++ b/tests/unit/test_kb_backend_rebuild_state.py
@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from misaka.db.models import KBDocument
+from misaka.db.models import KBChunk, KBDocument
 from misaka.services.knowledge.kb_service import KnowledgeBaseService
 
 
@@ -17,6 +17,15 @@ def test_backend_switch_marks_only_kbs_with_documents(db) -> None:
             file_name="notes.txt",
         )
     )
+    db.create_kb_chunks_batch([
+        KBChunk(
+            id="chunk-1",
+            document_id="doc-1",
+            knowledge_base_id=populated.id,
+            content="Indexed note",
+            is_embedded=1,
+        ),
+    ])
     db.update_knowledge_base(populated.id, document_count=1, chunk_count=1)
 
     service.mark_all_indexes_stale()
diff --git a/tests/unit/test_kb_index_safety.py b/tests/unit/test_kb_index_safety.py
new file mode 100644
index 0000000..0a99b6a
--- /dev/null
+++ b/tests/unit/test_kb_index_safety.py
@@ -0,0 +1,262 @@
+"""Regression tests for copy-on-write knowledge-base index operations."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+import pytest
+
+from misaka.services.knowledge.document_service import DocumentService
+from misaka.services.knowledge.kb_service import KnowledgeBaseService
+from misaka.services.knowledge.rag.abstractions import (
+    ChunkData,
+    DocumentParser,
+    EmbeddingConfig,
+    EmbeddingProvider,
+    ParsedDocument,
+    Reranker,
+    RerankerConfig,
+    RetrievalResult,
+    Retriever,
+    TextChunker,
+    VectorStore,
+)
+from misaka.services.knowledge.rag.factory import RAGComponentFactory
+from misaka.services.knowledge.rag_orchestrator import RAGOrchestrator
+
+
+class SafetyParser(DocumentParser):
+    def __init__(self) -> None:
+        self.fail = False
+
+    async def parse(self, file_path: str, file_type: str) -> ParsedDocument:
+        if self.fail:
+            raise RuntimeError("parser unavailable")
+        return ParsedDocument(text=Path(file_path).read_text(encoding="utf-8"))
+
+    def supported_types(self) -> list[str]:
+        return ["txt"]
+
+
+class SafetyChunker(TextChunker):
+    def chunk(
+        self,
+        text: str,
+        file_type: str,
+        chunk_size: int = 512,
+        chunk_overlap: int = 64,
+        metadata: dict | None = None,
+    ) -> list[ChunkData]:
+        return [ChunkData(content=text, index=0, metadata=metadata or {})]
+
+
+class SafetyEmbedding(EmbeddingProvider):
+    async def embed_texts(
+        self,
+        texts: list[str],
+        config: EmbeddingConfig,
+        batch_size: int = 32,
+    ) -> list[list[float]]:
+        return [[1.0, 0.0] for _ in texts]
+
+    async def embed_query(self, query: str, config: EmbeddingConfig) -> list[float]:
+        return [1.0, 0.0]
+
+    def get_dimensions(self, embedding: list[float]) -> int:
+        return len(embedding)
+
+
+class SafetyStore(VectorStore):
+    def __init__(self) -> None:
+        self.tables: dict[str, dict[str, ChunkData]] = {}
+        self.fail_drops = False
+
+    def ensure_table(self, table_name: str, dimensions: int) -> None:
+        self.tables.setdefault(table_name, {})
+
+    def add_chunks(
+        self, table_name: str, chunks: list[ChunkData], embeddings: list[list[float]],
+    ) -> None:
+        self.tables.setdefault(table_name, {}).update({
+            str(chunk.metadata["chunk_db_id"]): chunk for chunk in chunks
+        })
+
+    def search(
+        self, table_name: str, query_embedding: list[float], top_k: int = 5,
+    ) -> list[RetrievalResult]:
+        return [
+            RetrievalResult(
+                chunk_id=chunk_id,
+                content=chunk.content,
+                score=1.0,
+                metadata=chunk.metadata,
+            )
+            for chunk_id, chunk in self.tables.get(table_name, {}).items()
+        ][:top_k]
+
+    def delete_by_ids(self, table_name: str, chunk_ids: list[str]) -> None:
+        for chunk_id in chunk_ids:
+            self.tables.get(table_name, {}).pop(chunk_id, None)
+
+    def drop_table(self, table_name: str) -> None:
+        if self.fail_drops:
+            raise RuntimeError("remote cleanup unavailable")
+        self.tables.pop(table_name, None)
+
+    def close(self) -> None:
+        return
+
+
+class SafetyRetriever(Retriever):
+    async def retrieve(
+        self,
+        query: str,
+        query_embedding: list[float],
+        table_name: str,
+        top_k: int = 5,
+        chunks_for_bm25: list[ChunkData] | None = None,
+    ) -> list[RetrievalResult]:
+        return []
+
+
+class SafetyReranker(Reranker):
+    async def rerank(
+        self, query: str, results: list[RetrievalResult], config: RerankerConfig,
+    ) -> list[RetrievalResult]:
+        return results
+
+
+class SafetyFactory(RAGComponentFactory):
+    def __init__(self) -> None:
+        super().__init__(":memory:")
+        self.parser = SafetyParser()
+        self.store = SafetyStore()
+
+    def create_parser(self) -> DocumentParser:
+        return self.parser
+
+    def create_chunker(self) -> TextChunker:
+        return SafetyChunker()
+
+    def create_embedding_provider(self) -> EmbeddingProvider:
+        return SafetyEmbedding()
+
+    def create_vector_store(self) -> VectorStore:
+        return self.store
+
+    def create_retriever(self, vector_store: VectorStore):
+        return SafetyRetriever()
+
+    def create_reranker(self) -> Reranker:
+        return SafetyReranker()
+
+
+def _services(db, tmp_path, monkeypatch):
+    storage_root = tmp_path / "knowledge"
+    monkeypatch.setattr(
+        "misaka.services.knowledge.document_service.get_kb_storage_dir",
+        lambda kb_id: storage_root / kb_id,
+    )
+    monkeypatch.setattr(
+        "misaka.services.knowledge.kb_service.get_kb_storage_dir",
+        lambda kb_id: storage_root / kb_id,
+    )
+    factory = SafetyFactory()
+    orchestrator = RAGOrchestrator(factory, db)
+    kb_service = KnowledgeBaseService(db, orchestrator)
+    doc_service = DocumentService(db, orchestrator)
+    return factory, kb_service, doc_service
+
+
+def _config() -> EmbeddingConfig:
+    return EmbeddingConfig("embedding", "https://example.test", "key")
+
+
+async def test_reprocess_failure_keeps_active_index_and_statistics(
+    db, tmp_path, monkeypatch,
+) -> None:
+    factory, kb_service, doc_service = _services(db, tmp_path, monkeypatch)
+    kb = kb_service.create("Safety", embedding_model_id="embedding", embedding_router_config_id="r")
+    source = tmp_path / "notes.txt"
+    source.write_text("known good content", encoding="utf-8")
+    document = await doc_service.upload_document(kb.id, str(source), _config())
+    before = kb_service.get(kb.id)
+    before_chunks = db.get_kb_chunks_by_index(kb.id, before.active_index_version)
+
+    factory.parser.fail = True
+    with pytest.raises(RuntimeError, match="parser unavailable"):
+        await doc_service.reprocess_document(document.id, _config())
+
+    after = kb_service.get(kb.id)
+    assert after.active_index_version == before.active_index_version
+    assert db.get_kb_chunks_by_index(kb.id, after.active_index_version) == before_chunks
+    assert after.chunk_count == 1
+    assert kb_service.get_kb_for_chat_selection()[0]["id"] == kb.id
+
+
+async def test_rebuild_missing_source_preserves_active_index(db, tmp_path, monkeypatch) -> None:
+    _, kb_service, doc_service = _services(db, tmp_path, monkeypatch)
+    kb = kb_service.create("Safety", embedding_model_id="embedding", embedding_router_config_id="r")
+    source = tmp_path / "notes.txt"
+    source.write_text("known good content", encoding="utf-8")
+    document = await doc_service.upload_document(kb.id, str(source), _config())
+    before = kb_service.get(kb.id)
+    Path(document.storage_path).unlink()
+
+    result = await kb_service.rebuild_embeddings(kb.id, _config())
+
+    after = kb_service.get(kb.id)
+    assert result["error_count"] == 1
+    assert after.active_index_version == before.active_index_version
+    assert after.chunk_count == 1
+    assert kb_service.get_kb_for_chat_selection()[0]["id"] == kb.id
+
+
+async def test_vector_cleanup_failure_is_durable_and_retryable(db, tmp_path, monkeypatch) -> None:
+    factory, kb_service, doc_service = _services(db, tmp_path, monkeypatch)
+    kb = kb_service.create("Safety", embedding_model_id="embedding", embedding_router_config_id="r")
+    source = tmp_path / "notes.txt"
+    source.write_text("known good content", encoding="utf-8")
+    document = await doc_service.upload_document(kb.id, str(source), _config())
+    old_version = kb_service.get(kb.id).active_index_version
+
+    factory.store.fail_drops = True
+    await doc_service.reprocess_document(document.id, _config())
+
+    pending = db.get_pending_kb_cleanup_jobs()
+    assert len(pending) == 1
+    assert pending[0].index_version == old_version
+    assert db.get_kb_chunks_by_index(kb.id, old_version)
+
+    factory.store.fail_drops = False
+    assert await kb_service.retry_pending_cleanup() == 1
+    assert db.get_pending_kb_cleanup_jobs() == []
+    assert db.get_kb_chunks_by_index(kb.id, old_version) == []
+
+
+async def test_delete_cancels_and_waits_for_active_upload(db, tmp_path, monkeypatch) -> None:
+    factory, kb_service, doc_service = _services(db, tmp_path, monkeypatch)
+    kb = kb_service.create("Safety", embedding_model_id="embedding", embedding_router_config_id="r")
+    source = tmp_path / "notes.txt"
+    source.write_text("known good content", encoding="utf-8")
+
+    started = asyncio.Event()
+    release = asyncio.Event()
+    original_embed = SafetyEmbedding.embed_texts
+
+    async def slow_embed(self, texts, config, batch_size=32):
+        started.set()
+        await release.wait()
+        return await original_embed(self, texts, config, batch_size)
+
+    monkeypatch.setattr(SafetyEmbedding, "embed_texts", slow_embed)
+    upload = asyncio.create_task(doc_service.upload_document(kb.id, str(source), _config()))
+    await started.wait()
+    processing_doc = doc_service.get_documents(kb.id)[0]
+
+    assert await doc_service.delete_document(processing_doc.id, _config()) is True
+    with pytest.raises(asyncio.CancelledError):
+        await upload
+    assert doc_service.get_document(processing_doc.id) is None
+    assert kb_service.get_kb_for_chat_selection() == []
diff --git a/tests/unit/test_service_container.py b/tests/unit/test_service_container.py
index e2a9475..d4730a0 100644
--- a/tests/unit/test_service_container.py
+++ b/tests/unit/test_service_container.py
@@ -95,5 +95,7 @@ def test_configure_remote_vector_backend_rebuilds_services(
         assert container.rag_orchestrator is not old_orchestrator
         assert container.kb_service._orchestrator is container.rag_orchestrator
         assert container.document_service._orchestrator is container.rag_orchestrator
+        assert container.kb_service._index_manager._orchestrator is container.rag_orchestrator
+        assert container.document_service._index_manager._orchestrator is container.rag_orchestrator
         assert container.rag_orchestrator._factory._backend == "seekdb"
         assert container.rag_orchestrator._factory._seekdb_mode == "seekdb_remote"