feat: 自托管交付的安装、备份与恢复工具及部署文档 (#15)

- scripts/ops.py：install-check（依赖与资源版本、客户端不得低于服务端）、init-database（建空库
  并提示最小权限）、backup（全库 dump ＋ manifest，不含凭据）、restore（默认只写空库、
  --confirm 必需、覆盖需 --force、拒绝系统库与带 CREATE DATABASE/USE 的 dump、比对源库校验和）、
  verify（完整性 ＋ 可选两账号接口闭环）、smoke（干净实例上建两个演练账号走通学习闭环）
- scripts/bench.py：写明规模的人造数据集性能测量，记录数据量、机器与 p50/p95
- tests/test_lexgo_ops.py：版本解析、库名白名单、dump 安全性、manifest 字段白名单等无库测试
- Wiki 新增 Deployment-and-Operations 页面与 wiki-docs.json 映射（含升级回滚与已知限制）
- 更新 Architecture、Business-Rules、Local-Development、Product-Requirements、Home 与
  README/AGENTS；本单无 schema 与接口变化

AGENTS.md

@@ -27,6 +27,10 @@
 | 创建可选任务快照 | `python dev_scripts/harness.py archive 123 "修复登录超时"` |
 | 增量导出已有快照 | `python dev_scripts/harness.py export` |
 | 全量导出已有快照 | `python dev_scripts/harness.py export --all` |
+| 检查部署依赖 | `python scripts/ops.py install-check` |
+| 备份实例 | `python scripts/ops.py backup --out <目录>` |
+| 恢复到空库 | `python scripts/ops.py restore --dump <备份.sql.gz> --database <库名> --confirm` |
+| 校验实例 | `python scripts/ops.py verify --database <库名>` |
 
 ## 1. 永久规则
 
@@ -282,6 +286,7 @@ MVP 内所有单元任务通过后才能做 MVP 集成验收；MVP 通过后才
 - #4 独立阅读选择小样位于 `spikes/selection/`，`python spikes/selection/serve.py` 默认仅本机 5184。桌面鼠标/键盘与 11 项测试已验证，#4 已获用户验收并关闭；真实手机长按/手柄/滚动详细证据仍未提供；禁止把窄屏桌面当作真机验收。Intl.Segmenter 只用于 UI 范围验证，不替代 #3 NLP；释义保存只在内存。固定 LinguaCafe 源码对照和与 v1 的差异记录见架构 Wiki。
 - 2026-09-11 用户确认正式 NLP/词典采用全 Go。#5 已验收并合入 main；#6 使用 Go WordNet 解析和词形候选、Go Unicode 原文分片、schema v4 共享词典资源表，不调用 Python NLP。WordNet 3.0 ZIP 来源与摘要见 `server/wordnet-resource.json`，许可保留在 `server/WORDNET-LICENSE.txt`。词形候选不等于上下文消歧，不自动合并个人学习状态；#3 Python 小样只保留历史验证。当前词典仅英语释义，个人释义输入为临时草稿，持久化归 #7。
 - 2026-09-11 用户确认 #7 个人词条口径（三项由 Agent 定案）：身份为「学习者＋语言＋规范化词形」，大小写合并但**不按 lemma/候选合并**（`dog` 与 `dogs` 是两条记录）；首次保存默认「新词」；状态为 新词/学习中/已知/忽略，只有「学习中」带 1～7 级，对应原版 stage 2/1/0/-1～-7；例句只保存手输内容，不自动关联原文句子。schema v5 新增 `lexgo_terms`（唯一键加状态/等级检查约束），个人释义与共享词典分离且不进入审计日志；等级编辑 UI 归 #8/#12。
+- 2026-09-15 用户确认 #15 交付口径：只交付本机可复现的安装/备份/恢复材料并在本机演练，**不对外部署、不创建 release/tag、不邀请用户**；生产入口与 HTTPS 只写入文档；前端由反向代理托管 dist（不改后端代码）；试用实例从**空库**开始、管理员由显式 bootstrap 建立、不带默认密码；备份＝MySQL 全库 dump ＋ 环境配置（凭据只存运维密码库，不进仓库/日志），不新增定时备份；性能用人造数据集实测并写明环境，只作观察不给承诺。备份/恢复规则：`restore` 必须 `--confirm`、默认只写空库、覆盖需 `--force`、库名必须含 lexgo 且不能是系统库、拒绝带 CREATE DATABASE/USE 的 dump，恢复前后比对源库逐表内容校验和。**附件（#21）尚未实施，恢复契约目前只覆盖数据库**；真实回滚、HTTPS、多机与定时备份仍未验证。部署与运维规则见 Wiki 页 `Deployment-and-Operations`（镜像 `docs/11-deployment-and-operations.md`）。
 - 2026-09-15 用户确认 #14 显示与键盘口径：`theme ∈ {浅色,深色,跟随系统}`（默认跟随系统）与正文字号 `{标准,大,特大}`（1.0/1.15/1.3）**按账号保存在本机** `lexgo-learner-display:<账号 id>`，切换账号即换成该账号偏好或默认，退出回到默认，**不跨设备同步**；字号经 `--reader-font-scale` 只作用于阅读面（正文、释义内容、复习卡），不做全局缩放；深色用 `html[data-theme]` ＋ Element Plus 的 `html.dark`，`style.css` 的 `:root` 是文件内仅有的颜色字面量；阅读位置按账号＋章节保存滚动比例与该章 `content_sha256`，**正文换新版本后不恢复**；复习页 `空格`/`Enter` 显示答案、`1/2/3` 评分，输入类控件与聚焦按钮的按键不被劫持，带修饰键不拦截；移动验证用 390×844＋`hasTouch` 的 Playwright `mobile` 项目（桌面项目 `testIgnore: mobile-*`），**真机长按选择与手感仍需人工确认**，不得用模拟设备结果冒充真机。本单无 schema 与接口变化。
 - 2026-09-15 用户确认 #13 完成阅读与进度口径：`POST /api/v1/chapters/:id/complete` 只记已读、**不批量改变词语状态或等级**，只有 `ready` 章节可标记（其他 409），重复调用返回同一行且带 `duplicate`（不移动时间、不重复计数）；完成记录保存标记时的 `content_sha256`，**正文新版本后该章回到未读**（记录保留，重读后更新同一行），只改标题不影响，章节删除随外键级联；`GET /api/v1/progress` 统计只含本人与当前语言，已读与分母都只算可阅读（`ready`）章节，已知/学习中/新词/忽略分开计数，`dueNow` 与到期复习队列共用 `dueTermsQuery` 与同一服务端时钟；schema 升到 v7（新表 `lexgo_chapter_progress`，不用 ALTER TABLE），需显式 migrate。不做每日目标、日历、难度评分、统计导出与取消已读，也不做 X10 批量标已知。
 - 2026-09-15 既有缺陷记录：#32 编辑章节正文回到曾经用过的版本返回 500（`lexgo_ingest_jobs` 请求键与 `edit:<章节>:<内容摘要>` 冲突），已定位未修复，方案待用户确认；#13 的真实验证脚本因此自建 fixture 而不改写既有正文。

README.md

@@ -8,10 +8,10 @@
 - [英语分词与离线词典验证小样](spikes/english/README.md)（#3 已验收，独立本机入口）
 - [阅读选择验证小样](spikes/selection/README.md)（#4 已验收，真机详细测试证据缺口保留）
 - [项目档案](docs/00-project-profile.md) · [需求总览](docs/09-product-requirements-overview.md)
-- [工作量估算](docs/10-workload-estimate.md)：#2、#3、#4、#5、#6、#7、#8、#9、#10、#11、#12、#13、#14、#18 已验收，原规划中的 #5 已完成；剩余 #15 与新增 #21、#24 按工单复核；缺陷 #32 待确认修复方案；后续结合集成结果重估，旧全量研究仅供参考。
+- [工作量估算](docs/10-workload-estimate.md)：#2、#3、#4、#5、#6、#7、#8、#9、#10、#11、#12、#13、#14、#18 已验收，原规划中的 #5 已完成；#15 已实现待验收；新增 #21、#24 待排期；缺陷 #32 待确认修复方案；后续结合集成结果重估，旧全量研究仅供参考。
 - [四阶段实施总览 #16](https://git.ilapage.cn/OPC/lexgo/issues/16)：14 张单元工单，工程基础 → 技术验证 → 首条学习闭环 → 补齐 MVP；原型 v1 已获用户验收。两端使用账号（用户名）＋密码登录，不要求邮箱。
 - [原型工单 #1](https://git.ilapage.cn/OPC/lexgo/issues/1)：Quant-UX 桌面/手机原型 v1，预览入口与审核记录见工单及需求总览。
-- [工作流](docs/01-workflow.md) · [开发与验证](docs/04-local-development-and-verification.md)
+- [工作流](docs/01-workflow.md) · [开发与验证](docs/04-local-development-and-verification.md) · [部署与运维](docs/11-deployment-and-operations.md)
 
 ## 本地工程基础
 

docs/02-architecture-and-code-map.md

@@ -2,8 +2,8 @@
 generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
 wiki_page: Architecture-and-Code-Map
 wiki_url: https://git.ilapage.cn/OPC/lexgo/wiki/Architecture-and-Code-Map.-
-wiki_revision: b61d9e1b0d9ae9fc24cb026580149f2b582a2526
-synchronized_at: 2026-09-15T07:44:50Z
+wiki_revision: bd4a0e705abe722f153a7410435def91a21d472a
+synchronized_at: 2026-09-15T08:27:37Z
 <!-- gitea-wiki-mirror:end -->
 
 # 架构与代码地图
@@ -371,3 +371,21 @@ schema v7 新增 `lexgo_chapter_progress`：一章一行（`chapter_id` 主键
 `style.css` 收敛为一套语义调色板：`:root` 里 53 个变量是**文件内仅有的颜色字面量**，其余规则全部走 `var(...)`；`html[data-theme='dark']` 覆盖同一批变量，`html.dark` 同时映射 Element Plus 的暗色变量（`main.ts` 引入 `element-plus/theme-chalk/dark/css-vars.css`），所以页面与组件跟随同一个选择。字号经 `--reader-font-scale` 只作用于阅读面（章节正文、释义面板内容、复习卡的词与释义），不做全局缩放。`ReaderView.vue` 在章节就绪后把正文下方的位置恢复到上次比例（正文版本变化则不恢复，等待可滚动后再应用，最多约 1 秒后放弃），并在滚动时按 400ms 防抖保存。`ReviewView.vue` 加了 `useReviewShortcuts` 与一行快捷键提示；`DisplaySettings` 出现在七个学习页面的头部。
 
 移动测试项目：`playwright.config.ts` 增加 `mobile`（390×844，`isMobile` + `hasTouch`，Chromium），桌面项目用 `testIgnore` 排除 `mobile-*.spec.ts`，移动项目用 `testMatch` 只跑它们。`e2e/mobile-fixtures.ts` 提供移动端共用的 mock、无溢出检查与经 CDP `Input.dispatchTouchEvent` 的真实触摸滑动。
+
+## #15 自托管交付：部署拓扑与运维工具（2026-09-15）
+
+**没有 schema 变化、没有接口变化**：本单新增运维工具、部署文档与完整演练证据。
+
+| 新增文件 | 职责 |
+|---|---|
+| `scripts/ops.py` | `install-check`（依赖与资源版本）、`init-database`（建空库并提示最小权限）、`backup`（全库 dump ＋ `manifest.json`）、`restore`（默认只写空库，写后自动校验）、`verify`（数据库完整性 ＋ 可选两账号接口闭环）、`smoke`（在干净实例上建两个演练账号走通学习闭环） |
+| `scripts/bench.py` | 在写明规模的人造数据集上测量接口耗时，输出数据量、机器信息与 p50/p95，不做容量承诺 |
+| `tests/test_lexgo_ops.py` | 不需要数据库的规则测试：版本号解析、库名白名单、dump 是否带库名切换、manifest 字段白名单、审计敏感列清单 |
+
+**一次备份包含什么**：LexGo 把所有持久数据都放在 MySQL 里——账号与会话、书籍与章节原文、导入任务、词典归档（LONGBLOB）、个人词条、复习排期与作答、阅读进度、审计日志。因此备份 = 全库 dump ＋ `.env.local`（凭据单独从运维密码库取）。词典资源在显式导入后进入数据库，恢复即带走，运行时不下载。
+
+**恢复的安全边界**：`restore` 必须带 `--confirm`；库名必须含 `lexgo` 且不能是系统库；目标库已有数据时默认拒绝，覆盖需要 `--force`；拒绝加载带 `CREATE DATABASE`/`USE` 的旧式 dump（那会把数据写进文件里指定的库）；恢复前后比对**源库**逐表内容校验和，源库被改动就中止。
+
+**部署文档**：新建 Wiki 页 `Deployment-and-Operations`（`docs/11-deployment-and-operations.md`，按 `docs/templates/deployment.md` 结构），并加入 `wiki-docs.json` 映射。内容含服务概览、环境要求（含「MySQL 客户端版本不得低于服务端」这条实测规则）、首次部署、配置与凭据来源、日常运维、健康检查、升级与回滚、备份与恢复、已知限制。
+
+**视图与进程**：后端单二进制监听 `127.0.0.1:8000`；两份 SPA 由反向代理托管 `dist`，反代把 `/api/` 转发到后端并把未知路径回落到 `index.html`；本机开发用 supervisor 托管 `lexgo-api`/`lexgo-learner`/`lexgo-admin` 三个 program。

docs/03-business-rules-and-glossary.md

@@ -2,8 +2,8 @@
 generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
 wiki_page: Business-Rules-and-Glossary
 wiki_url: https://git.ilapage.cn/OPC/lexgo/wiki/Business-Rules-and-Glossary.-
-wiki_revision: 5a44b1713b3fd2374baa78314872539f3e2e23ab
-synchronized_at: 2026-09-15T07:11:02Z
+wiki_revision: 5f2912edd09256cf5fc52059145d65918e99f9cb
+synchronized_at: 2026-09-15T08:27:38Z
 <!-- gitea-wiki-mirror:end -->
 
 # 业务规则与术语
@@ -290,3 +290,17 @@ exact优先；未命中再按WordNet异常表/词尾规则查候选，词性顺
 **键盘操作**：复习页 `空格`/`Enter` 显示答案，`1` 认识/答对、`2` 不认识/答错、`3` 再学一次；界面给出快捷键提示，并保持「显示答案」与「答对」按钮自动获得焦点，键盘用户可以全程不碰鼠标。按键规则：**输入框、文本域、下拉与可编辑区域内的任何按键都不触发快捷键**；按下的键若属于当前聚焦控件（按钮与链接的 `空格`/`Enter`）则交给浏览器处理，避免重复触发；带修饰键的组合键一律不拦截。阅读器 `Esc` 关闭释义面板，面板内的 `Esc` 只关面板。
 
 **触摸与滚动**：没有任何阻止默认行为的触摸处理；释义面板在窄屏是底部固定面板，页面为其预留底部空间，关闭后阅读位置不跳动。**长按原生选择与手柄拖动的手感仍需真机确认**，本单的触摸证据来自模拟设备。
+
+## #15 交付与运维规则（2026-09-15）
+
+**数据边界**：一个 LexGo 实例的全部持久数据都在 MySQL 中（账号、空间、会话、审计日志、书籍与章节原文、导入任务、词典归档、词条、复习排期、复习作答、阅读进度）。备份的完整对象因此是「全库 dump ＋ 环境配置文件」；环境配置含凭据，单独保存在运维密码库，不放进备份目录。
+
+**备份规则**：`backup` 使用 `mysqldump --single-transaction --routines --triggers --hex-blob --no-tablespaces`，输出 `.sql.gz` 与 `manifest.json`。manifest 记录创建时间、库名、产品标识、schema 版本、Git 提交、dump 的 sha256 与体积、逐表行数、客户端与服务端版本、包含与排除的内容说明；**不记录任何凭据值**。备份文件含个人数据，按个人数据对待：受控存放、不进 Git、不进工单。
+
+**恢复规则**：恢复是写入操作，必须显式 `--confirm`；默认只写**空库**，目标库已有数据时拒绝，覆盖需要 `--force` 并在覆盖前另行备份；库名必须含 `lexgo` 且不能是 MySQL 系统库；带 `CREATE DATABASE`/`USE` 的 dump 一律拒绝（会写错库）；恢复前后比对源库逐表内容校验和，源库变化即中止。
+
+**升级与回退规则**：schema 迁移只新增对象（`CREATE TABLE IF NOT EXISTS`），旧二进制按版本号校验；回退到旧二进制前把 `lexgo_schema.version` 写回旧版本号。若迁移删改数据或旧二进制无法读写新结构，则按恢复流程处理。**真实回滚演练尚未执行**，属已知限制。
+
+**安装规则**：空库不含默认密码与任何演示数据；管理员只能由显式 `bootstrap` 建立，库中已有账号时拒绝再次执行、不覆盖既有管理员；应用账号只对该库拥有最小权限，不使用管理员账号运行应用；服务默认只监听 `127.0.0.1`，对外由反向代理转发。
+
+**验收账号规则**：试用验证使用虚构的演练账号，密码由演练进程生成、只经环境变量传入，不写入磁盘、日志、工单或截图。

docs/04-local-development-and-verification.md

@@ -2,8 +2,8 @@
 generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
 wiki_page: Local-Development-and-Verification
 wiki_url: https://git.ilapage.cn/OPC/lexgo/wiki/Local-Development-and-Verification.-
-wiki_revision: e6255c24e7b966573a9851b1e2356e370d3daef1
-synchronized_at: 2026-09-15T07:11:02Z
+wiki_revision: caa80888cf42fd7fd7f040e9fe1147203fe7bfe3
+synchronized_at: 2026-09-15T08:27:38Z
 <!-- gitea-wiki-mirror:end -->
 
 # 本地开发与验证
@@ -493,3 +493,70 @@ node --test spikes/english/view.test.mjs
 **真实链路**：真实学习端＋真实 Go API＋真实 MySQL 走了一遍——粘贴 40 行长章节并处理完成，滚到 45% 后离开再返回，位置被恢复（偏差 < 150px，页面提示「已回到上次阅读位置」）；切到深色＋特大后刷新仍保持，正文实际字号变大（> 18px）；在真实到期卡上按空格显示了答案（**故意不评分**，避免改动 fixture 账号的真实复习计划）；另用 390×844 触摸上下文在真实数据上检查了书库/生词本/进度无横向溢出。fixture 书籍在结束时删除；截图见 `.local/evidence/issue14-real-*.png`、`.local/evidence/issue14-mobile-*.png`、`issue14-theme-*.png`、`issue14-keyboard-review.png`。临时用例运行后删除。
 
 **未验证**：真实手机上的长按选择、手柄拖动、滚动惯性、软键盘遮挡与安全区表现无法由本机自动化产生——本单的触摸证据来自模拟设备；**真机验证仍需人工完成**，这一点在 #4、#11 就已记录，本单未能消除。也未做 Safari/Firefox 等其他引擎的验证（项目既有测试只在 Chromium 上跑）。
+
+## #15 交付、备份与恢复验证（2026-09-15）
+
+本单**无 schema 变化、无迁移**：新增运维脚本、部署文档与演练证据。所有命令在仓库根执行；MySQL 客户端用 8.4（`LEXGO_MYSQL_BIN=D:/mysql843/bin`），因为本机 PATH 上是 5.7 客户端，`install-check` 会把它判为不达标。
+
+### 运维命令
+
+| 用途 | 命令 |
+|---|---|
+| 依赖与资源检查 | `python scripts/ops.py install-check` |
+| 建空库并提示权限 | `python scripts/ops.py init-database --database <库名>` |
+| 备份 | `python scripts/ops.py backup --out <目录>` |
+| 恢复到空库 | `python scripts/ops.py restore --dump <备份.sql.gz> --database <库名> --confirm` |
+| 校验（数据完整性，可选接口闭环） | `python scripts/ops.py verify --database <库名> [--manifest <manifest.json>] [--api http://127.0.0.1:<端口> --user <前缀> --password-env <变量名>]` |
+| 干净实例两账号闭环演练 | `python scripts/ops.py smoke --api <url> --admin-user <账号> --user <前缀>` |
+| 性能测量 | `python scripts/bench.py --database <库名> --api <url> --user <账号>` |
+
+### 完整演练（2026-09-15，本机一次性通过，27 项顶层检查全部 `[ok]`）
+
+演练脚本（`.local/tmp/issue15-drill.py`，一次性工具）依次执行：
+
+1. **干净安装**：`init-database` 建空库 → `migrate`（schema v7，14 张表）→ 确认「新库没有账号、没有业务数据」→ `bootstrap` 建管理员 → **再次 `bootstrap` 被拒绝**（退出码 1，不覆盖管理员）→ 起独立 API 实例（:8010）。
+2. **两个演练账号走通闭环**（`ops.py smoke`，19 项全过）：管理员建 `trial_a`/`trial_b` → 新账号书库与进度为空 → 粘贴章节 → 章节处理完成 → 分词返回可点选词 → 词典查询返回明确状态 → 保存词义 → 另一账号查不到该词条 → 新词立刻到期 → 作答成功 → 重复作答不重复记账 → 完成章节只记已读 → 进度反映活动；`B` 读 `A` 的书 404、书库仍为空。
+3. **备份 lexgo_dev**（约 12 MB gz）：manifest 记录 schema 7、提交哈希、逐表行数、sha256；确认 manifest 无凭据字段；确认 dump 不含 `CREATE DATABASE`/`USE`（不会误写回源库）。
+4. **恢复到空库** `lexgo_restore_drill` 并自动校验（33 项全过）：逐表行数与备份一致（`sys_user` 6、`books` 27、`chapters` 33、`terms` 13、`term_reviews` 13、`review_answers` 55、`login_logs` 192 等）→ **逐表内容校验和与源库一致**（13 张表）→ **源库 lexgo_dev 未被改动**。
+5. **安全边界**：恢复默认拒绝覆盖已有库；拒绝写入系统库 `mysql`；不带 `--confirm` 直接拒绝。
+6. **恢复实例的两账号闭环与越权校验**（`:8011`，41 项全过）——恢复后的账号仍可登录、词条与到期队列仍在、答题与完成章节保持幂等、`B` 无法读取或修改 `A` 的数据。
+7. **收尾**：删除两个演练库，确认 `lexgo_dev` 与测试库保留，`lexgo_dev` 仍是 schema 7。
+
+### 性能观察（`scripts/bench.py`，人造数据集，单机无并发）
+
+数据集：1 个账号、20 本书 20 章（每章约 500 个词）、2000 词条、8000 条复习作答。环境：Windows 10（10.0.19044）、8 核、31.7 GB 内存、MySQL 8.4.3、Go 工具链由 `scripts/server.py` 固定为 1.26.5。
+
+| 接口 | 请求数 | p50 | p95 | 最大 | 错误 |
+|---|---|---|---|---|---|
+| `POST /login` | 10 | 82.2 ms | 93.3 ms | 93.3 ms | 0 |
+| `GET /books` | 30 | 5.9 ms | 18.1 ms | 18.2 ms | 0 |
+| `GET /chapters/1` | 30 | 8.3 ms | 20.1 ms | 22.0 ms | 0 |
+| `GET /chapters/1/tokens` | 30 | 10.0 ms | 23.6 ms | 30.3 ms | 0 |
+| `GET /terms`（第 1 页） | 30 | 10.2 ms | 20.4 ms | 21.5 ms | 0 |
+| `GET /terms?query`（搜索） | 30 | 12.3 ms | 24.9 ms | 27.0 ms | 0 |
+| `GET /progress` | 30 | 15.7 ms | 26.1 ms | 29.1 ms | 0 |
+| `GET /reviews/queue` | 30 | 22.7 ms | 32.3 ms | 34.2 ms | 0 |
+
+**这些是观察值，不是容量承诺**：单机、单进程、无并发压力；登录接口每地址每分钟限 30 次，因此登录只测 10 次，该数字不代表登录吞吐。原始报告：`.local/issue15-bench.json`。
+
+### 集成验证矩阵（复用既有 Go 测试，`LEXGO_TEST_DB_NAME=lexgo_test_issue13` 70 项全过）
+
+| 验收要求 | 覆盖用例（全部通过） |
+|---|---|
+| 越权与隔离 | `TestMySQLLibraryIsolationAndOwnership`、`TestMySQLTermIsolationAndInputRules`、`TestMySQLPhraseRulesAndIsolation`、`TestMySQLReviewAnswerErrorsAndOwnership`、`TestMySQLAccountIsolationAndRevocation` |
+| 任务崩溃重试 | `TestMySQLIngestRecoveryWithoutRestart`、`TestMySQLIngestRecoveryAfterRestart`、`TestMySQLIngestAttemptsAreBoundedAndManualRetryRestarts`、`TestMySQLRetryAfterContentRestoredPublishesSameChapter`、`TestMySQLRecoverySkipsSupersededJobs` |
+| 并发答题与并发删除 | `TestMySQLReviewConcurrentReplayOfOneAnswer`、`TestConcurrentDuplicateAccountHasOneWinner`、`TestMySQLConcurrentChapterDelete`、`TestMySQLDeleteDuringProcessing` |
+| 长文与输入上限 | `TestMySQLPasteRejectsInvalidInputAndLimits`、`TestMySQLTextUploadKeepsChapterLimit`、`TestMySQLTextUploadRejectsInvalidSubmissions`、`TestMySQLTextUploadRejectsInvalidSubmissions` |
+| 迁移与回退标记 | `TestMigrationRefusesUnownedOrUnsupportedSchema`、`TestMigrationCanResumeOwnedPartialSchema`、全部 `TestMigrationFrom*` |
+
+### 治理与其它测试
+
+`python -m unittest discover -s tests` 65 项通过（新增 9 项 `tests/test_lexgo_ops.py`，不需要数据库）；`harness.py check --strict` 与 `sync --check` 通过。本单未改前端与后端业务代码，仍复跑 `LEXGO_TEST_DB_NAME=lexgo_test_issue13 python scripts/server.py test-integration` 70 项通过。
+
+### 未验证
+
+- **真实回滚（升级后切回旧二进制）未演练**：文档给出规则（迁移只新增对象，回退前写回版本号），但没有在真实实例上做过升级/回滚全过程。
+- 未验证 HTTPS、域名、反代配置、多机与灰度；本机演练只在 `127.0.0.1` 上起两个临时实例。
+- 未做定时备份、监控与告警；备份与恢复均为人工触发。
+- 大库恢复耗时与磁盘空间上限未测；演练库与生产库规模不同。
+- 附件（封面/音频，#21）尚未实现，恢复范围目前只覆盖数据库。

docs/09-product-requirements-overview.md

@@ -2,8 +2,8 @@
 generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
 wiki_page: Product-Requirements-Overview
 wiki_url: https://git.ilapage.cn/OPC/lexgo/wiki/Product-Requirements-Overview.-
-wiki_revision: 358d7720d952a49b6827a96173a4f63387ac6989
-synchronized_at: 2026-09-15T07:44:50Z
+wiki_revision: a1847957e287c7eb1a54905e13632aeca4cd75f7
+synchronized_at: 2026-09-15T08:27:38Z
 <!-- gitea-wiki-mirror:end -->
 
 # 产品需求总览
@@ -287,3 +287,9 @@ F11 的完成阅读与基础进度已于 2026-09-15 通过用户验收：阅读
 F12 的电脑与手机基础体验已于 2026-09-15 通过用户验收：新增「显示」设置（浅色/深色/跟随系统；正文标准/大/特大），偏好按账号保存在本机、账号切换不串、不跨设备同步；阅读位置按账号与章节保留，正文换新版本后不套用旧位置；复习页支持空格显示答案与 1/2/3 评分，输入框内不误触全局快捷键，焦点环可见；手机端以 390×844 触摸视口验证八个界面无横向溢出、均可返回、触摸滑动可滚动且不误开释义面板。本次无数据库与接口变化。
 
 仍未实现并留给后续工单：自托管试用交付与完整恢复（#15）、书籍音频与封面附件（#21）、书库列表优化（#24），以及已定位未修复的缺陷 #32。PWA、离线能力、自定义配色、全局 UI 缩放与偏好跨设备同步明确不在范围内。**真实手机的长按选择与操作手感仍需人工确认**，本单的触摸证据来自模拟设备。
+
+## #15 交付状态更新（2026-09-15）
+
+B04 与 F01–F12 的集成交付已实现，待用户验收：新部署与运维 Wiki 页（首次部署、配置与凭据来源、日常运维、健康检查、升级回滚、备份恢复、已知限制），运维工具 `scripts/ops.py`（`install-check` / `init-database` / `backup` / `restore` / `verify` / `smoke`）与 `scripts/bench.py`。本机完成完整演练：空库安装并由管理员建立两个虚构演练账号走通学习闭环；备份开发库后恢复到空实例，逐表行数与内容校验和一致、源库未被改动，两账号闭环与越权校验全通过。性能在 20 章／2000 词条／8000 条作答的人造数据集上实测（p50 6–23 ms，登录 82 ms，均无错误），只作观察不给承诺。
+
+用户已确认的范围边界：本次只交付本机可复现的安装/备份/恢复材料并在本机演练，**不对外部署、不创建发布标签、不邀请用户**；生产入口与 HTTPS 只写入文档。**真实回滚、定时备份、监控告警、多机部署与 HTTPS 仍未验证**；附件（音频/封面，#21）尚未实现，因此恢复契约目前只覆盖数据库，待 #21 落地后由本单的恢复验收纳入附件。遗留缺陷 #32（编辑正文回到曾用版本返回 500）仍待确认修复方案。

docs/README.md

@@ -2,8 +2,8 @@
 generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
 wiki_page: Home
 wiki_url: https://git.ilapage.cn/OPC/lexgo/wiki/Home
-wiki_revision: 9f6485206c60c737b386fdf2f2946adf606d53e7
-synchronized_at: 2026-09-15T07:44:49Z
+wiki_revision: 22731c2b344b3e0c75eb892c897d36d1f153a441
+synchronized_at: 2026-09-15T08:27:37Z
 <!-- gitea-wiki-mirror:end -->
 
 # LexGo 文档入口
@@ -86,3 +86,5 @@ Quant-UX 原型 v1 已通过用户验收。[桌面预览](https://qux.ilapage.cn
 #13 完成章节与个人基础进度已于 2026-09-15 通过用户验收：阅读页正文下方提供显式的「标记本章已读」，只记录已读、不批量改变词语状态；重复标记返回同一条记录、不重复计数。书籍页章节列表显示「已读」，新的「进度」页面显示已读章节（只统计可阅读章节，形如「3 / 8 章（可阅读）」）、待复习、已知词、学习中、新词、忽略与已保存词条数，并按书显示已读进度；待复习与到期复习队列使用同一规则与同一时钟。正文被改成新版本后该章会回到未读（记录保留），重新标记即更新同一条记录；删除章节时记录随之消失。本次新增 schema v7（一张新表），需要显式迁移；开发库迁移前后各表计数一致。真实 API＋MySQL 37 项检查与浏览器页面数字对照数据库直查全部一致；PR #33 已 fast-forward-only 合入 main。
 
 #14 桌面与手机体验、主题与键盘操作已于 2026-09-15 通过用户验收：新增「显示」设置（浅色/深色/跟随系统，正文标准/大/特大），偏好按账号保存在本机，切换账号不会串；阅读位置按账号与章节保留，正文换成新版本后不再套用旧位置；复习页支持空格显示答案、1/2/3 评分，输入框内不会误触快捷键，所有可交互元素有可见焦点环。手机端用 390×844 触摸视口验证八个界面无横向溢出、都能返回、触摸滑动滚动而不误开释义面板，深色与特大字号刷新后保持。本次没有数据库与接口变化；PR #34 已 fast-forward-only 合入 main。真实手机的长按选择与手感仍需人工确认。
+
+#15 自托管试用交付已实现，待用户验收：新增运维工具（`scripts/ops.py`：依赖检查、建空库、备份、恢复、校验、两账号闭环演练；`scripts/bench.py`：写明规模的人造数据集性能测量）、新的 Wiki 部署与运维页，以及一次完整的本机演练——空库安装（建库 → 迁移 → 首次 bootstrap 建管理员 → 管理员建两个虚构演练账号 → 走通粘贴/阅读/查词/保存/复习/完成章节/进度）、备份 lexgo_dev、恢复到空实例并用第二个 API 实例完成两账号闭环与越权校验，恢复后逐表行数与内容校验和都和源库一致、源库未被改动。备份含全部用户数据（数据库）与环境配置（单独保管凭据），恢复默认只写空库、覆盖需要显式确认与 `--force`、拒绝写入系统库。性能在 20 章/2000 词条/8000 条作答的人造数据集上实测并记录环境，只作观察不给承诺。真实回滚、HTTPS、多机与定时备份仍未验证；附件（#21）尚未实施，恢复范围目前只覆盖数据库。

scripts/bench.py

@@ -0,0 +1,207 @@
+"""Measures the LexGo API on a stated synthetic dataset.
+
+The numbers this prints describe one machine and one dataset; they are an observation, not a
+capacity promise. Everything runs against a database whose name contains ``lexgo`` and that the
+operator creates for the run, so no real instance is touched.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import platform
+import statistics
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+import ops  # noqa: E402
+
+WORDS = ("curiosity", "door", "step", "reading", "language", "garden", "window", "music", "river",
+         "stone", "letter", "morning", "shadow", "bridge", "silence", "travel", "kitchen", "paper")
+
+
+def machine_facts() -> dict:
+    facts = {
+        "os": platform.platform(),
+        "cpu_count": os.cpu_count(),
+        "python": platform.python_version(),
+    }
+    try:
+        import shutil
+
+        if os.name == "nt":
+            output = subprocess.run(["wmic", "computersystem", "get", "TotalPhysicalMemory"],
+                                    capture_output=True, text=True, encoding="utf-8", errors="replace").stdout
+            numbers = [line.strip() for line in output.splitlines() if line.strip().isdigit()]
+            if numbers:
+                facts["memory_gb"] = round(int(numbers[0]) / (1024 ** 3), 1)
+        else:
+            with open("/proc/meminfo") as handle:
+                for line in handle:
+                    if line.startswith("MemTotal"):
+                        facts["memory_gb"] = round(int(line.split()[1]) / (1024 ** 2), 1)
+                        break
+        facts["go"] = subprocess.run(["go", "version"], capture_output=True, text=True,
+                                     encoding="utf-8", errors="replace").stdout.strip()
+        del shutil
+    except Exception:
+        pass
+    return facts
+
+
+def seed(database: str, chapters: int, chapter_words: int, terms: int, reviews: int) -> dict:
+    """Fills a fresh benchmark database with a synthetic learner, book, terms and schedules.
+
+    The account itself is left alone: it was created through the API beforehand, so its password
+    hash is a real one and the benchmark measures the same login path a user takes.
+    """
+    for table in ("lexgo_review_answers", "lexgo_term_reviews", "lexgo_terms", "lexgo_chapters",
+                  "lexgo_books", "lexgo_spaces", "lexgo_sessions"):
+        ops.query(database, "DELETE FROM " + table)
+    owner = int(ops.scalar(database, "SELECT user_id FROM sys_user ORDER BY user_id LIMIT 1", "0") or 0)
+    if not owner:
+        raise SystemExit("先创建一个账号再压测（例如 smoke 或 bootstrap）。")
+    ops.query(database, "INSERT INTO lexgo_spaces (owner_id, language) VALUES (%d, 'en')" % owner)
+    stamp = "2026-01-01 00:00:00.000"
+
+    body = " ".join(WORDS[index % len(WORDS)] for index in range(chapter_words))
+    for chapter in range(1, chapters + 1):
+        text = "%s chapter %d. " % (body, chapter)
+        sha = hashlib.sha256(text.encode("utf-8")).hexdigest()
+        ops.query(database, "INSERT INTO lexgo_books (id, owner_id, title, language, created_at, updated_at) "
+                            "VALUES (%d, %d, 'Bench book %d', 'en', '%s', '%s')" % (chapter, owner, chapter, stamp, stamp))
+        ops.query(database, "INSERT INTO lexgo_chapters (id, book_id, owner_id, ordinal, title, original_text, char_count, "
+                            "content_sha256, status, created_at, updated_at) VALUES (%d, %d, %d, 1, 'Bench %d', '%s', %d, "
+                            "'%s', 'ready', '%s', '%s')" % (chapter, chapter, owner, chapter, text.replace("'", "''"), len(text), sha, stamp, stamp))
+
+    rows = []
+    for term_id in range(1, terms + 1):
+        word = "%s%d" % (WORDS[term_id % len(WORDS)], term_id)
+        rows.append("(%d, %d, 'en', '%s', '%s', '虚构释义', '', 'new', 0, '%s', '%s')" % (term_id, owner, word, word, stamp, stamp))
+        if len(rows) == 500:
+            ops.query(database, "INSERT INTO lexgo_terms (id, owner_id, language, term, original_form, definition, examples, "
+                                "status, level, created_at, updated_at) VALUES " + ",".join(rows))
+            rows = []
+    if rows:
+        ops.query(database, "INSERT INTO lexgo_terms (id, owner_id, language, term, original_form, definition, examples, "
+                            "status, level, created_at, updated_at) VALUES " + ",".join(rows))
+
+    rows = []
+    for term_id in range(1, terms + 1):
+        rows.append("(%d, %d, 'en', DATE_ADD('%s', INTERVAL %d SECOND), 0, 0, 0)" % (term_id, owner, stamp, term_id % 900000))
+        if len(rows) == 500:
+            ops.query(database, "INSERT INTO lexgo_term_reviews (term_id, owner_id, language, due_at, review_count, "
+                                "correct_count, wrong_count) VALUES " + ",".join(rows))
+            rows = []
+    if rows:
+        ops.query(database, "INSERT INTO lexgo_term_reviews (term_id, owner_id, language, due_at, review_count, "
+                            "correct_count, wrong_count) VALUES " + ",".join(rows))
+
+    if reviews:
+        rows = []
+        for index in range(1, reviews + 1):
+            rows.append("(%d, %d, 'b%063d', %d, 'correct', 'applied', 'new', 'learning', 0, 1, '%s', '%s', 0, '%s')"
+                        % (index, owner, index, (index % terms) + 1, stamp, stamp, stamp))
+            if len(rows) == 500:
+                ops.query(database, "INSERT INTO lexgo_review_answers (id, owner_id, answer_key, term_id, grade, result, "
+                                    "status_before, status_after, level_before, level_after, due_at_before, due_at_after, "
+                                    "requeued, created_at) VALUES " + ",".join(rows))
+                rows = []
+        if rows:
+            ops.query(database, "INSERT INTO lexgo_review_answers (id, owner_id, answer_key, term_id, grade, result, "
+                                "status_before, status_after, level_before, level_after, due_at_before, due_at_after, "
+                                "requeued, created_at) VALUES " + ",".join(rows))
+
+    return {
+        "owner_id": owner,
+        "chapters": chapters,
+        "words_per_chapter": chapter_words,
+        "terms": terms,
+        "review_answers": reviews,
+        "row_counts": ops.row_counts(database),
+    }
+
+
+def measure(label: str, call, repeats: int) -> dict:
+    timings = []
+    errors = 0
+    for index in range(repeats):
+        started = time.perf_counter()
+        status, _ = call(index)
+        elapsed = (time.perf_counter() - started) * 1000
+        if status >= 400:
+            errors += 1
+        timings.append(elapsed)
+    timings.sort()
+    return {
+        "endpoint": label,
+        "requests": repeats,
+        "errors": errors,
+        "p50_ms": round(statistics.median(timings), 1),
+        "p95_ms": round(timings[min(len(timings) - 1, int(len(timings) * 0.95))], 1),
+        "max_ms": round(timings[-1], 1),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="在写明规模的人造数据集上测量接口耗时")
+    parser.add_argument("--database", default="lexgo_bench")
+    parser.add_argument("--api", default="http://127.0.0.1:8012")
+    parser.add_argument("--user", default="bench_admin", help="压测账号")
+    parser.add_argument("--chapters", type=int, default=20)
+    parser.add_argument("--chapter-words", type=int, default=500)
+    parser.add_argument("--terms", type=int, default=2000)
+    parser.add_argument("--reviews", type=int, default=8000)
+    parser.add_argument("--repeats", type=int, default=30)
+    parser.add_argument("--out", default=str(ROOT / ".local" / "issue15-bench.json"))
+    args = parser.parse_args()
+
+    ops.load_env()
+    database = ops.validate_database_name(args.database, "压测")
+    dataset = seed(database, args.chapters, args.chapter_words, args.terms, args.reviews)
+    print("数据集：" + json.dumps(dataset["row_counts"], ensure_ascii=False))
+
+    client = ops.Api(args.api)
+    password = os.environ.get("LEXGO_BENCH_PASSWORD", os.environ.get("LEXGO_TRIAL_PASSWORD", ""))
+    status, token = client.login(args.user, password)
+    if status != 200 or not token:
+        raise SystemExit(args.user + " 无法登录：请先在演练库上建号并设置 LEXGO_BENCH_PASSWORD。")
+
+    # The login endpoint allows 30 attempts per minute per address, so a sustained loop would
+    # measure the limiter instead of the login path. A short loop keeps the number meaningful.
+    login_repeats = max(1, min(args.repeats, 10))
+    results = [measure("POST /login", lambda _: client.login(args.user, password), login_repeats)]
+    results.append(measure("GET /books", lambda _: client.call("GET", "books", token), args.repeats))
+    results.append(measure("GET /chapters/1", lambda _: client.call("GET", "chapters/1", token), args.repeats))
+    results.append(measure("GET /chapters/1/tokens", lambda _: client.call("GET", "chapters/1/tokens", token), args.repeats))
+    results.append(measure("GET /terms (page 1)", lambda _: client.call("GET", "terms?page=1&limit=20", token), args.repeats))
+    results.append(measure("GET /terms?query (search)", lambda _: client.call("GET", "terms?query=curiosity&limit=20", token), args.repeats))
+    results.append(measure("GET /progress", lambda _: client.call("GET", "progress", token), args.repeats))
+    results.append(measure("GET /reviews/queue", lambda _: client.call("GET", "reviews/queue", token), args.repeats))
+
+    report = {
+        "measured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "api": args.api,
+        "database": database,
+        "dataset": dataset,
+        "machine": machine_facts(),
+        "mysql_server": ops.scalar("", "SELECT VERSION()", ""),
+        "results": results,
+        "note": "单机、单进程、无并发压力；数字是观察值，不是容量承诺。登录接口每分钟每地址限 30 次，"
+                "因此登录只测 10 次，且该数字不代表登录吞吐。",
+    }
+    Path(args.out).write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
+    for row in results:
+        print("  %-28s p50 %7.1f ms  p95 %7.1f ms  max %7.1f ms  错误 %d" %
+              (row["endpoint"], row["p50_ms"], row["p95_ms"], row["max_ms"], row["errors"]))
+    print("报告：" + args.out)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ops.py

@@ -0,0 +1,783 @@
+"""LexGo operations: dependency check, backup, restore and post-restore verification.
+
+Everything a self-hosted instance needs to be installed, backed up and restored is in MySQL:
+accounts, books and chapter text, personal terms, review schedules and answers, the imported
+WordNet archive and the audit logs. This script therefore never invents a second storage
+location: a dump plus the environment file is a complete backup.
+
+Safety rules, because a restore can overwrite a working instance:
+
+* a restore always needs ``--confirm``;
+* the target database name must contain ``lexgo`` and must not be a MySQL system schema;
+* a database that already holds LexGo data is refused unless ``--force`` is given;
+* credentials are read from the environment or ``.env.local`` and are never written into a
+  manifest, a log line or a dump name.
+
+Only the standard library is used, so the script runs on the deployment host without installing
+anything beyond the MySQL client.
+"""
+from __future__ import annotations
+
+import argparse
+import gzip
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import tempfile
+import time
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+ENV_FILE = ROOT / ".env.local"
+SYSTEM_SCHEMAS = {"mysql", "information_schema", "performance_schema", "sys"}
+# Columns that must never appear in the audit tables: they exist to prove that no credential,
+# request body or private learning content is stored there.
+AUDIT_BANNED_COLUMNS = {"password", "token", "body", "content", "definition", "text"}
+TABLES = [
+    "sys_user", "lexgo_spaces", "lexgo_sessions", "lexgo_login_logs", "lexgo_operation_logs",
+    "lexgo_books", "lexgo_chapters", "lexgo_ingest_jobs", "lexgo_dictionaries",
+    "lexgo_terms", "lexgo_term_reviews", "lexgo_review_answers", "lexgo_chapter_progress",
+]
+SCHEMA_VERSION = 7
+
+
+class OpsError(SystemExit):
+    """A failure the operator has to read and act on."""
+
+
+def load_env(path=ENV_FILE):
+    """Reads .env.local. Values are never printed; only key names are."""
+    values = {}
+    if path.exists():
+        for line in path.read_text(encoding="utf-8").splitlines():
+            line = line.strip()
+            if not line or line.startswith("#") or "=" not in line:
+                continue
+            key, value = line.split("=", 1)
+            values[key.strip()] = value.strip()
+    for key, value in values.items():
+        os.environ.setdefault(key, value)
+    return values
+
+
+def mysql_binary(name):
+    """Finds mysqldump/mysql via LEXGO_MYSQL_BIN, then PATH."""
+    configured = os.environ.get("LEXGO_MYSQL_BIN")
+    if configured:
+        candidate = Path(configured) / (name + (".exe" if os.name == "nt" else ""))
+        if candidate.exists():
+            return str(candidate)
+    found = shutil.which(name)
+    if not found:
+        raise OpsError(
+            "找不到 " + name + "：请安装 MySQL 客户端，或用 LEXGO_MYSQL_BIN 指向客户端目录。"
+        )
+    return found
+
+
+def connection_args(database=None):
+    """Builds client arguments. The password travels through the environment, not argv."""
+    args = [
+        "--host=" + os.environ.get("LEXGO_DB_HOST", "127.0.0.1"),
+        "--port=" + os.environ.get("LEXGO_DB_PORT", "3306"),
+        "--user=" + os.environ.get("LEXGO_DB_USER", ""),
+    ]
+    if database:
+        args.append(database)
+    return args
+
+
+def client_env():
+    env = dict(os.environ)
+    password = os.environ.get("LEXGO_DB_PASSWORD")
+    if password:
+        env["MYSQL_PWD"] = password
+    return env
+
+
+def run(command, env=None, capture=True):
+    result = subprocess.run(command, env=env, capture_output=capture, text=True, encoding="utf-8", errors="replace")
+    if result.returncode != 0:
+        message = (result.stderr or "").strip().splitlines()
+        # The last line of a client error is the one an operator needs; never echo arguments,
+        # which is why the command line itself is not part of the message.
+        raise OpsError("命令失败：" + (message[-1] if message else "未知错误"))
+    return result.stdout or ""
+
+
+def query(database, sql):
+    # SQL travels through stdin: a long INSERT would exceed the Windows command line limit.
+    result = subprocess.run([mysql_binary("mysql"), *connection_args(database), "--batch", "--skip-column-names"],
+                            input=sql.encode("utf-8"), env=client_env(), capture_output=True)
+    if result.returncode != 0:
+        lines = result.stderr.decode("utf-8", "replace").strip().splitlines()
+        raise OpsError("查询失败：" + (lines[-1] if lines else "未知错误"))
+    output = result.stdout.decode("utf-8", "replace")
+    return [line.split("\t") for line in output.strip().splitlines() if line]
+
+
+def scalar(database, sql, default=None):
+    rows = query(database, sql)
+    if not rows or not rows[0] or rows[0][0] == "NULL":
+        return default
+    return rows[0][0]
+
+
+def database_exists(database):
+    rows = query("", "SELECT SCHEMA_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME='%s'" % database)
+    return bool(rows)
+
+
+def sha256_of(path, chunk=1024 * 1024):
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        while True:
+            block = handle.read(chunk)
+            if not block:
+                break
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def git_commit():
+    try:
+        return subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(ROOT), capture_output=True, text=True,
+                              encoding="utf-8", errors="replace").stdout.strip()
+    except OSError:
+        return ""
+
+
+def version_of(text):
+    """Pulls the version out of a client or server banner.
+
+    A client banner reads "Ver 14.14 Distrib 5.7.38", where the first number is the protocol
+    version, so the version after "Distrib" is the one to compare.
+    """
+    banner = text or ""
+    match = re.search(r"Distrib\s+(\d+)\.(\d+)(?:\.(\d+))?", banner)
+    if not match:
+        match = re.search(r"(\d+)\.(\d+)(?:\.(\d+))?", banner)
+    if not match:
+        return None
+    return tuple(int(part) for part in match.groups(default="0"))
+
+
+def validate_database_name(name, action):
+    if not name or name in SYSTEM_SCHEMAS or "lexgo" not in name:
+        raise OpsError("拒绝在 %s 上执行：库名必须包含 lexgo 且不能是系统库。" % action)
+    return name
+
+
+def row_counts(database):
+    counts = {}
+    present = {row[0] for row in query("information_schema", "SELECT TABLE_NAME FROM TABLES WHERE TABLE_SCHEMA='%s'" % database)}
+    for table in TABLES:
+        if table in present:
+            counts[table] = int(scalar(database, "SELECT COUNT(*) FROM `%s`" % table, "0"))
+    return counts
+
+
+# ---------------------------------------------------------------- install-check
+
+
+def command_install_check(_args):
+    load_env()
+    checks = []
+    problems = []
+
+    def add(name, ok, detail):
+        checks.append((name, ok, detail))
+        if not ok:
+            problems.append(name)
+
+    # The client has to be at least as new as the server: an older mysqldump produces a dump the
+    # server cannot be restored from, which is worth catching before an incident rather than after.
+    try:
+        client = run([mysql_binary("mysql"), "--version"]).strip()
+    except OpsError as error:
+        client = str(error)
+    client_version = version_of(client)
+    add("MySQL 客户端存在", bool(client_version), client)
+    server = ""
+    database = os.environ.get("LEXGO_DB_NAME", "")
+    if database:
+        try:
+            server = scalar("", "SELECT VERSION()", "") or ""
+        except OpsError as error:
+            problems.append("数据库连接")
+            checks.append(("数据库连接", False, str(error)))
+    server_version = version_of(server)
+    if server_version:
+        add("MySQL 服务端为 8.x", server_version[0] == 8, server)
+    if client_version and server_version:
+        add("客户端版本不低于服务端", client_version >= server_version,
+            "客户端 " + ".".join(str(part) for part in client_version) + " / 服务端 " + ".".join(str(part) for part in server_version))
+    add("环境配置已加载", bool(database), "LEXGO_DB_NAME=" + (database or "（未设置）"))
+    for key in ("LEXGO_DB_USER", "LEXGO_DB_PASSWORD", "LEXGO_BOOTSTRAP_USERNAME"):
+        add("配置项 " + key + " 存在", bool(os.environ.get(key)), "已设置" if os.environ.get(key) else "未设置")
+    if not os.environ.get("LEXGO_LISTEN"):
+        print("  [info] LEXGO_LISTEN 未设置，服务将使用内置默认监听地址。")
+    # The dictionary resource is pinned by digest and imported explicitly, never downloaded at
+    # runtime; the pin file has to be present for an install to be reproducible.
+    resource = ROOT / "server" / "wordnet-resource.json"
+    if resource.exists():
+        pin = json.loads(resource.read_text(encoding="utf-8"))
+        add("WordNet 资源 pin", len(pin.get("sha256", "")) == 64 and bool(pin.get("source")), "sha256=" + pin.get("sha256", "")[:12] + "…")
+    else:
+        add("WordNet 资源 pin", False, "缺少 server/wordnet-resource.json")
+    for binary, label in (("go", "Go 工具链"), ("node", "Node"), ("pnpm", "pnpm")):
+        found = shutil.which(binary)
+        add(label + "（仅构建需要）", bool(found), found or "未安装：使用预构建产物时可不安装")
+    if os.environ.get("LEXGO_LISTEN", "").startswith("0.0.0.0"):
+        add("服务监听地址", False, "LEXGO_LISTEN 对外监听：" + os.environ["LEXGO_LISTEN"] + "，应由反向代理转发而不是直接暴露")
+
+    for name, ok, detail in checks:
+        print(("  [ok]   " if ok else "  [fail] ") + name + " — " + detail)
+    if problems:
+        print("\n检查未通过：" + "、".join(problems))
+        print("按 wiki 的 Deployment-and-Operations 页面补齐后再安装。")
+        raise SystemExit(1)
+    print("\n依赖检查通过。")
+    if database:
+        try:
+            version = scalar(database, "SELECT version FROM lexgo_schema WHERE id=1", None)
+            print("当前 schema 版本：" + (version or "未知（尚未迁移）"))
+        except OpsError:
+            print("当前 schema 版本：无法读取（数据库可能尚未初始化）")
+    return 0
+
+
+# ---------------------------------------------------------------- init-database
+
+
+def command_init_database(args):
+    """Creates the empty schema an installation starts from.
+
+    The server cannot connect to a database that does not exist yet, so this is the first step of
+    a fresh install. Grants are printed instead of applied: which account may reach the schema is
+    a decision for the operator's database policy, not for this tool.
+    """
+    load_env()
+    database = validate_database_name(args.database or os.environ.get("LEXGO_DB_NAME", ""), "建库")
+    if database_exists(database):
+        tables = int(scalar(database, "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='%s'" % database, "0") or 0)
+        print("库已存在：" + database + "（" + str(tables) + " 张表）")
+        if tables:
+            print("这是已有实例；需要升级时下一步是 python scripts/server.py migrate。")
+            return 0
+    else:
+        run([mysql_binary("mysql"), *connection_args(), "-e", "CREATE DATABASE `%s` CHARACTER SET utf8mb4" % database], env=client_env())
+        print("已创建库：" + database)
+    print(
+        "\n请为应用账号授予该库的最小权限（只限此库，不要使用管理员账号）：\n"
+        "  GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER, INDEX, DROP, REFERENCES\n"
+        "    ON `" + database + "`.* TO '<应用账号>'@'<来源主机>';\n"
+        "  FLUSH PRIVILEGES;\n"
+        "\n下一步：写入 .env.local（凭据取自运维密码库）后执行\n"
+        "  python scripts/server.py migrate\n"
+        "  python scripts/server.py bootstrap"
+    )
+    return 0
+
+
+# ---------------------------------------------------------------- backup
+
+
+def command_backup(args):
+    load_env()
+    database = validate_database_name(args.database or os.environ.get("LEXGO_DB_NAME", ""), "备份")
+    out_dir = Path(args.out) if args.out else ROOT / ".local" / "backups" / datetime.now().strftime("%Y%m%d-%H%M%S")
+    if out_dir.exists() and any(out_dir.iterdir()) and not args.force:
+        raise OpsError("输出目录非空：" + str(out_dir) + "（加 --force 覆盖）")
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    sql_path = out_dir / ("lexgo-" + stamp + ".sql")
+    gz_path = Path(str(sql_path) + ".gz")
+    command = [
+        mysql_binary("mysqldump"), *connection_args(),
+        "--single-transaction", "--routines", "--triggers", "--hex-blob", "--no-tablespaces",
+        "--default-character-set=utf8mb4",
+        # Deliberately without --databases: that option writes CREATE DATABASE and USE into the
+        # dump, which would send a restore straight back into the source schema instead of the
+        # target one. The client is given the target database instead.
+        database,
+    ]
+    print("正在备份 " + database + " …")
+    # The dump is captured and written as bytes: nothing about it is decoded or re-encoded on the
+    # way to disk, which keeps a binary-safe dump byte-exact.
+    dump = subprocess.run(command, env=client_env(), capture_output=True)
+    if dump.returncode != 0:
+        message = dump.stderr.decode("utf-8", "replace").strip().splitlines()
+        raise OpsError("mysqldump 失败：" + (message[-1] if message else "未知错误"))
+    with open(sql_path, "wb") as handle:
+        handle.write(dump.stdout)
+    with open(sql_path, "rb") as source, gzip.open(gz_path, "wb", compresslevel=6) as target:
+        shutil.copyfileobj(source, target)
+    sql_path.unlink()
+
+    manifest = {
+        "created_at": stamp,
+        "database": database,
+        "product": "lexgo",
+        "schema_version": int(scalar(database, "SELECT version FROM lexgo_schema WHERE id=1", "0") or 0),
+        "git_commit": git_commit(),
+        "dump_file": gz_path.name,
+        "dump_sha256": sha256_of(gz_path),
+        "dump_bytes": gz_path.stat().st_size,
+        "row_counts": row_counts(database),
+        "mysql_client": run([mysql_binary("mysqldump"), "--version"]).strip(),
+        "mysql_server": scalar("", "SELECT VERSION()", ""),
+        "contents": "accounts, spaces, sessions, audit logs, books, chapters, ingest jobs, "
+                    "dictionary archive, terms, review schedules, review answers, reading progress",
+        "excludes": "credentials and the environment file; back those up separately from the operations vault",
+    }
+    (out_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    print("备份完成：" + str(gz_path))
+    print("  schema 版本 " + str(manifest["schema_version"]) + "，提交 " + (manifest["git_commit"][:8] or "未知"))
+    print("  表行数 " + json.dumps(manifest["row_counts"], ensure_ascii=False))
+    print("  sha256 " + manifest["dump_sha256"][:16] + "…（完整值见 manifest.json）")
+    print("  凭据与 .env.local 不在其中，请用运维密码库单独保存。")
+    return 0
+
+
+# ---------------------------------------------------------------- restore
+
+
+def ensure_empty_target(database, force):
+    if not database_exists(database):
+        run([mysql_binary("mysql"), *connection_args(), "-e", "CREATE DATABASE `%s` CHARACTER SET utf8mb4" % database], env=client_env())
+        print("已创建空库 " + database)
+        return
+    existing = scalar(database, "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='%s'" % database, "0")
+    if int(existing or 0) == 0:
+        return
+    if not force:
+        raise OpsError(
+            "目标库 " + database + " 已有 " + str(existing) + " 张表。恢复默认只写空库；"
+            "确认要覆盖时再加 --force，并先对现有库做一次备份。"
+        )
+    print("警告：目标库已有数据，按 --force 覆盖。")
+    run([mysql_binary("mysql"), *connection_args(), "-e",
+         "DROP DATABASE `%s`; CREATE DATABASE `%s` CHARACTER SET utf8mb4" % (database, database)], env=client_env())
+
+
+def dump_targets_instead_of_source(dump):
+    """Reads the head of a dump and refuses one that would switch databases.
+
+    A dump taken with ``mysqldump --databases`` carries CREATE DATABASE and USE statements, so a
+    restore would write into the schema named inside the file rather than the requested target.
+    An older backup is rejected instead of being trusted.
+    """
+    with gzip.open(dump, "rb") as handle:
+        head = handle.read(256 * 1024).decode("utf-8", "replace")
+    return "CREATE DATABASE" in head.upper() or bool(re.search(r"^USE `", head, re.M))
+
+
+def dump_checksums(database):
+    """Per-table content checksums, used to prove a restore equals its source."""
+    checksums = {}
+    for table in TABLES:
+        rows = query(database, "CHECKSUM TABLE `%s`" % table)
+        if rows and len(rows[0]) > 1:
+            checksums[table] = rows[0][1]
+    return checksums
+
+
+def command_restore(args):
+    load_env()
+    if not args.confirm:
+        raise OpsError(
+            "恢复会写入数据库，需要显式确认：\n"
+            "  python scripts/ops.py restore --dump <文件.sql.gz> --database <库名> --confirm\n"
+            "默认只恢复到空库；覆盖已有库还要加 --force。"
+        )
+    dump = Path(args.dump)
+    if not dump.exists():
+        raise OpsError("找不到备份文件：" + str(dump))
+    database = validate_database_name(args.database, "恢复")
+    manifest = {}
+    manifest_path = dump.parent / "manifest.json"
+    if manifest_path.exists():
+        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+        expected = manifest.get("dump_sha256")
+        if expected and expected != sha256_of(dump):
+            raise OpsError("备份文件与 manifest 的 sha256 不一致，拒绝恢复。")
+        print("校验通过：备份文件与 manifest 的 sha256 一致。")
+    elif not args.skip_manifest_check:
+        raise OpsError("找不到 manifest.json，无法校验备份完整性（确认要继续时加 --skip-manifest-check）。")
+    if dump_targets_instead_of_source(dump):
+        raise OpsError(
+            "备份文件里带有 CREATE DATABASE / USE，恢复会写进文件里指定的库而不是目标库，拒绝执行。\n"
+            "请用本工具重新生成备份（它导出的是只含表数据的 dump）。"
+        )
+    # The source database must not change: its checksums are compared again afterwards.
+    source_database = manifest.get("database", "")
+    source_before = dump_checksums(source_database) if source_database and source_database != database and database_exists(source_database) else {}
+
+    ensure_empty_target(database, args.force)
+    print("正在恢复 " + dump.name + " 到 " + database + " …")
+    # The dump is decompressed to a temporary file and that file is handed to the client. Feeding
+    # a pipe from Python is unreliable on Windows for a stream this size, and a real file
+    # descriptor also means the client sees the plain SQL it expects.
+    work = Path(tempfile.mkdtemp(prefix="lexgo-restore-"))
+    plain = work / "dump.sql"
+    try:
+        with gzip.open(dump, "rb") as compressed, open(plain, "wb") as target:
+            shutil.copyfileobj(compressed, target)
+        with open(plain, "rb") as handle:
+            load = subprocess.run([mysql_binary("mysql"), *connection_args(database), "--default-character-set=utf8mb4"],
+                                  stdin=handle, env=client_env(), capture_output=True)
+        if load.returncode != 0:
+            message = load.stderr.decode("utf-8", "replace").strip().splitlines()
+            raise OpsError("恢复失败：" + (message[-1] if message else "未知错误"))
+    finally:
+        shutil.rmtree(work, ignore_errors=True)
+    print("恢复完成，开始校验。")
+    if source_before:
+        source_after = dump_checksums(source_database)
+        if source_after != source_before:
+            raise OpsError("源库 " + source_database + " 在校验过程中发生变化，已停止；请人工比对后再继续。")
+        print("源库 " + source_database + " 的内容校验和未变化。")
+    verify_args = argparse.Namespace(database=database, manifest=str(manifest_path) if manifest_path.exists() else "",
+                                     expect_schema=manifest.get("schema_version"), api="", user="", password_env="",
+                                     password_env_b="", skip_api=True)
+    return command_verify(verify_args)
+
+
+# ---------------------------------------------------------------- verify
+
+
+def integrity_checks(database):
+    checks = []
+
+    def add(name, ok, detail=""):
+        checks.append((name, ok, detail))
+
+    add("schema 表存在", int(scalar(database, "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='%s' AND TABLE_NAME='lexgo_schema'" % database, "0") or 0) == 1)
+    add("产品标识为 lexgo", scalar(database, "SELECT product FROM lexgo_schema WHERE id=1", "") == "lexgo")
+    add("没有孤立章节（章节属于其书）", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapters c LEFT JOIN lexgo_books b ON b.id=c.book_id WHERE b.id IS NULL", "0") or 0) == 0)
+    add("章节归属与书归属一致", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapters c JOIN lexgo_books b ON b.id=c.book_id WHERE c.owner_id<>b.owner_id", "0") or 0) == 0)
+    add("每个词条都有排期行", int(scalar(database, "SELECT COUNT(*) FROM lexgo_terms t LEFT JOIN lexgo_term_reviews r ON r.term_id=t.id WHERE r.term_id IS NULL", "0") or 0) == 0)
+    add("排期行都指向存在的词条", int(scalar(database, "SELECT COUNT(*) FROM lexgo_term_reviews r LEFT JOIN lexgo_terms t ON t.id=r.term_id WHERE t.id IS NULL", "0") or 0) == 0)
+    add("复习记录引用有效词条", int(scalar(database, "SELECT COUNT(*) FROM lexgo_review_answers a LEFT JOIN lexgo_terms t ON t.id=a.term_id WHERE t.id IS NULL", "0") or 0) == 0)
+    add("等级只出现在学习中词条", int(scalar(database, "SELECT COUNT(*) FROM lexgo_terms WHERE (status='learning' AND (level<1 OR level>7)) OR (status<>'learning' AND level<>0)", "0") or 0) == 0)
+    add("完成记录都指向存在的章节", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapter_progress p LEFT JOIN lexgo_chapters c ON c.id=p.chapter_id WHERE c.id IS NULL", "0") or 0) == 0)
+    add("词条语言与所属空间一致", int(scalar(database, "SELECT COUNT(*) FROM lexgo_terms t JOIN lexgo_spaces s ON s.owner_id=t.owner_id WHERE t.language<>s.language", "0") or 0) == 0)
+
+    # The audit tables must not gain a column that could hold a credential or private content.
+    for table in ("lexgo_login_logs", "lexgo_operation_logs"):
+        columns = {row[0].lower() for row in query("information_schema", "SELECT COLUMN_NAME FROM COLUMNS WHERE TABLE_SCHEMA='%s' AND TABLE_NAME='%s'" % (database, table))}
+        banned = sorted(columns & AUDIT_BANNED_COLUMNS)
+        add("审计表 " + table + " 不含敏感列", not banned, "命中：" + ",".join(banned) if banned else "列名白名单通过")
+    return checks
+
+
+class Api:
+    """A very small API client: the operations tool must not depend on the test harness."""
+
+    def __init__(self, base):
+        self.base = base.rstrip("/")
+
+    def call(self, method, path, token=None, body=None):
+        import urllib.error
+        import urllib.request
+
+        request = urllib.request.Request(self.base + "/api/v1/" + path, method=method)
+        if token:
+            request.add_header("Authorization", "Bearer " + token)
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            request.add_header("Content-Type", "application/json")
+        try:
+            with urllib.request.urlopen(request, payload, timeout=60) as response:
+                return response.status, json.loads(response.read().decode("utf-8"))
+        except urllib.error.HTTPError as error:
+            try:
+                return error.code, json.loads(error.read().decode("utf-8"))
+            except Exception:
+                return error.code, {}
+
+    def login(self, account, password):
+        status, payload = self.call("POST", "login", body={"username": account, "password": password})
+        return status, payload.get("data", {}).get("token", "")
+
+
+def report(checks):
+    """Prints one line per check and fails the command when any of them failed."""
+    failures = [name for name, ok, _ in checks if not ok]
+    for name, ok, detail in checks:
+        print(("  [ok]   " if ok else "  [fail] ") + name + ("— " + detail if detail else ""))
+    if failures:
+        print("\n检查未通过 " + str(len(failures)) + " 项：" + "、".join(failures))
+        raise SystemExit(1)
+    print("\n检查通过：" + str(len(checks)) + " 项全部成功。")
+    return 0
+
+
+def api_walk(api, user, password_a, password_b=None):
+    """Two accounts walk the learning loop on the restored instance and one must not see the other."""
+    checks = []
+    client = Api(api)
+
+    def add(name, ok, detail=""):
+        checks.append((name, ok, detail))
+
+    status, token_a = client.login(user + "_a", password_a)
+    add("账号 " + user + "_a 可以登录恢复实例", status == 200 and bool(token_a))
+    status, token_b = client.login(user + "_b", password_b or password_a)
+    add("账号 " + user + "_b 可以登录恢复实例", status == 200 and bool(token_b))
+    if not token_a or not token_b:
+        return checks
+
+    status, books_a = client.call("GET", "books", token_a)
+    items_a = books_a.get("data", {}).get("items", [])
+    add("账号 A 能看到自己的书库", status == 200 and len(items_a) > 0, str(len(items_a)) + " 本书")
+
+    status, progress_a = client.call("GET", "progress", token_a)
+    add("账号 A 能读到自己的进度", status == 200, json.dumps(progress_a.get("data", {}), ensure_ascii=False)[:120])
+
+    # Isolation: ids that belong to A must not resolve for B.
+    if items_a:
+        book_id = items_a[0]["id"]
+        status, _ = client.call("GET", "books/%d" % book_id, token_b)
+        add("B 读取 A 的书籍被拒绝", status == 404, "HTTP " + str(status))
+        status, detail = client.call("GET", "books/%d" % book_id, token_a)
+        chapters = [chapter for chapter in detail.get("data", {}).get("chapters", []) if chapter["status"] == "ready"]
+        if chapters:
+            chapter_id = chapters[0]["id"]
+            status, _ = client.call("GET", "chapters/%d" % chapter_id, token_b)
+            add("B 读取 A 的章节被拒绝", status == 404, "HTTP " + str(status))
+            # A completes the chapter twice: the mark must stay idempotent after a restore.
+            status, _ = client.call("POST", "chapters/%d/complete" % chapter_id, token_a)
+            status2, second = client.call("POST", "chapters/%d/complete" % chapter_id, token_a)
+            duplicate = second.get("data", {}).get("progress", {}).get("duplicate")
+            add("恢复后完成章节仍然幂等", status == 200 and status2 == 200 and bool(duplicate), "duplicate=" + str(duplicate))
+            status, _ = client.call("POST", "chapters/%d/complete" % chapter_id, token_b)
+            add("B 不能完成 A 的章节", status == 404, "HTTP " + str(status))
+
+    # The review loop works on restored schedules.
+    status, queue = client.call("GET", "reviews/queue", token_a)
+    items = queue.get("data", {}).get("items", [])
+    add("账号 A 的到期队列可读", status == 200, "到期 " + str(queue.get("data", {}).get("total")) + " 条")
+    if items:
+        item = items[0]
+        answer = {"answerId": "ops-verify-restore-0001", "grade": "correct", "expectedDueAt": item["dueAt"]}
+        status, result = client.call("POST", "reviews/%d/answers" % item["id"], token_a, answer)
+        add("恢复后答题成功并更新排期", status in (200, 201) and result.get("data", {}).get("result") == "applied",
+            "level " + str(result.get("data", {}).get("levelAfter")))
+        status, replay = client.call("POST", "reviews/%d/answers" % item["id"], token_a, answer)
+        add("重复提交同一答案不重复记账", status in (200, 201) and bool(replay.get("data", {}).get("duplicate")),
+            "duplicate=" + str(replay.get("data", {}).get("duplicate")))
+        status, _ = client.call("POST", "reviews/%d/answers" % item["id"], token_b,
+                                {"answerId": "ops-verify-restore-0002", "grade": "correct", "expectedDueAt": item["dueAt"]})
+        add("B 不能给 A 的词条答题", status == 404, "HTTP " + str(status))
+    return checks
+
+
+def command_smoke(args):
+    """Walks the whole learning loop on a freshly installed instance with two trial accounts."""
+    load_env()
+    client = Api(args.api)
+    password = os.environ.get(args.password_env, "")
+    if not password:
+        raise OpsError("环境变量 " + args.password_env + " 为空：请先为演练账号设置密码。")
+    checks = []
+
+    def add(name, ok, detail=""):
+        checks.append((name, ok, detail))
+
+    # 1. The first administrator exists and can log in: the installation is usable.
+    status, admin_token = client.login(args.admin_user, os.environ.get(args.admin_password_env, ""))
+    add("初始管理员可以登录", status == 200 and bool(admin_token))
+    if not admin_token:
+        return report(checks)
+
+    # 2. Two trial learners, created by the administrator. There is no self-registration.
+    for suffix in ("a", "b"):
+        account = args.user + "_" + suffix
+        status, payload = client.call("POST", "accounts", admin_token, {"username": account, "password": password})
+        if status == 201:
+            add("管理员创建演练账号 " + account, True, "HTTP 201")
+        else:
+            # An existing account is fine on a rerun; the password was set the first time.
+            existing = client.login(account, password)[0] == 200
+            add("演练账号 " + account + " 已存在且可登录", existing, payload.get("msg", "HTTP " + str(status)))
+
+    status, token_a = client.login(args.user + "_a", password)
+    status2, token_b = client.login(args.user + "_b", password)
+    add("两个演练账号都能登录", status == 200 and status2 == 200 and bool(token_a) and bool(token_b))
+    if not token_a or not token_b:
+        return report(checks)
+
+    add("新账号的学习空间是空的", client.call("GET", "books", token_a)[1].get("data", {}).get("items", []) == [])
+    add("新账号的进度从零开始", client.call("GET", "progress", token_a)[1].get("data", {}).get("totalChapters") == 0)
+
+    # 3. The learning loop on the new instance: paste, read, complete, save a word, review it.
+    text = "Curiosity opens the first door.\nThe second door stays closed.\n" * 6
+    status, pasted = client.call("POST", "books", token_a, {
+        "requestId": args.request_id, "title": "Fictional trial chapter", "text": text, "language": "en"})
+    add("演练账号可以粘贴章节", status == 201, "HTTP " + str(status))
+    if status != 201:
+        return report(checks)
+    book_id = pasted["data"]["book"]["id"]
+    chapter_id = pasted["data"]["chapter"]["id"]
+    ready = {}
+    for _ in range(40):
+        status, detail = client.call("GET", "chapters/%d" % chapter_id, token_a)
+        ready = detail.get("data", {}).get("chapter", {})
+        if ready.get("status") == "ready":
+            break
+        time.sleep(0.5)
+    add("章节在干净实例上处理完成", ready.get("status") == "ready", str(ready.get("status")))
+    if ready.get("status") != "ready":
+        return report(checks)
+
+    status, tokens = client.call("GET", "chapters/%d/tokens" % chapter_id, token_a)
+    words = [token for token in tokens.get("data", {}).get("tokens", []) if token.get("kind") == "word"]
+    add("章节可以分词并返回可点选的词", status == 200 and len(words) > 3, str(len(words)) + " 个词")
+
+    status, _ = client.call("GET", "terms/lookup?query=curiosity", token_a)
+    add("词典查询返回明确状态（含资源缺失时的降级）", status in (200, 404, 503), "lookup HTTP " + str(status))
+
+    if words:
+        word = words[0]
+        status, saved = client.call("POST", "terms", token_a, {
+            "chapterId": chapter_id, "start": word["start"], "end": word["end"],
+            "definition": "虚构释义", "status": "new"})
+        add("演练账号可以保存自己的词义", status in (200, 201), "HTTP " + str(status))
+        term_id = saved.get("data", {}).get("term", {}).get("id")
+        if term_id:
+            other_terms = client.call("GET", "terms?query=curiosity", token_b)[1].get("data", {}).get("total")
+            add("另一个账号看不到该词条", other_terms == 0, "B 查到 " + str(other_terms) + " 条")
+            queue = client.call("GET", "reviews/queue", token_a)[1]
+            items = queue.get("data", {}).get("items", [])
+            due = [entry for entry in items if entry["id"] == term_id]
+            add("新保存的词条立刻到期", bool(due), "到期 " + str(queue.get("data", {}).get("total")))
+            if due:
+                answer = {"answerId": args.answer_id, "grade": "correct", "expectedDueAt": due[0]["dueAt"]}
+                status, result = client.call("POST", "reviews/%d/answers" % term_id, token_a, answer)
+                add("到期复习可以作答", status in (200, 201) and result.get("data", {}).get("result") == "applied",
+                    "level " + str(result.get("data", {}).get("levelAfter")))
+                replay = client.call("POST", "reviews/%d/answers" % term_id, token_a, answer)[1]
+                add("重复作答不重复记账", replay.get("data", {}).get("duplicate") is True)
+
+    status, completed = client.call("POST", "chapters/%d/complete" % chapter_id, token_a)
+    add("完成章节只记已读", status == 200 and completed.get("data", {}).get("progress", {}).get("read") is True)
+
+    data = client.call("GET", "progress", token_a)[1].get("data", {})
+    add("进度反映新实例上的活动", data.get("readChapters", 0) >= 1 and data.get("savedTerms", 0) >= 1,
+        json.dumps({key: data.get(key) for key in ("readChapters", "totalChapters", "knownTerms", "learningTerms", "newTerms", "dueNow")}, ensure_ascii=False))
+
+    add("B 无法读取 A 的书籍", client.call("GET", "books/%d" % book_id, token_b)[0] == 404)
+    add("B 的书库仍然为空", client.call("GET", "books", token_b)[1].get("data", {}).get("items", []) == [])
+    return report(checks)
+
+
+def command_verify(args):
+    load_env()
+    database = validate_database_name(args.database or os.environ.get("LEXGO_DB_NAME", ""), "校验")
+    failures = []
+    print("校验数据库 " + database)
+    if not database_exists(database):
+        raise OpsError("数据库不存在：" + database)
+    version = int(scalar(database, "SELECT version FROM lexgo_schema WHERE id=1", "0") or 0)
+    if version != SCHEMA_VERSION:
+        print("  [warn] schema 版本 " + str(version) + "，本工具期望 " + str(SCHEMA_VERSION) + "（升级或回退阶段属正常）")
+
+    checks = integrity_checks(database)
+    if args.manifest:
+        manifest = json.loads(Path(args.manifest).read_text(encoding="utf-8"))
+        actual = row_counts(database)
+        for table, expected in manifest.get("row_counts", {}).items():
+            got = actual.get(table, 0)
+            checks.append(("表 " + table + " 行数与备份一致", got == expected, "备份 " + str(expected) + " / 现在 " + str(got)))
+    for name, ok, detail in checks:
+        print(("  [ok]   " if ok else "  [fail] ") + name + ("— " + detail if detail else ""))
+        if not ok:
+            failures.append(name)
+
+    if args.api:
+        print("校验接口 " + args.api + "（账号前缀 " + args.user + "）")
+        password = os.environ.get(args.password_env, "")
+        if not password:
+            raise OpsError("环境变量 " + args.password_env + " 为空，无法登录演练账号。")
+        password_b = os.environ.get(args.password_env_b, "") or password
+        for name, ok, detail in api_walk(args.api, args.user, password, password_b):
+            print(("  [ok]   " if ok else "  [fail] ") + name + ("— " + detail if detail else ""))
+            if not ok:
+                failures.append(name)
+
+    if failures:
+        print("\n校验未通过 " + str(len(failures)) + " 项：" + "、".join(failures))
+        raise SystemExit(1)
+    print("\n校验通过。")
+    return 0
+
+
+def build_parser():
+    parser = argparse.ArgumentParser(description="LexGo 运维工具：依赖检查、备份、恢复与校验")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    install = sub.add_parser("install-check", help="只读检查依赖与资源版本")
+    install.set_defaults(func=command_install_check)
+
+    init = sub.add_parser("init-database", help="为全新安装创建空库并提示所需权限")
+    init.add_argument("--database", default="", help="要创建的库名，默认取 LEXGO_DB_NAME")
+    init.set_defaults(func=command_init_database)
+
+    backup = sub.add_parser("backup", help="导出全库并写出 manifest")
+    backup.add_argument("--out", default="", help="输出目录，默认 .local/backups/<时间戳>")
+    backup.add_argument("--database", default="", help="要备份的库，默认取 LEXGO_DB_NAME")
+    backup.add_argument("--force", action="store_true", help="输出目录非空时覆盖")
+    backup.set_defaults(func=command_backup)
+
+    restore = sub.add_parser("restore", help="把备份恢复到空库（覆盖已有库要 --force）")
+    restore.add_argument("--dump", required=True, help="backup 生成的 .sql.gz")
+    restore.add_argument("--database", required=True, help="目标库名，必须包含 lexgo")
+    restore.add_argument("--confirm", action="store_true", help="确认执行写入")
+    restore.add_argument("--force", action="store_true", help="目标库已有数据时覆盖")
+    restore.add_argument("--skip-manifest-check", action="store_true", help="没有 manifest 时跳过完整性校验")
+    restore.set_defaults(func=command_restore)
+
+    verify = sub.add_parser("verify", help="校验恢复结果的完整性与两账号隔离")
+    verify.add_argument("--database", default="", help="要校验的库，默认取 LEXGO_DB_NAME")
+    verify.add_argument("--manifest", default="", help="与备份的行数逐表比对")
+    verify.add_argument("--expect-schema", type=int, default=None)
+    verify.add_argument("--api", default="", help="例如 http://127.0.0.1:8010 ，启用两账号闭环校验")
+    verify.add_argument("--user", default="", help="演练账号前缀，实际账号为 <前缀>_a 与 <前缀>_b")
+    verify.add_argument("--password-env", default="LEXGO_TRIAL_PASSWORD", help="存放演练账号 _a 密码的环境变量名")
+    verify.add_argument("--password-env-b", default="", help="存放演练账号 _b 密码的环境变量名，默认与 _a 相同")
+    verify.add_argument("--skip-api", action="store_true")
+    verify.set_defaults(func=command_verify)
+
+    smoke = sub.add_parser("smoke", help="在干净实例上创建两个演练账号并走通学习闭环")
+    smoke.add_argument("--api", required=True, help="例如 http://127.0.0.1:8010")
+    smoke.add_argument("--admin-user", required=True, help="初始管理员账号")
+    smoke.add_argument("--admin-password-env", default="LEXGO_BOOTSTRAP_PASSWORD", help="初始管理员密码所在的环境变量名")
+    smoke.add_argument("--user", required=True, help="演练账号前缀，实际账号为 <前缀>_a 与 <前缀>_b")
+    smoke.add_argument("--password-env", default="LEXGO_TRIAL_PASSWORD", help="演练账号密码所在的环境变量名")
+    smoke.add_argument("--request-id", default="ops-smoke-0001", help="粘贴章节的幂等请求号")
+    smoke.add_argument("--answer-id", default="ops-smoke-answer-0001", help="复习作答的幂等编号")
+    smoke.set_defaults(func=command_smoke)
+    return parser
+
+
+def main(argv=None):
+    args = build_parser().parse_args(argv)
+    try:
+        return args.func(args)
+    except OpsError as error:
+        print("错误：" + str(error), file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_lexgo_ops.py

@@ -0,0 +1,90 @@
+"""Unit tests for the operations helper rules that must hold without a database.
+
+The database paths (backup, restore, verify) are exercised by the #15 drill against real MySQL;
+what is checked here is the logic that decides whether an operation is allowed at all.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import ops  # noqa: E402
+
+
+class VersionParsingTests(unittest.TestCase):
+    def test_reads_the_version_after_distrib(self):
+        # A client banner's first number is the protocol version, not the server version.
+        self.assertEqual(ops.version_of("mysql  Ver 14.14 Distrib 5.7.38, for Win64 (x86_64)"), (5, 7, 38))
+        self.assertEqual(ops.version_of("mysql  Ver 8.4.3 for Win64 on x86_64"), (8, 4, 3))
+        self.assertIsNone(ops.version_of(""))
+        self.assertIsNone(ops.version_of("no version here"))
+
+    def test_compares_client_and_server_versions(self):
+        self.assertLess(ops.version_of("Ver 14.14 Distrib 5.7.38"), ops.version_of("8.4.3"))
+        self.assertGreaterEqual(ops.version_of("Ver 8.0.36"), ops.version_of("8.4.3".replace("8.4.3", "8.0.36")))
+        self.assertGreaterEqual(ops.version_of("8.4.3"), ops.version_of("8.4.3"))
+
+
+class DatabaseNameTests(unittest.TestCase):
+    def test_accepts_lexgo_names(self):
+        for name in ("lexgo_dev", "lexgo_prod", "lexgo_restore_drill"):
+            self.assertEqual(ops.validate_database_name(name, "测试"), name)
+
+    def test_refuses_system_and_unrelated_names(self):
+        for name in ("mysql", "information_schema", "performance_schema", "sys", "app_prod", ""):
+            with self.assertRaises(SystemExit):
+                ops.validate_database_name(name, "测试")
+
+
+class DumpSafetyTests(unittest.TestCase):
+    """A dump that carries CREATE DATABASE/USE would restore into the wrong schema."""
+
+    def _write_dump(self, body):
+        import gzip
+        import tempfile
+
+        handle = tempfile.NamedTemporaryFile(delete=False, suffix=".sql.gz")
+        handle.close()
+        with gzip.open(handle.name, "wb") as target:
+            target.write(body.encode("utf-8"))
+        self.addCleanup(Path(handle.name).unlink)
+        return Path(handle.name)
+
+    def test_flags_a_dump_that_switches_database(self):
+        dump = self._write_dump("-- MySQL dump\nCREATE DATABASE /*!32312 IF NOT EXISTS*/ `lexgo_dev`;\nUSE `lexgo_dev`;\nCREATE TABLE t (id INT);\n")
+        self.assertTrue(ops.dump_targets_instead_of_source(dump))
+
+    def test_flags_a_dump_with_use_only(self):
+        dump = self._write_dump("-- MySQL dump\nUSE `lexgo_dev`;\nCREATE TABLE t (id INT);\n")
+        self.assertTrue(ops.dump_targets_instead_of_source(dump))
+
+    def test_accepts_a_table_only_dump(self):
+        dump = self._write_dump("-- MySQL dump\nDROP TABLE IF EXISTS `lexgo_terms`;\nCREATE TABLE `lexgo_terms` (id INT);\nINSERT INTO `lexgo_terms` VALUES (1);\n")
+        self.assertFalse(ops.dump_targets_instead_of_source(dump))
+
+
+class ManifestTests(unittest.TestCase):
+    def test_manifest_keys_never_name_a_credential(self):
+        """The manifest is written to disk, so its field names are checked, not its prose."""
+        import re
+
+        source = (ROOT / "scripts" / "ops.py").read_text(encoding="utf-8")
+        block = source[source.index("    manifest = {"):source.index('(out_dir / "manifest.json")')]
+        keys = re.findall(r'^\s+"([a-z_]+)":', block, re.M)
+        self.assertIn("schema_version", keys)
+        self.assertIn("row_counts", keys)
+        for key in keys:
+            for banned in ("password", "secret", "token", "credential"):
+                self.assertNotIn(banned, key.lower(), key)
+
+    def test_audit_tables_declare_no_sensitive_column(self):
+        """The integrity check must keep naming the columns that may not appear in the audit logs."""
+        self.assertTrue({"password", "token", "body", "content"} <= ops.AUDIT_BANNED_COLUMNS)
+        self.assertIn("lexgo_login_logs", ops.TABLES)
+        self.assertIn("lexgo_operation_logs", ops.TABLES)
+
+
+if __name__ == "__main__":
+    unittest.main()