HEL-899: custom voice via reference audio on the unified API — reuse .12 panel zero-shot registry (server-side proxy, cg- namespaced IDs never collide with the 11 builtin voices, panel stays the only voices.json writer) + CosyVoice fresh-synthesis preview (unknown voice pre-checked: no silent default fallback) + persisted selection/manifest/usage journal in instance data space with atomic writes and full rollback; session-open captures the effective voice (selection if valid else worker-file value else default, D1 domain untouched when user never chose) pinned per call, mid-call changes land at next call at the earliest; explicit allow_asr gates FunASR transcription, preview/ASR journaled, preview writes no chat/memory/receipt rows; zero new deps (raw-bytes upload endpoint), zero real calls in tests: 18 new cases (formats/size/duration refusals, collision+concurrency, panel down/reject/lying, tts down, rollback, restart persistence, call pinning, identity/shared-state untouched) + offline chain script 12/12, suite 370; req governance v1.12: CG-010 boundary updated to custom-voice-first per user comment 01a17de1 (verbatim evidence + changes + manifest), checks pass vs origin/main and base 8dc84f8 with handoff-HEL-899

Co-authored-by: multica-agent <github@multica.ai>
2026-10-27 23:34:00 +08:00
19 changed files with 1692 additions and 27 deletions
@@ -1,3 +1,4 @@
 __pycache__/
 *.pyc
 web/node_modules
+.venv311/
@@ -25,6 +25,17 @@ agent_name: cybergril-voice-agent
 # CosyVoice TTS 健康探测（TTS 不在就诚实拒绝开通话，不假装能说话）
 tts_health_url: http://127.0.0.1:9880/health

+# ---- 自定义音色（HEL-899）----
+# .12 语音面板（8090，https 自签）：音色列表/zero-shot 登记的唯一写入者，
+# CyberGril 统一 API 服务端代理调用；浏览器永不直连，凭据不外发。
+panel_url: https://127.0.0.1:8090
+# CosyVoice 合成入口（试听现合成；/health 的热加载名单用于通话开启时校验）
+tts_url: http://127.0.0.1:9880
+# 共享 voices.json 只读回看（补全 ASR 自动转写的参考文本，供备份清单）；不配则留空
+registry_file: /opt/voice-agent/cosyvoice/voices.json
+# CyberGril 专属 worker 的音色状态文件（worker 会话入口读一次；通话中改选最迟下一通话生效）
+worker_state_file: /opt/voice-agent-cybergril/voice_state_cybergril.json
+
 # 浏览器接入 token 有效期（秒）：只够加入房间，短时限
 token_ttl_seconds: 300
 # 会话登记有效期（秒）：房间早空/挂断后登记自然过期
@@ -1,6 +1,6 @@
 {
  "schema_version": 1,
-  "version": "1.11",
+  "version": "1.12",
  "approval_comment_id": "01a1734c-1159-7a44-a81f-3105f1b8d98d",
  "owner_id": "e5c6c938-db48-4ff9-954f-4fb8b063a132",
  "scope": "完整要求；v1.11在v1.10的59项上新增成熟方案优先复用执行约束，原产品与验收边界保留；不代表已选定底座、授权迁移或功能通过。",
@@ -118,10 +118,11 @@
      "category": "产品",
      "topic": "连续语音与文字",
      "conclusion": "用户主动开启语音后自然连续交谈，支持开口打断她；结束语音后停止收音，同时保留文字聊天。",
-      "boundary": "响应速度、回声消除和打断效果需用现有语音服务实测，未承诺固定毫秒数。",
+      "boundary": "响应速度、回声消除和打断效果需用现有语音服务实测，未承诺固定毫秒数。音色入口以可自定义音色为准：在网页上传或录制参考音频→试听→应用到同一位她的后续通话，直接复用已装TTS的零样本克隆；不以仅选预设音色替代，先前参考包试听方案不采用。本轮通话测试授权总管自定音色、最长2小时/3次通话；自定义音色交付不等于用户已验收音色，也不改变换声音只换发声、不重建伴侣、不改出生底色/名字/关系/聊天与记忆的边界。",
      "source_comment_ids": [
        "01a15750-2cf9-7f23-9269-0f150d2c5e61",
-        "01a15750-e061-733b-9d80-38459d306b97"
+        "01a15750-e061-733b-9d80-38459d306b97",
+        "01a17de1-2892-78be-a422-eb6a11ffebfd"
      ]
    },
    {
@@ -108,6 +108,16 @@
        "GOV-006"
      ],
      "impact": "原59项保持；不预设保留自研底座，也不指定任何最终开源项目；未授权产品迁移、依赖安装、现网部署或改变资料/隐私边界。"
+    },
+    {
+      "version": "1.12",
+      "approval_comment_id": "01a17de1-2892-78be-a422-eb6a11ffebfd",
+      "approved_by": "e5c6c938-db48-4ff9-954f-4fb8b063a132",
+      "summary": "CG-010音色边界更新：以可自定义音色为准（网页上传/录制参考音频→试听→应用），直接复用已装TTS零样本克隆；不以仅选预设替代，参考包试听方案不采用；本轮通话测试授权总管自定音色、最长2小时/3次通话。",
+      "affected_ids": [
+        "CG-010"
+      ],
+      "impact": "只改音色入口与测试授权；换声音只换发声、不重建伴侣、不改底色/名字/关系/聊天/记忆的既有边界不变；交付与阶段通过不等于用户已验收音色；不新增付费服务、不改变部署与资料边界。"
    }
  ]
 }
@@ -39,3 +39,13 @@ v1.7：把原始31项、已确认视觉和后续用户裁定整理到同一入
 > 记住我的原则，能借鉴成熟方案的就直接借鉴，从零创造的难度太大需要的是一个专业的团队，不适用于我这个不专业的人加上没有决策性的AI的组合；另外我之前让你看的我跟另外智能体的交流内容希望你提取值得借鉴的地方

 旧记录未明确规定通用内核必须优先复用成熟实现；现新增GOV-006，减少从零研发和长期自行维护，不因沉没工作量而默认保留自研内核。原59项不删减。交流材料提炼为候选机制，具体技术栈、底座、情绪映射及迁移范围未获批；不将本次文档记录当功能通过或现网施工授权。
+
+## v1.12：音色入口定为可自定义音色
+
+来源评论：`01a17de1-2892-78be-a422-eb6a11ffebfd`（2026-10-27 15:01:10 UTC，回应总管阻塞通报 `01a17d45-f14e-7b62-9850-646e040c7318`）。白栖知原话：
+
+> .12已恢复，另外以上一条让我解压附件包试听，我不明白是什么意思，是让我听预设音色吗？如果是我觉得你的设计不够合理，我想要的是可以自定义音色，我的.12上安装的TTS是支持自定义音色的。至于最长2小时，3次通话的测试，我同意，音色你自己决定就行
+
+旧边界（CG-010/D1 路线）：音色按"先在现有中文女声里挑、都不满意再做零样本克隆"的条件路线推进，本轮以参考包编号试听为交付形式。新要求：**音色入口以可自定义音色为准**——在网页上传或录制参考音频→试听→应用到同一位她的后续通话，直接复用已装TTS的零样本克隆；不以仅选预设音色替代，参考包试听方案不采用。另明确：本轮通话测试授权总管自定音色、最长2小时/3次通话（操作授权，记 HEL-879 测试执行，不进需求边界）。
+
+影响范围：仅 CG-010 音色入口与测试授权；换声音只换发声、不重建伴侣、不改出生底色/名字/关系/聊天/记忆的既有边界不变；CG-010 其余条文与原59项不动。工程交付或测试通过不等于用户已验收音色，也不得把"用户已授权自定义音色"改写成"用户已选定某支音色"。原话记录见 `sources/comment-evidence-custom-voice-20261027.json`（施工员2号按平台原话逐字核对录入）。
@@ -1,5 +1,5 @@
 {
-  "baseline_version": "1.11",
+  "baseline_version": "1.12",
  "items": [
    {
      "requirement_id": "CG-001",
@@ -1,9 +1,11 @@
-# CyberGril 完整当前需求 · v1.11
+# CyberGril 完整当前需求 · v1.12

 完整要求；版本1.10在v1.9（交流静图认可）之上，把白栖知已体验的对坐动态小样 v2 整体效果（演示层面）记入 VIS-011 与 CG-004 边界；不证明其余页面、真实功能或完整产品已验收。

 v1.11新增GOV-006成熟方案优先复用；原59项与v1.10阶段验收边界保持。

+v1.12按白栖知2026-10-27原话（`01a17de1-2892-78be-a422-eb6a11ffebfd`）把CG-010音色边界更新为可自定义音色：网页上传/录制参考音频→试听→应用到同一位她的后续通话，直接复用已装TTS零样本克隆，不以仅选预设替代；先前参考包试听方案不采用。本轮通话测试授权总管自定音色、最长2小时/3次通话。该项交付不等于用户已验收音色。
+
 ## 阅读与效力

 用户明确的新决定覆盖旧决定并记录来源；完整当前记录优先于任务摘录、技术方案和旧仓库 SPEC。无明确用户裁定的冲突先查来源，不擅自取舍。
@@ -84,9 +86,9 @@ v1.11新增GOV-006成熟方案优先复用；原59项与v1.10阶段验收边界

 用户主动开启语音后自然连续交谈，支持开口打断她；结束语音后停止收音，同时保留文字聊天。

-边界：响应速度、回声消除和打断效果需用现有语音服务实测，未承诺固定毫秒数。
+边界：响应速度、回声消除和打断效果需用现有语音服务实测，未承诺固定毫秒数。音色入口以可自定义音色为准：在网页上传或录制参考音频→试听→应用到同一位她的后续通话，直接复用已装TTS的零样本克隆；不以仅选预设音色替代，先前参考包试听方案不采用。本轮通话测试授权总管自定音色、最长2小时/3次通话；自定义音色交付不等于用户已验收音色，也不改变换声音只换发声、不重建伴侣、不改出生底色/名字/关系/聊天与记忆的边界。

-来源评论：`01a15750-2cf9-7f23-9269-0f150d2c5e61`、`01a15750-e061-733b-9d80-38459d306b97`。
+来源评论：`01a15750-2cf9-7f23-9269-0f150d2c5e61`、`01a15750-e061-733b-9d80-38459d306b97`、`01a17de1-2892-78be-a422-eb6a11ffebfd`。

 ## CG-011 · 主动联系范围

@@ -0,0 +1,57 @@
+{
+  "baseline_version": "1.12",
+  "read_commit": "8dc84f88e9d5e961e17b246f24c23eaf526b012e",
+  "task": "HEL-899：把现有 CosyVoice 参考音频零样本克隆能力接入 CyberGril 统一 API（网页上传/录制参考音频→登记 cg- 专属 ID→试听现合成→应用到同一位她的后续通话）；直接复用 .12 语音面板 8090 的 zero-shot 注册与 CosyVoice 热加载、9880 /tts 合成，不训练新模型、不重造克隆引擎、不自建第二写者；本卡只做后台/协议与零真实调用自测，前端按合同接视觉总监，真实试听放统一审核",
+  "requirement_ids": [
+    "CG-006",
+    "CG-009",
+    "CG-010",
+    "CG-013",
+    "CG-014",
+    "CG-015",
+    "CG-016",
+    "CG-018",
+    "CG-019",
+    "CG-020",
+    "CG-021",
+    "AM-001",
+    "GOV-001",
+    "GOV-002",
+    "GOV-003",
+    "GOV-004",
+    "GOV-005",
+    "GOV-006",
+    "DEL-001"
+  ],
+  "deliverable": "src/cybergril/voice/custom.py（CustomVoiceService：8090 面板代理=共享 voices.json 唯一写者、cg-+8hex 服务端生成 ID 不撞既有 11 支、WAV 头服务端复验 RIFF/WAVE≤15MB≥2s、显式 allow_asr 才走 FunASR 转写、试听先验存在拦住 CosyVoice 未知音色静默回落 default、原子落盘+失败整体回滚、选择/元数据/资源流水持久在实例资料空间重启读回、备份清单 manifest）；router.py 五个新端点（GET/POST /api/voice/voices、POST /api/voice/preview、GET/POST /api/voice/selection）+ /api/voice/session 响应新增 voice=本次通话捕获音色（选择失效诚实回落 default；未选择不代写 worker 文件不替 D1 定音色）；session.py 会话记录 voice 字段+/health 名单读回；worker 语义不变（入口读一次→通话中改选最迟下一通话生效、绝不半句换人）；契约文档 docs/voice-custom-contract-HEL899.md（前端可直接实现的请求/响应/错误/录音格式/部署与回滚）；config/voice.example.yaml 增补（panel_url/tts_url/registry_file/worker_state_file，env 可覆盖）；零新依赖（登记端点原始字节体+查询串，特意不引 multipart 库）；tests/test_voice_custom.py 18 项 + scripts/verify_hel899_custom_voice.py 12/12；治理：v1.12（CG-010 边界更新，来源评论 01a17de1-2892-78be-a422-eb6a11ffebfd 逐字录入 sources/comment-evidence-custom-voice-20261027.json + manifest/baseline/coverage/changes 同步）",
+  "acceptance": [
+    "完整最小合同成立：列出可用声音（builtin 11 支+custom 标记、选择/试听状态、备份清单字段）、登记参考音频（服务端生成 cg- 专属 ID，不可能撞名覆盖既有 11 支，撞名重试不覆盖）、试听生成（每次现合成 WAV 24k/mono、无缓存不拿旧音频冒充）、读取/应用专属音色（换声音只换发声）全部经统一 API，浏览器不直连 8090/9880，凭据不下发",
+    "当前 session/chat 业务不分叉：worker/统一聊天链零改动；通话开启捕获有效音色并写入会话记录与 worker 状态文件，通话中改选不影响本通话、最迟下一通话生效（测试断言会话 voice 定住）、不打断或误认原轮次/receipt（receipt 链路未动）",
+    "原子与失败诚实：登记失败（面板 400/5xx/断连/谎报成功）不留下半条可选项、无元数据残留；应用失败整体回滚（选择状态与 worker 文件都回旧值）；服务不可用一律 503 中文文案；ASR 仅在 allow_asr=true 显式准许且参考文字为空时发生（面板 FunASR，非 LLM），每次记 register_asr/preview_tts 流水可核算",
+    "试听不写聊天/记忆/生活/回执（四表计数零变化测试断言）；自定义音色未试听不得应用（409），内置 D1 候选不强制；未知音色 404 且 TTS 零调用（前置拦住 CosyVoice 静默回落 default）；未知选择通话开启诚实回落 default 不冒充",
+    "持久化与身份不变：选择/元数据/试听凭据重启读回（重建服务同批文件测试）；换声音不改 companion/bazi_core/chat_message（逐字节对比测试）；通用助手 voice_state/persona_state/既有音色零写入零删除（桩端点计数零）；无人经 API 选择时不代写 worker 文件（D1 域不替用户定）",
+    "零真实调用：全链 fake 面板/TTS/LiveKit 传输，零真实 LLM/嵌入/ASR/TTS、零模型下载/安装、零共享音色实写、零部署、零现网触碰、未访问 4080、未动 HEL-879 现场"
+  ],
+  "deferred": [
+    "真实 CosyVoice 试听/真实 FunASR 转写/克隆相似度：统一审核卡在获批环境实测，复用合法剩余额度，不另开账绕 HEL-879 上限（最长2小时/3次通话授权属总管测试执行，不在本卡）",
+    "前端入口（上传/录音/试听/应用 UI 与入口呈现）：视觉总监按 docs/voice-custom-contract-HEL899.md §2 契约接线；本卡未动 web/",
+    "自定义音色删除与完整备份/恢复（CG-009/R6）：本卡最小合同不含删除（失败/多余登记不影响使用），备份清单字段已就位，恢复=按清单重新登记（面板无导出接口）",
+    "D1 最终音色决定仍归白栖知/总管；本轮测试音色已授权总管自定",
+    "R5 整体真机验收与 R1–R7 完整要求继续承接；自定义音色交付不等于 R5 完成或整产品完成"
+  ],
+  "requirement_changes": [
+    {
+      "summary": "v1.12：CG-010 音色边界更新为可自定义音色（网页上传/录制参考音频→试听→应用，直接复用已装TTS零样本克隆，不以仅选预设替代，参考包试听方案不采用）；影响仅 CG-010 音色入口与测试授权，换声音只换发声等既有边界不变",
+      "approval_comment_id": "01a17de1-2892-78be-a422-eb6a11ffebfd",
+      "record": "sources/comment-evidence-custom-voice-20261027.json + changes.json/changes.md v1.12 + baseline.json/coverage.json/current.md 同步"
+    }
+  ],
+  "verification": [
+    "PYTHONPATH=src .venv311/bin/python -m pytest tests -q：370 通过（基线 8dc84f8 全量 352 + 本卡 18：tests/test_voice_custom.py，全 fake 面板/TTS/LiveKit httpx.MockTransport + 临时文件 + 隔离 registry，零网络零真实模型调用）",
+    "scripts/verify_hel899_custom_voice.py：12/12 全过（真实 app+router+service，fake 传输：列表→无准许拒ASR→显式登记→未试听409→试听现合成→应用→会话捕获→通话中改选定住本通话→下一通话生效→capabilities.voice=false→重启读回选择/元数据/试听凭据）",
+    "ruff check src tests scripts deploy：本卡代码 0 错（既有 deploy/hel835_guard.py 的 SIM115 遗留不变，非本卡）",
+    "python3 scripts/check_requirements.py --base-ref origin/main 与 --base-ref 8dc84f8（实际基准）：含本 handoff 通过（v1.12 升版+新来源+变更记录合法，历史来源与变更未改写）",
+    ".12 只读取证（SSH 零写入零服务触碰）：/opt/voice-agent/web/server.py 实读核对 upload 校验（名字规则/15MB/WAV magic/32000字节/空参考文本→FunASR 转写/失败删 wav/voices.json 原子写）、/opt/voice-agent/cosyvoice/server.py 实读核对 mtime 热加载与 /tts 未知音色静默回落 default、voices.json 11 支音色清单——契约按实际源码写，不按记忆",
+    "未验证如实声明：真实 CosyVoice 合成与 FunASR 转写未调用（TTS 9880 停机中且本卡零真实调用）；浏览器端录音上传未实现（前端合同已交）；.12 未部署、未改任何现网配置/端口/数据"
+  ]
+}
@@ -15,5 +15,6 @@
  "comment-evidence-mature-reuse-20261025.json": "38ca076e26477e89cd161eaadb0fb00b0ee7bb5a26de2bd6ab41de52faaae79e",
  "mature-reuse-principle-20261025.md": "a2a75e071d3af5db722727bfaf8083bb60d07dcdd74e90d1e1374c1aafd2064d",
  "other-agent-discussion-20261025.txt": "42cd97029d73eace110e49e0566343dfeb25fe4f0633a10b78b126289938aaeb",
-  "comment-evidence-full-experience-20261026.json": "75dd55b8755bdad217b44c3f516c9372b046aae06dff511527014c7b209eccff"
+  "comment-evidence-full-experience-20261026.json": "75dd55b8755bdad217b44c3f516c9372b046aae06dff511527014c7b209eccff",
+  "comment-evidence-custom-voice-20261027.json": "ed72bd78218e1c0b4ca644bf933d6bdb2e3aa8cd2c0333f3410f6b60845bd3cb"
 }
@@ -0,0 +1,10 @@
+[
+  {
+    "id": "01a17de1-2892-78be-a422-eb6a11ffebfd",
+    "author_type": "member",
+    "author_id": "e5c6c938-db48-4ff9-954f-4fb8b063a132",
+    "content": ".12已恢复，另外以上一条让我解压附件包试听，我不明白是什么意思，是让我听预设音色吗？如果是我觉得你的设计不够合理，我想要的是可以自定义音色，我的.12上安装的TTS是支持自定义音色的。至于最长2小时，3次通话的测试，我同意，音色你自己决定就行",
+    "created_at": "2026-10-27T15:01:10Z",
+    "parent_id": "01a17d45-f14e-7b62-9850-646e040c7318"
+  }
+]
@@ -0,0 +1,205 @@
+# 自定义音色契约与部署说明（HEL-899）
+
+> 性质：**产品后端代码 + 前端可直接实现的请求/响应/错误/录音格式契约**。本卡零真实
+> LLM/嵌入/ASR/TTS 调用、零模型下载/安装、零共享音色实写、零部署；全部验证用
+> fake 面板/TTS 传输完成（见 §5）。基准 v1.11 → 本卡按治理机制升 v1.12
+> （CG-010 边界更新，来源评论 `01a17de1-2892-78be-a422-eb6a11ffebfd`）。
+> 读取提交：`8dc84f88e9d5e961e17b246f24c23eaf526b012e`（hel-879-assembly tip）。
+
+## 0. 用户原话（需求来源，2026-10-27）
+
+> ".12已恢复，另外以上一条让我解压附件包试听，我不明白是什么意思，是让我听预设音色吗？
+> 如果是我觉得你的设计不够合理，我想要的是可以自定义音色，我的.12上安装的TTS是支持
+> 自定义音色的。至于最长2小时，3次通话的测试，我同意，音色你自己决定就行"
+
+要点：① 自定义音色（不是仅选预设）；② 直接复用 .12 已装 TTS 的克隆能力；
+③ 本轮通话测试音色总管自定、最长 2 小时/3 次通话（授权给总管，不在本卡执行）。
+
+## 1. 能力边界（复用，不自建）
+
+- **登记**：复用 .12 语音面板（8090）`POST /api/voices/upload` 的 zero-shot 注册
+  （写共享 `voices.json` + CosyVoice mtime 热加载；参考文本留空时面板用 FunASR 转写）。
+  CyberGril 统一 API 是它的**服务端代理**：浏览器不直连 8090，凭据不进浏览器，
+  共享 `voices.json` 只有 8090 一个写者（无跨进程读-改-写竞态）。
+- **试听**：复用 CosyVoice（9880）`POST /tts`（`{"text","voice"}` → 24k mono int16
+  PCM 流）；API 现场包成 WAV 返回。每次试听都是新合成，**无缓存**。
+- **应用**：写 CyberGril 专属 worker 的 `voice_state_cybergril.json`（原子写）；
+  worker 在**会话入口读一次**（HEL-877 既有行为，未改）。
+- **不自建**克隆引擎、不训练模型、不改 CosyVoice/8090 任何代码与配置。
+
+### 共享注册表涉及的操作（全部经 8090，本服务不直接写共享文件）
+
+| 操作 | 途径 | 性质 |
+| --- | --- | --- |
+| 列表 | `GET /api/voices` | 只读 |
+| 登记 | `POST /api/voices/upload`（服务端生成 `cg-` 专属 ID） | 唯一写入，只新增 |
+| 参考文本回看 | 只读 `voices.json`（部署配置路径），补全 ASR 转写文本供备份清单 | 只读 |
+
+通用助手自己的 `voice_state.json`、`persona_state.json`、既有 11 支音色条目
+（bailu/bajie/default/female_soft/haimianbaobao/jiaojiao/male_warm/nezha/rain/
+sunwukong/xiaoyun）**从不触碰**；也不通过改面板全局默认偷换她的声音。
+
+### 命名空间
+
+- 新增音色 ID = `cg-` + 8 位十六进制（**服务端生成**，不收用户起名）→
+  结构上不可能撞上/覆盖既有 11 支，也不互相覆盖（撞名重试，见 §4 测试）。
+- 显示名 `label`（可中文，≤20 字）只存 CyberGril 资料空间，不进共享注册表。
+- 8090 的名字规则（`[a-zA-Z0-9_\-]{1,24}`）由 cg- 前缀天然满足。
+
+## 2. HTTP 契约（前端可直接实现）
+
+全部挂在统一 API（与现有 `/api/voice/*` 同前缀、同进程、同拒绝风格）。
+错误体一律 `{"detail": "<中文文案>"}`。
+
+### 2.1 `GET /api/voice/voices` — 列出可用声音
+
+```json
+{
+  "configured": true,
+  "voices": [
+    {"id": "rain", "label": "rain", "source": "builtin", "selected": false,
+     "previewed": false, "registered_at": null, "prompt_text": null, "prompt_wav": null},
+    {"id": "cg-9f2c1a77", "label": "她的声音", "source": "custom", "selected": true,
+     "previewed": true, "registered_at": 1793114934826,
+     "prompt_text": "这段参考音频念的文字", "prompt_wav": "/opt/voice-agent/cosyvoice/voice_samples/cg-9f2c1a77.wav"}
+  ],
+  "selection": {"voice": "cg-9f2c1a77", "label": "她的声音", "updated_at": 1793114940000}
+}
+```
+
+- `source`: `builtin`（11 支既有）| `custom`（`cg-` 前缀）。
+- `prompt_text`/`prompt_wav` 是备份清单字段（CG-009 承接），前端 UI 可不展示。
+- 实例未配置语音：`200 {"configured": false, "voices": [], "selection": null}`（诚实空态）。
+- 面板暂不可用：`503 {"detail": "音色登记服务这会儿不在，稍后再试试。"}`。
+
+### 2.2 `POST /api/voice/voices` — 登记参考音频
+
+**请求体就是参考音频原始 WAV 字节**（不引入 multipart 依赖）；参数走查询串：
+
+```
+POST /api/voice/voices?label=她的声音&prompt_text=这段音频念的文字&allow_asr=false
+Content-Type: audio/wav        （或 application/octet-stream）
+<body: WAV 字节>
+```
+
+| 参数 | 必填 | 说明 |
+| --- | --- | --- |
+| `label` | 否 | 显示名，≤20 字，不含控制字符/尖括号；缺省用 ID |
+| `prompt_text` | 否 | 参考文字（音频念的内容），≤500 字 |
+| `allow_asr` | 否 | `true` = 参考文字留空时**明确允许**面板 FunASR 自动转写（显式，不隐形） |
+
+`201` 响应即登记结果（同列表项结构）：
+```json
+{"id": "cg-9f2c1a77", "label": "她的声音", "prompt_text": "...",
+ "prompt_text_source": "user" | "asr", "prompt_wav": "...", "registered_at": 1793114934826}
+```
+
+服务端复验（不信任任何客户端声明；用户文件路径不是输入）：
+
+| 检查 | 拒绝（`422`） |
+| --- | --- |
+| WAV magic（RIFF/WAVE，fmt/data 可解析） | "只支持 WAV 格式的参考音频。" / "WAV 头不完整。" |
+| 大小 ≤ 15MB（同 8090） | "文件过大(>15MB)，请录 5–20 秒的一段。" |
+| 字节数与真实时长 ≥ 2 秒（同 8090 字节线 + 按头计算时长） | "参考音频太短（至少 2 秒），请录 5–20 秒。" |
+| 无参考文字且未明确允许转写 | "请填写这段音频的参考文字，或明确选择让系统转写（会用到语音识别）。" |
+
+其他错误：面板 400 透传原中文 detail（`422`）；面板 5xx/断连 `503`
+"音色登记服务这会儿不在，稍后再试。"；登记成功但列表未出现 `422`
+"登记结果没有出现在音色列表里，请重试。"（不留下半条可选项）。
+
+**建议录音格式**（前端 MediaRecorder 录完后转）：WAV（PCM），16k 或 24k 采样，
+单声道，16 位，5–20 秒，安静环境念 1–2 句完整的话。API 端校验是
+"合法 WAV + ≥2 秒"；时长上限由 15MB 大小线自然约束。
+
+### 2.3 `POST /api/voice/preview` — 试听（应用前必听）
+
+```
+POST /api/voice/preview
+{"voice": "cg-9f2c1a77", "text": "可选：自定义试听文字（≤200 字）"}
+```
+
+- 成功：`200 audio/wav`（24k/mono/16bit），现场新合成，绝不缓存旧音频。
+- 默认试听句："你好，这是我的声音，以后就用它陪你说话。"
+- 未知音色 `404` "这个声音现在不在列表里。"（**前置拦截**：CosyVoice 对未知
+  音色会静默回落 default，绝不让它"换人"冒充）；TTS 不在 `503`
+  "她的声音服务这会儿不在，稍后再试试。"；缺 `voice` `422`。
+- 试听不写聊天/记忆/生活/回执（测试断言四表零变化），只记一条资源流水（§4）。
+
+### 2.4 `GET /api/voice/selection` — 读取当前选择
+
+```json
+{"voice": "cg-9f2c1a77", "label": "她的声音", "updated_at": 1793114940000}
+```
+从未选择过：`{"voice": null, "label": "", "updated_at": null}`。
+
+### 2.5 `POST /api/voice/selection` — 应用给她
+
+```
+POST /api/voice/selection   {"voice": "cg-9f2c1a77"}
+```
+
+- 成功 `200 {"voice","label","updated_at"}`；未知音色 `404`；
+  **自定义音色未试听过 `409` "这个声音还没有试听过，先试听再给她用。"**
+  （内置 11 支是 D1 既有候选，不强制试听）；状态写不进 `503`
+  "她的音色状态这会儿写不进，这次没有换成功。"（已整体回滚，无半状态）。
+- 语义：**换声音只换发声**——不重建伴侣、不改出生底色/名字/关系/聊天/记忆
+  （测试断言 companion/bazi_core/chat_message 逐字节不变）。
+- 生效时机：worker 在会话入口读一次音色 → **通话中改选最迟下一通话生效**，
+  绝不半句换人；`POST /api/voice/session` 的响应新增 `"voice"` 字段 =
+  本次通话捕获的有效音色（选择已失效则诚实回落 `"default"`）。
+
+## 3. 状态与资源（可核算、可恢复）
+
+CyberGril 资料空间（实例数据库旁，与语音清理账同模式，重启读回）：
+
+| 文件 | 内容 |
+| --- | --- |
+| `<db>-voice-selection.json` | 当前选择（voice/label/updated_at），原子写 |
+| `<db>-voice-custom.json` | 自定义音色元数据 + 备份清单（id/label/prompt_text/prompt_wav/registered_at） |
+| `<db>-voice-usage.jsonl` | 资源流水：`{"ts","kind":"preview_tts"|"register_asr","voice"}`，有界保留（>5000 行裁到 2000） |
+
+- ASR 只在 `allow_asr=true` 且参考文字为空时发生（面板内 FunASR，非 LLM），
+  每次记 `register_asr` 流水；试听每次记 `preview_tts`。LLM 调用零新增。
+- 备份清单：`GET /api/voice/voices` 的 custom 项即完整清单（参考音频路径 +
+  参考文本 + 映射）；恢复 = 按清单重新登记（面板无导出接口，属 R6 备份卡范围）。
+- worker 文件 `voice_state_cybergril.json`：原子写且**保留文件里其他键**；
+  用户从未经 API 选择时，通话开启不代写该文件（部署/D1 域不替用户定）。
+
+## 4. 部署说明（本卡未执行；总工按 HEL-877 runbook §4 顺序）
+
+1. 前置与 HEL-877 相同：统一实例 8360/8361、TTS 9880（GPU≥6GB）、专属 worker 8082。
+2. `config/voice.yaml` 增补（全部可选，缺省即本机默认值）：
+   `panel_url: https://127.0.0.1:8090`、`tts_url: http://127.0.0.1:9880`、
+   `registry_file: /opt/voice-agent/cosyvoice/voices.json`、
+   `worker_state_file: /opt/voice-agent-cybergril/voice_state_cybergril.json`；
+   环境变量 `CYBERGRIL_VOICE_PANEL_URL/TTS_URL/REGISTRY_FILE/WORKER_STATE` 可覆盖。
+3. 依赖关系：音色登记/列表依赖 8090 面板进程在运行（.12 现状 pid 常驻）；
+   面板不在时相关端点诚实 503，语音会话不受影响（其健康检查仍只看 9880）。
+   API → 面板走本机回环 https（自签证书，服务端 `verify=False`，仅回环）。
+4. 零新增 Python 依赖（登记端点用原始字节体 + 查询串，特意不引 multipart 库）；
+   对 8090 的 multipart 代理由已有 httpx 客户端构造。
+5. 回滚：不部署即无痕；已部署时移除 voice.yaml 增补项即回到 HEL-877 行为，
+   资料空间三个文件留存无害。
+
+## 5. 自测与证据（全部零真实请求）
+
+- `tests/test_voice_custom.py` 18 项：正常登记/列表/选择全链、显式 ASR 准许、
+  格式/大小/时长拒绝、cg- 命名空间与撞名重试（含并发 4 路登记各自成立）、
+  面板拒绝/断连/谎报成功不留半条、未知音色 404 且 **TTS 零调用**（防静默回落
+  default）、TTS 停机诚实 503、试听零写聊天/记忆/回执、自定义未试听 409/内置
+  直选、写失败整体回滚、重启读回（选择/元数据/试听凭据）、通话捕获并定住/
+  失效回落 default/未选择不代写、通用助手状态零写入零删除、身份表零改写、
+  未配置端点诚实拒绝。
+- `scripts/verify_hel899_custom_voice.py`：12/12（真实 app + fake 面板/TTS/LiveKit：
+  列表→拒绝无准许 ASR→显式登记→未试听 409→试听→应用→会话捕获→通话中改选
+  定住本通话→下一通话生效→capabilities.voice=false→重启读回）。
+- 全量 `pytest tests -q`：370 通过（基线 352 + 本卡 18）。
+
+## 6. 未验证 / 承接（如实）
+
+- 真实 CosyVoice 试听、真实 FunASR 转写质量、克隆相似度：留统一审核卡在获批
+  环境实测（复用合法剩余额度，不另开账绕 HEL-879 上限）；本卡零真实调用。
+- 前端入口（上传/录音/试听/应用 UI）：视觉总监按 §2 契约接线（本卡未动 web/）。
+- 自定义音色删除、面板导出/完整备份（CG-009/R6）：后续卡（本卡最小合同不含删除，
+  失败/多余登记不影响使用；8090 面板自己的删除入口不代管）。
+- D1 最终音色决定：仍归白栖知/总管（本轮测试音色已授权总管自定）。
@@ -0,0 +1,191 @@
+"""HEL-899 offline one-shot chain check: custom voice registration -> preview
+-> apply -> call-time capture -> restart persistence, against the real app
+with a fake .12 voice stack (panel + CosyVoice) and fake LiveKit.
+
+Zero network, zero real LLM/embedding/ASR/TTS calls. Mirrors
+scripts/verify_hel877_voice_chain.py in spirit: real router + real services,
+fake transports only.
+
+Usage: PYTHONPATH=src .venv311/bin/python scripts/verify_hel899_custom_voice.py
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import sys
+import tempfile
+import wave
+from pathlib import Path
+
+import httpx
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tests"))
+
+from fastapi.testclient import TestClient  # noqa: E402
+from test_r2 import R2LLM  # noqa: E402
+
+from cybergril.api.app import create_app  # noqa: E402
+from cybergril.life.runtime import LifeConfig  # noqa: E402
+from cybergril.think.router import build_router  # noqa: E402
+from cybergril.think.scenes import Tier  # noqa: E402
+from cybergril.voice.custom import CustomVoiceConfig, CustomVoiceService  # noqa: E402
+from cybergril.voice.session import VoiceConfig, VoiceSessionService  # noqa: E402
+
+BUILTIN = ["default", "rain", "xiaoyun"]
+CREATE = {"name": "链路测试", "birth": "2000-10-18", "hour": "亥时", "relation": "朋友"}
+RESULTS: list[tuple[str, bool, str]] = []
+
+
+def check(name: str, ok: bool, note: str = "") -> None:
+    RESULTS.append((name, ok, note))
+    print(f"{'PASS' if ok else 'FAIL'}  {name}  {note}")
+
+
+def make_wav(seconds: float = 3.0) -> bytes:
+    buffer = io.BytesIO()
+    with wave.open(buffer, "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(16000)
+        wav.writeframes(b"\x00\x01" * int(16000 * seconds))
+    return buffer.getvalue()
+
+
+def _multipart_field(body: bytes, field: bytes) -> str:
+    for part in body.split(b"--"):
+        header, _, value = part.partition(b"\r\n\r\n")
+        if field + b'"' in header:
+            return value.strip(b"\r\n-").decode()
+    return ""
+
+
+def fake_transport(state: dict) -> httpx.MockTransport:
+    def handler(request: httpx.Request) -> httpx.Response:
+        host, path = request.url.host, request.url.path
+        if host == "panel.test":
+            if path == "/api/voices" and request.method == "GET":
+                return httpx.Response(200, json={"voices": sorted(state["ids"])})
+            if path == "/api/voices/upload":
+                name = _multipart_field(request.content, b"name")
+                prompt_text = _multipart_field(request.content, b"prompt_text")
+                if not prompt_text:  # 参考文本为空 → 面板 FunASR 转写（桩：固定文本）
+                    prompt_text = state["asr_text"]
+                state["ids"].append(name)
+                state["registry"][name] = {"prompt_text": prompt_text,
+                                           "prompt_wav": f"/samples/{name}.wav"}
+                if state.get("registry_file"):
+                    Path(state["registry_file"]).write_text(
+                        json.dumps(state["registry"], ensure_ascii=False), encoding="utf-8")
+                state["uploads"] += 1
+                return httpx.Response(200, json={"ok": True, "voice": name})
+            return httpx.Response(404)
+        if host == "tts.test":
+            if path == "/health":
+                return httpx.Response(200, json={"ok": True, "voices": sorted(state["ids"])})
+            if path == "/tts":
+                body = json.loads(request.content)
+                state["tts_calls"] += 1
+                return httpx.Response(200, content=f"{body['voice']}:".encode() + b"\x01\x00" * 240)
+        if host == "livekit.test":
+            return httpx.Response(200, json={"id": "AD_1", "room": "r"})
+        return httpx.Response(404)
+
+    return httpx.MockTransport(handler)
+
+
+def build(tmp: Path, state: dict):
+    combined = httpx.MockTransport(lambda request: fake_transport(state).handler(request))
+    voice = VoiceSessionService(
+        config=VoiceConfig(livekit_url="ws://voice.test:7880",
+                           livekit_internal="http://livekit.test",
+                           api_key="devkey", api_secret="test-secret-not-real",
+                           tts_health_url="http://tts.test/health"),
+        http=httpx.AsyncClient(transport=combined))
+    custom = CustomVoiceService(
+        config=CustomVoiceConfig(panel_url="https://panel.test", tts_url="http://tts.test",
+                                 registry_file=str(tmp / "voices.json"),
+                                 worker_state_file=str(tmp / "worker.json")),
+        http=httpx.AsyncClient(transport=fake_transport(state), verify=False),
+        selection_path=tmp / "sel.json", registry_path=tmp / "custom.json",
+        journal_path=tmp / "usage.jsonl")
+    provider = R2LLM()
+    router = build_router(providers={t: provider
+                                     for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)})
+    return create_app(db_path=":memory:", router=router,
+                      life_config=LifeConfig(enabled=True, poll_seconds=3600),
+                      semantic_mode="stub", voice_service=voice,
+                      custom_voice_service=custom), custom
+
+
+def main() -> int:
+    state = {"ids": list(BUILTIN), "registry": {}, "tts_calls": 0, "uploads": 0,
+             "asr_text": "自动转写参考文本。"}
+    with tempfile.TemporaryDirectory() as raw:
+        tmp = Path(raw)
+        registry_file = tmp / "voices.json"
+        registry_file.write_text(json.dumps(state["registry"]), encoding="utf-8")
+        state["registry_file"] = str(registry_file)
+        app, _custom = build(tmp, state)
+        with TestClient(app) as client:
+            client.post("/api/companion", json=CREATE)
+
+            listing = client.get("/api/voice/voices").json()
+            check("list builtin", {v["id"] for v in listing["voices"]} == set(BUILTIN))
+
+            refusal = client.post("/api/voice/voices", content=make_wav())
+            check("no-asr-consent refused", refusal.status_code == 422
+                  and state["uploads"] == 0, refusal.json().get("detail", ""))
+
+            registered = client.post("/api/voice/voices",
+                                     params={"allow_asr": "true", "label": "她的声音"},
+                                     content=make_wav())
+            record = registered.json()
+            check("register with explicit ASR", registered.status_code == 201
+                  and record["id"].startswith("cg-")
+                  and record["prompt_text_source"] == "asr", str(record))
+
+            unapplied = client.post("/api/voice/selection", json={"voice": record["id"]})
+            check("apply requires preview", unapplied.status_code == 409)
+
+            preview = client.post("/api/voice/preview", json={"voice": record["id"]})
+            check("preview fresh wav", preview.status_code == 200
+                  and preview.headers["content-type"] == "audio/wav"
+                  and state["tts_calls"] == 1)
+
+            applied = client.post("/api/voice/selection", json={"voice": record["id"]})
+            check("apply after preview", applied.status_code == 200
+                  and json.loads((tmp / "worker.json").read_text())["voice"] == record["id"])
+
+            opened = client.post("/api/voice/session").json()
+            check("session captures selection", opened["voice"] == record["id"])
+
+            client.post("/api/voice/selection", json={"voice": "rain"})
+            pinned = client.app.state.voice_service.get(opened["session_id"])
+            check("call keeps its voice", pinned.voice == record["id"])
+            second = client.post("/api/voice/session").json()
+            check("next call uses new voice", second["voice"] == "rain")
+
+            health = client.get("/api/health").json()
+            check("capabilities.voice stays false",
+                  health["capabilities"]["voice"] is False)
+
+        # “重启”：同批文件重建服务，选择/元数据/试听凭据都读回
+        app2, _custom2 = build(tmp, state)
+        with TestClient(app2) as client2:
+            client2.post("/api/companion", json=CREATE)
+            selection = client2.get("/api/voice/selection").json()
+            listing = client2.get("/api/voice/voices").json()
+            entry = next(v for v in listing["voices"] if v["id"] == record["id"])
+            check("selection survives restart", selection["voice"] == "rain")
+            check("custom entry survives restart", entry["label"] == "她的声音"
+                  and entry["previewed"] is True)
+
+    failed = [name for name, ok, _ in RESULTS if not ok]
+    print(f"\n{len(RESULTS) - len(failed)}/{len(RESULTS)} passed")
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -15,6 +15,7 @@ from typing import Literal
 from uuid import UUID
 from zoneinfo import ZoneInfo

+import httpx
 import yaml
 from anyio import CancelScope
 from fastapi import FastAPI, HTTPException, Query, Request
@@ -25,6 +26,7 @@ from pydantic import BaseModel, ConfigDict, Field, StrictInt, field_validator

 from ..config import CharacterConfig, ModelsConfig
 from ..contracts import LLMMessage
+from ..contracts.text import safe_text
 from ..core.bazi import (
    GAN_ELEM,
    Derived,
@@ -48,7 +50,13 @@ from ..memory.semantic import build_semantic
 from ..memory.store import Store
 from ..think.scenes import Scene
 from ..think.wiring import build_router_from_config
-from ..voice import VoiceSessionService, build_voice_router, load_voice_config
+from ..voice import (
+    CustomVoiceConfig,
+    CustomVoiceService,
+    VoiceSessionService,
+    build_voice_router,
+    load_voice_config,
+)

 HOURS = [
    "子时",
@@ -67,17 +75,6 @@ HOURS = [
 ERROR = "她这会儿没接上话，稍后再试试。你写的话还在。"


-def safe_text(value: str, maximum: int, label: str, required: bool = False) -> str:
-    value = value.strip()
-    if (
-        (required and not value)
-        or len(value) > maximum
-        or any(ord(c) < 32 or ord(c) == 127 or c in "<>" for c in value)
-    ):
-        raise ValueError(f"{label}请填写{1 if required else 0}–{maximum}字，不含控制字符或尖括号。")
-    return value
-
-
 class StrictBody(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

@@ -182,6 +179,7 @@ def create_app(
    semantic_mode: str | None = None,
    semantic_config: dict | None = None,
    voice_service=None,
+    custom_voice_service=None,
 ) -> FastAPI:
    """Dependencies can be injected for isolated fault/cancellation tests, never via browser flags."""
    db_path = db_path or os.environ.get(
@@ -291,6 +289,8 @@ def create_app(
                    await provider.aclose()
                if getattr(app.state, "voice_service", None):
                    await app.state.voice_service.aclose()
+                if getattr(app.state, "custom_voice_service", None):
+                    await app.state.custom_voice_service.aclose()
                store.close()
        finally:
            app.state.lifespan_active = False
@@ -315,8 +315,30 @@ def create_app(
                    Path(db_path).stem + "-voice-sessions.json"),
            ) if voice_config is not None else None
        )
+        # Custom voice service (HEL-899) rides the same config: same instance
+        # data space (selection/registry/journal beside the DB), dedicated
+        # HTTP client because the .12 voice panel is https with a self-signed
+        # cert the API must not fail to trust server-to-server.
+        if voice_config is not None:
+            in_memory = str(db_path) == ":memory:"
+            custom_voice_service = CustomVoiceService(
+                config=CustomVoiceConfig(
+                    panel_url=voice_config.panel_url,
+                    tts_url=voice_config.tts_url,
+                    registry_file=voice_config.registry_file,
+                    worker_state_file=voice_config.worker_state_file,
+                ),
+                http=httpx.AsyncClient(verify=False, timeout=httpx.Timeout(30, read=120)),
+                selection_path=None if in_memory else Path(db_path).with_name(
+                    Path(db_path).stem + "-voice-selection.json"),
+                registry_path=None if in_memory else Path(db_path).with_name(
+                    Path(db_path).stem + "-voice-custom.json"),
+                journal_path=None if in_memory else Path(db_path).with_name(
+                    Path(db_path).stem + "-voice-usage.jsonl"),
+            )
    app.state.voice_service = voice_service
-    app.include_router(build_voice_router(voice_service))
+    app.state.custom_voice_service = custom_voice_service
+    app.include_router(build_voice_router(voice_service, custom_voice_service))

    @app.exception_handler(RequestValidationError)
    async def invalid(_request, exc):
@@ -0,0 +1,14 @@
+"""跨层共享的纯文本校验（唯一实现，API 与语音模块共用）."""
+
+from __future__ import annotations
+
+
+def safe_text(value: str, maximum: int, label: str, required: bool = False) -> str:
+    value = value.strip()
+    if (
+        (required and not value)
+        or len(value) > maximum
+        or any(ord(c) < 32 or ord(c) == 127 or c in "<>" for c in value)
+    ):
+        raise ValueError(f"{label}请填写{1 if required else 0}–{maximum}字，不含控制字符或尖括号。")
+    return value
@@ -19,6 +19,10 @@ def __getattr__(name: str):  # lazy: API side must import without livekit instal
        from . import session

        return getattr(session, name)
+    if name in {"CustomVoiceConfig", "CustomVoiceService"}:
+        from . import custom
+
+        return getattr(custom, name)
    if name == "build_voice_router":
        from . import router

@@ -0,0 +1,423 @@
+"""参考音频自定义声音与专属音色选择（HEL-899，CyberGril 统一 API 侧）.
+
+复用既有能力，不自建克隆引擎（GOV-006）：
+
+* 音色登记 = .12 语音面板 ``POST /api/voices/upload``（写共享 ``voices.json``
+  + CosyVoice mtime 热加载；参考文本留空时面板用 FunASR 自动转写）。
+  本模块是它的**服务端代理**：浏览器永远不直连 8090，凭据/共享状态不进浏览器，
+  共享 ``voices.json`` 也只有 8090 这一个写者（跨进程读-改-写不再加竞态）。
+* 试听 = CosyVoice ``POST /tts``（返回 24k mono int16 PCM），本模块包成 WAV
+  给浏览器；每次试听都是新合成，不缓存旧音频冒充本次结果。
+* 应用 = 写 CyberGril 自己的音色状态（worker 的
+  ``voice_state_cybergril.json``），通话开启时由 router 捕获并落同一文件，
+  worker 在会话入口读一次——通话中改选最迟下一通话生效，绝不半句换人。
+
+命名空间：新增音色一律 ``cg-`` + 8 位十六进制（服务端生成，不收用户起名），
+不可能撞上/覆盖共享注册表里的既有 11 支音色，也不会互相覆盖。显示名
+（label，可中文）只存在 CyberGril 自己的资料空间，不进共享注册表。
+
+共享注册表涉及的操作（全部经 8090，本服务不直接改共享文件）：
+1. ``GET /api/voices`` 读列表（只读）；2. ``POST /api/voices/upload`` 以
+``cg-`` 专属 ID 登记新参考音频（唯一写入）；3. 登记成功后**只读**回看
+``voices.json``（部署配置的路径）补全 ASR 自动转写的参考文本，供备份清单。
+通用助手自己的 ``voice_state.json`` / ``persona_state.json`` 与既有音色
+条目从不触碰，也不通过改全局默认偷换她的声音。
+
+持久化（CyberGril 资料空间，实例数据库旁，与语音清理账同模式）：
+``<db>-voice-selection.json``（当前选择，重启读回）、
+``<db>-voice-custom.json``（自定义音色元数据 + 备份清单）、
+``<db>-voice-usage.jsonl``（ASR/TTS 资源使用流水，可核算；有界保留）。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import io
+import json
+import secrets
+import struct
+import time
+import wave
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import httpx
+
+from ..contracts.text import safe_text
+
+CUSTOM_PREFIX = "cg-"
+ID_HEX_CHARS = 8
+MAX_UPLOAD_BYTES = 15 * 1024 * 1024  # 与 8090 面板一致
+MIN_UPLOAD_BYTES = 16000 * 2  # 与 8090 面板一致（<2s @16k mono int16 的字节代理）
+MIN_SECONDS = 2.0  # 参考音频有效时长下限（比面板的字节检查更诚实）
+PREVIEW_TEXT_MAX = 200
+LABEL_MAX = 20
+PROMPT_TEXT_MAX = 500
+JOURNAL_KEEP_LINES = 2000
+JOURNAL_PRUNE_ABOVE = 5000
+DEFAULT_PREVIEW_TEXT = "你好，这是我的声音，以后就用它陪你说话。"
+STATE_VERSION = 1
+
+
+class RegistryUnavailable(Exception):
+    """8090 面板或 CosyVoice 暂不可用——如实上抛，不假装登记/试听成功。"""
+
+
+class StateWriteFailed(Exception):
+    """音色状态文件写不进——选择未成立，已整体回滚，不留下半状态。"""
+
+
+class RegistrationFailed(Exception):
+    """面板拒绝或返回异常——登记未成立，不留下半条可选项。"""
+
+
+@dataclass
+class CustomVoiceConfig:
+    panel_url: str = "https://127.0.0.1:8090"
+    tts_url: str = "http://127.0.0.1:9880"
+    # 只读回看（ASR 参考文本补全）；不配置则清单里该字段如实留空
+    registry_file: str | None = None
+    # CyberGril 专属 worker 的音色状态文件（worker 在会话入口读一次）
+    worker_state_file: str = "/opt/voice-agent-cybergril/voice_state_cybergril.json"
+
+
+def _atomic_write_json(path: Path, payload: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8")
+    tmp.replace(path)
+
+
+def parse_wav_header(data: bytes) -> dict:
+    """最小 WAV 头解析（RIFF/WAVE + fmt/data 定位），用于服务端复验.
+
+    只做诚实的格式事实读取：magic、声道数、采样率、位深、data 长度、时长。
+    不信任任何客户端声明，也不接受用户文件路径（输入只有上传字节）。
+    """
+    if len(data) < 44 or data[:4] != b"RIFF" or data[8:12] != b"WAVE":
+        raise ValueError("只支持 WAV 格式的参考音频。")
+    pos, channels, rate, bits, data_bytes = 12, None, None, None, 0
+    while pos + 8 <= len(data):
+        chunk_id = data[pos:pos + 4]
+        (size,) = struct.unpack_from("<I", data, pos + 4)
+        body = pos + 8
+        if chunk_id == b"fmt ":
+            if size < 16 or body + 16 > len(data):
+                raise ValueError("WAV 头不完整。")
+            fmt_code, channels, rate, _, _, bits = struct.unpack_from("<HHIIHH", data, body)
+            if fmt_code not in (1, 3, 0xFFFE):
+                raise ValueError("只支持 PCM/IEEE float WAV。")
+        elif chunk_id == b"data":
+            data_bytes = min(size, len(data) - body)
+        pos = body + size + (size & 1)
+    if not channels or not rate or not bits:
+        raise ValueError("WAV 头缺少格式信息。")
+    byte_rate = max(rate * channels * bits // 8, 1)
+    return {"channels": channels, "sample_rate": rate, "bits": bits,
+            "data_bytes": data_bytes, "seconds": data_bytes / byte_rate}
+
+
+def pcm_to_wav(pcm: bytes, sample_rate: int = 24000, channels: int = 1, width: int = 2) -> bytes:
+    """CosyVoice 的 audio/L16 裸 PCM 包成浏览器可播的 WAV（无缓存，每次现包）。"""
+    buffer = io.BytesIO()
+    with wave.open(buffer, "wb") as wav:
+        wav.setnchannels(channels)
+        wav.setsampwidth(width)
+        wav.setframerate(sample_rate)
+        wav.writeframes(pcm)
+    return buffer.getvalue()
+
+
+@dataclass
+class VoiceSelection:
+    voice: str | None = None
+    label: str = ""
+    updated_at: int = 0
+
+
+@dataclass
+class CustomVoiceService:
+    """统一 API 的自定义音色门面（一个实例一份文件，进程内锁串行化写）."""
+
+    config: CustomVoiceConfig = field(default_factory=CustomVoiceConfig)
+    http: httpx.AsyncClient = field(default_factory=httpx.AsyncClient)
+    clock: object = time.time
+    selection_path: Path | str | None = None
+    registry_path: Path | str | None = None  # 自定义音色元数据（备份清单同源）
+    journal_path: Path | str | None = None
+
+    def __post_init__(self) -> None:
+        self._lock = asyncio.Lock()
+        self._selection = self._load_selection()
+        self._custom: dict[str, dict] = self._load_custom()
+        self._prune_journal()
+
+    # -- 持久化（CyberGril 资料空间；重启读回） ------------------------------
+    def _load_selection(self) -> VoiceSelection:
+        if not self.selection_path:
+            return VoiceSelection()
+        try:
+            data = json.loads(Path(self.selection_path).read_text(encoding="utf-8"))
+            return VoiceSelection(voice=data.get("voice") or None,
+                                  label=str(data.get("label", "")),
+                                  updated_at=int(data.get("updated_at", 0)))
+        except (OSError, ValueError):
+            return VoiceSelection()
+
+    def _load_custom(self) -> dict[str, dict]:
+        if not self.registry_path:
+            return {}
+        try:
+            data = json.loads(Path(self.registry_path).read_text(encoding="utf-8"))
+            return dict(data.get("voices", {})) if data.get("version") == STATE_VERSION else {}
+        except (OSError, ValueError):
+            return {}
+
+    def _persist_selection(self) -> None:
+        if self.selection_path:
+            _atomic_write_json(Path(self.selection_path), {
+                "version": STATE_VERSION,
+                "voice": self._selection.voice,
+                "label": self._selection.label,
+                "updated_at": self._selection.updated_at,
+            })
+
+    def _persist_custom(self) -> None:
+        if self.registry_path:
+            _atomic_write_json(Path(self.registry_path), {
+                "version": STATE_VERSION, "voices": self._custom,
+            })
+
+    # -- 资源使用流水（可核算；ASR/TTS 才记，选择/列出不耗资源不记） ----------
+    def _journal(self, kind: str, voice: str) -> None:
+        if not self.journal_path:
+            return
+        try:
+            with Path(self.journal_path).open("a", encoding="utf-8") as fh:
+                fh.write(json.dumps({"ts": int(self.clock() * 1000),
+                                     "kind": kind, "voice": voice}, ensure_ascii=False) + "\n")
+        except OSError:
+            pass  # 流水尽力而为；核算缺失不影响功能诚实性（文档已注明）
+
+    def _prune_journal(self) -> None:
+        if not self.journal_path:
+            return
+        try:
+            lines = Path(self.journal_path).read_text(encoding="utf-8").splitlines()
+        except OSError:
+            return
+        if len(lines) > JOURNAL_PRUNE_ABOVE:
+            kept = [ln for ln in lines[-JOURNAL_KEEP_LINES:] if ln.strip()]
+            tmp = Path(self.journal_path).with_suffix(".jsonl.tmp")
+            tmp.write_text("\n".join(kept) + "\n", encoding="utf-8")
+            tmp.replace(self.journal_path)
+
+    def previewed_voices(self) -> set[str]:
+        """已成功试听过的音色（持久流水读回；重启不丢「应用前有可听试听」凭据）."""
+        if not self.journal_path:
+            return set()
+        try:
+            lines = Path(self.journal_path).read_text(encoding="utf-8").splitlines()
+        except OSError:
+            return set()
+        seen: set[str] = set()
+        for line in lines:
+            try:
+                entry = json.loads(line)
+            except ValueError:
+                continue
+            if entry.get("kind") == "preview_tts":
+                seen.add(str(entry.get("voice", "")))
+        return seen
+
+    # -- 共享注册表（8090 面板，唯一写者） ------------------------------------
+    async def registry_list(self) -> list[str]:
+        try:
+            response = await self.http.get(
+                f"{self.config.panel_url.rstrip('/')}/api/voices", timeout=5.0)
+            response.raise_for_status()
+            return sorted(response.json().get("voices", []))
+        except Exception as exc:
+            raise RegistryUnavailable(f"voices list: {type(exc).__name__}") from exc
+
+    def _read_registry_entry(self, voice_id: str) -> dict | None:
+        """只读回看共享 voices.json（部署配置路径）补全 ASR 参考文本；失败留空."""
+        path = self.config.registry_file
+        if not path:
+            return None
+        try:
+            data = json.loads(Path(path).read_text(encoding="utf-8"))
+            entry = data.get(voice_id)
+            return dict(entry) if isinstance(entry, dict) else None
+        except (OSError, ValueError):
+            return None
+
+    async def register(self, *, data: bytes, label: str, prompt_text: str,
+                       allow_asr: bool) -> dict:
+        """登记参考音频：服务端复验 → cg- 专属 ID → 面板 zero-shot 注册.
+
+        失败不留下半条可选项：面板拒绝/异常时不写任何 CyberGril 元数据，
+        面板自己对失败的 wav 有清理（ASR 失败即删，见其源码）。
+        """
+        if not data:
+            raise RegistrationFailed("请选择一段参考音频。")
+        if len(data) > MAX_UPLOAD_BYTES:
+            raise RegistrationFailed("文件过大(>15MB)，请录 5–20 秒的一段。")
+        try:
+            header = parse_wav_header(data)  # 非法格式在这里被拒
+            label = safe_text(label or "", LABEL_MAX, "声音名字") or ""
+            prompt_text = (safe_text(prompt_text or "", PROMPT_TEXT_MAX, "参考文字") or "").strip()
+        except ValueError as exc:
+            raise RegistrationFailed(str(exc)) from exc
+        if len(data) < MIN_UPLOAD_BYTES or header["seconds"] < MIN_SECONDS:
+            raise RegistrationFailed("参考音频太短（至少 2 秒），请录 5–20 秒。")
+        if not prompt_text and not allow_asr:
+            raise RegistrationFailed(
+                "请填写这段音频的参考文字，或明确选择让系统转写（会用到语音识别）。")
+        async with self._lock:
+            known = set(await self.registry_list())
+            for _ in range(6):
+                voice_id = CUSTOM_PREFIX + secrets.token_hex(ID_HEX_CHARS // 2)
+                if voice_id not in known and voice_id not in self._custom:
+                    break
+            else:  # pragma: no cover - 2^32 空间连撞 6 次属天文事件
+                raise RegistrationFailed("生成声音编号失败，请重试。")
+            files = {"file": (f"{voice_id}.wav", data, "audio/wav")}
+            form = {"name": voice_id, "prompt_text": prompt_text}
+            try:
+                response = await self.http.post(
+                    f"{self.config.panel_url.rstrip('/')}/api/voices/upload",
+                    data=form, files=files, timeout=60.0)
+            except Exception as exc:
+                raise RegistryUnavailable(f"voices upload: {type(exc).__name__}") from exc
+            if response.status_code >= 500:
+                raise RegistryUnavailable(f"voices upload: HTTP {response.status_code}")
+            if response.status_code >= 400:
+                # 面板拒绝（格式/大小/转写失败）：登记未成立，如实透传原因
+                detail = ""
+                try:
+                    detail = str(response.json().get("detail", ""))
+                except Exception:
+                    detail = response.text[:120]
+                raise RegistrationFailed(detail or f"音色登记被拒绝（HTTP {response.status_code}）。")
+            if voice_id not in set(await self.registry_list()):
+                raise RegistrationFailed("登记结果没有出现在音色列表里，请重试。")
+            entry = self._read_registry_entry(voice_id) or {}
+            record = {
+                "id": voice_id,
+                "label": label or voice_id,
+                "prompt_text": entry.get("prompt_text") or (prompt_text or None),
+                "prompt_text_source": "asr" if (not prompt_text and entry.get("prompt_text"))
+                else ("user" if prompt_text else None),
+                "prompt_wav": entry.get("prompt_wav"),
+                "registered_at": int(self.clock() * 1000),
+            }
+            self._custom[voice_id] = record
+            self._persist_custom()
+            if record["prompt_text_source"] == "asr":
+                self._journal("register_asr", voice_id)  # 显式获准的 FunASR 转写，可核算
+            return dict(record)
+
+    async def preview_pcm(self, *, voice: str, text: str | None) -> bytes:
+        """试听：先验存在（CosyVoice 对未知音色会静默回落 default，必须拦住），
+        再现场合成一段新 PCM。绝不缓存、绝不拿旧音频冒充本次结果。"""
+        if not voice or not await self._voice_exists(voice):
+            raise KeyError(voice)
+        try:
+            words = safe_text(text or DEFAULT_PREVIEW_TEXT, PREVIEW_TEXT_MAX, "试听文字") \
+                or DEFAULT_PREVIEW_TEXT
+        except ValueError as exc:
+            raise RegistrationFailed(str(exc)) from exc
+        try:
+            response = await self.http.post(
+                f"{self.config.tts_url.rstrip('/')}/tts",
+                json={"text": words, "voice": voice}, timeout=120.0)
+        except Exception as exc:
+            raise RegistryUnavailable(f"tts: {type(exc).__name__}") from exc
+        if response.status_code >= 400:
+            raise RegistryUnavailable(f"tts: HTTP {response.status_code}")
+        pcm = response.content or b""
+        if not pcm:
+            raise RegistryUnavailable("tts: empty audio")
+        self._journal("preview_tts", voice)
+        return pcm
+
+    async def _voice_exists(self, voice: str) -> bool:
+        return voice in set(await self.registry_list())
+
+    # -- 专属音色选择（换声音只换发声；人格/记忆/关系一概不动） ----------------
+    def selection(self) -> VoiceSelection:
+        return VoiceSelection(self._selection.voice, self._selection.label,
+                              self._selection.updated_at)
+
+    def custom_voices(self) -> dict[str, dict]:
+        return {vid: dict(rec) for vid, rec in self._custom.items()}
+
+    def read_worker_voice(self) -> str:
+        """worker 当前状态文件里的音色（部署/D1 域；无人经 API 选择时不代写）."""
+        path = Path(self.config.worker_state_file)
+        try:
+            return str(json.loads(path.read_text(encoding="utf-8")).get("voice", "")) or ""
+        except (OSError, ValueError):
+            return ""
+
+    def write_worker_voice(self, voice: str) -> None:
+        """原子写 worker 音色状态（保留文件里其他键）；失败抛 OSError，不半写."""
+        path = Path(self.config.worker_state_file)
+        try:
+            state = json.loads(path.read_text(encoding="utf-8"))
+            if not isinstance(state, dict):
+                state = {}
+        except (OSError, ValueError):
+            state = {}
+        state["voice"] = voice
+        _atomic_write_json(path, state)
+
+    async def apply_selection(self, *, voice: str) -> VoiceSelection:
+        """把她后续通话的发声换成指定音色（最迟下一通话生效）.
+
+        自定义（cg-）音色必须先成功试听过（「应用前有可听试听」）；内置
+        音色是既有 D1 候选，不强制。写 worker 状态文件与本地选择状态都
+        原子落盘；任一失败整体回滚，绝不留下两边不一致的半状态。
+        """
+        if not voice or not await self._voice_exists(voice):
+            raise KeyError(voice)
+        if voice.startswith(CUSTOM_PREFIX) and voice not in self.previewed_voices():
+            raise PermissionError(voice)
+        async with self._lock:
+            previous = VoiceSelection(self._selection.voice, self._selection.label,
+                                      self._selection.updated_at)
+            record = self._custom.get(voice)
+            try:
+                self._selection = VoiceSelection(
+                    voice=voice,
+                    label=(record["label"] if record else voice),
+                    updated_at=int(self.clock() * 1000))
+                self._persist_selection()
+                self.write_worker_voice(voice)
+            except OSError as exc:
+                self._selection = previous
+                with contextlib.suppress(OSError):
+                    self._persist_selection()  # 回滚本地状态
+                    if previous.voice:
+                        self.write_worker_voice(previous.voice)
+                raise StateWriteFailed(f"voice state write: {type(exc).__name__}") from exc
+            return VoiceSelection(self._selection.voice, self._selection.label,
+                                  self._selection.updated_at)
+
+    def manifest(self) -> dict:
+        """备份清单（CG-009 承接用）：参考音频路径/参考文本/映射 + 当前选择."""
+        return {
+            "version": STATE_VERSION,
+            "selection": {"voice": self._selection.voice, "label": self._selection.label},
+            "custom_voices": [dict(rec) for rec in
+                              sorted(self._custom.values(), key=lambda r: r["registered_at"])],
+            "shared_registry_note": (
+                "自定义音色登记在 .12 共享 voices.json（8090 面板写入）与 "
+                "voice_samples/<id>.wav；恢复时按 prompt_text + prompt_wav 重新登记即可。"
+            ),
+        }
+
+    async def aclose(self) -> None:
+        await self.http.aclose()
@@ -4,7 +4,12 @@ POST   /api/voice/session            open a room, dispatch the dedicated worker,
                                    return a short-lived join token (keys stay
                                    server-side); honest refusal when voice is
                                    not configured, no companion exists yet or
-                                    TTS is down.
+                                    TTS is down. HEL-899: the effective voice is
+                                    captured here (selection if still valid,
+                                    else the worker file's value, else default)
+                                    and pinned for THIS call — changes applied
+                                    mid-call take effect at the NEXT call at
+                                    the latest, never mid-sentence.
 DELETE /api/voice/session/{id}       end: delete the LiveKit room, which
                                    disconnects browser and worker, ends the
                                    agent job and releases the microphone.
@@ -15,6 +20,27 @@ DELETE /api/voice/session/{id}       end: delete the LiveKit room, which
 GET    /api/voice/status             honest switch state for the frontend
                                    (configured / tts reachable / agent name).

+Custom voice protocol (HEL-899) — the browser never talks to the .12 voice
+panel or CosyVoice directly; everything goes through this API:
+
+GET    /api/voice/voices             list builtin + her custom voices (labels,
+                                    selection/preview state, backup manifest
+                                    fields for custom entries).
+POST   /api/voice/voices             register a reference audio: raw WAV bytes
+                                    as the request body (no multipart — the
+                                    project takes no new dependency for one
+                                    endpoint), label/prompt_text/allow_asr as
+                                    query parameters; allow_asr explicitly
+                                    lets the existing FunASR transcribe when
+                                    no reference text is given.
+POST   /api/voice/preview            synthesize a FRESH sample with a voice and
+                                    return audio/wav (no caching, no chat/
+                                    memory/receipt writes).
+GET    /api/voice/selection          read her persisted voice selection.
+POST   /api/voice/selection          apply a voice to her subsequent calls
+                                    (custom voices require a successful preview
+                                    first; builtin D1 candidates do not).
+
 capabilities.voice in /api/health stays false: R5 is not real-accepted yet and
 this endpoint never claims end-to-end voice availability.
 """
@@ -22,19 +48,52 @@ this endpoint never claims end-to-end voice availability.
 from __future__ import annotations

 from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import Response

+from .custom import (
+    CustomVoiceService,
+    RegistrationFailed,
+    RegistryUnavailable,
+    StateWriteFailed,
+    pcm_to_wav,
+)
 from .session import CleanupAbandoned, VoiceSessionService

 REFUSAL_NOT_CONFIGURED = "语音还没有在这台服务上配置，不能假装能通话。"
 REFUSAL_NO_COMPANION = "还没有创建真实伴侣，语音也不会开始。"
 REFUSAL_TTS_DOWN = "她的声音服务这会儿不在，稍后再试试。"
 REFUSAL_LIVEKIT_DOWN = "通话服务暂时连不上，稍后再试试。"
+REFUSAL_PANEL_DOWN = "音色登记服务这会儿不在，稍后再试试。"
+REFUSAL_VOICE_STATE_WRITE = "她的音色状态这会儿写不进，这次没有换成功。"
 REFUSAL_CLEANUP_ABANDONED = (
    "这次通话的房间还没有删净，清理重试已达上限；已登记待人工收尾，请保留挂断状态。"
 )


-def build_voice_router(service: VoiceSessionService | None) -> APIRouter:
+def _capture_voice(service: VoiceSessionService, custom: CustomVoiceService | None) -> str:
+    """通话开启时的有效音色（HEL-899）：选择有效→选择；否则 worker 文件现值；
+    再无效→ default。用户从未经 API 选择时不改写 worker 文件（D1 域不代定）。"""
+    voices = service.last_health_voices()
+
+    def valid(candidate: str) -> bool:
+        return bool(candidate) and (voices is None or candidate in voices)
+
+    if custom is not None:
+        selection = custom.selection()
+        if selection.voice:
+            captured = selection.voice if valid(selection.voice) else "default"
+            try:
+                custom.write_worker_voice(captured)  # 本通话用捕获值；worker 入口读一次
+            except OSError:
+                raise HTTPException(503, REFUSAL_VOICE_STATE_WRITE) from None
+            return captured
+        current = custom.read_worker_voice()
+        return current if valid(current) else "default"
+    return ""
+
+
+def build_voice_router(service: VoiceSessionService | None,
+                       custom: CustomVoiceService | None = None) -> APIRouter:
    router = APIRouter(prefix="/api/voice", tags=["voice"])

    @router.get("/status")
@@ -59,8 +118,9 @@ def build_voice_router(service: VoiceSessionService | None) -> APIRouter:
            raise HTTPException(409, REFUSAL_NO_COMPANION)
        if not await service.tts_available():
            raise HTTPException(503, REFUSAL_TTS_DOWN)
+        captured = _capture_voice(service, custom)
        try:
-            session = await service.open_session(epoch=state.epoch)
+            session = await service.open_session(epoch=state.epoch, voice=captured)
        except Exception:
            raise HTTPException(503, REFUSAL_LIVEKIT_DOWN) from None
        return {
@@ -71,6 +131,7 @@ def build_voice_router(service: VoiceSessionService | None) -> APIRouter:
            "agent_name": service.config.agent_name,
            "epoch": session.epoch,
            "expires_at": int(session.expires_at * 1000),
+            "voice": session.voice,
        }

    @router.delete("/session/{session_id}")
@@ -101,4 +162,104 @@ def build_voice_router(service: VoiceSessionService | None) -> APIRouter:
            raise HTTPException(404, "这次通话已经不在了。")
        return {"ended": True, "room": session.room}

+    # -- custom voices (HEL-899) --------------------------------------------
+    @router.get("/voices")
+    async def voices() -> dict:
+        if custom is None:
+            return {"configured": False, "voices": [], "selection": None}
+        try:
+            listed = await custom.registry_list()
+        except RegistryUnavailable:
+            raise HTTPException(503, REFUSAL_PANEL_DOWN) from None
+        previewed = custom.previewed_voices()
+        selection = custom.selection()
+        records = custom.custom_voices()
+        items = []
+        for voice_id in listed:
+            record = records.get(voice_id)
+            items.append({
+                "id": voice_id,
+                "label": record["label"] if record else voice_id,
+                "source": "custom" if voice_id.startswith("cg-") else "builtin",
+                "selected": selection.voice == voice_id,
+                "previewed": voice_id in previewed,
+                "registered_at": record["registered_at"] if record else None,
+                # 备份清单字段（CG-009 承接）：自定义条目带参考文本与音频位置
+                "prompt_text": record["prompt_text"] if record else None,
+                "prompt_wav": record["prompt_wav"] if record else None,
+            })
+        return {
+            "configured": True,
+            "voices": items,
+            "selection": {"voice": selection.voice, "label": selection.label,
+                          "updated_at": selection.updated_at or None},
+        }
+
+    @router.post("/voices", status_code=201)
+    async def register_voice(request: Request, label: str = "", prompt_text: str = "",
+                             allow_asr: bool = False) -> dict:
+        if custom is None:
+            raise HTTPException(503, REFUSAL_NOT_CONFIGURED)
+        if not request.app.state.store.query_one("SELECT id FROM companion"):
+            raise HTTPException(409, REFUSAL_NO_COMPANION)
+        # 请求体就是参考音频原始字节（audio/wav/octet-stream）；登记参数走查询串，
+        # 不引入 multipart 依赖。录音/文件内容照旧服务端复验，不信任任何声明。
+        data = await request.body()
+        try:
+            record = await custom.register(data=data, label=label,
+                                           prompt_text=prompt_text, allow_asr=allow_asr)
+        except RegistrationFailed as exc:
+            raise HTTPException(422, str(exc)) from None
+        except RegistryUnavailable:
+            raise HTTPException(503, REFUSAL_PANEL_DOWN) from None
+        return record
+
+    @router.post("/preview")
+    async def preview(body: dict) -> Response:
+        if custom is None:
+            raise HTTPException(503, REFUSAL_NOT_CONFIGURED)
+        voice = str(body.get("voice") or "") if isinstance(body, dict) else ""
+        text = body.get("text") if isinstance(body, dict) else None
+        if not voice:
+            raise HTTPException(422, "请指定要试听的声音。")
+        if text is not None and not isinstance(text, str):
+            raise HTTPException(422, "试听文字格式无效。")
+        try:
+            pcm = await custom.preview_pcm(voice=voice, text=text)
+        except KeyError:
+            raise HTTPException(404, "这个声音现在不在列表里。") from None
+        except RegistrationFailed as exc:
+            raise HTTPException(422, str(exc)) from None
+        except RegistryUnavailable:
+            raise HTTPException(503, REFUSAL_TTS_DOWN) from None
+        return Response(content=pcm_to_wav(pcm), media_type="audio/wav")
+
+    @router.get("/selection")
+    async def get_selection() -> dict:
+        if custom is None:
+            raise HTTPException(503, REFUSAL_NOT_CONFIGURED)
+        selection = custom.selection()
+        return {"voice": selection.voice, "label": selection.label,
+                "updated_at": selection.updated_at or None}
+
+    @router.post("/selection")
+    async def apply_selection(body: dict) -> dict:
+        if custom is None:
+            raise HTTPException(503, REFUSAL_NOT_CONFIGURED)
+        voice = str(body.get("voice") or "") if isinstance(body, dict) else ""
+        if not voice:
+            raise HTTPException(422, "请指定要应用的声音。")
+        try:
+            selection = await custom.apply_selection(voice=voice)
+        except KeyError:
+            raise HTTPException(404, "这个声音现在不在列表里。") from None
+        except PermissionError:
+            raise HTTPException(409, "这个声音还没有试听过，先试听再给她用。") from None
+        except RegistryUnavailable:
+            raise HTTPException(503, REFUSAL_PANEL_DOWN) from None
+        except StateWriteFailed:
+            raise HTTPException(503, REFUSAL_VOICE_STATE_WRITE) from None
+        return {"voice": selection.voice, "label": selection.label,
+                "updated_at": selection.updated_at}
+
    return router
@@ -65,6 +65,11 @@ class VoiceConfig:
    token_ttl_seconds: float = DEFAULT_TOKEN_TTL
    session_ttl_seconds: float = DEFAULT_SESSION_TTL
    dispatch_timeout_seconds: float = 5.0
+    # HEL-899 自定义音色（复用 .12 语音面板与 CosyVoice；缺省即本机默认值）
+    panel_url: str = "https://127.0.0.1:8090"
+    tts_url: str = "http://127.0.0.1:9880"
+    registry_file: str | None = None  # 共享 voices.json 只读回看（ASR 参考文本补全）
+    worker_state_file: str = "/opt/voice-agent-cybergril/voice_state_cybergril.json"

    def sanitized(self) -> dict:
        return {
@@ -116,6 +121,10 @@ def load_voice_config(
        "api_key": os.environ.get("CYBERGRIL_VOICE_API_KEY"),
        "api_secret": os.environ.get("CYBERGRIL_VOICE_API_SECRET"),
        "tts_health_url": os.environ.get("CYBERGRIL_VOICE_TTS_HEALTH_URL"),
+        "panel_url": os.environ.get("CYBERGRIL_VOICE_PANEL_URL"),
+        "tts_url": os.environ.get("CYBERGRIL_VOICE_TTS_URL"),
+        "registry_file": os.environ.get("CYBERGRIL_VOICE_REGISTRY_FILE"),
+        "worker_state_file": os.environ.get("CYBERGRIL_VOICE_WORKER_STATE"),
    }
    values = {**data, **{k: v for k, v in overrides.items() if v}}
    required = ("livekit_url", "livekit_internal", "api_key", "api_secret")
@@ -131,6 +140,11 @@ def load_voice_config(
        tts_health_url=str(values.get("tts_health_url") or "http://127.0.0.1:9880/health"),
        token_ttl_seconds=float(values.get("token_ttl_seconds") or DEFAULT_TOKEN_TTL),
        session_ttl_seconds=float(values.get("session_ttl_seconds") or DEFAULT_SESSION_TTL),
+        panel_url=str(values.get("panel_url") or "https://127.0.0.1:8090"),
+        tts_url=str(values.get("tts_url") or "http://127.0.0.1:9880"),
+        registry_file=(str(values["registry_file"]) if values.get("registry_file") else None),
+        worker_state_file=str(values.get("worker_state_file")
+                              or "/opt/voice-agent-cybergril/voice_state_cybergril.json"),
    )


@@ -143,12 +157,13 @@ class VoiceSession:
    expires_at: float
    epoch: int
    dispatch_id: str = ""
+    voice: str = ""  # HEL-899: 本次通话捕获的有效音色（开启时定住，通话中改选不影响本通话）

    def as_state(self) -> dict:
        return {
            "session_id": self.session_id, "room": self.room, "identity": self.identity,
            "created_at": self.created_at, "expires_at": self.expires_at,
-            "epoch": self.epoch, "dispatch_id": self.dispatch_id,
+            "epoch": self.epoch, "dispatch_id": self.dispatch_id, "voice": self.voice,
        }

    @classmethod
@@ -158,6 +173,7 @@ class VoiceSession:
            identity=str(data.get("identity", "")), created_at=float(data.get("created_at", 0)),
            expires_at=float(data.get("expires_at", 0)), epoch=int(data.get("epoch", 0)),
            dispatch_id=str(data.get("dispatch_id", "")),
+            voice=str(data.get("voice", "")),
        )


@@ -193,6 +209,7 @@ class _CleanupDebt:
 class _Health:
    checked: float = 0.0
    ok: bool = False
+    voices: list[str] | None = None  # HEL-899: /health 顺带的热加载音色表（无则 None）


@dataclass
@@ -282,11 +299,25 @@ class VoiceSessionService:
            try:
                response = await self.http.get(self.config.tts_health_url, timeout=3.0)
                ok = response.status_code // 100 == 2
+                voices: list[str] | None = None
+                if ok:
+                    # CosyVoice /health = {"ok", "voices": [...]}（同请求内已热加载）；
+                    # 空体/异常体只当"探到了但没带名单"，不推翻 ok 判定。
+                    try:
+                        listed = response.json().get("voices")
+                        voices = sorted(str(v) for v in listed) if isinstance(listed, list) else None
+                    except Exception:
+                        voices = None
            except Exception:
                ok = False
-            self._tts_health = _Health(checked=self.clock(), ok=ok)
+                voices = None
+            self._tts_health = _Health(checked=self.clock(), ok=ok, voices=voices)
            return ok

+    def last_health_voices(self) -> list[str] | None:
+        """最近一次健康探测拿到的音色表（None=本次探测没带回名单）."""
+        return self._tts_health.voices
+
    # -- lifecycle ----------------------------------------------------------
    def _prune(self) -> None:
        """Expired registrations move into the cleanup ledger — never silently
@@ -387,7 +418,7 @@ class VoiceSessionService:
        this instance (bounded by the persisted ledger)."""
        self._kick_cleanup()

-    async def open_session(self, *, epoch: int) -> VoiceSession:
+    async def open_session(self, *, epoch: int, voice: str = "") -> VoiceSession:
        self._prune()
        self._kick_cleanup()
        alphabet = string.ascii_lowercase + string.digits
@@ -411,6 +442,7 @@ class VoiceSessionService:
            expires_at=now + self.config.session_ttl_seconds,
            epoch=epoch,
            dispatch_id=dispatch.dispatch_id,
+            voice=voice,
        )
        self._sessions[session.session_id] = session
        self._persist_state()
@@ -0,0 +1,510 @@
+"""Custom voice tests (HEL-899): reference-audio registration, preview,
+selection persistence, call-time capture — all against the real router and
+real services with fake panel/TTS transports. Zero network, zero real
+LLM/embedding/ASR/TTS calls."""
+
+from __future__ import annotations
+
+import io
+import json
+import sys
+import wave
+from pathlib import Path
+
+import httpx
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tests"))
+
+from test_r2 import R2LLM  # noqa: E402
+
+from cybergril.api.app import create_app  # noqa: E402
+from cybergril.life.runtime import LifeConfig  # noqa: E402
+from cybergril.think.router import build_router  # noqa: E402
+from cybergril.think.scenes import Tier  # noqa: E402
+from cybergril.voice.custom import CustomVoiceConfig, CustomVoiceService  # noqa: E402
+from cybergril.voice.session import VoiceConfig, VoiceSessionService  # noqa: E402
+
+CREATE = {"name": "语音测试", "birth": "2000-10-18", "hour": "亥时", "relation": "朋友"}
+BUILTIN = ["bailu", "bajie", "default", "female_soft", "haimianbaobao", "jiaojiao",
+           "male_warm", "nezha", "rain", "sunwukong", "xiaoyun"]
+
+
+def make_wav(seconds: float = 3.0, rate: int = 16000, channels: int = 1) -> bytes:
+    buffer = io.BytesIO()
+    with wave.open(buffer, "wb") as wav:
+        wav.setnchannels(channels)
+        wav.setsampwidth(2)
+        wav.setframerate(rate)
+        wav.writeframes(b"\x00\x01" * int(rate * channels * seconds))
+    return buffer.getvalue()
+
+
+class FakeVoiceStack:
+    """.12 语音面板(8090) + CosyVoice(9880) 的只读桩：记录每一次调用."""
+
+    def __init__(self, tmp: Path) -> None:
+        self.ids: list[str] = list(BUILTIN)
+        self.registry_file = tmp / "voices.json"
+        self.uploads: list[dict] = []
+        self.tts_calls: list[dict] = []
+        self.panel_down = False
+        self.tts_down = False
+        self.reject_upload = ""  # 面板 400 detail（如"自动转写失败"）
+        self.swallow = False  # 谎报成功：回 ok 但不进列表（测登记后核验）
+        self.generic_voice_writes = 0  # POST /api/voice —— 必须永远是 0
+        self.deletes: list[str] = []
+        self.transport = httpx.MockTransport(self._handle)
+
+    def _registry(self) -> dict:
+        return json.loads(self.registry_file.read_text(encoding="utf-8")) \
+            if self.registry_file.exists() else {}
+
+    def _write_registry(self, data: dict) -> None:
+        self.registry_file.write_text(json.dumps(data, ensure_ascii=False, indent=1),
+                                      encoding="utf-8")
+
+    def _handle(self, request: httpx.Request) -> httpx.Response:
+        host, path = request.url.host, request.url.path
+        if host == "panel.test":
+            if self.panel_down:
+                raise httpx.ConnectError("panel down")
+            if path == "/api/voices" and request.method == "GET":
+                return httpx.Response(200, json={"voices": sorted(self.ids)})
+            if path == "/api/voices/upload":
+                name = self._form_name(request)
+                prompt_text = self._form_text(request)
+                if self.reject_upload:
+                    return httpx.Response(400, json={"detail": self.reject_upload})
+                if not prompt_text:  # 面板在参考文本为空时走 FunASR（桩：固定转写）
+                    prompt_text = "这是自动转写出的参考文本，内容与音频一致。"
+                registry = self._registry()
+                registry[name] = {"prompt_text": prompt_text,
+                                  "prompt_wav": f"/opt/voice-agent/cosyvoice/voice_samples/{name}.wav"}
+                self._write_registry(registry)
+                self.uploads.append({"name": name, "prompt_text": prompt_text,
+                                     "bytes": len(request.content)})
+                if not self.swallow:
+                    self.ids.append(name)
+                return httpx.Response(200, json={"ok": True, "voice": name})
+            if path.startswith("/api/voices/") and request.method == "DELETE":
+                self.deletes.append(path.rsplit("/", 1)[-1])
+                return httpx.Response(200, json={"ok": True})
+            if path == "/api/voice":  # 通用助手自己的状态 —— CyberGril 绝不写
+                self.generic_voice_writes += 1
+                return httpx.Response(200, json={"ok": True})
+            return httpx.Response(404)
+        if host == "tts.test":
+            if path == "/health":
+                if self.tts_down:
+                    return httpx.Response(503)
+                return httpx.Response(200, json={"ok": True, "voices": sorted(self.ids)})
+            if path == "/tts":
+                if self.tts_down:
+                    return httpx.Response(503)
+                body = json.loads(request.content)
+                self.tts_calls.append(body)
+                marker = f"{body['voice']}|{body['text']}".encode()
+                return httpx.Response(200, content=marker + b"\x01\x00" * 240)
+            return httpx.Response(404)
+        return httpx.Response(404)
+
+    @staticmethod
+    def _form_name(request: httpx.Request) -> str:
+        return _multipart_field(request.content, b"name") or ""
+
+    @staticmethod
+    def _form_text(request: httpx.Request) -> str:
+        return _multipart_field(request.content, b"prompt_text") or ""
+
+
+def _multipart_field(body: bytes, field: bytes) -> str | None:
+    """极简 multipart 取字段（只为桩服务；产品代码不解析 multipart）."""
+    for part in body.split(b"--"):
+        header, _, value = part.partition(b"\r\n\r\n")
+        if field + b'"' in header:
+            return value.strip(b"\r\n-").decode()
+    return None
+
+
+class FakeLiveKit:
+    def __init__(self) -> None:
+        self.dispatches: list[str] = []
+        self.deleted_rooms: list[str] = []
+        self.transport = httpx.MockTransport(self._handle)
+
+    def _handle(self, request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/twirp/livekit.AgentDispatchService/CreateDispatch":
+            self.dispatches.append(request.url.path)
+            return httpx.Response(200, json={"id": "AD_1", "room": "r"})
+        if request.url.path == "/twirp/livekit.RoomService/DeleteRoom":
+            self.deleted_rooms.append(request.url.path)
+            return httpx.Response(200, content=b"")
+        return httpx.Response(404)
+
+
+def build_app(tmp: Path, stack: FakeVoiceStack, livekit: FakeLiveKit | None = None):
+    livekit = livekit or FakeLiveKit()
+    voice = VoiceSessionService(
+        config=VoiceConfig(
+            livekit_url="ws://voice.test:7880", livekit_internal="http://livekit.test",
+            api_key="devkey", api_secret="test-secret-not-real",
+            tts_health_url="http://tts.test/health"),
+        http=httpx.AsyncClient(transport=httpx.MockTransport(
+            lambda request: stack._handle(request) if request.url.host == "tts.test"
+            else livekit._handle(request))))
+    worker_state = tmp / "voice_state_cybergril.json"
+    custom = CustomVoiceService(
+        config=CustomVoiceConfig(
+            panel_url="https://panel.test", tts_url="http://tts.test",
+            registry_file=str(stack.registry_file),
+            worker_state_file=str(worker_state)),
+        http=httpx.AsyncClient(transport=stack.transport, verify=False),
+        selection_path=tmp / "sel.json", registry_path=tmp / "custom.json",
+        journal_path=tmp / "usage.jsonl")
+    provider = R2LLM()
+    router = build_router(providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)})
+    app = create_app(db_path=":memory:", router=router,
+                     life_config=LifeConfig(enabled=True, poll_seconds=3600),
+                     semantic_mode="stub", voice_service=voice,
+                     custom_voice_service=custom)
+    return app, custom, worker_state
+
+
+@pytest.fixture
+def stack(tmp_path):
+    return FakeVoiceStack(tmp_path)
+
+
+@pytest.fixture
+def instance(tmp_path, stack):
+    from fastapi.testclient import TestClient
+
+    app, custom, worker_state = build_app(tmp_path, stack)
+    with TestClient(app) as client:
+        client.post("/api/companion", json=CREATE)
+        yield client, stack, custom, worker_state
+
+
+def test_register_list_and_selection_flow(instance):
+    client, stack, custom, worker_state = instance
+    audio = make_wav(3)
+    registered = client.post(
+        "/api/voice/voices",
+        params={"label": "她的声音", "prompt_text": "这是参考文字"},
+        content=audio, headers={"Content-Type": "audio/wav"})
+    assert registered.status_code == 201
+    record = registered.json()
+    assert record["id"].startswith("cg-") and len(record["id"]) == 11
+    assert record["label"] == "她的声音"
+    assert record["prompt_text"] == "这是参考文字"
+    assert record["prompt_text_source"] == "user"
+    # 上传到达面板的是服务端生成的 cg- 专属 ID，不带用户文件名
+    assert stack.uploads[-1]["name"] == record["id"]
+    assert stack.uploads[-1]["bytes"] > 0
+
+    listing = client.get("/api/voice/voices").json()
+    by_id = {v["id"]: v for v in listing["voices"]}
+    assert set(by_id) == set(BUILTIN) | {record["id"]}
+    assert by_id[record["id"]]["source"] == "custom"
+    assert by_id[record["id"]]["label"] == "她的声音"
+    assert by_id["rain"]["source"] == "builtin"
+    assert listing["selection"]["voice"] is None
+
+    preview = client.post("/api/voice/preview", json={"voice": record["id"]})
+    assert preview.status_code == 200
+    assert preview.headers["content-type"] == "audio/wav"
+    with wave.open(io.BytesIO(preview.content)) as wav:
+        assert wav.getframerate() == 24000 and wav.getnchannels() == 1
+    assert stack.tts_calls[-1] == {"text": "你好，这是我的声音，以后就用它陪你说话。",
+                                   "voice": record["id"]}
+
+    applied = client.post("/api/voice/selection", json={"voice": record["id"]})
+    assert applied.status_code == 200 and applied.json()["voice"] == record["id"]
+    assert json.loads(worker_state.read_text())["voice"] == record["id"]
+    assert client.get("/api/voice/selection").json()["voice"] == record["id"]
+    listing = client.get("/api/voice/voices").json()
+    assert {v["id"] for v in listing["voices"] if v["selected"]} == {record["id"]}
+    # 备份清单字段在列表里（CG-009 承接）
+    entry = [v for v in listing["voices"] if v["id"] == record["id"]][0]
+    assert entry["prompt_wav"].endswith(record["id"] + ".wav")
+
+
+def test_register_requires_explicit_asr_consent(instance):
+    client, stack, _, _ = instance
+    refusal = client.post("/api/voice/voices", content=make_wav(3))
+    assert refusal.status_code == 422
+    assert "转写" in refusal.json()["detail"] or "参考文字" in refusal.json()["detail"]
+    assert stack.uploads == []  # 未获明确准许，不触发任何 ASR/上传
+    allowed = client.post("/api/voice/voices", params={"allow_asr": "true"},
+                          content=make_wav(3))
+    assert allowed.status_code == 201
+    assert allowed.json()["prompt_text_source"] == "asr"
+    journal = [json.loads(ln) for ln in
+               Path(instance[3].parent, "usage.jsonl").read_text().splitlines()]
+    assert any(e["kind"] == "register_asr" and e["voice"] == allowed.json()["id"]
+               for e in journal)
+
+
+def test_register_rejects_bad_formats(instance):
+    client, stack, _, _ = instance
+    assert client.post("/api/voice/voices", params={"prompt_text": "x"},
+                       content=b"ID3-not-a-wav").status_code == 422
+    assert client.post("/api/voice/voices", params={"prompt_text": "x"},
+                       content=make_wav(1.0)).status_code == 422  # <2s
+    header_only = b"RIFF" + (9999).to_bytes(4, "little") + b"WAVE" + b"junk" * 10
+    assert client.post("/api/voice/voices", params={"prompt_text": "x"},
+                       content=header_only).status_code == 422  # WAV magic 但头不完整
+    too_big = make_wav(2) + b"\x00" * (15 * 1024 * 1024 + 1)
+    response = client.post("/api/voice/voices", params={"prompt_text": "x"}, content=too_big)
+    assert response.status_code == 422 and "15MB" in response.json()["detail"]
+    assert stack.uploads == []
+
+
+def test_register_ids_never_collide_with_builtin_or_each_other(instance, monkeypatch):
+    client, stack, _, _ = instance
+    first = client.post("/api/voice/voices", params={"prompt_text": "一"},
+                        content=make_wav(2)).json()
+    second = client.post("/api/voice/voices", params={"prompt_text": "二"},
+                         content=make_wav(2)).json()
+    assert first["id"] != second["id"]
+    assert not any(v.startswith("cg-") for v in BUILTIN)
+    # 强制撞名：第一次生成的 ID 已在注册表里 → 必须换新 ID，而不是覆盖
+    registry = stack._registry()
+    registry["cg-deadbeef"] = {"prompt_text": "旧条目", "prompt_wav": "/old.wav"}
+    stack._write_registry(registry)
+    stack.ids.append("cg-deadbeef")  # 让首撞成立
+    calls = iter(["deadbeef", "cafef00d"])
+    monkeypatch.setattr("secrets.token_hex", lambda n: next(calls))
+    forced = client.post("/api/voice/voices", params={"prompt_text": "三"},
+                         content=make_wav(2)).json()
+    assert forced["id"] == "cg-cafef00d"
+    assert stack._registry()["cg-deadbeef"]["prompt_text"] == "旧条目"  # 未被覆盖
+
+
+async def test_concurrent_registers_are_serialized_and_distinct(tmp_path, stack):
+    """并发登记同名同音频：锁串行化 + 随机 ID，各自成立、互不覆盖."""
+    import asyncio
+
+    service = CustomVoiceService(
+        config=CustomVoiceConfig(panel_url="https://panel.test", tts_url="http://tts.test",
+                                 registry_file=str(stack.registry_file),
+                                 worker_state_file=str(tmp_path / "w.json")),
+        http=httpx.AsyncClient(transport=stack.transport, verify=False),
+        selection_path=tmp_path / "sel.json", registry_path=tmp_path / "custom.json",
+        journal_path=tmp_path / "usage.jsonl")
+    try:
+        records = await asyncio.gather(*[
+            service.register(data=make_wav(2), label="", prompt_text="并发", allow_asr=False)
+            for _ in range(4)])
+        ids = [record["id"] for record in records]
+        assert len(set(ids)) == 4 and all(v.startswith("cg-") for v in ids)
+        assert len(stack.uploads) == 4
+        assert set(service.custom_voices()) == set(ids)
+    finally:
+        await service.aclose()
+
+
+def test_register_panel_down_or_rejecting_leaves_no_half_option(instance):
+    client, stack, custom, _ = instance
+    stack.panel_down = True
+    assert client.post("/api/voice/voices", params={"prompt_text": "x"},
+                       content=make_wav(2)).status_code == 503
+    stack.panel_down = False
+    stack.reject_upload = "自动转写失败(boom), 请手动填写参考文本"
+    assert client.post("/api/voice/voices", params={"allow_asr": "true"},
+                       content=make_wav(2)).status_code == 422
+    assert custom.custom_voices() == {}
+    stack.reject_upload = ""
+    # 面板谎报成功（列表里没出现）也要拒绝，不登记元数据
+    stack.swallow = True
+    lying = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                        content=make_wav(2))
+    assert lying.status_code in (422, 503)
+    assert custom.custom_voices() == {}
+
+
+def test_preview_unknown_voice_never_reaches_tts(instance):
+    client, stack, _, _ = instance
+    response = client.post("/api/voice/preview", json={"voice": "ghost"})
+    assert response.status_code == 404
+    assert stack.tts_calls == []  # CosyVoice 对未知音色静默回落 default，必须前置拦住
+    assert client.post("/api/voice/preview", json={}).status_code == 422
+
+
+def test_preview_tts_down_honest(instance):
+    client, stack, _, _ = instance
+    registered = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                             content=make_wav(2)).json()
+    stack.tts_down = True
+    response = client.post("/api/voice/preview", json={"voice": registered["id"]})
+    assert response.status_code == 503
+    assert "声音服务" in response.json()["detail"]
+
+
+def test_preview_writes_no_chat_memory_or_receipts(instance):
+    client, stack, _, _ = instance
+    registered = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                             content=make_wav(2)).json()
+    app = client.app
+    counts = {table: app.state.store.query_one(f"SELECT COUNT(*) c FROM {table}")["c"]
+              for table in ("chat_message", "mem_episodic", "sharing_attempt", "life_event")}
+    assert client.post("/api/voice/preview",
+                       json={"voice": registered["id"], "text": "自定义试听句"}).status_code == 200
+    assert stack.tts_calls[-1]["text"] == "自定义试听句"
+    after = {table: app.state.store.query_one(f"SELECT COUNT(*) c FROM {table}")["c"]
+             for table in counts}
+    assert after == counts
+
+
+def test_apply_rules(instance):
+    client, stack, _, worker_state = instance
+    registered = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                             content=make_wav(2)).json()
+    # 自定义音色未试听 → 拒绝
+    unpreviewed = client.post("/api/voice/selection", json={"voice": registered["id"]})
+    assert unpreviewed.status_code == 409 and "试听" in unpreviewed.json()["detail"]
+    # 未知音色 → 404
+    assert client.post("/api/voice/selection", json={"voice": "ghost"}).status_code == 404
+    # 内置音色（D1 候选）不强制试听
+    builtin = client.post("/api/voice/selection", json={"voice": "rain"})
+    assert builtin.status_code == 200
+    assert json.loads(worker_state.read_text())["voice"] == "rain"
+    # 试听后自定义音色可应用
+    assert client.post("/api/voice/preview", json={"voice": registered["id"]}).status_code == 200
+    applied = client.post("/api/voice/selection", json={"voice": registered["id"]})
+    assert applied.status_code == 200 and applied.json()["label"] == registered["id"]
+
+
+def test_apply_rollback_on_state_write_failure(tmp_path, stack):
+    from fastapi.testclient import TestClient
+
+    (tmp_path / "blocker").write_text("not a dir")
+    app, custom, _ = build_app(tmp_path, stack)
+    custom.config.worker_state_file = str(tmp_path / "blocker" / "worker.json")
+    with TestClient(app) as client:
+        client.post("/api/companion", json=CREATE)
+        assert client.post("/api/voice/selection", json={"voice": "rain"}).status_code == 503
+        assert custom.selection().voice is None
+        assert client.get("/api/voice/selection").json()["voice"] is None
+
+
+def test_selection_persists_across_restart(tmp_path, stack):
+    from fastapi.testclient import TestClient
+
+    app, custom, worker_state = build_app(tmp_path, stack)
+    with TestClient(app) as client:
+        client.post("/api/companion", json=CREATE)
+        registered = client.post("/api/voice/voices",
+                                 params={"label": "留声", "prompt_text": "x"},
+                                 content=make_wav(2)).json()
+        client.post("/api/voice/preview", json={"voice": registered["id"]})
+        client.post("/api/voice/selection", json={"voice": registered["id"]})
+    # “重启”：同一批文件新建服务
+    app2, custom2, _ = build_app(tmp_path, stack)
+    with TestClient(app2) as client2:
+        client2.post("/api/companion", json=CREATE)
+        selection = client2.get("/api/voice/selection").json()
+        assert selection["voice"] == registered["id"]
+        listing = client2.get("/api/voice/voices").json()
+        entry = [v for v in listing["voices"] if v["id"] == registered["id"]][0]
+        assert entry["label"] == "留声" and entry["selected"] is True
+        assert entry["previewed"] is True  # 试听凭据也在持久流水里
+        # 重启后无需重新试听即可再应用
+        other = client2.post("/api/voice/voices", params={"prompt_text": "y"},
+                             content=make_wav(2)).json()
+        client2.post("/api/voice/preview", json={"voice": other["id"]})
+        assert client2.post("/api/voice/selection", json={"voice": other["id"]}).status_code == 200
+    assert json.loads(Path(worker_state).read_text())["voice"] == other["id"]
+
+
+def test_session_captures_voice_and_pins_for_the_call(instance):
+    client, stack, custom, worker_state = instance
+    registered = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                             content=make_wav(2)).json()
+    client.post("/api/voice/preview", json={"voice": registered["id"]})
+    client.post("/api/voice/selection", json={"voice": registered["id"]})
+    opened = client.post("/api/voice/session").json()
+    assert opened["voice"] == registered["id"]
+    assert json.loads(worker_state.read_text())["voice"] == registered["id"]
+    service_session = client.app.state.voice_service.get(opened["session_id"])
+    assert service_session.voice == registered["id"]
+    # 通话中改选：本通话定住，最迟下一通话生效
+    client.post("/api/voice/selection", json={"voice": "rain"})
+    assert client.app.state.voice_service.get(opened["session_id"]).voice == registered["id"]
+    second = client.post("/api/voice/session").json()
+    assert second["voice"] == "rain"
+
+
+def test_session_capture_falls_back_when_selection_invalid(instance):
+    client, stack, custom, worker_state = instance
+    registered = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                             content=make_wav(2)).json()
+    client.post("/api/voice/preview", json={"voice": registered["id"]})
+    client.post("/api/voice/selection", json={"voice": registered["id"]})
+    stack.ids.remove(registered["id"])  # 音色从注册表消失
+    client.app.state.voice_service._tts_health.checked = -1e9  # 重新探测拿新名单
+    opened = client.post("/api/voice/session").json()
+    assert opened["voice"] == "default"  # 诚实回落，不冒充原音色
+
+
+def test_session_without_selection_keeps_operator_voice(instance):
+    client, stack, custom, worker_state = instance
+    worker_state.write_text(json.dumps({"voice": "rain", "operator_note": "d1"}), encoding="utf-8")
+    opened = client.post("/api/voice/session").json()
+    assert opened["voice"] == "rain"  # D1/部署域现值如实生效
+    assert json.loads(worker_state.read_text()) == {"voice": "rain", "operator_note": "d1"}
+
+
+def test_shared_generic_state_never_touched(instance):
+    client, stack, custom, worker_state = instance
+    registered = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                             content=make_wav(2)).json()
+    client.post("/api/voice/preview", json={"voice": registered["id"]})
+    client.post("/api/voice/selection", json={"voice": registered["id"]})
+    client.post("/api/voice/session")
+    assert stack.generic_voice_writes == 0  # 通用助手 voice_state 从不写
+    assert stack.deletes == []              # 既有音色从不删
+    assert all(upload["name"].startswith("cg-") for upload in stack.uploads)
+    assert not any(v in stack._registry() for v in BUILTIN)  # 11 支既有条目原样
+
+
+def test_no_identity_rewrite(instance):
+    client, stack, custom, worker_state = instance
+    app = client.app
+    registered = client.post("/api/voice/voices", params={"prompt_text": "x"},
+                             content=make_wav(2)).json()
+    client.post("/api/voice/preview", json={"voice": registered["id"]})
+    before = {
+        "identity": app.state.store.query_one("SELECT identity_json FROM companion WHERE id=1"),
+        "bazi": app.state.store.query_one("SELECT pillars,derived FROM bazi_core WHERE id=1"),
+        "messages": app.state.store.query("SELECT id,role,text,turn_id,status FROM chat_message"),
+    }
+    client.post("/api/voice/selection", json={"voice": registered["id"]})
+    client.post("/api/voice/session")
+    after = {
+        "identity": app.state.store.query_one("SELECT identity_json FROM companion WHERE id=1"),
+        "bazi": app.state.store.query_one("SELECT pillars,derived FROM bazi_core WHERE id=1"),
+        "messages": app.state.store.query("SELECT id,role,text,turn_id,status FROM chat_message"),
+    }
+    assert before == after  # 换声音只换发声：底色/名字/关系/聊天/记忆一概不动
+
+
+def test_unconfigured_custom_endpoints_honest(tmp_path):
+    from fastapi.testclient import TestClient
+
+    provider = R2LLM()
+    router = build_router(providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)})
+    app = create_app(db_path=":memory:", router=router,
+                     life_config=LifeConfig(enabled=False), semantic_mode="stub")
+    with TestClient(app) as client:
+        listing = client.get("/api/voice/voices")
+        assert listing.status_code == 200 and listing.json()["configured"] is False
+        for method, path, kwargs in (
+                ("post", "/api/voice/voices", {"content": make_wav(2)}),
+                ("post", "/api/voice/preview", {"json": {"voice": "rain"}}),
+                ("post", "/api/voice/selection", {"json": {"voice": "rain"}})):
+            response = getattr(client, method)(path, **kwargs)
+            assert response.status_code == 503
+        assert client.get("/api/voice/selection").status_code == 503