HEL-879 timing closure: the three contract-approval findings fixed as implementation-timing only (v2 envelope fields/event kinds untouched, web/src untouched, no re-audit of passed binding/backup/budget work) - (1) open_session re-kicks the cleanup owner AFTER registration+persist: a real dispatch await yields the loop and the pre-dispatch kick's cleaner already exited with nothing to watch (manager repro: 20ms real dispatch await, TTL 25ms, 120ms of silence -> DELETE=0, expired registration still present, cleaner=None; fixed: exactly one DELETE, registration settled, no dangling coroutine - the bounded auto-expiry/shared-cap mechanism itself is unchanged), new regressions reproduce first on the old code with a genuinely-yielding dispatch stand-in (asyncio.sleep, not the instantly-returning MockTransport) plus an earlier-deadline wake negative; (2) speak_proactive_row takes the handle the real sync AgentSession.say returns (iscoroutine only unwraps await-style stand-ins - the real SpeechHandle itself defines __await__, so awaiting it would wait out the whole playout), publishes proactive_start IMMEDIATELY, registers the handle's own end/cancel, then waits out that one row (sequential rows kept); .12 real-SDK probe: start visible while handle_done=false, end only after, cancel-only on interrupt, v2 fields identical; (3) _after_playout gates assistant_final on the completed+legal-same-turn-receipt conditions BEFORE anything is sent - incomplete streams, missing/partial receipts and stale epochs close with turn_cancel only (reason incomplete/no-receipt), never final-then-unack; the complete+receipt path keeps BOTH final and ack; phi/interrupt/failure/unattributed fail-closed unchanged; real-SDK check extended to 14/14 ALL PASS on .12 (proactive timing, proactive interrupt, no-receipt no-final negatives all driving the actual worker module's registered callbacks), before/after evidence in docs/rework/hel879/timing-fix/ and sdk-check-hel879-timing.log, reusable probe generator scripts/gen_hel879_timing_probe.py; regression at this head: pytest 352/352 x3 consecutive runs, ruff(src/tests/tools) clean, verify_hel877_voice_chain 8/8, requirements checks 4/4 (origin/main and c121c42, each with/without handoff-HEL-879.json, handoff appended with this round's scope); zero real LLM/embed/ASR/TTS, zero downloads, .12 touched read-only via SSH stdin only, not deployed, nothing merged, PR#11 and hel-870-assembly untouched, real caps stay 0

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

docs/rework/hel879/sdk-check-hel879-timing.log

@@ -0,0 +1,15 @@
+HEL-877/879 real-SDK check on livekit-agents 1.8.8: ALL PASS
+  OK CyberGrillLLM/ChatApiLLMStream subclass real livekit-agents 1.8.8 LLM/LLMStream
+  OK latest_user_text reads a real ChatContext (items/role/text_content)
+  OK fake-SSE turn through real LLMStream emits ChatChunks, final text and receipt
+  OK chat.error phi is spoken verbatim via ChatChunk and carries no receipt
+  OK mint_token claims byte-parity with livekit.api.AccessToken (video/iss/sub/keys)
+  OK mint_server_token matches SDK no-grant auth header claims
+  OK worker.py SDK surface (SpeechHandle, UserInputTranscribedEvent.is_final) present
+  OK barge-in task cancel closes the in-flight HTTP stream immediately
+  OK installed-SDK source verified: speech tasks set agents_speech_handle around the pipeline; speech_created precedes the task; llm_node calls llm.chat inside it
+  OK ACTUAL worker module loaded; its registered callbacks driven on a real AgentSession: both-queued delayed generations bind exactly (manager interleaving), say never steals a stream, unattributed fails closed, interrupt cancels without final/receipt
+  OK assistant_interim seam: real LLMStream deltas feed per-speech_id v2 payloads (replace semantics); unattributed streams feed nothing
+  OK speak_proactive_row publishes proactive_start while the real SpeechHandle is still playing (sync-say shape), end/cancel on the handle's done callback
+  OK interrupted proactive row emits proactive_cancel(reason=interrupted), never proactive_end
+  OK incomplete/no-receipt attributed stream: actual worker callbacks emit turn_start -> turn_cancel only, no assistant_final, no ack

docs/rework/hel879/timing-fix/delayed-dispatch-after.json

@@ -0,0 +1 @@
+{"remote_delete_count_after_ttl": 1, "expired_but_registered": false, "cleanup_owner_present": false}

docs/rework/hel879/timing-fix/delayed-dispatch-before.json

@@ -0,0 +1 @@
+{"remote_delete_count_after_ttl": 0, "expired_but_registered": true, "cleanup_owner_present": false}

docs/rework/hel879/timing-fix/sdk-probe-after.json

@@ -0,0 +1 @@
+{"proactive_before_playout_finishes": {"handle_done": false, "events": ["proactive_start"]}, "proactive_after_playout_finishes": ["proactive_start", "proactive_end"], "proactive_payload_fields": {"v": 2, "kind": "proactive_start", "role": "assistant", "text": "offline sample", "speech_id": "speech_f28fdeb15e01", "message_id": 7}, "no_receipt_incomplete_actual_worker_events": ["turn_start", "turn_cancel"], "no_receipt_incomplete_confirmed": [], "complete_with_receipt_actual_worker_events": ["turn_start", "assistant_final", "turn_end"], "complete_with_receipt_confirmed": ["ok-turn"]}

docs/rework/hel879/timing-fix/sdk-probe-before.json

@@ -0,0 +1 @@
+{"proactive_before_playout_finishes": {"handle_done": false, "events": []}, "proactive_after_playout_finishes": ["proactive_start", "proactive_end"], "proactive_payload_fields": {"v": 2, "kind": "proactive_start", "role": "assistant", "text": "offline sample", "speech_id": "speech_2880d36ea013", "message_id": 7}, "no_receipt_incomplete_actual_worker_events": ["turn_start", "assistant_final", "turn_end"], "no_receipt_incomplete_confirmed": [], "complete_with_receipt_actual_worker_events": ["turn_start", "assistant_final", "turn_end"], "complete_with_receipt_confirmed": ["ok-turn"]}

docs/rework/hel879/timing-fix/summary.md

@@ -0,0 +1,37 @@
+# HEL-879 契约核准后 3 处实现时机收口（2026-10-27）
+
+核对对象：`hel-879-assembly @ 3228b1e0d7f822b7030f945e285eac74126771aa`（总管《HEL-879-契约核准与接续.md》）。
+本轮只修 3 处实现与已定契约 v2 不一致的问题：不改 `v` 信封字段、不改事件种类、不动 `web/src`、不重做已过绑定/备份/预算。
+
+## 修前复现（全部先复现失败，与总管证据一致）
+
+| # | 发现 | 修前证据 |
+|---|------|----------|
+| 1 | `open_session` 只在 `await create_dispatch` 前 kick；建房真实让出事件循环后 cleaner 空转退出，注册后无人再观察截止时间 | `delayed-dispatch-before.json`：`{"remote_delete_count_after_ttl": 0, "expired_but_registered": true, "cleanup_owner_present": false}`（与总管 delayed-dispatch.json 逐字段一致） |
+| 2 | `speak_proactive_row` 旧写法 `handle = await session.say(...)`；真实 SDK `say` 为同步函数、`await handle`＝等整段播完 → 播完才发 proactive_start | `sdk-probe-before.json`：`proactive_before_playout_finishes.events=[]`，播完后才 `["proactive_start","proactive_end"]` |
+| 3 | `_after_playout` 先按 `stream.text` 发 final/end 再查回执；注入 text 非空、completed=False、receipt=None 的确切归属流 → `turn_start→assistant_final→turn_end`，与 v2「无完整同轮回执只 cancel」相反 | 同上：`no_receipt_incomplete_actual_worker_events=["turn_start","assistant_final","turn_end"]`（实际 worker `bind_voice_session` 回调路径，真实 AgentSession/SpeechHandle） |
+
+新增专项测试在**未修复代码上先跑失败**（repro-first）：
+`tests/test_voice_rework_hel879.py::test_open_session_with_real_dispatch_yield_re_kicks_cleanup`、
+`::test_second_registration_wakes_a_sleeping_cleaner_for_earlier_deadline` → 修前 2 failed / 修后 2 passed。
+
+## 修复（精确小修）
+
+1. `src/cybergril/voice/session.py` `open_session`：注册成功并 `_persist_state()` 之后**重新 `_kick_cleanup()`**——建房 await 期间旧 cleaner 可能已因「无可观察对象」退出，新截止时间需要新的所有者；自动截止、共享上限、重启补偿、有界放弃全部保留。测试用**真正让出事件循环**的建房替身（`asyncio.sleep(0.02)`），不再只用立即返回的 MockTransport 替身。
+2. `src/cybergril/voice/worker.py` `speak_proactive_row`：先取 handle（`iscoroutine` 只解包 await 型替身——真实 SpeechHandle 自身可 await，不能误等）、**立即**发布 `proactive_start`、注册该 handle 的 `proactive_end/proactive_cancel`，再按需等待这一条播完（逐条顺序念保留）。
+3. `src/cybergril/voice/worker.py` `_after_playout`：`assistant_final` 前置必要门——流 `completed` 且 `should_ack`（合法同轮 receipt、非旧 epoch）才 `finish`+`confirm`；未完整/无回执/旧 epoch 一律只 `turn_cancel`（reason=incomplete/no-receipt），**绝不先发 final 再决定不确认**；φ/打断/失败/无归属 fail-closed 原样保留，完整+合法回执路径的 final 与 ack 均保留。
+
+## 修后验证
+
+- 本地：`delayed-dispatch-after.json`：`{"remote_delete_count_after_ttl": 1, "expired_but_registered": false, "cleanup_owner_present": false}`（cleaner 在全部结清后按设计自行退出，无悬空协程）。
+- `.12` 真 SDK（stdin 嵌入当前源码，零写入零网络，`/opt/voice-agent/agent/venv` livekit-agents 1.8.8）`sdk-probe-after.json`：
+  - `proactive_before_playout_finishes={"handle_done": false, "events": ["proactive_start"]}`（播放中即可见 start）；
+  - 播完 `["proactive_start","proactive_end"]`，字段 `v/kind/role/text/speech_id/message_id` 与 v2 完全一致；
+  - `no_receipt_incomplete=["turn_start","turn_cancel"]`、`confirmed=[]`；
+  - 完整+合法回执路径 `["turn_start","assistant_final","turn_end"]` + ack 保留。
+- 真SDK回归并入常驻检查 `scripts/gen_hel877_sdk_check.py`（**14/14 ALL PASS**，`../sdk-check-hel879-timing.log`）：新增「主动播报时机」「主动播报打断只有 cancel」「无回执不 final」三负例，均驱动实际 worker 模块注册的真实回调。
+- 产品套件：pytest **352/352**（3 次连跑）；ruff(src/tests/tools) **0 错**；`verify_hel877_voice_chain.py` **8/8**；需求检查 **4/4**（origin/main 与 c121c42 各带/不带 handoff-HEL-879.json）。字幕样例 `subtitle-samples/` 与 v2 契约文档无需再生成（字段与生命周期未变）。
+
+## 边界
+
+零真实 LLM/嵌入/ASR/TTS/麦克风；.12 仅 SSH stdin 执行只读检查（零写入零网络，stderr 仅既有 jwt 弱密钥告警）；零下载零安装；未部署、未合并；PR#11 与 `hel-870-assembly` 未动；权威账/冻结/共享服务/队员配置未碰；真实 cap 仍 0。R5 真机、完整 CG-011、R6/R7 继续承接，无需求变更（v1.11/60 项）。

scripts/gen_hel877_sdk_check.py

@@ -15,6 +15,15 @@ emitted before the task starts; (b) runtime — generations started inside that
 ContextVar bind exactly, including the manager's both-queued-then-delayed
 interleaving, a say speech sandwiched between user turns, an unattributed
 generation (fail-closed) and an interrupted turn (cancel, no receipt).
+
+HEL-879 timing closure (contract-approval round): the real SDK's
+AgentSession.say is a SYNC method returning an awaitable SpeechHandle (awaiting
+it awaits the whole playout) — speak_proactive_row must publish
+proactive_start DURING the playout with end/cancel on the handle's done
+callback; and a clearly-attributed but incomplete/no-receipt stream must close
+the subtitle turn with turn_cancel ONLY (contract v2: no complete same-turn
+receipt means no assistant_final), while the complete+receipt path keeps
+final AND the ack.
 """
 from __future__ import annotations
 
@@ -372,6 +381,95 @@ asyncio.run(interim_seam())
 ok.append("assistant_interim seam: real LLMStream deltas feed per-speech_id v2 "
           "payloads (replace semantics); unattributed streams feed nothing")
 
+# 11. HEL-879 timing closure: proactive_start DURING the playout on the real
+#     sync-say shape (a bare handle.__await__ would wait out the whole play).
+async def proactive_timing_case():
+    events: list[dict] = []
+    h = SpeechHandle.create()
+    class SyncSay:
+        def say(self, *args, **kwargs):  # real SDK shape: sync -> handle
+            asyncio.get_running_loop().call_later(.06, h._mark_done)
+            return h
+    task = asyncio.create_task(worker_mod.speak_proactive_row(
+        SyncSay(), seam_mod.SubtitleBus(events.append),
+        {"id": 7, "text": "offline sample"}))
+    await asyncio.sleep(.015)
+    assert h.done() is False
+    assert [e["kind"] for e in events] == ["proactive_start"], events
+    await task
+    await asyncio.sleep(.015)
+    assert [e["kind"] for e in events] == ["proactive_start", "proactive_end"]
+    start = events[0]
+    assert (start["v"], start["role"], start["message_id"], start["text"]) == \
+        (2, "assistant", 7, "offline sample")
+    assert start["speech_id"] == h.id
+
+asyncio.run(proactive_timing_case())
+ok.append("speak_proactive_row publishes proactive_start while the real "
+          "SpeechHandle is still playing (sync-say shape), end/cancel on the "
+          "handle's done callback")
+
+# 11b. an interrupted proactive row closes with cancel, never end
+async def proactive_interrupt_case():
+    events: list[dict] = []
+    h = SpeechHandle.create()
+    class SyncSay:
+        def say(self, *args, **kwargs):
+            return h
+    task = asyncio.create_task(worker_mod.speak_proactive_row(
+        SyncSay(), seam_mod.SubtitleBus(events.append),
+        {"id": 8, "text": "被打断的主动留言"}))
+    await asyncio.sleep(.005)  # start emitted, done callback registered
+    h.interrupt()
+    h._mark_done()
+    await task
+    await asyncio.sleep(.02)
+    assert [e["kind"] for e in events] == ["proactive_start", "proactive_cancel"], events
+    assert events[-1]["reason"] == "interrupted"
+
+asyncio.run(proactive_interrupt_case())
+ok.append("interrupted proactive row emits proactive_cancel(reason=interrupted), "
+          "never proactive_end")
+
+# 12. HEL-879 timing closure: a clearly-attributed but INCOMPLETE/no-receipt
+#     stream through the ACTUAL bind_voice_session callbacks must close the
+#     subtitle turn with turn_cancel ONLY — no assistant_final, no ack (v2).
+async def no_receipt_incomplete_case():
+    events: list[dict] = []
+    session = AgentSession()
+    ledger = receipt_mod.TurnLedger()
+    bus = seam_mod.SubtitleBus(events.append)
+    interim = seam_mod.AssistantInterimPublisher(bus)
+    confirmed: list[str] = []
+    async def epoch():
+        return 1
+    class Receipts:
+        async def confirm(self, *args):
+            confirmed.append(args[0])
+            return True
+    llm = types.SimpleNamespace(client=types.SimpleNamespace(current_epoch=epoch))
+    worker_mod.bind_voice_session(session, llm=llm, ledger=ledger, interim=interim,
+                                  bus=bus, stt=types.SimpleNamespace(),
+                                  receipts=Receipts())
+    h = SpeechHandle.create()
+    session.emit("speech_created", SpeechCreatedEvent(
+        user_initiated=True, source="generate_reply", speech_handle=h,
+        created_at=1.0))
+    stream = types.SimpleNamespace(
+        speech_id=h.id, error_phi=None, text="partial without receipt",
+        turn_id="incomplete", receipt=None, completed=False)
+    ledger.note_stream(stream, h)
+    interim.begin(h.id)
+    h._mark_done()
+    await asyncio.sleep(.06)
+    assert [e["kind"] for e in events] == ["turn_start", "turn_cancel"], events
+    assert events[-1]["reason"] == "incomplete"
+    assert confirmed == []
+
+asyncio.run(no_receipt_incomplete_case())
+ok.append("incomplete/no-receipt attributed stream: actual worker callbacks emit "
+          "turn_start -> turn_cancel only, no assistant_final, no ack")
+
 print("HEL-877/879 real-SDK check on livekit-agents", end=" ")
 import importlib.metadata as md
 print(md.version("livekit-agents") + ":", "ALL PASS")

scripts/gen_hel879_timing_probe.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+"""Generate the HEL-879 timing-fix probe for the .12 agent venv (real SDK).
+
+Reproduces the manager's two worker-side findings from the contract-approval
+round (sdk-say / sdk-extra payloads), against WHATEVER sources are in the tree
+when run — same script is used before AND after the fix:
+
+  1. speak_proactive_row on the REAL SDK shape: AgentSession.say is a SYNC
+     method returning an awaitable SpeechHandle (awaiting it awaits the whole
+     playout), so proactive_start must be published while the handle is still
+     playing (before done), with end/cancel on the handle's own done callback.
+  2. _after_playout through the ACTUAL bind_voice_session callbacks on a real
+     AgentSession: a clearly-attributed stream with text but completed=False
+     and receipt=None must NOT produce assistant_final — contract v2 says no
+     complete same-turn receipt means cancel, never final. The complete+receipt
+     path must still produce final AND the ack (checked in the same probe).
+
+Embeds repo voice-module sources in a self-contained stdin script; no files
+written on .12, no network, no service touched.
+"""
+from __future__ import annotations
+
+import base64
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULES = ["chat_api_client", "receipt", "seam", "llm_chat_api",
+           "stt_funasr", "tts_cosyvoice", "worker"]
+
+TEMPLATE = r'''
+import asyncio, base64, json, sys, types
+
+SOURCES = json.loads(base64.b64decode("__B64__"))
+
+pkg = types.ModuleType("cybergril"); pkg.__path__ = []
+voice = types.ModuleType("cybergril.voice"); voice.__path__ = []
+sys.modules["cybergril"] = pkg
+sys.modules["cybergril.voice"] = voice
+for name, src in SOURCES.items():
+    mod = types.ModuleType(f"cybergril.voice.{name}")
+    mod.__package__ = "cybergril.voice"
+    sys.modules[f"cybergril.voice.{name}"] = mod
+    exec(compile(src, f"<cybergril.voice.{name}>", "exec"), mod.__dict__)
+    setattr(voice, name, mod)
+
+from livekit.agents import AgentSession
+from livekit.agents.voice.speech_handle import SpeechHandle
+from livekit.agents.voice.events import SpeechCreatedEvent
+
+w = voice.worker
+out = {}
+
+# --- 1. proactive say timing on the REAL SpeechHandle shape -------------------
+async def proactive_case():
+    events = []
+    h = SpeechHandle.create()
+    class Say:
+        def say(self, *args, **kwargs):  # real SDK: sync, returns SpeechHandle
+            asyncio.get_running_loop().call_later(.06, h._mark_done)
+            return h
+    task = asyncio.create_task(w.speak_proactive_row(
+        Say(), voice.seam.SubtitleBus(events.append), {"id": 7, "text": "offline sample"}))
+    await asyncio.sleep(.015)
+    out["proactive_before_playout_finishes"] = {
+        "handle_done": h.done(),
+        "events": [x["kind"] for x in events]}
+    await task
+    await asyncio.sleep(.015)
+    out["proactive_after_playout_finishes"] = [x["kind"] for x in events]
+    out["proactive_payload_fields"] = {k: events[0][k] for k in ("v", "kind", "role", "text", "speech_id", "message_id")}
+
+asyncio.run(proactive_case())
+
+# --- 2. actual-worker no-receipt/incomplete stream ---------------------------
+async def no_receipt_case():
+    events = []
+    session = AgentSession()
+    ledger = voice.receipt.TurnLedger()
+    bus = voice.seam.SubtitleBus(events.append)
+    interim = voice.seam.AssistantInterimPublisher(bus)
+    async def epoch(): return 1
+    confirmed = []
+    class Receipts:
+        async def confirm(self, *args):
+            confirmed.append(args[0]); return True
+    llm = types.SimpleNamespace(client=types.SimpleNamespace(current_epoch=epoch))
+    w.bind_voice_session(session, llm=llm, ledger=ledger, interim=interim,
+                         bus=bus, stt=types.SimpleNamespace(), receipts=Receipts())
+    h = SpeechHandle.create()
+    session.emit("speech_created", SpeechCreatedEvent(
+        user_initiated=True, source="generate_reply", speech_handle=h, created_at=1.0))
+    stream = types.SimpleNamespace(
+        speech_id=h.id, error_phi=None, text="partial without receipt",
+        turn_id="incomplete", receipt=None, completed=False)
+    ledger.note_stream(stream, h)
+    interim.begin(h.id)
+    h._mark_done()
+    await asyncio.sleep(.06)
+    out["no_receipt_incomplete_actual_worker_events"] = [x["kind"] for x in events]
+    out["no_receipt_incomplete_confirmed"] = list(confirmed)
+
+asyncio.run(no_receipt_case())
+
+# --- 3. the complete+receipt path must keep final AND ack --------------------
+async def complete_case():
+    events = []
+    session = AgentSession()
+    ledger = voice.receipt.TurnLedger()
+    bus = voice.seam.SubtitleBus(events.append)
+    interim = voice.seam.AssistantInterimPublisher(bus)
+    async def epoch(): return 1
+    confirmed = []
+    class Receipts:
+        async def confirm(self, *args):
+            confirmed.append(args[0]); return True
+    llm = types.SimpleNamespace(client=types.SimpleNamespace(current_epoch=epoch))
+    w.bind_voice_session(session, llm=llm, ledger=ledger, interim=interim,
+                         bus=bus, stt=types.SimpleNamespace(), receipts=Receipts())
+    h = SpeechHandle.create()
+    session.emit("speech_created", SpeechCreatedEvent(
+        user_initiated=True, source="generate_reply", speech_handle=h, created_at=1.0))
+    receipt = {"message_id": 9, "turn_id": "ok-turn", "epoch": 1,
+               "token": "t" * 48, "text_hash": "z" * 64}
+    stream = types.SimpleNamespace(
+        speech_id=h.id, error_phi=None, text="her complete reply",
+        turn_id="ok-turn", receipt=receipt, completed=True)
+    ledger.note_stream(stream, h)
+    interim.begin(h.id)
+    h._mark_done()
+    await asyncio.sleep(.06)
+    out["complete_with_receipt_actual_worker_events"] = [x["kind"] for x in events]
+    out["complete_with_receipt_confirmed"] = list(confirmed)
+
+asyncio.run(complete_case())
+
+print(json.dumps(out, ensure_ascii=False))
+'''
+
+def main() -> int:
+    sources = {}
+    for name in MODULES:
+        path = ROOT / "src/cybergril/voice" / f"{name}.py"
+        sources[name] = path.read_text(encoding="utf-8")
+    payload = base64.b64encode(json.dumps(sources).encode()).decode()
+    script = TEMPLATE.replace("__B64__", payload)
+    out = ROOT / "scripts/_hel879_timing_probe_payload.py"
+    out.write_text(script, encoding="utf-8")
+    print(out)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/cybergril/voice/session.py

@@ -414,6 +414,11 @@ class VoiceSessionService:
         )
         self._sessions[session.session_id] = session
         self._persist_state()
+        # Re-kick AFTER the registration exists: a real dispatch await yields
+        # the loop, and the earlier kick's cleaner may have already found
+        # nothing to watch and exited — the new deadline needs an owner again
+        # (HEL-879 timing fix: the cleaner died mid-dispatch with zero deletes).
+        self._kick_cleanup()
         return session
 
     def token_for(self, session: VoiceSession) -> str:

src/cybergril/voice/worker.py

@@ -31,6 +31,8 @@ scripts/gen_hel877_sdk_check.py 不再手抄复刻）。
 from __future__ import annotations
 
 import asyncio
+import contextlib
+import inspect
 import json
 import os
 from pathlib import Path
@@ -148,13 +150,19 @@ def bind_voice_session(session: AgentSession, *, llm: CyberGrillLLM,
             interim.cancel(speech_id, "error_phi")
             outcomes[speech_id] = "error_phi"
             return
-        if stream.text:
-            interim.finish(speech_id, turn_id=stream.turn_id, text=stream.text)
         epoch = None
         try:
             epoch = await llm.client.current_epoch()
         except Exception:
             epoch = stream.receipt.get("epoch") if stream.receipt else None
+        # Contract v2 (HEL-879 timing fix): assistant_final means the archive
+        # really holds this turn — an incomplete stream, or one without a
+        # legal same-turn receipt (missing fields / stale epoch), closes with
+        # turn_cancel ONLY. The final is never sent first and retracted after.
+        if not getattr(stream, "completed", False):
+            interim.cancel(speech_id, "incomplete")
+            outcomes[speech_id] = "incomplete"
+            return
         if should_ack(
             played_to_end=True,
             interrupted=handle.interrupted,
@@ -162,9 +170,11 @@ def bind_voice_session(session: AgentSession, *, llm: CyberGrillLLM,
             receipt=stream.receipt,
             current_epoch=epoch,
         ):
+            interim.finish(speech_id, turn_id=stream.turn_id, text=stream.text)
             await receipts.confirm(stream.turn_id, stream.receipt)
             outcomes[speech_id] = "acked"
         else:
+            interim.cancel(speech_id, "no-receipt")
             outcomes[speech_id] = "no-receipt"
 
     @session.on("speech_created")
@@ -199,8 +209,17 @@ async def speak_proactive_row(session: AgentSession, bus: SubtitleBus, row: dict
 
     字幕生命周期绑定在 say 返回的 SpeechHandle 上（proactive_start ->
     proactive_end/proactive_cancel），不借用任何回执分支。
+
+    真实 SDK 的 AgentSession.say 是同步函数、返回可 await 的 SpeechHandle
+    （await 它＝等整段播完，HEL-879 timing 修正）：先取 handle、立刻发布
+    proactive_start 并注册该 handle 的 end/cancel，再等这一条播完——start
+    绝不能等声音播完后才发。await 型替身（测试桩）同样支持。
     """
-    handle = await session.say(row["text"], allow_interruptions=True)
+    result = session.say(row["text"], allow_interruptions=True)
+    # iscoroutine (NOT isawaitable): the real SpeechHandle itself defines
+    # __await__ (awaiting it = waiting out the playout) — only a coroutine
+    # from an await-style stand-in needs resolving to reach the handle.
+    handle = await result if inspect.iscoroutine(result) else result
     speech_id = getattr(handle, "id", "")
     bus.emit("proactive_start", role="assistant", text=row["text"],
              speech_id=speech_id, message_id=row["id"])
@@ -214,6 +233,11 @@ async def speak_proactive_row(session: AgentSession, bus: SubtitleBus, row: dict
             bus.emit("proactive_end", role="assistant", speech_id=sid, message_id=mid)
 
     handle.add_done_callback(_on_say_done)
+    # 逐条念：等 THIS 行播完再返回（替身 handle 不可 await 时直接返回；
+    # 播放中的异常已由 done 回调按 cancel 收口，这里不再向上抛）。
+    if inspect.isawaitable(handle):
+        with contextlib.suppress(Exception):
+            await handle
 
 
 async def _maybe_open_proactive(session: AgentSession, room, bus: SubtitleBus,

tests/test_voice_rework_hel879.py

@@ -432,6 +432,77 @@ async def test_cleaner_stays_alive_while_a_session_is_registered():
         await service.aclose()
 
 
+@pytest.mark.asyncio
+async def test_open_session_with_real_dispatch_yield_re_kicks_cleanup(monkeypatch):
+    """总管契约核准轮第 1 项精确复现：建房 await 真正让出事件循环（真实 HTTP
+    一定如此），pre-dispatch kick 的清理所有者在等待期间已空转退出；注册成功
+    并持久化之后必须重新唤醒/重建清理所有者——过 TTL 且不再来任何请求也能
+    到期转债并真实删房。旧代码：DELETE=0、过期登记仍在、cleaner=None。"""
+    deletes: list[str] = []
+    dispatch_yielded = asyncio.Event()
+
+    async def slow_create(*args, **kwargs):
+        await asyncio.sleep(0.02)  # real HTTP always gives the loop a chance
+        dispatch_yielded.set()
+        return SimpleNamespace(dispatch_id="AD_delayed")
+
+    async def slow_delete(*args, **kwargs):
+        deletes.append(kwargs["room"])
+
+    monkeypatch.setattr(livekit_wire, "create_dispatch", slow_create)
+    monkeypatch.setattr(livekit_wire, "delete_room", slow_delete)
+    service = _service(StandinLiveKit(), session_ttl_seconds=0.025)
+    try:
+        session = await service.open_session(epoch=1)
+        assert dispatch_yielded.is_set(), "dispatch await must really yield here"
+        # the registration's deadline has a live owner again right after open
+        assert service._cleaner is not None and not service._cleaner.done()
+        await asyncio.sleep(0.12)  # past TTL with NO further request at all
+        assert deletes == [session.room]  # exactly one delete, no duplicates
+        assert session.session_id not in service._sessions
+        assert service.cleanup_debt(session.session_id) is None
+        # settled: the cleaner exits by itself, no dangling coroutine
+        for _ in range(100):
+            if service._cleaner is None or service._cleaner.done():
+                break
+            await asyncio.sleep(0.02)
+        assert service._cleaner is None or service._cleaner.done()
+    finally:
+        await service.aclose()
+        assert service._cleaner is None
+
+
+@pytest.mark.asyncio
+async def test_second_registration_wakes_a_sleeping_cleaner_for_earlier_deadline(
+        monkeypatch):
+    """注册后的 kick 也唤醒仍在睡眠的清理所有者：更短 TTL 的新会话不必等
+    旧会话的截止时间才被观察（min(deadlines) 立即重算）。"""
+    deletes: list[str] = []
+
+    async def create(*args, **kwargs):
+        await asyncio.sleep(0.02)
+        return SimpleNamespace(dispatch_id="AD_2")
+
+    async def delete(*args, **kwargs):
+        deletes.append(kwargs["room"])
+
+    monkeypatch.setattr(livekit_wire, "create_dispatch", create)
+    monkeypatch.setattr(livekit_wire, "delete_room", delete)
+    service = _service(StandinLiveKit(), session_ttl_seconds=5.0, backoff=0.005)
+    try:
+        first = await service.open_session(epoch=1)  # long TTL keeps cleaner sleeping
+        service.config.session_ttl_seconds = 0.025
+        second = await service.open_session(epoch=1)  # short TTL + real dispatch await
+        for _ in range(100):
+            if deletes:
+                break
+            await asyncio.sleep(0.02)
+        assert deletes == [second.room]
+        assert first.session_id in service._sessions  # the long-TTL call is untouched
+    finally:
+        await service.aclose()
+
+
 @pytest.mark.asyncio
 async def test_end_after_abandoned_cap_sends_no_further_deletes():
     """总管复核 A-2 精确复现：上限 2，后台到顶后再 3 次 end 不得再发删除