HEL-906 rework: persistence fail-closed stop + conservative pre-record accounting

- asr_usage: ledger write/read failure, corruption or mid-window vanish now
  stop all window outbound (fail-closed), preserve known consumption
  (monotonic, no zero-budget default), log + record guard_stop (durable
  across same-window restarts)
- asr_usage: conservative reserve before send (failed/cancelled sends are
  NOT refunded) + separately tracked confirmed-delivered seconds; replay
  restores both
- stt_funasr: reserve_audio before ws.send, note_audio_delivered after
  successful await only
- tests: 24 deterministic offline cases incl. manager's budget-regression
  repro, read/corrupt/missing rejection, record-then-raise/cancel fake ws,
  guard_stop across restart

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

src/cybergril/voice/asr_usage.py

@@ -6,12 +6,18 @@
 worker 的真实送数路径上：
 
 - FunASRSTT（voice/stt_funasr.py）在每次 websocket 建连前 check_connect、
-  建连结果 note_connect_result、每个真实发送帧发送前 precheck_audio、
-  发送成功后 note_audio_sent；worker（voice/worker.py）按 job 注入 guard
-  并把 on_exhausted 接到 ctx.shutdown。
+  建连结果 note_connect_result、每个真实发送帧发送前 reserve_audio
+  （保守预记＋越限拒绝）、``await ws.send`` 成功返回后 note_audio_delivered
+  （确证送达量）；worker（voice/worker.py）按 job 注入 guard 并把
+  on_exhausted 接到 ctx.shutdown。
 - 音频按**实际发送帧**的 sample_rate 与样本数计时（重采样后的输出帧按其
   自身采样率计），同一帧重复发送按重复计；墙钟、开房/删房流水不参与
   音频计时，事后成功流水也不追溯改数。
+- **保守计费 / 确证送达双口径**（HEL-906 复核返工）：音频在发送**前**
+  按整帧预记入预算（conservative），发送失败或被取消**不退还**——帧可能
+  已经出站；只有 ``await ws.send`` 成功返回的帧才累计确证送达量
+  （confirmed）。保守量是硬闸依据，确证量只是更低的下界核对；不得把
+  预记量宣称全部确证送达。
 - 连接事实逐条落账：每次建连尝试（含失败）、失败后的重试（retry）、
   接收侧死亡（connection_died）。FunASR 一句一连的惰性建连里，正常说完
   一轮后的下一次建连不是 retry，只有上一次尝试失败/连接死亡/流异常结束
@@ -20,9 +26,10 @@ worker 的真实送数路径上：
 账本（追加写 JSONL，默认 /opt/voice-agent-cybergril/asr-usage.jsonl）独立
 于实例侧会话账本 ``<db>-voice-sessions.json``：删房成功会把通话从会话账
 本移除，但这里的逐通历史（call_start/audio/connect_attempt/
-connection_died/call_end/block）永不改写、不随删房消失；窗口总量由重放
-账本恢复，同 window_id 重启进程不能重置预算，旧窗口耗尽状态也被持久化
-（不得把旧余额或新开账当作新许可）。
+connection_died/call_end/block/guard_stop）永不改写、不随删房消失；窗口
+总量由重放账本恢复，同 window_id 重启进程不能重置预算，旧窗口耗尽状态
+与持久化停机状态（guard_stop）也被持久化（不得把旧余额或新开账当作
+新许可）。
 
 窗口语义沿用 think/budget.py 的 fail-closed 先例：
 
@@ -32,16 +39,24 @@ connection_died/call_end/block）永不改写、不随删房消失；窗口总
   在发送前拒绝（AsrWindowClosed），worker 经 on_exhausted 结束本通 job；
   上限只能收紧不能放宽，候选值见 docs/rework/hel902/r5-followup-package.md §4。
 
+**持久化 fail-closed（HEL-906 复核返工）**：账本写失败、读失败、行损坏
+或账本中途消失时，guard 停止本窗口一切出站（新连接/新音频一律拒绝）、
+保全内存已知消耗（窗口计数只增不减，禁止默认零预算或按旧账回退）、
+记 error 日志与停机原因，并把 ``guard_stop`` 落账——同窗口重启进程重放
+到该记录仍保持停机。故障下不继续通话（不要求支持）。
+
 程序能保证 / 剩余只能监看（如实边界）：
 
-- 保证：布防进程内逐帧发送前硬停（帧不越上限一个字节）、逐通与全窗
-  计数不因删房/重启丢失、同窗重启不能重置预算。
+- 保证：布防进程内逐帧发送前硬停（帧不越上限一个字节；保守预记含
+  发送失败/取消的不确定出站）、逐通与全窗计数不因删房/重启丢失、
+  同窗重启不能重置预算、持久化故障即停不回退。
 - 跨进程有界收敛：账本是共同事实，每次建连检查与 resync_seconds 间隔
   重放账本；两进程并行通话时对端最近 ≤ max(resync, flush) 间隔的音频
-  可能未计入本轮判定（有界过冲）。
-- 只能监看：进程崩溃前 ≤ flush_seconds 的已发音频可能少计；接收侧死亡
-  未伴随发送失败时按 connection_died 计连接事实、不改已计音频；真实
-  FunASR/LiveKit 行为仍以窗口前 .12 实测为准。
+  可能未计入本轮判定（有界过冲）。**不承诺全窗口跨进程绝不越限**。
+- 只能监看：进程崩溃前 ≤ flush_seconds 的已发音频可能少计（确证量同样
+  只在 flush 后落账）；接收侧死亡未伴随发送失败时按 connection_died
+  计连接事实、不改已计音频；真实 FunASR/LiveKit 行为仍以窗口前 .12
+  实测为准。
 """
 
 from __future__ import annotations
@@ -49,6 +64,7 @@ from __future__ import annotations
 import argparse
 import contextlib
 import json
+import logging
 import os
 import secrets
 import time
@@ -59,6 +75,8 @@ DEFAULT_ASR_USAGE_LEDGER = "/opt/voice-agent-cybergril/asr-usage.jsonl"
 DEFAULT_FLUSH_SECONDS = 5.0
 DEFAULT_RESYNC_SECONDS = 5.0
 
+_logger = logging.getLogger("cybergril.voice.asr_usage")
+
 # 允许浮点求和后恰好等于上限（“≤上限”而不是“<上限”）
 _EPSILON = 1e-9
 
@@ -117,7 +135,8 @@ def policy_from_env(environ: dict | None = None) -> AsrWindowPolicy:
 
 @dataclass
 class _AudioBuf:
-    seconds: float = 0.0
+    seconds: float = 0.0  # 保守预记（发送前计入，失败/取消不退还）
+    confirmed_seconds: float = 0.0  # 确证送达（await ws.send 成功返回）
     frames: int = 0
     bytes: int = 0
     sample_rate: int = 0
@@ -129,7 +148,8 @@ class CallUsage:
     connect_attempts: int = 0
     connect_failures: int = 0
     connect_retries: int = 0
-    audio_seconds: float = 0.0
+    audio_seconds: float = 0.0  # 保守计费口径（含不确定出站）
+    confirmed_audio_seconds: float = 0.0  # 确证送达口径（≤ audio_seconds）
     retry_pending: bool = False  # 上一次连接事实是失败 → 下次建连是 retry
     _would_retry: bool = False  # check_connect 为本次在途尝试标记的 retry 身份
 
@@ -137,21 +157,17 @@ class CallUsage:
 @dataclass
 class _ReplayState:
     window_audio_seconds: float = 0.0
+    window_confirmed_seconds: float = 0.0
+    stopped: bool = False
+    stop_reason: str = ""
     calls: dict[str, CallUsage] = field(default_factory=dict)
 
 
-def summarize_ledger(path: Path | str, window_id: str | None = None) -> dict:
-    """重放账本成汇总（操作员核对口径；不修改账本）.
-
-    window_id 为 None 时按账本里出现过的窗口分组汇总。
-    """
+def _fold_lines(lines: list[str], window_id: str | None) -> tuple[dict[str, _ReplayState], int, int]:
+    """把 JSONL 行折叠为 {window_id: _ReplayState}；返回 (windows, records, parse_errors)."""
     windows: dict[str, _ReplayState] = {}
     records = 0
     parse_errors = 0
-    try:
-        lines = Path(path).read_text(encoding="utf-8").splitlines()
-    except OSError:
-        lines = []
     for line in lines:
         line = line.strip()
         if not line:
@@ -183,16 +199,42 @@ def summarize_ledger(path: Path | str, window_id: str | None = None) -> dict:
             cu.retry_pending = True
         elif kind == "audio":
             seconds = float(rec.get("seconds", 0.0) or 0.0)
+            # 旧记录（返工前）只在发送成功后计 seconds，即全部确证送达
+            confirmed = rec.get("confirmed_seconds")
+            confirmed_s = seconds if confirmed is None else float(confirmed or 0.0)
             cu.audio_seconds = round(cu.audio_seconds + seconds, 6)
+            cu.confirmed_audio_seconds = round(cu.confirmed_audio_seconds + confirmed_s, 6)
             state.window_audio_seconds = round(state.window_audio_seconds + seconds, 6)
+            state.window_confirmed_seconds = round(
+                state.window_confirmed_seconds + confirmed_s, 6)
         elif kind == "call_end" and str(rec.get("reason")) == "error":
             cu.retry_pending = True
+        elif kind == "guard_stop" and not state.stopped:
+            state.stopped = True
+            state.stop_reason = str(rec.get("reason") or "ledger-stopped")
+    return windows, records, parse_errors
+
+
+def summarize_ledger(path: Path | str, window_id: str | None = None) -> dict:
+    """重放账本成汇总（操作员核对口径；不修改账本）.
+
+    window_id 为 None 时按账本里出现过的窗口分组汇总。宽容模式：读失败
+    按空、坏行只计数——只用于人工核对；guard 内部用严格读（见 _replay）。
+    """
+    try:
+        lines = Path(path).read_text(encoding="utf-8").splitlines()
+    except OSError:
+        lines = []
+    windows, records, parse_errors = _fold_lines(lines, window_id)
     return {
         "records": records,
         "parse_errors": parse_errors,
         "windows": {
             wid: {
                 "audio_seconds": round(st.window_audio_seconds, 6),
+                "confirmed_audio_seconds": round(st.window_confirmed_seconds, 6),
+                "stopped": st.stopped,
+                "stop_reason": st.stop_reason,
                 "calls": {
                     call: {
                         "streams": cu.streams,
@@ -200,6 +242,7 @@ def summarize_ledger(path: Path | str, window_id: str | None = None) -> dict:
                         "connect_failures": cu.connect_failures,
                         "connect_retries": cu.connect_retries,
                         "audio_seconds": round(cu.audio_seconds, 6),
+                        "confirmed_audio_seconds": round(cu.confirmed_audio_seconds, 6),
                         "retry_pending": cu.retry_pending,
                     }
                     for call, cu in st.calls.items()
@@ -229,31 +272,70 @@ class AsrUsageGuard:
         self._unlabeled = 0
         self._pending: dict[str, _AudioBuf] = {}
         self.calls: dict[str, CallUsage] = {}
-        self.window_audio_seconds = 0.0
+        self.window_audio_seconds = 0.0  # 保守计费（发送前预记，含不确定出站）
+        self.window_confirmed_seconds = 0.0  # 确证送达（await 成功返回）
         self.ledger_errors = 0
+        self.persistence_stop_reason: str | None = None  # 非 None = 本窗出站已停
+        self._ledger_seen = False  # 见过账本存在（读成功或写成功过）
         self._last_resync = -1e18
         self._replay()
 
     # -- 账本 ---------------------------------------------------------------
     def _replay(self) -> None:
+        """严格重放：读失败/损坏/账本中途消失 → fail-closed 停机，不套用数据."""
         self._last_resync = self._clock()
-        if self.ledger_path is None or not self.ledger_path.is_file():
+        if self.ledger_path is None:
             return
-        state = summarize_ledger(self.ledger_path, self.policy.window_id)["windows"].get(
-            self.policy.window_id
-        )
+        try:
+            exists = self.ledger_path.is_file()
+        except OSError as exc:
+            _logger.error("asr usage ledger stat failed: %s: %s", self.ledger_path, exc)
+            self._mark_persistence_stop("ledger-read-failed",
+                                        detail=f"stat: {type(exc).__name__}", fire=False)
+            return
+        if not exists:
+            if self._ledger_seen:
+                # 读过/写过的账本中途消失 = 已落账事实丢失，不许默认零预算
+                _logger.error("asr usage ledger vanished mid-window: %s", self.ledger_path)
+                self._mark_persistence_stop("ledger-missing", fire=False)
+            return
+        try:
+            lines = self.ledger_path.read_text(encoding="utf-8").splitlines()
+        except OSError as exc:
+            _logger.error("asr usage ledger read failed: %s: %s", self.ledger_path, exc)
+            self._mark_persistence_stop("ledger-read-failed",
+                                        detail=type(exc).__name__, fire=False)
+            return
+        self._ledger_seen = True
+        windows, _records, parse_errors = _fold_lines(lines, self.policy.window_id)
+        if parse_errors:
+            # 损坏账本不可信：不套用任何重放数据，内存已知消耗保全原值
+            _logger.error("asr usage ledger corrupt (%d bad lines): %s",
+                          parse_errors, self.ledger_path)
+            self._mark_persistence_stop(
+                "ledger-corrupt", detail=f"{parse_errors} bad lines", fire=False)
+            return
+        state = windows.get(self.policy.window_id)
         if state is None:
             return
-        # 只重放计数事实；重放结果不覆盖本进程在途未落账的音频缓冲
-        self.window_audio_seconds = float(state["audio_seconds"])
-        for call, cu in state["calls"].items():
+        # 保全已知消耗：同窗计数只增不减（重放/内存取大，绝不倒退回旧账）
+        self.window_audio_seconds = max(
+            self.window_audio_seconds, state.window_audio_seconds)
+        self.window_confirmed_seconds = max(
+            self.window_confirmed_seconds, state.window_confirmed_seconds)
+        for call, cu in state.calls.items():
             local = self.calls.setdefault(call, CallUsage())
-            local.streams = max(local.streams, cu["streams"])
-            local.connect_attempts = max(local.connect_attempts, cu["connect_attempts"])
-            local.connect_failures = max(local.connect_failures, cu["connect_failures"])
-            local.connect_retries = max(local.connect_retries, cu["connect_retries"])
-            local.audio_seconds = max(local.audio_seconds, cu["audio_seconds"])
-            local.retry_pending = bool(cu.get("retry_pending", False))
+            local.streams = max(local.streams, cu.streams)
+            local.connect_attempts = max(local.connect_attempts, cu.connect_attempts)
+            local.connect_failures = max(local.connect_failures, cu.connect_failures)
+            local.connect_retries = max(local.connect_retries, cu.connect_retries)
+            local.audio_seconds = max(local.audio_seconds, cu.audio_seconds)
+            local.confirmed_audio_seconds = max(
+                local.confirmed_audio_seconds, cu.confirmed_audio_seconds)
+            local.retry_pending = bool(local.retry_pending or cu.retry_pending)
+        if state.stopped:
+            # 其他/先前进程记录的持久化停机：同窗重启不恢复出站
+            self._mark_persistence_stop(state.stop_reason or "ledger-stopped", fire=False)
 
     def _maybe_resync(self) -> None:
         if self.policy.resync_seconds < 0:
@@ -274,9 +356,33 @@ class AsrUsageGuard:
             self.ledger_path.parent.mkdir(parents=True, exist_ok=True)
             with self.ledger_path.open("a", encoding="utf-8") as fh:
                 fh.write(line + "\n")
-        except OSError:
-            # 账本写失败不破坏通话，只降级可核算性；进程内闸仍按内存计数执行
+            self._ledger_seen = True
+        except OSError as exc:
+            # HEL-906 复核：写失败不再只降级可核算性——账本不可信即停本窗出站
             self.ledger_errors += 1
+            _logger.error("asr usage ledger append failed: %s: %s", self.ledger_path, exc)
+            self._mark_persistence_stop(
+                "ledger-write-failed", detail=f"{type(exc).__name__}: {exc}", fire=True)
+
+    def _mark_persistence_stop(self, reason: str, *, detail: str = "", fire: bool) -> None:
+        """持久化故障停机：保全已知消耗、留日志与停机原因、落账 guard_stop."""
+        if self.persistence_stop_reason is not None:
+            return  # 保留首个停机原因，不重复触发
+        self.persistence_stop_reason = reason
+        _logger.error("asr usage ledger unusable — stopping window outbound: "
+                      "reason=%s detail=%s ledger=%s", reason, detail, self.ledger_path)
+        self._append({"kind": "guard_stop", "reason": reason, "detail": (detail or "")[:200]})
+        if fire and not self._exhausted_fired:
+            self._exhausted_fired = True
+            if self._on_exhausted is not None:
+                with contextlib.suppress(Exception):
+                    self._on_exhausted(reason)
+
+    def _gate_persistence_stop(self, call: str, what: str) -> None:
+        if self.persistence_stop_reason is None:
+            return
+        self._block(call, what, self.persistence_stop_reason,
+                    detail="usage ledger unusable; window outbound stopped")
 
     def _block(self, call: str, what: str, reason: str, detail: str = "") -> None:
         self._append({"kind": "block", "call": call, "what": what, "reason": reason,
@@ -300,6 +406,7 @@ class AsrUsageGuard:
         self._maybe_resync()
         if not self.policy.armed:
             self._block(call, "new_stream", "window-not-armed")
+        self._gate_persistence_stop(call, "new_stream")
         if self.window_audio_seconds + self._pending_total() >= \
                 self.policy.max_audio_seconds - _EPSILON:
             self._block(call, "new_stream", "window-audio-exhausted")
@@ -315,6 +422,7 @@ class AsrUsageGuard:
         would_retry = cu.retry_pending
         if not self.policy.armed:
             self._block(call, "new_connection", "window-not-armed")
+        self._gate_persistence_stop(call, "new_connection")
         if self.window_audio_seconds + self._pending_total() >= \
                 self.policy.max_audio_seconds - _EPSILON:
             self._block(call, "new_connection", "window-audio-exhausted")
@@ -351,6 +459,7 @@ class AsrUsageGuard:
         seconds = samples / sample_rate
         if not self.policy.armed:
             self._block(call, "audio", "window-not-armed")
+        self._gate_persistence_stop(call, "audio")
         pending = self._pending.get(call)
         pending_seconds = pending.seconds if pending else 0.0
         total = self.window_audio_seconds + pending_seconds + seconds
@@ -361,8 +470,13 @@ class AsrUsageGuard:
                         f"frame={seconds:.3f}s cap={self.policy.max_audio_seconds:.3f}s"),
             )
 
-    def note_audio_sent(self, call: str, sample_rate: int, samples: int, nbytes: int = 0) -> None:
-        """真实发送成功后计数（重发按重复计）；达到 flush 阈值聚合落账."""
+    def reserve_audio(self, call: str, sample_rate: int, samples: int, nbytes: int = 0) -> None:
+        """发送前保守预记整帧（HEL-906 复核口径）.
+
+        越限帧在此拒绝（不出站）；计入预算后发送失败/取消**不退还**——帧
+        可能已出站。确证送达量另记 note_audio_delivered，二者不得混同。
+        """
+        self.precheck_audio(call, sample_rate, samples)
         cu = self.calls.setdefault(call, CallUsage())
         seconds = round(samples / sample_rate, 6)
         cu.audio_seconds = round(cu.audio_seconds + seconds, 6)
@@ -374,13 +488,25 @@ class AsrUsageGuard:
         if buf.seconds >= self.policy.flush_seconds:
             self.flush_audio(call)
 
+    def note_audio_delivered(self, call: str, sample_rate: int, samples: int,
+                             nbytes: int = 0) -> None:
+        """确证送达量单独累计（``await ws.send`` 成功返回后调用；不重复计费）."""
+        cu = self.calls.setdefault(call, CallUsage())
+        seconds = round(samples / sample_rate, 6)
+        cu.confirmed_audio_seconds = round(cu.confirmed_audio_seconds + seconds, 6)
+        buf = self._pending.setdefault(call, _AudioBuf())
+        buf.confirmed_seconds = round(buf.confirmed_seconds + seconds, 6)
+
     def flush_audio(self, call: str) -> None:
         """把该通缓冲中的音频计数聚合落账（utterance 边界 / 流结束时调用）."""
         buf = self._pending.pop(call, None)
-        if buf is None or buf.seconds <= 0:
+        if buf is None or (buf.seconds <= 0 and buf.confirmed_seconds <= 0):
             return
         self.window_audio_seconds = round(self.window_audio_seconds + buf.seconds, 6)
+        self.window_confirmed_seconds = round(
+            self.window_confirmed_seconds + buf.confirmed_seconds, 6)
         self._append({"kind": "audio", "call": call, "seconds": buf.seconds,
+                      "confirmed_seconds": buf.confirmed_seconds,
                       "frames": buf.frames, "bytes": buf.bytes, "sample_rate": buf.sample_rate})
 
     def stream_end(self, call: str, reason: str) -> None:
@@ -392,6 +518,7 @@ class AsrUsageGuard:
         self._append({
             "kind": "call_end", "call": call, "reason": reason,
             "audio_seconds": round(cu.audio_seconds, 6) if cu else 0.0,
+            "confirmed_audio_seconds": round(cu.confirmed_audio_seconds, 6) if cu else 0.0,
             "connect_attempts": cu.connect_attempts if cu else 0,
             "connect_failures": cu.connect_failures if cu else 0,
             "connect_retries": cu.connect_retries if cu else 0,
@@ -404,15 +531,18 @@ class AsrUsageGuard:
 
     def summary(self) -> dict:
         pending = round(sum(b.seconds for b in self._pending.values()), 6)
+        pending_confirmed = round(sum(b.confirmed_seconds for b in self._pending.values()), 6)
         return {
             "window": self.policy.window_id,
             "armed": self.policy.armed,
             "max_audio_seconds": self.policy.max_audio_seconds,
             "max_connect_retries_per_call": self.policy.max_connect_retries_per_call,
             "audio_seconds_flushed": round(self.window_audio_seconds, 6),
+            "audio_seconds_confirmed": round(self.window_confirmed_seconds + pending_confirmed, 6),
             "audio_seconds_with_pending": round(self.window_audio_seconds + pending, 6),
             "ledger": str(self.ledger_path) if self.ledger_path else "",
             "ledger_errors": self.ledger_errors,
+            "persistence_stop_reason": self.persistence_stop_reason,
             "calls": {
                 call: {
                     "streams": cu.streams,
@@ -420,6 +550,7 @@ class AsrUsageGuard:
                     "connect_failures": cu.connect_failures,
                     "connect_retries": cu.connect_retries,
                     "audio_seconds": round(cu.audio_seconds, 6),
+                    "audio_seconds_confirmed": round(cu.confirmed_audio_seconds, 6),
                 }
                 for call, cu in self.calls.items()
             },
@@ -427,12 +558,16 @@ class AsrUsageGuard:
 
     def summary_line(self) -> str:
         s = self.summary()
-        return (
+        line = (
             f"window={s['window'] or 'not-armed'} "
             f"audio={s['audio_seconds_with_pending']:.3f}/{s['max_audio_seconds']:.3f}s "
+            f"confirmed={s['audio_seconds_confirmed']:.3f}s "
             f"retries<= {s['max_connect_retries_per_call']}/call "
             f"ledger={s['ledger']}"
         )
+        if s["persistence_stop_reason"]:
+            line += f" STOPPED[{s['persistence_stop_reason']}]"
+        return line
 
 
 def build_usage_from_env(

src/cybergril/voice/stt_funasr.py

@@ -14,10 +14,11 @@
 
 HEL-906 用量接缝: FunASRSTT 可注入 AsrUsageGuard(voice/asr_usage.py, worker
 按 job 构造). 每次建连前 check_connect、建连结果 note_connect_result、
-每个真实发送帧发送前 precheck_audio(越限在发送前拒绝)、发送成功后
-note_audio_sent; 流结束/utterance 结束 flush_audio, 流异常时 retry 标记
-由 guard 落账. usage=None 时行为与本卡之前完全一致(计数只在布防窗口生效,
-未布防 guard 本身即 fail-closed).
+每个真实发送帧发送前 reserve_audio(保守预记＋越限帧发送前拒绝)、
+``await ws.send`` 成功返回后 note_audio_delivered(确证送达量)；发送失败/
+取消不退还预记——帧可能已出站(HEL-906 复核口径). 流结束/utterance 结束
+flush_audio, 流异常时 retry 标记由 guard 落账. usage=None 时行为与本卡
+之前完全一致(计数只在布防窗口生效, 未布防 guard 本身即 fail-closed).
 """
 
 from __future__ import annotations
@@ -175,14 +176,17 @@ class FunASRRecognizeStream(RecognizeStream):
             return u
 
         async def send_frame(f) -> None:
-            # HEL-906: 发送前检查(越限帧一个字节都不出站), 发送成功才计数
+            # HEL-906 复核: 发送前保守预记(越限帧一个字节不出站、账本故障/
+            # 触线在此拒绝)；await 成功返回才记确证送达。失败/取消不退还预记
+            # ——帧可能已出站，保守计费不退。
             payload = bytes(f.data)
             if usage is not None:
-                usage.precheck_audio(call_id, f.sample_rate, f.samples_per_channel)
+                usage.reserve_audio(call_id, f.sample_rate, f.samples_per_channel,
+                                    len(payload))
             await utt.ws.send(payload)  # type: ignore[union-attr]
             if usage is not None:
-                usage.note_audio_sent(call_id, f.sample_rate,
-                                      f.samples_per_channel, len(payload))
+                usage.note_audio_delivered(call_id, f.sample_rate,
+                                           f.samples_per_channel, len(payload))
 
         async def recv_loop(u: _Utterance) -> None:
             assert u.ws is not None

tests/test_voice_asr_usage.py

@@ -6,10 +6,14 @@
 1. 纯计数/闸/账本恢复（guard 单元）：fail-closed 默认、按实际帧样本计时、
    触线前拒绝、重试口径、落盘恢复、跨窗口隔离、崩溃少计有界。
 2. 真实接缝：用假 livekit/websockets 模块（隔离假 provider、假帧）导入
-  真实的 voice/stt_funasr.py 与 voice/worker.py，驱动真实 ``_run`` 送数
-   路径——建连检查、发送前拒绝、发送后计数、接收侧死亡、流异常重试、
-   utterance 聚合落账全部走产品代码；worker entrypoint 布线（guard 注入、
-   房间名标识、触线 ctx.shutdown、未布防拒绝）逐项断言。
+   真实的 voice/stt_funasr.py 与 voice/worker.py，驱动真实 ``_run`` 送数
+   路径——建连检查、发送前拒绝、保守预记/确证送达、接收侧死亡、流异常
+   重试、utterance 聚合落账全部走产品代码；worker entrypoint 布线（guard
+   注入、房间名标识、触线 ctx.shutdown、未布防拒绝）逐项断言。
+3. 2026-10-28 总管定点复核返工：账本写失败→重放不得回退预算（复现脚本的
+   回归路径）、读失败/损坏/中途丢失 fail-closed 停机并保全已知消耗、
+   guard_stop 跨重启仍停、假 ws 先记 payload 再 raise/取消后预算仍包含
+   该帧（保守计费不退还、确证量单列）。
 
 零真实 ASR/TTS/LiveKit 调用（fake 连接全部在进程内），零安装。
 """
@@ -82,19 +86,31 @@ class FakeResampler:
 
 
 class FakeWS:
-    """进程内 FunASR websocket 假 provider：记录每个出站字节."""
+    """进程内 FunASR websocket 假 provider：记录每个出站字节.
+
+    fail_send 模拟「帧可能已出站」的不确定失败：先把 payload 记入 sent
+    再 raise/cancel（控制消息不受 fail_bytes_after 影响）。
+    """
 
     def __init__(self, server_msgs: list[str] | None = None, recv_fail: bool = False) -> None:
         self.sent: list[bytes | str] = []
         self.server_msgs = list(server_msgs or [])
         self.recv_fail = recv_fail
         self.closed = False
-        self.fail_send = False
+        self.fail_send = ""  # "" | "raise" | "cancel"
+        self.fail_bytes_after: int | None = None  # 第 N 个音频帧之后开始失败
 
     async def send(self, payload: bytes | str) -> None:
-        if self.fail_send:
-            raise OSError("send refused")
-        self.sent.append(payload)
+        self.sent.append(payload)  # 先记：即使随后失败，帧也可能已经出站
+        if isinstance(payload, (bytes, bytearray)) and self.fail_send:
+            if self.fail_bytes_after is not None and self.fail_bytes_after > 0:
+                self.fail_bytes_after -= 1
+                return  # 前 N 个音频帧正常送达
+            self.fail_bytes_after = None  # 计数耗尽：之后的音频帧都按失败语义
+            if self.fail_send == "raise":
+                raise OSError("send failed after outbound")
+            if self.fail_send == "cancel":
+                raise asyncio.CancelledError()
 
     async def close(self) -> None:
         self.closed = True
@@ -447,15 +463,32 @@ def test_guard_audio_counted_by_actual_frame_samples(tmp_path):
     guard.note_connect_result(call, ok=True)
     # 16k 直接路径：160 样本 = 0.01s；重采样路径按输出帧 16k 计
     for _ in range(10):
-        guard.precheck_audio(call, 16000, 160)
-        guard.note_audio_sent(call, 16000, 160, 320)
+        guard.reserve_audio(call, 16000, 160, 320)
+        guard.note_audio_delivered(call, 16000, 160, 320)
     for _ in range(5):
-        guard.precheck_audio(call, 16000, 160)
-        guard.note_audio_sent(call, 16000, 160, 320)  # 重发按重复计
+        guard.reserve_audio(call, 16000, 160, 320)
+        guard.note_audio_delivered(call, 16000, 160, 320)  # 重发按重复计
     guard.stream_end(call, "closed")
     assert guard.calls[call].audio_seconds == pytest.approx(0.15)
+    assert guard.calls[call].confirmed_audio_seconds == pytest.approx(0.15)
     audio = [r for r in read_records(tmp_path) if r["kind"] == "audio"]
     assert sum(r["seconds"] for r in audio) == pytest.approx(0.15)
+    assert sum(r["confirmed_seconds"] for r in audio) == pytest.approx(0.15)
+
+
+def test_guard_uncertain_outbound_not_refunded(tmp_path):
+    """发送前预记、失败/取消不退还（保守计费）；确证量只记成功返回的帧."""
+    guard = make_guard(tmp_path)
+    call = guard.begin_stream("room-a")
+    guard.reserve_audio(call, 16000, 160)  # 发送失败的帧：保守预记保留
+    guard.reserve_audio(call, 16000, 160)
+    guard.note_audio_delivered(call, 16000, 160)  # 只有这帧确证送达
+    guard.flush_audio(call)
+    assert guard.calls[call].audio_seconds == pytest.approx(0.02)
+    assert guard.calls[call].confirmed_audio_seconds == pytest.approx(0.01)
+    audio = [r for r in read_records(tmp_path) if r["kind"] == "audio"][-1]
+    assert audio["seconds"] == pytest.approx(0.02)
+    assert audio["confirmed_seconds"] == pytest.approx(0.01)
 
 
 def test_guard_blocks_crossing_frame_before_send(tmp_path):
@@ -464,14 +497,14 @@ def test_guard_blocks_crossing_frame_before_send(tmp_path):
     call = guard.begin_stream("room-a")
     guard.check_connect(call)
     guard.note_connect_result(call, ok=True)
-    guard.precheck_audio(call, 16000, 16000)  # 1.0s
-    guard.note_audio_sent(call, 16000, 16000, 32000)
+    guard.reserve_audio(call, 16000, 16000, 32000)  # 1.0s
+    guard.note_audio_delivered(call, 16000, 16000, 32000)
     guard.flush_audio(call)
-    guard.precheck_audio(call, 16000, 16000)  # 恰好等于上限：允许（≤上限）
-    guard.note_audio_sent(call, 16000, 16000, 32000)
+    guard.reserve_audio(call, 16000, 16000)  # 恰好等于上限：允许（≤上限）
+    guard.note_audio_delivered(call, 16000, 16000, 32000)
     guard.flush_audio(call)
     with pytest.raises(AsrWindowClosed):  # 第 3 秒会在发送前拒绝
-        guard.precheck_audio(call, 16000, 16000)
+        guard.reserve_audio(call, 16000, 16000)
     assert stops == ["window-audio-exhausted"]  # 回调恰好一次
     # 同窗新通也立即拒绝（新连接/新音频都不放行）
     with pytest.raises(AsrWindowClosed):
@@ -526,16 +559,18 @@ def test_guard_ledger_recovery_same_window_no_reset(tmp_path):
     guard1.check_connect(call)
     guard1.note_connect_result(call, ok=False, error="x")
     for _ in range(150):
-        guard1.precheck_audio(call, 16000, 160)
-        guard1.note_audio_sent(call, 16000, 160, 320)
+        guard1.reserve_audio(call, 16000, 160, 320)
+        guard1.note_audio_delivered(call, 16000, 160, 320)
     guard1.flush_audio(call)
     guard1.stream_end(call, "closed")
     # 模拟进程重启：同窗口同账本重建 guard，预算与逐通历史都在
     guard2 = make_guard(tmp_path, max_audio_seconds=3.0)
     assert guard2.window_audio_seconds == pytest.approx(1.5)
+    assert guard2.window_confirmed_seconds == pytest.approx(1.5)
     cu = guard2.calls["room-a"]
     assert (cu.connect_attempts, cu.connect_failures, cu.connect_retries) == (2, 1, 0)
     assert cu.audio_seconds == pytest.approx(1.5)
+    assert cu.confirmed_audio_seconds == pytest.approx(1.5)
     assert cu.retry_pending is True  # 最后一次连接事实是失败
     # 耗尽同窗口剩余预算后拒绝；重开同窗口 id 不能重置
     call = guard2.begin_stream("room-a")
@@ -543,8 +578,8 @@ def test_guard_ledger_recovery_same_window_no_reset(tmp_path):
     guard2.note_connect_result(call, ok=True)  # retry_pending=True → 这是 retry
     with pytest.raises(AsrWindowClosed):
         for _ in range(200):
-            guard2.precheck_audio(call, 16000, 160)
-            guard2.note_audio_sent(call, 16000, 160, 320)
+            guard2.reserve_audio(call, 16000, 160, 320)
+            guard2.note_audio_delivered(call, 16000, 160, 320)
     guard2.flush_audio(call)
 
 
@@ -556,15 +591,15 @@ def test_guard_window_isolation_does_not_stop_others(tmp_path):
         g1.check_connect(call)
         g1.note_connect_result(call, ok=True)
         for _ in range(10):
-            g1.precheck_audio(call, 16000, 160)
-            g1.note_audio_sent(call, 16000, 160, 320)
+            g1.reserve_audio(call, 16000, 160, 320)
+            g1.note_audio_delivered(call, 16000, 160, 320)
     g2 = make_guard(tmp_path, window_id="w2", max_audio_seconds=100.0)
     call = g2.begin_stream("room-a")  # 同账本、不同窗口：完全不受 W1 影响
     g2.check_connect(call)
     g2.note_connect_result(call, ok=True)
     for _ in range(10):
-        g2.precheck_audio(call, 16000, 160)
-        g2.note_audio_sent(call, 16000, 160, 320)
+        g2.reserve_audio(call, 16000, 160, 320)
+        g2.note_audio_delivered(call, 16000, 160, 320)
     g2.stream_end(call, "closed")
     assert g2.calls["room-a"].audio_seconds == pytest.approx(0.1)
 
@@ -576,15 +611,16 @@ def test_guard_crash_undercount_bounded_by_flush(tmp_path):
     guard.check_connect(call)
     guard.note_connect_result(call, ok=True)
     for _ in range(30):
-        guard.precheck_audio(call, 16000, 160)
-        guard.note_audio_sent(call, 16000, 160, 320)  # 0.3s < flush 阈值
+        guard.reserve_audio(call, 16000, 160, 320)
+        guard.note_audio_delivered(call, 16000, 160, 320)  # 0.3s < flush 阈值
     # 模拟进程直接消失（没有 stream_end/flush）
     replayed = summarize_ledger(tmp_path / "asr-usage.jsonl", "w-test")
     assert replayed["windows"]["w-test"]["audio_seconds"] == 0.0
     # 正常路径（flush/stream_end）计数完整落账
     guard2 = make_guard(tmp_path, flush_seconds=5.0)
     call = guard2.begin_stream("room-b")
-    guard2.note_audio_sent(call, 16000, 16000, 32000)
+    guard2.reserve_audio(call, 16000, 16000, 32000)
+    guard2.note_audio_delivered(call, 16000, 16000, 32000)
     guard2.stream_end(call, "closed")
     assert summarize_ledger(tmp_path / "asr-usage.jsonl", "w-test")["windows"][
         "w-test"]["calls"]["room-b"]["audio_seconds"] == pytest.approx(1.0)
@@ -595,7 +631,8 @@ def test_summarize_cli_reproducible(tmp_path, monkeypatch):
     call = guard.begin_stream("room-a")
     guard.check_connect(call)
     guard.note_connect_result(call, ok=True)
-    guard.note_audio_sent(call, 16000, 16000, 32000)
+    guard.reserve_audio(call, 16000, 16000, 32000)
+    guard.note_audio_delivered(call, 16000, 16000, 32000)
     guard.stream_end(call, "closed")
     repo_root = Path(__file__).resolve().parents[1]
     proc = subprocess.run(
@@ -631,11 +668,13 @@ async def test_seam_counts_fed_audio_both_sample_rates(fake_sdk, tmp_path):
     cu = guard.calls["cg-voice-abc"]
     # 16k 直接路径 0.3s + 48k→16k 重采样路径 0.3s（按实际输出帧计时）
     assert cu.audio_seconds == pytest.approx(0.6)
+    assert cu.confirmed_audio_seconds == pytest.approx(0.6)  # 全部送达成功
     assert cu.connect_attempts == 2  # 两轮 utterance 两次建连
     assert (cu.connect_failures, cu.connect_retries) == (0, 0)
     records = read_records(tmp_path)
     audio_records = [r for r in records if r["kind"] == "audio"]
     assert [round(r["seconds"], 6) for r in audio_records] == [0.3, 0.3]  # utterance 边界聚合
+    assert [round(r["confirmed_seconds"], 6) for r in audio_records] == [0.3, 0.3]
     assert all(r["call"] == "cg-voice-abc" for r in records)
     ends = [r for r in records if r["kind"] == "call_end"]
     assert [r["reason"] for r in ends] == ["closed"]
@@ -719,9 +758,10 @@ async def test_seam_window_exhaustion_stops_before_crossing(fake_sdk, tmp_path):
     stt = funasr.FunASRSTT(usage=guard, call_label="room-cap")
     with pytest.raises(AsrWindowClosed):
         await drive(funasr, stt, [frame_16k() for _ in range(6)] + [None])
-    # 0.05s = 5 帧放行，第 6 帧发送前拒绝：出站字节恰好 5 帧
+    # 0.05s = 5 帧放行，第 6 帧发送前拒绝：出站字节恰好 5 帧（provider 未被触线帧触碰）
     assert len(fake_sdk.websockets.bytes_sent()) == 5
     assert guard.calls["room-cap"].audio_seconds == pytest.approx(0.05)
+    assert guard.calls["room-cap"].confirmed_audio_seconds == pytest.approx(0.05)
     assert stops == ["window-audio-exhausted"]
     records = read_records(tmp_path)
     assert records[-1]["kind"] == "call_end"
@@ -808,10 +848,10 @@ async def test_worker_entrypoint_wires_guard(fake_sdk, tmp_path, monkeypatch, ca
     assert "asr_window=" in out and "hel902-w1" in out
     # 触线回调真的接到 ctx.shutdown：预检越限帧 → 结束本通 job
     call = guard.begin_stream("cg-voice-room1")
-    guard.note_audio_sent(call, 16000, 16000, 32000)
+    guard.reserve_audio(call, 16000, 16000, 32000)
     guard.flush_audio(call)
     with pytest.raises(AsrWindowClosed):
-        guard.precheck_audio(call, 16000, 16000 * 2700)  # 会越 2700s 上限的帧
+        guard.reserve_audio(call, 16000, 16000 * 2700)  # 会越 2700s 上限的帧
     assert ctx.shutdown_calls, "on_exhausted 必须真的调用 ctx.shutdown 结束本通"
 
 
@@ -829,3 +869,209 @@ async def test_worker_entrypoint_default_is_fail_closed(fake_sdk, monkeypatch, c
     with pytest.raises(AsrWindowClosed):
         guard.begin_stream("cg-voice-room2")
     assert "asr_window=" in capsys.readouterr().out
+
+
+# ---------------------------------------------------------------------------
+# 4) 2026-10-28 总管定点复核返工：持久化 fail-closed 与不确定出站口径
+# ---------------------------------------------------------------------------
+
+class FakeClock:
+    def __init__(self, t: float = 0.0) -> None:
+        self.t = t
+
+    def __call__(self) -> float:
+        return self.t
+
+    def advance(self, dt: float) -> None:
+        self.t += dt
+
+
+def _rework_guard(tmp_path: Path, clock: FakeClock, *, on_exhausted=None,
+                  **overrides) -> AsrUsageGuard:
+    values = dict(window_id="w-rework", max_audio_seconds=10.0,
+                  flush_seconds=1e9, resync_seconds=5.0)
+    values.update(overrides)
+    return AsrUsageGuard(AsrWindowPolicy(**values), tmp_path / "asr-usage.jsonl",
+                         clock=clock, on_exhausted=on_exhausted)
+
+
+def _feed(guard: AsrUsageGuard, call: str, seconds: float, *, deliver: bool = True) -> None:
+    frames = round(seconds * 100)
+    for _ in range(frames):
+        guard.reserve_audio(call, 16000, 160, 320)
+        if deliver:
+            guard.note_audio_delivered(call, 16000, 160, 320)
+
+
+def test_ledger_write_failure_never_regresses_budget(tmp_path, monkeypatch):
+    """总管复现的回归路径：写失败→flush 丢缓冲→重放按旧账回退，修复后不再发生."""
+    import pathlib
+
+    stops: list[str] = []
+    clock = FakeClock(0.0)
+    guard = _rework_guard(tmp_path, clock, max_audio_seconds=1.5,
+                          on_exhausted=stops.append)
+    call = guard.begin_stream("room")
+    guard.check_connect(call)
+    guard.note_connect_result(call, ok=True)
+    _feed(guard, call, 0.8)
+    guard.flush_audio(call)  # 0.8s 落账成功
+    assert guard.window_audio_seconds == pytest.approx(0.8)
+
+    real_open = pathlib.Path.open
+
+    def fail_append(self, mode="r", *args, **kwargs):
+        if "a" in (mode or ""):
+            raise OSError("append refused (injected)")
+        return real_open(self, mode, *args, **kwargs)
+
+    monkeypatch.setattr(pathlib.Path, "open", fail_append)
+
+    _feed(guard, call, 0.5)
+    guard.flush_audio(call)  # 0.5s 落账失败：内存保全、立即停机、不许回退
+    assert guard.window_audio_seconds == pytest.approx(1.3)
+    assert guard.persistence_stop_reason == "ledger-write-failed"
+    assert stops == ["ledger-write-failed"]  # 停机回调恰好一次并带原因
+
+    clock.advance(6.0)  # 越过 resync：重放不得把内存拉回旧账本值 0.8
+    with pytest.raises(AsrWindowClosed) as ei:
+        _feed(guard, call, 0.6)  # 复现脚本里的「再送 0.6s」：现在被拒
+    assert "ledger-write-failed" in str(ei.value)
+    assert guard.window_audio_seconds == pytest.approx(1.3)  # 无预算倒退
+    # 写失败注入仍生效期间 block 记录同样写不落账——停机事实以日志与内存为准
+    assert not [r for r in read_records(tmp_path) if r["kind"] == "block"]
+    # 停机状态在 summary/summary_line 可见（操作员可核对）
+    assert guard.summary()["persistence_stop_reason"] == "ledger-write-failed"
+    assert "ledger-write-failed" in guard.summary_line()
+
+
+def test_ledger_read_failure_stops_outbound_preserving_memory(tmp_path, monkeypatch):
+    import pathlib
+
+    clock = FakeClock(0.0)
+    guard = _rework_guard(tmp_path, clock)
+    call = guard.begin_stream("room")
+    guard.check_connect(call)
+    guard.note_connect_result(call, ok=True)
+    _feed(guard, call, 1.0)
+    guard.flush_audio(call)
+
+    def boom(self, *args, **kwargs):
+        raise OSError("read refused (injected)")
+
+    monkeypatch.setattr(pathlib.Path, "read_text", boom)
+    clock.advance(6.0)  # resync 触发严格读：读失败 → fail-closed
+    with pytest.raises(AsrWindowClosed):
+        guard.reserve_audio(call, 16000, 160, 320)
+    assert guard.persistence_stop_reason == "ledger-read-failed"
+    assert guard.window_audio_seconds == pytest.approx(1.0)  # 不清零、不倒退
+    with pytest.raises(AsrWindowClosed):  # 新通也不放行
+        guard.begin_stream("room-2")
+
+
+def test_ledger_corruption_stops_outbound(tmp_path):
+    clock = FakeClock(0.0)
+    guard = _rework_guard(tmp_path, clock)
+    call = guard.begin_stream("room")
+    guard.check_connect(call)
+    guard.note_connect_result(call, ok=True)
+    _feed(guard, call, 1.0)
+    guard.flush_audio(call)
+
+    with (tmp_path / "asr-usage.jsonl").open("a", encoding="utf-8") as fh:
+        fh.write("{corrupt-not-json\n")  # 中途损坏一行
+    clock.advance(6.0)
+    with pytest.raises(AsrWindowClosed):
+        guard.reserve_audio(call, 16000, 160, 320)
+    assert guard.persistence_stop_reason == "ledger-corrupt"
+    assert guard.window_audio_seconds == pytest.approx(1.0)  # 损坏不套用、内存保全
+    with pytest.raises(AsrWindowClosed):
+        guard.check_connect("room")
+
+
+def test_ledger_missing_after_seen_stops_outbound(tmp_path):
+    clock = FakeClock(0.0)
+    guard = _rework_guard(tmp_path, clock)
+    call = guard.begin_stream("room")
+    guard.check_connect(call)
+    guard.note_connect_result(call, ok=True)
+    _feed(guard, call, 1.0)
+    guard.flush_audio(call)  # 账本已存在（读过/写过）
+
+    (tmp_path / "asr-usage.jsonl").unlink()  # 中途消失 = 已落账事实丢失
+    clock.advance(6.0)
+    with pytest.raises(AsrWindowClosed):
+        guard.reserve_audio(call, 16000, 160, 320)
+    assert guard.persistence_stop_reason == "ledger-missing"
+    assert guard.window_audio_seconds == pytest.approx(1.0)
+    # 停机事实已落回重建的账本（guard_stop）：汇总 CLI 可见
+    assert summarize_ledger(tmp_path / "asr-usage.jsonl", "w-rework")[
+        "windows"]["w-rework"]["stopped"] is True
+
+
+def test_guard_stop_record_keeps_window_stopped_across_restart(tmp_path):
+    ledger = tmp_path / "asr-usage.jsonl"
+    lines = [
+        {"kind": "audio", "window": "w", "call": "room", "seconds": 2.0,
+         "confirmed_seconds": 2.0, "t": 1},
+        {"kind": "guard_stop", "window": "w", "reason": "ledger-corrupt", "t": 2},
+    ]
+    ledger.write_text("".join(json.dumps(x) + "\n" for x in lines), encoding="utf-8")
+    guard = AsrUsageGuard(
+        AsrWindowPolicy(window_id="w", max_audio_seconds=10.0), ledger)
+    # 同窗重启不恢复出站；已知消耗（2.0s）仍被重放保全
+    assert guard.persistence_stop_reason == "ledger-corrupt"
+    assert guard.window_audio_seconds == pytest.approx(2.0)
+    with pytest.raises(AsrWindowClosed):
+        guard.begin_stream("room-new")
+
+
+async def test_seam_uncertain_outbound_counted_conservatively(fake_sdk, tmp_path):
+    """假 ws 先记下 payload 再 raise/取消：预算仍包含该帧，确证量不包含."""
+    import cybergril.voice.stt_funasr as funasr
+
+    guard = make_guard(tmp_path)
+
+    def flaky(mode: str, ok_frames: int) -> FakeWS:
+        ws = FakeWS(server_msgs=[])
+        ws.fail_send = mode
+        ws.fail_bytes_after = ok_frames
+        return ws
+
+    # 场景 1：第 3 帧送出后 raise（帧可能已出站）
+    fake_sdk.websockets.ws_factory = lambda: flaky("raise", 2)
+    stt = funasr.FunASRSTT(usage=guard, call_label="room-raise")
+    with pytest.raises(OSError):
+        await drive(funasr, stt, [frame_16k() for _ in range(3)] + [None])
+    cu = guard.calls["room-raise"]
+    assert cu.audio_seconds == pytest.approx(0.03)  # 保守计费含失败帧
+    assert cu.confirmed_audio_seconds == pytest.approx(0.02)  # 确证只有前两帧
+    ws_raise = fake_sdk.websockets.sockets[-1]
+    assert len([p for p in ws_raise.sent if isinstance(p, bytes)]) == 3  # 帧确实已出站
+    rec = [r for r in read_records(tmp_path)
+           if r["kind"] == "audio" and r["call"] == "room-raise"][-1]
+    assert rec["seconds"] == pytest.approx(0.03)
+    assert rec["confirmed_seconds"] == pytest.approx(0.02)
+    assert [r for r in read_records(tmp_path)
+            if r["kind"] == "call_end" and r["call"] == "room-raise"][-1]["reason"] == "error"
+
+    # 场景 2：第 2 帧送出后被取消：同样不退还预记
+    fake_sdk.websockets.ws_factory = lambda: flaky("cancel", 1)
+    stt2 = funasr.FunASRSTT(usage=guard, call_label="room-cancel")
+    with pytest.raises(asyncio.CancelledError):
+        await drive(funasr, stt2, [frame_16k() for _ in range(2)] + [None])
+    cu2 = guard.calls["room-cancel"]
+    assert cu2.audio_seconds == pytest.approx(0.02)
+    assert cu2.confirmed_audio_seconds == pytest.approx(0.01)
+    ws_cancel = fake_sdk.websockets.sockets[-1]
+    assert len([p for p in ws_cancel.sent if isinstance(p, bytes)]) == 2
+
+    # 触线拒绝仍然不碰 provider：越限帧在 reserve 阶段就拒绝，ws.send 不执行
+    guard_cap = AsrUsageGuard(
+        make_policy(max_audio_seconds=0.01), tmp_path / "cap.jsonl")
+    fake_sdk.websockets.ws_factory = None
+    stt3 = funasr.FunASRSTT(usage=guard_cap, call_label="room-cap2")
+    with pytest.raises(AsrWindowClosed):
+        await drive(funasr, stt3, [frame_16k() for _ in range(3)] + [None])
+    ws_cap = fake_sdk.websockets.sockets[-1]
+    assert len([p for p in ws_cap.sent if isinstance(p, bytes)]) == 1  # 只有第 1 帧出站