HEL-906: continuous-call ASR usage metering + one-shot window guard in dedicated worker seam

- asr_usage.py (stdlib-only): append-only JSONL ledger, per-call/window audio
  seconds counted from actually-sent frames (sample-rate aware, resend counted),
  connect attempt/failure/retry semantics (normal per-utterance reconnects are
  not retries), fail-closed default (real cap=0 unless a window is armed),
  pre-send blocking so no byte crosses the cap, ledger replay for restart
  recovery and cross-window isolation, resumable summarize CLI
- stt_funasr.py: guard hooks on the real send path (check_connect before
  open_utt, precheck before send, count after send, recv-side death recorded,
  per-utterance flush aggregation); usage=None keeps prior behavior
- worker.py: entrypoint builds guard from env per job, room name as call id,
  on_exhausted -> ctx.shutdown ends the call, window summary in start log
- tests: 17 deterministic offline tests incl. fake livekit/websockets SDK
  driving the real _run seam, sample-rate variants, silence/cancel/failure/
  reconnect/repeated-end, cap rejection before crossing, ledger persistence
  across restart, room deletion keeps history, other windows unaffected

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

src/cybergril/voice/asr_usage.py

@@ -0,0 +1,464 @@
+"""连续通话 ASR 实际用量计数与一次性窗口闸（HEL-906）.
+
+补齐 HEL-905 复审留下的 §4 待补阻塞：连续通话内 FunASR 的**实际送入音频
+量**与**连接尝试/重连/失败**没有本仓库可执行的程序计数。本模块只依赖
+标准库（API 进程与离线测试环境都能直接导入），计数全部发生在专属
+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。
+- 音频按**实际发送帧**的 sample_rate 与样本数计时（重采样后的输出帧按其
+  自身采样率计），同一帧重复发送按重复计；墙钟、开房/删房流水不参与
+  音频计时，事后成功流水也不追溯改数。
+- 连接事实逐条落账：每次建连尝试（含失败）、失败后的重试（retry）、
+  接收侧死亡（connection_died）。FunASR 一句一连的惰性建连里，正常说完
+  一轮后的下一次建连不是 retry，只有上一次尝试失败/连接死亡/流异常结束
+  之后的建连才计 retry。
+
+账本（追加写 JSONL，默认 /opt/voice-agent-cybergril/asr-usage.jsonl）独立
+于实例侧会话账本 ``<db>-voice-sessions.json``：删房成功会把通话从会话账
+本移除，但这里的逐通历史（call_start/audio/connect_attempt/
+connection_died/call_end/block）永不改写、不随删房消失；窗口总量由重放
+账本恢复，同 window_id 重启进程不能重置预算，旧窗口耗尽状态也被持久化
+（不得把旧余额或新开账当作新许可）。
+
+窗口语义沿用 think/budget.py 的 fail-closed 先例：
+
+- 未布防（CYBERGRIL_ASR_WINDOW_ID 未设，产品默认）：一切新连接与新音频
+  在发送前拒绝（真实 cap=0），拒绝本身落账；
+- 布防（获批的一次性窗口启动时由窗口负责人显式设置环境变量）：触线即
+  在发送前拒绝（AsrWindowClosed），worker 经 on_exhausted 结束本通 job；
+  上限只能收紧不能放宽，候选值见 docs/rework/hel902/r5-followup-package.md §4。
+
+程序能保证 / 剩余只能监看（如实边界）：
+
+- 保证：布防进程内逐帧发送前硬停（帧不越上限一个字节）、逐通与全窗
+  计数不因删房/重启丢失、同窗重启不能重置预算。
+- 跨进程有界收敛：账本是共同事实，每次建连检查与 resync_seconds 间隔
+  重放账本；两进程并行通话时对端最近 ≤ max(resync, flush) 间隔的音频
+  可能未计入本轮判定（有界过冲）。
+- 只能监看：进程崩溃前 ≤ flush_seconds 的已发音频可能少计；接收侧死亡
+  未伴随发送失败时按 connection_died 计连接事实、不改已计音频；真实
+  FunASR/LiveKit 行为仍以窗口前 .12 实测为准。
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import json
+import os
+import secrets
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+DEFAULT_ASR_USAGE_LEDGER = "/opt/voice-agent-cybergril/asr-usage.jsonl"
+DEFAULT_FLUSH_SECONDS = 5.0
+DEFAULT_RESYNC_SECONDS = 5.0
+
+# 允许浮点求和后恰好等于上限（“≤上限”而不是“<上限”）
+_EPSILON = 1e-9
+
+
+class AsrWindowClosed(Exception):
+    """The armed window (or the fail-closed default) refuses this connection/audio."""
+
+
+@dataclass(frozen=True)
+class AsrWindowPolicy:
+    window_id: str = ""  # 空 = 未布防：真实 cap=0，拒绝一切 ASR 连接/音频
+    max_audio_seconds: float = 0.0  # 全窗口实际送入 ASR 音频上限
+    max_connect_retries_per_call: int = 0  # 每通连接重试上限（失败后的再建连）
+    flush_seconds: float = DEFAULT_FLUSH_SECONDS  # 音频计数聚合落账阈值
+    resync_seconds: float = DEFAULT_RESYNC_SECONDS  # 跨进程账本重放间隔
+
+    @property
+    def armed(self) -> bool:
+        return bool(self.window_id)
+
+
+def _env_float(env: dict, name: str, default: float) -> float:
+    raw = (env.get(name) or "").strip()
+    if not raw:
+        return default
+    try:
+        value = float(raw)
+    except ValueError:
+        return 0.0  # 配置损坏按 fail-closed 处理
+    return value if value > 0 else 0.0
+
+
+def _env_int(env: dict, name: str, default: int) -> int:
+    raw = (env.get(name) or "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return 0
+    return value if value > 0 else 0
+
+
+def policy_from_env(environ: dict | None = None) -> AsrWindowPolicy:
+    env = dict(os.environ) if environ is None else environ
+    return AsrWindowPolicy(
+        window_id=(env.get("CYBERGRIL_ASR_WINDOW_ID") or "").strip(),
+        max_audio_seconds=_env_float(env, "CYBERGRIL_ASR_WINDOW_MAX_AUDIO_SECONDS", 0.0),
+        max_connect_retries_per_call=_env_int(
+            env, "CYBERGRIL_ASR_MAX_CONNECT_RETRIES_PER_CALL", 0
+        ),
+        flush_seconds=_env_float(env, "CYBERGRIL_ASR_USAGE_FLUSH_SECONDS", DEFAULT_FLUSH_SECONDS),
+        resync_seconds=_env_float(env, "CYBERGRIL_ASR_USAGE_RESYNC_SECONDS", DEFAULT_RESYNC_SECONDS),
+    )
+
+
+@dataclass
+class _AudioBuf:
+    seconds: float = 0.0
+    frames: int = 0
+    bytes: int = 0
+    sample_rate: int = 0
+
+
+@dataclass
+class CallUsage:
+    streams: int = 0
+    connect_attempts: int = 0
+    connect_failures: int = 0
+    connect_retries: int = 0
+    audio_seconds: float = 0.0
+    retry_pending: bool = False  # 上一次连接事实是失败 → 下次建连是 retry
+    _would_retry: bool = False  # check_connect 为本次在途尝试标记的 retry 身份
+
+
+@dataclass
+class _ReplayState:
+    window_audio_seconds: float = 0.0
+    calls: dict[str, CallUsage] = field(default_factory=dict)
+
+
+def summarize_ledger(path: Path | str, window_id: str | None = None) -> dict:
+    """重放账本成汇总（操作员核对口径；不修改账本）.
+
+    window_id 为 None 时按账本里出现过的窗口分组汇总。
+    """
+    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:
+            continue
+        records += 1
+        try:
+            rec = json.loads(line)
+        except ValueError:
+            parse_errors += 1
+            continue
+        wid = str(rec.get("window", ""))
+        if window_id is not None and wid != window_id:
+            continue
+        state = windows.setdefault(wid, _ReplayState())
+        call = str(rec.get("call", ""))
+        cu = state.calls.setdefault(call, CallUsage())
+        kind = rec.get("kind")
+        if kind == "call_start":
+            cu.streams += 1
+        elif kind == "connect_attempt":
+            cu.connect_attempts += 1
+            if rec.get("retry"):
+                cu.connect_retries += 1
+            if not bool(rec.get("ok")):
+                cu.connect_failures += 1
+            cu.retry_pending = not bool(rec.get("ok"))
+        elif kind == "connection_died":
+            cu.connect_failures += 1
+            cu.retry_pending = True
+        elif kind == "audio":
+            seconds = float(rec.get("seconds", 0.0) or 0.0)
+            cu.audio_seconds = round(cu.audio_seconds + seconds, 6)
+            state.window_audio_seconds = round(state.window_audio_seconds + seconds, 6)
+        elif kind == "call_end" and str(rec.get("reason")) == "error":
+            cu.retry_pending = True
+    return {
+        "records": records,
+        "parse_errors": parse_errors,
+        "windows": {
+            wid: {
+                "audio_seconds": round(st.window_audio_seconds, 6),
+                "calls": {
+                    call: {
+                        "streams": cu.streams,
+                        "connect_attempts": cu.connect_attempts,
+                        "connect_failures": cu.connect_failures,
+                        "connect_retries": cu.connect_retries,
+                        "audio_seconds": round(cu.audio_seconds, 6),
+                        "retry_pending": cu.retry_pending,
+                    }
+                    for call, cu in st.calls.items()
+                },
+            }
+            for wid, st in windows.items()
+        },
+    }
+
+
+class AsrUsageGuard:
+    """worker 侧计数与闸（每个 job 一个实例，注入 FunASRSTT）."""
+
+    def __init__(
+        self,
+        policy: AsrWindowPolicy,
+        ledger_path: Path | str | None = None,
+        *,
+        clock=time.time,
+        on_exhausted=None,
+    ) -> None:
+        self.policy = policy
+        self.ledger_path = Path(ledger_path) if ledger_path else None
+        self._clock = clock
+        self._on_exhausted = on_exhausted
+        self._exhausted_fired = False
+        self._unlabeled = 0
+        self._pending: dict[str, _AudioBuf] = {}
+        self.calls: dict[str, CallUsage] = {}
+        self.window_audio_seconds = 0.0
+        self.ledger_errors = 0
+        self._last_resync = -1e18
+        self._replay()
+
+    # -- 账本 ---------------------------------------------------------------
+    def _replay(self) -> None:
+        self._last_resync = self._clock()
+        if self.ledger_path is None or not self.ledger_path.is_file():
+            return
+        state = summarize_ledger(self.ledger_path, self.policy.window_id)["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():
+            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))
+
+    def _maybe_resync(self) -> None:
+        if self.policy.resync_seconds < 0:
+            return
+        if self._clock() - self._last_resync < self.policy.resync_seconds:
+            return
+        self._replay()
+
+    def _append(self, record: dict) -> None:
+        if self.ledger_path is None:
+            return
+        line = json.dumps(
+            {**record, "window": self.policy.window_id, "t": round(self._clock(), 3)},
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+        try:
+            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_errors += 1
+
+    def _block(self, call: str, what: str, reason: str, detail: str = "") -> None:
+        self._append({"kind": "block", "call": call, "what": what, "reason": reason,
+                      "detail": detail})
+        if not self._exhausted_fired:
+            self._exhausted_fired = True
+            if self._on_exhausted is not None:
+                with contextlib.suppress(Exception):
+                    self._on_exhausted(reason)
+        raise AsrWindowClosed(f"asr window stop [{reason}] {what} call={call} {detail}".strip())
+
+    # -- worker 接缝（stt_funasr / worker 真实调用） -------------------------
+    def _pending_total(self) -> float:
+        return round(sum(b.seconds for b in self._pending.values()), 6)
+
+    def begin_stream(self, call: str = "") -> str:
+        """一个 RecognizeStream 开始（worker 按房间名标识；返回生效 call id）."""
+        if not call:
+            self._unlabeled += 1
+            call = f"call-{self._unlabeled}-{secrets.token_hex(3)}"
+        self._maybe_resync()
+        if not self.policy.armed:
+            self._block(call, "new_stream", "window-not-armed")
+        if self.window_audio_seconds + self._pending_total() >= \
+                self.policy.max_audio_seconds - _EPSILON:
+            self._block(call, "new_stream", "window-audio-exhausted")
+        cu = self.calls.setdefault(call, CallUsage())
+        cu.streams += 1
+        self._append({"kind": "call_start", "call": call, "stream_seq": cu.streams})
+        return call
+
+    def check_connect(self, call: str) -> None:
+        """建连前检查（可抛 AsrWindowClosed；之后必须 note_connect_result）."""
+        self._maybe_resync()
+        cu = self.calls.setdefault(call, CallUsage())
+        would_retry = cu.retry_pending
+        if not self.policy.armed:
+            self._block(call, "new_connection", "window-not-armed")
+        if self.window_audio_seconds + self._pending_total() >= \
+                self.policy.max_audio_seconds - _EPSILON:
+            self._block(call, "new_connection", "window-audio-exhausted")
+        if would_retry and cu.connect_retries >= self.policy.max_connect_retries_per_call:
+            self._block(
+                call, "new_connection", "connect-retries-exhausted",
+                detail=f"retries={cu.connect_retries}",
+            )
+        cu._would_retry = would_retry
+
+    def note_connect_result(self, call: str, *, ok: bool, error: str = "") -> None:
+        cu = self.calls.setdefault(call, CallUsage())
+        cu.connect_attempts += 1
+        retry = cu._would_retry
+        if retry:
+            cu.connect_retries += 1
+        if not ok:
+            cu.connect_failures += 1
+        cu.retry_pending = not ok
+        cu._would_retry = False
+        self._append({"kind": "connect_attempt", "call": call, "seq": cu.connect_attempts,
+                      "ok": ok, "retry": retry, "error": (error or "")[:200]})
+
+    def note_connection_died(self, call: str, error: str = "") -> None:
+        """连接接收侧死亡（未伴随建连失败）：计连接失败并置 retry 标记."""
+        cu = self.calls.setdefault(call, CallUsage())
+        cu.connect_failures += 1
+        cu.retry_pending = True
+        self._append({"kind": "connection_died", "call": call, "error": (error or "")[:200]})
+
+    def precheck_audio(self, call: str, sample_rate: int, samples: int) -> None:
+        """发送前检查：这一帧会让窗口越限就在发送前拒绝（帧不越限一个字节）."""
+        self._maybe_resync()
+        seconds = samples / sample_rate
+        if not self.policy.armed:
+            self._block(call, "audio", "window-not-armed")
+        pending = self._pending.get(call)
+        pending_seconds = pending.seconds if pending else 0.0
+        total = self.window_audio_seconds + pending_seconds + seconds
+        if total > self.policy.max_audio_seconds + _EPSILON:
+            self._block(
+                call, "audio", "window-audio-exhausted",
+                detail=(f"window={self.window_audio_seconds:.3f}s pending={pending_seconds:.3f}s "
+                        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 阈值聚合落账."""
+        cu = self.calls.setdefault(call, CallUsage())
+        seconds = round(samples / sample_rate, 6)
+        cu.audio_seconds = round(cu.audio_seconds + seconds, 6)
+        buf = self._pending.setdefault(call, _AudioBuf())
+        buf.seconds = round(buf.seconds + seconds, 6)
+        buf.frames += 1
+        buf.bytes += nbytes
+        buf.sample_rate = sample_rate
+        if buf.seconds >= self.policy.flush_seconds:
+            self.flush_audio(call)
+
+    def flush_audio(self, call: str) -> None:
+        """把该通缓冲中的音频计数聚合落账（utterance 边界 / 流结束时调用）."""
+        buf = self._pending.pop(call, None)
+        if buf is None or buf.seconds <= 0:
+            return
+        self.window_audio_seconds = round(self.window_audio_seconds + buf.seconds, 6)
+        self._append({"kind": "audio", "call": call, "seconds": buf.seconds,
+                      "frames": buf.frames, "bytes": buf.bytes, "sample_rate": buf.sample_rate})
+
+    def stream_end(self, call: str, reason: str) -> None:
+        """一个 RecognizeStream 结束（closed/error/window_exhausted）."""
+        self.flush_audio(call)
+        cu = self.calls.get(call)
+        if cu is not None and reason == "error":
+            cu.retry_pending = True
+        self._append({
+            "kind": "call_end", "call": call, "reason": reason,
+            "audio_seconds": round(cu.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,
+        })
+
+    # -- 观测 ---------------------------------------------------------------
+    def pending_audio_seconds(self, call: str) -> float:
+        buf = self._pending.get(call)
+        return buf.seconds if buf else 0.0
+
+    def summary(self) -> dict:
+        pending = round(sum(b.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_with_pending": round(self.window_audio_seconds + pending, 6),
+            "ledger": str(self.ledger_path) if self.ledger_path else "",
+            "ledger_errors": self.ledger_errors,
+            "calls": {
+                call: {
+                    "streams": cu.streams,
+                    "connect_attempts": cu.connect_attempts,
+                    "connect_failures": cu.connect_failures,
+                    "connect_retries": cu.connect_retries,
+                    "audio_seconds": round(cu.audio_seconds, 6),
+                }
+                for call, cu in self.calls.items()
+            },
+        }
+
+    def summary_line(self) -> str:
+        s = self.summary()
+        return (
+            f"window={s['window'] or 'not-armed'} "
+            f"audio={s['audio_seconds_with_pending']:.3f}/{s['max_audio_seconds']:.3f}s "
+            f"retries<= {s['max_connect_retries_per_call']}/call "
+            f"ledger={s['ledger']}"
+        )
+
+
+def build_usage_from_env(
+    environ: dict | None = None,
+    *,
+    clock=time.time,
+    on_exhausted=None,
+    ledger_path: Path | str | None = None,
+) -> AsrUsageGuard:
+    """worker entrypoint 的唯一构造入口（env 只在窗口启动时由负责人显式设置）."""
+    env = os.environ if environ is None else environ
+    path = ledger_path or env.get("CYBERGRIL_ASR_USAGE_LEDGER") or DEFAULT_ASR_USAGE_LEDGER
+    return AsrUsageGuard(policy_from_env(env), path, clock=clock, on_exhausted=on_exhausted)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        prog="python -m cybergril.voice.asr_usage",
+        description="重放 ASR 用量账本为汇总 JSON（只读，不修改账本）",
+    )
+    parser.add_argument("ledger", help="asr-usage.jsonl 路径")
+    parser.add_argument("--window", default=None, help="只汇总指定 window id")
+    args = parser.parse_args(argv)
+    print(json.dumps(summarize_ledger(args.ledger, args.window), ensure_ascii=False, indent=1))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/cybergril/voice/stt_funasr.py

@@ -11,6 +11,13 @@
   -> FINAL_TRANSCRIPT, 之后为本轮重建 websocket(FunASR 一次会话一句话)
 
 一个 RecognizeStream 对应一个房间会话, websocket 按轮次惰性重建.
+
+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).
 """
 
 from __future__ import annotations
@@ -31,15 +38,21 @@ from livekit.agents.stt import (
 )
 from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS, APIConnectOptions
 
+from .asr_usage import AsrUsageGuard, AsrWindowClosed
+
 FUNASR_WS = "ws://127.0.0.1:10095"
 
 
 class FunASRSTT(stt.STT):
-    def __init__(self, endpoint: str = FUNASR_WS) -> None:
+    def __init__(self, endpoint: str = FUNASR_WS, *,
+                 usage: AsrUsageGuard | None = None,
+                 call_label: str = "") -> None:
         super().__init__(
             capabilities=stt.STTCapabilities(streaming=True, interim_results=True)
         )
         self._endpoint = endpoint
+        self._usage = usage
+        self._call_label = call_label
         # 回声抑制(worker 运行时注入): agent 说话期间, 短文本或与刚说内容
         # 高度相似的识别结果视为音箱回声, 不作为新轮次提交
         self.agent_speaking = False
@@ -75,7 +88,9 @@ class FunASRSTT(stt.STT):
     def stream(
         self, *, language=None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
     ) -> RecognizeStream:
-        return _LoggedRecognizeStream(stt=self, conn_options=conn_options, endpoint=self._endpoint)
+        return _LoggedRecognizeStream(stt=self, conn_options=conn_options,
+                                       endpoint=self._endpoint,
+                                       usage=self._usage, call_label=self._call_label)
 
 
 class _Utterance:
@@ -86,12 +101,16 @@ class _Utterance:
         self.ws: websockets.WebSocketClientProtocol | None = None
         self.closed = False
         self.last_text = ""
+        self.recv_failed = False  # 接收侧死亡(未伴随建连失败), HEL-906 计数用
 
 
 class FunASRRecognizeStream(RecognizeStream):
-    def __init__(self, *, stt: FunASRSTT, conn_options: APIConnectOptions, endpoint: str) -> None:
+    def __init__(self, *, stt: FunASRSTT, conn_options: APIConnectOptions, endpoint: str,
+                 usage: AsrUsageGuard | None = None, call_label: str = "") -> None:
         super().__init__(stt=stt, conn_options=conn_options)
         self._endpoint = endpoint
+        self._usage = usage
+        self._call_label = call_label
 
     def _ev(self, ev_type: SpeechEventType, text: str) -> SpeechEvent:
         return SpeechEvent(
@@ -100,6 +119,23 @@ class FunASRRecognizeStream(RecognizeStream):
         )
 
     async def _run(self) -> None:
+        if self._usage is None:
+            await self._pump(None, "")
+            return
+        call_id = self._usage.begin_stream(self._call_label)
+        end_reason = "closed"
+        try:
+            await self._pump(self._usage, call_id)
+        except AsrWindowClosed:
+            end_reason = "window_exhausted"
+            raise
+        except BaseException:
+            end_reason = "error"
+            raise
+        finally:
+            self._usage.stream_end(call_id, end_reason)
+
+    async def _pump(self, usage: AsrUsageGuard | None, call_id: str) -> None:
         utt: _Utterance | None = None
         recv_task: asyncio.Task | None = None
         finals: asyncio.Queue[str] = asyncio.Queue()
@@ -107,7 +143,17 @@ class FunASRRecognizeStream(RecognizeStream):
         n_frames = 0
 
         async def open_utt() -> _Utterance:
-            ws = await websockets.connect(self._endpoint, max_size=2**24)
+            if usage is not None:
+                usage.check_connect(call_id)
+            try:
+                ws = await websockets.connect(self._endpoint, max_size=2**24)
+            except Exception as exc:
+                if usage is not None:
+                    usage.note_connect_result(
+                        call_id, ok=False, error=f"{type(exc).__name__}: {exc}")
+                raise
+            if usage is not None:
+                usage.note_connect_result(call_id, ok=True)
             cfg = {
                 "mode": "2pass",
                 "wav_name": "livekit",
@@ -117,11 +163,27 @@ class FunASRRecognizeStream(RecognizeStream):
                 "audio_fs": 16000,
                 "itn": True,
             }
-            await ws.send(json.dumps(cfg))
+            try:
+                await ws.send(json.dumps(cfg))
+            except Exception as exc:
+                if usage is not None:
+                    usage.note_connection_died(
+                        call_id, error=f"{type(exc).__name__}: {exc}")
+                raise
             u = _Utterance(self._endpoint)
             u.ws = ws
             return u
 
+        async def send_frame(f) -> None:
+            # HEL-906: 发送前检查(越限帧一个字节都不出站), 发送成功才计数
+            payload = bytes(f.data)
+            if usage is not None:
+                usage.precheck_audio(call_id, f.sample_rate, f.samples_per_channel)
+            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))
+
         async def recv_loop(u: _Utterance) -> None:
             assert u.ws is not None
             pending: list[str] = []  # 去抖合并中的 final 片段
@@ -165,7 +227,7 @@ class FunASRRecognizeStream(RecognizeStream):
                             self._ev(SpeechEventType.INTERIM_TRANSCRIPT, text)
                         )
             except Exception:
-                pass
+                u.recv_failed = True  # 接收侧死亡: HEL-906 计入连接失败事实
 
         try:
             async for ev in self._input_ch:
@@ -195,6 +257,10 @@ class FunASRRecognizeStream(RecognizeStream):
                                 await recv_task
                         with contextlib.suppress(Exception):
                             await utt.ws.close()
+                        if usage is not None:
+                            if utt.recv_failed:
+                                usage.note_connection_died(call_id)
+                            usage.flush_audio(call_id)  # utterance 边界聚合落账
                         utt = None
                     continue
                 if not isinstance(ev, rtc.AudioFrame):
@@ -210,9 +276,9 @@ class FunASRRecognizeStream(RecognizeStream):
                     if resampler is None:
                         resampler = rtc.AudioResampler(frame.sample_rate, 16000)
                     for f in resampler.push(frame):
-                        await utt.ws.send(bytes(f.data))
+                        await send_frame(f)
                 else:
-                    await utt.ws.send(bytes(frame.data))
+                    await send_frame(frame)
         finally:
             if recv_task is not None:
                 recv_task.cancel()

src/cybergril/voice/worker.py

@@ -20,6 +20,14 @@
       见 docs/voice-subtitle-contract-v2.md）。
 音色: 独立 voice_state_cybergril.json；D1 未拍板前回落 default，不替用户定。
 
+HEL-906 连续通话 ASR 用量: entrypoint 按 job 构造 AsrUsageGuard 注入
+FunASRSTT（房间名作为每通标识）。未布防（无 CYBERGRIL_ASR_WINDOW_ID）
+即 fail-closed 拒绝一切 ASR 连接/音频（真实 cap=0）；获批的一次性窗口
+由窗口负责人启动时显式设置环境变量（见 voice/asr_usage.py 与
+docs/rework/hel902/r5-followup-package.md §4）。触线时 guard 在发送前
+拒绝并经 on_exhausted 回调 ctx.shutdown 结束本通 job；逐通/全窗计数落
+asr-usage.jsonl，不随删房丢失。
+
 运行位置: /opt/voice-agent-cybergril/（独立目录，复用 .12 现有 agent venv 与
 LiveKit/FunASR/CosyVoice；专属健康端口 CYBERGRIL_WORKER_HEALTH_PORT——
 livekit-agents worker 内建 HTTP 健康检查，GET / 返回 OK/503）。
@@ -40,6 +48,7 @@ from pathlib import Path
 from livekit.agents import Agent, AgentSession, JobContext, JobProcess, WorkerOptions, cli
 from livekit.plugins.silero import VAD
 
+from .asr_usage import build_usage_from_env
 from .llm_chat_api import CyberGrillLLM
 from .receipt import ReceiptSender, TurnLedger, should_ack
 from .seam import AssistantInterimPublisher, SubtitleBus, pending_proactive
@@ -266,8 +275,19 @@ async def _maybe_open_proactive(session: AgentSession, room, bus: SubtitleBus,
     return spoken_ids
 
 
+def _shutdown_job(ctx, reason: str) -> None:
+    """AsrUsageGuard 触线回调: 结束本通 job（异步 shutdown 兼容为任务）."""
+    print(f"[cybergril-worker] asr window stop: {reason}; ending this call", flush=True)
+    with contextlib.suppress(Exception):
+        result = ctx.shutdown()
+        if inspect.isawaitable(result):
+            asyncio.get_running_loop().create_task(result)
+
+
 async def entrypoint(ctx: JobContext) -> None:
-    stt = FunASRSTT()
+    usage = build_usage_from_env(
+        on_exhausted=lambda reason: _shutdown_job(ctx, reason))
+    stt = FunASRSTT(usage=usage, call_label=getattr(ctx.room, "name", ""))
     # agent 说话期间短文本/相似文本视为回声丢弃（同通用 worker 的实测做法）
     stt.agent_speaking = False
     stt.agent_recent_text = ""
@@ -305,7 +325,8 @@ async def entrypoint(ctx: JobContext) -> None:
               flush=True)
     print(
         f"[cybergril-worker] session started (api={API_BASE}, "
-        f"voice={read_default_voice()}, agent_name={AGENT_NAME})",
+        f"voice={read_default_voice()}, agent_name={AGENT_NAME}, "
+        f"asr_window=({usage.summary_line()}))",
         flush=True,
     )
 

tests/test_voice_asr_usage.py

@@ -0,0 +1,831 @@
+"""HEL-906 连续通话 ASR 实际用量计数与窗口闸的确定性离线测试.
+
+两层验证（复审要求"接缝必须真的被专属 worker 调用，不能只有独立计数类
+测试"）：
+
+1. 纯计数/闸/账本恢复（guard 单元）：fail-closed 默认、按实际帧样本计时、
+   触线前拒绝、重试口径、落盘恢复、跨窗口隔离、崩溃少计有界。
+2. 真实接缝：用假 livekit/websockets 模块（隔离假 provider、假帧）导入
+  真实的 voice/stt_funasr.py 与 voice/worker.py，驱动真实 ``_run`` 送数
+   路径——建连检查、发送前拒绝、发送后计数、接收侧死亡、流异常重试、
+   utterance 聚合落账全部走产品代码；worker entrypoint 布线（guard 注入、
+   房间名标识、触线 ctx.shutdown、未布防拒绝）逐项断言。
+
+零真实 ASR/TTS/LiveKit 调用（fake 连接全部在进程内），零安装。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import enum
+import json
+import subprocess
+import sys
+import types
+from pathlib import Path
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from cybergril.voice.asr_usage import (
+    AsrUsageGuard,
+    AsrWindowClosed,
+    AsrWindowPolicy,
+    build_usage_from_env,
+    policy_from_env,
+    summarize_ledger,
+)
+
+# ---------------------------------------------------------------------------
+# fake livekit / websockets SDK（只满足本仓库代码用到的表面，进程内假 provider）
+# ---------------------------------------------------------------------------
+
+_FAKE_MODULES = (
+    "livekit",
+    "livekit.rtc",
+    "livekit.agents",
+    "livekit.agents.stt",
+    "livekit.agents.llm",
+    "livekit.agents.tts",
+    "livekit.agents.types",
+    "livekit.plugins",
+    "livekit.plugins.silero",
+    "websockets",
+)
+
+
+class FakeFrame:
+    def __init__(self, data: bytes, sample_rate: int, samples_per_channel: int) -> None:
+        self.data = data
+        self.sample_rate = sample_rate
+        self.samples_per_channel = samples_per_channel
+
+
+class FakeResampler:
+    """线性比例重采样替身：输出帧数与输入样本数严格成比例（16bit/单声道）."""
+
+    def __init__(self, in_rate: int, out_rate: int) -> None:
+        self.out_rate = out_rate
+        self.ratio = out_rate / in_rate
+        self._in_samples = 0
+        self._out_samples = 0
+
+    def push(self, frame: FakeFrame) -> list[FakeFrame]:
+        self._in_samples += frame.samples_per_channel
+        target = round(self._in_samples * self.ratio)
+        n = target - self._out_samples
+        if n <= 0:
+            return []
+        self._out_samples = target
+        return [FakeFrame(b"\x00" * (n * 2), self.out_rate, n)]
+
+
+class FakeWS:
+    """进程内 FunASR websocket 假 provider：记录每个出站字节."""
+
+    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
+
+    async def send(self, payload: bytes | str) -> None:
+        if self.fail_send:
+            raise OSError("send refused")
+        self.sent.append(payload)
+
+    async def close(self) -> None:
+        self.closed = True
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        while True:
+            if self.closed:
+                raise StopAsyncIteration
+            if self.recv_fail:
+                raise ConnectionError("recv died")
+            if self.server_msgs:
+                return self.server_msgs.pop(0)
+            await asyncio.sleep(0.005)
+
+
+class FakeWebsockets:
+    """websockets.connect 假 provider：可编程失败与假服务端响应."""
+
+    def __init__(self) -> None:
+        self.attempts = 0
+        self.fail_attempts: set[int] = set()
+        self.sockets: list[FakeWS] = []
+        self.ws_factory = None
+
+    def connect(self, url: str, max_size: int | None = None):
+        async def _connect() -> FakeWS:
+            self.attempts += 1
+            if self.attempts in self.fail_attempts:
+                raise OSError(f"connect refused (attempt {self.attempts})")
+            if self.ws_factory is not None:
+                ws = self.ws_factory()
+            else:
+                ws = FakeWS(server_msgs=[json.dumps(
+                    {"mode": "2pass-offline", "text": "测试终稿"})])
+            self.sockets.append(ws)
+            return ws
+
+        return _connect()
+
+    def bytes_sent(self) -> list[bytes]:
+        return [p for ws in self.sockets for p in ws.sent if isinstance(p, bytes)]
+
+
+class _InCh:
+    def __init__(self) -> None:
+        self.q: asyncio.Queue = asyncio.Queue()
+
+    def push(self, item) -> None:
+        self.q.put_nowait(item)
+
+    def end(self) -> None:
+        self.q.put_nowait(None)
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        return await self.q.get()
+
+
+class _OutCh:
+    def __init__(self) -> None:
+        self.q: asyncio.Queue = asyncio.Queue()
+
+    def send_nowait(self, item) -> None:
+        self.q.put_nowait(item)
+
+    def empty(self) -> bool:
+        return self.q.empty()
+
+    def get_nowait(self):
+        return self.q.get_nowait()
+
+
+class SpeechEventType(enum.Enum):
+    START_OF_SPEECH = "start_of_speech"
+    INTERIM_TRANSCRIPT = "interim_transcript"
+    FINAL_TRANSCRIPT = "final_transcript"
+    END_OF_SPEECH = "end_of_speech"
+
+
+class SpeechData:
+    def __init__(self, *, language: str, text: str) -> None:
+        self.language = language
+        self.text = text
+
+
+class SpeechEvent:
+    def __init__(self, *, type, alternatives) -> None:
+        self.type = type
+        self.alternatives = alternatives
+
+
+def _kwargs_ns(**kwargs):
+    return SimpleNamespace(**kwargs)
+
+
+class FakeRecognizeStream:
+    class _FlushSentinel:
+        pass
+
+    def __init__(self, *, stt=None, conn_options=None) -> None:
+        self._stt = stt
+        self._conn_options = conn_options
+        self._input_ch = _InCh()
+        self._event_ch = _OutCh()
+        self._run_task: asyncio.Task | None = None
+
+    def _ensure_run(self) -> asyncio.Task:
+        if self._run_task is None:
+            self._run_task = asyncio.get_running_loop().create_task(self._run())
+        return self._run_task
+
+    def __aiter__(self):
+        self._ensure_run()
+        return self
+
+    async def __anext__(self):
+        self._ensure_run()
+        while True:
+            if not self._event_ch.empty():
+                return self._event_ch.get_nowait()
+            task = self._run_task
+            if task.done():
+                exc = task.exception()
+                if exc is not None:
+                    raise exc
+                raise StopAsyncIteration
+            await asyncio.sleep(0.005)
+
+
+class FakeAgentSession:
+    last: FakeAgentSession | None = None  # 测试替身的类级记录
+
+    def __init__(self, **kwargs) -> None:
+        self.kwargs = kwargs
+        self.handlers: dict[str, list] = {}
+        self.started = None
+        FakeAgentSession.last = self
+
+    def on(self, ev: str):
+        def deco(fn):
+            self.handlers.setdefault(ev, []).append(fn)
+            return fn
+
+        return deco
+
+    async def start(self, *, room=None, agent=None) -> None:
+        self.started = (room, agent)
+
+
+class FakeCtx:
+    """JobContext 替身：记录 shutdown 调用."""
+
+    def __init__(self, room_name: str) -> None:
+        self.room = SimpleNamespace(
+            name=room_name,
+            local_participant=SimpleNamespace(
+                publish_data=self._publish_data),
+        )
+        self.proc = SimpleNamespace(userdata={"vad": object()})
+        self.shutdown_calls: list[str] = []
+
+    async def _publish_data(self, data, topic=None, reliable=False) -> None:
+        return None
+
+    async def connect(self) -> None:
+        return None
+
+    def shutdown(self, reason: str = "", message: str = "") -> None:
+        self.shutdown_calls.append(reason or message or "shutdown")
+
+
+class _VAD:
+    @classmethod
+    def load(cls):
+        return cls()
+
+
+def _install_fake_sdk(monkeypatch) -> SimpleNamespace:
+    rtc_mod = types.ModuleType("livekit.rtc")
+    rtc_mod.AudioFrame = FakeFrame
+    rtc_mod.AudioResampler = FakeResampler
+
+    stt_mod = types.ModuleType("livekit.agents.stt")
+    stt_mod.STT = type("STT", (), {"__init__": lambda self, *, capabilities=None: setattr(
+        self, "capabilities", capabilities)})
+    stt_mod.STTCapabilities = _kwargs_ns
+    stt_mod.SpeechData = SpeechData
+    stt_mod.SpeechEvent = SpeechEvent
+    stt_mod.SpeechEventType = SpeechEventType
+    stt_mod.RecognizeStream = FakeRecognizeStream
+
+    llm_mod = types.ModuleType("livekit.agents.llm")
+    for name in ("LLM", "ChatChunk", "ChatContext", "ChoiceDelta", "LLMStream", "Tool"):
+        setattr(llm_mod, name, type(name, (), {"__init__": lambda self, *a, **k: None}))
+
+    tts_mod = types.ModuleType("livekit.agents.tts")
+    tts_mod.TTS = type("TTS", (), {"__init__": lambda self, *, capabilities=None,
+                                   sample_rate=None, num_channels=None: None})
+    tts_mod.TTSCapabilities = _kwargs_ns
+    for name in ("AudioEmitter", "ChunkedStream"):
+        setattr(tts_mod, name, type(name, (), {}))
+
+    types_mod = types.ModuleType("livekit.agents.types")
+    types_mod.APIConnectOptions = type("APIConnectOptions", (), {})
+    types_mod.DEFAULT_API_CONNECT_OPTIONS = types_mod.APIConnectOptions()
+    types_mod.NOT_GIVEN = object()
+    types_mod.NotGivenOr = object
+
+    agents_mod = types.ModuleType("livekit.agents")
+    agents_mod.Agent = type("Agent", (), {"__init__": lambda self, *a, **k: None})
+    agents_mod.AgentSession = FakeAgentSession
+    for name in ("JobContext", "JobProcess"):
+        setattr(agents_mod, name, type(name, (), {}))
+    agents_mod.WorkerOptions = type("WorkerOptions", (), {"__init__": lambda self, **k: None})
+    agents_mod.cli = type("cli", (), {"run_app": staticmethod(lambda *a, **k: None)})
+    agents_mod.APIConnectionError = type("APIConnectionError", (Exception,), {})
+    agents_mod.APIStatusError = type("APIStatusError", (Exception,), {})
+    agents_mod.stt = stt_mod
+    agents_mod.llm = llm_mod
+    agents_mod.tts = tts_mod
+    agents_mod.types = types_mod
+
+    silero_mod = types.ModuleType("livekit.plugins.silero")
+    silero_mod.VAD = _VAD
+    plugins_mod = types.ModuleType("livekit.plugins")
+    plugins_mod.silero = silero_mod
+
+    livekit_mod = types.ModuleType("livekit")
+    livekit_mod.rtc = rtc_mod
+    livekit_mod.agents = agents_mod
+    livekit_mod.plugins = plugins_mod
+
+    websockets_mod = types.ModuleType("websockets")
+    fake_ws_module = FakeWebsockets()
+    websockets_mod.connect = fake_ws_module.connect
+
+    for name, mod in (
+        ("livekit", livekit_mod),
+        ("livekit.rtc", rtc_mod),
+        ("livekit.agents", agents_mod),
+        ("livekit.agents.stt", stt_mod),
+        ("livekit.agents.llm", llm_mod),
+        ("livekit.agents.tts", tts_mod),
+        ("livekit.agents.types", types_mod),
+        ("livekit.plugins", plugins_mod),
+        ("livekit.plugins.silero", silero_mod),
+        ("websockets", websockets_mod),
+    ):
+        monkeypatch.setitem(sys.modules, name, mod)
+    for cached in ("cybergril.voice.stt_funasr", "cybergril.voice.worker",
+                   "cybergril.voice.tts_cosyvoice", "cybergril.voice.llm_chat_api"):
+        monkeypatch.delitem(sys.modules, cached, raising=False)
+    return SimpleNamespace(websockets=fake_ws_module, rtc=rtc_mod, stt=stt_mod)
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+def make_policy(**overrides) -> AsrWindowPolicy:
+    values = dict(window_id="w-test", max_audio_seconds=1000.0,
+                  max_connect_retries_per_call=6, flush_seconds=5.0,
+                  resync_seconds=1e9)
+    values.update(overrides)
+    return AsrWindowPolicy(**values)
+
+
+def make_guard(tmp_path: Path, *, on_exhausted=None, **overrides) -> AsrUsageGuard:
+    return AsrUsageGuard(make_policy(**overrides), tmp_path / "asr-usage.jsonl",
+                         on_exhausted=on_exhausted)
+
+
+def frame_16k(n_samples: int = 160) -> FakeFrame:
+    return FakeFrame(b"\x00" * (n_samples * 2), 16000, n_samples)
+
+
+def frame_48k(n_samples: int = 480) -> FakeFrame:
+    return FakeFrame(b"\x00" * (n_samples * 2), 48000, n_samples)
+
+
+def read_records(tmp_path: Path) -> list[dict]:
+    path = tmp_path / "asr-usage.jsonl"
+    if not path.exists():
+        return []
+    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()
+            if line.strip()]
+
+
+async def drive(module, stt, script, *, timeout: float = 10.0):
+    """按脚本驱动真实 RecognizeStream（帧/flush/None=结束），等待跑完."""
+    stream = stt.stream()
+    task = stream._ensure_run()
+    for item in script:
+        if item is None:
+            stream._input_ch.end()
+        elif isinstance(item, str) and item == "flush":
+            stream._input_ch.push(module.RecognizeStream._FlushSentinel())
+        else:
+            stream._input_ch.push(item)
+        await asyncio.sleep(0)
+    await asyncio.wait_for(task, timeout=timeout)
+    return stream
+
+
+# ---------------------------------------------------------------------------
+# 1) guard 单元：计数口径、fail-closed、触线、恢复、隔离
+# ---------------------------------------------------------------------------
+
+def test_policy_env_armed_and_fail_closed_default():
+    assert policy_from_env({}).armed is False  # 未布防 = 真实 cap=0
+    policy = policy_from_env({
+        "CYBERGRIL_ASR_WINDOW_ID": "hel902-w1",
+        "CYBERGRIL_ASR_WINDOW_MAX_AUDIO_SECONDS": "2700",
+        "CYBERGRIL_ASR_MAX_CONNECT_RETRIES_PER_CALL": "6",
+    })
+    assert policy.armed
+    assert policy.max_audio_seconds == 2700.0
+    assert policy.max_connect_retries_per_call == 6
+    # 配置损坏按 fail-closed（0）处理，不放宽
+    bad = policy_from_env({"CYBERGRIL_ASR_WINDOW_ID": "w",
+                           "CYBERGRIL_ASR_WINDOW_MAX_AUDIO_SECONDS": "abc"})
+    assert bad.max_audio_seconds == 0.0
+
+
+def test_unarmed_guard_blocks_everything_and_records(tmp_path):
+    guard = make_guard(tmp_path, window_id="")
+    with pytest.raises(AsrWindowClosed):
+        guard.begin_stream("room-a")
+    with pytest.raises(AsrWindowClosed):
+        guard.check_connect("room-a")
+    with pytest.raises(AsrWindowClosed):
+        guard.precheck_audio("room-a", 16000, 160)
+    blocks = [r for r in read_records(tmp_path) if r["kind"] == "block"]
+    assert {(b["what"], b["reason"]) for b in blocks} == {
+        ("new_stream", "window-not-armed"),
+        ("new_connection", "window-not-armed"),
+        ("audio", "window-not-armed"),
+    }
+
+
+def test_guard_audio_counted_by_actual_frame_samples(tmp_path):
+    guard = make_guard(tmp_path)
+    call = guard.begin_stream("room-a")
+    guard.check_connect(call)
+    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)
+    for _ in range(5):
+        guard.precheck_audio(call, 16000, 160)
+        guard.note_audio_sent(call, 16000, 160, 320)  # 重发按重复计
+    guard.stream_end(call, "closed")
+    assert guard.calls[call].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)
+
+
+def test_guard_blocks_crossing_frame_before_send(tmp_path):
+    stops: list[str] = []
+    guard = make_guard(tmp_path, max_audio_seconds=2.0, on_exhausted=stops.append)
+    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.flush_audio(call)
+    guard.precheck_audio(call, 16000, 16000)  # 恰好等于上限：允许（≤上限）
+    guard.note_audio_sent(call, 16000, 16000, 32000)
+    guard.flush_audio(call)
+    with pytest.raises(AsrWindowClosed):  # 第 3 秒会在发送前拒绝
+        guard.precheck_audio(call, 16000, 16000)
+    assert stops == ["window-audio-exhausted"]  # 回调恰好一次
+    # 同窗新通也立即拒绝（新连接/新音频都不放行）
+    with pytest.raises(AsrWindowClosed):
+        guard.begin_stream("room-b")
+    blocks = [r for r in read_records(tmp_path) if r["kind"] == "block"]
+    assert {b["reason"] for b in blocks} == {"window-audio-exhausted"}
+    assert guard.window_audio_seconds == pytest.approx(2.0)  # 一个字节没多
+
+
+def test_guard_retry_semantics_only_after_failure(tmp_path):
+    guard = make_guard(tmp_path, max_connect_retries_per_call=2)
+    call = guard.begin_stream("room-a")
+
+    guard.check_connect(call)  # 初次建连不是 retry
+    guard.note_connect_result(call, ok=True)
+    guard.check_connect(call)  # 正常说完一轮后的下一次建连也不是 retry
+    guard.note_connect_result(call, ok=True)
+    assert guard.calls[call].connect_retries == 0
+
+    guard.check_connect(call)  # 失败 → retry_pending
+    guard.note_connect_result(call, ok=False, error="OSError: refused")
+    guard.check_connect(call)  # 失败后的再建连 = retry #1
+    guard.note_connect_result(call, ok=True)
+    guard.check_connect(call)  # 再次失败 → retry_pending
+    guard.note_connect_result(call, ok=False, error="OSError: refused")
+    guard.check_connect(call)  # retry #2（用尽上限）
+    guard.note_connect_result(call, ok=True)
+    cu = guard.calls[call]
+    assert (cu.connect_attempts, cu.connect_failures, cu.connect_retries) == (6, 2, 2)
+
+    guard.check_connect(call)  # 上次成功 → 非 retry：不受 retry 上限约束
+    guard.note_connect_result(call, ok=True)
+    guard.check_connect(call)  # 失败 → 置 retry_pending
+    guard.note_connect_result(call, ok=False, error="OSError: refused")
+    with pytest.raises(AsrWindowClosed):  # retry #3 在建连前被拒
+        guard.check_connect(call)
+    cu = guard.calls[call]
+    assert (cu.connect_attempts, cu.connect_failures, cu.connect_retries) == (8, 3, 2)
+
+    # 新的一通有独立 retry 预算（每通口径）
+    call2 = guard.begin_stream("room-b")
+    guard.check_connect(call2)
+    guard.note_connect_result(call2, ok=True)
+    assert guard.calls["room-b"].connect_attempts == 1
+
+
+def test_guard_ledger_recovery_same_window_no_reset(tmp_path):
+    guard1 = make_guard(tmp_path, max_audio_seconds=3.0)
+    call = guard1.begin_stream("room-a")
+    guard1.check_connect(call)
+    guard1.note_connect_result(call, ok=True)
+    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.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)
+    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.retry_pending is True  # 最后一次连接事实是失败
+    # 耗尽同窗口剩余预算后拒绝；重开同窗口 id 不能重置
+    call = guard2.begin_stream("room-a")
+    guard2.check_connect(call)
+    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.flush_audio(call)
+
+
+def test_guard_window_isolation_does_not_stop_others(tmp_path):
+    """W1 耗尽不误停别的窗口/会话（一次性窗口各有独立预算）."""
+    g1 = make_guard(tmp_path, window_id="w1", max_audio_seconds=0.05)
+    with pytest.raises(AsrWindowClosed):
+        call = g1.begin_stream("room-a")
+        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)
+    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.stream_end(call, "closed")
+    assert g2.calls["room-a"].audio_seconds == pytest.approx(0.1)
+
+
+def test_guard_crash_undercount_bounded_by_flush(tmp_path):
+    """如实边界：崩溃前 ≤flush 阈值的已发音频未落账（有界少计）."""
+    guard = make_guard(tmp_path, flush_seconds=5.0)
+    call = guard.begin_stream("room-a")
+    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 阈值
+    # 模拟进程直接消失（没有 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.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)
+
+
+def test_summarize_cli_reproducible(tmp_path, monkeypatch):
+    guard = make_guard(tmp_path)
+    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.stream_end(call, "closed")
+    repo_root = Path(__file__).resolve().parents[1]
+    proc = subprocess.run(
+        [sys.executable, "-m", "cybergril.voice.asr_usage",
+         str(tmp_path / "asr-usage.jsonl"), "--window", "w-test"],
+        capture_output=True, text=True, timeout=60,
+        env={"PYTHONPATH": str(repo_root / "src"), "PATH": "/usr/bin:/bin"},
+        check=True,
+    )
+    out = json.loads(proc.stdout)
+    call_summary = out["windows"]["w-test"]["calls"]["room-a"]
+    assert call_summary["audio_seconds"] == pytest.approx(1.0)
+    assert call_summary["connect_attempts"] == 1
+    assert out["parse_errors"] == 0
+
+
+# ---------------------------------------------------------------------------
+# 2) 真实接缝：stt_funasr._run 在假 provider/假帧上走完整送数路径
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def fake_sdk(monkeypatch):
+    return _install_fake_sdk(monkeypatch)
+
+
+async def test_seam_counts_fed_audio_both_sample_rates(fake_sdk, tmp_path):
+    import cybergril.voice.stt_funasr as funasr
+
+    guard = make_guard(tmp_path)
+    stt = funasr.FunASRSTT(usage=guard, call_label="cg-voice-abc")
+    await drive(funasr, stt, [frame_16k() for _ in range(30)] + ["flush"]
+                + [frame_48k() for _ in range(30)] + [None])
+    cu = guard.calls["cg-voice-abc"]
+    # 16k 直接路径 0.3s + 48k→16k 重采样路径 0.3s（按实际输出帧计时）
+    assert cu.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 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"]
+    # 假 provider 侧：每帧恰好一次出站，cfg/is_speaking 各一条文本
+    audio_payloads = fake_sdk.websockets.bytes_sent()
+    assert len(audio_payloads) == 60
+    assert all(len(p) == 320 for p in audio_payloads)
+
+
+async def test_seam_silence_and_cancel_counted(fake_sdk, tmp_path):
+    import cybergril.voice.stt_funasr as funasr
+
+    guard = make_guard(tmp_path)
+    stt = funasr.FunASRSTT(usage=guard, call_label="room-silent")
+    # 静音数据帧只要真实送出就计数；不 flush 直接结束（用户取消/挂断）
+    silence = FakeFrame(b"\x00" * 320, 16000, 160)
+    await drive(funasr, stt, [silence] * 20 + [None])
+    cu = guard.calls["room-silent"]
+    assert cu.audio_seconds == pytest.approx(0.2)
+    records = read_records(tmp_path)
+    assert [r["reason"] for r in records if r["kind"] == "call_end"] == ["closed"]
+    # 反复结束：第二通历史不覆盖第一通
+    stt2 = funasr.FunASRSTT(usage=guard, call_label="room-silent-2")
+    await drive(funasr, stt2, [frame_16k(1600)] + [None])  # 0.1s
+    assert guard.calls["room-silent-2"].audio_seconds == pytest.approx(0.1)
+    assert len([r for r in read_records(tmp_path)
+                if r["kind"] == "call_start"]) == 2
+
+
+async def test_seam_connect_failure_marks_error_and_retry_denied(fake_sdk, tmp_path):
+    import cybergril.voice.stt_funasr as funasr
+
+    guard = make_guard(tmp_path, max_connect_retries_per_call=2)
+    stt = funasr.FunASRSTT(usage=guard, call_label="room-flaky")
+    fake_sdk.websockets.fail_attempts = {1, 3, 4}
+
+    with pytest.raises(OSError):  # 初次建连失败 → 流异常结束
+        await drive(funasr, stt, [frame_16k(), None])
+    records = read_records(tmp_path)
+    assert records[-1]["kind"] == "call_end" and records[-1]["reason"] == "error"
+
+    stt2 = funasr.FunASRSTT(usage=guard, call_label="room-flaky")
+    await drive(funasr, stt2, [frame_16k(), "flush", None])  # retry#1 成功
+    assert guard.calls["room-flaky"].connect_retries == 1
+
+    stt3 = funasr.FunASRSTT(usage=guard, call_label="room-flaky")
+    with pytest.raises(OSError):  # 上次成功后这次是普通建连，但连接失败
+        await drive(funasr, stt3, [frame_16k(), None])
+
+    stt4 = funasr.FunASRSTT(usage=guard, call_label="room-flaky")
+    with pytest.raises(OSError):  # retry#2（用尽上限）且仍失败
+        await drive(funasr, stt4, [frame_16k(), None])
+
+    stt5 = funasr.FunASRSTT(usage=guard, call_label="room-flaky")
+    attempts_before = fake_sdk.websockets.attempts
+    with pytest.raises(AsrWindowClosed):  # retry#3 建连前被拒：连假 provider 都不碰
+        await drive(funasr, stt5, [frame_16k(), None])
+    assert fake_sdk.websockets.attempts == attempts_before
+    blocks = [r for r in read_records(tmp_path) if r["kind"] == "block"]
+    assert blocks[-1]["reason"] == "connect-retries-exhausted"
+    cu = guard.calls["room-flaky"]
+    assert (cu.connect_attempts, cu.connect_failures, cu.connect_retries) == (4, 3, 2)
+
+    stt5 = funasr.FunASRSTT(usage=guard, call_label="room-flaky")
+    with pytest.raises(AsrWindowClosed):  # 该通 retry 预算已用尽：再建连被拒
+        await drive(funasr, stt5, [frame_16k(), None])
+    # 新的一通有独立 retry 预算：两次正常 utterance 建连都放行
+    stt6 = funasr.FunASRSTT(usage=guard, call_label="room-flaky-2")
+    await drive(funasr, stt6, [frame_16k(), "flush", frame_16k(), "flush", None])
+    assert guard.calls["room-flaky-2"].connect_attempts == 2
+    assert guard.calls["room-flaky-2"].connect_retries == 0
+
+
+async def test_seam_window_exhaustion_stops_before_crossing(fake_sdk, tmp_path):
+    import cybergril.voice.stt_funasr as funasr
+
+    stops: list[str] = []
+    guard = AsrUsageGuard(
+        make_policy(max_audio_seconds=0.05),
+        tmp_path / "asr-usage.jsonl", on_exhausted=stops.append)
+    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 帧
+    assert len(fake_sdk.websockets.bytes_sent()) == 5
+    assert guard.calls["room-cap"].audio_seconds == pytest.approx(0.05)
+    assert stops == ["window-audio-exhausted"]
+    records = read_records(tmp_path)
+    assert records[-1]["kind"] == "call_end"
+    assert records[-1]["reason"] == "window_exhausted"
+    assert any(r["kind"] == "block" for r in records)
+    # 本窗口后续会话：新流在建流即被拒
+    stt2 = funasr.FunASRSTT(usage=guard, call_label="room-cap-2")
+    with pytest.raises(AsrWindowClosed):
+        await drive(funasr, stt2, [frame_16k(), None])
+
+
+async def test_seam_recv_death_recorded_as_connection_died(fake_sdk, tmp_path):
+    import cybergril.voice.stt_funasr as funasr
+
+    guard = make_guard(tmp_path)
+    fake_sdk.websockets.ws_factory = lambda: FakeWS(recv_fail=True)
+    stt = funasr.FunASRSTT(usage=guard, call_label="room-dead")
+    await drive(funasr, stt, [frame_16k(), "flush", None], timeout=6)
+    records = read_records(tmp_path)
+    died = [r for r in records if r["kind"] == "connection_died"]
+    assert len(died) == 1 and died[0]["call"] == "room-dead"
+    assert guard.calls["room-dead"].connect_failures == 1
+    assert guard.calls["room-dead"].retry_pending is True
+    # 死亡后的下一次建连计为 retry（接收侧死亡未伴随建连失败也如实计数）
+    fake_sdk.websockets.ws_factory = None
+    stt2 = funasr.FunASRSTT(usage=guard, call_label="room-dead")
+    await drive(funasr, stt2, [frame_16k(), "flush", None])
+    assert guard.calls["room-dead"].connect_retries == 1
+
+
+async def test_ledger_survives_room_deletion(fake_sdk, tmp_path):
+    """删房成功会话账本移除该通；ASR 账本逐通历史原样保留."""
+    import cybergril.voice.stt_funasr as funasr
+    from cybergril.voice.session import VoiceConfig, VoiceSessionService
+
+    guard = make_guard(tmp_path)
+    stt = funasr.FunASRSTT(usage=guard, call_label="cg-voice-keep")
+    await drive(funasr, stt, [frame_16k(1600), "flush", None])
+    before = read_records(tmp_path)
+
+    service = VoiceSessionService(
+        config=VoiceConfig(livekit_url="ws://v.test", livekit_internal="http://v.test",
+                           api_key="devkey", api_secret="s"),
+        http=httpx.AsyncClient(transport=httpx.MockTransport(_fake_twirp)))
+    session = await service.open_session(epoch=1)
+    ended = await service.end_session(session.session_id)
+    assert ended is not None
+    assert service.get(session.session_id) is None  # 会话账本已移除
+    assert read_records(tmp_path) == before  # ASR 账本一字未动
+    await service.aclose()
+
+
+def _fake_twirp(request: httpx.Request) -> httpx.Response:
+    if request.url.path.endswith("CreateDispatch"):
+        return httpx.Response(200, json={"id": "AD_1"})
+    return httpx.Response(200, content=b"")
+
+
+# ---------------------------------------------------------------------------
+# 3) worker 接线：entrypoint 真的构造并注入 guard，触线真的关 job
+# ---------------------------------------------------------------------------
+
+async def test_worker_entrypoint_wires_guard(fake_sdk, tmp_path, monkeypatch, capsys):
+    import cybergril.voice.worker as worker
+
+    ledger = tmp_path / "worker-asr.jsonl"
+    env = {"CYBERGRIL_ASR_WINDOW_ID": "hel902-w1",
+           "CYBERGRIL_ASR_WINDOW_MAX_AUDIO_SECONDS": "2700",
+           "CYBERGRIL_ASR_MAX_CONNECT_RETRIES_PER_CALL": "6",
+           "CYBERGRIL_ASR_USAGE_LEDGER": str(ledger)}
+    monkeypatch.setattr(worker, "build_usage_from_env",
+                        lambda on_exhausted=None: build_usage_from_env(
+                            env, on_exhausted=on_exhausted, ledger_path=ledger))
+    ctx = FakeCtx("cg-voice-room1")
+    await worker.entrypoint(ctx)
+    captured = FakeAgentSession.last
+    assert captured is not None and captured.started is not None
+    stt_obj = captured.kwargs["stt"]
+    assert stt_obj._call_label == "cg-voice-room1"  # 房间名 = 每通标识
+    guard = stt_obj._usage
+    assert isinstance(guard, AsrUsageGuard)
+    assert guard.policy.window_id == "hel902-w1"
+    out = capsys.readouterr().out
+    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.flush_audio(call)
+    with pytest.raises(AsrWindowClosed):
+        guard.precheck_audio(call, 16000, 16000 * 2700)  # 会越 2700s 上限的帧
+    assert ctx.shutdown_calls, "on_exhausted 必须真的调用 ctx.shutdown 结束本通"
+
+
+async def test_worker_entrypoint_default_is_fail_closed(fake_sdk, monkeypatch, capsys):
+    import cybergril.voice.worker as worker
+
+    monkeypatch.setattr(
+        worker, "build_usage_from_env",
+        lambda on_exhausted=None: build_usage_from_env({}, on_exhausted=on_exhausted,
+                                                       ledger_path=None))
+    ctx = FakeCtx("cg-voice-room2")
+    await worker.entrypoint(ctx)
+    guard = FakeAgentSession.last.kwargs["stt"]._usage
+    assert guard.policy.armed is False  # 未布防 = 真实 cap=0
+    with pytest.raises(AsrWindowClosed):
+        guard.begin_stream("cg-voice-room2")
+    assert "asr_window=" in capsys.readouterr().out