HEL-908: terminal chat.cancelled + no-revive gate for text interrupt; reasoning/content separation at the adapter seam

Bug 1 (recorded call1): server-side cancel ended the voice SSE without a
terminal frame BEFORE the first delta (worker.log:108 SDK retry 13ms after
the text POST); the adapter raised a connection error, livekit-agents 1.8.8
silently retried the same turn_id, the retry re-acquired the gate, cancelled
the user's text turn and re-spoke a full ~32s reply. The HEL-907 claim that
the _retry_on_chunk_sent guard never worked is corrected: the guard only
applies after output (output_sent), verified against the installed SDK
source. Fix: explicit chat.cancelled terminal frame; a cancelled turn_id is
refused re-activation (never touches the newer active turn); the worker ends
cancelled turns on the SDK's own 499 path (no retry) and stops the cached
playout via SpeechHandle.interrupt(); a server_cancelled latch short-circuits
any re-entry with zero network. Failed-turn retries, completed-turn replay
and receipt validation are unchanged (dedicated tests).

Bug 2 (recorded row id=12): the endpoint streams implicit-open reasoning
plus a stray </think> inline in content into subtitles and TTS. Fix:
ReasoningSeparator at the OpenAI-compat adapter seam (cross-chunk safe,
reasoning_content field side-channel on LLMDelta.reasoning, never persisted)
plus a turn-level split before persist/final/receipt. Recorded stream
replays clean at every chunk size; normal zh/en content preserved
byte-for-byte; documented residual: implicit-open reasoning whose closer
lies beyond the 128-char hold streams partially (persistence stays clean).

Offline verification: .venv311 full suite 445 passed / 5 skipped (2
pre-existing test_r3 failures fail identically at 5625de6); real-SDK harness
(livekit-agents 1.8.8, agent venv) 5/5; reasoning separation 37/37; ruff
clean. Zero real requests, zero deployment, no dependency changes.

Co-authored-by: multica-agent <github@multica.ai>

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…/</think> 进入字幕与 TTS』：端点把隐式开头（无 <think>、模板预填）的推理原文+stray </think> 混入 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 字段与 <think> 约定，不造推理引擎、不改提示词、不删英文），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 </think>）经真实 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() 断言，不冒充端到端毫秒指标",
+    "思考分流流式残留：隐式开头且 </think> 落在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 本文件通过（日志随交付）"
+  ]
+}

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项；来源和覆盖完整。仍须人工核对授权语义与实际体验。

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…`/`</think>` 进入字幕与 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</think>\n\n大圣，您这定身法…"`——**`content` 里只有 stray `</think>`、没有开头 `<think>`**（服务端 chat template 预填了开头），推理原文因此流进 chat.delta → 字幕与 TTS（tts-window-tail.log 两个 segment 起头即 `Considering the limi…` 与 `</think>`）。同一端点同窗口亦有完全干净回复（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 字符分块全零泄漏；`</think>` 跨 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. **思考分流流式残留**：隐式开头（无 `<think>`）且收尾 `</think>` 落在 128 字符 hold 之外的推理，前缀仍会部分流出到字幕/TTS（无法在流式下提前判定）；落库文本与 receipt 始终干净（轮级兜底，专项用例声明该边界）。彻底消除需上游自行分流（`reasoning_content` 字段或服务端 reasoning parser）——零真实请求边界内无法探测该端点能力，这是**最小缺口**，不虚称已完全修好。
+2. 修复后的真实端到端复测（真实房间/真人打断时延/真机）未做，交回总管按授权窗口安排；停播路径用真实 SDK 的 `SpeechHandle.interrupt()` 断言证明，不冒充端到端毫秒指标。
+3. 前端零改动（新帧 kind 被现有 ChatPanel 静默忽略，行为不差于现状）；把“被新轮打断”做成可见界面状态属界面核心，交视觉总监。

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

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(

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 字段与
+    <think> 约定块从可见正文中分离到这里；仅供诊断，绝不持久化、不下发
+    chat.delta、不进字幕或 TTS。"""
 
 
 @dataclass(frozen=True, slots=True)

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 <think> 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 {},

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</think>\\n\\n大圣，您这定身法…
+
+i.e. reasoning text arrives in ``content`` with NO opening ``<think>`` (the
+serving chat template pre-fills it) and a stray ``</think>`` 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 ``<think>`` 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 ``</think>``
+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>"
+THINK_CLOSE = "</think>"
+
+# Held prefix before the stream is known visible: comfortably above the
+# recorded implicit-open reasoning length (~104 chars incl. </think>).
+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 <think>) 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 ``</think>`` (opened explicitly or implicitly) is reasoning, later
+    explicit ``<think>...</think>`` 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):]

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

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

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()

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</think>\\n\\n大圣，…"
+into ``content`` — implicit-open reasoning (no ``<think>``, 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</think>\n\n"
+    "大圣，您这定身法没点准啊，我可不是什么妖精妖怪，我是一行代码，"
+    "连肉都没有，怕硌着您的金箍棒。"
+)
+RECORDED_REASONING = (
+    "Considering the limited time by the user, I have to give the solution "
+    "based on the thinking directly now.\n</think>"
+)
+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("<think>隐藏</think>可见")))
+def test_explicit_tags_split_across_chunks(split_at):
+    text = "<think>隐藏</think>可见"
+    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("先说。<think>中途思考</think>后说。", 3)
+    assert visible == "先说。后说。"
+
+
+def test_unclosed_think_yields_no_visible_text():
+    separator = ReasoningSeparator()
+    visible, _ = separator.feed("<think>只想不说到结尾")
+    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) + "</think>"
+    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("<think>a</think>答") == ("答", "<think>a</think>")
+    assert split_reasoning("<think>a") == ("", "<think>a")
+    assert split_reasoning("正常回答") == ("正常回答", "")
+    visible, reasoning = split_reasoning("<think>a</think>一<think>b</think>二")
+    assert visible == "一二"
+    assert reasoning == "<think>a</think><think>b</think>"
+
+
+# ── 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 "</think>" 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()

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 ``</think>``
+   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</think>\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 </think> 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(["<think>只想不说到结尾"])]
+    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

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()))

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"