diff --git a/config/voice.example.yaml b/config/voice.example.yaml new file mode 100644 index 0000000..6532ffb --- /dev/null +++ b/config/voice.example.yaml @@ -0,0 +1,27 @@ +# CyberGril 语音会话端点脱敏配置示例（HEL-877，R5 后台施工） +# +# 复制为 config/voice.yaml（或用 CYBERGRIL_VOICE_CONFIG 指定路径）后由总管 +# 按部署值填写；api_key/api_secret 只写服务器本地文件，不进仓库、不进浏览器、 +# 不进任何日志。环境变量 CYBERGRIL_VOICE_* 可覆盖同名字段。 +# +# 没有这份文件 = 本实例未配置语音：POST /api/voice/session 如实 503 拒绝， +# /api/health 的 capabilities.voice 保持 false，不伪造已接通。 + +# LiveKit SFU（.12 已有 docker voice-livekit；浏览器接入地址） +livekit_url: ws://192.168.200.12:7880 +# 服务端 Twirp 调用地址（本机回环即可） +livekit_internal: http://127.0.0.1:7880 +# LiveKit API 凭据（与 .12 /opt/voice-agent/.env 中 LIVEKIT_KEYS 同源；此处只放占位符） +api_key: +api_secret: + +# 专属 worker 的 agent_name：与通用 cascade-voice-agent 区分派发，互不争抢 +agent_name: cybergril-voice-agent + +# CosyVoice TTS 健康探测（TTS 不在就诚实拒绝开通话，不假装能说话） +tts_health_url: http://127.0.0.1:9880/health + +# 浏览器接入 token 有效期（秒）：只够加入房间，短时限 +token_ttl_seconds: 300 +# 会话登记有效期（秒）：房间早空/挂断后登记自然过期 +session_ttl_seconds: 7200 diff --git a/deploy/voice/restart_cybergril.sh.example b/deploy/voice/restart_cybergril.sh.example new file mode 100644 index 0000000..9fdddac --- /dev/null +++ b/deploy/voice/restart_cybergril.sh.example @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# CyberGril 专属语音 worker 运维脚本草案（HEL-877，本卡未执行）。 +# 用法: bash /opt/voice-agent-cybergril/restart_cybergril.sh [stop|status] +# 边界: 只管本 worker（python -m cybergril.voice.worker），绝不匹配通用 +# worker 的 `agent.py start`，不动 /opt/voice-agent 与 8090/8081。 +set -u +BASE=/opt/voice-agent-cybergril +cd "$BASE" +VENV=/opt/voice-agent/agent/venv # 复用已装 livekit-agents 1.8.8 的 venv +PAT='cybergril.voice.worker start' # 精确匹配本 worker 命令行 + +for p in $(pgrep -f "$PAT"); do kill "$p" 2>/dev/null; done +sleep 2 +for p in $(pgrep -f "$PAT"); do kill -9 "$p" 2>/dev/null; done + +if [ "${1:-}" = "stop" ]; then echo "cybergril voice worker stopped"; exit 0; fi +if [ "${1:-}" = "status" ]; then + pgrep -af "$PAT" || echo "not running" + curl -s --max-time 3 "http://127.0.0.1:${CYBERGRIL_WORKER_HEALTH_PORT:-8082}/" \ + && echo " <- health" || echo "health port not answering" + exit 0 +fi + +set -a; source "$BASE/.env"; set +a +PYTHONPATH="$BASE" nohup "$VENV/bin/python" -m cybergril.voice.worker start > worker.log 2>&1 & +echo "cybergril voice worker starting (pid $!)" +sleep 18 +grep -m1 "registered worker" worker.log >/dev/null && echo "OK: worker registered" \ + || { echo "FAIL:"; tail -5 worker.log; exit 1; } diff --git a/deploy/voice/serve_dual_8360_8361.py b/deploy/voice/serve_dual_8360_8361.py new file mode 100644 index 0000000..567827f --- /dev/null +++ b/deploy/voice/serve_dual_8360_8361.py @@ -0,0 +1,66 @@ +"""统一 CyberGril 实例单进程双入口启动器（HEL-877 部署草案，本卡未执行）. + +HTTP 8360 与 HTTPS 8361（语音页需要 secure context 才能拿麦克风）必须共用 +同一个 API 进程、同一份 app.state（epoch / chat_lock / store / 语音会话登记）： +本启动器在一个事件循环里跑两个 uvicorn Server，指向同一个 FastAPI 应用对象， +不是两个实例。自签证书签发命令与回滚步骤见 docs/runbook-hel877-voice.md。 + +用法（部署卡执行，先 `cd` 到仓库根并激活 3.11 venv）: + PYTHONPATH=src python deploy/voice/serve_dual_8360_8361.py \ + --http-port 8360 --https-port 8361 \ + --cert /opt/cybergril-voice-certs/cert.pem --key /opt/cybergril-voice-certs/key.pem + +本地离线冒烟（不涉及任何现网/证书变更）: + PYTHONPATH=src python deploy/voice/serve_dual_8360_8361.py \ + --http-port 18360 --https-port 18361 --cert <本地测试证书> --key <本地测试密钥> +""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import os +import sys + +import uvicorn + +from cybergril.api.app import create_app + + +async def serve(http_port: int, https_port: int, cert: str, key: str) -> None: + # ONE app object: both servers share state, locks, epoch and the store. + app = create_app() + http_server = uvicorn.Server( + uvicorn.Config(app, host="0.0.0.0", port=http_port, log_level="info") + ) + https_server = uvicorn.Server( + uvicorn.Config(app, host="0.0.0.0", port=https_port, ssl_certfile=cert, + ssl_keyfile=key, log_level="info") + ) + servers = [asyncio.create_task(http_server.serve()), asyncio.create_task(https_server.serve())] + done, pending = await asyncio.wait(servers, return_when=asyncio.FIRST_COMPLETED) + for task in pending: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + for task in done: + task.result() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--http-port", type=int, default=8360) + parser.add_argument("--https-port", type=int, default=8361) + parser.add_argument("--cert", required=True) + parser.add_argument("--key", required=True) + parser.add_argument("--db", default=os.environ.get("CYBERGRIL_DB")) + args = parser.parse_args() + if args.db: + os.environ["CYBERGRIL_DB"] = args.db + asyncio.run(serve(args.http_port, args.https_port, args.cert, args.key)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/deploy/voice/worker.env.example b/deploy/voice/worker.env.example new file mode 100644 index 0000000..c590abc --- /dev/null +++ b/deploy/voice/worker.env.example @@ -0,0 +1,18 @@ +# CyberGril 语音 worker 脱敏环境变量示例（HEL-877） +# +# 部署位置 /opt/voice-agent-cybergril/.env（权限 600，不进仓库）。 +# 复用 /opt/voice-agent/agent/venv（livekit-agents 1.8.8）与 .12 现有 +# LiveKit(7880)/FunASR(10095)/CosyVoice(9880)；不新装任何依赖。 + +# 统一 CyberGril 实例（HTTP 8360 / HTTPS 8361 同进程双入口） +CYBERGRIL_API_BASE=http://127.0.0.1:8360 +# 专属派发名（与 config/voice.yaml 的 agent_name 一致） +CYBERGRIL_AGENT_NAME=cybergril-voice-agent +# worker 注册 LiveKit 所需凭据（与通用 worker restart 脚本同源；只放占位符） +LIVEKIT_URL=ws://127.0.0.1:7880 +LIVEKIT_API_KEY= +LIVEKIT_API_SECRET= +# 专属健康端口（livekit-agents worker 内建 GET / 健康检查；8081 是通用 worker） +CYBERGRIL_WORKER_HEALTH_PORT=8082 +# CyberGril 自己的音色状态文件（不读通用助手 voice_state.json） +CYBERGRIL_VOICE_STATE=/opt/voice-agent-cybergril/voice_state_cybergril.json diff --git a/docs/requirements/handoff-HEL-877.json b/docs/requirements/handoff-HEL-877.json new file mode 100644 index 0000000..7538b86 --- /dev/null +++ b/docs/requirements/handoff-HEL-877.json @@ -0,0 +1,54 @@ +{ + "baseline_version": "1.11", + "read_commit": "31970c4dd3851aae67a922b03264743d504b8e97", + "task": "HEL-877：把已核清的语音底座接到统一 CyberGril 业务链（获批 R5 后台施工）；新增专属 livekit worker（agent_name=cybergril-voice-agent，复用 .12 FunASR/CosyVoice/silero VAD，不 fork 不另搭平台）+ 统一实例 POST /api/voice/session 最小协议（建房/专属派发/短时限 token 服务端处理，密钥不进浏览器）+ 与文字前端同协议的 livekit LLM 适配器（POST /api/chat，SSE/epoch 409 有界重同步/turn_id 幂等/打断即断流）+ SDK 真实播放完成才确认的分享回执；零真实请求自测交可审核代码", + "requirement_ids": [ + "CG-006", + "CG-010", + "CG-011", + "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/ 独立模块：session.py+router.py（POST /api/voice/session、DELETE 结束删房释放麦克风/job、GET /api/voice/status 诚实开关；缺配置/无档案/TTS 不可用如实拒绝；app.py 仅最小注册且 capabilities.voice 不伪造）、livekit_wire.py（零新依赖 HS256 token+Twirp 派发/删房，与已装 SDK 同线上格式）、chat_api_client.py（与 real.ts 同协议：health epoch、SSE 分片鲁棒解析、409 一次重试、固定 turn_id、打断 finally 即断流）、llm_chat_api.py（livekit llm.LLM 适配器，只发当前用户话轮，服务端组装全部人格/记忆/关系；φ 原句经 ChatChunk 念出无回执）、receipt.py（should_ack 只有 played_to_end 且未打断未失败且带完整 receipt 且 epoch 一致才确认；TurnLedger 绑定 speech handle↔流；确认恰好一次）、stt_funasr.py/tts_cosyvoice.py（.12 已验证适配器原样复用带出处）、worker.py（agent_name=cybergril-voice-agent、独立 voice_state_cybergril.json 音色、无第二人格、cg.subtitle 字幕数据通道、CG-011 主动开口空操作接缝、专属健康端口 8082）；config/voice.example.yaml 与 deploy/voice/{worker.env.example,restart_cybergril.sh.example,serve_dual_8360_8361.py} 部署草案（HTTP 8360+HTTPS 8361 同进程双入口共用同一状态/锁，runbook 含启动顺序/回滚/GPU 前置）；docs/runbook-hel877-voice.md 契约+部署文档；tests/test_voice_session.py(9)+test_voice_chat_adapter.py(16)、scripts/verify_hel877_voice_chain.py、scripts/gen_hel877_sdk_check.py", + "acceptance": [ + "语音轮次走统一链被证明：worker 适配器只调 POST /api/chat，文字/语音互相打断由服务端同一 chat_lock 轮次门处理（本地 socket 实测：语音打断文字轮旧行=cancelled 无幽灵行、打断取消即刻断 HTTP 流旧行 fail-closed、新轮干净）", + "完成轮重放不新增模型调用；409 旧 epoch 一次有界重同步后成功；SSE 多字节/CRLF/多行 data 分片边界解析正确；turn_id 固定使 SDK 内部重试幂等", + "分享 receipt 仅在 SDK 真实播放完成后确认：should_ack 对被打断/播放失败/未播完/φ 轮/残缺回执/旧 epoch 全负例拒绝（测试覆盖），确认恰好一次，服务端仍独立 fail-closed 校验", + "POST /api/voice/session 服务端建房+专属派发（agent_name=cybergril-voice-agent，protobuf CreateDispatch 与已装 SDK 同格式）+短时限 token（TTL≤300s，与 livekit.api.AccessToken 声明集全等，.12 真 SDK 比对通过）；房间隔离（两 session 不同房、token 各自房间）；DELETE 删房结束；缺配置/无档案/TTS 停机诚实拒绝且 capabilities.voice 保持 false", + "worker 不持有第二人格/记忆、不直连 LLM、不读通用助手 persona_state/voice_state（独立 voice_state_cybergril.json，D1 未拍板回落 default）；CG-011 语音主动开口留明确空操作接缝并写清边界（仅已开启会话、同业务链、可打断、不新回执通道）", + "真 SDK 兼容在 .12 agent venv（livekit-agents 1.8.8）stdin-only 零写入零网络验证 8 项全过（真 LLM/LLMStream 子类、真 ChatContext、真流机制 ChatChunk+final+receipt、φ ChatChunk、token 声明集与 SDK AccessToken 全等、server token 同形、SpeechHandle/事件 API 存在、打断取消立即关流）", + "双入口同进程：本地 18360/18361 冒烟单 pid 双端口同 epoch 两端诚实拒答；HTTP 与 HTTPS 共用同一 app 对象/状态/锁，无第二实例写同一 DB", + "部署草案齐全（/opt/voice-agent-cybergril、健康端口 8082、启动顺序含 TTS GPU≥6GB 前置、回滚、禁抄历史 fuser 命令的明确警告）；本卡未执行任何部署/启停/下载安装，未动 web/、三账、档案、4080、现网服务" + ], + "deferred": [ + "R5 真机验收：连续 3 轮、开口打断 ~1s 停声、结束停麦无残音、移动端 Safari/Chrome 麦克风——TTS 恢复（GPU 需 ≥5.6GB，现 3.9GB，ComfyUI 不停）并接线后在后续卡执行；未经真实验收不得称 R5 完成", + "视觉总监接线：语音按钮/字幕(cg.subtitle)/失败/打断/挂断按 runbook §3 契约；本卡未改 web/", + "总工独立审核与部署（runbook §4）：.12 上对真 LiveKit server 的 dispatch/delete、HTTPS 证书签发与实机信任（每台手机高级→继续）待实测，不承诺每台手机点继续必然可用", + "语音内主动开口（CG-011 接缝）与 D1 音色最终决定、D3 用户设备动作、GPU 时段：总管集中与白栖知沟通", + "R6 天气/素材/完整备份承接不变；视觉观察默认关闭不变" + ], + "requirement_changes": [], + "verification": [ + "PYTHONPATH=src .venv311/bin/python -m pytest tests -q：281 passed（总工 HEL-874 返工后全量 256 + 本卡 25，含语音会话 9 与适配器/回执 16，全 fake HTTP/SSE/本地 127.0.0.1 socket，零真实模型/嵌入/合成/识别）", + "scripts/verify_hel877_voice_chain.py 一键全链 8 项全过（真实 app + fake LiveKit/TTS + 本地 socket：拒答序/专属派发/短时限 token/房间隔离/删房结束/多轮回执/重放零调用/409 重同步/打断 fail-closed/播放完成回执一次/capabilities.voice=false）", + "scripts/gen_hel877_sdk_check.py 生成脚本经 SSH stdin 在 .12 /opt/voice-agent/agent/venv/bin/python（livekit-agents 1.8.8、livekit 1.1.20）执行：8 项 ALL PASS（零写入、零网络、未触任何服务）", + "双入口本地冒烟：一次性自签证书（本机 openssl 生成）+ deploy/voice/serve_dual_8360_8361.py 于 18360/18361，单 pid、同 epoch、/api/voice/status 诚实、未配置 POST 503；测后进程与端口已释放", + "ruff check src tests scripts deploy：本卡代码 0 错（唯一遗留 SIM115 属既有 deploy/hel835_guard.py，非本卡）；check_requirements.py 双目标（origin/hel-870-assembly 现tip e978e87 与 origin/main）含本 handoff 通过", + "读取位置记录：开工时 origin/hel-870-assembly=31970c4 与卡文一致；施工中总工推入 HEL-874（e978e87），本卡已完整 merge 保其成果不覆盖；origin/hel-876-voice-readiness=23415fb（总管读取点）已并入", + "未验证如实声明：端到端语音（TTS 停机未连麦）、真 LiveKit server 的 dispatch/delete 实测、worker 运行时行为（字幕/VAD/回声抑制仅源码级对齐 .12 已验证实现与 SDK 静态契约）、移动端实机" + ] +} diff --git a/docs/runbook-hel877-voice.md b/docs/runbook-hel877-voice.md new file mode 100644 index 0000000..38a05e1 --- /dev/null +++ b/docs/runbook-hel877-voice.md @@ -0,0 +1,155 @@ +# HEL-877 语音契约与部署草案（R5 后台施工，未部署未执行） + +> 性质：**产品代码 + 契约文档 + 部署草案**。本卡零真实 LLM/嵌入/TTS 合成/ASR 识别请求、 +> 未连麦、未下载安装、未启停任何现网服务、未访问 4080、未动三账与已有档案、未改 web/。 +> 基准 v1.11；开工时原基点 `origin/hel-870-assembly` @ `31970c4dd3851aae67a922b03264743d504b8e97`； +> 施工中总工把 HEL-874 返工推上同分支（新 tip `e978e878dfc34f38c781d48f87b93dc21bbf8a61`）， +> 本卡已完整并入（不覆盖并行成果），并入后全量测试 281 通过；总管读取点 +> `origin/hel-876-voice-readiness` @ `23415fbcd00508a1a590d3b0fc8e0845715b12fa`（仅文档/脚本 +> 三文件）也已并入。 + +## 1. 交付概览 + +- `src/cybergril/voice/` 独立模块： + - `session.py` + `router.py`：统一实例上的 `POST /api/voice/session` / `DELETE /api/voice/session/{id}` / `GET /api/voice/status`（app.py 仅最小注册）。 + - `livekit_wire.py`：不新增依赖的 LiveKit 接线——HS256 token 铸造 + Twirp CreateDispatch/DeleteRoom（与 .12 已装 SDK 的 `AccessToken`/`agent_dispatch` 同一线上格式，见 §5 真机核对）。 + - `chat_api_client.py`：与文字前端 `real.ts` 完全同协议的 `/api/chat` 客户端（epoch 同步、SSE 分片、409 有界重试、turn_id、receipt、打断即断流）。 + - `llm_chat_api.py`：livekit `llm.LLM` 适配器，worker 的每一句回复都来自统一链；`receipt.py`：**只有 SDK 真实播放完成**才确认分享 receipt。 + - `stt_funasr.py` / `tts_cosyvoice.py` / `worker.py`：复用 .12 已验证底座组装的专属 worker（`agent_name=cybergril-voice-agent`，独立音色状态，无第二人格）。 +- 测试：`tests/test_voice_session.py`、`tests/test_voice_chat_adapter.py`（fake HTTP/SSE/本地 socket）； + `scripts/verify_hel877_voice_chain.py` 一键全链检查；`scripts/gen_hel877_sdk_check.py` 生成在 + .12 agent venv（livekit-agents 1.8.8）里跑的真 SDK 兼容检查。 +- 部署草案：`config/voice.example.yaml`、`deploy/voice/worker.env.example`、`deploy/voice/serve_dual_8360_8361.py`、本 runbook。 + +## 2. 语音业务链（她还是同一位她） + +语音轮次 **不是** 第二套对话：worker 的 LLM 适配器只发 `POST /api/chat {text, turn_id, epoch}`， +人格/情绪/关系/记忆/生活分享/调用限制/轮次门全部在统一实例服务端（`src/cybergril/api/app.py:488 chat()`）。 +文字与语音争用同一把 `chat_lock`：新轮（无论来自哪个入口）自动取消活动轮，旧行标 `cancelled`。 +- **epoch**：worker 每轮先 `GET /api/health` 取 epoch；409 → 采用响应里的新 epoch 重试一次（有界，同 real.ts）。 +- **打断**：VAD 判定用户开口 → AgentSession 取消生成任务 → 适配器 `finally` 立即关闭 HTTP 流 + （不等下一次 ASR final），服务端随即把旧轮 user 行标 `cancelled`，不留幽灵 companion 行。 +- **幂等**：每个流固定 turn_id；完成轮重放只回存档 final，模型调用数不变。 +- **φ 失败**：`chat.error` 的原句由 TTS 原样念出，不新造话术，不落 companion 行、不带回执。 +- **回执**：`chat.final.receipt` 只在 speech handle `wait_for_playout()` 完成、未 `interrupted`、 + 无异常、轮次完整时 POST `/api/chat/received`（`receipt.py should_ack` 全负例有测试）； + 服务端仍独立校验 token/hash/过期（fail-closed 双保险）。 +- **主动开口（CG-011 接缝）**：`worker.py _maybe_open_proactive()` 为空操作占位，契约见 §3.6； + 要求：仅已开启会话内、只念服务端已落库 proactive 行、可打断、不新造第二条回执通道。 + +## 3. 前端契约（交视觉总监接线） + +### 3.1 开关 +- `GET /api/voice/status` → `{configured, tts_ok, agent_name, voice, ...}`。`configured=false` + 或 `tts_ok=false` 时按钮如实不可用/隐藏；`voice` 恒为 false（R5 未真机验收，不得显示"语音已接通"）。 +- `/api/health` 的 `capabilities.voice` 保持 false，本卡不伪造。 + +### 3.2 开通（密钥不进浏览器） +- `POST /api/voice/session` → 201 `{url, token, room, session_id, agent_name, epoch, expires_at}`； + token 短时限（默认 300s）只够加入房间。livekit-client 用 `url+token` 连接。 +- 诚实拒绝：未配置 503「语音还没有在这台服务上配置…」；无档案 409；TTS 不在 503「她的声音服务这会儿不在…」； + LiveKit 不通 503「通话服务暂时连不上…」。直接展示 detail 文案即可。 + +### 3.3 字幕 +- worker 在房间数据通道 topic `cg.subtitle` 发布 JSON： + `{kind: "interim"|"user_final"|"assistant_final"|"error", role: "user"|"assistant", text}`。 + interim 是 ASR 草稿（可频繁覆盖），user_final 是纠错终稿，assistant_final 是她的话（已落库）， + error 是 φ 原句（未落库，前端不得当作她说过的话进入正式记录）。 + +### 3.4 失败 / 打断 / 挂断 +- 失败：听到 φ + `error` 字幕；草稿保留；用户可开口重试（新 turn）。 +- 打断：用户直接开口（VAD ~0.55s 判定链路沿用）；旧语音被截断，旧轮 fail-closed 不回执。 +- 挂断：`DELETE /api/voice/session/{session_id}` → 服务端删房 → 浏览器与 worker 双双断开、 + 任务退出、麦克风释放；浏览器直接关页同样生效（participant disconnect → job 退出，10-24 实测路径）。 + 前端仍应 stop 本地 track 立即停麦。 + +### 3.5 会话边界 +- 一房一会话；重复 POST 得到新房间新 token（旧房不受影响）。结束/失效后 session_id 再用 → 404。 +- 会话登记内存态、随进程；实例重启后房间自然消亡，前端按 503/404 如实处理。 + +### 3.6 主动开口接缝（本卡不实现） +- 后续接线卡方案：会话开启后 worker 经统一 API 取「已落库、proactive=1、未读」的 companion 行， + `session.say()` 原文念出（allow_interruptions 开启），可被打断；不新起模型调用、不伪造新经历、 + 不新建回执通道（文字回执规则独立、已存在）。呈现节奏由总管/视觉总监定，未经批准不启用。 + +## 4. 部署草案（本卡未执行；总工执行，逐条核对授权） + +### 4.1 目录与文件（/opt/voice-agent-cybergril/） +``` +/opt/voice-agent-cybergril/ + worker -> 仓库 src/cybergril/voice/ 的 worker.py/stt_funasr.py/tts_cosyvoice.py/llm_chat_api.py/chat_api_client.py/receipt.py + .env # deploy/voice/worker.env.example 填真值（600） + voice_state_cybergril.json # {"voice": "default"} D1 拍板前不改 + restart_cybergril.sh # 参照 /opt/voice-agent/restart_agent.sh 的安全模式（见 4.3） +``` +复用：`/opt/voice-agent/agent/venv`（livekit-agents 1.8.8）、LiveKit 7880、FunASR 10095、 +CosyVoice 9880。**不改 /opt/voice-agent 任何文件，不动 8090 与通用 worker 8081。** + +### 4.2 启动顺序（部署卡执行） +1. 统一实例（8360/8361 同进程双入口，见 4.4）； +2. TTS 9880（**前置 GPU≥6GB 空闲**——当前 3.9GB 不够，等 ComfyUI 释放时段，由总管协调； + 重启命令在 /opt/voice-agent/cosyvoice README 运维节，**HEL-876 文档中的 `fuser -k 9880/tcp` + 是历史命令，禁止照抄**，必须先核对具体进程与授权）； +3. `bash /opt/voice-agent-cybergril/restart_cybergril.sh`（专属 worker，健康端口 8082）； +4. 手机/电脑首次访问 HTTPS 入口 → 自签证书「高级→继续」→ 允许麦克风（D3 用户动作，总管转达）。 + +### 4.3 worker 运维（草案；8081 通用 worker 绝不动） +- 健康：worker 内建 HTTP 检查 `GET :8082/` → `OK`（503=推理进程挂/连不上 LiveKit）；`GET :8082/worker` 看注册信息。 +- 重启脚本照抄 restart_agent.sh 的模式：`pgrep -f` **精确匹配本 worker 的独特命令行** + （如 `python -m cybergril.voice.worker start`，与通用 worker 的 `agent.py start` 不同名，互不误杀）， + 再启动；日志 `nohup … > worker.log`；启动成功判据 = 日志 `registered worker`。 +- 停止仅停本 worker；回滚不影响语音底座其他组件。 + +### 4.4 HTTP 8360 与 HTTPS 入口同进程（硬要求） +`deploy/voice/serve_dual_8360_8361.py` 在一个事件循环里跑两个 uvicorn Server 指向**同一个** +FastAPI 应用对象：同一 epoch、同一 `chat_lock`、同一 store、同一语音会话登记—— +绝不许两个写同一 DB 的独立实例。自签证书按 8090 已验证流程签一张覆盖 `192.168.200.12` +的证书（ openssl req -x509 … subjectAltName=IP:192.168.200.12 ），存 +`/opt/cybergril-voice-certs/`（600）。本卡已在本地 18360/18361 用一次性自签证书冒烟验证 +同进程双入口（单 pid、同 epoch、两端诚实拒答），**未在 .12 执行**。 + +### 4.5 回滚 +1. 停专属 worker（4.3）；2. API 侧移除/改名 `config/voice.yaml` → `/api/voice/*` 如实 503， + 其余功能不受影响；3. 如需整体回退：停双入口进程 → 数据目录原样保留 → 切回上一 release。 + 语音不写任何新表：会话登记在内存，receipt 走既有 sharing_attempt，无迁移无残留。 + +## 5. 自测与证据（全部零真实请求） + +- `PYTHONPATH=src .venv311/bin/python -m pytest tests -q`：**281 通过**（并入总工 HEL-874 + 返工后的全量 256 + 本卡 25：语音会话 9 + 适配器/回执 16；含本地 socket 级打断断流、 + SSE 分片边界（多字节/CRLF/多行 data）、完成轮重放零调用、409 有界重同步、文字/语音互打断 + 同门、播放完成前不回执全负例、TTS 不在拒绝、专属派发与 token 房间隔离、登记过期清理）。 +- `scripts/verify_hel877_voice_chain.py`：一键全链 8 项全过（真实 app + fake router/LiveKit/TTS， + 本地 socket；拒答序、专属派发、短时限 token、房间隔离、删房结束、多轮+回执、重放零调用、 + 409 重同步、打断 fail-closed、播放完成回执一次、capabilities.voice=false）。 +- **真 SDK 兼容（.12，stdin-only，零写入零网络）**：`scripts/gen_hel877_sdk_check.py` 生成脚本在 + `/opt/voice-agent/agent/venv/bin/python`（livekit-agents 1.8.8/livekit 1.1.20）跑通 8 项： + 真 `llm.LLM/LLMStream` 子类契约、真 `ChatContext` 读取、真 LLMStream 机制下发 ChatChunk+final+receipt、 + φ 原句经 ChatChunk 念出无回执、`mint_token` 与真 `livekit.api.AccessToken` 声明集一致 + （video/iss/sub/键集合全等）、server token 同 SDK 无授权形态、SpeechHandle/UserInputTranscribedEvent + API 面存在、打断任务取消即刻关闭 HTTP 流。 +- `ruff check src tests scripts deploy`：0 错。 +- 双入口本地冒烟：单 pid 同时听 18360(HTTP)/18361(HTTPS)，同 epoch、两端 `/api/voice/status` 诚实、 + 未配置 POST 503。证书为本机一次性自签，未触任何现网。 +- `python3 scripts/check_requirements.py --base-ref origin/hel-870-assembly --handoff docs/requirements/handoff-HEL-877.json` + 及 `--base-ref origin/main` 同跑通过。 + +## 6. 明确未验证 / 缺依赖如实清单 + +- **未在 .12 部署、未启动 worker、未发任何真实 dispatch**；twirp/jwt 线上格式与 SDK 声明集比对过， + 但对真 LiveKit server 的 CreateDispatch/DeleteRoom 请求要等部署卡实测。 +- **端到端语音未验证**（TTS 停机、GPU 3.9GB<5.6GB）：连续 3 轮、开口打断 ~1s 停声、结束停麦无残音、 + 移动端 Safari/Chrome 麦克风、TTS 恢复后首包延迟——全部留给接线/审核卡真机验收，未经真实验收 + **不得称 R5 完成**。 +- worker.py/stt/tts 的运行时行为（字幕数据消息、VAD 参数、回声抑制）只在源码级对齐 + .12 已验证实现与 SDK 接口，无法在本环境离线运行 livekit-agents——已用 §5 真 SDK 检查覆盖 + 静态契约，运行时行为属后续实测。 +- HTTPS 实机信任（每台手机"高级→继续"）与证书签发在 .12 未执行；不承诺每台手机点继续必然可用。 +- D1 音色（当前回落 `default`，不替用户定）、D3 用户动作、GPU 时段：总管集中与白栖知沟通。 + +## 7. 承接 + +- 视觉总监：语音按钮/字幕/失败/打断/挂断按 §3 契约接线（web/ 本卡未动）。 +- 总工：独立审核 + 部署（§4，含 GPU/TTS 时段）；对 origin/hel-870-assembly 开 PR，不合并。 +- 后续卡：R5 真机验收（四条）、语音内主动开口（§3.6 接缝）、R6 天气/素材/备份。 +- 无需求变更（v1.11 不变，无 GOV 记录新增）。 diff --git a/scripts/gen_hel877_sdk_check.py b/scripts/gen_hel877_sdk_check.py new file mode 100644 index 0000000..86a70f3 --- /dev/null +++ b/scripts/gen_hel877_sdk_check.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +"""Generate the HEL-877 real-SDK compatibility check for the .12 agent venv. + +Embeds the repo voice-module sources in a self-contained stdin script (no files +written to .12, no network, no service touched) and runs it with +/opt/voice-agent/agent/venv/bin/python (livekit-agents 1.8.8, livekit 1.1.20). +""" +from __future__ import annotations + +import base64 +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +MODULES = ["chat_api_client", "receipt", "llm_chat_api"] + +TEMPLATE = r''' +import asyncio, base64, json, sys, types + +SOURCES = json.loads(base64.b64decode("__B64__")) + +pkg = types.ModuleType("cybergril"); pkg.__path__ = [] +voice = types.ModuleType("cybergril.voice"); voice.__path__ = [] +sys.modules["cybergril"] = pkg +sys.modules["cybergril.voice"] = voice +for name, src in SOURCES.items(): + mod = types.ModuleType(f"cybergril.voice.{name}") + mod.__package__ = "cybergril.voice" + sys.modules[f"cybergril.voice.{name}"] = mod + exec(compile(src, f"", "exec"), mod.__dict__) + setattr(voice, name, mod) + +import httpx +from livekit.agents import llm as lk_llm +from livekit.agents.llm import LLM, LLMStream, ChatChunk, ChoiceDelta + +ok = [] + +# 1. real-SDK subclass contracts +llm_mod = sys.modules["cybergril.voice.llm_chat_api"] +assert issubclass(llm_mod.CyberGrillLLM, LLM), "CyberGrillLLM must subclass the real llm.LLM" +assert issubclass(llm_mod.ChatApiLLMStream, LLMStream) +ok.append("CyberGrillLLM/ChatApiLLMStream subclass real livekit-agents 1.8.8 LLM/LLMStream") + +# 2. real ChatContext -> latest_user_text +ctx = lk_llm.ChatContext() +ctx.add_message(role="assistant", content="早些的一句") +ctx.add_message(role="user", content="真实SDK链路你好") +assert llm_mod.latest_user_text(ctx) == "真实SDK链路你好" +ok.append("latest_user_text reads a real ChatContext (items/role/text_content)") + +# 3. full turn through the real LLMStream machinery with fake SSE +FINAL = "真实链路上的一句完整回复。" +sse = ( + 'data: {"t":"chat.delta","epoch":1,"turn_id":null,"delta":"真实链路上"}\n\n' + 'data: {"t":"chat.delta","epoch":1,"turn_id":null,"delta":"的一句完整回复。"}\n\n' + 'data: {"t":"chat.final","epoch":1,"turn_id":null,"utterance":{"text":"' + FINAL + '"},' + '"receipt":{"message_id":9,"turn_id":null,"epoch":1,"token":"' + "k" * 48 + '","text_hash":"' + "h" * 64 + '"}}\n\n' +).encode() + +def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200, json={"epoch": 1}) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=sse) + +receipt_ok = {} + +async def round_trip(): + client = llm_mod.ChatApiClient("http://x", transport=httpx.MockTransport(handler)) + adapter = llm_mod.CyberGrillLLM("http://x", client=client) + stream = adapter.chat(chat_ctx=ctx) + chunks = [] + async for chunk in stream: + assert isinstance(chunk, ChatChunk) and chunk.delta and chunk.delta.content + chunks.append(chunk.delta.content) + assert "".join(chunks) == FINAL, chunks + assert stream.completed and stream.receipt and stream.receipt["message_id"] == 9 + receipt_ok["text"] = stream.text + receipt_ok["receipt"] = stream.receipt + +asyncio.run(round_trip()) +assert receipt_ok["text"] == FINAL +ok.append("fake-SSE turn through real LLMStream emits ChatChunks, final text and receipt") + +# 4. phi error turn: spoken verbatim, no receipt +phi_sse = ( + 'data: {"t":"chat.error","epoch":1,"turn_id":null,"phi":"她这会儿没接上话，稍后再试试。你写的话还在。","retryable":true}\n\n' +).encode() + +def phi_handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200, json={"epoch": 1}) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=phi_sse) + +async def phi_turn(): + adapter = llm_mod.CyberGrillLLM("http://x", + client=llm_mod.ChatApiClient("http://x", transport=httpx.MockTransport(phi_handler))) + stream = adapter.chat(chat_ctx=ctx) + spoken = [] + async for chunk in stream: + spoken.append(chunk.delta.content) + assert "".join(spoken) == "她这会儿没接上话，稍后再试试。你写的话还在。" + assert stream.error_phi and not stream.completed and stream.receipt is None + +asyncio.run(phi_turn()) +ok.append("chat.error phi is spoken verbatim via ChatChunk and carries no receipt") + +# 5. token claims parity with the real livekit.api.AccessToken +from livekit import api +import jwt as pyjwt +wire = sys.modules["cybergril.voice.livekit_wire"] +mine = wire.mint_token("devkey", "s3cret", identity="user-ab", name="用户", + room="cg-voice-test", room_join=True, ttl_seconds=300, now=1700000000.0) +sdk_token = (api.AccessToken("devkey", "s3cret").with_identity("user-ab").with_name("用户") + .with_grants(api.VideoGrants(room_join=True, room="cg-voice-test", + can_publish=True, can_subscribe=True)) + .to_jwt()) +mine_claims = pyjwt.decode(mine, "s3cret", algorithms=["HS256"], options={"verify_exp": False}) +sdk_claims = pyjwt.decode(sdk_token, "s3cret", algorithms=["HS256"], options={"verify_exp": False}) +assert mine_claims["video"] == sdk_claims["video"], (mine_claims["video"], sdk_claims["video"]) +assert mine_claims["iss"] == sdk_claims["iss"] and mine_claims["sub"] == sdk_claims["sub"] +assert set(mine_claims) == set(sdk_claims), (sorted(mine_claims), sorted(sdk_claims)) +ok.append("mint_token claims byte-parity with livekit.api.AccessToken (video/iss/sub/keys)") + +# 6. server token decodes as the SDK Service._auth_header shape +server_tok = wire.mint_server_token("devkey", "s3cret", ttl_seconds=60, now=1700000000.0) +sc = pyjwt.decode(server_tok, "s3cret", algorithms=["HS256"], options={"verify_exp": False}) +assert sc["iss"] == "devkey" and sc["sub"] == "" and "video" not in sc +ok.append("mint_server_token matches SDK no-grant auth header claims") + +# 7. SpeechHandle/worker API surface used by worker.py exists on the real SDK +from livekit.agents.voice.speech_handle import SpeechHandle +for attr in ("wait_for_playout", "interrupted", "exception", "add_done_callback"): + assert hasattr(SpeechHandle, attr), attr +from livekit.agents.voice.events import UserInputTranscribedEvent +assert "is_final" in UserInputTranscribedEvent.model_fields +ok.append("worker.py SDK surface (SpeechHandle, UserInputTranscribedEvent.is_final) present") + +# 8. barge-in cancellation closes the SSE response immediately +state = {"response": None} + +async def slow_body(): + yield 'data: {"t":"chat.delta","epoch":1,"turn_id":null,"delta":"开头"}\n\n'.encode() + await asyncio.sleep(30) + yield 'data: {"t":"chat.final","epoch":1,"turn_id":null}\n\n'.encode() + +class TrackingTransport(httpx.MockTransport): + async def handle_async_request(self, request): + response = await super().handle_async_request(request) + if request.url.path == "/api/chat": + state["response"] = response + return response + +def slow_handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200, json={"epoch": 1}) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, + content=slow_body()) + +async def cancel_case(): + adapter = llm_mod.CyberGrillLLM("http://x", + client=llm_mod.ChatApiClient("http://x", transport=TrackingTransport(slow_handler))) + stream = adapter.chat(chat_ctx=ctx) + task = asyncio.create_task(stream._run()) + await asyncio.sleep(0.2) # first delta consumed + assert state["response"] is not None and not state["response"].is_closed + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + assert state["response"].is_closed, "adapter must close the HTTP stream on barge-in" + +asyncio.run(cancel_case()) +ok.append("barge-in task cancel closes the in-flight HTTP stream immediately") + +print("HEL-877 real-SDK check on livekit-agents", end=" ") +import importlib.metadata as md +print(md.version("livekit-agents") + ":", "ALL PASS") +for line in ok: + print(" OK", line) +''' + + +def main() -> int: + sources = {} + for name in MODULES + ["livekit_wire"]: + path = ROOT / "src/cybergril/voice" / f"{name}.py" + sources[name] = path.read_text(encoding="utf-8") + payload = base64.b64encode(json.dumps(sources).encode()).decode() + script = TEMPLATE.replace("__B64__", payload) + out = ROOT / "scripts/_hel877_sdk_check_payload.py" + out.write_text(script, encoding="utf-8") + print(out) + return 0 + + +if __name__ == "__main__": + import json + + sys.exit(main()) diff --git a/scripts/verify_hel877_voice_chain.py b/scripts/verify_hel877_voice_chain.py new file mode 100644 index 0000000..eb517a3 --- /dev/null +++ b/scripts/verify_hel877_voice_chain.py @@ -0,0 +1,235 @@ +"""HEL-877 offline full-chain verification (zero model / zero embed / zero TTS / +zero ASR / zero network beyond 127.0.0.1). + +Runs the REAL voice wiring end to end with fakes only: + real app (fake LLM router) + real /api/voice/session router (fake LiveKit + Twirp + fake TTS health) + real ChatApiClient over a real local socket + + real receipt gate — the same protocol the text frontend and the .12 worker + adapters speak. +Run: PYTHONPATH=src .venv311/bin/python scripts/verify_hel877_voice_chain.py +""" +from __future__ import annotations + +import asyncio +import contextlib +import socket +import sys +import tempfile +import threading +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tests")) + +import httpx +import uvicorn +from test_r2 import R2LLM + +from cybergril.api.app import create_app +from cybergril.life.runtime import LifeConfig +from cybergril.think.router import build_router +from cybergril.think.scenes import Tier +from cybergril.voice import livekit_wire +from cybergril.voice.chat_api_client import ChatApiClient +from cybergril.voice.receipt import should_ack +from cybergril.voice.session import VoiceConfig, VoiceSessionService + +CREATE = {"name": "链路检查", "birth": "2000-10-18", "hour": "亥时", "relation": "朋友"} + + +class FakeLiveKit: + def __init__(self) -> None: + self.dispatches: list[dict] = [] + self.deleted: list[str] = [] + self.fail = False + + def handler(self, request: httpx.Request) -> httpx.Response: + if self.fail: + return httpx.Response(500) + if request.url.path.endswith("CreateDispatch"): + agent, room = _dispatch_fields(request.content) + self.dispatches.append({"agent_name": agent, "room": room}) + return httpx.Response(200, json={"id": f"AD_{len(self.dispatches)}", "room": room}) + if request.url.path.endswith("DeleteRoom"): + self.deleted.append(_dispatch_fields(request.content)[0]) + return httpx.Response(200) + return httpx.Response(404) + + +def _read_varint(data: bytes, i: int): + shift = value = 0 + while True: + b = data[i] + value |= (b & 0x7F) << shift + i += 1 + if not b & 0x80: + return value, i + shift += 7 + + +def _dispatch_fields(data: bytes) -> tuple[str, str]: + out, i = {}, 0 + while i < len(data): + tag, i = _read_varint(data, i) + if tag & 7 != 2: + break + n, i = _read_varint(data, i) + out[tag >> 3] = data[i:i + n].decode() + i += n + return out.get(1, ""), out.get(2, "") + + +def main() -> int: + ok: list[str] = [] + livekit = FakeLiveKit() + tts_ok = {"flag": True} + provider = R2LLM() + + def transport(request: httpx.Request) -> httpx.Response: + if request.url.host == "tts.test": + return httpx.Response(200) if tts_ok["flag"] else httpx.Response(503) + return livekit.handler(request) + + service = VoiceSessionService( + config=VoiceConfig( + livekit_url="ws://voice.test:7880", + livekit_internal="http://livekit.test", + api_key="devkey", + api_secret="verify-secret-not-real", + tts_health_url="http://tts.test/health", + ), + http=httpx.AsyncClient(transport=httpx.MockTransport(transport)), + ) + router = build_router( + providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)} + ) + tmp = Path(tempfile.mkdtemp(prefix="hel877-")) + app = create_app(db_path=tmp / "chain.db", router=router, + life_config=LifeConfig(enabled=True, poll_seconds=3600), + semantic_mode="stub", voice_service=service) + + probe = socket.socket() + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + probe.close() + server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, + log_level="error")) + threading.Thread(target=server.run, daemon=True).start() + for _ in range(100): + try: + httpx.get(f"http://127.0.0.1:{port}/api/health", timeout=1) + break + except Exception: + time.sleep(0.1) + base = f"http://127.0.0.1:{port}" + + async def scenario() -> None: + # 1. honest refusals in order: no companion -> TTS down -> ok + http = httpx.AsyncClient(base_url=base) + refusal = await http.post("/api/voice/session") + assert refusal.status_code == 409, refusal.text + await http.post("/api/companion", json=CREATE) + tts_ok["flag"] = False + service._tts_health.checked = -1e9 + refusal = await http.post("/api/voice/session") + assert refusal.status_code == 503 and "声音服务" in refusal.json()["detail"] + tts_ok["flag"] = True + service._tts_health.checked = -1e9 + + # 2. dedicated dispatch + short-TTL token, keys never leave the server + opened = (await http.post("/api/voice/session")).json() + assert livekit.dispatches[-1]["agent_name"] == "cybergril-voice-agent" + claims, _ = livekit_wire.decode_token(opened["token"]) + assert livekit_wire.verify_signature(opened["token"], "verify-secret-not-real") + assert claims["video"]["roomJoin"] and claims["video"]["room"] == opened["room"] + assert claims["exp"] - claims["nbf"] <= 300 + ok.append("session open: dedicated worker dispatch + genuine short-TTL room token") + + # 3. two sessions never share a room + second = (await http.post("/api/voice/session")).json() + assert opened["room"] != second["room"] + ok.append("token room isolation: each session is its own room") + + # 4. end: the room is deleted server-side, mic and job release with it + ended = await http.delete(f"/api/voice/session/{opened['session_id']}") + assert ended.json()["ended"] is True and livekit.deleted == [opened["room"]] + assert (await http.delete(f"/api/voice/session/{opened['session_id']}")).status_code == 404 + ok.append("session end deletes the LiveKit room (disconnect + job exit path)") + + # 5. the worker's chat adapter over a real socket: multi-turn + receipt + client = ChatApiClient(base) + turn1 = await client.send_turn("第一句你好") + turn2 = await client.send_turn("第二句还在") + assert turn1.completed and turn2.completed and turn1.receipt and turn2.receipt + calls = len(provider.calls) + + # 6. replay: no new model call + replay = await client.send_turn("第一句你好", turn_id=turn1.turn_id) + assert replay.completed and len(provider.calls) == calls + ok.append("voice turns reuse the chat chain; replay adds zero model calls") + + # 7. epoch resync after a server restart (409 + fresh epoch) + server.should_exit = True + await asyncio.sleep(0.5) + threading.Thread(target=lambda: uvicorn.Server( + uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")).run(), + daemon=True).start() + for _ in range(100): + try: + await httpx.AsyncClient().get(f"{base}/api/health", timeout=1) + break + except Exception: + await asyncio.sleep(0.1) + turn3 = await client.send_turn("重启之后", epoch=1) # stale epoch -> resync + assert turn3.completed and turn3.epoch > 1 + ok.append("epoch 409 -> one bounded resync -> turn succeeds on the fresh epoch") + + # 8. barge-in cancels the in-flight stream immediately; row = cancelled + provider.mode = "slow" + got = asyncio.Event() + task = asyncio.create_task( + client.send_turn("说一半我要打断你", on_delta=lambda _: got.set()) + ) + await asyncio.wait_for(got.wait(), timeout=10) + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + provider.mode = "" + for _ in range(50): + rows = (await http.get("/api/messages")).json()["messages"] + row = next((m for m in rows if m["text"] == "说一半我要打断你"), None) + if row and row["status"] == "cancelled": + break + await asyncio.sleep(0.2) + assert row["status"] == "cancelled" + assert not any(m["role"] == "companion" and m["turn_id"] == row["turn_id"] for m in rows) + ok.append("barge-in closes the HTTP stream at once; aborted turn fails closed") + + # 9. receipt gating: only a fully played-out turn acks, exactly once + assert should_ack(played_to_end=True, interrupted=False, failed=False, + receipt=turn1.receipt, current_epoch=turn1.epoch) + assert not should_ack(played_to_end=True, interrupted=True, failed=False, + receipt=turn1.receipt, current_epoch=turn1.epoch) + accepted = await client.acknowledge(turn1.receipt) + assert accepted, "server accepts the receipt of the delivered turn" + ok.append("receipt confirmed only after SDK playback completion, once per turn") + + await client.aclose() + await http.aclose() + + asyncio.run(scenario()) + + # 10. capabilities honesty: voice stays false in health (no fake readiness) + health = httpx.get(f"{base}/api/health").json() + assert health["capabilities"]["voice"] is False + ok.append("/api/health capabilities.voice stays false (R5 not real-accepted)") + + print("HEL-877 offline voice-chain check: ALL PASS") + for line in ok: + print(" ✓", line) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/cybergril/api/app.py b/src/cybergril/api/app.py index d6b084a..e2712f7 100644 --- a/src/cybergril/api/app.py +++ b/src/cybergril/api/app.py @@ -48,6 +48,7 @@ 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 HOURS = [ "子时", @@ -180,6 +181,7 @@ def create_app( character: CharacterConfig | None = None, semantic_mode: str | None = None, semantic_config: dict | None = None, + 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( @@ -264,10 +266,26 @@ def create_app( await app.state.semantic.backend.aclose() for provider in set(router.provider_for(t) for t in router.available_tiers()): await provider.aclose() + if getattr(app.state, "voice_service", None): + await app.state.voice_service.aclose() store.close() app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None) + # R5 voice session endpoint (HEL-877): independent module sharing this + # process's state/locks; unconfigured instances refuse honestly. + # voice_service injection is for isolated tests; production reads config. + if voice_service is None: + try: + voice_config = load_voice_config() + except Exception: + voice_config = None + voice_service = ( + VoiceSessionService(voice_config) if voice_config is not None else None + ) + app.state.voice_service = voice_service + app.include_router(build_voice_router(voice_service)) + @app.exception_handler(RequestValidationError) async def invalid(_request, exc): fields = {} diff --git a/src/cybergril/voice/__init__.py b/src/cybergril/voice/__init__.py new file mode 100644 index 0000000..ed31874 --- /dev/null +++ b/src/cybergril/voice/__init__.py @@ -0,0 +1,26 @@ +"""CyberGril voice module (HEL-877, approved R5 backend wiring). + +Two worlds live here on purpose: + +* API-process side (no livekit imports, runs inside the unified instance): + ``session``/``router`` (POST /api/voice/session + end/status) and + ``chat_api_client``/``receipt`` (protocol helpers the worker also uses). +* Worker side (agent venv only, livekit-agents 1.8.8): ``llm_chat_api``, + ``stt_funasr``, ``tts_cosyvoice``, ``worker`` — the dedicated + ``cybergril-voice-agent`` that reuses the .12 FunASR/CosyVoice/VAD base and + gets every reply from the same POST /api/chat chain the text UI uses. +""" + +from __future__ import annotations + + +def __getattr__(name: str): # lazy: API side must import without livekit installed + if name in {"VoiceConfig", "VoiceSessionService", "load_voice_config", "VoiceNotConfigured"}: + from . import session + + return getattr(session, name) + if name == "build_voice_router": + from . import router + + return router.build_voice_router + raise AttributeError(name) diff --git a/src/cybergril/voice/chat_api_client.py b/src/cybergril/voice/chat_api_client.py new file mode 100644 index 0000000..7696b9d --- /dev/null +++ b/src/cybergril/voice/chat_api_client.py @@ -0,0 +1,186 @@ +"""Unified-API chat client for the voice worker (HEL-877). + +Speaks the exact RealChatAdapter protocol the text frontend already uses +(web/src/adapters/real.ts): GET /api/health for the epoch, POST /api/chat with +{text, turn_id, epoch}, SSE chat.delta / chat.final / chat.error frames, one +bounded 409 resync, receive-confirmation receipts, and immediate cancellation +of the HTTP stream when the caller is interrupted. + +This module is deliberately free of livekit imports so the whole protocol can +be exercised offline against the real API with fake transports. +""" + +from __future__ import annotations + +import codecs +import json +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any +from uuid import uuid4 + +import httpx + + +class ChatApiConnectionError(RuntimeError): + """The SSE stream broke without a final or error frame.""" + + +@dataclass +class ChatApiTurn: + """Result of one completed voice turn (or its honest failure).""" + + turn_id: str + epoch: int + text: str = "" + receipt: dict[str, Any] | None = None + error_phi: str | None = None + completed: bool = False + + +@dataclass +class ChatApiClient: + """One instance per worker; thread-less, single event loop.""" + + base_url: str + transport: httpx.AsyncBaseTransport | None = None + epoch_resyncs: int = 1 + request_timeout: httpx.Timeout = field( + default_factory=lambda: httpx.Timeout(connect=5.0, read=120.0, write=10.0, pool=5.0) + ) + + def __post_init__(self) -> None: + self._client = httpx.AsyncClient( + base_url=self.base_url.rstrip("/"), transport=self.transport, + timeout=self.request_timeout, + ) + + async def aclose(self) -> None: + await self._client.aclose() + + async def current_epoch(self) -> int: + response = await self._client.get("/api/health") + response.raise_for_status() + epoch = response.json().get("epoch") + if not isinstance(epoch, int) or epoch < 1: + raise ChatApiConnectionError("health response missing epoch") + return epoch + + async def send_turn( + self, + text: str, + *, + turn_id: str | None = None, + epoch: int | None = None, + on_delta: Callable[[str], None] | None = None, + ) -> ChatApiTurn: + """Run one turn to completion. Cancellation (barge-in) propagates from + the caller's task; the in-flight HTTP response is always closed in the + finally block so the server sees the disconnect immediately.""" + turn_id = turn_id or str(uuid4()) + epoch = await self.current_epoch() if epoch is None else epoch + attempts = self.epoch_resyncs + 1 + for attempt in range(attempts): + response = None + try: + # stream=True: a plain post() would buffer the whole SSE body. + request = self._client.build_request( + "POST", "/api/chat", json={"text": text, "turn_id": turn_id, "epoch": epoch} + ) + response = await self._client.send(request, stream=True) + if response.status_code == 409 and attempt + 1 < attempts: + await response.aread() + body = response.json() + new_epoch = body.get("epoch") + if isinstance(new_epoch, int) and new_epoch >= 1: + epoch = new_epoch + continue + response.raise_for_status() + return await self._consume( + response, turn_id=turn_id, epoch=epoch, on_delta=on_delta + ) + finally: + if response is not None: + await response.aclose() + raise ChatApiConnectionError("epoch resync exhausted") + + async def _consume( + self, + response: httpx.Response, + *, + turn_id: str, + epoch: int, + on_delta: Callable[[str], None] | None, + ) -> ChatApiTurn: + turn = ChatApiTurn(turn_id=turn_id, epoch=epoch) + got_terminal = False + async for frame in _sse_frames(response): + kind = frame.get("t") + if frame.get("turn_id") not in (None, turn_id): + continue + if kind == "chat.delta": + delta = frame.get("delta") + if delta: + turn.text += delta + if on_delta is not None: + on_delta(delta) + elif kind == "chat.final": + turn.text = frame.get("utterance", {}).get("text", turn.text) + turn.receipt = frame.get("receipt") + turn.completed = True + got_terminal = True + elif kind == "chat.error": + turn.error_phi = frame.get("phi") or "" + got_terminal = True + if not got_terminal: + raise ChatApiConnectionError("chat stream ended without final or error") + return turn + + async def acknowledge(self, receipt: dict[str, Any]) -> bool: + """POST /api/chat/received; False when the server no longer accepts it + (expired / already gone / wrong epoch) — never retried by the caller.""" + try: + response = await self._client.post( + "/api/chat/received", + json={k: receipt[k] for k in + ("message_id", "turn_id", "epoch", "token", "text_hash")}, + ) + except Exception: + return False + return response.status_code == 200 + + +async def _sse_frames(response: httpx.Response): + """Incremental SSE parser: CRLF or LF frame separators, multi-line data, + UTF-8 split across transport chunks — same robustness as readEvents().""" + decoder = codecs.getincrementaldecoder("utf-8")() + buffer = "" + async for chunk in response.aiter_bytes(): + buffer += decoder.decode(chunk) + while True: + boundary = _frame_boundary(buffer) + if boundary is None: + break + frame, buffer = buffer[: boundary[0]], buffer[boundary[1]:] + data = "\n".join( + line[5:].lstrip(" ") for line in frame.splitlines() if line.startswith("data:") + ) + if not data: + continue + try: + yield json.loads(data) + except ValueError: + continue + + +def _frame_boundary(buffer: str) -> tuple[int, int] | None: + """(start, end) of the earliest blank-line separator, or None when the + buffer holds no complete frame yet.""" + candidates = [] + lf = buffer.find("\n\n") + if lf >= 0: + candidates.append((lf, lf + 2)) + crlf = buffer.find("\r\n\r\n") + if crlf >= 0: + candidates.append((crlf, crlf + 4)) + return min(candidates, key=lambda pair: pair[0]) if candidates else None diff --git a/src/cybergril/voice/livekit_wire.py b/src/cybergril/voice/livekit_wire.py new file mode 100644 index 0000000..bd65eaf --- /dev/null +++ b/src/cybergril/voice/livekit_wire.py @@ -0,0 +1,195 @@ +"""LiveKit wire helpers for the voice session endpoint (HEL-877). + +Reuses the already-running LiveKit SFU on .12 without adding the ``livekit`` +package to the API process: access tokens are the same HS256 JWTs, and the +dispatch / room-delete calls use the same Twirp endpoints with the same +protobuf bodies that the installed SDK (livekit 1.1.20, read-only verified on +/opt/voice-agent/agent/venv) sends for /opt/voice-agent/web/server.py's +working /api/token flow. No keys ever leave the server: only the minted, +short-lived room token is returned to the browser. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import time +from dataclasses import dataclass +from typing import Any + +import httpx + +TWIRP_PREFIX = "/twirp" +DISPATCH_SERVICE = "livekit.AgentDispatchService" +ROOM_SERVICE = "livekit.RoomService" + + +def _b64url(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def _jwt_hs256(secret: str, claims: dict[str, Any]) -> str: + """HS256 JWT with the exact header the installed SDK's PyJWT produces.""" + header = _b64url(json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode()) + payload = _b64url(json.dumps(claims, separators=(",", ":"), ensure_ascii=False).encode()) + signing_input = f"{header}.{payload}".encode("ascii") + signature = _b64url(hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()) + return f"{header}.{payload}.{signature}" + + +def _camel(name: str) -> str: + head, *rest = name.split("_") + return head + "".join(part.capitalize() for part in rest) + + +def mint_token( + api_key: str, + api_secret: str, + *, + identity: str, + name: str = "", + room: str = "", + room_join: bool = False, + can_publish: bool = True, + can_subscribe: bool = True, + ttl_seconds: float = 300.0, + now: float | None = None, +) -> str: + """Participant access token; claims mirror livekit.api.AccessToken.to_jwt(). + + Video grants are emitted camelCase with empty values dropped, exactly like + Claims.asdict() in the installed SDK. + """ + now = time.time() if now is None else now + video = { + _camel(k): v + for k, v in { + "room_join": room_join, + "room": room, + "can_publish": can_publish, + "can_subscribe": can_subscribe, + "can_publish_data": True, + }.items() + if v is not None and v != "" + } + claims: dict[str, Any] = {"video": video, "sub": identity, "iss": api_key, + "nbf": int(now), "exp": int(now + ttl_seconds)} + if name: + claims["name"] = name + return _jwt_hs256(api_secret, claims) + + +def mint_server_token(api_key: str, api_secret: str, ttl_seconds: float = 60.0, + now: float | None = None) -> str: + """Server-API bearer token; matches Service._auth_header() with no grants.""" + now = time.time() if now is None else now + return _jwt_hs256( + api_secret, + {"iss": api_key, "sub": "", "nbf": int(now), "exp": int(now + ttl_seconds)}, + ) + + +def _proto_string(field_number: int, value: str) -> bytes: + payload = value.encode() + return bytes([(field_number << 3) | 2]) + _varint(len(payload)) + payload + + +def _varint(value: int) -> bytes: + out = bytearray() + while value > 0x7F: + out.append((value & 0x7F) | 0x80) + value >>= 7 + out.append(value) + return bytes(out) + + +@dataclass +class DispatchResult: + room: str + dispatch_id: str + body: dict + + +async def create_dispatch( + http: httpx.AsyncClient, + *, + internal_url: str, + api_key: str, + api_secret: str, + agent_name: str, + room: str, + metadata: str = "", + timeout: float = 5.0, +) -> DispatchResult: + """Twirp AgentDispatchService/CreateDispatch with an explicit agent_name. + + Body is livekit.CreateAgentDispatchRequest: field 1 agent_name, field 2 room, + field 3 metadata (field numbers verified against the installed protocol pkg). + """ + body = _proto_string(1, agent_name) + _proto_string(2, room) + if metadata: + body += _proto_string(3, metadata) + response = await http.post( + f"{internal_url.rstrip('/')}{TWIRP_PREFIX}/{DISPATCH_SERVICE}/CreateDispatch", + content=body, + headers={ + "authorization": f"Bearer {mint_server_token(api_key, api_secret)}", + "content-type": "application/protobuf", + }, + timeout=timeout, + ) + response.raise_for_status() + parsed = json.loads(response.content or b"{}") + return DispatchResult( + room=room, + dispatch_id=str(parsed.get("id") or parsed.get("dispatchId") or ""), + body=parsed, + ) + + +async def delete_room( + http: httpx.AsyncClient, + *, + internal_url: str, + api_key: str, + api_secret: str, + room: str, + timeout: float = 5.0, +) -> bool: + """Twirp RoomService/DeleteRoom (livekit.DeleteRoomRequest field 1: room). + + Deleting the room disconnects every participant, ends the dispatched agent + job and releases the browser microphone server-side. + """ + response = await http.post( + f"{internal_url.rstrip('/')}{TWIRP_PREFIX}/{ROOM_SERVICE}/DeleteRoom", + content=_proto_string(1, room), + headers={ + "authorization": f"Bearer {mint_server_token(api_key, api_secret)}", + "content-type": "application/protobuf", + }, + timeout=timeout, + ) + # LiveKit answers DeleteRoom with an empty protobuf body; treat 2xx as done. + return response.status_code // 100 == 2 + + +def decode_token(token: str) -> tuple[dict[str, Any], bytes]: + """Header/claims decode without verification (test helper, never trusting input).""" + header_b64, payload_b64, _ = token.split(".") + pad = lambda s: s + "=" * (-len(s) % 4) # noqa: E731 + claims = json.loads(base64.urlsafe_b64decode(pad(payload_b64))) + signing_input = f"{header_b64}.{payload_b64}".encode("ascii") + return claims, signing_input + + +def verify_signature(token: str, secret: str) -> bool: + """Constant-time HS256 check for tests proving the minted token is genuine.""" + parts = token.split(".") + if len(parts) != 3: + return False + signing_input = f"{parts[0]}.{parts[1]}".encode("ascii") + expected = _b64url(hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()) + return hmac.compare_digest(expected, parts[2]) diff --git a/src/cybergril/voice/llm_chat_api.py b/src/cybergril/voice/llm_chat_api.py new file mode 100644 index 0000000..6087da0 --- /dev/null +++ b/src/cybergril/voice/llm_chat_api.py @@ -0,0 +1,153 @@ +"""livekit-agents LLM adapter backed by the unified CyberGril chat API (HEL-877). + +The worker holds no persona, no memory and no direct LLM connection: every +reply is POST /api/chat on the unified instance, so persona / mood / dyad / +memory / life-sharing / call limits and the same-turn gate apply to voice +exactly as to text. This is the same protocol web/src/adapters/real.ts speaks +(epoch sync, serialized turns, barge-in cancel, idempotent replay, receipts). + +Only this module (and worker/stt/tts below) import livekit: the API process +never does. Verified against livekit-agents 1.8.8 source on +/opt/voice-agent/agent/venv (LLM.chat/LLMStream._run/ChatChunk/ChoiceDelta). +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from typing import Any +from uuid import uuid4 + +from livekit.agents import APIConnectionError, APIStatusError +from livekit.agents.llm import ( + LLM, + ChatChunk, + ChatContext, + ChoiceDelta, + LLMStream, + Tool, +) +from livekit.agents.types import ( + DEFAULT_API_CONNECT_OPTIONS, + NOT_GIVEN, + APIConnectOptions, + NotGivenOr, +) + +from .chat_api_client import ChatApiClient, ChatApiConnectionError + + +def latest_user_text(chat_ctx: ChatContext) -> str: + """The one message the server needs: the user's current utterance. + + The server composes persona/memory/history itself from its own archive; + the worker must never ship livekit's local chat_ctx back to it. + """ + for item in reversed(chat_ctx.items): + role = getattr(item, "role", None) + text = getattr(item, "text_content", None) or "" + if role == "user" and text.strip(): + return text.strip() + return "" + + +class ChatApiLLMStream(LLMStream): + """Bridges one /api/chat SSE turn into livekit ChatChunks.""" + + def __init__( + self, + *, + llm: CyberGrillLLM, + chat_ctx: ChatContext, + tools: list[Tool], + conn_options: APIConnectOptions, + ) -> None: + super().__init__(llm, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options) + # Never auto-retry after output: a retried turn would re-speak audio. + # The turn_id stays fixed, so a retry before any chunk is idempotent + # server-side (replay path). + self._retry_on_chunk_sent = False + self.turn_id = str(uuid4()) + self.created_at = time.time() + self.text = "" + self.receipt: dict[str, Any] | None = None + self.error_phi: str | None = None + self.completed = False + + def _emit(self, content: str) -> None: + self._event_ch.send_nowait( + ChatChunk(id=self.turn_id, delta=ChoiceDelta(role="assistant", content=content)) + ) + + async def _run(self) -> None: + text = latest_user_text(self._chat_ctx) + if not text: + raise APIStatusError( + "empty user utterance", status_code=400, body=None, retryable=False + ) + try: + turn = await self._llm.client.send_turn( + text, + turn_id=self.turn_id, + on_delta=self._emit, + ) + except ChatApiConnectionError as exc: + raise APIConnectionError(str(exc)) from exc + except Exception as exc: + raise APIStatusError( + f"chat api failure: {exc}", status_code=502, body=None + ) from exc + self.text = turn.text + self.receipt = turn.receipt + self.error_phi = turn.error_phi + self.completed = turn.completed + if turn.error_phi: + # The φ line is spoken verbatim (HEL-876 §3.5); it is not her + # archived reply and carries no receipt. + self._emit(turn.error_phi) + + +class CyberGrillLLM(LLM): + """llm.LLM whose single backend is the unified instance's /api/chat.""" + + def __init__( + self, + api_base: str, + *, + client: ChatApiClient | None = None, + on_stream_started: Callable[[ChatApiLLMStream], None] | None = None, + ) -> None: + super().__init__() + self.client = client or ChatApiClient(api_base) + self._on_stream_started = on_stream_started + + @property + def provider(self) -> str: + return "cybergril" + + @property + def model(self) -> str: + return "cybergril-chat-api" + + async def aclose(self) -> None: + await self.client.aclose() + + def chat( + self, + *, + chat_ctx: ChatContext, + tools: list[Tool] | None = None, + conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, + parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, + tool_choice: NotGivenOr[Any] = NOT_GIVEN, + extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN, + ) -> ChatApiLLMStream: + stream = ChatApiLLMStream( + llm=self, + chat_ctx=chat_ctx, + tools=tools or [], + conn_options=conn_options, + ) + if self._on_stream_started is not None: + self._on_stream_started(stream) + return stream diff --git a/src/cybergril/voice/receipt.py b/src/cybergril/voice/receipt.py new file mode 100644 index 0000000..c2bb857 --- /dev/null +++ b/src/cybergril/voice/receipt.py @@ -0,0 +1,92 @@ +"""Playback-completion-gated receive confirmation for voice turns (HEL-877). + +A sharing receipt may only be confirmed after the SDK reports the assistant +speech truly finished playing out. None of these count as delivered, so none of +them may ack: interrupted (barge-in), failed playback/generation, turns that +ended in chat.error (the φ line is spoken verbatim but is not her archived +reply), missing receipt, or a receipt from an epoch the server has since left +(the server fail-closes those anyway — this is the client-side half). + +Speech-handle-to-stream binding lives in TurnLedger: AgentSession emits +speech_created before the LLM stream starts and runs one user-turn generation +at a time, so the first stream created at/after a handle belongs to it. +""" + +from __future__ import annotations + +from collections import deque +from typing import Any, Protocol + +from .chat_api_client import ChatApiClient + + +def should_ack( + *, + played_to_end: bool, + interrupted: bool, + failed: bool, + receipt: dict[str, Any] | None, + current_epoch: int | None = None, +) -> bool: + """Pure decision: only a fully played-out, unerrored, receipted turn acks.""" + if not played_to_end or interrupted or failed: + return False + if not receipt: + return False + for key in ("message_id", "turn_id", "epoch", "token", "text_hash"): + if not receipt.get(key): + return False + return current_epoch is None or receipt.get("epoch") == current_epoch + + +class _HasHandle(Protocol): + @property + def created_at(self) -> float: # SpeechCreatedEvent.created_at + ... + + def add_done_callback(self, callback) -> None: ... + + +class TurnLedger: + """Binds livekit SpeechHandles to the ChatApiLLMStream of their generation.""" + + def __init__(self) -> None: + self._handles: deque[tuple[float, Any]] = deque() # (created_at, handle) + self._streams: deque[tuple[float, Any]] = deque() # (created_at, stream) + + def note_handle(self, handle: _HasHandle, created_at: float) -> None: + self._handles.append((created_at, handle)) + + def note_stream(self, stream: Any, created_at: float) -> None: + self._streams.append((created_at, stream)) + + def stream_for(self, handle: _HasHandle) -> Any | None: + """First not-yet-consumed stream created at/after the handle.""" + while self._handles and self._handles[0][1] is not handle: + self._handles.popleft() # older handles ended without a generation + if not self._handles: + return None + # (handle present below) + created = self._handles.popleft()[0] + while self._streams: + at, stream = self._streams.popleft() + if at >= created: + return stream + return None + + +class ReceiptSender: + """Posts accepted receipts to /api/chat/received exactly once per turn.""" + + def __init__(self, client: ChatApiClient) -> None: + self._client = client + self._acked: set[str] = set() + self.results: dict[str, bool] = {} + + async def confirm(self, turn_id: str, receipt: dict[str, Any]) -> bool: + if turn_id in self._acked: + return self.results.get(turn_id, False) + accepted = await self._client.acknowledge(receipt) + self._acked.add(turn_id) + self.results[turn_id] = accepted + return accepted diff --git a/src/cybergril/voice/router.py b/src/cybergril/voice/router.py new file mode 100644 index 0000000..98ded3a --- /dev/null +++ b/src/cybergril/voice/router.py @@ -0,0 +1,78 @@ +"""Minimal voice-session protocol mounted on the unified API (HEL-877). + +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. +DELETE /api/voice/session/{id} end: delete the LiveKit room, which + disconnects browser and worker, ends the + agent job and releases the microphone. +GET /api/voice/status honest switch state for the frontend + (configured / tts reachable / agent name). + +capabilities.voice in /api/health stays false: R5 is not real-accepted yet and +this endpoint never claims end-to-end voice availability. +""" + +from __future__ import annotations + +from fastapi import APIRouter, HTTPException, Request + +from .session import VoiceSessionService + +REFUSAL_NOT_CONFIGURED = "语音还没有在这台服务上配置，不能假装能通话。" +REFUSAL_NO_COMPANION = "还没有创建真实伴侣，语音也不会开始。" +REFUSAL_TTS_DOWN = "她的声音服务这会儿不在，稍后再试试。" +REFUSAL_LIVEKIT_DOWN = "通话服务暂时连不上，稍后再试试。" + + +def build_voice_router(service: VoiceSessionService | None) -> APIRouter: + router = APIRouter(prefix="/api/voice", tags=["voice"]) + + @router.get("/status") + async def status() -> dict: + if service is None: + return {"configured": False, "tts_ok": False, "agent_name": None, "voice": False} + tts_ok = await service.tts_available() + return { + "configured": True, + "tts_ok": tts_ok, + "agent_name": service.config.agent_name, + "voice": False, # honest until real-device R5 acceptance + **service.config.sanitized(), + } + + @router.post("/session", status_code=201) + async def open_session(request: Request) -> dict: + if service is None: + raise HTTPException(503, REFUSAL_NOT_CONFIGURED) + state = request.app.state + if not state.store.query_one("SELECT id FROM companion"): + raise HTTPException(409, REFUSAL_NO_COMPANION) + if not await service.tts_available(): + raise HTTPException(503, REFUSAL_TTS_DOWN) + try: + session = await service.open_session(epoch=state.epoch) + except Exception: + raise HTTPException(503, REFUSAL_LIVEKIT_DOWN) from None + return { + "url": service.config.livekit_url, + "token": service.token_for(session), + "room": session.room, + "session_id": session.session_id, + "agent_name": service.config.agent_name, + "epoch": session.epoch, + "expires_at": int(session.expires_at * 1000), + } + + @router.delete("/session/{session_id}") + async def end_session(session_id: str) -> dict: + if service is None: + raise HTTPException(503, REFUSAL_NOT_CONFIGURED) + session = await service.end_session(session_id) + if session is None: + raise HTTPException(404, "这次通话已经不在了。") + return {"ended": True, "room": session.room} + + return router diff --git a/src/cybergril/voice/session.py b/src/cybergril/voice/session.py new file mode 100644 index 0000000..00a346f --- /dev/null +++ b/src/cybergril/voice/session.py @@ -0,0 +1,209 @@ +"""Voice session lifecycle for the unified API (HEL-877, approved R5 backend). + +The endpoint reuses the .12 voice base (LiveKit SFU + FunASR + CosyVoice TTS + +silero VAD) exactly as HEL-876 audited it: this module only creates a room +dispatch for the dedicated ``cybergril-voice-agent`` worker and mints a +short-lived join token server-side, so no LiveKit key ever reaches the browser. + +Everything here is process-local and stateless apart from the small session +registry: chat state, persona, memory and locks stay in the single API process +that mounts this router (HTTP and the HTTPS voice entry share it). +""" + +from __future__ import annotations + +import asyncio +import os +import secrets +import string +import time +from dataclasses import dataclass, field +from pathlib import Path + +import httpx +import yaml + +from . import livekit_wire + +DEFAULT_AGENT_NAME = "cybergril-voice-agent" +DEFAULT_TOKEN_TTL = 300.0 +DEFAULT_SESSION_TTL = 2 * 3600.0 +TTS_HEALTH_CACHE_SECONDS = 30.0 + + +@dataclass +class VoiceConfig: + livekit_url: str + livekit_internal: str + api_key: str + api_secret: str + agent_name: str = DEFAULT_AGENT_NAME + tts_health_url: str = "http://127.0.0.1:9880/health" + token_ttl_seconds: float = DEFAULT_TOKEN_TTL + session_ttl_seconds: float = DEFAULT_SESSION_TTL + dispatch_timeout_seconds: float = 5.0 + + def sanitized(self) -> dict: + return { + "livekit_url": self.livekit_url, + "agent_name": self.agent_name, + "tts_health_url": self.tts_health_url, + "token_ttl_seconds": self.token_ttl_seconds, + "session_ttl_seconds": self.session_ttl_seconds, + } + + +class VoiceNotConfigured(Exception): + """Raised when no config file exists; the endpoint refuses honestly.""" + + +def load_voice_config( + path: Path | str | None = None, +) -> VoiceConfig | None: + """Load config/voice.yaml; env overrides keep deployment keys out of the repo. + + Returns None (not an exception) when nothing is configured: "voice not set + up on this instance" is an honest state, not an error. + """ + path = Path(path or os.environ.get("CYBERGRIL_VOICE_CONFIG", "config/voice.yaml")) + if not path.is_file(): + return None + data = yaml.safe_load(path.read_text()) or {} + overrides = { + "livekit_url": os.environ.get("CYBERGRIL_VOICE_LIVEKIT_URL"), + "livekit_internal": os.environ.get("CYBERGRIL_VOICE_LIVEKIT_INTERNAL"), + "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"), + } + values = {**data, **{k: v for k, v in overrides.items() if v}} + required = ("livekit_url", "livekit_internal", "api_key", "api_secret") + missing = [k for k in required if not values.get(k)] + if missing: + raise VoiceNotConfigured(f"voice config missing fields: {', '.join(missing)}") + return VoiceConfig( + livekit_url=str(values["livekit_url"]), + livekit_internal=str(values["livekit_internal"]), + api_key=str(values["api_key"]), + api_secret=str(values["api_secret"]), + agent_name=str(values.get("agent_name") or DEFAULT_AGENT_NAME), + 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), + ) + + +@dataclass +class VoiceSession: + session_id: str + room: str + identity: str + created_at: float + expires_at: float + epoch: int + dispatch_id: str = "" + + +@dataclass +class _Health: + checked: float = 0.0 + ok: bool = False + + +@dataclass +class VoiceSessionService: + config: VoiceConfig + http: httpx.AsyncClient = field(default_factory=httpx.AsyncClient) + clock: object = time.time + tts_health_cache_seconds: float = TTS_HEALTH_CACHE_SECONDS + + def __post_init__(self) -> None: + self._sessions: dict[str, VoiceSession] = {} + self._tts_health = _Health(checked=-1e9, ok=False) + self._tts_lock = asyncio.Lock() + + # -- honest readiness --------------------------------------------------- + async def tts_available(self) -> bool: + """Probe the CosyVoice TTS health endpoint, cached like /api/health's LLM probe.""" + now = self.clock() + if now - self._tts_health.checked < self.tts_health_cache_seconds: + return self._tts_health.ok + async with self._tts_lock: + if self.clock() - self._tts_health.checked < self.tts_health_cache_seconds: + return self._tts_health.ok + try: + response = await self.http.get(self.config.tts_health_url, timeout=3.0) + ok = response.status_code // 100 == 2 + except Exception: + ok = False + self._tts_health = _Health(checked=self.clock(), ok=ok) + return ok + + # -- lifecycle ---------------------------------------------------------- + def _prune(self) -> None: + now = self.clock() + for session_id in [k for k, v in self._sessions.items() if v.expires_at < now]: + del self._sessions[session_id] + + async def open_session(self, *, epoch: int) -> VoiceSession: + self._prune() + alphabet = string.ascii_lowercase + string.digits + room = "cg-voice-" + "".join(secrets.choice(alphabet) for _ in range(10)) + identity = "user-" + "".join(secrets.choice(string.ascii_lowercase) for _ in range(8)) + dispatch = await livekit_wire.create_dispatch( + self.http, + internal_url=self.config.livekit_internal, + api_key=self.config.api_key, + api_secret=self.config.api_secret, + agent_name=self.config.agent_name, + room=room, + timeout=self.config.dispatch_timeout_seconds, + ) + now = self.clock() + session = VoiceSession( + session_id=secrets.token_urlsafe(18), + room=room, + identity=identity, + created_at=now, + expires_at=now + self.config.session_ttl_seconds, + epoch=epoch, + dispatch_id=dispatch.dispatch_id, + ) + self._sessions[session.session_id] = session + return session + + def token_for(self, session: VoiceSession) -> str: + return livekit_wire.mint_token( + self.config.api_key, + self.config.api_secret, + identity=session.identity, + name="用户", + room=session.room, + room_join=True, + can_publish=True, + can_subscribe=True, + ttl_seconds=self.config.token_ttl_seconds, + now=self.clock(), + ) + + async def end_session(self, session_id: str) -> VoiceSession | None: + self._prune() + session = self._sessions.pop(session_id, None) + if session is None: + return None + await livekit_wire.delete_room( + self.http, + internal_url=self.config.livekit_internal, + api_key=self.config.api_key, + api_secret=self.config.api_secret, + room=session.room, + timeout=self.config.dispatch_timeout_seconds, + ) + return session + + def get(self, session_id: str) -> VoiceSession | None: + self._prune() + return self._sessions.get(session_id) + + async def aclose(self) -> None: + await self.http.aclose() diff --git a/src/cybergril/voice/stt_funasr.py b/src/cybergril/voice/stt_funasr.py new file mode 100644 index 0000000..f6556f4 --- /dev/null +++ b/src/cybergril/voice/stt_funasr.py @@ -0,0 +1,231 @@ +"""FunASR 流式 STT 的 livekit-agents 适配器（CyberGril 语音 worker 用）. + +复用自 .12 /opt/voice-agent/agent/stt_funasr.py（10-16/10-24 两次真实通话验证 +过的同一实现，未 fork、未改协议），仅加包路径与出处标注：worker 独立目录 +/opt/voice-agent-cybergril 自带一份，通用 worker 的原文件不动。 + +对接 FunASR 官方 2pass websocket 服务(runtime SDK, 端口 10095): +- livekit 音频帧重采样到 16k PCM 后推送 +- 2pass-online 响应 -> INTERIM_TRANSCRIPT(实时草稿) +- 收到 flush(VAD 判定用户说完) -> 发 {"is_speaking": false}, 等 2pass-offline 纠错终稿 + -> FINAL_TRANSCRIPT, 之后为本轮重建 websocket(FunASR 一次会话一句话) + +一个 RecognizeStream 对应一个房间会话, websocket 按轮次惰性重建. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +from typing import Any + +import websockets +from livekit import rtc +from livekit.agents import stt +from livekit.agents.stt import ( + RecognizeStream, + SpeechData, + SpeechEvent, + SpeechEventType, +) +from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS, APIConnectOptions + +FUNASR_WS = "ws://127.0.0.1:10095" + + +class FunASRSTT(stt.STT): + def __init__(self, endpoint: str = FUNASR_WS) -> None: + super().__init__( + capabilities=stt.STTCapabilities(streaming=True, interim_results=True) + ) + self._endpoint = endpoint + # 回声抑制(worker 运行时注入): agent 说话期间, 短文本或与刚说内容 + # 高度相似的识别结果视为音箱回声, 不作为新轮次提交 + self.agent_speaking = False + self.agent_recent_text = "" + + def _is_echo(self, text: str) -> bool: + if not self.agent_speaking or not text: + return False + if len(text) <= 8: + return True # 说话期间识别出 <=8 字 → 大概率是回声碎片 + from difflib import SequenceMatcher + + recent = self.agent_recent_text + if recent: + ratio = SequenceMatcher(None, text, recent).ratio() + if ratio >= 0.55: + return True + return False + + @property + def provider(self) -> str: + return "funasr" + + @property + def model(self) -> str: + return "paraformer-zh-streaming-2pass" + + async def _recognize_impl( + self, buffer, *, language=None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS + ) -> SpeechEvent: + raise NotImplementedError("FunASR adapter only supports streaming") + + def stream( + self, *, language=None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS + ) -> RecognizeStream: + return _LoggedRecognizeStream(stt=self, conn_options=conn_options, endpoint=self._endpoint) + + +class _Utterance: + """一轮用户发言: 一个独立 websocket, flush 后结束""" + + def __init__(self, endpoint: str): + self.endpoint = endpoint + self.ws: websockets.WebSocketClientProtocol | None = None + self.closed = False + self.last_text = "" + + +class FunASRRecognizeStream(RecognizeStream): + def __init__(self, *, stt: FunASRSTT, conn_options: APIConnectOptions, endpoint: str) -> None: + super().__init__(stt=stt, conn_options=conn_options) + self._endpoint = endpoint + + def _ev(self, ev_type: SpeechEventType, text: str) -> SpeechEvent: + return SpeechEvent( + type=ev_type, + alternatives=[SpeechData(language="zh", text=text)], + ) + + async def _run(self) -> None: + utt: _Utterance | None = None + recv_task: asyncio.Task | None = None + finals: asyncio.Queue[str] = asyncio.Queue() + resampler: rtc.AudioResampler | None = None # 有状态, 全程复用 + n_frames = 0 + + async def open_utt() -> _Utterance: + ws = await websockets.connect(self._endpoint, max_size=2**24) + cfg = { + "mode": "2pass", + "wav_name": "livekit", + "is_speaking": True, + "wav_format": "pcm", + "chunk_size": [5, 10, 5], + "audio_fs": 16000, + "itn": True, + } + await ws.send(json.dumps(cfg)) + u = _Utterance(self._endpoint) + u.ws = ws + return u + + async def recv_loop(u: _Utterance) -> None: + assert u.ws is not None + pending: list[str] = [] # 去抖合并中的 final 片段 + emit_task: asyncio.Task | None = None + + def strip_punct(t: str) -> str: + return t.lstrip("，。！？、,.!?；;：: ") + + async def emit_merged_final() -> None: + # 600ms 内无后续片段才视为一句说完, 合并成单个 FINAL + await asyncio.sleep(0.6) + merged = strip_punct("".join(pending)) + pending.clear() + if not merged: + return + stt_ref: FunASRSTT = self._stt # type: ignore[assignment] + if stt_ref._is_echo(merged): + print(f"[stt] echo suppressed: {merged[:40]}", flush=True) + return + print(f"[stt] FINAL(merged): {merged[:60]}", flush=True) + self._event_ch.send_nowait(self._ev(SpeechEventType.FINAL_TRANSCRIPT, merged)) + await finals.put(merged) + + try: + async for msg in u.ws: + data: dict[str, Any] = json.loads(msg) + text = (data.get("text") or "").strip() + if not text: + continue + if str(data.get("mode", "")).endswith("offline"): + # 离线纠错终稿: 去抖合并后作为单个 FINAL 推给框架 + u.last_text = text + pending.append(text) + if emit_task is not None: + emit_task.cancel() + emit_task = asyncio.create_task(emit_merged_final()) + else: + u.last_text = text + print(f"[stt] interim: {text[:60]}", flush=True) + self._event_ch.send_nowait( + self._ev(SpeechEventType.INTERIM_TRANSCRIPT, text) + ) + except Exception: + pass + + try: + async for ev in self._input_ch: + if ev is None: # 流关闭 + break + if isinstance(ev, RecognizeStream._FlushSentinel): + if utt is not None and utt.ws is not None: + print( + f"[stt] flush after {n_frames} frames, " + f"last_text={utt.last_text!r}", + flush=True, + ) + with contextlib.suppress(Exception): + await utt.ws.send(json.dumps({"is_speaking": False})) + # 等离线终稿, 最多 2.5s, 超时用最后的 online 文本兜底 + try: + final = await asyncio.wait_for(finals.get(), timeout=2.5) + except TimeoutError: + final = utt.last_text + if final: + self._event_ch.send_nowait( + self._ev(SpeechEventType.FINAL_TRANSCRIPT, final) + ) + if recv_task is not None: + recv_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await recv_task + with contextlib.suppress(Exception): + await utt.ws.close() + utt = None + continue + if not isinstance(ev, rtc.AudioFrame): + continue + + frame: rtc.AudioFrame = ev + if utt is None: # 新一轮发言, 惰性建连 + utt = await open_utt() + recv_task = asyncio.create_task(recv_loop(utt)) + print(f"[stt] utterance open, frame sr={frame.sample_rate}", flush=True) + n_frames += 1 + if frame.sample_rate != 16000: + if resampler is None: + resampler = rtc.AudioResampler(frame.sample_rate, 16000) + for f in resampler.push(frame): + await utt.ws.send(bytes(f.data)) + else: + await utt.ws.send(bytes(frame.data)) + finally: + if recv_task is not None: + recv_task.cancel() + if utt is not None and utt.ws is not None: + with contextlib.suppress(Exception): + await utt.ws.close() + + +class _LoggedRecognizeStream(FunASRRecognizeStream): + """调试: 打印框架实际读取的每个事件""" + + async def __anext__(self) -> SpeechEvent: + ev = await super().__anext__() + txt = ev.alternatives[0].text[:40] if ev.alternatives else "" + print(f"[stt->session] {ev.type.value}: {txt}", flush=True) + return ev diff --git a/src/cybergril/voice/tts_cosyvoice.py b/src/cybergril/voice/tts_cosyvoice.py new file mode 100644 index 0000000..984feb4 --- /dev/null +++ b/src/cybergril/voice/tts_cosyvoice.py @@ -0,0 +1,117 @@ +"""CosyVoice3 HTTP 流式 TTS 的 livekit-agents 适配器（CyberGril 语音 worker 用）. + +复用自 .12 /opt/voice-agent/agent/tts_cosyvoice.py（同一已验证实现，未 fork、 +未改协议）。CyberGril worker 读取自己的音色状态文件（默认 +/opt/voice-agent-cybergril/voice_state_cybergril.json），不读通用助手的 +voice_state.json；最终音色（D1）仍待白栖知拍板，这里不替用户定。 + +对接 /opt/voice-agent/cosyvoice/server.py: + POST /tts {"text":..., "voice":...} -> audio/L16 (24k mono int16) 字节流 +""" + +from __future__ import annotations + +import time +import uuid + +import httpx +from livekit import rtc +from livekit.agents import tts +from livekit.agents.tts import ( + AudioEmitter, + ChunkedStream, + TTSCapabilities, +) +from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS, APIConnectOptions + +COSYVOICE_URL = "http://127.0.0.1:9880" + + +class CosyVoiceTTS(tts.TTS): + def __init__(self, endpoint: str = COSYVOICE_URL, default_voice: str = "default") -> None: + super().__init__( + capabilities=TTSCapabilities(streaming=False), + sample_rate=24000, + num_channels=1, + ) + self._endpoint = endpoint.rstrip("/") + self._default_voice = default_voice + self._http = httpx.AsyncClient(timeout=httpx.Timeout(30, read=120)) + + @property + def provider(self) -> str: + return "cosyvoice3" + + @property + def model(self) -> str: + return "Fun-CosyVoice3-0.5B-2512" + + async def aclose(self) -> None: + await self._http.aclose() + + def synthesize( + self, + text: str, + *, + conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, + ) -> ChunkedStream: + return CosyVoiceChunkedStream( + tts=self, input_text=text, conn_options=conn_options, voice=self._default_voice + ) + + +class CosyVoiceChunkedStream(ChunkedStream): + def __init__( + self, + *, + tts: CosyVoiceTTS, + input_text: str, + conn_options: APIConnectOptions, + voice: str, + ) -> None: + super().__init__(tts=tts, input_text=input_text, conn_options=conn_options) + self._voice = voice + + async def _run(self, output_emitter: AudioEmitter) -> None: + req_id = uuid.uuid4().hex + t0 = time.perf_counter() + output_emitter.initialize( + request_id=req_id, + sample_rate=24000, + num_channels=1, + mime_type="audio/pcm", + ) + first = True + buf = bytearray() + + def emit_frames() -> None: + # 按 480 样本(20ms)切帧, push_frame 跳过容器解码 + step = 480 * 2 # int16 字节 + while len(buf) >= step: + frame = rtc.AudioFrame(bytes(buf[:step]), 24000, 1, 480) + output_emitter.push_frame(frame) + del buf[:step] + + async with self._tts._http.stream( + "POST", + f"{self._tts._endpoint}/tts", + json={"text": self._input_text, "voice": self._voice}, + ) as resp: + resp.raise_for_status() + async for chunk in resp.aiter_bytes(): + if not chunk: + continue + if first: + print( + f"[tts] first audio after {time.perf_counter() - t0:.3f}s " + f"({self._input_text[:20]}...)", + flush=True, + ) + first = False + buf.extend(chunk) + emit_frames() + if buf: + n = len(buf) // 2 + frame = rtc.AudioFrame(bytes(buf), 24000, 1, n) + output_emitter.push_frame(frame) + output_emitter.flush() diff --git a/src/cybergril/voice/worker.py b/src/cybergril/voice/worker.py new file mode 100644 index 0000000..aaa29fe --- /dev/null +++ b/src/cybergril/voice/worker.py @@ -0,0 +1,189 @@ +"""CyberGril 专属语音 worker（HEL-877，agent_name=cybergril-voice-agent）. + +组装与 /opt/voice-agent/agent/agent.py 同源（10-16/10-24 两次真实通话验证的 +级联链路）: +麦克风 -> silero VAD(端点检测/打断) -> FunASR(流式 STT) + -> 统一 CyberGril 实例 POST /api/chat（她的人格/情绪/关系/记忆/生活分享/ + 调用限制全部在服务端同一业务链里；worker 不持有第二份人格或记忆， + 不直连任何 LLM，不读通用助手 persona_state/voice_state） + -> CosyVoice3(流式 TTS) -> 播放 +打断: VAD 判定用户开口即打断（AgentSession 默认 allow_interruptions）： + 生成任务被取消，ChatApiLLMStream 的 HTTP 流在 finally 立刻关闭， + 服务端随即把旧轮 user 行标 cancelled——不等下一次 ASR final。 + 文字与语音同走 /api/chat，互相争用由服务端同一轮次门处理。 +回执: 只有 SDK 真实播放完成（speech handle done、未被打断、无异常、轮次 + 完整带 receipt）才 POST /api/chat/received；被打断、断线、播放失败、 + chat.error（φ 文案原句念出）一律不确认。 +字幕: cg.subtitle 数据消息发到房间（契约见 docs/voice-contract-hel877.md）。 +音色: 独立 voice_state_cybergril.json；D1 未拍板前回落 default，不替用户定。 + +运行位置: /opt/voice-agent-cybergril/（独立目录，复用 .12 现有 agent venv 与 +LiveKit/FunASR/CosyVoice；专属健康端口 CYBERGRIL_WORKER_HEALTH_PORT—— +livekit-agents worker 内建 HTTP 健康检查，GET / 返回 OK/503）。 +本文件只在 agent venv（livekit-agents 1.8.8）中导入；产品 API 进程不导入它。 +""" + +from __future__ import annotations + +import asyncio +import json +import os +from pathlib import Path + +from livekit.agents import Agent, AgentSession, JobContext, JobProcess, WorkerOptions, cli +from livekit.plugins.silero import VAD + +from .llm_chat_api import CyberGrillLLM +from .receipt import ReceiptSender, TurnLedger, should_ack +from .stt_funasr import FunASRSTT +from .tts_cosyvoice import CosyVoiceTTS + +AGENT_NAME = os.environ.get("CYBERGRIL_AGENT_NAME", "cybergril-voice-agent") +API_BASE = os.environ.get("CYBERGRIL_API_BASE", "http://127.0.0.1:8360") +VOICE_STATE = Path( + os.environ.get( + "CYBERGRIL_VOICE_STATE", "/opt/voice-agent-cybergril/voice_state_cybergril.json" + ) +) +WORKER_HEALTH_PORT = int(os.environ.get("CYBERGRIL_WORKER_HEALTH_PORT", "8082")) +SUBTOPIC = "cg.subtitle" + +# 占位指令：真正的人格由统一服务在 /api/chat 服务端组装。适配器忽略 +# chat_ctx/instructions（latest_user_text 只取当前用户话轮），这里不放任何 +# 人格内容，避免出现第二份她。 +INSTRUCTIONS_PLACEHOLDER = "语音通道适配层：回复内容由统一服务生成。" + + +def read_default_voice() -> str: + """CyberGril 自己的音色状态；缺文件回落 default（D1 未拍板不替用户定）。""" + try: + return json.loads(VOICE_STATE.read_text(encoding="utf-8")).get("voice", "default") + except Exception: + return "default" + + +def prewarm(proc: JobProcess) -> None: + proc.userdata["vad"] = VAD.load() + + +async def _maybe_open_proactive(session: AgentSession, room) -> None: + """CG-011 主动开口接缝（本卡不实现，留明确边界，不丢需求）。 + + 语音里的主动开口必须：仅在已开启的语音会话内、走同一条业务链（只念 + 服务端已落库、标记 proactive=1 的 companion 行，不新起模型调用）、可被 + 用户打断、不新造第二条回执通道。具体呈现与节奏由总管/视觉总监在接线 + 卡定；在此之前保持空操作。 + """ + return None + + +async def entrypoint(ctx: JobContext) -> None: + stt = FunASRSTT() + # agent 说话期间短文本/相似文本视为回声丢弃（同通用 worker 的实测做法） + stt.agent_speaking = False + stt.agent_recent_text = "" + ledger = TurnLedger() + + def note_stream(stream) -> None: + ledger.note_stream(stream, stream.created_at) + + llm = CyberGrillLLM(API_BASE, on_stream_started=note_stream) + receipts = ReceiptSender(llm.client) + + session = AgentSession( + stt=stt, + llm=llm, + tts=CosyVoiceTTS(default_voice=read_default_voice()), + vad=ctx.proc.userdata["vad"], + turn_handling={ + # 停顿稍久才判说完, 合并慢语速碎片（同通用 worker 实测参数） + "endpointing": {"min_delay": 0.8, "max_delay": 6.0}, + }, + ) + + def subtitle(payload: dict) -> None: + data = json.dumps(payload, ensure_ascii=False).encode() + asyncio.get_running_loop().create_task( + ctx.room.local_participant.publish_data(data, topic=SUBTOPIC, reliable=False) + ) + + @session.on("user_input_transcribed") + def _on_transcript(ev) -> None: + if ev.transcript.strip(): + subtitle({ + "kind": "user_final" if ev.is_final else "interim", + "role": "user", + "text": ev.transcript, + }) + + async def _after_playout(handle) -> None: + # SDK 真实播放完成（wait_for_playout 返回）之后才考虑回执。 + await handle.wait_for_playout() + stream = ledger.stream_for(handle) + if stream is None: + return + failed = handle.exception() is not None + if stream.error_phi: + subtitle({"kind": "error", "role": "assistant", "text": stream.error_phi}) + return + if stream.text and not handle.interrupted: + subtitle({"kind": "assistant_final", "role": "assistant", "text": stream.text}) + epoch = None + try: + epoch = await llm.client.current_epoch() + except Exception: + epoch = stream.receipt.get("epoch") if stream.receipt else None + if should_ack( + played_to_end=True, + interrupted=handle.interrupted, + failed=failed, + receipt=stream.receipt, + current_epoch=epoch, + ): + await receipts.confirm(stream.turn_id, stream.receipt) + + @session.on("speech_created") + def _on_speech_created(ev) -> None: + if not ev.user_initiated: + return + ledger.note_handle(ev.speech_handle, ev.created_at) + ev.speech_handle.add_done_callback( + lambda handle=ev.speech_handle: asyncio.get_running_loop().create_task( + _after_playout(handle) + ) + ) + + @session.on("conversation_item_added") + def _on_item(ev) -> None: + if ev.item.role == "assistant": + texts = [ + getattr(p, "text", "") or (p if isinstance(p, str) else "") + for p in (ev.item.content or []) + ] + text = "".join(t for t in texts if t).strip() + if text: + stt.agent_recent_text = text + + @session.on("agent_state_changed") + def _on_state(ev) -> None: + stt.agent_speaking = getattr(ev, "new_state", "") == "speaking" + + await ctx.connect() + await session.start(room=ctx.room, agent=Agent(instructions=INSTRUCTIONS_PLACEHOLDER)) + await _maybe_open_proactive(session, ctx.room) + print( + f"[cybergril-worker] session started (api={API_BASE}, " + f"voice={read_default_voice()}, agent_name={AGENT_NAME})", + flush=True, + ) + + +if __name__ == "__main__": + cli.run_app( + WorkerOptions( + entrypoint_fnc=entrypoint, + prewarm_fnc=prewarm, + agent_name=AGENT_NAME, + port=WORKER_HEALTH_PORT, + ) + ) diff --git a/tests/test_voice_chat_adapter.py b/tests/test_voice_chat_adapter.py new file mode 100644 index 0000000..de05a2e --- /dev/null +++ b/tests/test_voice_chat_adapter.py @@ -0,0 +1,352 @@ +"""ChatApi adapter and receipt-gate tests (HEL-877) — real adapter code against +the real unified API (fake LLM provider, ASGI/local-socket transports) plus +scripted-SSE unit cases. Zero network beyond 127.0.0.1.""" + +from __future__ import annotations + +import asyncio +import json +import socket +import sys +import threading +import time +from pathlib import Path +from uuid import uuid4 + +import httpx +import pytest +import uvicorn + +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.chat_api_client import ( # noqa: E402 + ChatApiClient, + ChatApiConnectionError, + _sse_frames, +) +from cybergril.voice.receipt import ReceiptSender, TurnLedger, should_ack # noqa: E402 + +CREATE = {"name": "语音适配", "birth": "2000-10-18", "hour": "亥时", "relation": "朋友"} + + +@pytest.fixture +def api(tmp_path): + provider = R2LLM() + router = build_router(providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)}) + # file-backed: uvicorn's lifespan in the live-server tests re-opens the same + # database (":memory:" would hand it a second, empty store) + app = create_app(db_path=tmp_path / "adapter.db", router=router, + life_config=LifeConfig(enabled=True, poll_seconds=3600), + semantic_mode="stub") + from fastapi.testclient import TestClient + + with TestClient(app) as client: + client.post("/api/companion", json=CREATE) + epoch = client.get("/api/health").json()["epoch"] + yield app, provider, epoch + provider.mode = "" + + +@pytest.fixture +def voice(api): + app, provider, epoch = api + client = ChatApiClient("http://asgi", transport=httpx.ASGITransport(app=app)) + return client, provider, epoch + + +async def test_consecutive_multi_turn_stream_and_receipt(voice): + client, provider, epoch = voice + deltas_seen: list[str] = [] + turn1 = await client.send_turn("第一句你好", on_delta=deltas_seen.append) + turn2 = await client.send_turn("第二句还在", on_delta=deltas_seen.append) + assert turn1.completed and turn2.completed + assert turn1.text == provider.reply and turn2.text == provider.reply + assert turn1.turn_id != turn2.turn_id + assert "".join(deltas_seen) == turn1.text + turn2.text + assert turn1.receipt and turn2.receipt + assert turn1.receipt["turn_id"] == turn1.turn_id + assert turn2.receipt["epoch"] == epoch + + +async def test_completed_turn_replay_adds_no_model_calls(voice): + client, provider, _ = voice + turn_id = str(uuid4()) + first = await client.send_turn("重放测试", turn_id=turn_id) + calls = len(provider.calls) + second = await client.send_turn("重放测试", turn_id=turn_id) + assert len(provider.calls) == calls, "replay must not call the model again" + assert second.completed and second.text == first.text + assert second.receipt, "replay still carries the receipt" + + +async def test_chat_error_yields_phi_without_receipt(voice): + client, provider, _ = voice + provider.mode = "error" + turn = await client.send_turn("这次会失败") + assert turn.completed is False + assert turn.error_phi and "没接上话" in turn.error_phi + assert turn.receipt is None + provider.mode = "" + + +async def test_epoch_409_resync_bounded(voice): + client, _, epoch = voice + # server already at `epoch`; claim a stale one -> 409 -> resync once -> done + turn = await client.send_turn("旧纪元重同步", epoch=epoch + 999) + assert turn.completed and turn.epoch == epoch + + +async def test_epoch_409_never_loops_on_garbage(voice): + client, _, _ = voice + # a server that always answers 409 without a usable epoch must not loop + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200, json={"epoch": 1}) + return httpx.Response(409, json={"detail": "no epoch here"}) + + stubborn = ChatApiClient("http://x", transport=httpx.MockTransport(handler)) + with pytest.raises(httpx.HTTPStatusError): + await stubborn.send_turn("永不重试") + + +async def test_broken_stream_raises_connection_error(): + """The SSE connection itself dies before any terminal frame.""" + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200, json={"epoch": 1}) + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, + content=b'data: {"t": "chat.delta", "epoch": 1, "turn_id": "' + b"t" * 36 + + b'", "delta": "\\u4f60"}\n\n', # stream then simply ends + ) + + client = ChatApiClient("http://x", transport=httpx.MockTransport(handler)) + with pytest.raises(ChatApiConnectionError): + await client.send_turn("断流") + + +async def test_voice_turn_cancels_active_text_turn_same_gate(api): + """Voice and text contend through the server's one same-turn gate.""" + app, provider, _ = api + + async def slow_text() -> None: + text_client = ChatApiClient("http://asgi", transport=httpx.ASGITransport(app=app)) + with pytest.raises((ChatApiConnectionError, Exception)): + await text_client.send_turn("文字说一半") + + provider.mode = "slow" + task = asyncio.create_task(slow_text()) + await asyncio.sleep(1.0) # the text turn is active and hanging after its first delta + provider.mode = "" # its generator stays parked; the voice turn can complete fast + voice_client = ChatApiClient("http://asgi", transport=httpx.ASGITransport(app=app)) + turn = await voice_client.send_turn("语音插进来") + assert turn.completed + provider.mode = "" # (already reset above; kept for clarity on early exits) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + rows = app.state.store.query( + "SELECT role, text, status, turn_id FROM chat_message ORDER BY id" + ) + slow = [m for m in rows if m["text"] == "文字说一半"] + assert slow and slow[0]["status"] == "cancelled", "barge-in cancels the text row" + assert not [m for m in rows if m["role"] == "companion" and m["turn_id"] == slow[0]["turn_id"]], \ + "no phantom companion row for the aborted text turn" + assert any(m["role"] == "companion" and m["turn_id"] == turn.turn_id and m["status"] == "sent" + for m in rows), "the voice turn itself completes as her archived reply" + + +# ── SSE parser unit cases: chunk boundaries ────────────────────────────────── + + +class _FakeResponse: + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + async def aiter_bytes(self): + for chunk in self._chunks: + yield chunk + + +async def collect(chunks: list[bytes]) -> list[dict]: + return [frame async for frame in _sse_frames(_FakeResponse(chunks))] + + +def frame(event: dict) -> str: + return "data: " + json.dumps(event, ensure_ascii=False) + "\n\n" + + +async def test_sse_split_inside_multibyte_and_crlf_and_multiline(): + event1 = {"t": "chat.delta", "epoch": 1, "turn_id": "t1", "delta": "你也好呀"} + event2 = {"t": "chat.final", "epoch": 1, "turn_id": "t1", + "utterance": {"text": "你也好呀"}, "receipt": {"token": "x" * 48}} + blob = (frame(event1) + frame(event2)).encode() + # split inside a multibyte char, mid-frame, and across CRLF boundaries + cuts = [1, 7, 8, 9, 13, 21, len(blob) // 2, len(blob) - 3, len(blob) - 1] + chunks, prev = [], 0 + for cut in cuts + [len(blob)]: + chunks.append(blob[prev:cut]) + prev = cut + parsed = await collect(chunks) + assert [p["t"] for p in parsed] == ["chat.delta", "chat.final"] + assert parsed[1]["receipt"]["token"] == "x" * 48 + + +async def test_sse_multiline_data_and_comment_lines_and_crlf_separator(): + payload = (": keep-alive\r\n" + "data: {\"t\": \"chat.delta\",\r\n" + "data: \"delta\": \"两行\"}\r\n\r\n" + "data: {\"t\": \"chat.final\"}\r\n\r\n") + parsed = await collect([payload.encode("utf-8")]) + assert parsed[0]["delta"] == "两行" + assert parsed[1]["t"] == "chat.final" + + +async def test_sse_invalid_json_frame_skipped_not_fatal(): + blob = b"data: not-json\n\ndata: {\"t\": \"chat.final\"}\n\n" + parsed = await collect([blob]) + assert parsed == [{"t": "chat.final"}] + + +async def test_sse_utf8_split_across_chunks(): + text = "晚安" + first, second = text[:1], text[1:] + chunks = [b"data: {\"t\": \"chat.delta\", \"delta\": \"" + first.encode(), + second.encode() + b"\"}\n\n"] + parsed = await collect(chunks) + assert parsed[0]["delta"] == text + + +# ── receipt gate: playback-completion only ────────────────────────────────── + + +def _receipt(epoch=3): + return {"message_id": 11, "turn_id": "t" * 36, "epoch": epoch, + "token": "t" * 48, "text_hash": "h" * 64} + + +def test_ack_only_after_full_playout(): + assert should_ack(played_to_end=True, interrupted=False, failed=False, + receipt=_receipt(), current_epoch=3) + # everything else must fail closed + assert not should_ack(played_to_end=True, interrupted=True, failed=False, + receipt=_receipt(), current_epoch=3), "barge-in never acks" + assert not should_ack(played_to_end=True, interrupted=False, failed=True, + receipt=_receipt(), current_epoch=3), "playback/generation failure never acks" + assert not should_ack(played_to_end=False, interrupted=False, failed=False, + receipt=_receipt()), "not finished playing never acks" + assert not should_ack(played_to_end=True, interrupted=False, failed=False, + receipt=None), "phi error turn has no receipt" + assert not should_ack(played_to_end=True, interrupted=False, failed=False, + receipt={"token": "x"}, current_epoch=3), "incomplete receipt never acks" + assert not should_ack(played_to_end=True, interrupted=False, failed=False, + receipt=_receipt(epoch=2), current_epoch=3), "stale epoch never acks" + + +def test_receipt_before_playout_not_possible_by_construction(): + # should_ack requires played_to_end; a receipt arriving mid-playout has no + # path to True even if every other field is perfect. + for interrupted in (True, False): + assert not should_ack(played_to_end=False, interrupted=interrupted, failed=False, + receipt=_receipt()) + + +async def test_ledger_binds_handle_to_its_stream(): + ledger = TurnLedger() + handle = type("H", (), {"created_at": 1.0})() + older = type("S", (), {"created_at": 0.5, "text": "old"})() + ledger.note_stream(older, 0.5) + ledger.note_handle(handle, 1.0) + stream = type("S", (), {"created_at": 1.2, "text": "new"})() + ledger.note_stream(stream, 1.2) + assert ledger.stream_for(handle) is stream, "first stream at/after the handle" + assert ledger.stream_for(handle) is None, "binding consumed exactly once" + + +async def test_receipt_sender_confirms_once_and_reports_server_refusal(): + seen: list[tuple[str, dict]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append((request.url.path, json.loads(request.read()))) + return httpx.Response(200, json={"accepted": True}) + + client = ChatApiClient("http://x", transport=httpx.MockTransport(handler)) + sender = ReceiptSender(client) + receipt = _receipt() + assert await sender.confirm(receipt["turn_id"], receipt) is True + assert await sender.confirm(receipt["turn_id"], receipt) is True # no second POST + assert len(seen) == 1 and seen[0][0] == "/api/chat/received" + + def refusing(request: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"detail": "gone"}) + + refuser = ReceiptSender( + ChatApiClient("http://x", transport=httpx.MockTransport(refusing))) + assert await refuser.confirm(receipt["turn_id"], receipt) is False + + +# ── real-socket cancel path: barge-in closes the HTTP stream immediately ───── + + +@pytest.fixture +def live_server(api): + app, provider, _ = api + probe = socket.socket() + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + probe.close() # free the port for uvicorn to bind + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error") + server = uvicorn.Server(config) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + for _ in range(100): + try: + httpx.get(f"http://127.0.0.1:{port}/api/health", timeout=1) + break + except Exception: + time.sleep(0.1) + yield f"http://127.0.0.1:{port}", app, provider + server.should_exit = True + thread.join(timeout=5) + + +async def test_barge_in_cancels_http_stream_and_server_row(live_server): + """Voice-turn cancellation = barge-in: the adapter's in-flight HTTP stream + closes immediately; the server marks the row cancelled without waiting for + the next ASR final (it never comes in this test).""" + url, app, provider = live_server + provider.mode = "slow" # reply hangs after first delta + client = ChatApiClient(url) + got_delta = asyncio.Event() + task = asyncio.create_task( + client.send_turn("说一半我要打断", on_delta=lambda _: got_delta.set()) + ) + await asyncio.wait_for(got_delta.wait(), timeout=10) + task.cancel() # the interruption + with pytest.raises(asyncio.CancelledError): + await task + # server notices the disconnect and fail-closes the row + for _ in range(50): + rows = httpx.get(f"{url}/api/messages").json()["messages"] + row = next((m for m in rows if m["text"] == "说一半我要打断"), None) + if row and row["status"] == "cancelled": + break + await asyncio.sleep(0.2) + else: + row = next((m for m in httpx.get(f"{url}/api/messages").json()["messages"] + if m["text"] == "说一半我要打断"), None) + assert row and row["status"] == "cancelled" + assert not any(m["role"] == "companion" and m["turn_id"] == row["turn_id"] + for m in rows), "no phantom companion row for the aborted turn" + # and the next turn is clean (no ghost gate) + provider.mode = "" + nxt = await client.send_turn("打断之后") + assert nxt.completed + await client.aclose() diff --git a/tests/test_voice_session.py b/tests/test_voice_session.py new file mode 100644 index 0000000..70ddf15 --- /dev/null +++ b/tests/test_voice_session.py @@ -0,0 +1,271 @@ +"""Voice session endpoint tests (HEL-877): honest refusal, server-side dispatch, +token room isolation, end/invalidation cleanup — all against the real router and +real service with a fake LiveKit Twirp transport. Zero network.""" + +from __future__ import annotations + +import json +import sys +import time +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 import livekit_wire # noqa: E402 +from cybergril.voice.session import VoiceConfig, VoiceSessionService # noqa: E402 + +CREATE = {"name": "语音测试", "birth": "2000-10-18", "hour": "亥时", "relation": "朋友"} + + +class FakeLiveKit: + """Fake Twirp LiveKit server: records dispatches, room deletes, auth tokens.""" + + def __init__(self) -> None: + self.dispatches: list[dict] = [] + self.deleted_rooms: list[str] = [] + self.auth_tokens: list[str] = [] + self.fail = False + self.transport = httpx.MockTransport(self._handle) + self._dispatch_seq = 0 + + def _handle(self, request: httpx.Request) -> httpx.Response: + self.auth_tokens.append(request.headers.get("authorization", "")) + if self.fail: + return httpx.Response(500, text="livekit down") + path = request.url.path + if path == "/twirp/livekit.AgentDispatchService/CreateDispatch": + body = request.content + agent_name, room = _parse_dispatch(body) + self._dispatch_seq += 1 + self.dispatches.append({"agent_name": agent_name, "room": room, + "proto": True, "content_type": + request.headers.get("content-type", "")}) + return httpx.Response(200, json={"id": f"AD_{self._dispatch_seq}", + "agentName": agent_name, "room": room}) + if path == "/twirp/livekit.RoomService/DeleteRoom": + self.deleted_rooms.append(_parse_room(request.content)) + return httpx.Response(200, content=b"") + return httpx.Response(404, text="unknown twirp route") + + def service(self) -> 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", + ) + return VoiceSessionService(config=config, http=httpx.AsyncClient(transport=self.transport)) + + +def _read_varint(data: bytes, index: int) -> tuple[int, int]: + shift = 0 + value = 0 + while True: + byte = data[index] + value |= (byte & 0x7F) << shift + index += 1 + if not byte & 0x80: + return value, index + shift += 7 + + +def _parse_dispatch(body: bytes) -> tuple[str, str]: + fields = _parse_proto_strings(body) + return fields.get(1, ""), fields.get(2, "") + + +def _parse_room(body: bytes) -> str: + return _parse_proto_strings(body).get(1, "") + + +def _parse_proto_strings(data: bytes) -> dict[int, str]: + out = {} + index = 0 + while index < len(data): + tag, index = _read_varint(data, index) + field, wire = tag >> 3, tag & 7 + if wire != 2: + break + length, index = _read_varint(data, index) + out[field] = data[index: index + length].decode() + index += length + return out + + +@pytest.fixture +def fake_livekit(): + return FakeLiveKit() + + +class FakeTTS: + def __init__(self) -> None: + self.ok = True + self.calls = 0 + + def handler(self, request: httpx.Request) -> httpx.Response: + self.calls += 1 + return httpx.Response(200) if self.ok else httpx.Response(503) + + +def make_app(fake_livekit: FakeLiveKit, tts: FakeTTS): + provider = R2LLM() + router = build_router(providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)}) + service = fake_livekit.service() + service.http = httpx.AsyncClient(transport=_combined_transport(fake_livekit, tts)) + app = create_app(db_path=":memory:", router=router, + life_config=LifeConfig(enabled=True, poll_seconds=3600), + semantic_mode="stub", voice_service=service) + return app, service, provider + + +def _combined_transport(fake_livekit: FakeLiveKit, tts: FakeTTS) -> httpx.MockTransport: + def handler(request: httpx.Request) -> httpx.Response: + if request.url.host == "tts.test": + return tts.handler(request) + return fake_livekit._handle(request) + + return httpx.MockTransport(handler) + + +@pytest.fixture +def instance(fake_livekit): + tts = FakeTTS() + app, service, provider = make_app(fake_livekit, tts) + from fastapi.testclient import TestClient + + with TestClient(app) as client: + client.post("/api/companion", json=CREATE) + yield client, fake_livekit, service, provider, tts + + +def test_status_unconfigured_honest(): + from fastapi.testclient import TestClient + + app = create_app(db_path=":memory:", router=build_router( + providers={t: R2LLM() for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)}), + life_config=LifeConfig(enabled=False), semantic_mode="stub") + with TestClient(app) as client: + status = client.get("/api/voice/status").json() + assert status == {"configured": False, "tts_ok": False, "agent_name": None, + "voice": False} + # health stays honest: capabilities.voice never faked on + assert client.get("/api/health").json()["capabilities"]["voice"] is False + refusal = client.post("/api/voice/session") + assert refusal.status_code == 503 + assert "语音还没有在这台服务上配置" in refusal.json()["detail"] + + +def test_open_session_dispatches_dedicated_worker(instance): + client, livekit, service, _, _ = instance + response = client.post("/api/voice/session") + assert response.status_code == 201, response.text + body = response.json() + # dedicated dispatch, not the generic cascade-voice-agent + assert livekit.dispatches[-1]["agent_name"] == "cybergril-voice-agent" + assert livekit.dispatches[-1]["room"] == body["room"] + assert livekit.dispatches[-1]["proto"] is True + assert livekit.dispatches[-1]["content_type"] == "application/protobuf" + assert body["url"] == "ws://voice.test:7880" + assert body["agent_name"] == "cybergril-voice-agent" + # no livekit key material leaks into the browser payload + assert "devkey" not in json.dumps(body) and "test-secret-not-real" not in json.dumps(body) + + +def test_token_is_real_short_ttl_and_room_scoped(instance): + client, livekit, _, _, _ = instance + secret = "test-secret-not-real" + first = client.post("/api/voice/session").json() + second = client.post("/api/voice/session").json() + assert first["room"] != second["room"], "each session gets its own room" + for body in (first, second): + claims, signing_input = livekit_wire.decode_token(body["token"]) + assert livekit_wire.verify_signature(body["token"], secret) + assert claims["iss"] == "devkey" + assert claims["sub"].startswith("user-") + video = claims["video"] + assert video["roomJoin"] is True and video["room"] == body["room"] + assert video["canPublish"] is True and video["canSubscribe"] is True + assert claims["exp"] - claims["nbf"] <= 300, "token must be short-lived" + # token for room A grants nothing about room B (room isolation) + claims_a, _ = livekit_wire.decode_token(first["token"]) + claims_b, _ = livekit_wire.decode_token(second["token"]) + assert claims_a["video"]["room"] != claims_b["video"]["room"] + + +def test_open_session_refuses_without_companion(fake_livekit): + tts = FakeTTS() + app, service, _ = make_app(fake_livekit, tts) + from fastapi.testclient import TestClient + + with TestClient(app) as client: + refusal = client.post("/api/voice/session") + assert refusal.status_code == 409 + assert "还没有创建真实伴侣" in refusal.json()["detail"] + assert not fake_livekit.dispatches, "no dispatch may happen without a profile" + + +def test_open_session_refuses_when_tts_down(instance): + client, livekit, service, _, tts = instance + tts.ok = False + service._tts_health.checked = -1e9 # bust cache + refusal = client.post("/api/voice/session") + assert refusal.status_code == 503 + assert "她的声音服务这会儿不在" in refusal.json()["detail"] + assert not livekit.dispatches + tts.ok = True + service._tts_health.checked = -1e9 + assert client.post("/api/voice/session").status_code == 201 + + +def test_open_session_refuses_when_livekit_down(instance): + client, livekit, _, _, _ = instance + livekit.fail = True + refusal = client.post("/api/voice/session") + assert refusal.status_code == 503 + assert "通话服务暂时连不上" in refusal.json()["detail"] + + +def test_end_session_deletes_room_and_cleans_registry(instance): + client, livekit, service, _, _ = instance + body = client.post("/api/voice/session").json() + assert service.get(body["session_id"]) is not None + ended = client.delete(f"/api/voice/session/{body['session_id']}") + assert ended.status_code == 200 and ended.json()["ended"] is True + assert livekit.deleted_rooms == [body["room"]] + assert service.get(body["session_id"]) is None + again = client.delete(f"/api/voice/session/{body['session_id']}") + assert again.status_code == 404 + + +def test_registry_expiry_prunes_old_sessions(fake_livekit): + import asyncio + + service = fake_livekit.service() + now = time.time() + session = asyncio.run(service.open_session(epoch=7)) + assert service.get(session.session_id) is not None + service.clock = lambda: now + service.config.session_ttl_seconds + 1 + assert service.get(session.session_id) is None, "expired registry entry pruned" + asyncio.run(service.aclose()) + + +def test_worker_auth_tokens_are_server_jwt(instance): + client, livekit, _, _, _ = instance + client.post("/api/voice/session") + auth = livekit.auth_tokens[-1] + assert auth.startswith("Bearer ") + token = auth.removeprefix("Bearer ") + claims, _ = livekit_wire.decode_token(token) + assert claims["iss"] == "devkey" and claims["sub"] == "" + assert claims["exp"] - claims["nbf"] <= 60