diff --git a/src/cybergril/voice/asr_usage.py b/src/cybergril/voice/asr_usage.py index 5ffb22f..7e015ae 100644 --- a/src/cybergril/voice/asr_usage.py +++ b/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)独立 于实例侧会话账本 ``-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( diff --git a/src/cybergril/voice/stt_funasr.py b/src/cybergril/voice/stt_funasr.py index d2c04dc..418d188 100644 --- a/src/cybergril/voice/stt_funasr.py +++ b/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 diff --git a/tests/test_voice_asr_usage.py b/tests/test_voice_asr_usage.py index 451a1f8..c8504fc 100644 --- a/tests/test_voice_asr_usage.py +++ b/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 帧出站