From e59b20f60fff034a7268febd3348b15fc767edc4 Mon Sep 17 00:00:00 2001
From: cai <nb666@nbcai.cc>
Date: Sun, 28 Jun 2026 16:18:44 +0800
Subject: [PATCH] feat: support streaming livekit helper
---
.env.example | 7
docs/runtime-contract.md | 206 ++
.cargo/config.toml | 16
src/audio.rs | 1037 +++++++---
.dockerignore | 3
.gitignore | 1
tools/validate-turn-stream-fixture.mjs | 147 +
generate-local-fixture.sh | 32
src/main.rs | 2646 ++++++++++++++++++++++++++-
Cargo.lock | 17
README.md | 259 ++
Dockerfile | 38
fixtures/turn-stream-mp3-chunks.ndjson | 6
run-local.sh | 387 ++-
Cargo.toml | 12
fixtures/turn-stream-happy.ndjson | 5
src/service.rs | 908 +++++++++
17 files changed, 5,021 insertions(+), 706 deletions(-)
diff --git a/.cargo/config.toml b/.cargo/config.toml
index 28aeac3..dfba410 100644
--- a/.cargo/config.toml
+++ b/.cargo/config.toml
@@ -1,8 +1,8 @@
-[source.crates-io]
-replace-with = "rsproxy"
-
-[source.rsproxy]
-registry = "sparse+https://rsproxy.cn/index/"
-
-[net]
-retry = 5
+[source.crates-io]
+replace-with = "rsproxy"
+
+[source.rsproxy]
+registry = "sparse+https://rsproxy.cn/index/"
+
+[net]
+retry = 5
diff --git a/.dockerignore b/.dockerignore
index 1f5535f..7daf6c2 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -1,5 +1,2 @@
target/
.local/
-.tmp/
-.git/
-.env
diff --git a/.env.example b/.env.example
index 8b02633..19f0f6f 100644
--- a/.env.example
+++ b/.env.example
@@ -27,3 +27,10 @@
# Local wrapper behavior.
COMBRABO_VOICE_RUNTIME_HELPER_MODE=auto
COMBRABO_VOICE_RUNTIME_HELPER_IMAGE=combrabo-voice-runtime-helper:local
+
+# Service mode control plane. Values below are placeholders only.
+# lmrobot-app calls helper with Authorization: Bearer {CV_HELPER_AUTH_TOKEN}.
+CV_HELPER_SERVICE_ENABLED=false
+CV_HELPER_SERVICE_BIND=127.0.0.1:18080
+CV_HELPER_AUTH_TOKEN=replace-with-local-helper-token
+CV_RUNTIME_TURN_BRIDGE_TOKEN=replace-with-local-turn-bridge-token
diff --git a/.gitignore b/.gitignore
index 7074eb8..1b373d3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,2 @@
.local/
-.tmp/
target/
diff --git a/Cargo.lock b/Cargo.lock
index b5e60c6..03c9ec2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -231,6 +231,8 @@
version = "0.1.0"
dependencies = [
"anyhow",
+ "base64 0.22.1",
+ "futures-util",
"hound",
"libwebrtc",
"livekit",
@@ -2296,12 +2298,14 @@
"sync_wrapper",
"tokio",
"tokio-rustls",
+ "tokio-util",
"tower",
"tower-http",
"tower-service",
"url",
"wasm-bindgen",
"wasm-bindgen-futures",
+ "wasm-streams",
"web-sys",
"webpki-roots",
]
@@ -3314,6 +3318,19 @@
]
[[package]]
+name = "wasm-streams"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65"
+dependencies = [
+ "futures-util",
+ "js-sys",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
+[[package]]
name = "wasmparser"
version = "0.244.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index a2b6ff9..a6d823d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -4,13 +4,15 @@
edition = "2024"
license = "MIT"
-[dependencies]
-anyhow = "1.0.100"
-hound = "3.5.1"
-libwebrtc = "0.3.36"
+[dependencies]
+anyhow = "1.0.100"
+base64 = "0.22.1"
+futures-util = "0.3.31"
+hound = "3.5.1"
+libwebrtc = "0.3.36"
livekit = "0.7.45"
minimp3 = "0.5.1"
-reqwest = { version = "0.12.24", default-features = false, features = ["rustls-tls"] }
+reqwest = { version = "0.12.24", default-features = false, features = ["rustls-tls", "json", "stream"] }
serde = { version = "1.0.228", features = ["derive"] }
serde_json = "1.0.145"
tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "signal", "time"] }
diff --git a/Dockerfile b/Dockerfile
index 3d55983..7024ec5 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,21 +1,21 @@
-# syntax=docker/dockerfile:1.7
-
-FROM rust:1.88-bookworm AS builder
-
-WORKDIR /app
-ENV CARGO_REGISTRIES_CRATES_IO_PROTOCOL=sparse
-ENV CARGO_HTTP_TIMEOUT=120
-ENV CARGO_NET_RETRY=5
-
-COPY Cargo.toml Cargo.lock ./
-COPY .cargo ./.cargo
-COPY src ./src
-
-RUN --mount=type=cache,target=/usr/local/cargo/registry \
- --mount=type=cache,target=/app/target \
- cargo fetch --locked \
- && cargo build --release --locked \
- && cp /app/target/release/combrabo-voice-runtime-helper /usr/local/bin/combrabo-voice-runtime-helper
+# syntax=docker/dockerfile:1.7
+
+FROM rust:1.88-bookworm AS builder
+
+WORKDIR /app
+ENV CARGO_REGISTRIES_CRATES_IO_PROTOCOL=sparse
+ENV CARGO_HTTP_TIMEOUT=120
+ENV CARGO_NET_RETRY=5
+
+COPY Cargo.toml Cargo.lock ./
+COPY .cargo ./.cargo
+COPY src ./src
+
+RUN --mount=type=cache,target=/usr/local/cargo/registry \
+ --mount=type=cache,target=/app/target \
+ cargo fetch --locked \
+ && cargo build --release --locked \
+ && cp /app/target/release/combrabo-voice-runtime-helper /usr/local/bin/combrabo-voice-runtime-helper
FROM debian:bookworm-slim
@@ -23,6 +23,6 @@
&& apt-get install -y --no-install-recommends ca-certificates libstdc++6 libgcc-s1 libglib2.0-0 \
&& rm -rf /var/lib/apt/lists/*
-COPY --from=builder /usr/local/bin/combrabo-voice-runtime-helper /usr/local/bin/combrabo-voice-runtime-helper
+COPY --from=builder /usr/local/bin/combrabo-voice-runtime-helper /usr/local/bin/combrabo-voice-runtime-helper
ENTRYPOINT ["/usr/local/bin/combrabo-voice-runtime-helper"]
diff --git a/README.md b/README.md
index 020355c..9ea88e6 100644
--- a/README.md
+++ b/README.md
@@ -1,100 +1,269 @@
-# lm-livekit-helper
+# Combrabo Voice Runtime Helper
-`lm-livekit-helper` 是 Combrabo Voice 的 LiveKit 媒体 worker。它从 `lmrobot-app` 接收每次通话的 `CV_*` 运行参数，加入 LiveKit 房间，发布 bot 音轨，播放主动问候音频，并在 S7 阶段承接用户语音 turn 的媒体输入输出。
+这是 `lmrobot-app` Combrabo Voice 一阶段的最小 external runtime helper。它的长期定位是 LiveKit media worker：连接房间、发布 bot 音轨、订阅用户音轨、搬运 / 观测 PCM、输出脱敏诊断。
-它不是业务服务，不承担鉴权、数据库、订单、ASR/LLM/TTS 权威、消息落库或 diagnostics 聚合。业务权威仍在 `lmrobot-app`。
+当前职责分两层：
-## 当前职责
+一阶段固定主动问候职责有 4 件事：
-1. 读取 `lmrobot-app` 注入的 `CV_*` 环境变量；
-2. 使用 bot token 连接 LiveKit room；
+1. 读取 `CombraboVoiceRuntimeServiceImpl` 注入的 `CV_*` 环境变量；
+2. 使用 bot token 连接 local LiveKit room；
3. 发布 bot 本地音轨；
-4. 播放 WAV / MP3 主动问候音频；
-5. 在 smoke 模式下通过 LiveKit reliable Data Message 下发 `device_output`；
-6. S7 阶段通过 Java runtime turn bridge 复用后端 ASR / TextChat / TTS / Message 能力。
+4. 播放一段固定问候音频，并保持连接直到 `calls/end` 触发 stop。
+
+2026-06-24 起进入 helper 下一阶段入口：默认开启用户上行音频观测，先做远端用户音轨订阅和音频帧摘要日志，不做 ASR/LLM/TTS。
+
+它**不承担**鉴权、数据库、订单、旧 TRTC、角色 / 提示词、正式 ASR/LLM/TTS 编排、消息写入、计费或 diagnostics 聚合。上述业务能力继续由 `lmrobot-app` Java 后端复用现有体系承接。
+
+本定位已按 `cb-sdk` 真实链路校准：服务端 worker 负责 LiveKit 用户音频输入与 bot 音频输出，后端 speech-runtime 负责 ASR / Agent / TTS / turn 状态。`lmrobot` 当前 helper 只对齐 media worker / rtc 边界，不新建第二套业务后端。
## 本机构建
-```bash
-cargo build --release
+helper 编译固定分为“开发调试线”和“部署验证线”。详细口径见：
+
+```text
+doc/task/202606/0615-nativesdk-combrabo-voice-migration/29-helper编译与验证线路说明.md
```
-helper 目录内带 `.cargo/config.toml`，本机构建建议通过本仓库根目录执行，确保使用同一套 registry / retry 配置。
+本机开发调试优先复用已下载的 LiveKit WebRTC 预编译缓存：
+
+```bash
+export LK_CUSTOM_WEBRTC="$HOME/.cache/combrabo/livekit-webrtc/mac-arm64-release-webrtc-51ef663"
+```
+
+然后执行：
+
+```bash
+cargo fmt --manifest-path tools/combrabo-voice-runtime-helper/Cargo.toml --check
+cargo check --manifest-path tools/combrabo-voice-runtime-helper/Cargo.toml
+cargo test --manifest-path tools/combrabo-voice-runtime-helper/Cargo.toml
+```
+
+`LK_CUSTOM_WEBRTC` 指向的是 macOS host 开发调试缓存，不能直接用于 Linux Docker 镜像构建。
+
+```bash
+cargo build --manifest-path tools/combrabo-voice-runtime-helper/Cargo.toml --release
+```
+
+helper 目录内带 `.cargo/config.toml`，本机构建建议通过 `run-local.sh` 或进入 helper 目录执行，确保使用同一套 registry / retry 配置。
+
+## Turn stream 快速校验
+
+TTS streaming 相关改动优先跑一条不依赖 LiveKit / WebRTC native 编译链路的快速线：
+
+```bash
+node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-happy.ndjson
+node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-mp3-chunks.ndjson
+```
+
+这条快速线只校验 NDJSON contract、事件顺序和 `pcm_s16le` / `mp3` chunk 基本约束，用于提前发现 `replyPlaybackMode`、`reply_state`、`reply_audio_chunk`、`turn_completed` 等字段破坏。它不能替代 Docker 镜像构建、真实 LiveKit smoke 或 iPhone 真机验收。
## 本机运行
推荐通过包装脚本启动：
```bash
-./run-local.sh
+./tools/combrabo-voice-runtime-helper/run-local.sh
```
-如果本地还没有 release 二进制，脚本会先执行构建。macOS 默认优先使用 Docker 模式，避免 host 侧 WebRTC native 依赖链路反复阻塞。
+如果本地还没有 release 二进制，脚本会先执行一次构建。
-可预先准备运行模式：
+开发调试时建议显式带上本机 WebRTC 缓存：
```bash
-./run-local.sh --prepare
-./run-local.sh --prepare --rebuild
+LK_CUSTOM_WEBRTC="$HOME/.cache/combrabo/livekit-webrtc/mac-arm64-release-webrtc-51ef663" \
+COMBRABO_VOICE_RUNTIME_HELPER_MODE=host \
+./tools/combrabo-voice-runtime-helper/run-local.sh --prepare
```
+
+## Service 模式
+
+`lmrobot-app` 线上形态不再适合每次通话直接用命令行拉起 helper。当前 helper 支持 service 外壳：
+
+```bash
+CV_HELPER_SERVICE_ENABLED=true \
+CV_HELPER_SERVICE_BIND=127.0.0.1:18080 \
+CV_HELPER_AUTH_TOKEN=local-helper-token \
+CV_RUNTIME_TURN_BRIDGE_TOKEN=local-turn-bridge-token \
+./target/release/combrabo-voice-runtime-helper
+```
+
+service 模式下，helper 只暴露控制面接口：
+
+- `GET /health`
+- `GET /internal/combrabo-voice/health`
+- `POST /internal/combrabo-voice/sessions/start`
+- `GET /internal/combrabo-voice/sessions/{callId}`
+- `POST /internal/combrabo-voice/sessions/{callId}/stop`
+
+`sessions/start` 收到 Java 传入的 LiveKit bot 入房材料后，会拉起现有 worker 子进程承接媒体链路。helper service 本身不做 ASR / LLM / TTS / 消息 / 计费，也不持久化业务数据。
+
+鉴权口径：
+
+- Java 调 helper 控制面使用 `Authorization: Bearer {helperAuthToken}`。
+- `authProfile` 只是非敏感 alias，首版允许 `default / local-dev / dev`。
+- helper 根据自身部署配置读取同名 profile 的 `turnBridgeToken`，再用于调用 Java internal turn bridge。
+- token 只能来自本机私有 `.env`、Jenkins credentials 或服务器 ENC 配置，不写入 Git。
+
+Docker service mode 本地启动时，`run-local.sh` 会把 `CV_HELPER_SERVICE_BIND=127.0.0.1:18080`
+映射为容器内 `0.0.0.0:18080` 并发布到宿主 `127.0.0.1:18080`。因此本机探活使用：
+
+```bash
+CV_HELPER_SERVICE_ENABLED=true \
+CV_HELPER_SERVICE_BIND=127.0.0.1:18080 \
+CV_HELPER_AUTH_TOKEN=local-helper-token \
+CV_RUNTIME_TURN_BRIDGE_TOKEN=local-turn-bridge-token \
+COMBRABO_VOICE_RUNTIME_HELPER_MODE=docker \
+./run-local.sh
+
+curl http://127.0.0.1:18080/health
+```
+
+`sessions/start` 会等待 worker 回报真实 readiness：worker 在 LiveKit connect 后输出
+`bot_participant_joined`，在 bot audio track 发布后输出 `bot_track_ready`。service 捕获这些
+结构化事件后才返回 `STARTED + botTrackReady=true`；如果 worker 失败或超时，则返回结构化
+`RUNTIME_START_FAILED` / `RUNTIME_START_TIMEOUT`。
## 本机固定问候音频
-helper 支持 WAV / MP3 音频自动识别。主动问候 `prepare` 生成的 MP3 可以直接进入 helper；本地 smoke 仍可使用 WAV fixture 作为固定兜底样本。
+helper 当前 MVP 支持 **WAV / MP3** 音频自动识别。主动问候 `prepare` 生成的 MP3 可以直接进入 helper；本地 smoke 仍可使用 WAV fixture 作为固定兜底样本。
-在 macOS local 环境，可先生成一份本机固定问候 WAV fixture：
+在 macOS local 环境，推荐先生成一份本机固定问候 WAV fixture：
```bash
-./generate-local-fixture.sh
+./tools/combrabo-voice-runtime-helper/generate-local-fixture.sh
```
默认输出：
```text
-.local/greeting-local.wav
+tools/combrabo-voice-runtime-helper/.local/greeting-local.wav
```
-## lmrobot-app 接线示例
+## 推荐 launch-command
+
+`lmrobot-app` local 推荐设置：
```bash
-export COMBRABO_VOICE_RUNTIME_WORKDIR=/opt/lmrobot/lm-livekit-helper
-export COMBRABO_VOICE_RUNTIME_LAUNCH_COMMAND=/opt/lmrobot/lm-livekit-helper/run-local.sh
-export COMBRABO_VOICE_RUNTIME_HELPER_MODE=docker
-export COMBRABO_VOICE_RUNTIME_HELPER_IMAGE=registry.example.com/lm-livekit-helper:git-sha
-export COMBRABO_VOICE_RUNTIME_FALLBACK_GREETING_AUDIO_PATH=/opt/lmrobot/lm-livekit-helper/.local/greeting-local.wav
+export COMBRABO_VOICE_RUNTIME_WORKDIR=/Users/ar/Applications/combrabo/wwww
+export COMBRABO_VOICE_RUNTIME_LAUNCH_COMMAND=./tools/combrabo-voice-runtime-helper/run-local.sh
+export COMBRABO_VOICE_RUNTIME_FALLBACK_GREETING_AUDIO_PATH=/Users/ar/Applications/combrabo/wwww/tools/combrabo-voice-runtime-helper/.local/greeting-local.wav
```
-`CV_LIVEKIT_URL`、`CV_LIVEKIT_ROOM_ID`、`CV_LIVEKIT_BOT_TOKEN`、`CV_LIVEKIT_BOT_PARTICIPANT_IDENTITY` 等连接材料由 `lmrobot-app` 在每次 `calls/start` 时生成并注入，不写入仓库。
+## 主动问候音频质量诊断
-## LiveKit Data 设备输出 smoke
+helper 会在播放主动问候前输出脱敏音质摘要，覆盖：
-需要验证 NativeSDK 是否能收到设备输出 Data Message 时，在启动 `lmrobot-app` 前打开：
+- 源格式、源字节数、源采样率、源声道数；
+- 解码样本数、解码时长、目标采样率、目标声道数；
+- 20ms frame 数、理论播放时长、实际推帧 wall duration、推帧漂移；
+- RMS、peak、削波样本数、静音比例；
+- MP3 `SkippedData` / `InsufficientData` 计数。
+
+本地排查音质问题时，可以额外设置 debug dump 目录：
```bash
-export CV_DEVICE_OUTPUT_SMOKE_ENABLED=true
+export CV_AUDIO_DEBUG_DUMP_DIR=/tmp/combrabo-voice-audio-debug
```
-helper 会在 bot 进房并发布音轨后，向 `CV_LIVEKIT_USER_PARTICIPANT_IDENTITY` 指向的 SDK client 发送 reliable Data Message：
+开启后 helper 会保留两类文件：
-- `topic`: `device_output`
-- `type`: `device_output`
-- `schemaVersion`: `1.0`
-- `commandCode`: 例如 `vibration.start`
-- `params`: 设备参数
+```text
+<callId>-greeting-source.<wav|mp3>
+<callId>-greeting-target.wav
+```
-如需指定接收方，可使用英文逗号分隔：
+其中 `source` 是原始主动问候音频，`target` 是推送给 LiveKit 前的 `48kHz/mono/16-bit` WAV。文件只用于本地回听排查，不进入 Git、不写入协作事件正文，不上传到线上环境。
+
+## 用户上行音频观测
+
+helper 默认开启下一阶段入口观测：
```bash
-export CV_DEVICE_OUTPUT_DESTINATION_IDENTITIES=client-identity-a,client-identity-b
+export CV_ENABLE_USER_AUDIO_OBSERVER=true
```
-如果没有指定接收方且没有 `CV_LIVEKIT_USER_PARTICIPANT_IDENTITY`，helper 会广播到 room。设备控制建议继续使用 reliable/ordered Data Message；ACK / 执行结果首版建议走 HTTP，便于落库、重试和排查。
+如需临时回退为纯固定问候播放器，可关闭：
-## 文档
+```bash
+export CV_ENABLE_USER_AUDIO_OBSERVER=false
+```
-- [Runtime Contract](docs/runtime-contract.md)
-- [Jenkins Build And Deploy Runbook](docs/jenkins-build-deploy.md)
+开启后 helper 会监听 LiveKit `TrackSubscribed` 事件，并对远端用户音频轨道输出脱敏日志：
-## 安全要求
+- `runtime helper user_track_subscribe_requested`
+- `runtime helper user_track_subscribed`
+- `runtime helper user_audio_frame_received`
+- `runtime helper user_audio_frame_summary`
+- `runtime helper user_audio_stream_ended`
-仓库内只允许提交 `.env.example` 这类脱敏样例。禁止提交真实 LiveKit token、API key、secret、roomId、participantIdentity、用户语音内容、ASR 文本、LLM 回复全文或完整 prompt。
+这些日志只记录 `call_id`、`trace_id`、脱敏 participant alias、脱敏 track sid、采样率、声道、帧数和时间摘要。禁止记录 token、room secret、真实 participantIdentity、完整 roomId、音频内容、ASR 文本、用户语音内容或 AI 回复文本。
+
+## 轻量 VAD / turn detection
+
+在确认 helper 能收到用户上行音频帧后，helper 默认开启轻量 VAD：
+
+```bash
+export CV_ENABLE_SIMPLE_VAD=true
+```
+
+首版 VAD 不引入外部模型，只基于上行 PCM frame 的 RMS / peak 做保守阈值判断。配置了 Java runtime turn bridge 后，helper 会在有效 `vad_speech_end` 后把本轮 PCM 写成 `user.wav` artifact，并调用 Java 内部 bridge；helper 仍不做 ASR / LLM / TTS / 消息落库。
+
+可调参数：
+
+```bash
+export CV_VAD_RMS_THRESHOLD=0.012
+export CV_VAD_PEAK_THRESHOLD=0.08
+export CV_VAD_START_FRAMES=5
+export CV_VAD_END_SILENCE_MS=700
+export CV_VAD_MIN_SPEECH_MS=300
+export CV_VAD_MAX_TURN_MS=10000
+export CV_VAD_INITIAL_IGNORE_MS=500
+export CV_VAD_GATE_UNTIL_GREETING_DONE=true
+export CV_VAD_POST_GREETING_DELAY_MS=800
+```
+
+Java turn bridge 相关配置：
+
+```bash
+export CV_RUNTIME_TURN_BRIDGE_URL=http://127.0.0.1:11102/internal/sdk/combrabo-voice/runtime/turns
+export CV_RUNTIME_TURN_BRIDGE_TOKEN=local-dev-token
+export CV_RUNTIME_TURN_ARTIFACT_DIR=/tmp/combrabo-voice-runtime/turn-artifacts
+export CV_RUNTIME_SESSION_NONCE=runtime-session-nonce
+```
+
+这些值由 `lmrobot-app` 启动 runtime helper 时注入。Docker 模式会透传 env，并把 `CV_RUNTIME_TURN_ARTIFACT_DIR` 挂载进容器。日志不得打印 `CV_RUNTIME_TURN_BRIDGE_TOKEN`。
+
+本机 Docker 模式下，`run-local.sh` 会自动把 `CV_RUNTIME_TURN_BRIDGE_URL` 中的 `127.0.0.1` / `localhost` 改写为 `host.docker.internal`，避免 helper 容器把 Java internal turn bridge 误解析为容器自身。
+
+开启后新增脱敏事件：
+
+- `runtime helper vad_disabled_greeting`
+- `runtime helper vad_enable_scheduled`
+- `runtime helper vad_enabled`
+- `runtime helper vad_ignored_before_enabled`
+- `runtime helper vad_speech_start`
+- `runtime helper vad_speech_end`
+- `runtime helper vad_speech_too_short`
+- `runtime helper vad_no_speech_summary`
+- `runtime helper turn_artifact_written`
+- `runtime helper turn_bridge_completed`
+- `runtime helper turn_bridge_failed`
+- `runtime helper turn_bridge_skipped`
+
+这些事件只记录 `call_id`、`trace_id`、脱敏 participant alias、脱敏 track sid、turn 序号、起止时间、帧数、样本数、RMS / peak 摘要、结束原因、artifact 字节数、bridge HTTP / reasonCode 摘要。它们不记录用户音频内容、ASR 文本、LLM 回复、TTS URL、bridge token 或本机绝对 artifact 路径。
+
+默认启用 `CV_VAD_GATE_UNTIL_GREETING_DONE=true`，即主动问候播放完成前不产生 `vad_speech_start/end`。问候播放结束后，helper 会按 `CV_VAD_POST_GREETING_DELAY_MS` 延迟开启 VAD；门禁期间收到的用户音频帧只输出 `vad_ignored_before_enabled` 摘要，用于证明上行音频仍在，但不会形成用户 turn。
+
+当前 local 真机复测默认 `CV_VAD_POST_GREETING_DELAY_MS=800`，用于在主动问候实际写完后保留短尾缓冲，避免 15 秒固定禁听窗口吞掉用户首句。若需要做受控回声隔离实验，可以临时调大该值，但不得作为产品化默认值。
+
+## 当前边界
+
+- `CV_GREETING_AUDIO_FILE`：优先读取本地 WAV / MP3 文件；
+- `CV_GREETING_AUDIO_URL`：支持下载后按 WAV / MP3 解析；
+- 没有问候音频时，helper 仍会发布 bot track 并保持连接，但不会主动播放音频；
+- 如果传入了问候音频路径/URL，但内容不是可解析的 WAV / MP3，helper 会启动失败并把错误返回给 `calls/start`。
+- 用户上行音频观测只证明 helper 能订阅用户音轨并收到帧，不代表 ASR/LLM/TTS 动态对话已经完成。
+- 轻量 VAD 在 bridge 配置完整时会生成 `user.wav` artifact 并调用 Java internal turn bridge；bridge 配置缺失时只记录 `turn_bridge_skipped`，不阻断 helper 进程。
+- 当前阶段只完成用户 turn artifact 和 Java bridge 请求；第二段 AI 回复音频写回 bot track 属于下一阶段。
+- 主动问候、固定 TTS 和后续动态回复必须统一写入 helper 发布的 LiveKit bot 音轨；不允许另起本地播放旁路。
diff --git a/docs/runtime-contract.md b/docs/runtime-contract.md
index d4446f3..1ac5424 100644
--- a/docs/runtime-contract.md
+++ b/docs/runtime-contract.md
@@ -20,6 +20,212 @@
`lmrobot-app` remains the business authority for call state, role permission, greeting prepare/consume, ASR, prompt/history, LLM, TTS, message persistence, activity, diagnostics and reasonCode.
+## Turn stream fixture fast path
+
+For TTS streaming contract changes, run the fixture validator before Docker or iPhone smoke:
+
+```bash
+node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-happy.ndjson
+node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-mp3-chunks.ndjson
+```
+
+The fixture validates only the protocol shape and fast-path invariants:
+
+- `reply_playback_mode_selected` appears before audio chunks.
+- `reply_playback_started` appears before the first audio chunk.
+- `reply_audio_chunk.audioChunk.format` is `pcm_s16le`, `mp3`, `mpeg` or `wav`.
+- `pcm_s16le` chunks are `48000Hz` mono and 16-bit aligned.
+- Encoded chunks such as `mp3` must have non-empty payload bytes. This validates contract shape only; decoder and LiveKit write behavior still require runtime smoke.
+- A terminal event exists.
+
+This is not a media smoke. Docker build, LiveKit room join and iPhone `first_reply_remote_audio` still remain the runtime acceptance path.
+
+## LiveKit Data Message contract
+
+The helper publishes reply playback state through LiveKit reliable Data Message:
+
+```text
+topic = combrabo_voice.reply_state
+```
+
+The topic is the LiveKit channel. The payload `type` is the business message type and must not reuse the topic value:
+
+```json
+{
+ "type": "reply_state",
+ "schemaVersion": "1.0",
+ "callId": "cv_xxx",
+ "traceId": "trace_xxx",
+ "turnId": "turn_xxx",
+ "replyPlaybackMode": "streaming_tts",
+ "state": "reply_playback_started",
+ "seq": 3,
+ "tsMs": 1234567890
+}
+```
+
+## Runtime modes
+
+The helper has two runtime modes:
+
+- `worker` mode: the original one-call process. It reads `CV_*` variables and joins one LiveKit room.
+- `service` mode: a long-running control-plane wrapper. It exposes internal HTTP endpoints and spawns one worker child process per session.
+
+Service mode is enabled by either:
+
+```bash
+CV_HELPER_SERVICE_ENABLED=true ./combrabo-voice-runtime-helper
+./combrabo-voice-runtime-helper service
+```
+
+Service mode does not change the media worker boundary. It only replaces `lmrobot-app -> ProcessBuilder` with `lmrobot-app -> helper HTTP control plane`.
+
+## Service mode endpoints
+
+All session endpoints require:
+
+```http
+Authorization: Bearer {helperAuthToken}
+X-Voice-Trace-Id: {traceId}
+```
+
+### Health
+
+```http
+GET /internal/combrabo-voice/health
+```
+
+For Jenkins and ops probes, `/health` is also supported and returns the same body.
+
+Response:
+
+```json
+{
+ "code": 0,
+ "msg": "",
+ "data": {
+ "status": "UP",
+ "version": "0.1.0",
+ "mode": "service"
+ }
+}
+```
+
+### Start session
+
+```http
+POST /internal/combrabo-voice/sessions/start
+Idempotency-Key: {callId}
+Content-Type: application/json
+```
+
+Request body follows the `lmrobot-app` service-mode contract:
+
+```json
+{
+ "callId": "cv_xxx",
+ "traceId": "trace_xxx",
+ "runtimeSessionNonce": "nonce_xxx",
+ "authProfile": "local-dev",
+ "livekit": {
+ "url": "ws://127.0.0.1:7880",
+ "roomId": "room_xxx",
+ "botToken": "dynamic_bot_token",
+ "botParticipantIdentity": "bot_xxx",
+ "userParticipantIdentity": "user_xxx"
+ },
+ "turnBridge": {
+ "url": "http://127.0.0.1:19102/internal/sdk/combrabo-voice/runtime/turns/stream"
+ },
+ "audio": {
+ "firstAudioSource": "fixed_greeting_tts",
+ "greetingAudio": {
+ "type": "local_file",
+ "pathRef": "greeting/cv_xxx.mp3",
+ "format": "mp3"
+ }
+ },
+ "runtime": {
+ "turnArtifactDir": "/tmp/combrabo-voice/artifacts",
+ "audioDebugDumpEnabled": false
+ }
+}
+```
+
+`botToken` is dynamic LiveKit connection material. It is accepted only through this internal control plane and must not be logged.
+
+Success response:
+
+```json
+{
+ "code": 0,
+ "msg": "",
+ "data": {
+ "callId": "cv_xxx",
+ "status": "STARTED",
+ "runtimeSessionId": "rt_cv_xxx",
+ "botParticipantJoined": true,
+ "botTrackReady": true,
+ "firstAudioSource": "fixed_greeting_tts"
+ }
+}
+```
+
+`sessions/start` must not treat a spawned process as ready. In service mode the worker emits
+`cv_activity` lines on stdout. The service updates its session registry from these events and
+waits for:
+
+- `bot_participant_joined`
+- `bot_track_ready`
+
+Only after both are observed can `botParticipantJoined=true` and `botTrackReady=true` be returned.
+If the worker exits first, return `RUNTIME_START_FAILED`; if the ready window expires, return
+`RUNTIME_START_TIMEOUT`.
+
+### Query session
+
+```http
+GET /internal/combrabo-voice/sessions/{callId}
+```
+
+The response contains only state aliases, never token, roomId or participantIdentity.
+
+### Stop session
+
+```http
+POST /internal/combrabo-voice/sessions/{callId}/stop
+Content-Type: application/json
+```
+
+```json
+{
+ "reason": "client_end",
+ "runtimeSessionNonce": "nonce_xxx"
+}
+```
+
+Repeated stop returns `code=0` with `alreadyStopped=true`. If nonce mismatches, helper returns `RUNTIME_SESSION_MISMATCH`.
+
+## Service mode auth profiles
+
+`authProfile` is an alias, not a token. The first version allows `default`, `local-dev` and `dev`.
+
+For local smoke, helper accepts these environment variables:
+
+```bash
+CV_HELPER_AUTH_TOKEN=local-helper-token
+CV_RUNTIME_TURN_BRIDGE_TOKEN=local-turn-bridge-token
+```
+
+Profile-specific override is supported by suffix:
+
+```bash
+CV_HELPER_AUTH_TOKEN_LOCAL_DEV=local-helper-token
+CV_TURN_BRIDGE_TOKEN_LOCAL_DEV=local-turn-bridge-token
+```
+
+The helper never receives `turnBridgeToken` in the `sessions/start` body. It resolves the token from its own deployment config, using `authProfile`.
+
## Required environment variables
| Name | Source | Notes |
diff --git a/fixtures/turn-stream-happy.ndjson b/fixtures/turn-stream-happy.ndjson
new file mode 100644
index 0000000..b5f48b3
--- /dev/null
+++ b/fixtures/turn-stream-happy.ndjson
@@ -0,0 +1,5 @@
+{"event":"reply_playback_mode_selected","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_001","seq":1,"replyPlaybackMode":"streaming_tts"}
+{"event":"reply_state","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_001","seq":2,"replyPlaybackMode":"streaming_tts","state":"reply_output_pending"}
+{"event":"reply_state","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_001","seq":3,"replyPlaybackMode":"streaming_tts","state":"reply_playback_started"}
+{"event":"reply_audio_chunk","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_001","seq":4,"replyPlaybackMode":"streaming_tts","audioChunk":{"chunkSeq":1,"format":"pcm_s16le","sampleRate":48000,"channels":1,"payloadBase64":"AAAAAAAAAAA=","last":true}}
+{"event":"turn_completed","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_001","seq":5,"replyPlaybackMode":"streaming_tts","completion":{"messageId":"msg_fixture_alias","audioChunkCount":1}}
diff --git a/fixtures/turn-stream-mp3-chunks.ndjson b/fixtures/turn-stream-mp3-chunks.ndjson
new file mode 100644
index 0000000..1c4064a
--- /dev/null
+++ b/fixtures/turn-stream-mp3-chunks.ndjson
@@ -0,0 +1,6 @@
+{"event":"reply_playback_mode_selected","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_mp3_001","seq":1,"replyPlaybackMode":"streaming_tts","diagnostics":{"ttsProvider":"elevenlabs","providerStreamingSupported":true,"streamBridgeMode":"stream"}}
+{"event":"reply_state","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_mp3_001","seq":2,"replyPlaybackMode":"streaming_tts","state":"reply_output_pending"}
+{"event":"reply_state","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_mp3_001","seq":3,"replyPlaybackMode":"streaming_tts","state":"reply_playback_started"}
+{"event":"reply_audio_chunk","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_mp3_001","seq":4,"replyPlaybackMode":"streaming_tts","audioChunk":{"chunkSeq":1,"format":"mp3","sampleRate":44100,"channels":1,"payloadBase64":"SUQzBAAAAAAA","last":false}}
+{"event":"reply_audio_chunk","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_mp3_001","seq":5,"replyPlaybackMode":"streaming_tts","audioChunk":{"chunkSeq":2,"format":"mp3","sampleRate":44100,"channels":1,"payloadBase64":"//uQZAAAAAAAAAA=","last":true}}
+{"event":"turn_completed","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn_fixture_mp3_001","seq":6,"replyPlaybackMode":"streaming_tts","completion":{"messageId":"msg_fixture_alias","audioChunkCount":2}}
diff --git a/generate-local-fixture.sh b/generate-local-fixture.sh
index 07b6fb0..1b209f1 100755
--- a/generate-local-fixture.sh
+++ b/generate-local-fixture.sh
@@ -1,16 +1,16 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-OUTPUT_DIR="${SCRIPT_DIR}/.local"
-TEXT="${1:-こんにちは、来てくれてうれしいです。今日はゆっくりお話ししましょう。}"
-TMP_AIFF="${OUTPUT_DIR}/greeting-local.aiff"
-OUTPUT_WAV="${OUTPUT_DIR}/greeting-local.wav"
-
-mkdir -p "${OUTPUT_DIR}"
-
-/usr/bin/say -v Kyoko -o "${TMP_AIFF}" "${TEXT}"
-/usr/bin/afconvert -f WAVE -d LEI16@16000 -c 1 "${TMP_AIFF}" "${OUTPUT_WAV}"
-rm -f "${TMP_AIFF}"
-
-echo "Generated local greeting fixture: ${OUTPUT_WAV}"
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+OUTPUT_DIR="${SCRIPT_DIR}/.local"
+TEXT="${1:-こんにちは、来てくれてうれしいです。今日はゆっくりお話ししましょう。}"
+TMP_AIFF="${OUTPUT_DIR}/greeting-local.aiff"
+OUTPUT_WAV="${OUTPUT_DIR}/greeting-local.wav"
+
+mkdir -p "${OUTPUT_DIR}"
+
+/usr/bin/say -v Kyoko -o "${TMP_AIFF}" "${TEXT}"
+/usr/bin/afconvert -f WAVE -d LEI16@16000 -c 1 "${TMP_AIFF}" "${OUTPUT_WAV}"
+rm -f "${TMP_AIFF}"
+
+echo "Generated local greeting fixture: ${OUTPUT_WAV}"
diff --git a/run-local.sh b/run-local.sh
index 2e304e7..337bd55 100755
--- a/run-local.sh
+++ b/run-local.sh
@@ -1,159 +1,246 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
+#!/usr/bin/env bash
+set -euo pipefail
+
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-WORKSPACE_ROOT="${COMBRABO_VOICE_RUNTIME_HELPER_WORKDIR:-${SCRIPT_DIR}}"
+WORKSPACE_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
MANIFEST_PATH="${SCRIPT_DIR}/Cargo.toml"
BINARY_PATH="${SCRIPT_DIR}/target/release/combrabo-voice-runtime-helper"
-DOCKERFILE_PATH="${SCRIPT_DIR}/Dockerfile"
-IMAGE_NAME="${COMBRABO_VOICE_RUNTIME_HELPER_IMAGE:-combrabo-voice-runtime-helper:local}"
-MODE="${COMBRABO_VOICE_RUNTIME_HELPER_MODE:-auto}"
-
-usage() {
- cat <<'EOF'
-Usage:
- run-local.sh [--prepare] [--rebuild]
-
-Options:
- --prepare Prepare the current helper runtime mode without starting the helper.
- --rebuild Force docker image rebuild in docker mode.
-EOF
-}
-
-PREPARE_ONLY=false
-FORCE_REBUILD=false
-
-while [[ $# -gt 0 ]]; do
- case "$1" in
- --prepare)
- PREPARE_ONLY=true
- shift
- ;;
- --rebuild)
- FORCE_REBUILD=true
- shift
- ;;
- -h|--help)
- usage
- exit 0
- ;;
- *)
- echo "Unknown argument: $1" >&2
- usage >&2
- exit 1
- ;;
- esac
-done
-
-resolve_mode() {
- local requested="$1"
- if [[ "${requested}" != "auto" ]]; then
- printf '%s' "${requested}"
- return 0
- fi
- if [[ "$(uname -s)" == "Darwin" ]] && command -v docker >/dev/null 2>&1; then
- printf '%s' "docker"
- return 0
- fi
- printf '%s' "host"
-}
-
-prepare_host() {
- (cd "${SCRIPT_DIR}" && cargo build --release)
-}
-
-prepare_docker() {
- if ! command -v docker >/dev/null 2>&1; then
- echo "Docker is required for docker helper mode, but docker is not installed." >&2
- exit 1
- fi
- if [[ "${FORCE_REBUILD}" == "true" ]] || ! docker image inspect "${IMAGE_NAME}" >/dev/null 2>&1; then
- docker build -t "${IMAGE_NAME}" "${SCRIPT_DIR}"
- fi
-}
-
-rewrite_livekit_url_for_docker() {
- local url="$1"
- if [[ "${url}" =~ ^(ws|wss)://(127\.0\.0\.1|localhost)([:/].*)?$ ]]; then
- printf '%s://host.docker.internal%s' "${BASH_REMATCH[1]}" "${BASH_REMATCH[3]:-}"
- return 0
- fi
- printf '%s' "${url}"
-}
-
-run_host() {
- if [[ ! -x "${BINARY_PATH}" ]]; then
- prepare_host
- fi
- exec "${BINARY_PATH}"
-}
-
-run_docker() {
- prepare_docker
- local livekit_url="${CV_LIVEKIT_URL:-}"
- local -a docker_args
- docker_args=(
- --rm
- --name "cv-helper-${CV_CALL_ID:-manual}"
+DOCKERFILE_PATH="${SCRIPT_DIR}/Dockerfile"
+IMAGE_NAME="${COMBRABO_VOICE_RUNTIME_HELPER_IMAGE:-combrabo-voice-runtime-helper:local}"
+MODE="${COMBRABO_VOICE_RUNTIME_HELPER_MODE:-auto}"
+
+usage() {
+ cat <<'EOF'
+Usage:
+ run-local.sh [--prepare] [--rebuild]
+
+Options:
+ --prepare Prepare the current helper runtime mode without starting the helper.
+ --rebuild Force docker image rebuild in docker mode.
+EOF
+}
+
+PREPARE_ONLY=false
+FORCE_REBUILD=false
+
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --prepare)
+ PREPARE_ONLY=true
+ shift
+ ;;
+ --rebuild)
+ FORCE_REBUILD=true
+ shift
+ ;;
+ -h|--help)
+ usage
+ exit 0
+ ;;
+ *)
+ echo "Unknown argument: $1" >&2
+ usage >&2
+ exit 1
+ ;;
+ esac
+done
+
+resolve_mode() {
+ local requested="$1"
+ if [[ "${requested}" != "auto" ]]; then
+ printf '%s' "${requested}"
+ return 0
+ fi
+ if [[ "$(uname -s)" == "Darwin" ]] && command -v docker >/dev/null 2>&1; then
+ printf '%s' "docker"
+ return 0
+ fi
+ printf '%s' "host"
+}
+
+prepare_host() {
+ (cd "${SCRIPT_DIR}" && cargo build --release)
+}
+
+prepare_docker() {
+ if ! command -v docker >/dev/null 2>&1; then
+ echo "Docker is required for docker helper mode, but docker is not installed." >&2
+ exit 1
+ fi
+ if [[ "${FORCE_REBUILD}" == "true" ]] || ! docker image inspect "${IMAGE_NAME}" >/dev/null 2>&1; then
+ docker build -t "${IMAGE_NAME}" "${SCRIPT_DIR}"
+ fi
+}
+
+rewrite_livekit_url_for_docker() {
+ local url="$1"
+ if [[ "${url}" =~ ^(ws|wss)://(127\.0\.0\.1|localhost)([:/].*)?$ ]]; then
+ printf '%s://host.docker.internal%s' "${BASH_REMATCH[1]}" "${BASH_REMATCH[3]:-}"
+ return 0
+ fi
+ printf '%s' "${url}"
+}
+
+rewrite_http_url_for_docker() {
+ local url="$1"
+ if [[ "${url}" =~ ^(http|https)://(127\.0\.0\.1|localhost)([:/].*)?$ ]]; then
+ printf '%s://host.docker.internal%s' "${BASH_REMATCH[1]}" "${BASH_REMATCH[3]:-}"
+ return 0
+ fi
+ printf '%s' "${url}"
+}
+
+is_truthy() {
+ case "${1:-}" in
+ true|TRUE|1|yes|YES|on|ON)
+ return 0
+ ;;
+ *)
+ return 1
+ ;;
+ esac
+}
+
+docker_service_host() {
+ local bind_addr="${1:-127.0.0.1:18080}"
+ if [[ "${bind_addr}" =~ ^([^:]+):([0-9]+)$ ]]; then
+ printf '%s' "${BASH_REMATCH[1]}"
+ return 0
+ fi
+ printf '%s' "127.0.0.1"
+}
+
+docker_service_port() {
+ local bind_addr="${1:-127.0.0.1:18080}"
+ if [[ "${bind_addr}" =~ ^([^:]+):([0-9]+)$ ]]; then
+ printf '%s' "${BASH_REMATCH[2]}"
+ return 0
+ fi
+ printf '%s' "18080"
+}
+
+run_host() {
+ if [[ ! -x "${BINARY_PATH}" ]]; then
+ prepare_host
+ fi
+ exec "${BINARY_PATH}"
+}
+
+run_docker() {
+ prepare_docker
+ local livekit_url="${CV_LIVEKIT_URL:-}"
+ local -a docker_args
+ docker_args=(
+ --rm
+ --name "cv-helper-${CV_CALL_ID:-manual}"
-v "${WORKSPACE_ROOT}:${WORKSPACE_ROOT}"
- -w "${WORKSPACE_ROOT}"
- -e CV_CALL_ID
- -e CV_TRACE_ID
- -e CV_GAME_MATCH_ID
- -e CV_LIVEKIT_URL
- -e CV_LIVEKIT_ROOM_ID
- -e CV_LIVEKIT_BOT_TOKEN
- -e CV_LIVEKIT_BOT_PARTICIPANT_IDENTITY
- -e CV_LIVEKIT_USER_PARTICIPANT_IDENTITY
- -e CV_FIRST_AUDIO_SOURCE
+ -w "${WORKSPACE_ROOT}"
+ -e CV_CALL_ID
+ -e CV_TRACE_ID
+ -e CV_GAME_MATCH_ID
+ -e CV_LIVEKIT_URL
+ -e CV_LIVEKIT_ROOM_ID
+ -e CV_LIVEKIT_BOT_TOKEN
+ -e CV_LIVEKIT_BOT_PARTICIPANT_IDENTITY
+ -e CV_LIVEKIT_USER_PARTICIPANT_IDENTITY
+ -e CV_FIRST_AUDIO_SOURCE
-e CV_GREETING_SOURCE
-e CV_GREETING_AUDIO_FILE
-e CV_GREETING_AUDIO_URL
-e CV_GREETING_TEXT
- -e CV_ROLE_ID
- -e CV_DEVICE_OUTPUT_SMOKE_ENABLED
- -e CV_DEVICE_OUTPUT_DESTINATION_IDENTITIES
+ -e CV_AUDIO_DEBUG_DUMP_DIR
+ -e CV_ENABLE_USER_AUDIO_OBSERVER
+ -e CV_ENABLE_SIMPLE_VAD
+ -e CV_VAD_RMS_THRESHOLD
+ -e CV_VAD_PEAK_THRESHOLD
+ -e CV_VAD_START_FRAMES
+ -e CV_VAD_END_SILENCE_MS
+ -e CV_VAD_MIN_SPEECH_MS
+ -e CV_VAD_MAX_TURN_MS
+ -e CV_VAD_INITIAL_IGNORE_MS
+ -e CV_VAD_GATE_UNTIL_GREETING_DONE
+ -e CV_VAD_POST_GREETING_DELAY_MS
+ -e CV_RUNTIME_TURN_BRIDGE_URL
+ -e CV_RUNTIME_TURN_BRIDGE_TOKEN
+ -e CV_RUNTIME_TURN_ARTIFACT_DIR
+ -e CV_RUNTIME_SESSION_NONCE
+ -e CV_HELPER_SERVICE_ENABLED
+ -e CV_HELPER_SERVICE_BIND
+ -e CV_HELPER_AUTH_TOKEN
+ -e CV_HELPER_AUTH_TOKEN_LOCAL_DEV
+ -e CV_TURN_BRIDGE_TOKEN_LOCAL_DEV
+ -e CV_HELPER_AUTH_TOKEN_DEV
+ -e CV_TURN_BRIDGE_TOKEN_DEV
)
- if [[ -n "${livekit_url}" ]]; then
- export CV_LIVEKIT_URL="$(rewrite_livekit_url_for_docker "${livekit_url}")"
- fi
- if [[ -n "${CV_GREETING_AUDIO_FILE:-}" && -f "${CV_GREETING_AUDIO_FILE}" ]]; then
- local audio_parent
- audio_parent="$(cd "$(dirname "${CV_GREETING_AUDIO_FILE}")" && pwd)"
- if [[ "${audio_parent}" != "${WORKSPACE_ROOT}" && "${audio_parent}" != "${WORKSPACE_ROOT}/"* ]]; then
- docker_args+=(-v "${audio_parent}:${audio_parent}:ro")
- fi
- fi
- exec docker run "${docker_args[@]}" "${IMAGE_NAME}"
-}
-
-RUNTIME_MODE="$(resolve_mode "${MODE}")"
-
-if [[ "${PREPARE_ONLY}" == "true" ]]; then
- case "${RUNTIME_MODE}" in
- host)
- prepare_host
- ;;
- docker)
- prepare_docker
- ;;
- *)
- echo "Unsupported helper mode: ${RUNTIME_MODE}" >&2
- exit 1
- ;;
- esac
- exit 0
-fi
-
-case "${RUNTIME_MODE}" in
- host)
- run_host
- ;;
- docker)
- run_docker
- ;;
- *)
- echo "Unsupported helper mode: ${RUNTIME_MODE}" >&2
- exit 1
- ;;
-esac
+ if is_truthy "${CV_HELPER_SERVICE_ENABLED:-}"; then
+ local service_bind service_host service_port
+ service_bind="${CV_HELPER_SERVICE_BIND:-127.0.0.1:18080}"
+ service_host="$(docker_service_host "${service_bind}")"
+ service_port="$(docker_service_port "${service_bind}")"
+ export CV_HELPER_SERVICE_BIND="0.0.0.0:${service_port}"
+ docker_args+=(-p "${service_host}:${service_port}:${service_port}")
+ fi
+ if [[ -n "${livekit_url}" ]]; then
+ export CV_LIVEKIT_URL="$(rewrite_livekit_url_for_docker "${livekit_url}")"
+ fi
+ if [[ -n "${CV_RUNTIME_TURN_BRIDGE_URL:-}" ]]; then
+ export CV_RUNTIME_TURN_BRIDGE_URL="$(rewrite_http_url_for_docker "${CV_RUNTIME_TURN_BRIDGE_URL}")"
+ fi
+ if [[ -n "${CV_GREETING_AUDIO_FILE:-}" && -f "${CV_GREETING_AUDIO_FILE}" ]]; then
+ local audio_parent
+ audio_parent="$(cd "$(dirname "${CV_GREETING_AUDIO_FILE}")" && pwd)"
+ if [[ "${audio_parent}" != "${WORKSPACE_ROOT}" && "${audio_parent}" != "${WORKSPACE_ROOT}/"* ]]; then
+ docker_args+=(-v "${audio_parent}:${audio_parent}:ro")
+ fi
+ fi
+ if [[ -n "${CV_AUDIO_DEBUG_DUMP_DIR:-}" ]]; then
+ mkdir -p "${CV_AUDIO_DEBUG_DUMP_DIR}"
+ local debug_dump_dir
+ debug_dump_dir="$(cd "${CV_AUDIO_DEBUG_DUMP_DIR}" && pwd)"
+ export CV_AUDIO_DEBUG_DUMP_DIR="${debug_dump_dir}"
+ if [[ "${debug_dump_dir}" != "${WORKSPACE_ROOT}" && "${debug_dump_dir}" != "${WORKSPACE_ROOT}/"* ]]; then
+ docker_args+=(-v "${debug_dump_dir}:${debug_dump_dir}")
+ fi
+ fi
+ if [[ -n "${CV_RUNTIME_TURN_ARTIFACT_DIR:-}" ]]; then
+ mkdir -p "${CV_RUNTIME_TURN_ARTIFACT_DIR}"
+ local turn_artifact_dir
+ turn_artifact_dir="$(cd "${CV_RUNTIME_TURN_ARTIFACT_DIR}" && pwd)"
+ export CV_RUNTIME_TURN_ARTIFACT_DIR="${turn_artifact_dir}"
+ if [[ "${turn_artifact_dir}" != "${WORKSPACE_ROOT}" && "${turn_artifact_dir}" != "${WORKSPACE_ROOT}/"* ]]; then
+ docker_args+=(-v "${turn_artifact_dir}:${turn_artifact_dir}")
+ fi
+ fi
+ exec docker run "${docker_args[@]}" "${IMAGE_NAME}"
+}
+
+RUNTIME_MODE="$(resolve_mode "${MODE}")"
+
+if [[ "${PREPARE_ONLY}" == "true" ]]; then
+ case "${RUNTIME_MODE}" in
+ host)
+ prepare_host
+ ;;
+ docker)
+ prepare_docker
+ ;;
+ *)
+ echo "Unsupported helper mode: ${RUNTIME_MODE}" >&2
+ exit 1
+ ;;
+ esac
+ exit 0
+fi
+
+case "${RUNTIME_MODE}" in
+ host)
+ run_host
+ ;;
+ docker)
+ run_docker
+ ;;
+ *)
+ echo "Unsupported helper mode: ${RUNTIME_MODE}" >&2
+ exit 1
+ ;;
+esac
diff --git a/src/audio.rs b/src/audio.rs
index 01d3a28..1aadcb4 100644
--- a/src/audio.rs
+++ b/src/audio.rs
@@ -1,317 +1,720 @@
-use std::{fs, io::Cursor, path::Path};
-
-use anyhow::{Context, Result, anyhow};
-use hound::{SampleFormat, WavReader};
-use minimp3::{Decoder as Mp3Decoder, Error as Mp3Error, Frame as Mp3Frame};
-use reqwest::Client;
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct PcmFrame {
- pub data: Vec<i16>,
- pub sample_rate: u32,
- pub num_channels: u32,
- pub samples_per_channel: u32,
-}
-
-impl PcmFrame {
- pub fn new(
- data: Vec<i16>,
- sample_rate: u32,
- num_channels: u32,
- samples_per_channel: u32,
- ) -> Self {
- Self {
- data,
- sample_rate,
- num_channels,
- samples_per_channel,
- }
- }
-}
-
-pub async fn load_pre_recorded_frames(
- http: &Client,
- audio_file: Option<&str>,
- audio_url: Option<&str>,
- target_sample_rate_hz: u32,
- target_num_channels: u16,
-) -> Result<Option<Vec<PcmFrame>>> {
- if let Some(path) = audio_file.filter(|value| !value.trim().is_empty()) {
- let audio_bytes = fs::read(Path::new(path))
- .with_context(|| format!("failed to read greeting audio file {path}"))?;
- return decode_audio_frames(&audio_bytes, target_sample_rate_hz, target_num_channels)
- .map(Some);
- }
- if let Some(url) = audio_url.filter(|value| !value.trim().is_empty()) {
- let response = http
- .get(url)
- .send()
- .await
- .with_context(|| format!("failed to fetch greeting audio {url}"))?;
- if !response.status().is_success() {
- return Err(anyhow!(
- "greeting audio fetch failed for {url} with status {}",
- response.status()
- ));
- }
- let bytes = response
- .bytes()
- .await
- .with_context(|| format!("failed to read greeting audio body {url}"))?;
- return decode_audio_frames(&bytes, target_sample_rate_hz, target_num_channels).map(Some);
- }
- Ok(None)
-}
-
-fn decode_audio_frames(
- audio_bytes: &[u8],
- target_sample_rate_hz: u32,
- target_num_channels: u16,
-) -> Result<Vec<PcmFrame>> {
- if is_wav(audio_bytes) {
- return decode_wav_frames(audio_bytes, target_sample_rate_hz, target_num_channels);
- }
- if is_mp3(audio_bytes) {
- return decode_mp3_frames(audio_bytes, target_sample_rate_hz, target_num_channels);
- }
-
- let wav_result = decode_wav_frames(audio_bytes, target_sample_rate_hz, target_num_channels);
- if wav_result.is_ok() {
- return wav_result;
- }
- let wav_error = wav_result.err();
- decode_mp3_frames(audio_bytes, target_sample_rate_hz, target_num_channels).map_err(
- |mp3_error| {
- anyhow!(
- "failed to decode greeting audio as wav or mp3: wav={}, mp3={}",
- wav_error
- .map(|error| error.to_string())
- .unwrap_or_else(|| "unknown".to_string()),
- mp3_error
- )
- },
- )
-}
-
-fn decode_wav_frames(
- wav_bytes: &[u8],
- target_sample_rate_hz: u32,
- target_num_channels: u16,
-) -> Result<Vec<PcmFrame>> {
- let cursor = Cursor::new(wav_bytes.to_vec());
- let mut reader = WavReader::new(cursor).context("failed to open greeting wav bytes")?;
- let spec = reader.spec();
- let src_channels = spec.channels.max(1);
- let mut samples = match (spec.sample_format, spec.bits_per_sample) {
- (SampleFormat::Int, 16) => reader
- .samples::<i16>()
- .collect::<std::result::Result<Vec<_>, _>>()
- .context("failed to decode 16-bit wav samples")?,
- (SampleFormat::Float, 32) => reader
- .samples::<f32>()
- .map(|sample| sample.map(|value| (value.clamp(-1.0, 1.0) * i16::MAX as f32) as i16))
- .collect::<std::result::Result<Vec<_>, _>>()
- .context("failed to decode float wav samples")?,
- _ => {
- return Err(anyhow!(
- "unsupported greeting wav format: {:?} {}-bit",
- spec.sample_format,
- spec.bits_per_sample
- ));
- }
- };
-
- samples = remap_channels(samples, src_channels, target_num_channels);
- samples = resample_linear(
- samples,
- spec.sample_rate,
- target_sample_rate_hz,
- target_num_channels,
- );
-
- Ok(chunk_pcm_samples(
- samples,
- target_sample_rate_hz,
- target_num_channels,
- ))
-}
-
-fn decode_mp3_frames(
- mp3_bytes: &[u8],
- target_sample_rate_hz: u32,
- target_num_channels: u16,
-) -> Result<Vec<PcmFrame>> {
- let cursor = Cursor::new(mp3_bytes.to_vec());
- let mut decoder = Mp3Decoder::new(cursor);
- let mut samples = Vec::new();
-
- loop {
- match decoder.next_frame() {
- Ok(Mp3Frame {
- data,
- sample_rate,
- channels,
- ..
- }) => {
- let src_rate = u32::try_from(sample_rate)
- .map_err(|_| anyhow!("unsupported mp3 sample rate {sample_rate}"))?;
- let src_channels = u16::try_from(channels)
- .map_err(|_| anyhow!("unsupported mp3 channel count {channels}"))?;
- let mut frame_samples = remap_channels(data, src_channels, target_num_channels);
- frame_samples = resample_linear(
- frame_samples,
- src_rate,
- target_sample_rate_hz,
- target_num_channels,
- );
- samples.extend(frame_samples);
- }
- Err(Mp3Error::Eof) => break,
- Err(Mp3Error::SkippedData) | Err(Mp3Error::InsufficientData) => continue,
- Err(error) => return Err(anyhow!("failed to decode mp3 frame: {error:?}")),
- }
- }
-
- if samples.is_empty() {
- return Err(anyhow!("decoded mp3 contains no audio samples"));
- }
-
- Ok(chunk_pcm_samples(
- samples,
- target_sample_rate_hz,
- target_num_channels,
- ))
-}
-
-fn chunk_pcm_samples(
- samples: Vec<i16>,
- target_sample_rate_hz: u32,
- target_num_channels: u16,
-) -> Vec<PcmFrame> {
- const CHUNK_MS: u32 = 20;
- let samples_per_chunk = ((target_sample_rate_hz / 1000) * CHUNK_MS).max(1) as usize
- * usize::from(target_num_channels);
- let mut frames = Vec::new();
- for chunk in samples.chunks(samples_per_chunk) {
- let mut frame_data = chunk.to_vec();
- if frame_data.len() < samples_per_chunk {
- frame_data.resize(samples_per_chunk, 0);
- }
- let samples_per_channel =
- (frame_data.len() / usize::from(target_num_channels.max(1))) as u32;
- frames.push(PcmFrame::new(
- frame_data,
- target_sample_rate_hz,
- u32::from(target_num_channels),
- samples_per_channel,
- ));
- }
- frames
-}
-
-fn is_wav(bytes: &[u8]) -> bool {
- bytes.len() >= 12 && bytes.starts_with(b"RIFF") && &bytes[8..12] == b"WAVE"
-}
-
-fn is_mp3(bytes: &[u8]) -> bool {
- bytes.starts_with(b"ID3") || (bytes.len() >= 2 && bytes[0] == 0xff && bytes[1] & 0xe0 == 0xe0)
-}
-
-fn remap_channels(samples: Vec<i16>, src_channels: u16, dst_channels: u16) -> Vec<i16> {
- if src_channels == dst_channels {
- return samples;
- }
-
- let src_channels = usize::from(src_channels.max(1));
- let dst_channels = usize::from(dst_channels.max(1));
- let frames = samples.chunks(src_channels);
- let mut output = Vec::new();
- for frame in frames {
- match (src_channels, dst_channels) {
- (1, 2) => {
- let sample = frame.first().copied().unwrap_or_default();
- output.push(sample);
- output.push(sample);
- }
- (2, 1) => {
- let left = frame.first().copied().unwrap_or_default() as i32;
- let right = frame.get(1).copied().unwrap_or_default() as i32;
- output.push(((left + right) / 2) as i16);
- }
- _ => {
- for channel in 0..dst_channels {
- output.push(
- frame
- .get(channel % src_channels)
- .copied()
- .unwrap_or_default(),
- );
- }
- }
- }
- }
- output
-}
-
-fn resample_linear(samples: Vec<i16>, src_rate: u32, dst_rate: u32, channels: u16) -> Vec<i16> {
- if src_rate == dst_rate {
- return samples;
- }
-
- let channels = usize::from(channels.max(1));
- let src_frames = samples.len() / channels;
- if src_frames <= 1 {
- return samples;
- }
- let ratio = dst_rate as f64 / src_rate as f64;
- let dst_frames = ((src_frames as f64) * ratio).round().max(1.0) as usize;
- let mut output = Vec::with_capacity(dst_frames * channels);
-
- for dst_frame in 0..dst_frames {
- let src_pos = (dst_frame as f64) / ratio;
- let src_index = src_pos.floor() as usize;
- let next_index = (src_index + 1).min(src_frames - 1);
- let frac = (src_pos - src_index as f64) as f32;
- for channel in 0..channels {
- let a = samples[src_index * channels + channel] as f32;
- let b = samples[next_index * channels + channel] as f32;
- let mixed = a + (b - a) * frac;
- output.push(mixed.round().clamp(i16::MIN as f32, i16::MAX as f32) as i16);
- }
- }
-
- output
-}
-
-#[cfg(test)]
-mod tests {
- use super::load_pre_recorded_frames;
-
- #[tokio::test]
- async fn load_pre_recorded_frames_reads_local_wav() {
- let dir = std::env::temp_dir().join("cv-runtime-helper-tests");
- std::fs::create_dir_all(&dir).expect("create temp dir");
- let path = dir.join("fixture.wav");
- let spec = hound::WavSpec {
- channels: 1,
- sample_rate: 16_000,
- bits_per_sample: 16,
- sample_format: hound::SampleFormat::Int,
- };
- let mut writer = hound::WavWriter::create(&path, spec).expect("create wav");
- for _ in 0..16_000 {
- writer.write_sample(512_i16).expect("write sample");
- }
- writer.finalize().expect("finalize wav");
-
- let http = reqwest::Client::new();
- let frames = load_pre_recorded_frames(&http, path.to_str(), None, 48_000, 1)
- .await
- .expect("load frames")
- .expect("frames");
-
- assert!(!frames.is_empty());
- assert_eq!(frames[0].sample_rate, 48_000);
- assert_eq!(frames[0].num_channels, 1);
- }
-}
+use std::{
+ fs,
+ io::Cursor,
+ path::{Path, PathBuf},
+};
+
+use anyhow::{Context, Result, anyhow};
+use hound::{SampleFormat, WavReader, WavSpec, WavWriter};
+use minimp3::{Decoder as Mp3Decoder, Error as Mp3Error, Frame as Mp3Frame};
+use reqwest::Client;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PcmFrame {
+ pub data: Vec<i16>,
+ pub sample_rate: u32,
+ pub num_channels: u32,
+ pub samples_per_channel: u32,
+}
+
+impl PcmFrame {
+ pub fn new(
+ data: Vec<i16>,
+ sample_rate: u32,
+ num_channels: u32,
+ samples_per_channel: u32,
+ ) -> Self {
+ Self {
+ data,
+ sample_rate,
+ num_channels,
+ samples_per_channel,
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct LoadedAudio {
+ pub frames: Vec<PcmFrame>,
+ pub diagnostics: AudioDiagnostics,
+}
+
+#[derive(Clone, Debug)]
+pub struct AudioDiagnostics {
+ pub source_kind: &'static str,
+ pub source_format: &'static str,
+ pub source_bytes: usize,
+ pub source_sample_rate_hz: u32,
+ pub source_num_channels: u16,
+ pub decoded_sample_count: usize,
+ pub decoded_duration_ms: u64,
+ pub target_sample_rate_hz: u32,
+ pub target_num_channels: u16,
+ pub target_sample_count: usize,
+ pub target_duration_ms: u64,
+ pub frame_count: usize,
+ pub rms: f64,
+ pub peak: f64,
+ pub clipped_sample_count: usize,
+ pub silence_ratio: f64,
+ pub mp3_skipped_data_count: usize,
+ pub mp3_insufficient_data_count: usize,
+ pub debug_source_path: Option<String>,
+ pub debug_pcm_wav_path: Option<String>,
+}
+
+struct DecodeResult {
+ samples: Vec<i16>,
+ diagnostics: AudioDiagnostics,
+}
+
+pub async fn load_pre_recorded_frames(
+ http: &Client,
+ audio_file: Option<&str>,
+ audio_url: Option<&str>,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+ debug_dump_dir: Option<&str>,
+ call_id: &str,
+ debug_label: &str,
+) -> Result<Option<LoadedAudio>> {
+ if let Some(path) = audio_file.filter(|value| !value.trim().is_empty()) {
+ let audio_bytes = fs::read(Path::new(path))
+ .with_context(|| format!("failed to read greeting audio file {path}"))?;
+ return decode_audio_frames(
+ &audio_bytes,
+ "local_file",
+ target_sample_rate_hz,
+ target_num_channels,
+ debug_dump_dir,
+ call_id,
+ debug_label,
+ )
+ .map(Some);
+ }
+ if let Some(url) = audio_url.filter(|value| !value.trim().is_empty()) {
+ let response = http
+ .get(url)
+ .send()
+ .await
+ .with_context(|| format!("failed to fetch greeting audio {url}"))?;
+ if !response.status().is_success() {
+ return Err(anyhow!(
+ "greeting audio fetch failed for {url} with status {}",
+ response.status()
+ ));
+ }
+ let bytes = response
+ .bytes()
+ .await
+ .with_context(|| format!("failed to read greeting audio body {url}"))?;
+ return decode_audio_frames(
+ &bytes,
+ "downloaded_url",
+ target_sample_rate_hz,
+ target_num_channels,
+ debug_dump_dir,
+ call_id,
+ debug_label,
+ )
+ .map(Some);
+ }
+ Ok(None)
+}
+
+pub fn decode_audio_bytes_to_frames(
+ audio_bytes: &[u8],
+ source_kind: &'static str,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+ debug_dump_dir: Option<&str>,
+ call_id: &str,
+ debug_label: &str,
+) -> Result<LoadedAudio> {
+ decode_audio_frames(
+ audio_bytes,
+ source_kind,
+ target_sample_rate_hz,
+ target_num_channels,
+ debug_dump_dir,
+ call_id,
+ debug_label,
+ )
+}
+
+fn decode_audio_frames(
+ audio_bytes: &[u8],
+ source_kind: &'static str,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+ debug_dump_dir: Option<&str>,
+ call_id: &str,
+ debug_label: &str,
+) -> Result<LoadedAudio> {
+ if is_wav(audio_bytes) {
+ return finalize_decode_result(
+ audio_bytes,
+ decode_wav_samples(
+ audio_bytes,
+ source_kind,
+ target_sample_rate_hz,
+ target_num_channels,
+ )?,
+ target_sample_rate_hz,
+ target_num_channels,
+ debug_dump_dir,
+ call_id,
+ debug_label,
+ );
+ }
+ if is_mp3(audio_bytes) {
+ return finalize_decode_result(
+ audio_bytes,
+ decode_mp3_samples(
+ audio_bytes,
+ source_kind,
+ target_sample_rate_hz,
+ target_num_channels,
+ )?,
+ target_sample_rate_hz,
+ target_num_channels,
+ debug_dump_dir,
+ call_id,
+ debug_label,
+ );
+ }
+
+ let wav_result = decode_wav_samples(
+ audio_bytes,
+ source_kind,
+ target_sample_rate_hz,
+ target_num_channels,
+ );
+ if wav_result.is_ok() {
+ return finalize_decode_result(
+ audio_bytes,
+ wav_result?,
+ target_sample_rate_hz,
+ target_num_channels,
+ debug_dump_dir,
+ call_id,
+ debug_label,
+ );
+ }
+ let wav_error = wav_result.err();
+ let mp3_result = decode_mp3_samples(
+ audio_bytes,
+ source_kind,
+ target_sample_rate_hz,
+ target_num_channels,
+ );
+ match mp3_result {
+ Ok(result) => finalize_decode_result(
+ audio_bytes,
+ result,
+ target_sample_rate_hz,
+ target_num_channels,
+ debug_dump_dir,
+ call_id,
+ debug_label,
+ ),
+ Err(mp3_error) => Err(anyhow!(
+ "failed to decode greeting audio as wav or mp3: wav={}, mp3={}",
+ wav_error
+ .map(|error| error.to_string())
+ .unwrap_or_else(|| "unknown".to_string()),
+ mp3_error
+ )),
+ }
+}
+
+fn decode_wav_samples(
+ wav_bytes: &[u8],
+ source_kind: &'static str,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+) -> Result<DecodeResult> {
+ let cursor = Cursor::new(wav_bytes.to_vec());
+ let mut reader = WavReader::new(cursor).context("failed to open greeting wav bytes")?;
+ let spec = reader.spec();
+ let src_channels = spec.channels.max(1);
+ let decoded_samples = match (spec.sample_format, spec.bits_per_sample) {
+ (SampleFormat::Int, 16) => reader
+ .samples::<i16>()
+ .collect::<std::result::Result<Vec<_>, _>>()
+ .context("failed to decode 16-bit wav samples")?,
+ (SampleFormat::Float, 32) => reader
+ .samples::<f32>()
+ .map(|sample| sample.map(|value| (value.clamp(-1.0, 1.0) * i16::MAX as f32) as i16))
+ .collect::<std::result::Result<Vec<_>, _>>()
+ .context("failed to decode float wav samples")?,
+ _ => {
+ return Err(anyhow!(
+ "unsupported greeting wav format: {:?} {}-bit",
+ spec.sample_format,
+ spec.bits_per_sample
+ ));
+ }
+ };
+
+ let decoded_sample_count = decoded_samples.len();
+ let mut samples = remap_channels(decoded_samples, src_channels, target_num_channels);
+ samples = resample_linear(
+ samples,
+ spec.sample_rate,
+ target_sample_rate_hz,
+ target_num_channels,
+ );
+
+ Ok(DecodeResult {
+ diagnostics: build_diagnostics(
+ source_kind,
+ "wav",
+ wav_bytes.len(),
+ spec.sample_rate,
+ src_channels,
+ decoded_sample_count,
+ &samples,
+ target_sample_rate_hz,
+ target_num_channels,
+ 0,
+ 0,
+ ),
+ samples,
+ })
+}
+
+fn decode_mp3_samples(
+ mp3_bytes: &[u8],
+ source_kind: &'static str,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+) -> Result<DecodeResult> {
+ let cursor = Cursor::new(mp3_bytes.to_vec());
+ let mut decoder = Mp3Decoder::new(cursor);
+ let mut samples = Vec::new();
+ let mut decoded_sample_count = 0usize;
+ let mut source_sample_rate_hz = 0u32;
+ let mut source_num_channels = 0u16;
+ let mut skipped_data_count = 0usize;
+ let mut insufficient_data_count = 0usize;
+
+ loop {
+ match decoder.next_frame() {
+ Ok(Mp3Frame {
+ data,
+ sample_rate,
+ channels,
+ ..
+ }) => {
+ let src_rate = u32::try_from(sample_rate)
+ .map_err(|_| anyhow!("unsupported mp3 sample rate {sample_rate}"))?;
+ let src_channels = u16::try_from(channels)
+ .map_err(|_| anyhow!("unsupported mp3 channel count {channels}"))?;
+ if source_sample_rate_hz == 0 {
+ source_sample_rate_hz = src_rate;
+ }
+ if source_num_channels == 0 {
+ source_num_channels = src_channels;
+ }
+ decoded_sample_count += data.len();
+ let mut frame_samples = remap_channels(data, src_channels, target_num_channels);
+ frame_samples = resample_linear(
+ frame_samples,
+ src_rate,
+ target_sample_rate_hz,
+ target_num_channels,
+ );
+ samples.extend(frame_samples);
+ }
+ Err(Mp3Error::Eof) => break,
+ Err(Mp3Error::SkippedData) => {
+ skipped_data_count += 1;
+ continue;
+ }
+ Err(Mp3Error::InsufficientData) => {
+ insufficient_data_count += 1;
+ continue;
+ }
+ Err(error) => return Err(anyhow!("failed to decode mp3 frame: {error:?}")),
+ }
+ }
+
+ if samples.is_empty() {
+ return Err(anyhow!("decoded mp3 contains no audio samples"));
+ }
+
+ Ok(DecodeResult {
+ diagnostics: build_diagnostics(
+ source_kind,
+ "mp3",
+ mp3_bytes.len(),
+ source_sample_rate_hz,
+ source_num_channels.max(1),
+ decoded_sample_count,
+ &samples,
+ target_sample_rate_hz,
+ target_num_channels,
+ skipped_data_count,
+ insufficient_data_count,
+ ),
+ samples,
+ })
+}
+
+fn finalize_decode_result(
+ audio_bytes: &[u8],
+ mut decode_result: DecodeResult,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+ debug_dump_dir: Option<&str>,
+ call_id: &str,
+ debug_label: &str,
+) -> Result<LoadedAudio> {
+ if let Some(dir) = debug_dump_dir.filter(|value| !value.trim().is_empty()) {
+ let paths = dump_debug_audio(
+ dir,
+ call_id,
+ debug_label,
+ decode_result.diagnostics.source_format,
+ audio_bytes,
+ &decode_result.samples,
+ target_sample_rate_hz,
+ target_num_channels,
+ )?;
+ decode_result.diagnostics.debug_source_path = paths.0;
+ decode_result.diagnostics.debug_pcm_wav_path = paths.1;
+ }
+
+ let frames = chunk_pcm_samples(
+ decode_result.samples,
+ target_sample_rate_hz,
+ target_num_channels,
+ );
+ decode_result.diagnostics.frame_count = frames.len();
+
+ Ok(LoadedAudio {
+ frames,
+ diagnostics: decode_result.diagnostics,
+ })
+}
+
+fn build_diagnostics(
+ source_kind: &'static str,
+ source_format: &'static str,
+ source_bytes: usize,
+ source_sample_rate_hz: u32,
+ source_num_channels: u16,
+ decoded_sample_count: usize,
+ target_samples: &[i16],
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+ mp3_skipped_data_count: usize,
+ mp3_insufficient_data_count: usize,
+) -> AudioDiagnostics {
+ let target_sample_count = target_samples.len();
+ let decoded_duration_ms = duration_ms(
+ decoded_sample_count,
+ source_sample_rate_hz,
+ source_num_channels,
+ );
+ let target_duration_ms = duration_ms(
+ target_sample_count,
+ target_sample_rate_hz,
+ target_num_channels,
+ );
+ let (rms, peak, clipped_sample_count, silence_ratio) = quality_stats(target_samples);
+ AudioDiagnostics {
+ source_kind,
+ source_format,
+ source_bytes,
+ source_sample_rate_hz,
+ source_num_channels,
+ decoded_sample_count,
+ decoded_duration_ms,
+ target_sample_rate_hz,
+ target_num_channels,
+ target_sample_count,
+ target_duration_ms,
+ frame_count: 0,
+ rms,
+ peak,
+ clipped_sample_count,
+ silence_ratio,
+ mp3_skipped_data_count,
+ mp3_insufficient_data_count,
+ debug_source_path: None,
+ debug_pcm_wav_path: None,
+ }
+}
+
+fn duration_ms(sample_count: usize, sample_rate_hz: u32, channels: u16) -> u64 {
+ if sample_count == 0 || sample_rate_hz == 0 || channels == 0 {
+ return 0;
+ }
+ let frames = sample_count as f64 / f64::from(channels.max(1));
+ ((frames / f64::from(sample_rate_hz)) * 1000.0).round() as u64
+}
+
+fn quality_stats(samples: &[i16]) -> (f64, f64, usize, f64) {
+ if samples.is_empty() {
+ return (0.0, 0.0, 0, 0.0);
+ }
+ const SILENCE_THRESHOLD: i32 = 256;
+ let mut sum_squares = 0f64;
+ let mut peak = 0i32;
+ let mut clipped = 0usize;
+ let mut silence = 0usize;
+ for &sample in samples {
+ let abs = i32::from(sample).abs();
+ peak = peak.max(abs);
+ if abs >= i32::from(i16::MAX) {
+ clipped += 1;
+ }
+ if abs <= SILENCE_THRESHOLD {
+ silence += 1;
+ }
+ let normalized = f64::from(sample) / f64::from(i16::MAX);
+ sum_squares += normalized * normalized;
+ }
+ let rms = (sum_squares / samples.len() as f64).sqrt();
+ let peak = f64::from(peak) / f64::from(i16::MAX);
+ let silence_ratio = silence as f64 / samples.len() as f64;
+ (round4(rms), round4(peak), clipped, round4(silence_ratio))
+}
+
+fn round4(value: f64) -> f64 {
+ (value * 10_000.0).round() / 10_000.0
+}
+
+fn dump_debug_audio(
+ dir: &str,
+ call_id: &str,
+ debug_label: &str,
+ source_format: &str,
+ source_bytes: &[u8],
+ target_samples: &[i16],
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+) -> Result<(Option<String>, Option<String>)> {
+ let base_dir = PathBuf::from(dir);
+ fs::create_dir_all(&base_dir)
+ .with_context(|| format!("failed to create audio debug dump dir {dir}"))?;
+ let safe_call_id = sanitize_file_segment(call_id);
+ let safe_label = sanitize_file_segment(debug_label);
+ let source_path = base_dir.join(format!(
+ "{safe_call_id}-{safe_label}-source.{source_format}"
+ ));
+ fs::write(&source_path, source_bytes).with_context(|| {
+ format!(
+ "failed to write audio debug source {}",
+ source_path.display()
+ )
+ })?;
+
+ let pcm_path = base_dir.join(format!("{safe_call_id}-{safe_label}-target.wav"));
+ write_debug_wav(
+ &pcm_path,
+ target_samples,
+ target_sample_rate_hz,
+ target_num_channels,
+ )?;
+
+ Ok((
+ Some(source_path.to_string_lossy().to_string()),
+ Some(pcm_path.to_string_lossy().to_string()),
+ ))
+}
+
+fn write_debug_wav(path: &Path, samples: &[i16], sample_rate_hz: u32, channels: u16) -> Result<()> {
+ let spec = WavSpec {
+ channels,
+ sample_rate: sample_rate_hz,
+ bits_per_sample: 16,
+ sample_format: SampleFormat::Int,
+ };
+ let mut writer = WavWriter::create(path, spec)
+ .with_context(|| format!("failed to create audio debug wav {}", path.display()))?;
+ for &sample in samples {
+ writer
+ .write_sample(sample)
+ .with_context(|| format!("failed to write audio debug wav {}", path.display()))?;
+ }
+ writer
+ .finalize()
+ .with_context(|| format!("failed to finalize audio debug wav {}", path.display()))
+}
+
+pub fn write_pcm_wav(
+ path: &Path,
+ samples: &[i16],
+ sample_rate_hz: u32,
+ channels: u16,
+) -> Result<()> {
+ write_debug_wav(path, samples, sample_rate_hz, channels)
+}
+
+fn sanitize_file_segment(value: &str) -> String {
+ let sanitized: String = value
+ .chars()
+ .map(|ch| {
+ if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.' {
+ ch
+ } else {
+ '_'
+ }
+ })
+ .collect();
+ if sanitized.is_empty() {
+ "unknown".to_string()
+ } else {
+ sanitized
+ }
+}
+
+fn chunk_pcm_samples(
+ samples: Vec<i16>,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+) -> Vec<PcmFrame> {
+ const CHUNK_MS: u32 = 20;
+ let samples_per_chunk = ((target_sample_rate_hz / 1000) * CHUNK_MS).max(1) as usize
+ * usize::from(target_num_channels);
+ let mut frames = Vec::new();
+ for chunk in samples.chunks(samples_per_chunk) {
+ let mut frame_data = chunk.to_vec();
+ if frame_data.len() < samples_per_chunk {
+ frame_data.resize(samples_per_chunk, 0);
+ }
+ let samples_per_channel =
+ (frame_data.len() / usize::from(target_num_channels.max(1))) as u32;
+ frames.push(PcmFrame::new(
+ frame_data,
+ target_sample_rate_hz,
+ u32::from(target_num_channels),
+ samples_per_channel,
+ ));
+ }
+ frames
+}
+
+fn is_wav(bytes: &[u8]) -> bool {
+ bytes.len() >= 12 && bytes.starts_with(b"RIFF") && &bytes[8..12] == b"WAVE"
+}
+
+fn is_mp3(bytes: &[u8]) -> bool {
+ bytes.starts_with(b"ID3") || (bytes.len() >= 2 && bytes[0] == 0xff && bytes[1] & 0xe0 == 0xe0)
+}
+
+fn remap_channels(samples: Vec<i16>, src_channels: u16, dst_channels: u16) -> Vec<i16> {
+ if src_channels == dst_channels {
+ return samples;
+ }
+
+ let src_channels = usize::from(src_channels.max(1));
+ let dst_channels = usize::from(dst_channels.max(1));
+ let frames = samples.chunks(src_channels);
+ let mut output = Vec::new();
+ for frame in frames {
+ match (src_channels, dst_channels) {
+ (1, 2) => {
+ let sample = frame.first().copied().unwrap_or_default();
+ output.push(sample);
+ output.push(sample);
+ }
+ (2, 1) => {
+ let left = frame.first().copied().unwrap_or_default() as i32;
+ let right = frame.get(1).copied().unwrap_or_default() as i32;
+ output.push(((left + right) / 2) as i16);
+ }
+ _ => {
+ for channel in 0..dst_channels {
+ output.push(
+ frame
+ .get(channel % src_channels)
+ .copied()
+ .unwrap_or_default(),
+ );
+ }
+ }
+ }
+ }
+ output
+}
+
+fn resample_linear(samples: Vec<i16>, src_rate: u32, dst_rate: u32, channels: u16) -> Vec<i16> {
+ if src_rate == dst_rate {
+ return samples;
+ }
+
+ let channels = usize::from(channels.max(1));
+ let src_frames = samples.len() / channels;
+ if src_frames <= 1 {
+ return samples;
+ }
+ let ratio = dst_rate as f64 / src_rate as f64;
+ let dst_frames = ((src_frames as f64) * ratio).round().max(1.0) as usize;
+ let mut output = Vec::with_capacity(dst_frames * channels);
+
+ for dst_frame in 0..dst_frames {
+ let src_pos = (dst_frame as f64) / ratio;
+ let src_index = src_pos.floor() as usize;
+ let next_index = (src_index + 1).min(src_frames - 1);
+ let frac = (src_pos - src_index as f64) as f32;
+ for channel in 0..channels {
+ let a = samples[src_index * channels + channel] as f32;
+ let b = samples[next_index * channels + channel] as f32;
+ let mixed = a + (b - a) * frac;
+ output.push(mixed.round().clamp(i16::MIN as f32, i16::MAX as f32) as i16);
+ }
+ }
+
+ output
+}
+
+#[cfg(test)]
+mod tests {
+ use super::load_pre_recorded_frames;
+
+ #[tokio::test]
+ async fn load_pre_recorded_frames_reads_local_wav() {
+ let dir = std::env::temp_dir().join("cv-runtime-helper-tests");
+ std::fs::create_dir_all(&dir).expect("create temp dir");
+ let path = dir.join("fixture.wav");
+ let spec = hound::WavSpec {
+ channels: 1,
+ sample_rate: 16_000,
+ bits_per_sample: 16,
+ sample_format: hound::SampleFormat::Int,
+ };
+ let mut writer = hound::WavWriter::create(&path, spec).expect("create wav");
+ for _ in 0..16_000 {
+ writer.write_sample(512_i16).expect("write sample");
+ }
+ writer.finalize().expect("finalize wav");
+
+ let http = reqwest::Client::new();
+ let loaded = load_pre_recorded_frames(
+ &http,
+ path.to_str(),
+ None,
+ 48_000,
+ 1,
+ Some(dir.to_string_lossy().as_ref()),
+ "test-call",
+ "greeting",
+ )
+ .await
+ .expect("load frames")
+ .expect("frames");
+
+ assert!(!loaded.frames.is_empty());
+ assert_eq!(loaded.frames[0].sample_rate, 48_000);
+ assert_eq!(loaded.frames[0].num_channels, 1);
+ assert_eq!(loaded.diagnostics.source_format, "wav");
+ assert_eq!(loaded.diagnostics.source_sample_rate_hz, 16_000);
+ assert_eq!(loaded.diagnostics.target_sample_rate_hz, 48_000);
+ assert!(loaded.diagnostics.debug_source_path.is_some());
+ assert!(loaded.diagnostics.debug_pcm_wav_path.is_some());
+ }
+}
diff --git a/src/main.rs b/src/main.rs
index 07f4e9f..9e99d6c 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,30 +1,53 @@
mod audio;
+mod service;
-use std::{env, sync::Arc, time::Duration};
+use std::{
+ env, fs,
+ path::{Path, PathBuf},
+ sync::{
+ Arc,
+ atomic::{AtomicBool, Ordering},
+ },
+ time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
use anyhow::{Context, Result, anyhow};
-use audio::load_pre_recorded_frames;
+use audio::{AudioDiagnostics, PcmFrame, load_pre_recorded_frames};
+use base64::{Engine as _, engine::general_purpose};
+use futures_util::StreamExt;
use libwebrtc::{
audio_source::native::NativeAudioSource,
+ audio_stream::native::NativeAudioStream,
prelude::{AudioFrame, AudioSourceOptions, RtcAudioSource},
};
use livekit::{
options::TrackPublishOptions,
- prelude::{DataPacket, LocalAudioTrack, LocalTrack, ParticipantIdentity, Room, RoomOptions},
+ prelude::{
+ DataPacket, LocalAudioTrack, LocalTrack, ParticipantIdentity, RemoteAudioTrack,
+ RemoteTrack, Room, RoomEvent, RoomOptions,
+ },
};
use reqwest::Client;
-use serde::Serialize;
-use tokio::time::sleep;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::time::{sleep, sleep_until};
+use tokio::{sync::mpsc::UnboundedReceiver, task::JoinHandle};
use tracing::{info, warn};
const TARGET_SAMPLE_RATE_HZ: u32 = 48_000;
const TARGET_NUM_CHANNELS: u16 = 1;
const TRACK_NAME: &str = "bot-main-audio";
-const DEVICE_OUTPUT_TOPIC: &str = "device_output";
#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
init_tracing();
+ if service::service_mode_enabled() {
+ return service::run_service().await;
+ }
+ run_worker().await
+}
+
+async fn run_worker() -> Result<()> {
let config = Config::from_env()?;
let http = Client::builder()
.use_rustls_tls()
@@ -37,10 +60,13 @@
config.greeting_audio_url.as_deref(),
TARGET_SAMPLE_RATE_HZ,
TARGET_NUM_CHANNELS,
+ config.audio_debug_dump_dir.as_deref(),
+ &config.call_id,
+ "greeting",
)
.await?;
- let (room, _events) = Room::connect(
+ let (room, events) = Room::connect(
config.livekit_url.as_str(),
config.bot_token.as_str(),
RoomOptions::default(),
@@ -52,42 +78,134 @@
info!(
call_id = %config.call_id,
trace_id = %config.trace_id,
- room_id = %config.room_id,
+ room_alias = %redact(&config.room_id),
participant_alias = %redact(&config.bot_participant_identity),
greeting_source = %config.greeting_source,
"combrabo voice runtime helper connected"
);
+ emit_activity(
+ &config.call_id,
+ &config.trace_id,
+ None,
+ "bot_participant_joined",
+ "ok",
+ None,
+ None,
+ json!({"participantAlias": redact(&config.bot_participant_identity)}),
+ );
+
+ let vad_enabled_gate = Arc::new(AtomicBool::new(
+ !config.simple_vad_enabled || !config.simple_vad_gate_until_greeting_done,
+ ));
+ if config.simple_vad_enabled && config.simple_vad_gate_until_greeting_done {
+ info!(
+ call_id = %config.call_id,
+ trace_id = %config.trace_id,
+ post_greeting_delay_ms = config.simple_vad_post_greeting_delay_ms,
+ "runtime helper vad_disabled_greeting"
+ );
+ } else if config.simple_vad_enabled {
+ info!(
+ call_id = %config.call_id,
+ trace_id = %config.trace_id,
+ "runtime helper vad_enabled"
+ );
+ emit_activity(
+ &config.call_id,
+ &config.trace_id,
+ None,
+ "vad_enabled",
+ "ok",
+ None,
+ None,
+ json!({"reason": "gate_disabled"}),
+ );
+ }
let sink = BotAudioOutputSink::publish(
room.clone(),
&config.room_id,
&config.bot_participant_identity,
+ &config.call_id,
+ &config.trace_id,
TRACK_NAME,
TARGET_SAMPLE_RATE_HZ,
u32::from(TARGET_NUM_CHANNELS),
+ config.user_participant_identity.clone(),
)
.await?;
+ let sink = Arc::new(sink);
- if config.device_output_smoke_enabled {
- publish_device_output_smoke(room.as_ref(), &config).await?;
- }
+ let user_audio_observer = spawn_user_audio_observer(
+ events,
+ &config,
+ vad_enabled_gate.clone(),
+ http.clone(),
+ sink.clone(),
+ );
- if let Some(frames) = greeting_frames {
+ if let Some(loaded_audio) = greeting_frames {
+ let diagnostics = loaded_audio.diagnostics;
+ let frames = loaded_audio.frames;
+ log_audio_diagnostics(&config, &diagnostics);
info!(
call_id = %config.call_id,
frame_count = frames.len(),
+ theoretical_duration_ms = diagnostics.target_duration_ms,
greeting_source = %config.greeting_source,
"runtime helper starting greeting playback"
);
- for frame in &frames {
+ emit_activity(
+ &config.call_id,
+ &config.trace_id,
+ None,
+ "greeting_write_started",
+ "ok",
+ None,
+ None,
+ json!({
+ "frameCount": frames.len(),
+ "greetingTheoreticalDurationMs": diagnostics.target_duration_ms,
+ }),
+ );
+ let playback_started_at = Instant::now();
+ let pacing_started_at = tokio::time::Instant::now();
+ for (index, frame) in frames.iter().enumerate() {
sink.write_pcm_frame(frame).await?;
- sleep(Duration::from_millis(20)).await;
+ sleep_until(pacing_started_at + Duration::from_millis(((index + 1) as u64) * 20)).await;
}
+ let playback_wall_ms = playback_started_at.elapsed().as_millis() as i64;
+ let drift_ms = playback_wall_ms - diagnostics.target_duration_ms as i64;
sink.clear_buffer();
info!(
call_id = %config.call_id,
greeting_source = %config.greeting_source,
+ frame_count = frames.len(),
+ theoretical_duration_ms = diagnostics.target_duration_ms,
+ push_wall_duration_ms = playback_wall_ms,
+ push_drift_ms = drift_ms,
"runtime helper finished greeting playback"
+ );
+ emit_activity(
+ &config.call_id,
+ &config.trace_id,
+ None,
+ "greeting_write_finished",
+ "ok",
+ None,
+ None,
+ json!({
+ "frameCount": frames.len(),
+ "greetingTheoreticalDurationMs": diagnostics.target_duration_ms,
+ "greetingPushWallDurationMs": playback_wall_ms,
+ "greetingPushDriftMs": drift_ms,
+ }),
+ );
+ schedule_vad_gate_enable(
+ vad_enabled_gate,
+ &config,
+ config.simple_vad_post_greeting_delay_ms,
+ "greeting_finished",
);
} else {
warn!(
@@ -95,9 +213,12 @@
greeting_source = %config.greeting_source,
"runtime helper started without greeting audio; keeping published track alive"
);
+ schedule_vad_gate_enable(vad_enabled_gate, &config, 0, "no_greeting_audio");
}
wait_for_shutdown_signal().await?;
+ user_audio_observer.abort();
+ let _ = user_audio_observer.await;
if let Err(error) = sink.close().await {
warn!(
@@ -133,12 +254,70 @@
room_id: String,
bot_token: String,
bot_participant_identity: String,
+ user_participant_identity: Option<String>,
greeting_source: String,
greeting_audio_file: Option<String>,
greeting_audio_url: Option<String>,
- role_id: String,
- device_output_smoke_enabled: bool,
- device_output_destination_identities: Vec<ParticipantIdentity>,
+ audio_debug_dump_dir: Option<String>,
+ runtime_turn_bridge_url: Option<String>,
+ runtime_turn_bridge_token: Option<String>,
+ runtime_turn_bridge_mode: String,
+ runtime_turn_artifact_dir: Option<String>,
+ runtime_session_nonce: Option<String>,
+ user_audio_observer_enabled: bool,
+ simple_vad_enabled: bool,
+ simple_vad_gate_until_greeting_done: bool,
+ simple_vad_post_greeting_delay_ms: u64,
+ simple_vad_config: SimpleVadConfig,
+}
+
+#[derive(Clone)]
+struct TurnBridgeConfig {
+ bridge_url: Option<String>,
+ bridge_token: Option<String>,
+ bridge_mode: String,
+ artifact_dir: Option<String>,
+ runtime_session_nonce: Option<String>,
+ audio_debug_dump_dir: Option<String>,
+}
+
+impl TurnBridgeConfig {
+ fn from_config(config: &Config) -> Self {
+ Self {
+ bridge_url: config.runtime_turn_bridge_url.clone(),
+ bridge_token: config.runtime_turn_bridge_token.clone(),
+ bridge_mode: config.runtime_turn_bridge_mode.clone(),
+ artifact_dir: config.runtime_turn_artifact_dir.clone(),
+ runtime_session_nonce: config.runtime_session_nonce.clone(),
+ audio_debug_dump_dir: config.audio_debug_dump_dir.clone(),
+ }
+ }
+
+ fn is_ready(&self) -> bool {
+ self.bridge_url
+ .as_ref()
+ .is_some_and(|value| !value.is_empty())
+ && self
+ .bridge_token
+ .as_ref()
+ .is_some_and(|value| !value.is_empty())
+ && self
+ .artifact_dir
+ .as_ref()
+ .is_some_and(|value| !value.is_empty())
+ && self
+ .runtime_session_nonce
+ .as_ref()
+ .is_some_and(|value| !value.is_empty())
+ }
+
+ fn is_stream_mode(&self) -> bool {
+ self.bridge_mode.eq_ignore_ascii_case("stream")
+ || self
+ .bridge_url
+ .as_deref()
+ .is_some_and(|value| value.trim_end_matches('/').ends_with("/stream"))
+ }
}
impl Config {
@@ -150,18 +329,115 @@
room_id: required_env("CV_LIVEKIT_ROOM_ID")?,
bot_token: required_env("CV_LIVEKIT_BOT_TOKEN")?,
bot_participant_identity: required_env("CV_LIVEKIT_BOT_PARTICIPANT_IDENTITY")?,
+ user_participant_identity: optional_env("CV_LIVEKIT_USER_PARTICIPANT_IDENTITY"),
greeting_source: env::var("CV_GREETING_SOURCE")
.unwrap_or_else(|_| "no_audio".to_string()),
greeting_audio_file: optional_env("CV_GREETING_AUDIO_FILE"),
greeting_audio_url: optional_env("CV_GREETING_AUDIO_URL"),
- role_id: env::var("CV_ROLE_ID").unwrap_or_else(|_| "90".to_string()),
- device_output_smoke_enabled: bool_env("CV_DEVICE_OUTPUT_SMOKE_ENABLED"),
- device_output_destination_identities: device_output_destinations(
- optional_env("CV_DEVICE_OUTPUT_DESTINATION_IDENTITIES"),
- optional_env("CV_LIVEKIT_USER_PARTICIPANT_IDENTITY"),
- ),
+ audio_debug_dump_dir: optional_env("CV_AUDIO_DEBUG_DUMP_DIR"),
+ runtime_turn_bridge_url: optional_env("CV_RUNTIME_TURN_BRIDGE_URL"),
+ runtime_turn_bridge_token: optional_env("CV_RUNTIME_TURN_BRIDGE_TOKEN"),
+ runtime_turn_bridge_mode: env::var("CV_RUNTIME_TURN_BRIDGE_MODE")
+ .unwrap_or_else(|_| "json".to_string()),
+ runtime_turn_artifact_dir: optional_env("CV_RUNTIME_TURN_ARTIFACT_DIR"),
+ runtime_session_nonce: optional_env("CV_RUNTIME_SESSION_NONCE"),
+ user_audio_observer_enabled: bool_env("CV_ENABLE_USER_AUDIO_OBSERVER", true),
+ simple_vad_enabled: bool_env("CV_ENABLE_SIMPLE_VAD", true),
+ simple_vad_gate_until_greeting_done: bool_env("CV_VAD_GATE_UNTIL_GREETING_DONE", true),
+ simple_vad_post_greeting_delay_ms: u64_env("CV_VAD_POST_GREETING_DELAY_MS", 800),
+ simple_vad_config: SimpleVadConfig::from_env(),
})
}
+}
+
+fn schedule_vad_gate_enable(
+ gate: Arc<AtomicBool>,
+ config: &Config,
+ delay_ms: u64,
+ reason: &'static str,
+) {
+ if !config.simple_vad_enabled || !config.simple_vad_gate_until_greeting_done {
+ return;
+ }
+
+ let call_id = config.call_id.clone();
+ let trace_id = config.trace_id.clone();
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ delay_ms,
+ reason,
+ "runtime helper vad_enable_scheduled"
+ );
+ emit_activity(
+ &call_id,
+ &trace_id,
+ None,
+ "vad_enable_scheduled",
+ "ok",
+ None,
+ None,
+ json!({
+ "vadEnableDelayMs": delay_ms,
+ "reason": reason,
+ }),
+ );
+
+ tokio::spawn(async move {
+ if delay_ms > 0 {
+ sleep(Duration::from_millis(delay_ms)).await;
+ }
+ gate.store(true, Ordering::Release);
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ delay_ms,
+ reason,
+ "runtime helper vad_enabled"
+ );
+ emit_activity(
+ &call_id,
+ &trace_id,
+ None,
+ "vad_enabled",
+ "ok",
+ None,
+ None,
+ json!({
+ "vadEnableDelayMs": delay_ms,
+ "reason": reason,
+ }),
+ );
+ });
+}
+
+fn log_audio_diagnostics(config: &Config, diagnostics: &AudioDiagnostics) {
+ info!(
+ call_id = %config.call_id,
+ trace_id = %config.trace_id,
+ greeting_source = %config.greeting_source,
+ source_kind = diagnostics.source_kind,
+ source_format = diagnostics.source_format,
+ source_bytes = diagnostics.source_bytes,
+ source_sample_rate_hz = diagnostics.source_sample_rate_hz,
+ source_num_channels = diagnostics.source_num_channels,
+ decoded_sample_count = diagnostics.decoded_sample_count,
+ decoded_duration_ms = diagnostics.decoded_duration_ms,
+ target_sample_rate_hz = diagnostics.target_sample_rate_hz,
+ target_num_channels = diagnostics.target_num_channels,
+ target_sample_count = diagnostics.target_sample_count,
+ target_duration_ms = diagnostics.target_duration_ms,
+ frame_count = diagnostics.frame_count,
+ rms = diagnostics.rms,
+ peak = diagnostics.peak,
+ clipped_sample_count = diagnostics.clipped_sample_count,
+ silence_ratio = diagnostics.silence_ratio,
+ mp3_skipped_data_count = diagnostics.mp3_skipped_data_count,
+ mp3_insufficient_data_count = diagnostics.mp3_insufficient_data_count,
+ debug_source_path = diagnostics.debug_source_path.as_deref().unwrap_or(""),
+ debug_pcm_wav_path = diagnostics.debug_pcm_wav_path.as_deref().unwrap_or(""),
+ "runtime helper greeting audio quality diagnostics"
+ );
}
fn required_env(key: &str) -> Result<String> {
@@ -183,34 +459,37 @@
})
}
-fn bool_env(key: &str) -> bool {
- matches!(
- env::var(key)
- .unwrap_or_default()
- .trim()
- .to_ascii_lowercase()
- .as_str(),
- "1" | "true" | "yes" | "y" | "on"
- )
+fn bool_env(key: &str, default_value: bool) -> bool {
+ match env::var(key) {
+ Ok(value) => match value.trim().to_ascii_lowercase().as_str() {
+ "1" | "true" | "yes" | "on" => true,
+ "0" | "false" | "no" | "off" => false,
+ _ => default_value,
+ },
+ Err(_) => default_value,
+ }
}
-fn device_output_destinations(
- configured: Option<String>,
- user_identity: Option<String>,
-) -> Vec<ParticipantIdentity> {
- let identities = configured
- .filter(|value| !value.trim().is_empty())
- .map(|value| {
- value
- .split(',')
- .map(str::trim)
- .filter(|identity| !identity.is_empty())
- .map(ToOwned::to_owned)
- .collect::<Vec<_>>()
- })
- .or_else(|| user_identity.map(|identity| vec![identity]))
- .unwrap_or_default();
- identities.into_iter().map(Into::into).collect()
+fn f64_env(key: &str, default_value: f64) -> f64 {
+ env::var(key)
+ .ok()
+ .and_then(|value| value.trim().parse::<f64>().ok())
+ .filter(|value| value.is_finite() && *value >= 0.0)
+ .unwrap_or(default_value)
+}
+
+fn u64_env(key: &str, default_value: u64) -> u64 {
+ env::var(key)
+ .ok()
+ .and_then(|value| value.trim().parse::<u64>().ok())
+ .unwrap_or(default_value)
+}
+
+fn u32_env(key: &str, default_value: u32) -> u32 {
+ env::var(key)
+ .ok()
+ .and_then(|value| value.trim().parse::<u32>().ok())
+ .unwrap_or(default_value)
}
fn redact(value: &str) -> String {
@@ -220,118 +499,2017 @@
format!("{}***{}", &value[..4], &value[value.len() - 4..])
}
-#[derive(Serialize)]
-struct DeviceOutputMessage {
- #[serde(rename = "type")]
- message_type: &'static str,
- #[serde(rename = "schemaVersion")]
- schema_version: &'static str,
- #[serde(rename = "callId")]
+fn safe_error(value: &str) -> String {
+ let sanitized = value.replace(['\r', '\n'], " ");
+ let trimmed = sanitized.trim();
+ let mut output: String = trimmed.chars().take(180).collect();
+ if trimmed.chars().count() > 180 {
+ output.push_str("...");
+ }
+ output
+}
+
+fn current_time_millis() -> u64 {
+ SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .map(|value| value.as_millis() as u64)
+ .unwrap_or_default()
+}
+
+fn emit_activity(
+ call_id: &str,
+ trace_id: &str,
+ turn_id: Option<&str>,
+ event_name: &str,
+ result: &str,
+ reason_code: Option<&str>,
+ retryable: Option<bool>,
+ extension: serde_json::Value,
+) {
+ let payload = json!({
+ "type": "cv_activity",
+ "callId": call_id,
+ "traceId": trace_id,
+ "turnId": turn_id,
+ "eventName": event_name,
+ "result": result,
+ "reasonCode": reason_code,
+ "retryable": retryable,
+ "extension": extension,
+ });
+ println!("{payload}");
+}
+
+fn spawn_user_audio_observer(
+ events: UnboundedReceiver<RoomEvent>,
+ config: &Config,
+ vad_enabled_gate: Arc<AtomicBool>,
+ http: Client,
+ sink: Arc<BotAudioOutputSink>,
+) -> JoinHandle<()> {
+ let call_id = config.call_id.clone();
+ let trace_id = config.trace_id.clone();
+ let enabled = config.user_audio_observer_enabled;
+ let simple_vad_enabled = config.simple_vad_enabled;
+ let simple_vad_config = config.simple_vad_config.clone();
+ let turn_bridge_config = TurnBridgeConfig::from_config(config);
+
+ tokio::spawn(async move {
+ if !enabled {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ "runtime helper user audio observer disabled"
+ );
+ return;
+ }
+ observe_user_audio_events(
+ events,
+ call_id,
+ trace_id,
+ simple_vad_enabled,
+ simple_vad_config,
+ vad_enabled_gate,
+ turn_bridge_config,
+ http,
+ sink,
+ )
+ .await;
+ })
+}
+
+async fn observe_user_audio_events(
+ mut events: UnboundedReceiver<RoomEvent>,
call_id: String,
- #[serde(rename = "roleId")]
- role_id: String,
- #[serde(rename = "traceId")]
trace_id: String,
- #[serde(rename = "ackMode")]
- ack_mode: &'static str,
- #[serde(rename = "sensorInstructions")]
- sensor_instructions: Vec<SensorInstruction>,
+ simple_vad_enabled: bool,
+ simple_vad_config: SimpleVadConfig,
+ vad_enabled_gate: Arc<AtomicBool>,
+ turn_bridge_config: TurnBridgeConfig,
+ http: Client,
+ sink: Arc<BotAudioOutputSink>,
+) {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ simple_vad_enabled,
+ vad_rms_threshold = simple_vad_config.rms_threshold,
+ vad_peak_threshold = simple_vad_config.peak_threshold,
+ vad_start_frames = simple_vad_config.start_frames,
+ vad_end_silence_ms = simple_vad_config.end_silence_ms,
+ vad_min_speech_ms = simple_vad_config.min_speech_ms,
+ vad_max_turn_ms = simple_vad_config.max_turn_ms,
+ vad_initial_ignore_ms = simple_vad_config.initial_ignore_ms,
+ vad_gate_enabled = vad_enabled_gate.load(Ordering::Acquire),
+ "runtime helper user_track_subscribe_requested"
+ );
+
+ while let Some(event) = events.recv().await {
+ match event {
+ RoomEvent::TrackSubscribed {
+ track: RemoteTrack::Audio(track),
+ publication: _,
+ participant,
+ } => {
+ let participant_alias = redact(&participant.identity().to_string());
+ let track_sid_alias = redact(&track.sid().to_string());
+ let track_name = track.name();
+ let track_source = format!("{:?}", track.source());
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ track_name = %track_name,
+ track_source = %track_source,
+ "runtime helper user_track_subscribed"
+ );
+ spawn_user_audio_frame_observer(
+ track,
+ call_id.clone(),
+ trace_id.clone(),
+ participant_alias,
+ track_sid_alias,
+ simple_vad_enabled,
+ simple_vad_config.clone(),
+ vad_enabled_gate.clone(),
+ turn_bridge_config.clone(),
+ http.clone(),
+ sink.clone(),
+ );
+ }
+ RoomEvent::TrackSubscribed {
+ track: RemoteTrack::Video(track),
+ publication: _,
+ participant,
+ } => {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %redact(&participant.identity().to_string()),
+ track_sid_alias = %redact(&track.sid().to_string()),
+ "runtime helper ignored non-audio subscribed track"
+ );
+ }
+ RoomEvent::TrackSubscriptionFailed {
+ participant,
+ error,
+ track_sid,
+ } => {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %redact(&participant.identity().to_string()),
+ track_sid_alias = %redact(&track_sid.to_string()),
+ error = %error,
+ "runtime helper user_track_subscription_failed"
+ );
+ }
+ RoomEvent::Disconnected { reason } => {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ reason = ?reason,
+ "runtime helper room event stream disconnected"
+ );
+ break;
+ }
+ _ => {}
+ }
+ }
}
-#[derive(Serialize)]
-struct SensorInstruction {
- #[serde(rename = "commandId")]
- command_id: String,
- #[serde(rename = "sensorType")]
- sensor_type: &'static str,
- #[serde(rename = "operationType")]
- operation_type: &'static str,
- step: i32,
- #[serde(rename = "durationSec", skip_serializing_if = "Option::is_none")]
- duration_sec: Option<i32>,
- extension: String,
+async fn handle_finished_turn(
+ http: &Client,
+ bridge_config: &TurnBridgeConfig,
+ sink: &BotAudioOutputSink,
+ call_id: &str,
+ trace_id: &str,
+ turn: FinishedSpeechTurn,
+) {
+ let turn_pipeline_started_at = Instant::now();
+ if !bridge_config.is_ready() {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ turn_index = turn.turn_index,
+ duration_ms = turn.duration_ms,
+ "runtime helper turn_bridge_skipped"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_bridge_skipped",
+ "skipped",
+ Some("TURN_BRIDGE_NOT_CONFIGURED"),
+ Some(true),
+ json!({
+ "durationMs": turn.duration_ms,
+ "frameCount": turn.frame_count,
+ "sampleCount": turn.sample_count,
+ }),
+ );
+ return;
+ }
+
+ let artifact_root = PathBuf::from(bridge_config.artifact_dir.as_deref().unwrap_or_default());
+ match write_user_turn_artifact(&artifact_root, call_id, &turn) {
+ Ok((path_ref, byte_size)) => {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ turn_index = turn.turn_index,
+ duration_ms = turn.duration_ms,
+ frame_count = turn.frame_count,
+ sample_count = turn.sample_count,
+ byte_size,
+ end_reason = %turn.end_reason,
+ "runtime helper turn_artifact_written"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_bridge_requested",
+ "ok",
+ None,
+ None,
+ json!({
+ "turnDurationMs": turn.duration_ms,
+ "turnArtifactBytes": byte_size,
+ "frameCount": turn.frame_count,
+ "sampleCount": turn.sample_count,
+ "endReason": turn.end_reason,
+ }),
+ );
+ if bridge_config.is_stream_mode() {
+ match request_turn_bridge_stream(
+ http,
+ bridge_config,
+ sink,
+ call_id,
+ trace_id,
+ &turn,
+ &path_ref,
+ byte_size,
+ turn_pipeline_started_at,
+ )
+ .await
+ {
+ Ok(outcome) => {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ audio_chunk_count = outcome.audio_chunk_count,
+ device_output_count = outcome.device_output_count,
+ "runtime helper turn_stream_completed"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_completed",
+ "ok",
+ None,
+ None,
+ json!({
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ "audioChunkCount": outcome.audio_chunk_count,
+ "deviceOutputCount": outcome.device_output_count,
+ "replyPlaybackMode": outcome.reply_playback_mode,
+ }),
+ );
+ }
+ Err(error) => {
+ let safe = safe_error(&error.to_string());
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ error = %safe,
+ "runtime helper turn_stream_failed"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_failed",
+ "failed",
+ Some("TURN_STREAM_FAILED"),
+ Some(true),
+ json!({
+ "stage": "turn_bridge_stream",
+ "error": safe,
+ }),
+ );
+ }
+ }
+ return;
+ }
+ match request_turn_bridge(
+ http,
+ bridge_config,
+ call_id,
+ trace_id,
+ &turn,
+ &path_ref,
+ byte_size,
+ turn_pipeline_started_at,
+ )
+ .await
+ {
+ Ok(outcome) => {
+ for output in &outcome.device_outputs {
+ if let Err(error) = sink
+ .publish_device_output(call_id, trace_id, &turn.turn_id, output)
+ .await
+ {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ reason_code = "BOT_DATA_WRITE_FAILED",
+ error = %safe_error(&error.to_string()),
+ "runtime helper device_output_failed"
+ );
+ }
+ }
+ let Some(reply_audio_artifact) = outcome.reply_audio_artifact else {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ reason_code = "REPLY_AUDIO_MISSING",
+ "runtime helper turn_failed"
+ );
+ return;
+ };
+ if let Err(error) = write_reply_audio_artifact(
+ http,
+ bridge_config,
+ sink,
+ call_id,
+ trace_id,
+ &turn,
+ &reply_audio_artifact,
+ turn_pipeline_started_at,
+ )
+ .await
+ {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ reason_code = "BOT_AUDIO_WRITE_FAILED",
+ error = %safe_error(&error.to_string()),
+ "runtime helper turn_failed"
+ );
+ return;
+ }
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ "runtime helper turn_completed"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_completed",
+ "ok",
+ None,
+ None,
+ json!({
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ }),
+ );
+ }
+ Err(error) => warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ error = %safe_error(&error.to_string()),
+ "runtime helper turn_bridge_failed"
+ ),
+ }
+ }
+ Err(error) => warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ error = %safe_error(&error.to_string()),
+ "runtime helper turn_artifact_write_failed"
+ ),
+ }
}
-async fn publish_device_output_smoke(room: &Room, config: &Config) -> Result<()> {
- let payload = build_device_output_smoke_payload(config);
- let payload_json =
- serde_json::to_vec(&payload).context("failed to serialize device output smoke payload")?;
- let destination_count = config.device_output_destination_identities.len();
- room.local_participant()
- .publish_data(DataPacket {
- reliable: true,
- payload: payload_json,
- topic: Some(DEVICE_OUTPUT_TOPIC.to_string()),
- destination_identities: config.device_output_destination_identities.clone(),
- })
+fn write_user_turn_artifact(
+ artifact_root: &Path,
+ call_id: &str,
+ turn: &FinishedSpeechTurn,
+) -> Result<(String, u64)> {
+ require_safe_segment(call_id)?;
+ require_safe_segment(&turn.turn_id)?;
+ if turn.samples.is_empty() {
+ return Err(anyhow!("empty turn samples"));
+ }
+ let path_ref = format!("{}/{}/user.wav", call_id, turn.turn_id);
+ let output_path = normalize_path_lexically(&artifact_root.join(&path_ref));
+ let root = normalize_path_lexically(artifact_root);
+ if !output_path.starts_with(&root) {
+ return Err(anyhow!("turn artifact path escapes root"));
+ }
+ if let Some(parent) = output_path.parent() {
+ fs::create_dir_all(parent).context("failed to create turn artifact dir")?;
+ }
+ audio::write_pcm_wav(
+ &output_path,
+ &turn.samples,
+ TARGET_SAMPLE_RATE_HZ,
+ TARGET_NUM_CHANNELS,
+ )?;
+ let byte_size = fs::metadata(&output_path)
+ .context("failed to stat turn artifact")?
+ .len();
+ Ok((path_ref, byte_size))
+}
+
+async fn request_turn_bridge_stream(
+ http: &Client,
+ bridge_config: &TurnBridgeConfig,
+ sink: &BotAudioOutputSink,
+ call_id: &str,
+ trace_id: &str,
+ turn: &FinishedSpeechTurn,
+ path_ref: &str,
+ byte_size: u64,
+ turn_pipeline_started_at: Instant,
+) -> Result<RuntimeTurnStreamOutcome> {
+ let bridge_started_at = Instant::now();
+ let request = RuntimeTurnRequest {
+ call_id: call_id.to_string(),
+ trace_id: trace_id.to_string(),
+ turn_id: turn.turn_id.clone(),
+ audio_artifact: RuntimeTurnAudioArtifact {
+ artifact_type: "local_file".to_string(),
+ path_ref: path_ref.to_string(),
+ format: "wav".to_string(),
+ sample_rate: TARGET_SAMPLE_RATE_HZ,
+ channels: u32::from(TARGET_NUM_CHANNELS),
+ duration_ms: turn.duration_ms,
+ byte_size,
+ },
+ };
+ let response = http
+ .post(bridge_config.bridge_url.as_deref().unwrap_or_default())
+ .header(
+ "X-CV-Runtime-Token",
+ bridge_config.bridge_token.as_deref().unwrap_or_default(),
+ )
+ .header("X-CV-Call-Id", call_id)
+ .header("X-CV-Trace-Id", trace_id)
+ .header(
+ "X-CV-Runtime-Session-Nonce",
+ bridge_config
+ .runtime_session_nonce
+ .as_deref()
+ .unwrap_or_default(),
+ )
+ .json(&request)
+ .send()
.await
- .map_err(|error| anyhow!("failed to publish livekit device output data: {error}"))?;
+ .context("failed to post turn stream bridge")?;
+ let status = response.status();
+ if !status.is_success() {
+ let body_len = response.text().await.map(|body| body.len()).unwrap_or(0);
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ http_status = status.as_u16(),
+ body_len,
+ "runtime helper turn_stream_http_failed"
+ );
+ return Err(anyhow!("turn stream bridge http failed"));
+ }
info!(
- call_id = %config.call_id,
- trace_id = %config.trace_id,
- topic = DEVICE_OUTPUT_TOPIC,
- reliable = true,
- command_count = payload.sensor_instructions.len(),
- destination_count,
- "runtime helper published device output smoke data"
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ "runtime helper turn_stream_connected"
+ );
+ let mut byte_stream = response.bytes_stream();
+ let mut line_buffer: Vec<u8> = Vec::new();
+ let mut state = RuntimeTurnStreamState::default();
+ while let Some(chunk) = byte_stream.next().await {
+ let chunk = chunk.context("failed to read turn stream chunk")?;
+ line_buffer.extend_from_slice(&chunk);
+ while let Some(newline_index) = line_buffer.iter().position(|value| *value == b'\n') {
+ let line: Vec<u8> = line_buffer.drain(..=newline_index).collect();
+ if let Some(event) = parse_turn_stream_event_line(&line)? {
+ handle_turn_stream_event(
+ bridge_config,
+ sink,
+ call_id,
+ trace_id,
+ turn,
+ event,
+ &mut state,
+ turn_pipeline_started_at,
+ )
+ .await?;
+ }
+ }
+ }
+ if !line_buffer.is_empty() {
+ if let Some(event) = parse_turn_stream_event_line(&line_buffer)? {
+ handle_turn_stream_event(
+ bridge_config,
+ sink,
+ call_id,
+ trace_id,
+ turn,
+ event,
+ &mut state,
+ turn_pipeline_started_at,
+ )
+ .await?;
+ }
+ }
+ if !state.completed {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ audio_chunk_count = state.audio_chunk_count,
+ "runtime helper turn_stream_completed_without_final_event"
+ );
+ return Err(anyhow!("turn stream ended without turn_completed"));
+ }
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_bridge_completed",
+ "ok",
+ None,
+ None,
+ json!({
+ "bridgeWallDurationMs": bridge_started_at.elapsed().as_millis() as u64,
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ "audioChunkCount": state.audio_chunk_count,
+ "deviceOutputCount": state.device_output_count,
+ "replyPlaybackMode": state.reply_playback_mode.as_str(),
+ }),
+ );
+ Ok(RuntimeTurnStreamOutcome {
+ reply_playback_mode: state.reply_playback_mode,
+ audio_chunk_count: state.audio_chunk_count,
+ device_output_count: state.device_output_count,
+ })
+}
+
+fn parse_turn_stream_event_line(line: &[u8]) -> Result<Option<RuntimeTurnStreamEvent>> {
+ let line = trim_ascii_whitespace(line);
+ if line.is_empty() {
+ return Ok(None);
+ }
+ serde_json::from_slice(line)
+ .context("failed to parse turn stream event")
+ .map(Some)
+}
+
+fn diagnostic_str<'a>(diagnostics: Option<&'a serde_json::Value>, key: &str) -> Option<&'a str> {
+ diagnostics?.get(key)?.as_str()
+}
+
+fn diagnostic_bool(diagnostics: Option<&serde_json::Value>, key: &str) -> Option<bool> {
+ diagnostics?.get(key)?.as_bool()
+}
+
+fn diagnostic_u64(diagnostics: Option<&serde_json::Value>, key: &str) -> Option<u64> {
+ diagnostics?.get(key)?.as_u64()
+}
+
+async fn handle_turn_stream_event(
+ bridge_config: &TurnBridgeConfig,
+ sink: &BotAudioOutputSink,
+ call_id: &str,
+ trace_id: &str,
+ turn: &FinishedSpeechTurn,
+ event: RuntimeTurnStreamEvent,
+ state: &mut RuntimeTurnStreamState,
+ turn_pipeline_started_at: Instant,
+) -> Result<()> {
+ let event_type = event.event_type();
+ match event_type.as_deref() {
+ Some("reply_playback_mode_selected") => {
+ if let Some(reply_playback_mode) = event.reply_playback_mode.as_deref() {
+ state.reply_playback_mode = reply_playback_mode.to_string();
+ }
+ let diagnostics = event.diagnostics.as_ref();
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ reply_playback_mode = %state.reply_playback_mode,
+ streaming_enabled = ?diagnostic_bool(diagnostics, "streamingEnabled"),
+ provider_streaming_supported = ?diagnostic_bool(diagnostics, "providerStreamingSupported"),
+ first_chunk_received = ?diagnostic_bool(diagnostics, "firstChunkReceived"),
+ stream_chunk_count = ?diagnostic_u64(diagnostics, "streamChunkCount"),
+ fallback_reason = %diagnostic_str(diagnostics, "fallbackReason").unwrap_or("none"),
+ fallback_stage = %diagnostic_str(diagnostics, "fallbackStage").unwrap_or("none"),
+ stream_bridge_mode = %diagnostic_str(diagnostics, "streamBridgeMode").unwrap_or("unknown"),
+ tts_provider = %diagnostic_str(diagnostics, "ttsProvider").unwrap_or("unknown"),
+ "runtime helper reply_playback_mode_selected"
+ );
+ }
+ Some("reply_state") => {
+ if let Some(reply_state) = event.state.as_deref() {
+ publish_reply_state_from_stream_event(
+ sink,
+ call_id,
+ trace_id,
+ &turn.turn_id,
+ &state.reply_playback_mode,
+ reply_state,
+ event.seq,
+ )
+ .await?;
+ if reply_state == "reply_playback_started" {
+ state.playback_started_sent = true;
+ }
+ }
+ }
+ Some("reply_audio_chunk") => {
+ let audio_chunk = event
+ .audio_chunk
+ .as_ref()
+ .ok_or_else(|| anyhow!("reply_audio_chunk event missing audioChunk"))?;
+ if !state.playback_started_sent {
+ state.reply_state_seq = state.reply_state_seq.saturating_add(1);
+ sink.publish_reply_state(
+ call_id,
+ trace_id,
+ &turn.turn_id,
+ &state.reply_playback_mode,
+ "reply_playback_started",
+ state.reply_state_seq,
+ )
+ .await?;
+ state.playback_started_sent = true;
+ }
+ let written_frames = write_stream_audio_chunk(
+ bridge_config,
+ sink,
+ call_id,
+ trace_id,
+ turn,
+ audio_chunk,
+ state,
+ turn_pipeline_started_at,
+ )
+ .await?;
+ if written_frames > 0 {
+ state.audio_chunk_count = state.audio_chunk_count.saturating_add(1);
+ }
+ }
+ Some("device_output") => {
+ if let Some(output) = event.device_output.as_ref() {
+ sink.publish_device_output(call_id, trace_id, &turn.turn_id, output)
+ .await?;
+ state.device_output_count = state.device_output_count.saturating_add(1);
+ }
+ }
+ Some("turn_completed") => {
+ state.completed = true;
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ message_id_alias = %event.completion.as_ref().and_then(|value| value.message_id.as_deref()).map(redact).unwrap_or_else(|| "none".to_string()),
+ audio_chunk_count = state.audio_chunk_count,
+ "runtime helper turn_stream_final_received"
+ );
+ }
+ Some("turn_failed") => {
+ let error = event.error.as_ref();
+ let reason_code = error
+ .and_then(|value| value.reason_code.as_deref())
+ .unwrap_or("TURN_STREAM_FAILED");
+ let stage = error
+ .and_then(|value| value.stage.as_deref())
+ .unwrap_or("turn_bridge_stream");
+ let retryable = error.and_then(|value| value.retryable).unwrap_or(false);
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ reason_code = %reason_code,
+ stage = %stage,
+ retryable = retryable,
+ "runtime helper turn_stream_failed_event"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_failed",
+ "failed",
+ Some(reason_code),
+ Some(retryable),
+ json!({
+ "stage": stage,
+ "replyPlaybackMode": state.reply_playback_mode.as_str(),
+ }),
+ );
+ return Err(anyhow!("turn stream failed event"));
+ }
+ Some("turn_cancelled") => {
+ state.completed = true;
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ "runtime helper turn_stream_cancelled_event"
+ );
+ publish_reply_state_from_stream_event(
+ sink,
+ call_id,
+ trace_id,
+ &turn.turn_id,
+ &state.reply_playback_mode,
+ "reply_playback_cancelled",
+ event.seq,
+ )
+ .await?;
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_cancelled",
+ "ok",
+ None,
+ Some(false),
+ json!({
+ "replyPlaybackMode": state.reply_playback_mode.as_str(),
+ "audioChunkCount": state.audio_chunk_count,
+ }),
+ );
+ }
+ Some("activity") => {
+ if let Some(activity) = event.activity.as_ref() {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ activity_event = %activity.event_type.as_deref().unwrap_or("unknown"),
+ stage = %activity.stage.as_deref().unwrap_or("unknown"),
+ reason_code = %activity.reason_code.as_deref().unwrap_or("none"),
+ "runtime helper turn_stream_activity"
+ );
+ }
+ }
+ Some(other) => {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ event_type = %other,
+ "runtime helper ignored turn stream event"
+ );
+ }
+ None => {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ "runtime helper ignored turn stream event without type"
+ );
+ }
+ }
+ Ok(())
+}
+
+async fn publish_reply_state_from_stream_event(
+ sink: &BotAudioOutputSink,
+ call_id: &str,
+ trace_id: &str,
+ turn_id: &str,
+ reply_playback_mode: &str,
+ state: &str,
+ seq: Option<u64>,
+) -> Result<()> {
+ sink.publish_reply_state(
+ call_id,
+ trace_id,
+ turn_id,
+ reply_playback_mode,
+ state,
+ seq.unwrap_or(0),
+ )
+ .await
+}
+
+async fn write_stream_audio_chunk(
+ bridge_config: &TurnBridgeConfig,
+ sink: &BotAudioOutputSink,
+ call_id: &str,
+ trace_id: &str,
+ turn: &FinishedSpeechTurn,
+ audio_chunk: &RuntimeTurnStreamAudioChunk,
+ state: &mut RuntimeTurnStreamState,
+ turn_pipeline_started_at: Instant,
+) -> Result<usize> {
+ let payload_base64 = audio_chunk
+ .payload_base64
+ .as_deref()
+ .map(str::trim)
+ .filter(|value| !value.is_empty())
+ .ok_or_else(|| anyhow!("reply_audio_chunk payloadBase64 missing"))?;
+ let payload = general_purpose::STANDARD
+ .decode(payload_base64)
+ .context("failed to decode reply_audio_chunk payloadBase64")?;
+ let format = audio_chunk
+ .format
+ .as_deref()
+ .unwrap_or("pcm_s16le")
+ .trim()
+ .to_ascii_lowercase();
+ let frames = if format == "pcm_s16le" {
+ pcm_s16le_payload_to_frames(
+ &payload,
+ audio_chunk.sample_rate.unwrap_or(TARGET_SAMPLE_RATE_HZ),
+ audio_chunk
+ .channels
+ .unwrap_or(u32::from(TARGET_NUM_CHANNELS)),
+ )?
+ } else if matches!(format.as_str(), "mp3" | "mpeg" | "wav") {
+ state.encoded_audio_buffer.extend_from_slice(&payload);
+ match audio::decode_audio_bytes_to_frames(
+ &state.encoded_audio_buffer,
+ "stream_chunk",
+ TARGET_SAMPLE_RATE_HZ,
+ TARGET_NUM_CHANNELS,
+ bridge_config.audio_debug_dump_dir.as_deref(),
+ call_id,
+ &format!("stream-reply-{}", turn.turn_id),
+ ) {
+ Ok(loaded) => {
+ state.encoded_audio_buffer.clear();
+ loaded.frames
+ }
+ Err(error) if !audio_chunk.last.unwrap_or(false) => {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ chunk_seq = audio_chunk.chunk_seq.unwrap_or_default(),
+ format = %format,
+ error = %safe_error(&error.to_string()),
+ "runtime helper stream_audio_chunk_decode_waiting_for_more_data"
+ );
+ return Ok(0);
+ }
+ Err(error) => return Err(error).context("failed to decode final stream audio chunk"),
+ }
+ } else {
+ return Err(anyhow!("unsupported reply_audio_chunk format {format}"));
+ };
+ if frames.is_empty() {
+ return Ok(0);
+ }
+
+ if !state.first_audio_frame_written {
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "bot_reply_audio_write_started",
+ "ok",
+ None,
+ None,
+ json!({
+ "replyPlaybackMode": state.reply_playback_mode.as_str(),
+ "format": format.as_str(),
+ "chunkSeq": audio_chunk.chunk_seq,
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ }),
+ );
+ }
+ let pacing_started_at = tokio::time::Instant::now();
+ for (index, frame) in frames.iter().enumerate() {
+ sink.write_pcm_frame(frame).await?;
+ if !state.first_audio_frame_written {
+ state.first_audio_frame_written = true;
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "bot_reply_first_audio_frame_written",
+ "ok",
+ None,
+ None,
+ json!({
+ "replyPlaybackMode": state.reply_playback_mode.as_str(),
+ "format": format.as_str(),
+ "chunkSeq": audio_chunk.chunk_seq,
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ }),
+ );
+ }
+ sleep_until(pacing_started_at + Duration::from_millis(((index + 1) as u64) * 20)).await;
+ }
+ if audio_chunk.last.unwrap_or(false) {
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "bot_reply_audio_write_finished",
+ "ok",
+ None,
+ None,
+ json!({
+ "replyPlaybackMode": state.reply_playback_mode.as_str(),
+ "format": format.as_str(),
+ "chunkSeq": audio_chunk.chunk_seq,
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ }),
+ );
+ }
+ Ok(frames.len())
+}
+
+fn pcm_s16le_payload_to_frames(
+ payload: &[u8],
+ sample_rate: u32,
+ channels: u32,
+) -> Result<Vec<PcmFrame>> {
+ if sample_rate != TARGET_SAMPLE_RATE_HZ || channels != u32::from(TARGET_NUM_CHANNELS) {
+ return Err(anyhow!(
+ "unsupported pcm_s16le stream format: sample_rate={}, channels={}",
+ sample_rate,
+ channels
+ ));
+ }
+ if payload.len() % 2 != 0 {
+ return Err(anyhow!("pcm_s16le payload has odd byte length"));
+ }
+ let samples: Vec<i16> = payload
+ .chunks_exact(2)
+ .map(|chunk| i16::from_le_bytes([chunk[0], chunk[1]]))
+ .collect();
+ if samples.is_empty() {
+ return Ok(Vec::new());
+ }
+ let samples_per_channel = (sample_rate / 50).max(1);
+ let frame_sample_count = samples_per_channel as usize * channels as usize;
+ let mut frames = Vec::new();
+ for chunk in samples.chunks(frame_sample_count) {
+ let chunk_samples_per_channel = (chunk.len() / channels as usize) as u32;
+ if chunk_samples_per_channel == 0 {
+ continue;
+ }
+ frames.push(PcmFrame::new(
+ chunk.to_vec(),
+ sample_rate,
+ channels,
+ chunk_samples_per_channel,
+ ));
+ }
+ Ok(frames)
+}
+
+fn trim_ascii_whitespace(value: &[u8]) -> &[u8] {
+ let mut start = 0;
+ let mut end = value.len();
+ while start < end && value[start].is_ascii_whitespace() {
+ start += 1;
+ }
+ while end > start && value[end - 1].is_ascii_whitespace() {
+ end -= 1;
+ }
+ &value[start..end]
+}
+
+async fn request_turn_bridge(
+ http: &Client,
+ bridge_config: &TurnBridgeConfig,
+ call_id: &str,
+ trace_id: &str,
+ turn: &FinishedSpeechTurn,
+ path_ref: &str,
+ byte_size: u64,
+ turn_pipeline_started_at: Instant,
+) -> Result<RuntimeTurnBridgeOutcome> {
+ let bridge_started_at = Instant::now();
+ let request = RuntimeTurnRequest {
+ call_id: call_id.to_string(),
+ trace_id: trace_id.to_string(),
+ turn_id: turn.turn_id.clone(),
+ audio_artifact: RuntimeTurnAudioArtifact {
+ artifact_type: "local_file".to_string(),
+ path_ref: path_ref.to_string(),
+ format: "wav".to_string(),
+ sample_rate: TARGET_SAMPLE_RATE_HZ,
+ channels: u32::from(TARGET_NUM_CHANNELS),
+ duration_ms: turn.duration_ms,
+ byte_size,
+ },
+ };
+ let response = http
+ .post(bridge_config.bridge_url.as_deref().unwrap_or_default())
+ .header(
+ "X-CV-Runtime-Token",
+ bridge_config.bridge_token.as_deref().unwrap_or_default(),
+ )
+ .header("X-CV-Call-Id", call_id)
+ .header("X-CV-Trace-Id", trace_id)
+ .header(
+ "X-CV-Runtime-Session-Nonce",
+ bridge_config
+ .runtime_session_nonce
+ .as_deref()
+ .unwrap_or_default(),
+ )
+ .json(&request)
+ .send()
+ .await
+ .context("failed to post turn bridge")?;
+ let status = response.status();
+ let body = response
+ .text()
+ .await
+ .context("failed to read turn bridge response")?;
+ let body_len = body.len();
+ let parsed: Option<RuntimeTurnCommonResult<RuntimeTurnResponseData>> =
+ serde_json::from_str(&body).ok();
+ if !status.is_success() {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ http_status = status.as_u16(),
+ body_len,
+ code = parsed.as_ref().map(|value| value.code).unwrap_or_default(),
+ reason_code = %parsed.as_ref().and_then(|value| value.msg.as_deref()).unwrap_or("unknown"),
+ "runtime helper turn_bridge_http_failed"
+ );
+ return Err(anyhow!("turn bridge http failed"));
+ }
+ let Some(result) = parsed else {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ body_len,
+ "runtime helper turn_bridge_response_invalid"
+ );
+ return Err(anyhow!("turn bridge response invalid"));
+ };
+ if result.code != 0 {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ code = result.code,
+ reason_code = %result.msg.as_deref().unwrap_or("unknown"),
+ retryable = result.retryable.unwrap_or(false),
+ stage = %result.stage.as_deref().unwrap_or("unknown"),
+ "runtime helper turn_bridge_business_failed"
+ );
+ return Err(anyhow!("turn bridge business failed"));
+ }
+ let data = result.data;
+ let reply_audio_artifact = data
+ .as_ref()
+ .and_then(|value| value.reply_audio_artifact.as_ref());
+ let device_outputs = data
+ .as_ref()
+ .and_then(|value| value.device_outputs.clone())
+ .unwrap_or_default();
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ response_turn_id = %data.as_ref().and_then(|value| value.turn_id.as_deref()).unwrap_or("unknown"),
+ message_id_alias = %data.as_ref().and_then(|value| value.message_id.as_deref()).map(redact).unwrap_or_else(|| "none".to_string()),
+ reply_audio_present = reply_audio_artifact.is_some(),
+ reply_audio_type = %reply_audio_artifact.and_then(|value| value.artifact_type.as_deref()).unwrap_or("none"),
+ reply_audio_path_alias = %reply_audio_artifact.and_then(|value| value.path_ref.as_deref()).map(redact).unwrap_or_else(|| "none".to_string()),
+ reply_audio_format = %reply_audio_artifact.and_then(|value| value.format.as_deref()).unwrap_or("none"),
+ device_output_count = device_outputs.len(),
+ "runtime helper turn_bridge_completed"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "turn_bridge_completed",
+ "ok",
+ None,
+ None,
+ json!({
+ "bridgeWallDurationMs": bridge_started_at.elapsed().as_millis() as u64,
+ "replyAudioPresent": reply_audio_artifact.is_some(),
+ "deviceOutputCount": device_outputs.len(),
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ }),
+ );
+ Ok(RuntimeTurnBridgeOutcome {
+ reply_audio_artifact: reply_audio_artifact.cloned(),
+ device_outputs,
+ })
+}
+
+async fn write_reply_audio_artifact(
+ http: &Client,
+ bridge_config: &TurnBridgeConfig,
+ sink: &BotAudioOutputSink,
+ call_id: &str,
+ trace_id: &str,
+ turn: &FinishedSpeechTurn,
+ artifact: &RuntimeTurnReplyAudioArtifact,
+ turn_pipeline_started_at: Instant,
+) -> Result<()> {
+ let artifact_type = artifact.artifact_type.as_deref().unwrap_or_default().trim();
+ if artifact_type != "local_file" {
+ return Err(anyhow!("unsupported reply audio artifact type"));
+ }
+ let path_ref = artifact
+ .path_ref
+ .as_deref()
+ .map(str::trim)
+ .filter(|value| !value.is_empty())
+ .ok_or_else(|| anyhow!("reply audio pathRef missing"))?;
+ let artifact_root = PathBuf::from(bridge_config.artifact_dir.as_deref().unwrap_or_default());
+ let audio_path = resolve_artifact_path(&artifact_root, path_ref)?;
+ let audio_path_string = audio_path.to_string_lossy().to_string();
+ let reply_audio_format = artifact.format.as_deref().unwrap_or("unknown");
+
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ reply_audio_type = %artifact_type,
+ reply_audio_path_alias = %redact(path_ref),
+ reply_audio_format = %reply_audio_format,
+ "runtime helper bot_reply_audio_write_started"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "bot_reply_audio_write_started",
+ "ok",
+ None,
+ None,
+ json!({
+ "replyAudioFormat": reply_audio_format,
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ }),
+ );
+
+ let reply_debug_label = format!("reply-{}", turn.turn_id);
+ let loaded_audio = load_pre_recorded_frames(
+ http,
+ Some(&audio_path_string),
+ None,
+ TARGET_SAMPLE_RATE_HZ,
+ TARGET_NUM_CHANNELS,
+ bridge_config.audio_debug_dump_dir.as_deref(),
+ call_id,
+ &reply_debug_label,
+ )
+ .await?
+ .ok_or_else(|| anyhow!("reply audio artifact decode returned empty"))?;
+ let AudioDiagnostics {
+ source_format,
+ source_bytes,
+ source_sample_rate_hz,
+ source_num_channels,
+ target_duration_ms,
+ frame_count,
+ rms,
+ peak,
+ clipped_sample_count,
+ silence_ratio,
+ mp3_skipped_data_count,
+ mp3_insufficient_data_count,
+ debug_source_path,
+ debug_pcm_wav_path,
+ ..
+ } = loaded_audio.diagnostics;
+ let frames = loaded_audio.frames;
+
+ let playback_started_at = Instant::now();
+ let pacing_started_at = tokio::time::Instant::now();
+ for (index, frame) in frames.iter().enumerate() {
+ sink.write_pcm_frame(frame).await?;
+ sleep_until(pacing_started_at + Duration::from_millis(((index + 1) as u64) * 20)).await;
+ }
+ let playback_wall_ms = playback_started_at.elapsed().as_millis() as i64;
+ let drift_ms = playback_wall_ms - target_duration_ms as i64;
+ sink.clear_buffer();
+
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ source_format,
+ source_bytes,
+ source_sample_rate_hz,
+ source_num_channels,
+ frame_count,
+ theoretical_duration_ms = target_duration_ms,
+ push_wall_duration_ms = playback_wall_ms,
+ push_drift_ms = drift_ms,
+ rms = round4(rms),
+ peak = round4(peak),
+ clipped_sample_count,
+ silence_ratio = round4(silence_ratio),
+ mp3_skipped_data_count,
+ mp3_insufficient_data_count,
+ debug_source_path = debug_source_path.as_deref().unwrap_or(""),
+ debug_pcm_wav_path = debug_pcm_wav_path.as_deref().unwrap_or(""),
+ "runtime helper bot_reply_audio_write_finished"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn.turn_id),
+ "bot_reply_audio_write_finished",
+ "ok",
+ None,
+ None,
+ json!({
+ "sourceFormat": source_format,
+ "sourceBytes": source_bytes,
+ "sourceSampleRateHz": source_sample_rate_hz,
+ "sourceNumChannels": source_num_channels,
+ "frameCount": frame_count,
+ "theoreticalDurationMs": target_duration_ms,
+ "pushWallDurationMs": playback_wall_ms,
+ "pushDriftMs": drift_ms,
+ "rms": round4(rms),
+ "peak": round4(peak),
+ "clippedSampleCount": clipped_sample_count,
+ "silenceRatio": round4(silence_ratio),
+ "mp3SkippedDataCount": mp3_skipped_data_count,
+ "mp3InsufficientDataCount": mp3_insufficient_data_count,
+ "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
+ }),
);
Ok(())
}
-fn build_device_output_smoke_payload(config: &Config) -> DeviceOutputMessage {
- DeviceOutputMessage {
- message_type: DEVICE_OUTPUT_TOPIC,
- schema_version: "1.0",
- call_id: config.call_id.clone(),
- role_id: config.role_id.clone(),
- trace_id: config.trace_id.clone(),
- ack_mode: "http",
- sensor_instructions: vec![
- SensorInstruction {
- command_id: command_id(&config.call_id, 1),
- sensor_type: "Vibrator",
- operation_type: "VibratorStart",
- step: 1,
- duration_sec: None,
- extension: r#"{"levelList":[{"level":1,"percent":"0.6"}]}"#.to_string(),
- },
- SensorInstruction {
- command_id: command_id(&config.call_id, 2),
- sensor_type: "Vibrator",
- operation_type: "VibratorUp",
- step: 1,
- duration_sec: None,
- extension: r#"{"delta":1}"#.to_string(),
- },
- SensorInstruction {
- command_id: command_id(&config.call_id, 3),
- sensor_type: "Pump",
- operation_type: "JiaStart",
- step: 1,
- duration_sec: None,
- extension: r#"{"levelList":[{"level":1,"percent":"0.5"}]}"#.to_string(),
- },
- SensorInstruction {
- command_id: command_id(&config.call_id, 4),
- sensor_type: "Heating",
- operation_type: "HeatingStart",
- step: 1,
- duration_sec: Some(3),
- extension: r#"{"target":"warm"}"#.to_string(),
- },
- ],
+fn resolve_artifact_path(artifact_root: &Path, path_ref: &str) -> Result<PathBuf> {
+ let path_ref = path_ref.trim();
+ if path_ref.is_empty() {
+ return Err(anyhow!("empty artifact pathRef"));
+ }
+ let relative = Path::new(path_ref);
+ if relative.is_absolute() {
+ return Err(anyhow!("absolute artifact pathRef is not allowed"));
+ }
+ let mut safe_relative = PathBuf::new();
+ for component in relative.components() {
+ match component {
+ std::path::Component::Normal(segment) => {
+ let segment = segment
+ .to_str()
+ .ok_or_else(|| anyhow!("non-utf8 artifact pathRef segment"))?;
+ require_safe_segment(segment)?;
+ safe_relative.push(segment);
+ }
+ std::path::Component::CurDir => {}
+ _ => return Err(anyhow!("unsafe artifact pathRef component")),
+ }
+ }
+ if safe_relative.as_os_str().is_empty() {
+ return Err(anyhow!("artifact pathRef has no safe components"));
+ }
+ let root = normalize_path_lexically(artifact_root);
+ let output_path = normalize_path_lexically(&root.join(safe_relative));
+ if !output_path.starts_with(&root) {
+ return Err(anyhow!("reply artifact path escapes root"));
+ }
+ Ok(output_path)
+}
+
+#[derive(Serialize)]
+struct RuntimeTurnRequest {
+ #[serde(rename = "callId")]
+ call_id: String,
+ #[serde(rename = "traceId")]
+ trace_id: String,
+ #[serde(rename = "turnId")]
+ turn_id: String,
+ #[serde(rename = "audioArtifact")]
+ audio_artifact: RuntimeTurnAudioArtifact,
+}
+
+#[derive(Serialize)]
+struct RuntimeTurnAudioArtifact {
+ #[serde(rename = "type")]
+ artifact_type: String,
+ #[serde(rename = "pathRef")]
+ path_ref: String,
+ format: String,
+ #[serde(rename = "sampleRate")]
+ sample_rate: u32,
+ channels: u32,
+ #[serde(rename = "durationMs")]
+ duration_ms: u64,
+ #[serde(rename = "byteSize")]
+ byte_size: u64,
+}
+
+#[derive(Deserialize)]
+struct RuntimeTurnCommonResult<T> {
+ code: i64,
+ msg: Option<String>,
+ data: Option<T>,
+ stage: Option<String>,
+ retryable: Option<bool>,
+}
+
+#[derive(Default)]
+struct RuntimeTurnBridgeOutcome {
+ reply_audio_artifact: Option<RuntimeTurnReplyAudioArtifact>,
+ device_outputs: Vec<RuntimeTurnDeviceOutput>,
+}
+
+struct RuntimeTurnStreamOutcome {
+ reply_playback_mode: String,
+ audio_chunk_count: u64,
+ device_output_count: u64,
+}
+
+struct RuntimeTurnStreamState {
+ reply_playback_mode: String,
+ reply_state_seq: u64,
+ playback_started_sent: bool,
+ first_audio_frame_written: bool,
+ completed: bool,
+ audio_chunk_count: u64,
+ device_output_count: u64,
+ encoded_audio_buffer: Vec<u8>,
+}
+
+impl Default for RuntimeTurnStreamState {
+ fn default() -> Self {
+ Self {
+ reply_playback_mode: "full_tts_fallback".to_string(),
+ reply_state_seq: 0,
+ playback_started_sent: false,
+ first_audio_frame_written: false,
+ completed: false,
+ audio_chunk_count: 0,
+ device_output_count: 0,
+ encoded_audio_buffer: Vec::new(),
+ }
}
}
-fn command_id(call_id: &str, sequence: u8) -> String {
- format!("{call_id}-device-smoke-{sequence:03}")
+#[derive(Deserialize)]
+struct RuntimeTurnResponseData {
+ #[serde(rename = "turnId")]
+ turn_id: Option<String>,
+ #[serde(rename = "messageId")]
+ message_id: Option<String>,
+ #[serde(rename = "replyAudioArtifact")]
+ reply_audio_artifact: Option<RuntimeTurnReplyAudioArtifact>,
+ #[serde(rename = "deviceOutputs")]
+ device_outputs: Option<Vec<RuntimeTurnDeviceOutput>>,
+}
+
+#[derive(Deserialize)]
+struct RuntimeTurnStreamEvent {
+ #[serde(rename = "type", alias = "event")]
+ event_type: Option<String>,
+ #[serde(rename = "seq")]
+ seq: Option<u64>,
+ #[serde(rename = "replyPlaybackMode")]
+ reply_playback_mode: Option<String>,
+ state: Option<String>,
+ #[serde(rename = "audioChunk")]
+ audio_chunk: Option<RuntimeTurnStreamAudioChunk>,
+ activity: Option<RuntimeTurnStreamActivity>,
+ #[serde(rename = "deviceOutput")]
+ device_output: Option<RuntimeTurnDeviceOutput>,
+ error: Option<RuntimeTurnStreamError>,
+ completion: Option<RuntimeTurnStreamCompletion>,
+ diagnostics: Option<serde_json::Value>,
+}
+
+impl RuntimeTurnStreamEvent {
+ fn event_type(&self) -> Option<String> {
+ self.event_type.as_deref().map(|value| match value {
+ "reply_playback_mode_selected" => "reply_playback_mode_selected".to_string(),
+ "reply_state" => "reply_state".to_string(),
+ "reply_audio_chunk" => "reply_audio_chunk".to_string(),
+ "device_output" => "device_output".to_string(),
+ "turn_completed" => "turn_completed".to_string(),
+ "turn_failed" => "turn_failed".to_string(),
+ "turn_cancelled" => "turn_cancelled".to_string(),
+ "activity" => "activity".to_string(),
+ other => other.to_string(),
+ })
+ }
+}
+
+#[derive(Deserialize)]
+struct RuntimeTurnStreamAudioChunk {
+ #[serde(rename = "chunkSeq", alias = "seq")]
+ chunk_seq: Option<u64>,
+ format: Option<String>,
+ #[serde(rename = "sampleRate")]
+ sample_rate: Option<u32>,
+ channels: Option<u32>,
+ #[serde(rename = "payloadBase64", alias = "audioBase64")]
+ payload_base64: Option<String>,
+ last: Option<bool>,
+}
+
+#[derive(Deserialize)]
+struct RuntimeTurnStreamActivity {
+ #[serde(rename = "eventType", alias = "event")]
+ event_type: Option<String>,
+ stage: Option<String>,
+ #[serde(rename = "reasonCode")]
+ reason_code: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct RuntimeTurnStreamError {
+ #[serde(rename = "reasonCode")]
+ reason_code: Option<String>,
+ stage: Option<String>,
+ retryable: Option<bool>,
+}
+
+#[derive(Deserialize)]
+struct RuntimeTurnStreamCompletion {
+ #[serde(rename = "messageId")]
+ message_id: Option<String>,
+}
+
+#[derive(Clone, Deserialize)]
+struct RuntimeTurnReplyAudioArtifact {
+ #[serde(rename = "type")]
+ artifact_type: Option<String>,
+ #[serde(rename = "pathRef")]
+ path_ref: Option<String>,
+ format: Option<String>,
+}
+
+#[derive(Clone, Deserialize)]
+struct RuntimeTurnDeviceOutput {
+ #[serde(rename = "commandId")]
+ command_id: Option<String>,
+ #[serde(rename = "commandCode")]
+ command_code: Option<String>,
+ params: Option<serde_json::Value>,
+}
+
+fn require_safe_segment(value: &str) -> Result<()> {
+ if value.is_empty()
+ || value.contains('/')
+ || value.contains("..")
+ || !value
+ .chars()
+ .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.'))
+ {
+ return Err(anyhow!("unsafe path segment"));
+ }
+ Ok(())
+}
+
+fn normalize_path_lexically(path: &Path) -> PathBuf {
+ let mut normalized = PathBuf::new();
+ for component in path.components() {
+ match component {
+ std::path::Component::CurDir => {}
+ std::path::Component::ParentDir => {
+ normalized.pop();
+ }
+ _ => normalized.push(component.as_os_str()),
+ }
+ }
+ normalized
+}
+
+fn spawn_user_audio_frame_observer(
+ track: RemoteAudioTrack,
+ call_id: String,
+ trace_id: String,
+ participant_alias: String,
+ track_sid_alias: String,
+ simple_vad_enabled: bool,
+ simple_vad_config: SimpleVadConfig,
+ vad_enabled_gate: Arc<AtomicBool>,
+ turn_bridge_config: TurnBridgeConfig,
+ http: Client,
+ sink: Arc<BotAudioOutputSink>,
+) -> JoinHandle<()> {
+ tokio::spawn(async move {
+ let mut stream = NativeAudioStream::new(
+ track.rtc_track(),
+ TARGET_SAMPLE_RATE_HZ as i32,
+ i32::from(TARGET_NUM_CHANNELS),
+ );
+ let started_at = Instant::now();
+ let mut frame_count: u64 = 0;
+ let mut sample_count: u64 = 0;
+ let mut simple_vad = if simple_vad_enabled {
+ Some(SimpleVad::new(simple_vad_config))
+ } else {
+ None
+ };
+
+ while let Some(frame) = stream.next().await {
+ frame_count += 1;
+ sample_count += u64::from(frame.samples_per_channel) * u64::from(frame.num_channels);
+ let elapsed_ms = started_at.elapsed().as_millis() as u64;
+ if frame_count == 1 {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ sample_rate_hz = frame.sample_rate,
+ num_channels = frame.num_channels,
+ samples_per_channel = frame.samples_per_channel,
+ first_frame_elapsed_ms = elapsed_ms,
+ "runtime helper user_audio_frame_received"
+ );
+ } else if frame_count % 250 == 0 {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ frame_count,
+ sample_count,
+ observed_wall_ms = elapsed_ms,
+ "runtime helper user_audio_frame_summary"
+ );
+ }
+
+ if let Some(vad) = simple_vad.as_mut() {
+ if vad_enabled_gate.load(Ordering::Acquire) {
+ if let Some(turn) = vad.observe_frame(
+ &call_id,
+ &trace_id,
+ &participant_alias,
+ &track_sid_alias,
+ frame_count,
+ elapsed_ms,
+ &frame,
+ ) {
+ handle_finished_turn(
+ &http,
+ &turn_bridge_config,
+ &sink,
+ &call_id,
+ &trace_id,
+ turn,
+ )
+ .await;
+ }
+ } else {
+ vad.observe_disabled_frame(
+ &call_id,
+ &trace_id,
+ &participant_alias,
+ &track_sid_alias,
+ frame_count,
+ elapsed_ms,
+ );
+ }
+ }
+ }
+
+ if let Some(vad) = simple_vad.as_mut() {
+ if let Some(turn) = vad.finish_stream(
+ &call_id,
+ &trace_id,
+ &participant_alias,
+ &track_sid_alias,
+ started_at.elapsed().as_millis() as u64,
+ ) {
+ handle_finished_turn(&http, &turn_bridge_config, &sink, &call_id, &trace_id, turn)
+ .await;
+ }
+ }
+
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ frame_count,
+ sample_count,
+ observed_wall_ms = started_at.elapsed().as_millis() as u64,
+ "runtime helper user_audio_stream_ended"
+ );
+ })
+}
+
+#[derive(Clone)]
+struct SimpleVadConfig {
+ rms_threshold: f64,
+ peak_threshold: f64,
+ start_frames: u32,
+ end_silence_ms: u64,
+ min_speech_ms: u64,
+ max_turn_ms: u64,
+ initial_ignore_ms: u64,
+}
+
+impl SimpleVadConfig {
+ fn from_env() -> Self {
+ Self {
+ rms_threshold: f64_env("CV_VAD_RMS_THRESHOLD", 0.012),
+ peak_threshold: f64_env("CV_VAD_PEAK_THRESHOLD", 0.08),
+ start_frames: u32_env("CV_VAD_START_FRAMES", 5).max(1),
+ end_silence_ms: u64_env("CV_VAD_END_SILENCE_MS", 700).max(100),
+ min_speech_ms: u64_env("CV_VAD_MIN_SPEECH_MS", 300).max(1),
+ max_turn_ms: u64_env("CV_VAD_MAX_TURN_MS", 10_000).max(1_000),
+ initial_ignore_ms: u64_env("CV_VAD_INITIAL_IGNORE_MS", 500),
+ }
+ }
+
+ fn end_silence_frames(&self, frame_duration_ms: u64) -> u32 {
+ let frame_duration_ms = frame_duration_ms.max(1);
+ self.end_silence_ms.div_ceil(frame_duration_ms) as u32
+ }
+}
+
+struct SimpleVad {
+ config: SimpleVadConfig,
+ turn_index: u64,
+ in_speech: bool,
+ voiced_run_frames: u32,
+ silence_run_frames: u32,
+ speech_start_elapsed_ms: u64,
+ speech_frame_count: u64,
+ speech_sample_count: u64,
+ speech_rms_sum: f64,
+ speech_peak: f64,
+ ignored_before_enabled_frames: u64,
+ pre_speech_frames: Vec<Vec<i16>>,
+ speech_samples: Vec<i16>,
+}
+
+struct FinishedSpeechTurn {
+ turn_index: u64,
+ turn_id: String,
+ samples: Vec<i16>,
+ duration_ms: u64,
+ frame_count: u64,
+ sample_count: u64,
+ end_reason: String,
+}
+
+impl SimpleVad {
+ fn new(config: SimpleVadConfig) -> Self {
+ Self {
+ config,
+ turn_index: 0,
+ in_speech: false,
+ voiced_run_frames: 0,
+ silence_run_frames: 0,
+ speech_start_elapsed_ms: 0,
+ speech_frame_count: 0,
+ speech_sample_count: 0,
+ speech_rms_sum: 0.0,
+ speech_peak: 0.0,
+ ignored_before_enabled_frames: 0,
+ pre_speech_frames: Vec::new(),
+ speech_samples: Vec::new(),
+ }
+ }
+
+ fn observe_disabled_frame(
+ &mut self,
+ call_id: &str,
+ trace_id: &str,
+ participant_alias: &str,
+ track_sid_alias: &str,
+ frame_count: u64,
+ elapsed_ms: u64,
+ ) {
+ self.reset_current_turn();
+ self.ignored_before_enabled_frames = self.ignored_before_enabled_frames.saturating_add(1);
+ if self.ignored_before_enabled_frames == 1 || self.ignored_before_enabled_frames % 250 == 0
+ {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ frame_count,
+ ignored_frame_count = self.ignored_before_enabled_frames,
+ elapsed_ms,
+ "runtime helper vad_ignored_before_enabled"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ None,
+ "ignored_user_audio_before_vad_enabled",
+ "ok",
+ None,
+ None,
+ json!({
+ "frameCount": frame_count,
+ "ignoredFrameCount": self.ignored_before_enabled_frames,
+ "elapsedMs": elapsed_ms,
+ }),
+ );
+ }
+ }
+
+ fn observe_frame(
+ &mut self,
+ call_id: &str,
+ trace_id: &str,
+ participant_alias: &str,
+ track_sid_alias: &str,
+ frame_count: u64,
+ elapsed_ms: u64,
+ frame: &AudioFrame<'_>,
+ ) -> Option<FinishedSpeechTurn> {
+ let frame_duration_ms = frame_duration_ms(frame);
+ let (rms, peak) = pcm_energy_stats(frame.data.as_ref());
+ if elapsed_ms < self.config.initial_ignore_ms {
+ return None;
+ }
+
+ let voiced = rms >= self.config.rms_threshold || peak >= self.config.peak_threshold;
+ if !self.in_speech {
+ self.remember_pre_speech_frame(frame);
+ }
+ if voiced {
+ self.voiced_run_frames = self.voiced_run_frames.saturating_add(1);
+ self.silence_run_frames = 0;
+ } else {
+ self.voiced_run_frames = 0;
+ self.silence_run_frames = self.silence_run_frames.saturating_add(1);
+ }
+
+ if !self.in_speech {
+ if self.voiced_run_frames >= self.config.start_frames {
+ self.turn_index += 1;
+ self.in_speech = true;
+ self.speech_start_elapsed_ms = elapsed_ms
+ .saturating_sub(u64::from(self.config.start_frames) * frame_duration_ms);
+ self.speech_frame_count = u64::from(self.config.start_frames);
+ self.speech_sample_count = u64::from(frame.samples_per_channel)
+ * u64::from(frame.num_channels)
+ * u64::from(self.config.start_frames);
+ self.speech_rms_sum = rms * f64::from(self.config.start_frames);
+ self.speech_peak = peak;
+ self.speech_samples = self
+ .pre_speech_frames
+ .iter()
+ .flat_map(|samples| samples.iter().copied())
+ .collect();
+ let turn_id = format!("turn-{:04}", self.turn_index);
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ turn_index = self.turn_index,
+ frame_count,
+ speech_start_elapsed_ms = self.speech_start_elapsed_ms,
+ rms = round4(rms),
+ peak = round4(peak),
+ "runtime helper vad_speech_start"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn_id),
+ "vad_speech_start",
+ "ok",
+ None,
+ None,
+ json!({
+ "turnIndex": self.turn_index,
+ "frameCount": frame_count,
+ "speechStartElapsedMs": self.speech_start_elapsed_ms,
+ "rms": round4(rms),
+ "peak": round4(peak),
+ }),
+ );
+ }
+ return None;
+ }
+
+ self.speech_samples.extend_from_slice(frame.data.as_ref());
+ self.speech_frame_count += 1;
+ self.speech_sample_count +=
+ u64::from(frame.samples_per_channel) * u64::from(frame.num_channels);
+ self.speech_rms_sum += rms;
+ self.speech_peak = self.speech_peak.max(peak);
+
+ let speech_duration_ms = elapsed_ms.saturating_sub(self.speech_start_elapsed_ms);
+ if speech_duration_ms >= self.config.max_turn_ms {
+ return self.finish_turn(
+ call_id,
+ trace_id,
+ participant_alias,
+ track_sid_alias,
+ elapsed_ms,
+ "max_turn_ms",
+ );
+ }
+
+ if !voiced && self.silence_run_frames >= self.config.end_silence_frames(frame_duration_ms) {
+ return self.finish_turn(
+ call_id,
+ trace_id,
+ participant_alias,
+ track_sid_alias,
+ elapsed_ms,
+ "silence",
+ );
+ }
+ None
+ }
+
+ fn finish_stream(
+ &mut self,
+ call_id: &str,
+ trace_id: &str,
+ participant_alias: &str,
+ track_sid_alias: &str,
+ elapsed_ms: u64,
+ ) -> Option<FinishedSpeechTurn> {
+ if self.in_speech {
+ return self.finish_turn(
+ call_id,
+ trace_id,
+ participant_alias,
+ track_sid_alias,
+ elapsed_ms,
+ "stream_end",
+ );
+ } else if self.turn_index == 0 {
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ observed_wall_ms = elapsed_ms,
+ "runtime helper vad_no_speech_summary"
+ );
+ }
+ None
+ }
+
+ fn finish_turn(
+ &mut self,
+ call_id: &str,
+ trace_id: &str,
+ participant_alias: &str,
+ track_sid_alias: &str,
+ elapsed_ms: u64,
+ end_reason: &str,
+ ) -> Option<FinishedSpeechTurn> {
+ let speech_duration_ms = elapsed_ms.saturating_sub(self.speech_start_elapsed_ms);
+ let event_name = if speech_duration_ms < self.config.min_speech_ms {
+ "runtime helper vad_speech_too_short"
+ } else {
+ "runtime helper vad_speech_end"
+ };
+ let avg_rms = if self.speech_frame_count == 0 {
+ 0.0
+ } else {
+ self.speech_rms_sum / self.speech_frame_count as f64
+ };
+
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ participant_alias = %participant_alias,
+ track_sid_alias = %track_sid_alias,
+ turn_index = self.turn_index,
+ speech_start_elapsed_ms = self.speech_start_elapsed_ms,
+ speech_end_elapsed_ms = elapsed_ms,
+ speech_duration_ms,
+ speech_frame_count = self.speech_frame_count,
+ speech_sample_count = self.speech_sample_count,
+ avg_rms = round4(avg_rms),
+ peak = round4(self.speech_peak),
+ end_reason = %end_reason,
+ "{}", event_name
+ );
+
+ let finished_turn = if speech_duration_ms >= self.config.min_speech_ms {
+ let turn_id = format!("turn-{:04}", self.turn_index);
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(&turn_id),
+ "vad_speech_end",
+ "ok",
+ None,
+ None,
+ json!({
+ "turnIndex": self.turn_index,
+ "speechStartElapsedMs": self.speech_start_elapsed_ms,
+ "speechEndElapsedMs": elapsed_ms,
+ "speechDurationMs": speech_duration_ms,
+ "speechFrameCount": self.speech_frame_count,
+ "speechSampleCount": self.speech_sample_count,
+ "avgRms": round4(avg_rms),
+ "peak": round4(self.speech_peak),
+ "endReason": end_reason,
+ }),
+ );
+ Some(FinishedSpeechTurn {
+ turn_index: self.turn_index,
+ turn_id,
+ samples: std::mem::take(&mut self.speech_samples),
+ duration_ms: speech_duration_ms,
+ frame_count: self.speech_frame_count,
+ sample_count: self.speech_sample_count,
+ end_reason: end_reason.to_string(),
+ })
+ } else {
+ None
+ };
+ self.reset_current_turn();
+ finished_turn
+ }
+
+ fn reset_current_turn(&mut self) {
+ self.in_speech = false;
+ self.voiced_run_frames = 0;
+ self.silence_run_frames = 0;
+ self.speech_start_elapsed_ms = 0;
+ self.speech_frame_count = 0;
+ self.speech_sample_count = 0;
+ self.speech_rms_sum = 0.0;
+ self.speech_peak = 0.0;
+ self.speech_samples.clear();
+ self.pre_speech_frames.clear();
+ }
+
+ fn remember_pre_speech_frame(&mut self, frame: &AudioFrame<'_>) {
+ self.pre_speech_frames.push(frame.data.as_ref().to_vec());
+ let max_frames = self.config.start_frames as usize;
+ if self.pre_speech_frames.len() > max_frames {
+ let remove_count = self.pre_speech_frames.len() - max_frames;
+ self.pre_speech_frames.drain(0..remove_count);
+ }
+ }
+}
+
+fn frame_duration_ms(frame: &AudioFrame<'_>) -> u64 {
+ if frame.sample_rate == 0 {
+ return 10;
+ }
+ ((u64::from(frame.samples_per_channel) * 1000) / u64::from(frame.sample_rate)).max(1)
+}
+
+fn pcm_energy_stats(samples: &[i16]) -> (f64, f64) {
+ if samples.is_empty() {
+ return (0.0, 0.0);
+ }
+ let mut square_sum = 0.0;
+ let mut peak = 0.0;
+ for sample in samples {
+ let normalized = f64::from(*sample) / f64::from(i16::MAX);
+ square_sum += normalized * normalized;
+ let abs = normalized.abs();
+ if abs > peak {
+ peak = abs;
+ }
+ }
+ ((square_sum / samples.len() as f64).sqrt(), peak)
+}
+
+fn round4(value: f64) -> f64 {
+ (value * 10_000.0).round() / 10_000.0
}
struct BotAudioOutputSink {
room: Arc<Room>,
rtc_source: NativeAudioSource,
track: LocalAudioTrack,
+ device_output_destination_identity: Option<String>,
}
impl BotAudioOutputSink {
@@ -339,9 +2517,12 @@
room: Arc<Room>,
room_name: &str,
participant_identity: &str,
+ call_id: &str,
+ trace_id: &str,
track_name: &str,
sample_rate: u32,
num_channels: u32,
+ device_output_destination_identity: Option<String>,
) -> Result<Self> {
let rtc_source = NativeAudioSource::new(
AudioSourceOptions::default(),
@@ -353,6 +2534,8 @@
track_name,
RtcAudioSource::Native(rtc_source.clone()),
);
+ let room_alias = redact(room_name);
+ let participant_alias = redact(participant_identity);
room.local_participant()
.publish_track(
@@ -362,26 +2545,205 @@
.await
.map_err(|error| {
anyhow!(
- "failed to publish bot audio track in room {room_name} for participant {participant_identity}: {error}"
+ "failed to publish bot audio track in room {room_alias} for participant {participant_alias}: {error}"
)
})?;
info!(
- room_id = %room_name,
- participant_alias = %redact(participant_identity),
+ room_alias = %room_alias,
+ participant_alias = %participant_alias,
track_name = %track_name,
sample_rate,
num_channels,
"runtime helper published bot audio track"
+ );
+ emit_activity(
+ call_id,
+ trace_id,
+ None,
+ "bot_track_ready",
+ "ok",
+ None,
+ None,
+ json!({
+ "trackName": track_name,
+ "sampleRate": sample_rate,
+ "numChannels": num_channels,
+ }),
);
Ok(Self {
room,
rtc_source,
track,
+ device_output_destination_identity,
})
}
+ async fn publish_device_output(
+ &self,
+ call_id: &str,
+ trace_id: &str,
+ turn_id: &str,
+ output: &RuntimeTurnDeviceOutput,
+ ) -> Result<()> {
+ let command_id = output
+ .command_id
+ .as_deref()
+ .map(str::trim)
+ .filter(|value| !value.is_empty())
+ .ok_or_else(|| anyhow!("device output commandId missing"))?;
+ let command_code = output
+ .command_code
+ .as_deref()
+ .map(str::trim)
+ .filter(|value| !value.is_empty())
+ .ok_or_else(|| anyhow!("device output commandCode missing"))?;
+ let payload = serde_json::json!({
+ "type": "device_output",
+ "schemaVersion": "1.0",
+ "callId": call_id,
+ "traceId": trace_id,
+ "turnId": turn_id,
+ "commandId": command_id,
+ "commandCode": command_code,
+ "params": output.params.clone().unwrap_or_else(|| serde_json::json!({})),
+ "source": {
+ "kind": "voice_command"
+ },
+ });
+ let destinations = self
+ .device_output_destination_identity
+ .as_ref()
+ .filter(|value| !value.trim().is_empty())
+ .map(|value| vec![ParticipantIdentity(value.trim().to_string())])
+ .unwrap_or_default();
+ let destination_count = destinations.len();
+ self.room
+ .local_participant()
+ .publish_data(DataPacket {
+ payload: serde_json::to_vec(&payload)
+ .context("failed to encode device output data message")?,
+ topic: Some("device_output".to_string()),
+ reliable: true,
+ destination_identities: destinations,
+ })
+ .await
+ .map_err(|error| anyhow!("failed to publish device output data message: {error}"))?;
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn_id,
+ command_id_alias = %redact(command_id),
+ command_code = %command_code,
+ destination_count,
+ "runtime helper device_output_sent"
+ );
+ Ok(())
+ }
+
+ async fn publish_reply_state(
+ &self,
+ call_id: &str,
+ trace_id: &str,
+ turn_id: &str,
+ reply_playback_mode: &str,
+ state: &str,
+ seq: u64,
+ ) -> Result<()> {
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(turn_id),
+ "reply_state_send_started",
+ "ok",
+ None,
+ None,
+ json!({
+ "state": state,
+ "seq": seq,
+ "replyPlaybackMode": reply_playback_mode,
+ }),
+ );
+ let payload = serde_json::json!({
+ "type": "reply_state",
+ "schemaVersion": "1.0",
+ "callId": call_id,
+ "traceId": trace_id,
+ "turnId": turn_id,
+ "replyPlaybackMode": reply_playback_mode,
+ "state": state,
+ "seq": seq,
+ "tsMs": current_time_millis(),
+ });
+ let destinations = self
+ .device_output_destination_identity
+ .as_ref()
+ .filter(|value| !value.trim().is_empty())
+ .map(|value| vec![ParticipantIdentity(value.trim().to_string())])
+ .unwrap_or_default();
+ let destination_count = destinations.len();
+ let publish_result = self
+ .room
+ .local_participant()
+ .publish_data(DataPacket {
+ payload: serde_json::to_vec(&payload)
+ .context("failed to encode reply state data message")?,
+ topic: Some("combrabo_voice.reply_state".to_string()),
+ reliable: true,
+ destination_identities: destinations,
+ })
+ .await
+ .map_err(|error| anyhow!("failed to publish reply state data message: {error}"));
+ match publish_result {
+ Ok(_) => {
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(turn_id),
+ "reply_state_send_finished",
+ "ok",
+ None,
+ None,
+ json!({
+ "state": state,
+ "seq": seq,
+ "replyPlaybackMode": reply_playback_mode,
+ "destinationCount": destination_count,
+ }),
+ );
+ info!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn_id,
+ state = %state,
+ seq,
+ reply_playback_mode = %reply_playback_mode,
+ destination_count,
+ "runtime helper reply_state_sent"
+ );
+ Ok(())
+ }
+ Err(error) => {
+ emit_activity(
+ call_id,
+ trace_id,
+ Some(turn_id),
+ "reply_state_send_failed",
+ "failed",
+ Some("REPLY_STATE_SEND_FAILED"),
+ Some(true),
+ json!({
+ "state": state,
+ "seq": seq,
+ "replyPlaybackMode": reply_playback_mode,
+ }),
+ );
+ Err(error)
+ }
+ }
+ }
+
async fn write_pcm_frame(&self, frame: &audio::PcmFrame) -> Result<()> {
let audio_frame = AudioFrame {
data: frame.data.as_slice().into(),
diff --git a/src/service.rs b/src/service.rs
new file mode 100644
index 0000000..a579636
--- /dev/null
+++ b/src/service.rs
@@ -0,0 +1,908 @@
+use std::{
+ collections::HashMap,
+ env,
+ io::{BufRead, BufReader, Read, Write},
+ net::{TcpListener, TcpStream},
+ process::{Child, Command, Stdio},
+ sync::{Arc, Mutex},
+ thread,
+ time::{Duration, SystemTime, UNIX_EPOCH},
+};
+
+use anyhow::{Context, Result, anyhow};
+use serde::Deserialize;
+use serde_json::{Value, json};
+use tracing::{info, warn};
+
+const DEFAULT_BIND_ADDR: &str = "127.0.0.1:18080";
+const HTTP_READ_LIMIT_BYTES: usize = 1024 * 1024;
+const DEFAULT_START_READY_TIMEOUT_MS: u64 = 10_000;
+
+pub fn service_mode_enabled() -> bool {
+ env::args().any(|arg| arg == "service" || arg == "--service")
+ || bool_env("CV_HELPER_SERVICE_ENABLED", false)
+}
+
+pub async fn run_service() -> Result<()> {
+ let config = ServiceConfig::from_env()?;
+ let listener = TcpListener::bind(&config.bind_addr)
+ .with_context(|| format!("failed to bind helper service {}", config.bind_addr))?;
+ let state = Arc::new(ServiceState {
+ sessions: Mutex::new(HashMap::new()),
+ config: config.clone(),
+ });
+ info!(
+ bind_addr = %config.bind_addr,
+ auth_profiles = ?config.auth_profiles.keys().collect::<Vec<_>>(),
+ "combrabo voice helper service started"
+ );
+
+ for stream in listener.incoming() {
+ match stream {
+ Ok(stream) => {
+ let state = state.clone();
+ thread::spawn(move || {
+ if let Err(error) = handle_connection(stream, state) {
+ warn!(error = %error, "helper service connection failed");
+ }
+ });
+ }
+ Err(error) => warn!(error = %error, "helper service accept failed"),
+ }
+ }
+ Ok(())
+}
+
+fn handle_connection(mut stream: TcpStream, state: Arc<ServiceState>) -> Result<()> {
+ stream.set_read_timeout(Some(Duration::from_secs(5)))?;
+ let request = HttpRequest::read(&mut stream)?;
+ let response = route(request, state);
+ response.write(&mut stream)?;
+ Ok(())
+}
+
+fn route(request: HttpRequest, state: Arc<ServiceState>) -> HttpResponse {
+ if request.method == "GET"
+ && (request.path == "/health" || request.path == "/internal/combrabo-voice/health")
+ {
+ return ok(json!({
+ "status": "UP",
+ "service": "lm-livekit-helper",
+ "version": env!("CARGO_PKG_VERSION"),
+ "mode": "service"
+ }));
+ }
+ if !authorized(&request, &state.config) {
+ return error(
+ 401,
+ 2001003404,
+ "HELPER_AUTH_FAILED",
+ "auth",
+ false,
+ trace_id(&request),
+ );
+ }
+ if request.method == "POST" && request.path == "/internal/combrabo-voice/sessions/start" {
+ return start_session(&request, state);
+ }
+ if request.method == "GET"
+ && request
+ .path
+ .starts_with("/internal/combrabo-voice/sessions/")
+ {
+ let call_id = request
+ .path
+ .trim_start_matches("/internal/combrabo-voice/sessions/")
+ .trim_matches('/');
+ return get_session(call_id, state);
+ }
+ if request.method == "POST"
+ && request
+ .path
+ .starts_with("/internal/combrabo-voice/sessions/")
+ && request.path.ends_with("/stop")
+ {
+ let call_id = request
+ .path
+ .trim_start_matches("/internal/combrabo-voice/sessions/")
+ .trim_end_matches("/stop")
+ .trim_matches('/');
+ return stop_session(call_id, &request, state);
+ }
+ error(
+ 404,
+ 2001003404,
+ "HELPER_ROUTE_NOT_FOUND",
+ "helper_http",
+ false,
+ trace_id(&request),
+ )
+}
+
+fn start_session(request: &HttpRequest, state: Arc<ServiceState>) -> HttpResponse {
+ let start_req: SessionStartRequest = match serde_json::from_slice(&request.body) {
+ Ok(value) => value,
+ Err(_) => {
+ return error(
+ 400,
+ 2001003404,
+ "HELPER_REQUEST_INVALID",
+ "validate",
+ false,
+ trace_id(request),
+ );
+ }
+ };
+ let trace_id = start_req.trace_id.clone().or_else(|| trace_id(request));
+ if start_req.call_id.trim().is_empty() || start_req.runtime_session_nonce.trim().is_empty() {
+ return error(
+ 400,
+ 2001003404,
+ "HELPER_REQUEST_INVALID",
+ "validate",
+ false,
+ trace_id,
+ );
+ }
+ let auth_profile = start_req
+ .auth_profile
+ .as_deref()
+ .filter(|value| !value.trim().is_empty())
+ .unwrap_or("default")
+ .to_string();
+ let profile = match state.config.auth_profiles.get(&auth_profile) {
+ Some(value) => value.clone(),
+ None => {
+ return error(
+ 200,
+ 2001003404,
+ "HELPER_AUTH_PROFILE_NOT_FOUND",
+ "auth",
+ true,
+ trace_id,
+ );
+ }
+ };
+ let mut sessions = match state.sessions.lock() {
+ Ok(value) => value,
+ Err(_) => {
+ return error(
+ 200,
+ 2001003404,
+ "HELPER_SESSION_LOCK_FAILED",
+ "runtime_ready",
+ true,
+ trace_id,
+ );
+ }
+ };
+ if let Some(existing) = sessions.get_mut(&start_req.call_id) {
+ let _ = refresh_child_status(existing);
+ if existing.runtime_session_nonce == start_req.runtime_session_nonce
+ && existing.status != "STOPPED"
+ {
+ return ok(session_start_data(existing));
+ }
+ return error(
+ 200,
+ 2001003404,
+ "RUNTIME_SESSION_CONFLICT",
+ "runtime_ready",
+ true,
+ trace_id,
+ );
+ }
+ let child = match spawn_worker(&start_req, &profile, &state.config) {
+ Ok(value) => value,
+ Err(spawn_error) => {
+ warn!(
+ call_id = %start_req.call_id,
+ trace_id = ?trace_id,
+ error = %spawn_error,
+ "helper service failed to spawn worker"
+ );
+ return error(
+ 200,
+ 2001003404,
+ "RUNTIME_START_FAILED",
+ "runtime_ready",
+ true,
+ trace_id,
+ );
+ }
+ };
+ let runtime_session_id = format!("rt_{}", start_req.call_id);
+ let mut child = child;
+ let child_stdout = child.stdout.take();
+ let session = HelperSession {
+ call_id: start_req.call_id.clone(),
+ trace_id: trace_id.clone(),
+ runtime_session_nonce: start_req.runtime_session_nonce.clone(),
+ runtime_session_id,
+ status: "STARTING".to_string(),
+ bot_participant_joined: false,
+ bot_track_ready: false,
+ first_audio_source: start_req
+ .audio
+ .as_ref()
+ .and_then(|value| value.first_audio_source.clone())
+ .unwrap_or_else(|| "fixed_greeting_tts".to_string()),
+ active_turn_id: None,
+ last_event_type: Some("helper_worker_started".to_string()),
+ last_event_at: now_millis(),
+ child: Some(child),
+ };
+ sessions.insert(start_req.call_id.clone(), session);
+ drop(sessions);
+
+ if let Some(stdout) = child_stdout {
+ spawn_worker_stdout_monitor(start_req.call_id.clone(), state.clone(), stdout);
+ }
+
+ match wait_for_session_ready(&start_req.call_id, state, &trace_id) {
+ WaitReadyResult::Ready(response) => ok(response),
+ WaitReadyResult::Failed(response) => response,
+ }
+}
+
+fn get_session(call_id: &str, state: Arc<ServiceState>) -> HttpResponse {
+ let mut sessions = match state.sessions.lock() {
+ Ok(value) => value,
+ Err(_) => {
+ return error(
+ 200,
+ 2001003404,
+ "HELPER_SESSION_LOCK_FAILED",
+ "runtime_ready",
+ true,
+ None,
+ );
+ }
+ };
+ let Some(session) = sessions.get_mut(call_id) else {
+ return error(
+ 200,
+ 2001003404,
+ "CALL_NOT_FOUND",
+ "runtime_ready",
+ false,
+ None,
+ );
+ };
+ let _ = refresh_child_status(session);
+ ok(json!({
+ "callId": &session.call_id,
+ "status": &session.status,
+ "botParticipantJoined": session.bot_participant_joined,
+ "botTrackReady": session.bot_track_ready,
+ "activeTurnId": session.active_turn_id.as_ref(),
+ "lastEventType": session.last_event_type.as_ref(),
+ "lastEventAt": session.last_event_at.to_string(),
+ }))
+}
+
+fn stop_session(call_id: &str, request: &HttpRequest, state: Arc<ServiceState>) -> HttpResponse {
+ let stop_req: SessionStopRequest = serde_json::from_slice(&request.body).unwrap_or_default();
+ let mut sessions = match state.sessions.lock() {
+ Ok(value) => value,
+ Err(_) => {
+ return error(
+ 200,
+ 2001003404,
+ "HELPER_SESSION_LOCK_FAILED",
+ "runtime_stop_requested",
+ true,
+ trace_id(request),
+ );
+ }
+ };
+ let Some(mut session) = sessions.remove(call_id) else {
+ return ok(json!({
+ "callId": call_id,
+ "status": "STOPPED",
+ "alreadyStopped": true,
+ }));
+ };
+ if stop_req
+ .runtime_session_nonce
+ .as_ref()
+ .is_some_and(|value| *value != session.runtime_session_nonce)
+ {
+ sessions.insert(call_id.to_string(), session);
+ return error(
+ 200,
+ 2001003404,
+ "RUNTIME_SESSION_MISMATCH",
+ "runtime_stop_requested",
+ true,
+ trace_id(request),
+ );
+ }
+ if let Some(child) = session.child.as_mut() {
+ let _ = child.kill();
+ let _ = child.wait();
+ }
+ session.status = "STOPPED".to_string();
+ ok(json!({
+ "callId": session.call_id,
+ "status": "STOPPED",
+ "alreadyStopped": false,
+ }))
+}
+
+fn spawn_worker(
+ req: &SessionStartRequest,
+ profile: &AuthProfile,
+ config: &ServiceConfig,
+) -> Result<Child> {
+ let exe = env::current_exe().context("failed to resolve helper executable")?;
+ let mut command = Command::new(exe);
+ command
+ .env_remove("CV_HELPER_SERVICE_ENABLED")
+ .env("CV_CALL_ID", &req.call_id)
+ .env(
+ "CV_TRACE_ID",
+ req.trace_id
+ .clone()
+ .unwrap_or_else(|| format!("trace_{}", req.call_id)),
+ )
+ .env("CV_RUNTIME_CALL_ID", &req.call_id)
+ .env(
+ "CV_RUNTIME_TRACE_ID",
+ req.trace_id
+ .clone()
+ .unwrap_or_else(|| format!("trace_{}", req.call_id)),
+ )
+ .env("CV_RUNTIME_SESSION_NONCE", &req.runtime_session_nonce)
+ .env("CV_LIVEKIT_URL", &req.livekit.url)
+ .env("CV_LIVEKIT_ROOM_ID", &req.livekit.room_id)
+ .env("CV_LIVEKIT_BOT_TOKEN", &req.livekit.bot_token)
+ .env(
+ "CV_LIVEKIT_BOT_PARTICIPANT_IDENTITY",
+ &req.livekit.bot_participant_identity,
+ )
+ .env("CV_RUNTIME_TURN_BRIDGE_TOKEN", &profile.turn_bridge_token)
+ .stdout(Stdio::piped())
+ .stderr(Stdio::inherit());
+ if let Some(value) = &req.livekit.user_participant_identity {
+ command.env("CV_LIVEKIT_USER_PARTICIPANT_IDENTITY", value);
+ }
+ if let Some(turn_bridge) = &req.turn_bridge {
+ command.env("CV_RUNTIME_TURN_BRIDGE_URL", &turn_bridge.url);
+ }
+ if let Some(runtime) = &req.runtime {
+ if let Some(value) = &runtime.turn_artifact_dir {
+ command.env("CV_RUNTIME_TURN_ARTIFACT_DIR", value);
+ }
+ if let Some(true) = runtime.audio_debug_dump_enabled {
+ if let Some(value) = &config.audio_debug_dump_dir {
+ command.env("CV_AUDIO_DEBUG_DUMP_DIR", value);
+ }
+ }
+ }
+ if let Some(audio) = &req.audio {
+ if let Some(first_audio_source) = &audio.first_audio_source {
+ command.env("CV_FIRST_AUDIO_SOURCE", first_audio_source);
+ command.env("CV_GREETING_SOURCE", first_audio_source);
+ }
+ if let Some(greeting_audio) = &audio.greeting_audio {
+ match greeting_audio.r#type.as_deref() {
+ Some("remote_url") => {
+ if let Some(value) = &greeting_audio.path_ref {
+ command.env("CV_GREETING_AUDIO_URL", value);
+ }
+ }
+ _ => {
+ if let Some(value) = &greeting_audio.path_ref {
+ command.env("CV_GREETING_AUDIO_FILE", value);
+ }
+ }
+ }
+ }
+ }
+ command.spawn().map_err(|error| anyhow!(error))
+}
+
+fn spawn_worker_stdout_monitor(
+ call_id: String,
+ state: Arc<ServiceState>,
+ stdout: std::process::ChildStdout,
+) {
+ thread::spawn(move || {
+ let reader = BufReader::new(stdout);
+ for line in reader.lines() {
+ let Ok(line) = line else {
+ continue;
+ };
+ println!("{line}");
+ if let Ok(value) = serde_json::from_str::<Value>(&line) {
+ apply_worker_event(&call_id, &state, &value);
+ }
+ }
+ });
+}
+
+fn apply_worker_event(call_id: &str, state: &Arc<ServiceState>, value: &Value) {
+ if value.get("type").and_then(Value::as_str) != Some("cv_activity") {
+ return;
+ }
+ if value.get("callId").and_then(Value::as_str) != Some(call_id) {
+ return;
+ }
+ let event_name = value
+ .get("eventName")
+ .and_then(Value::as_str)
+ .unwrap_or("worker_event");
+ let result = value.get("result").and_then(Value::as_str).unwrap_or("ok");
+ let Ok(mut sessions) = state.sessions.lock() else {
+ return;
+ };
+ let Some(session) = sessions.get_mut(call_id) else {
+ return;
+ };
+ session.last_event_type = Some(event_name.to_string());
+ session.last_event_at = now_millis();
+ if result != "ok" {
+ session.status = "FAILED".to_string();
+ session.bot_participant_joined = false;
+ session.bot_track_ready = false;
+ return;
+ }
+ match event_name {
+ "helper_worker_started" => {
+ session.status = "STARTING".to_string();
+ }
+ "bot_participant_joined" => {
+ session.bot_participant_joined = true;
+ }
+ "bot_track_ready" => {
+ session.bot_participant_joined = true;
+ session.bot_track_ready = true;
+ session.status = "STARTED".to_string();
+ }
+ _ => {}
+ }
+}
+
+enum WaitReadyResult {
+ Ready(Value),
+ Failed(HttpResponse),
+}
+
+fn wait_for_session_ready(
+ call_id: &str,
+ state: Arc<ServiceState>,
+ trace_id: &Option<String>,
+) -> WaitReadyResult {
+ let deadline = SystemTime::now()
+ .checked_add(Duration::from_millis(state.config.start_ready_timeout_ms))
+ .unwrap_or_else(SystemTime::now);
+ loop {
+ let snapshot = {
+ let mut sessions = match state.sessions.lock() {
+ Ok(value) => value,
+ Err(_) => {
+ return WaitReadyResult::Failed(error(
+ 200,
+ 2001003404,
+ "HELPER_SESSION_LOCK_FAILED",
+ "runtime_ready",
+ true,
+ trace_id.clone(),
+ ));
+ }
+ };
+ let Some(session) = sessions.get_mut(call_id) else {
+ return WaitReadyResult::Failed(error(
+ 200,
+ 2001003404,
+ "CALL_NOT_FOUND",
+ "runtime_ready",
+ false,
+ trace_id.clone(),
+ ));
+ };
+ let _ = refresh_child_status(session);
+ (
+ session.status.clone(),
+ session.bot_participant_joined,
+ session.bot_track_ready,
+ session_start_data(session),
+ )
+ };
+ if snapshot.0 == "STARTED" && snapshot.1 && snapshot.2 {
+ return WaitReadyResult::Ready(snapshot.3);
+ }
+ if snapshot.0 == "FAILED" || snapshot.0 == "STOPPED" {
+ return WaitReadyResult::Failed(error(
+ 200,
+ 2001003404,
+ "RUNTIME_START_FAILED",
+ "runtime_ready",
+ true,
+ trace_id.clone(),
+ ));
+ }
+ if SystemTime::now() >= deadline {
+ mark_session_start_timeout(call_id, &state);
+ return WaitReadyResult::Failed(error(
+ 200,
+ 2001003404,
+ "RUNTIME_START_TIMEOUT",
+ "runtime_ready",
+ true,
+ trace_id.clone(),
+ ));
+ }
+ thread::sleep(Duration::from_millis(50));
+ }
+}
+
+fn mark_session_start_timeout(call_id: &str, state: &Arc<ServiceState>) {
+ let Ok(mut sessions) = state.sessions.lock() else {
+ return;
+ };
+ let Some(session) = sessions.get_mut(call_id) else {
+ return;
+ };
+ if let Some(child) = session.child.as_mut() {
+ let _ = child.kill();
+ let _ = child.wait();
+ }
+ session.status = "FAILED".to_string();
+ session.bot_participant_joined = false;
+ session.bot_track_ready = false;
+ session.last_event_type = Some("worker_ready_timeout".to_string());
+ session.last_event_at = now_millis();
+}
+
+fn refresh_child_status(session: &mut HelperSession) -> Result<()> {
+ if let Some(child) = session.child.as_mut() {
+ if let Some(status) = child.try_wait()? {
+ session.status = if status.success() {
+ "STOPPED".to_string()
+ } else {
+ "FAILED".to_string()
+ };
+ session.bot_participant_joined = false;
+ session.bot_track_ready = false;
+ session.last_event_type = Some("worker_exited".to_string());
+ session.last_event_at = now_millis();
+ }
+ }
+ Ok(())
+}
+
+fn session_start_data(session: &HelperSession) -> Value {
+ json!({
+ "callId": &session.call_id,
+ "status": &session.status,
+ "runtimeSessionId": &session.runtime_session_id,
+ "botParticipantJoined": session.bot_participant_joined,
+ "botTrackReady": session.bot_track_ready,
+ "firstAudioSource": &session.first_audio_source,
+ })
+}
+
+fn authorized(request: &HttpRequest, config: &ServiceConfig) -> bool {
+ let Some(authorization) = request.headers.get("authorization") else {
+ return false;
+ };
+ let Some(token) = authorization.strip_prefix("Bearer ") else {
+ return false;
+ };
+ config
+ .auth_profiles
+ .values()
+ .any(|profile| constant_time_eq(token.as_bytes(), profile.helper_auth_token.as_bytes()))
+}
+
+fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
+ if left.len() != right.len() {
+ return false;
+ }
+ left.iter()
+ .zip(right.iter())
+ .fold(0u8, |acc, (l, r)| acc | (l ^ r))
+ == 0
+}
+
+fn trace_id(request: &HttpRequest) -> Option<String> {
+ request.headers.get("x-voice-trace-id").cloned()
+}
+
+fn ok(data: Value) -> HttpResponse {
+ HttpResponse::json(
+ 200,
+ json!({
+ "code": 0,
+ "msg": "",
+ "data": data
+ }),
+ )
+}
+
+fn error(
+ http_status: u16,
+ code: i64,
+ reason_code: &str,
+ stage: &str,
+ retryable: bool,
+ trace_id: Option<String>,
+) -> HttpResponse {
+ HttpResponse::json(
+ http_status,
+ json!({
+ "code": code,
+ "msg": reason_code,
+ "data": {
+ "reasonCode": reason_code,
+ "stage": stage,
+ "retryable": retryable,
+ "traceId": trace_id
+ }
+ }),
+ )
+}
+
+fn now_millis() -> u128 {
+ SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .unwrap_or_default()
+ .as_millis()
+}
+
+#[derive(Clone)]
+struct ServiceConfig {
+ bind_addr: String,
+ audio_debug_dump_dir: Option<String>,
+ start_ready_timeout_ms: u64,
+ auth_profiles: HashMap<String, AuthProfile>,
+}
+
+impl ServiceConfig {
+ fn from_env() -> Result<Self> {
+ let bind_addr = env::var("CV_HELPER_SERVICE_BIND")
+ .ok()
+ .filter(|value| !value.trim().is_empty())
+ .unwrap_or_else(|| DEFAULT_BIND_ADDR.to_string());
+ let mut auth_profiles = HashMap::new();
+ for profile in ["default", "local-dev", "dev"] {
+ if let Some(auth_profile) = AuthProfile::from_env(profile) {
+ auth_profiles.insert(profile.to_string(), auth_profile);
+ }
+ }
+ if auth_profiles.is_empty() {
+ return Err(anyhow!("missing helper auth profile config"));
+ }
+ Ok(Self {
+ bind_addr,
+ audio_debug_dump_dir: env::var("CV_AUDIO_DEBUG_DUMP_DIR").ok(),
+ start_ready_timeout_ms: env::var("CV_HELPER_SERVICE_START_READY_TIMEOUT_MS")
+ .ok()
+ .and_then(|value| value.trim().parse::<u64>().ok())
+ .filter(|value| *value > 0)
+ .unwrap_or(DEFAULT_START_READY_TIMEOUT_MS),
+ auth_profiles,
+ })
+ }
+}
+
+#[derive(Clone)]
+struct AuthProfile {
+ helper_auth_token: String,
+ turn_bridge_token: String,
+}
+
+impl AuthProfile {
+ fn from_env(profile: &str) -> Option<Self> {
+ let suffix = profile
+ .chars()
+ .map(|ch| if ch == '-' { '_' } else { ch })
+ .collect::<String>()
+ .to_ascii_uppercase();
+ let helper_auth_token = env::var(format!("CV_HELPER_AUTH_TOKEN_{suffix}"))
+ .or_else(|_| env::var("CV_HELPER_AUTH_TOKEN"))
+ .ok()?;
+ let turn_bridge_token = env::var(format!("CV_TURN_BRIDGE_TOKEN_{suffix}"))
+ .or_else(|_| env::var("CV_RUNTIME_TURN_BRIDGE_TOKEN"))
+ .ok()?;
+ Some(Self {
+ helper_auth_token,
+ turn_bridge_token,
+ })
+ }
+}
+
+struct ServiceState {
+ sessions: Mutex<HashMap<String, HelperSession>>,
+ config: ServiceConfig,
+}
+
+struct HelperSession {
+ call_id: String,
+ trace_id: Option<String>,
+ runtime_session_nonce: String,
+ runtime_session_id: String,
+ status: String,
+ bot_participant_joined: bool,
+ bot_track_ready: bool,
+ first_audio_source: String,
+ active_turn_id: Option<String>,
+ last_event_type: Option<String>,
+ last_event_at: u128,
+ child: Option<Child>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SessionStartRequest {
+ call_id: String,
+ trace_id: Option<String>,
+ runtime_session_nonce: String,
+ livekit: LiveKitStart,
+ turn_bridge: Option<TurnBridgeStart>,
+ auth_profile: Option<String>,
+ audio: Option<AudioStart>,
+ runtime: Option<RuntimeStart>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct LiveKitStart {
+ url: String,
+ room_id: String,
+ bot_token: String,
+ bot_participant_identity: String,
+ user_participant_identity: Option<String>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct TurnBridgeStart {
+ url: String,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct AudioStart {
+ first_audio_source: Option<String>,
+ greeting_audio: Option<GreetingAudioStart>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct GreetingAudioStart {
+ r#type: Option<String>,
+ path_ref: Option<String>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct RuntimeStart {
+ turn_artifact_dir: Option<String>,
+ audio_debug_dump_enabled: Option<bool>,
+}
+
+#[derive(Default, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SessionStopRequest {
+ runtime_session_nonce: Option<String>,
+}
+
+struct HttpRequest {
+ method: String,
+ path: String,
+ headers: HashMap<String, String>,
+ body: Vec<u8>,
+}
+
+impl HttpRequest {
+ fn read(stream: &mut TcpStream) -> Result<Self> {
+ let mut buffer = Vec::new();
+ let mut temp = [0u8; 4096];
+ let header_end;
+ loop {
+ let read = stream.read(&mut temp)?;
+ if read == 0 {
+ return Err(anyhow!("connection closed before request header"));
+ }
+ buffer.extend_from_slice(&temp[..read]);
+ if buffer.len() > HTTP_READ_LIMIT_BYTES {
+ return Err(anyhow!("http request too large"));
+ }
+ if let Some(index) = find_header_end(&buffer) {
+ header_end = index;
+ break;
+ }
+ }
+ let header_text = String::from_utf8_lossy(&buffer[..header_end]).to_string();
+ let mut lines = header_text.split("\r\n");
+ let request_line = lines
+ .next()
+ .ok_or_else(|| anyhow!("missing request line"))?;
+ let parts = request_line.split_whitespace().collect::<Vec<_>>();
+ if parts.len() < 2 {
+ return Err(anyhow!("invalid request line"));
+ }
+ let method = parts[0].to_string();
+ let path = parts[1].split('?').next().unwrap_or(parts[1]).to_string();
+ let mut headers = HashMap::new();
+ for line in lines {
+ if let Some((key, value)) = line.split_once(':') {
+ headers.insert(key.trim().to_ascii_lowercase(), value.trim().to_string());
+ }
+ }
+ let content_length = headers
+ .get("content-length")
+ .and_then(|value| value.parse::<usize>().ok())
+ .unwrap_or(0);
+ let body_start = header_end + 4;
+ let mut body = buffer[body_start..].to_vec();
+ while body.len() < content_length {
+ let read = stream.read(&mut temp)?;
+ if read == 0 {
+ break;
+ }
+ body.extend_from_slice(&temp[..read]);
+ if body.len() > HTTP_READ_LIMIT_BYTES {
+ return Err(anyhow!("http request body too large"));
+ }
+ }
+ body.truncate(content_length);
+ Ok(Self {
+ method,
+ path,
+ headers,
+ body,
+ })
+ }
+}
+
+struct HttpResponse {
+ status: u16,
+ body: Vec<u8>,
+}
+
+impl HttpResponse {
+ fn json(status: u16, body: Value) -> Self {
+ Self {
+ status,
+ body: serde_json::to_vec(&body).unwrap_or_else(|_| b"{}".to_vec()),
+ }
+ }
+
+ fn write(self, stream: &mut TcpStream) -> Result<()> {
+ let reason = match self.status {
+ 200 => "OK",
+ 400 => "Bad Request",
+ 401 => "Unauthorized",
+ 404 => "Not Found",
+ _ => "Internal Server Error",
+ };
+ write!(
+ stream,
+ "HTTP/1.1 {} {}\r\nContent-Type: application/json;charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+ self.status,
+ reason,
+ self.body.len()
+ )?;
+ stream.write_all(&self.body)?;
+ Ok(())
+ }
+}
+
+fn find_header_end(buffer: &[u8]) -> Option<usize> {
+ buffer.windows(4).position(|window| window == b"\r\n\r\n")
+}
+
+fn bool_env(key: &str, default_value: bool) -> bool {
+ match env::var(key) {
+ Ok(value) => match value.trim().to_ascii_lowercase().as_str() {
+ "1" | "true" | "yes" | "on" => true,
+ "0" | "false" | "no" | "off" => false,
+ _ => default_value,
+ },
+ Err(_) => default_value,
+ }
+}
diff --git a/tools/validate-turn-stream-fixture.mjs b/tools/validate-turn-stream-fixture.mjs
new file mode 100755
index 0000000..710365f
--- /dev/null
+++ b/tools/validate-turn-stream-fixture.mjs
@@ -0,0 +1,147 @@
+#!/usr/bin/env node
+
+import fs from 'node:fs'
+import path from 'node:path'
+
+const file = process.argv[2] || 'fixtures/turn-stream-happy.ndjson'
+const absolute = path.resolve(file)
+const raw = fs.readFileSync(absolute, 'utf8')
+const lines = raw.split(/\r?\n/).map((line) => line.trim()).filter(Boolean)
+
+const allowedEvents = new Set([
+ 'reply_playback_mode_selected',
+ 'reply_state',
+ 'reply_audio_chunk',
+ 'device_output',
+ 'turn_completed',
+ 'turn_failed',
+ 'turn_cancelled',
+ 'activity',
+])
+
+let playbackMode = null
+let sawPlaybackStarted = false
+let audioChunkCount = 0
+let sawCompleted = false
+let sawFailed = false
+let sawCancelled = false
+let lastSeq = 0
+let callId = null
+let traceId = null
+let turnId = null
+const supportedAudioFormats = new Set(['pcm_s16le', 'mp3', 'mpeg', 'wav'])
+
+function fail(message) {
+ console.error(`turn stream fixture invalid: ${message}`)
+ process.exit(1)
+}
+
+function requireSameIdentity(event) {
+ if (!event.callId || !event.traceId || !event.turnId) {
+ fail(`${event.event} missing callId/traceId/turnId`)
+ }
+ callId ??= event.callId
+ traceId ??= event.traceId
+ turnId ??= event.turnId
+ if (event.callId !== callId || event.traceId !== traceId || event.turnId !== turnId) {
+ fail(`${event.event} identity mismatch`)
+ }
+}
+
+for (const [index, line] of lines.entries()) {
+ let event
+ try {
+ event = JSON.parse(line)
+ } catch (error) {
+ fail(`line ${index + 1} is not JSON`)
+ }
+
+ if (!allowedEvents.has(event.event)) {
+ fail(`line ${index + 1} has unsupported event ${event.event}`)
+ }
+ requireSameIdentity(event)
+ if (!Number.isInteger(event.seq) || event.seq <= lastSeq) {
+ fail(`${event.event} seq must be strictly increasing`)
+ }
+ lastSeq = event.seq
+
+ if (event.event === 'reply_playback_mode_selected') {
+ if (!['streaming_tts', 'full_tts_fallback'].includes(event.replyPlaybackMode)) {
+ fail('reply_playback_mode_selected has invalid replyPlaybackMode')
+ }
+ playbackMode = event.replyPlaybackMode
+ }
+
+ if (event.event === 'reply_state') {
+ if (!event.state) {
+ fail('reply_state missing state')
+ }
+ if (event.state === 'reply_playback_started') {
+ sawPlaybackStarted = true
+ }
+ }
+
+ if (event.event === 'reply_audio_chunk') {
+ if (!playbackMode) {
+ fail('reply_audio_chunk arrived before reply_playback_mode_selected')
+ }
+ if (!sawPlaybackStarted) {
+ fail('reply_audio_chunk arrived before reply_playback_started')
+ }
+ const chunk = event.audioChunk
+ if (!chunk) {
+ fail('reply_audio_chunk missing audioChunk')
+ }
+ if (!supportedAudioFormats.has(chunk.format)) {
+ fail(`fixture fast path only accepts ${Array.from(supportedAudioFormats).join('/')}, got ${chunk.format}`)
+ }
+ const payload = Buffer.from(chunk.payloadBase64 || '', 'base64')
+ if (payload.length === 0) {
+ fail(`${chunk.format} payload must be non-empty`)
+ }
+ if (chunk.format === 'pcm_s16le') {
+ if (chunk.sampleRate !== 48000 || chunk.channels !== 1) {
+ fail('pcm_s16le chunk must be 48000Hz mono')
+ }
+ if (payload.length % 2 !== 0) {
+ fail('pcm_s16le payload must be 16-bit aligned')
+ }
+ } else {
+ if (chunk.sampleRate != null && !Number.isInteger(chunk.sampleRate)) {
+ fail(`${chunk.format} sampleRate must be an integer when present`)
+ }
+ if (chunk.channels != null && !Number.isInteger(chunk.channels)) {
+ fail(`${chunk.format} channels must be an integer when present`)
+ }
+ }
+ audioChunkCount += 1
+ }
+
+ if (event.event === 'turn_completed') {
+ sawCompleted = true
+ }
+ if (event.event === 'turn_failed') {
+ sawFailed = true
+ }
+ if (event.event === 'turn_cancelled') {
+ sawCancelled = true
+ }
+}
+
+if (lines.length === 0) {
+ fail('fixture is empty')
+}
+if (!sawCompleted && !sawFailed && !sawCancelled) {
+ fail('fixture must end with turn_completed, turn_failed or turn_cancelled')
+}
+if (sawCompleted && audioChunkCount === 0) {
+ fail('completed streaming fixture must contain at least one audio chunk')
+}
+
+console.log(JSON.stringify({
+ ok: true,
+ file: path.relative(process.cwd(), absolute),
+ eventCount: lines.length,
+ audioChunkCount,
+ terminal: sawCompleted ? 'turn_completed' : sawFailed ? 'turn_failed' : 'turn_cancelled',
+}))
--
Gitblit v1.9.3