diff --git a/docs/requirements/handoff-HEL-908.json b/docs/requirements/handoff-HEL-908.json new file mode 100644 index 0000000..e476a66 --- /dev/null +++ b/docs/requirements/handoff-HEL-908.json @@ -0,0 +1,48 @@ +{ + "baseline_version": "1.12", + "read_commit": "5625de6606a607a9d4fd583fe8a52d93d6d2b4ea", + "task": "HEL-908：HEL-797 已授权定点返修，修复 HEL-907 窗口实证的两项缺陷。其一『她在播语音时网页发文字→文字轮被取消、无回复、旧语音继续约32秒』：实证根因是取消发生在首 delta 前（worker.log:108 的 SDK 重试警告在文字 POST 后 13ms），服务端取消收尾无终态帧，worker 适配器把它当连接错误，livekit-agents 1.8.8 LLMStream._main_task 静默重试同一 turn_id，重试重新激活已取消轮并反杀新文字轮、重新生成整段语音播完；原报告『_retry_on_chunk_sent 守卫从未生效』按总管只读核对与实测源码确认不成立（守卫仅在有输出后适用）。其二『Considering the limi…/ 进入字幕与 TTS』：端点把隐式开头（无 、模板预填）的推理原文+stray 混入 content（DB id=12 原文），适配器此前无任何 reasoning 分离。全部修复离线复现→验证，零真实请求、零部署、零依赖变更。", + "requirement_ids": [ + "CG-006", + "CG-010", + "CG-013", + "CG-014", + "CG-015", + "CG-016", + "CG-018", + "CG-019", + "CG-020", + "CG-021", + "AM-001", + "GOV-001", + "GOV-002", + "GOV-003", + "GOV-004", + "GOV-005", + "GOV-006", + "DEL-001" + ], + "deliverable": "双侧最小修复+离线证据。服务端 src/cybergril/api/app.py：取消路径显式下发 chat.cancelled 终态帧（取消≠断网）；chat_lock 拒绝 status=cancelled 的旧轮重新取得活动轮（迟到重试不再反杀新轮，幂等回 chat.cancelled；failed 轮仍可重试）；落库/chat.final/receipt 前轮级 split_reasoning 兜底。worker：chat_api_client.py 识别 chat.cancelled 终态；llm_chat_api.py 以 SDK 自身 499 语义正常结束（不重试不抛出）并经 SpeechHandle.interrupt()（LiveKit 既有打断路径）停掉已缓存音频，server_cancelled 重入闸零网络短路；worker.py 零改动。思考分流：新增 providers/llm/reasoning.py（流式状态机+轮级分流，复用 reasoning_content 字段与 约定，不造推理引擎、不改提示词、不删英文），openai_compat.py 接入并把 reasoning_content 增量导入 LLMDelta.reasoning（contracts 新增诊断字段，绝不持久化/下发/播报）。测试：tests/test_rework_hel908.py（app 级取消语义与泄漏回放）、tests/test_reasoning_separation_hel908.py（37 例含记录流逐字回放）、tests/test_sdk_livekit_hel908.py（真实 SDK harness，pytest 可收集、agent venv 可独立运行）；既有 test_voice_chat_adapter.py 一例更新到新终态语义、test_api.py 断连用例内容补句号（意图不变）。证据 docs/rework/hel908/（summary.md+tests-remote.log）。", + "acceptance": [ + "问题1离线复现→修复：ScriptedLLM 首delta前挂起模拟记录时序——文字 POST 后被取消的语音流收到 chat.cancelled 终态干净结束（不再异常）；迟到同 turn_id 重试被拒绝且零额外模型调用；文字轮保留并获回复（user 行 sent+companion 行落库）；语音轮无 phantom companion 行", + "首delta后取消（真实 socket uvicorn）：已发 delta 保留、随后终态取消；真实 SDK harness 5/5——首delta前取消不重试（1次POST）、SpeechHandle.interrupt() 恰一次、取消后重入立即499零网络、真实连接错误仍按 SDK 有限重试（max_retry+1后诚实失败）、完成轮流式+receipt不变", + "问题2离线复现→修复：DB id=12 原文（隐式开头推理+stray ）经真实 OpenAICompatLLM+真实 app 全链回放——chat.delta 拼接、落库行、chat.final、receipt text_hash 全部只绑干净可见回复；未闭合思考→空正文→诚实 chat.error（user 行 failed，无 phantom 回复）；中英文正常正文逐字节保留（不以删英文掩盖）", + "既有韧性不破坏：已完成轮幂等重放仍零额外模型调用；failed 轮仍可重试；receipt 校验语义不变（专项用例）", + "规范环境全量：.venv311（requirements.lock 一致，只读复用）445 passed/5 skipped；2 个 test_r3 失败在施工基准 5625de6 上同样失败（预先存在，证据留存）；ruff src/ tests/ 全绿；零真实请求、零部署、未触碰任何现网服务/端口/依赖/锁文件" + ], + "deferred": [ + "修复后真实端到端复测（真实房间、真人打断时延、真机 Safari/Chrome）须按授权窗口安排，本卡零真实请求边界内未做；停播证明是真实 SDK SpeechHandle.interrupt() 断言，不冒充端到端毫秒指标", + "思考分流流式残留：隐式开头且 落在128字符hold外的推理前缀仍会部分流出到字幕/TTS（流式下无法提前判定），落库与 receipt 始终干净；彻底消除需上游自行分流（reasoning_content 字段或服务端 reasoning parser），零真实请求下无法探测该端点能力——最小缺口如实声明", + "被新轮打断的可见界面状态（文字轮显示『被新对话取代』等）属界面核心，交视觉总监；本卡前端零改动（新帧 kind 被现有 ChatPanel 静默忽略，行为不差于现状）", + "scripts/gen_hel877_sdk_check.py 在施工基准上即因 HEL-906 的 stt_funasr→asr_usage 导入而失效（harness 模块清单未随更新，base_exit=2 证据留存），交总管决定是否另行安排维护；本卡 SDK 验证由新 harness 覆盖", + "R1–R7 完整功能（真实用户麦克风打断、视频/口型、记忆长跑等）继续按获批依赖承接，本卡不构成整产品完成" + ], + "requirement_changes": [], + "verification": [ + "docs/rework/hel908/summary.md：根因时间轴（worker.log:108 13ms 证据、DB id=5/6/8、_retry_on_chunk_sent 纠偏对照源码）、最小变更表、残留声明", + "docs/rework/hel908/tests-remote.log：.venv311 全量 445 passed/5 skipped/2 pre-existing failed + agent venv SDK harness 5/5 PASS 实况", + "tests/test_sdk_livekit_hel908.py 可在任何装有 livekit-agents 1.8.8 的环境复跑（pytest 或 --standalone），命令写在文件头", + "预先存在失败的基准证据：test_r3 两例与 gen_hel877 harness 在 5625de6 上同样失败（本卡留存对照运行记录）", + "需求检查：check_requirements.py 对 origin/main 与 5625de6 各带 --handoff 本文件通过（日志随交付）" + ] +} diff --git a/docs/rework/hel908/check-requirements.log b/docs/rework/hel908/check-requirements.log new file mode 100644 index 0000000..5687cae --- /dev/null +++ b/docs/rework/hel908/check-requirements.log @@ -0,0 +1,8 @@ +== --base-ref origin/main == +需求结构检查通过：60项；来源和覆盖完整。仍须人工核对授权语义与实际体验。 +== --base-ref origin/main --handoff == +需求结构检查通过：60项；来源和覆盖完整。仍须人工核对授权语义与实际体验。 +== --base-ref 5625de6 == +需求结构检查通过：60项；来源和覆盖完整。仍须人工核对授权语义与实际体验。 +== --base-ref 5625de6 --handoff == +需求结构检查通过：60项；来源和覆盖完整。仍须人工核对授权语义与实际体验。 diff --git a/docs/rework/hel908/summary.md b/docs/rework/hel908/summary.md new file mode 100644 index 0000000..95b9a8f --- /dev/null +++ b/docs/rework/hel908/summary.md @@ -0,0 +1,87 @@ +# HEL-908 · 定点返修记录：交谈打断丢字 + 内部思考外泄（2026-10-28） + +> 授权：本卡 HEL-908（主卡 HEL-797 定点返修）。全部验证离线、零真实请求、零部署。 +> 施工基准：`agent/agent/47adf6961f8a` @ `5625de6606a607a9d4fd583fe8a52d93d6d2b4ea`（完整 SHA 已核）。 +> 需求基准 v1.12/60 项，本卡零需求变更。 + +## 0. 结论速览 + +| 项 | 结果 | +| --- | --- | +| 问题1：她在播语音时网页发文字 → 文字轮被取消、无回复、旧语音继续约32秒 | ✅ 已修复并离线复现→验证（服务端+worker 双侧，穿真实 SDK 路径） | +| 问题2：`Considering the limi…`/`` 进入字幕与 TTS | ✅ 已修复并离线复现→验证（记录流逐字回放）；1 项有界残留已如实声明（§4） | +| HEL-907 原报告“`_retry_on_chunk_sent` 守卫从未生效”的推断 | ❌ 不成立（按总管只读核对纠偏，本卡实证确认，见 §1.2） | +| 既有有限重试/已完成幂等重放/receipt 校验 | ✅ 未被破坏（专项测试证明） | +| 真实端到端复测（真人麦克风/真房间） | ⛔ 本卡零真实请求边界内未做，交回总管按窗口安排 | + +## 1. 问题1根因（源码级 + 实证时间轴） + +### 1.1 事实链（HEL-907 证据 + 本卡新增核对） + +- `worker.log:108`：`failed to generate LLM completion: chat stream ended without final or error …, retrying in 0.1s`，时间 `12:17:58.455`；用户文字 POST t0=`12:17:58.442`（call1.json `t0_text_post_ms=1793189878442`）——**SSE 中止与 SDK 重试发生在文字 POST 后 13ms**。 +- 该警告来自 livekit-agents 1.8.8 `LLMStream._main_task` 的 APIError 重试分支（attempt 1）→ 证明重试走的是 SDK 自己的重试环，重试同一 `turn_id` 重新执行 `_run` → `send_turn`（api.log：health+POST 同端口 38996 两次）。 +- DB：语音轮 user 行 id=5 两度 pending（重试 `ON CONFLICT … DO UPDATE` 重新激活），文字轮 user 行 id=6 `cancelled`，语音重试生成完整回复 id=8（ts=882175，重试 POST 后约 2.6 秒重新生成）并经 TTS 全部播完（t0 后 32.0 秒音频，`old_audio_stop_latency_s=32.015`）。 +- 关键事实：**取消发生在首 delta 之前**——12:17:58.442 时该轮一个 chat.delta 都没到（TTS 首包在其后 3.5 秒；字幕全部增量来自重试后的重新生成），因此 `LLMStream._event_ch.output_sent=False`。 + +### 1.2 纠偏：原报告“`_retry_on_chunk_sent=False` 永远不起作用”不成立 + +对照 .12 实际安装的 livekit-agents 1.8.8 源码（只读核对 `livekit/agents/llm/llm.py`）： + +- `_LLMEventChannel.send_nowait` 在**送入**含 content 的 chunk 时置 `output_sent=True`； +- `_main_task` 对 APIError：`if not self._retry_on_chunk_sent and self._event_ch.output_sent: e.retryable=False`；`APIStatusError` 499 直接正常返回（不重试不抛出）。 + +即守卫**只在已有输出后才适用**；本缺陷路径取消发生在首 chunk 前（`output_sent=False`），守卫根本不在适用范围——不是“守卫失效”，是“取消被当成了临时断网”。本卡 SDK harness 的 `real_connection_errors_still_retry` 用例同时证明：真实连接错误的重试路径未被本次修复全局禁用。 + +### 1.3 修复（双侧最小改动） + +- 服务端 `src/cybergril/api/app.py`： + 1. `generate()` 取消路径在结束 SSE 前**显式下发终态帧 `chat.cancelled`**（取消≠断网：取消是协议事件，断网才走连接错误语义；客户端已断开时保持原样静默结束）。 + 2. `chat_lock` 段新增：`status='cancelled'` 的旧轮**拒绝重新激活**，直接幂等回 `chat.cancelled`，且**不触碰当前活动轮**——同一已取消 `turn_id` 永远无法再取得活动轮所有权（迟到的重试再也不能反杀新文字轮）。`failed` 轮仍可重试（既有韧性不变，专项用例覆盖）。 +- worker `src/cybergril/voice/`： + - `chat_api_client.py`：`_consume` 识别 `chat.cancelled` 终态帧 → `ChatApiTurn.cancelled=True`，不再当 `ChatApiConnectionError`。 + - `llm_chat_api.py`：取消终态 → 复用 SDK 自身的 499（Client Closed Request）语义 `raise APIStatusError(status_code=499)`，`_main_task` 正常返回、**不重试不抛出**；同时经 `SpeechHandle.interrupt()`（SDK 既有打断路径，VAD 抢话同款）**停掉已缓存的旧音频播放**，不止停止生成。另加 `server_cancelled` 重入闸：一旦该流得知取消，`_run` 再被任何路径重入都立即 499、零网络调用。 +- `worker.py`：零改动（interrupt 经适配器归属到确切 SpeechHandle；`_after_playout` 对 interrupted 的 fail-closed 收口、字幕 turn_cancel、回执放弃全部走既有路径）。 + +## 2. 问题2根因与修复 + +- 证据：DB 行 id=12（call3）原文 = `"Considering the limited time by the user, I have to give the solution based on the thinking directly now.\n\n\n大圣，您这定身法…"`——**`content` 里只有 stray ``、没有开头 ``**（服务端 chat template 预填了开头），推理原文因此流进 chat.delta → 字幕与 TTS（tts-window-tail.log 两个 segment 起头即 `Considering the limi…` 与 ``）。同一端点同窗口亦有完全干净回复（id=14），证明泄漏是按次混入而非固定。 +- 定位：`openai_compat.py` 适配器此前只读 `delta.content`，**没有任何 reasoning 分离**；`reasoning_content` 字段（DeepSeek 系约定）也未读。 +- 修复（复用 OpenAI 兼容生态的两个既有分流约定，不造推理引擎、不改提示词、不删英文）： + - 新增 `src/cybergril/providers/llm/reasoning.py`：流式 `ReasoningSeparator` 状态机（hold/think/visible；跨 chunk 标签安全；stray closer=隐式开头思考的收口；中文句末/128 字符解除 hold）+ 轮级 `split_reasoning`。 + - `openai_compat.py`：`reasoning_content`/`reasoning` 字段增量 → `LLMDelta.reasoning`（仅诊断，绝不持久化/下发/播报）；`content` 过分离器后才是 `delta.text`。 + - `contracts/utterance.py`：`LLMDelta.reasoning` 字段（默认空，向后兼容）。 + - `app.py`：落库/chat.final/receipt 前再做轮级 `split_reasoning`（对超出 hold 的流式残留兜底，receipt 绑定干净文本）。 + - 边界：原始模型输出 →（适配器分流）→ 可见正文才可持久化/字幕/TTS；reasoning 仅存在于 `LLMDelta.reasoning` 诊断通道，落地即弃。 + +## 3. 最小变更表 + +| 文件 | 变更 | +| --- | --- | +| `src/cybergril/api/app.py` | 取消路径下发 `chat.cancelled`；拒绝 cancelled 轮重新激活；落库前轮级分流兜底 | +| `src/cybergril/voice/chat_api_client.py` | `chat.cancelled` 终态帧解析（`ChatApiTurn.cancelled`） | +| `src/cybergril/voice/llm_chat_api.py` | 499 取消终态（不重试）+ `SpeechHandle.interrupt()` 停播 + `server_cancelled` 重入闸 | +| `src/cybergril/providers/llm/reasoning.py` | 新增：流式思考分流状态机 + 轮级分流 | +| `src/cybergril/providers/llm/openai_compat.py` | 接入分流器；读 `reasoning_content` 字段 | +| `src/cybergril/contracts/utterance.py` | `LLMDelta.reasoning` 诊断字段 | +| `tests/test_rework_hel908.py` | 新增：app 级取消语义（首 delta 前取消/首 delta 后取消/迟到重试拒绝/文字轮存活/think 泄漏回放/未闭合思考 fail-closed/重放与失败重试不变） | +| `tests/test_reasoning_separation_hel908.py` | 新增：分流单测（记录流逐字回放、跨 chunk 标签、推理字段、未闭合、中英文正文保留、残留声明） | +| `tests/test_sdk_livekit_hel908.py` | 新增：真实 SDK 路径 harness（pytest 可收集；agent venv 无 pytest 可独立运行） | +| `tests/test_voice_chat_adapter.py` | 既有用例更新到新终态语义（被取消方收到 `chat.cancelled` 干净结束，不再异常） | +| `tests/test_api.py` | 断连用例的上游内容补句号（hold 语义下首帧可流出；用例意图——断连关闭上游——不变） | + +## 4. 验证（全部离线、零真实请求） + +- **规范环境**：`.12 /opt/cybergril-hel879/current/.venv311`（requirements.lock 逐字一致，只读复用；代码在 `/tmp` 副本运行，未装任何依赖）。 + - 全量：**445 passed / 5 skipped**（5 skipped=SDK harness 需 livekit，见下）/ 2 failed=`test_r3` 基线/迁移两例——**在施工基准 5625de6 上同样失败（预先存在，与本卡无关，已留存证据）**。 + - `tests-remote.log`：全量 + SDK harness 输出。 +- **真实 SDK 路径**（`.12 /opt/voice-agent/agent/venv`，livekit-agents 1.8.8，只读复用）：`tests/test_sdk_livekit_hel908.py --standalone` **5/5 PASS**——首 delta 前取消不重试（1 次 POST）、首 delta 后取消保留已发 chunk 且 `SpeechHandle.interrupt()` 恰好一次、取消后重入立即 499 零网络、**真实连接错误仍按 SDK 有限重试（max_retry+1 次后诚实失败）**、完成轮流式与 receipt 不变。 +- **记录流回放**：`tests/test_reasoning_separation_hel908.py` 37/37（含 id=12 原文按 1/2/3/5/7/13/1000 字符分块全零泄漏；`` 跨 SSE 帧拆分；中英文正文逐字节保留；未闭合思考→空正文→诚实 `chat.error`）。 +- 本地环境（py3.10，与 lock 版本不一致）：reasoning 37/37、ruff 全绿；app 级用例因 starlette/anyio 版本差异在本地环境基线即不可运行（已在 .12 规范环境执行）。 +- ruff：`src/ tests/` 全绿。 +- `scripts/gen_hel877_sdk_check.py` 在 agent venv 于**施工基准上即失败**（HEL-906 引入 `stt_funasr→asr_usage` 导入后 harness 模块清单未更新，与本卡无关，base_exit=2 证据留存）；本卡 SDK 验证由新 harness 覆盖。 + +## 5. 残留与未测（如实） + +1. **思考分流流式残留**：隐式开头（无 ``）且收尾 `` 落在 128 字符 hold 之外的推理，前缀仍会部分流出到字幕/TTS（无法在流式下提前判定）；落库文本与 receipt 始终干净（轮级兜底，专项用例声明该边界）。彻底消除需上游自行分流（`reasoning_content` 字段或服务端 reasoning parser）——零真实请求边界内无法探测该端点能力，这是**最小缺口**，不虚称已完全修好。 +2. 修复后的真实端到端复测（真实房间/真人打断时延/真机）未做，交回总管按授权窗口安排；停播路径用真实 SDK 的 `SpeechHandle.interrupt()` 断言证明，不冒充端到端毫秒指标。 +3. 前端零改动（新帧 kind 被现有 ChatPanel 静默忽略，行为不差于现状）；把“被新轮打断”做成可见界面状态属界面核心，交视觉总监。 diff --git a/docs/rework/hel908/tests-remote.log b/docs/rework/hel908/tests-remote.log new file mode 100644 index 0000000..e4eb453 --- /dev/null +++ b/docs/rework/hel908/tests-remote.log @@ -0,0 +1,15 @@ + +-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html +=========================== short test summary info ============================ +FAILED tests/test_r3.py::test_populated_migration_idempotence_and_old_code_rollback +FAILED tests/test_r3.py::test_three_endpoint_bytes_equal_product_baseline - s... +2 failed, 445 passed, 5 skipped, 1 warning in 51.70s +=== SDK harness (agent venv, livekit-agents 1.8.8) === +failed to generate LLM completion: message='chat api failure: network is genuinely down', status_code=502, retryable=True, retrying in 0.1s +failed to generate LLM completion: message='chat api failure: network is genuinely down', status_code=502, retryable=True, retrying in 0.05s +PASS pre_first_delta_cancel_never_retries +PASS mid_stream_cancel_interrupts_handle_and_never_retries +PASS reentry_after_cancel_short_circuits_without_network +PASS real_connection_errors_still_retry_not_globally_disabled +PASS completed_turn_still_streams_with_receipt +5/5 passed diff --git a/src/cybergril/api/app.py b/src/cybergril/api/app.py index 3117ad4..ae19679 100644 --- a/src/cybergril/api/app.py +++ b/src/cybergril/api/app.py @@ -48,6 +48,7 @@ from ..life.sharing import Sharing from ..memory.episodic import Memories from ..memory.semantic import build_semantic from ..memory.store import Store +from ..providers.llm.reasoning import split_reasoning from ..think.scenes import Scene from ..think.wiring import build_router_from_config from ..voice import ( @@ -574,6 +575,18 @@ def create_app( ) if previous and previous["text"] != data.text: raise HTTPException(409, "相同轮次不能改写消息。") + if previous and previous["status"] == "cancelled": + # HEL-908: a turn the server cancelled (superseded by a newer + # turn) never re-acquires the active-turn slot — a late retry + # of the same turn_id must not cancel the newer turn that + # superseded it. Answer with the same terminal cancel frame, + # idempotently, without touching any state. + async def cancelled_turn(): + yield event(data, "chat.cancelled") + + return StreamingResponse( + cancelled_turn(), media_type="text/event-stream" + ) if app.state.active: active = app.state.active if active["turn_id"] != data.turn_id: @@ -691,7 +704,14 @@ def create_app( done, _ = await asyncio.wait( [pending, cancelling], return_when=asyncio.FIRST_COMPLETED ) - if cancelling in done or await request.is_disconnected(): + if cancelling in done: + # HEL-908: cancel is terminal and explicit — the + # still-connected caller learns this turn was + # superseded instead of mistaking the stream end + # for a network failure and retrying it. + yield event(data, "chat.cancelled") + return + if await request.is_disconnected(): return try: delta = pending.result() @@ -708,9 +728,18 @@ def create_app( completed = True if delta.done: break - if active["cancel"].is_set() or await request.is_disconnected(): + if active["cancel"].is_set(): + yield event(data, "chat.cancelled") return - if not text.strip() or not completed: + if await request.is_disconnected(): + return + if not completed: + raise RuntimeError("Empty upstream response") + # HEL-908: reasoning never becomes her archived reply — the + # turn-level split is the belt-and-braces behind the adapter's + # streaming separator (residual >HOLD_LIMIT leaks die here). + text, _dropped_reasoning = split_reasoning(text) + if not text.strip(): raise RuntimeError("Empty upstream response") with store.conn: inserted = store.execute( diff --git a/src/cybergril/contracts/utterance.py b/src/cybergril/contracts/utterance.py index 1520c8b..54a6510 100644 --- a/src/cybergril/contracts/utterance.py +++ b/src/cybergril/contracts/utterance.py @@ -83,6 +83,10 @@ class LLMDelta: done: bool = False finish_reason: str | None = None usage: dict[str, int] = field(default_factory=dict) + reasoning: str = "" + """模型内部思考增量（HEL-908 分流）：适配层把 reasoning_content 字段与 + 约定块从可见正文中分离到这里；仅供诊断，绝不持久化、不下发 + chat.delta、不进字幕或 TTS。""" @dataclass(frozen=True, slots=True) diff --git a/src/cybergril/providers/llm/openai_compat.py b/src/cybergril/providers/llm/openai_compat.py index ee804e3..5a2929f 100644 --- a/src/cybergril/providers/llm/openai_compat.py +++ b/src/cybergril/providers/llm/openai_compat.py @@ -24,6 +24,7 @@ from ...contracts import ( ) from ..base import ProviderError from .base import LLMProvider +from .reasoning import ReasoningSeparator class OpenAICompatLLM(LLMProvider): @@ -107,11 +108,20 @@ class OpenAICompatLLM(LLMProvider): f"LLM {self.name} HTTP {resp.status_code}: {body[:400]}" ) + # HEL-908: reasoning/user-content separation at the adapter seam — + # reasoning_content deltas (DeepSeek-style field) go to the + # reasoning side-channel, inline conventions are stripped + # cross-chunk; only visible text ever reaches delta.text. + separator = ReasoningSeparator() + async for line in resp.aiter_lines(): if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": + tail_visible, tail_reasoning = separator.finish() + if tail_visible or tail_reasoning: + yield LLMDelta(text=tail_visible, reasoning=tail_reasoning) yield LLMDelta(done=True) return try: @@ -122,12 +132,18 @@ class OpenAICompatLLM(LLMProvider): choice = (obj.get("choices") or [{}])[0] delta = choice.get("delta") or {} text = delta.get("content") or "" + field_reasoning = ( + delta.get("reasoning_content") or delta.get("reasoning") or "" + ) + visible, dropped = separator.feed(text) if text else ("", "") + reasoning = dropped + field_reasoning finish = choice.get("finish_reason") usage = obj.get("usage") or {} - if text or finish or usage: + if visible or reasoning or finish or usage: yield LLMDelta( - text=text, + text=visible, + reasoning=reasoning, done=False, finish_reason=finish, usage=usage if isinstance(usage, dict) else {}, diff --git a/src/cybergril/providers/llm/reasoning.py b/src/cybergril/providers/llm/reasoning.py new file mode 100644 index 0000000..e8d02f4 --- /dev/null +++ b/src/cybergril/providers/llm/reasoning.py @@ -0,0 +1,201 @@ +"""Reasoning / user-content separation at the model-adapter seam (HEL-908). + +The dialogue endpoint (halogen-qwen3.8-flash-next, OpenAI-compatible) is a +Qwen3-family model served raw: some replies open a reasoning phase inline in +``content``. The recorded evidence (HEL-907 window, companion.db row id=12 and +tts-window-tail.log) shows the exact shape: + + Considering the limited time by the user, ... now.\\n\\n\\n大圣，您这定身法… + +i.e. reasoning text arrives in ``content`` with NO opening ```` (the +serving chat template pre-fills it) and a stray ```` closes it, while +other replies from the same endpoint carry no tags at all, and DeepSeek-style +endpoints instead stream a separate ``reasoning_content`` delta field. + +This module is the single boundary between raw model output and user-visible +content: visible text is the ONLY thing that may be persisted as her reply, +streamed as chat.delta (subtitles) or spoken (TTS). Reasoning is exposed on +``LLMDelta.reasoning`` for diagnostics and dropped everywhere else — it is +never persisted. This is the same ```` text convention vLLM/SGLang +reasoning parsers implement server-side (GOV-006: reuse the mature +convention, do not build a reasoning engine): no prompt changes, no model +behavior changes, and nothing is deleted except the tag-delimited reasoning +itself. + +Streaming caveat (documented residual, offline-verifiable): an implicit-open +reasoning phase is indistinguishable from visible text until its ```` +arrives, so the separator holds back the beginning of every stream until +either a tag appears, a Chinese sentence ends, or HOLD_LIMIT characters +accumulate. The recorded leak (closer at ~104 chars) is fully covered; +reasoning whose closer lies beyond HOLD_LIMIT streams partially visible — +``split_reasoning`` still cleans the archived reply and the receipt text at +turn level (api/app.py). Removing the residual needs the upstream to separate +reasoning itself (reasoning_content field or a server-side parser), which +cannot be probed under the zero-real-request rule of this rework. +""" + +from __future__ import annotations + +THINK_OPEN = "" +THINK_CLOSE = "" + +# Held prefix before the stream is known visible: comfortably above the +# recorded implicit-open reasoning length (~104 chars incl. ). +HOLD_LIMIT = 128 +# A finished Chinese sentence ends the hold: her replies are Chinese, and the +# recorded reasoning contains none of these before its closer. ASCII "." and +# newline are deliberately NOT flush triggers (both occur inside reasoning). +_SENTENCE_ENDS = "。！？!?" + + +def _partial_tag_tail(buffer: str, tags: tuple[str, ...]) -> int: + """Length of the longest buffer suffix that could still grow into one of + ``tags`` (cross-chunk tag safety); 0 when no suffix can.""" + longest = max(len(t) for t in tags) + for size in range(min(len(buffer), longest - 1), 0, -1): + if any(tag.startswith(buffer[-size:]) for tag in tags): + return size + return 0 + + +class ReasoningSeparator: + """Stateful streaming separator for one upstream reply. + + ``feed``/``finish`` return ``(visible_delta, reasoning_delta)``; the + concatenation of all visible deltas plus the finish tail is the visible + text, everything else is reasoning. Untagged content is always preserved + byte-for-byte (Chinese or English) — only tag-delimited reasoning is + dropped. + """ + + def __init__(self, *, hold_limit: int = HOLD_LIMIT) -> None: + self._hold_limit = hold_limit + self._mode = "hold" # hold | think | visible + self._buffer = "" + self._reasoning = "" + self._strip_pending = False # drop leading ws until real content (post-think) + + @property + def reasoning(self) -> str: + """All reasoning seen so far (diagnostics only; never persisted).""" + return self._reasoning + + def feed(self, chunk: str) -> tuple[str, str]: + self._buffer += chunk + return self._drain(final=False) + + def finish(self) -> tuple[str, str]: + """End of stream: resolve remaining buffer (partial tags are text).""" + return self._drain(final=True) + + def _drain(self, *, final: bool) -> tuple[str, str]: + visible = "" + reasoning = "" + while True: + if self._mode == "hold": + open_at = self._buffer.find(THINK_OPEN) + close_at = self._buffer.find(THINK_CLOSE) + if close_at != -1 and (open_at == -1 or close_at < open_at): + # implicit-open (template pre-filled ) or a normal + # block that opened at/after position 0: everything through + # the closer is reasoning + dropped = self._buffer[: close_at + len(THINK_CLOSE)] + self._reasoning += dropped + reasoning += dropped + self._buffer = self._buffer[close_at + len(THINK_CLOSE):] + self._mode = "visible" + self._strip_pending = True + continue + if open_at != -1: + visible += self._buffer[:open_at] + self._buffer = self._buffer[open_at + len(THINK_OPEN):] + self._mode = "think" + continue + if final: + self._mode = "visible" # untagged short reply: never swallow + continue + sentence_end = min( + ( + i + for i in (self._buffer.find(c) for c in _SENTENCE_ENDS) + if i != -1 + ), + default=-1, + ) + if sentence_end != -1 or len(self._buffer) >= self._hold_limit: + self._mode = "visible" + continue + return visible, reasoning + if self._mode == "think": + close_at = self._buffer.find(THINK_CLOSE) + if close_at != -1: + dropped = self._buffer[: close_at + len(THINK_CLOSE)] + self._reasoning += dropped + reasoning += dropped + self._buffer = self._buffer[close_at + len(THINK_CLOSE):] + self._mode = "visible" + self._strip_pending = True + continue + keep = 0 if final else _partial_tag_tail(self._buffer, (THINK_CLOSE,)) + cut = len(self._buffer) - keep + if cut > 0: + self._reasoning += self._buffer[:cut] + self._buffer = self._buffer[cut:] + return visible, "" + # visible + if self._strip_pending and self._buffer: + stripped = self._buffer.lstrip() + if stripped: + self._buffer = stripped + self._strip_pending = False + else: # whitespace only so far; more may follow + self._buffer = "" + open_at = self._buffer.find(THINK_OPEN) + close_at = self._buffer.find(THINK_CLOSE) + if close_at != -1 and (open_at == -1 or close_at < open_at): + # stray closer after output already streamed: drop the marker; + # split_reasoning still cleans the turn-level text + visible += self._buffer[:close_at] + self._reasoning += THINK_CLOSE + reasoning += THINK_CLOSE + self._buffer = self._buffer[close_at + len(THINK_CLOSE):] + self._strip_pending = True + continue + if open_at != -1: + visible += self._buffer[:open_at] + self._buffer = self._buffer[open_at + len(THINK_OPEN):] + self._mode = "think" + continue + if final: + visible += self._buffer + self._buffer = "" + return visible, reasoning + keep = _partial_tag_tail(self._buffer, (THINK_OPEN, THINK_CLOSE)) + cut = len(self._buffer) - keep + if cut > 0: + visible += self._buffer[:cut] + self._buffer = self._buffer[cut:] + return visible, reasoning + + +def split_reasoning(text: str) -> tuple[str, str]: + """Turn-level final split of one complete reply: everything through the + FIRST ```` (opened explicitly or implicitly) is reasoning, later + explicit ``...`` blocks are reasoning too, the rest is the + visible reply. Belt-and-braces for streams that leaked past HOLD_LIMIT — + the archived reply and its receipt always bind the clean text.""" + rest = text + reasoning = "" + close_at = rest.find(THINK_CLOSE) + if close_at != -1: + reasoning += rest[: close_at + len(THINK_CLOSE)] + rest = rest[close_at + len(THINK_CLOSE):].lstrip() + while True: + open_at = rest.find(THINK_OPEN) + if open_at == -1: + return rest, reasoning + close_at = rest.find(THINK_CLOSE, open_at) + if close_at == -1: + return rest[:open_at], reasoning + rest[open_at:] + reasoning += rest[open_at: close_at + len(THINK_CLOSE)] + rest = rest[:open_at] + rest[close_at + len(THINK_CLOSE):] diff --git a/src/cybergril/voice/chat_api_client.py b/src/cybergril/voice/chat_api_client.py index 8d6ea82..883da1d 100644 --- a/src/cybergril/voice/chat_api_client.py +++ b/src/cybergril/voice/chat_api_client.py @@ -36,6 +36,9 @@ class ChatApiTurn: receipt: dict[str, Any] | None = None error_phi: str | None = None completed: bool = False + cancelled: bool = False + """The SERVER ended this turn as superseded (chat.cancelled, HEL-908). + Terminal, not a connection error: the same turn_id must never be retried.""" @dataclass @@ -142,6 +145,12 @@ class ChatApiClient: elif kind == "chat.error": turn.error_phi = frame.get("phi") or "" got_terminal = True + elif kind == "chat.cancelled": + # HEL-908: the server cancelled this turn (superseded by a + # newer one). A clean terminal — the caller ends the turn; a + # retry of the same turn_id would only be refused again. + turn.cancelled = True + got_terminal = True if not got_terminal: raise ChatApiConnectionError("chat stream ended without final or error") return turn diff --git a/src/cybergril/voice/llm_chat_api.py b/src/cybergril/voice/llm_chat_api.py index 8b28263..13c1353 100644 --- a/src/cybergril/voice/llm_chat_api.py +++ b/src/cybergril/voice/llm_chat_api.py @@ -94,6 +94,11 @@ class ChatApiLLMStream(LLMStream): self.receipt: dict[str, Any] | None = None self.error_phi: str | None = None self.completed = False + # HEL-908: once the SERVER has cancelled this turn (chat.cancelled), + # the turn_id is dead — re-running _run must never re-acquire the + # server's active-turn slot (a late retry would otherwise cancel the + # newer turn, the exact HEL-907 defect). + self.server_cancelled = False # Exact ownership (set by CyberGrillLLM.chat from the SDK generation # context): the SpeechHandle whose pipeline task created this stream, # and its id as the subtitle turn key. None/"" = unattributed. @@ -112,7 +117,24 @@ class ChatApiLLMStream(LLMStream): ChatChunk(id=self.turn_id, delta=ChoiceDelta(role="assistant", content=content)) ) + def _interrupt_owner(self) -> None: + """Stop this generation's cached playout through the SDK's own + SpeechHandle.interrupt (the same path user barge-in takes), so a + server-side cancel stops already-synthesized audio — not just + generation. Any failure here must not mask the cancelled end.""" + handle = self.speech_handle + if handle is None: + return + with contextlib.suppress(Exception): + handle.interrupt() + async def _run(self) -> None: + if self.server_cancelled: + # A cancelled turn never re-acquires ownership, whatever re-runs + # _run (SDK retry loop, reconnect): no network call at all. + raise APIStatusError( + "chat turn cancelled by server", status_code=499, body=None + ) text = latest_user_text(self._chat_ctx) if not text: raise APIStatusError( @@ -130,6 +152,17 @@ class ChatApiLLMStream(LLMStream): raise APIStatusError( f"chat api failure: {exc}", status_code=502, body=None ) from exc + if turn.cancelled: + # HEL-908: cancelled is a TERMINAL state, never a transient + # network error. End on the SDK's own client-closed-request path: + # LLMStream._main_task (verified against installed 1.8.8 source) + # returns gracefully for APIStatusError 499 — no retry, no + # exception — and interrupt() stops the cached playout. + self.server_cancelled = True + self._interrupt_owner() + raise APIStatusError( + "chat turn cancelled by server", status_code=499, body=None + ) self.text = turn.text self.receipt = turn.receipt self.error_phi = turn.error_phi diff --git a/tests/test_api.py b/tests/test_api.py index c8dde0d..66987c0 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -314,7 +314,10 @@ async def test_disconnect_closes_real_openai_http_stream(tmp_path): async def completion(): async def chunks(): try: - yield 'data: {"choices":[{"delta":{"content":"部分"}}]}\n\n' + # ends with a sentence terminator so the HEL-908 reasoning + # separator flushes it immediately (this test is about upstream + # closure on disconnect, not about hold semantics) + yield 'data: {"choices":[{"delta":{"content":"部分。"}}]}\n\n' await asyncio.sleep(30) finally: closed.set() diff --git a/tests/test_reasoning_separation_hel908.py b/tests/test_reasoning_separation_hel908.py new file mode 100644 index 0000000..7292ecd --- /dev/null +++ b/tests/test_reasoning_separation_hel908.py @@ -0,0 +1,246 @@ +"""HEL-908 reasoning separation tests: the adapter seam between raw model +output and user-visible content, against the REAL ReasoningSeparator and the +REAL OpenAICompatLLM adapter (fake SSE transports, zero network). + +The recorded evidence (HEL-907 window, companion.db row id=12 and +tts-window-tail.log) is replayed verbatim: the dialogue endpoint streamed +"Considering the limited time by the user, ... now.\\n\\n\\n大圣，…" +into ``content`` — implicit-open reasoning (no ````, the chat template +pre-fills it) that reached subtitles AND TTS. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import httpx +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from cybergril.providers.llm.openai_compat import OpenAICompatLLM +from cybergril.providers.llm.reasoning import ( + HOLD_LIMIT, + ReasoningSeparator, + split_reasoning, +) + +# companion.db row id=12 (call3, turn 3a67b253) — verbatim recorded stream. +RECORDED_RAW = ( + "Considering the limited time by the user, I have to give the solution " + "based on the thinking directly now.\n\n\n" + "大圣，您这定身法没点准啊，我可不是什么妖精妖怪，我是一行代码，" + "连肉都没有，怕硌着您的金箍棒。" +) +RECORDED_REASONING = ( + "Considering the limited time by the user, I have to give the solution " + "based on the thinking directly now.\n" +) +RECORDED_VISIBLE = ( + "大圣，您这定身法没点准啊，我可不是什么妖精妖怪，我是一行代码，" + "连肉都没有，怕硌着您的金箍棒。" +) + + +def run_stream(text: str, chunk_size: int) -> tuple[str, str]: + separator = ReasoningSeparator() + visible_parts: list[str] = [] + for i in range(0, len(text), chunk_size): + v, _ = separator.feed(text[i : i + chunk_size]) + visible_parts.append(v) + v, _ = separator.finish() + visible_parts.append(v) + return "".join(visible_parts), separator.reasoning + + +# ── streaming separator ────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("chunk", [1, 2, 3, 5, 7, 13, 1000]) +def test_recorded_stream_leaks_nothing_at_any_chunk_size(chunk): + visible, reasoning = run_stream(RECORDED_RAW, chunk) + assert visible == RECORDED_VISIBLE + assert reasoning == RECORDED_REASONING + + +@pytest.mark.parametrize("split_at", range(1, len("隐藏可见"))) +def test_explicit_tags_split_across_chunks(split_at): + text = "隐藏可见" + head, tail = text[:split_at], text[split_at:] + separator = ReasoningSeparator() + v1, _ = separator.feed(head) + v2, _ = separator.feed(tail) + v3, _ = separator.finish() + assert v1 + v2 + v3 == "可见" + + +def test_explicit_think_block_anywhere_in_stream(): + visible, _ = run_stream("先说。中途思考后说。", 3) + assert visible == "先说。后说。" + + +def test_unclosed_think_yields_no_visible_text(): + separator = ReasoningSeparator() + visible, _ = separator.feed("只想不说到结尾") + tail_visible, _ = separator.finish() + assert visible + tail_visible == "" + + +def test_safe_no_think_chinese_streams_immediately(): + separator = ReasoningSeparator() + visible, _ = separator.feed("哈哈，那走吧。") + assert visible == "哈哈，那走吧。" + + +def test_normal_chinese_body_preserved_byte_for_byte(): + text = "今天水母网我背，你负责举着泡泡喊“出发”。\n\n抓完水母我们就坐在海滩上。" + visible, reasoning = run_stream(text, 4) + assert visible == text + assert reasoning == "" + + +def test_normal_english_body_preserved_byte_for_byte(): + text = ( + "Sure, listening to you. A reply that happens to be English is normal " + "user-facing content and must survive the separator untouched, " + "including newlines\nand punctuation." + ) + visible, reasoning = run_stream(text, 5) + assert visible == text + assert reasoning == "" + + +def test_untagged_short_stream_flushes_at_finish(): + separator = ReasoningSeparator() + v, _ = separator.feed("嗯") # below HOLD_LIMIT, no sentence end: held + assert v == "" + v, _ = separator.finish() + assert v == "嗯" + + +def test_residual_beyond_hold_limit_is_cleaned_at_turn_level(): + """Documented residual: implicit-open reasoning whose closer lies beyond + HOLD_LIMIT streams partially visible; the turn-level split still returns + the clean reply for persistence and the receipt.""" + long_reasoning = "r" * (HOLD_LIMIT + 50) + "" + text = long_reasoning + "干净的回答。" + streamed_visible, _ = run_stream(text, 64) + assert streamed_visible != "干净的回答。" # partial streaming leak (known) + final_visible, dropped = split_reasoning(text) + assert final_visible == "干净的回答。" + assert long_reasoning in dropped + + +# ── turn-level final split ─────────────────────────────────────────────────── + + +def test_split_recorded_row_is_clean(): + visible, reasoning = split_reasoning(RECORDED_RAW) + assert visible == RECORDED_VISIBLE + assert reasoning == RECORDED_REASONING + + +def test_split_explicit_blocks_and_unclosed(): + assert split_reasoning("a答") == ("答", "a") + assert split_reasoning("a") == ("", "a") + assert split_reasoning("正常回答") == ("正常回答", "") + visible, reasoning = split_reasoning("a一b二") + assert visible == "一二" + assert reasoning == "ab" + + +# ── REAL OpenAICompatLLM adapter with fake upstream SSE ───────────────────── + + +def sse_response(deltas: list[dict]) -> httpx.Response: + lines = [] + for d in deltas: + lines.append("data: " + json.dumps({"choices": [{"delta": d}]}, ensure_ascii=False)) + lines.append("data: [DONE]") + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, + content=("\n\n".join(lines) + "\n\n").encode(), + ) + + +async def collect_provider(provider: OpenAICompatLLM) -> tuple[str, str]: + text = reasoning = "" + async for delta in provider.stream([type("M", (), {"role": "user", "content": "你好"})()]): + text += delta.text + reasoning += delta.reasoning + return text, reasoning + + +async def test_adapter_separates_reasoning_field_from_content(): + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return sse_response( + [ + {"reasoning_content": "先在心里盘算一下"}, + {"content": ""}, + {"content": RECORDED_VISIBLE[:4]}, + {"content": RECORDED_VISIBLE[4:]}, + ] + ) + + provider = OpenAICompatLLM(base_url="http://isolated/v1", model="m") + provider._client = httpx.AsyncClient( + base_url="http://isolated/v1", transport=httpx.MockTransport(handler) + ) + try: + text, reasoning = await collect_provider(provider) + assert text == RECORDED_VISIBLE + assert reasoning == "先在心里盘算一下" + finally: + await provider.aclose() + + +async def test_adapter_filters_recorded_inline_leak(): + def handler(request: httpx.Request) -> httpx.Response: + # recorded id=12 shape: reasoning text + stray closer, no opening tag; + # chunks split the closing tag across SSE frames + return sse_response( + [ + {"content": RECORDED_RAW[:60]}, + {"content": RECORDED_RAW[60:104]}, + {"content": RECORDED_RAW[104:112]}, + {"content": RECORDED_RAW[112:]}, + ] + ) + + provider = OpenAICompatLLM(base_url="http://isolated/v1", model="m") + provider._client = httpx.AsyncClient( + base_url="http://isolated/v1", transport=httpx.MockTransport(handler) + ) + try: + text, reasoning = await collect_provider(provider) + assert text == RECORDED_VISIBLE, "reasoning must never reach delta.text" + assert "Considering" in reasoning and "" in reasoning + finally: + await provider.aclose() + + +async def test_adapter_preserves_english_content_and_flags(): + def handler(request: httpx.Request) -> httpx.Response: + return sse_response( + [ + {"content": "Plain English reply, kept fully. "}, + {"reasoning_content": "internal note"}, + {"content": "Second sentence."}, + ] + ) + + provider = OpenAICompatLLM(base_url="http://isolated/v1", model="m") + provider._client = httpx.AsyncClient( + base_url="http://isolated/v1", transport=httpx.MockTransport(handler) + ) + try: + text, reasoning = await collect_provider(provider) + assert text == "Plain English reply, kept fully. Second sentence." + assert reasoning == "internal note" + finally: + await provider.aclose() diff --git a/tests/test_rework_hel908.py b/tests/test_rework_hel908.py new file mode 100644 index 0000000..25581e8 --- /dev/null +++ b/tests/test_rework_hel908.py @@ -0,0 +1,366 @@ +"""HEL-908 rework tests: text-interrupt turn semantics through the REAL app. + +Replays the recorded HEL-907 window defects offline (fake scripted LLM +provider, real FastAPI app + real ChatApiClient protocol client, zero network +beyond 127.0.0.1): + +1. call1 third-turn defect: her voice turn was cancelled by the user's text + POST BEFORE its first delta; the SSE then ended without a terminal frame, + the worker's adapter raised a connection error, livekit-agents silently + RETRIED the same turn_id, and that retry cancelled the text turn and + re-spoke a full ~32s reply (worker.log line 108, api.log POST #5, + chat_message ids 5/6/8). +2. call3 think leak: the model's implicit-open reasoning + stray ```` + streamed into chat.delta and TTS (companion.db row id=12). +""" + +from __future__ import annotations + +import asyncio +import hashlib +import sys +from pathlib import Path +from uuid import uuid4 + +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 conftest import engineering_budget # noqa: E402 +from test_r2 import R2LLM # noqa: E402 + +from cybergril.api.app import create_app # noqa: E402 +from cybergril.contracts import LLMDelta # 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 ChatApiClient # noqa: E402 + +CREATE = {"name": "返修", "birth": "2000-10-18", "hour": "亥时", "relation": "朋友"} + +# companion.db row id=12 (call3), verbatim: implicit-open reasoning + stray +# closer with NO opening tag, then her actual reply. +RECORDED_RAW = ( + "Considering the limited time by the user, I have to give the solution " + "based on the thinking directly now.\n\n\n" + "大圣，您这定身法没点准啊，我可不是什么妖精妖怪，我是一行代码，" + "连肉都没有，怕硌着您的金箍棒。" +) +RECORDED_VISIBLE = ( + "大圣，您这定身法没点准啊，我可不是什么妖精妖怪，我是一行代码，" + "连肉都没有，怕硌着您的金箍棒。" +) +RECORDED_CHUNKS = [RECORDED_RAW[:60], RECORDED_RAW[60:104], + RECORDED_RAW[104:112], RECORDED_RAW[112:]] + + +class ScriptedLLM(R2LLM): + """R2 controlled provider with HEL-908 scripted dialogue behaviour.""" + + def __init__(self): + super().__init__() + self.script: list[str] = [] + self.hang_mode = "" # "" | "before" | "after" + self.hang_release = asyncio.Event() # unset: a hang parks until released + self.reply = "文字轮正常拿到了回复。" + + async def stream(self, messages, **kwargs): + if "只返回JSON，不要Markdown" in messages[0].content: + async for delta in super().stream(messages, **kwargs): + yield delta + return + self.calls.append(messages) + if self.hang_mode == "before": + await self.hang_release.wait() + if self.mode == "error": + raise OSError("scripted upstream fault") + for piece in self.script: + yield LLMDelta(text=piece) + if not self.script: + yield LLMDelta(text=self.reply) + if self.hang_mode == "after": + await self.hang_release.wait() + yield LLMDelta(done=True) + + +@pytest.fixture +def rework(tmp_path): + provider = ScriptedLLM() + router = build_router( + providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)}, + budget=engineering_budget(), + ) + app = create_app(db_path=tmp_path / "hel908.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 = "" + + +def sse_chunks(raw_pieces: list[str]) -> bytes: + """SSE lines for a raw content stream (mirrors the endpoint's wire shape).""" + import json as _json + + lines = [ + "data: " + _json.dumps({"choices": [{"delta": {"content": piece}}]}, + ensure_ascii=False) + for piece in raw_pieces + ] + lines.append("data: [DONE]") + return ("\n\n".join(lines) + "\n\n").encode() + + +@pytest.fixture +def adapter_app(tmp_path): + """The REAL OpenAICompatLLM behind the real app, fed by a scripted + OpenAI-compatible SSE gateway (the seam where the reasoning separation + lives, exercised end to end).""" + from cybergril.providers.llm.openai_compat import OpenAICompatLLM + + state = {"answers": [sse_chunks(RECORDED_CHUNKS)]} + + async def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/models"): + return httpx.Response(200, json={"data": [{"id": "recorded"}]}) + answer = state["answers"][0] + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=answer + ) + + provider = OpenAICompatLLM(base_url="http://recorded/v1", model="recorded") + provider._client = httpx.AsyncClient( + base_url="http://recorded/v1", transport=httpx.MockTransport(handler) + ) + router = build_router( + providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)}, + budget=engineering_budget(), + ) + app = create_app(db_path=tmp_path / "hel908-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, state, epoch + + +def api_client(app) -> ChatApiClient: + return ChatApiClient("http://asgi", transport=httpx.ASGITransport(app=app)) + + +@pytest.fixture +def live_rework(tmp_path): + """Real-socket variant (uvicorn on 127.0.0.1, ephemeral port) for the + cases that need genuine incremental SSE delivery — httpx.ASGITransport + does not stream response bodies chunk-by-chunk mid-response. Same pattern + as test_voice_chat_adapter's live_server: its own app instance on the same + file-backed database (one lifespan owner per app).""" + import socket + import threading + import time as _time + + import uvicorn + + provider = ScriptedLLM() + router = build_router( + providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)}, + budget=engineering_budget(), + ) + served = create_app(db_path=tmp_path / "hel908-live.db", router=router, + life_config=LifeConfig(enabled=True, poll_seconds=3600), + semantic_mode="stub") + from fastapi.testclient import TestClient + + with TestClient(served) as client: + client.post("/api/companion", json=CREATE) + epoch = client.get("/api/health").json()["epoch"] + + probe = socket.socket() + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + probe.close() + server = uvicorn.Server(uvicorn.Config(served, host="127.0.0.1", port=port, + log_level="error")) + 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}", served, provider, epoch + server.should_exit = True + thread.join(timeout=5) + + +def rows(app): + return app.state.store.query( + "SELECT id,role,text,status,turn_id FROM chat_message ORDER BY id" + ) + + +async def test_text_interrupt_pre_first_delta_no_retry_no_revive(rework): + """The recorded call1 path: cancel arrives before the first delta. + + The cancelled voice stream must end on an explicit chat.cancelled + terminal; a late retry of the same turn_id must be refused without + re-acquiring the gate (so it can never cancel the text turn), and no + regeneration may run.""" + app, provider, epoch = rework + provider.hang_mode = "before" # voice turn parks before its first token + voice_tid = str(uuid4()) + deltas: list[str] = [] + voice = asyncio.create_task( + api_client(app).send_turn("你是不是早就想睡我了呀", turn_id=voice_tid, + epoch=epoch, on_delta=deltas.append) + ) + await asyncio.sleep(0.5) # voice turn is the active one, pre-first-delta + provider.hang_mode = "" + + text = await api_client(app).send_turn("（文字通道）我改用打字了", epoch=epoch) + assert text.completed and text.text == provider.reply, \ + "the newer text turn keeps the gate and gets its reply" + + cancelled_voice = await voice + assert cancelled_voice.cancelled and not cancelled_voice.completed + assert deltas == [], "recorded case: no chunk was ever sent before cancel" + + calls_after_window = len(provider.calls) + retry = await api_client(app).send_turn("你是不是早就想睡我了呀", + turn_id=voice_tid, epoch=epoch) + assert retry.cancelled and not retry.completed, \ + "a cancelled turn_id is terminal — the refusal answers chat.cancelled" + assert len(provider.calls) == calls_after_window, \ + "the refused retry must not regenerate (no re-spoken 32s reply)" + + assert len(provider.calls) == 2 # exactly: voice attempt + text turn + table = rows(app) + voice_rows = [r for r in table if r["turn_id"] == voice_tid] + assert voice_rows and voice_rows[0]["status"] == "cancelled" + assert not [r for r in table + if r["role"] == "companion" and r["turn_id"] == voice_tid], \ + "the superseded voice turn never becomes her archived reply" + text_rows = [r for r in table if r["text"] == "（文字通道）我改用打字了"] + assert text_rows and text_rows[0]["status"] == "sent" + assert any(r["role"] == "companion" and r["status"] == "sent" + for r in table), "the text turn's reply is archived" + + +async def test_text_interrupt_after_deltas_cancels_cleanly(live_rework): + """Cancel AFTER the voice turn already streamed: the caller keeps the + deltas it saw, then receives the terminal cancel frame. Real socket: the + delta must arrive DURING the parked stream (ASGITransport buffers).""" + url, app, provider, epoch = live_rework + provider.script = ["哈哈，那走吧。"] + provider.hang_mode = "after" + got_delta = asyncio.Event() + deltas: list[str] = [] + voice_tid = str(uuid4()) + live_client = ChatApiClient(url) + voice = asyncio.create_task( + live_client.send_turn("语音说到一半", turn_id=voice_tid, epoch=epoch, + on_delta=lambda d: (deltas.append(d), got_delta.set())) + ) + await asyncio.wait_for(got_delta.wait(), timeout=10) + provider.hang_mode = "" + + text = await live_client.send_turn("文字打断", epoch=epoch) + assert text.completed + + cancelled_voice = await voice + assert cancelled_voice.cancelled + assert "".join(deltas) == "哈哈，那走吧。" + assert not [r for r in rows(app) + if r["role"] == "companion" and r["turn_id"] == voice_tid] + await live_client.aclose() + + +async def test_think_leak_replay_clean_deltas_archive_and_receipt(adapter_app): + """The recorded id=12 stream through the real adapter and the whole app: + chat.delta, the archived row and the receipt all bind the clean visible + reply only.""" + app, _, epoch = adapter_app + deltas: list[str] = [] + turn = await api_client(app).send_turn("老孙来也", epoch=epoch, + on_delta=deltas.append) + assert turn.completed + assert turn.text == RECORDED_VISIBLE + assert "".join(deltas) == RECORDED_VISIBLE, \ + "reasoning and must never stream to subtitles/TTS" + assert not any("Considering" in d or "think" in d for d in deltas) + + archived = [r for r in rows(app) if r["role"] == "companion"] + assert archived and archived[0]["text"] == RECORDED_VISIBLE + assert turn.receipt and turn.receipt["text_hash"] == \ + hashlib.sha256(RECORDED_VISIBLE.encode()).hexdigest() + assert await api_client(app).acknowledge(turn.receipt) is True + + +async def test_unclosed_think_fails_closed(adapter_app): + """A reply that never leaves its think phase has no visible content: the + turn fails honestly instead of archiving reasoning as her words.""" + app, state, epoch = adapter_app + state["answers"] = [sse_chunks(["只想不说到结尾"])] + turn = await api_client(app).send_turn("这次只有思考", epoch=epoch) + assert not turn.completed and not turn.cancelled + assert turn.error_phi and "没接上话" in turn.error_phi + assert not [r for r in rows(app) if r["role"] == "companion"] + failed = [r for r in rows(app) if r["text"] == "这次只有思考"] + assert failed and failed[0]["status"] == "failed" + + +async def test_completed_replay_and_failed_retry_still_work(rework): + """The refusal branch must not disturb replay idempotence or the existing + retry-after-failure resilience (only 'cancelled' is terminal).""" + app, provider, epoch = rework + tid = str(uuid4()) + first = await api_client(app).send_turn("正常一轮", turn_id=tid, epoch=epoch) + assert first.completed + + replay = await api_client(app).send_turn("正常一轮", turn_id=tid, epoch=epoch) + assert replay.completed and replay.text == first.text + calls = len(provider.calls) + + provider.mode = "error" + failed_tid = str(uuid4()) + failed = await api_client(app).send_turn("会失败的一轮", turn_id=failed_tid, + epoch=epoch) + assert not failed.completed and failed.error_phi + provider.mode = "" + retried = await api_client(app).send_turn("会失败的一轮", turn_id=failed_tid, + epoch=epoch) + assert retried.completed, "failed turns stay retryable (only cancelled is final)" + assert len(provider.calls) == calls + 2 + + +async def test_client_treats_cancelled_frame_as_terminal(rework): + """Unit-level protocol check: chat.cancelled ends the turn client-side + without raising (a raise is what made livekit retry it as a network + failure).""" + import json as _json + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200, json={"epoch": 1}) + turn_id = _json.loads(request.read())["turn_id"] + frame = _json.dumps( + {"t": "chat.cancelled", "epoch": 1, "turn_id": turn_id} + ).encode() + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, + content=b"data: " + frame + b"\n\n", + ) + + client = ChatApiClient("http://x", transport=httpx.MockTransport(handler)) + turn = await client.send_turn("只收一帧") + assert turn.cancelled and not turn.completed and not turn.error_phi diff --git a/tests/test_sdk_livekit_hel908.py b/tests/test_sdk_livekit_hel908.py new file mode 100644 index 0000000..159449e --- /dev/null +++ b/tests/test_sdk_livekit_hel908.py @@ -0,0 +1,246 @@ +"""HEL-908 SDK-path harness: the cancelled-turn fix through the REAL +livekit-agents 1.8.8 LLMStream machinery. + +These checks drive the installed SDK's own retry/cancel code (``LLMStream +._main_task``, verified read-only on .12) against fake SSE transports — no +room, no LLM/ASR/TTS call, zero network beyond the in-process transport. + +They need ``livekit.agents`` importable, i.e. the agent venv on .12 +(`/opt/voice-agent/agent/venv`, which has no pytest). Two ways to run: + + pytest tests/test_sdk_livekit_hel908.py + /opt/voice-agent/agent/venv/bin/python \\ + tests/test_sdk_livekit_hel908.py --standalone + +The HEL-907 defect under test: worker.log line-108's retry +("chat stream ended without final or error ... retrying in 0.1s") fired 13ms +after the user's text POST — the cancelled stream ended WITHOUT a terminal +frame BEFORE any chunk was sent, so ``output_sent`` was False, the +``_retry_on_chunk_sent`` guard was inapplicable, and the SDK silently re-ran +``_run``: the retried POST cancelled the user's text turn and re-spoke a full +~32s reply (api.log POST #5, chat_message ids 5/6/8). +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +import httpx + +try: + import pytest +except ImportError: # the agent venv has no pytest; standalone mode only + pytest = None # type: ignore[assignment] + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +try: # the SDK only exists in the agent venv / worker environments + import livekit.agents.llm as lk_llm + from livekit.agents import APIConnectionError, APIStatusError + from livekit.agents.types import APIConnectOptions + + HAVE_SDK = True + SKIP_REASON = "" +except Exception as exc: # pragma: no cover - environment-dependent + HAVE_SDK = False + SKIP_REASON = f"livekit-agents not importable in this env: {exc}" + +if pytest is not None: + pytestmark = pytest.mark.skipif(not HAVE_SDK, reason=SKIP_REASON or "sdk present") + + +class _Raises: + """Minimal pytest.raises stand-in so the standalone runner (agent venv, + no pytest installed) executes the very same assertions.""" + + def __init__(self, exc_type): + self.exc_type = exc_type + self.value = None + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + assert exc_type is not None and issubclass(exc_type, self.exc_type), ( + f"expected {self.exc_type.__name__}, got {exc_type.__name__ if exc_type else exc_type}" + ) + self.value = exc + return True + + +def raises(exc_type): + return pytest.raises(exc_type) if pytest is not None else _Raises(exc_type) + +if HAVE_SDK: + FAST = APIConnectOptions(timeout=10, retry_interval=0.05, max_retry=2) + + +def sse(frames: list[str]) -> bytes: + return "".join(f"data: {frame}\n\n" for frame in frames).encode() + + +def chat_ctx(): + ctx = lk_llm.ChatContext() + ctx.add_message(role="user", content="你好呀") + return ctx + + +class Backend: + """Scripted /api/chat backend: one canned answer per POST (Exception + instances are raised), counting every POST it serves.""" + + def __init__(self, answers: list): + self.answers = answers + self.chat_posts = 0 + + def transport(self) -> httpx.MockTransport: + backend = self + + async def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200, json={"epoch": 1}) + backend.chat_posts += 1 + answer = backend.answers[(backend.chat_posts - 1) % len(backend.answers)] + if isinstance(answer, Exception): + raise answer + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=answer + ) + + return httpx.MockTransport(handler) + + +def make_llm(answers: list): + from cybergril.voice.chat_api_client import ChatApiClient + from cybergril.voice.llm_chat_api import CyberGrillLLM + + backend = Backend(answers) + return CyberGrillLLM( + "http://x", client=ChatApiClient("http://x", transport=backend.transport()) + ), backend + + +class InterruptSpy: + """Stands in for the SDK SpeechHandle: records interrupt() calls only.""" + + def __init__(self) -> None: + self.interrupts = 0 + + def interrupt(self): + self.interrupts += 1 + return self + + +CANCELLED = sse(['{"t":"chat.cancelled","epoch":1,"turn_id":null}']) +TWO_DELTAS_THEN_CANCELLED = sse([ + '{"t":"chat.delta","epoch":1,"turn_id":null,"delta":"哈哈，"}', + '{"t":"chat.delta","epoch":1,"turn_id":null,"delta":"那走吧。"}', + '{"t":"chat.cancelled","epoch":1,"turn_id":null}', +]) +COMPLETED = sse([ + '{"t":"chat.delta","epoch":1,"turn_id":null,"delta":"完整"}', + '{"t":"chat.delta","epoch":1,"turn_id":null,"delta":"回复。"}', + '{"t":"chat.final","epoch":1,"turn_id":null,"utterance":{"text":"完整回复。"},' + '"receipt":{"message_id":9,"turn_id":null,"epoch":1,"token":"' + "k" * 48 + '","text_hash":"' + "h" * 64 + '"}}', +]) + + +async def _collect(stream): + chunks: list[str] = [] + try: + async for chunk in stream: + if chunk.delta and chunk.delta.content: + chunks.append(chunk.delta.content) + finally: + await stream.aclose() + return chunks + + +async def test_pre_first_delta_cancel_never_retries(): + """The recorded call1 path: cancel before any chunk. The 499 end means + the SDK's retry loop never re-runs _run — exactly one POST.""" + llm, backend = make_llm([CANCELLED]) + stream = llm.chat(chat_ctx=chat_ctx(), conn_options=FAST) + chunks = await _collect(stream) + assert chunks == [] + assert stream.server_cancelled and not stream.completed + assert backend.chat_posts == 1, "a cancelled turn must never be retried" + + +async def test_mid_stream_cancel_interrupts_handle_and_never_retries(): + llm, backend = make_llm([TWO_DELTAS_THEN_CANCELLED]) + stream = llm.chat(chat_ctx=chat_ctx(), conn_options=FAST) + spy = InterruptSpy() + stream.speech_handle = spy + chunks = await _collect(stream) + assert chunks == ["哈哈，", "那走吧。"], "chunks already sent stay visible" + assert stream.server_cancelled + assert spy.interrupts == 1, \ + "server-side cancel must stop the cached playout via SpeechHandle.interrupt" + assert backend.chat_posts == 1 + + +async def test_reentry_after_cancel_short_circuits_without_network(): + llm, backend = make_llm([CANCELLED]) + stream = llm.chat(chat_ctx=chat_ctx(), conn_options=FAST) + await _collect(stream) + assert backend.chat_posts == 1 + with raises(APIStatusError) as excinfo: + await stream._run() + assert excinfo.value.status_code == 499 + assert backend.chat_posts == 1, \ + "a cancelled turn_id never re-acquires the gate, not even in-process" + + +async def test_real_connection_errors_still_retry_not_globally_disabled(): + llm, backend = make_llm([httpx.ConnectError("network is genuinely down")]) + stream = llm.chat(chat_ctx=chat_ctx(), conn_options=FAST) + with raises(APIConnectionError): + await _collect(stream) + assert backend.chat_posts == FAST.max_retry + 1, \ + "transient network failures keep the SDK's own bounded retry" + + +async def test_completed_turn_still_streams_with_receipt(): + llm, backend = make_llm([COMPLETED]) + stream = llm.chat(chat_ctx=chat_ctx(), conn_options=FAST) + chunks = await _collect(stream) + assert chunks == ["完整", "回复。"] + assert stream.completed and stream.receipt and stream.text == "完整回复。" + assert backend.chat_posts == 1 + + +# ── standalone runner (agent venv has no pytest) ───────────────────────────── + + +async def _standalone() -> int: + checks = [ + ("pre_first_delta_cancel_never_retries", test_pre_first_delta_cancel_never_retries), + ("mid_stream_cancel_interrupts_handle_and_never_retries", + test_mid_stream_cancel_interrupts_handle_and_never_retries), + ("reentry_after_cancel_short_circuits_without_network", + test_reentry_after_cancel_short_circuits_without_network), + ("real_connection_errors_still_retry_not_globally_disabled", + test_real_connection_errors_still_retry_not_globally_disabled), + ("completed_turn_still_streams_with_receipt", + test_completed_turn_still_streams_with_receipt), + ] + failures = 0 + for name, check in checks: + try: + await check() + print(f"PASS {name}") + except Exception as exc: # noqa: BLE001 - report and continue + failures += 1 + print(f"FAIL {name}: {type(exc).__name__}: {exc}") + print(f"{len(checks) - failures}/{len(checks)} passed") + return 1 if failures else 0 + + +if __name__ == "__main__": + if not HAVE_SDK: + print(f"SKIP: {SKIP_REASON}") + raise SystemExit(77) + raise SystemExit(asyncio.run(_standalone())) diff --git a/tests/test_voice_chat_adapter.py b/tests/test_voice_chat_adapter.py index 2a34b18..4f78e77 100644 --- a/tests/test_voice_chat_adapter.py +++ b/tests/test_voice_chat_adapter.py @@ -139,13 +139,18 @@ async def test_broken_stream_raises_connection_error(): async def test_voice_turn_cancels_active_text_turn_same_gate(api): - """Voice and text contend through the server's one same-turn gate.""" + """Voice and text contend through the server's one same-turn gate. + + HEL-908: the cancelled side now ends with an explicit chat.cancelled + terminal frame — a clean end, not a connection error the caller might + mistake for a transient failure and retry (the HEL-907 defect family).""" app, provider, _ = api + outcome: dict = {} + 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("文字说一半") + outcome["turn"] = await text_client.send_turn("文字说一半") provider.mode = "slow" task = asyncio.create_task(slow_text()) @@ -155,8 +160,10 @@ async def test_voice_turn_cancels_active_text_turn_same_gate(api): 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) + await asyncio.wait_for(task, timeout=10) + superseded = outcome["turn"] + assert superseded.cancelled and not superseded.completed, \ + "the cancelled side ends on the explicit chat.cancelled terminal frame" rows = app.state.store.query( "SELECT role, text, status, turn_id FROM chat_message ORDER BY id"