diff --git a/README.md b/README.md index b4e00bc..db6501d 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,7 @@ Docker, TURN ports, and production notes: [Quick start guide](./docs/quickstart. | **Transport** | WebRTC audio over WHIP ([RFC 9725](https://www.rfc-editor.org/rfc/rfc9725.html)) — one HTTP POST, no signaling socket. Opus/RTP both ways | | **Connectivity** | Built-in Pion STUN/TURN on UDP *and* TCP 3478 — no external coturn. A network handover or NAT rebind is recovered by ICE restart on the same session, so the conversation survives it | | **Turn-taking** | Adaptive VAD that tracks each call's noise floor, plus a debounce that merges mid-sentence pauses into one turn | -| **Interruption** | Barge-in that ducks agent audio, filters backchannels ("mm-hm"), and cancels in-flight LLM and TTS on a confirmed interrupt | +| **Interruption** | Barge-in that ducks agent audio, filters backchannels ("mm-hm"), and cancels in-flight LLM and TTS on a confirmed interrupt. On paths with no echo cancellation, such as telephony, the threshold is bounded by what the agent just sent so it never interrupts itself | | **Streaming** | Streaming STT → streaming LLM → chunk-streaming TTS, so audio starts before synthesis finishes | | **Sessions & events** | Server-generated session IDs, multi-peer sessions, DataChannel events for transcript, response, state, and per-turn latency | | **Reach** | Browser, mobile, backend, CLI, [SIP telephony](https://github.com/streamcoreai/sip-server), and [ESP32](https://github.com/streamcoreai/esp32) endpoints | diff --git a/README.zh-CN.md b/README.zh-CN.md index 3006a9b..2c84e0d 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -75,7 +75,7 @@ Docker、TURN 端口与生产部署注意事项见[快速开始指南](./docs/qu | **传输** | 基于 WHIP（[RFC 9725](https://www.rfc-editor.org/rfc/rfc9725.html)）的 WebRTC 音频 —— 一次 HTTP POST，无需常驻信令连接。双向 Opus/RTP | | **连通性** | 内置 Pion STUN/TURN，同时监听 UDP 与 TCP 3478 —— 无需额外的 coturn。网络切换或 NAT 重新绑定可在同一会话上通过 ICE restart 恢复，对话不会因此中断 | | **轮次控制** | 自适应 VAD 跟踪每通电话的噪声基线，并用去抖把句中停顿合并为同一轮 | -| **插话打断** | Barge-in 先压低智能体音量，过滤 "嗯嗯" 这类回应词，确认打断后取消进行中的 LLM 与 TTS | +| **插话打断** | Barge-in 先压低智能体音量，过滤 "嗯嗯" 这类回应词，确认打断后取消进行中的 LLM 与 TTS。在没有回声消除的链路（如电话）上，打断阈值会以智能体刚发出的音频为下限，因此不会被自己的声音打断 | | **流式链路** | 流式 STT → 流式 LLM → 分块流式 TTS，合成尚未结束音频就已开始播放 | | **会话与事件** | 服务端生成会话 ID、多 peer 会话，DataChannel 推送转写、回复、状态与每轮延迟 | | **接入范围** | 浏览器、移动端、后端服务、CLI、[SIP 电话](https://github.com/streamcoreai/sip-server) 与 [ESP32](https://github.com/streamcoreai/esp32) 设备 | diff --git a/config.toml.example b/config.toml.example index a775c85..64d36d2 100644 --- a/config.toml.example +++ b/config.toml.example @@ -29,6 +29,16 @@ turn_merge_ms = 350 # Debounce window for merging finals into one turn. # rag_prefetch = false # Start retrieval during the merge window instead of after it # readback_bargein_guard_enabled = false # Ignore weak barge-ins while the agent reads values back +# Echo reference for barge-in, decided per session. On a path with no AEC +# (SIP/PCMU telephony) the agent's own voice comes back loud enough to look +# like an interruption; a browser cancels it in getUserMedia and needs no +# bound. "auto" follows what each client declares on the WHIP URL (aec=none), +# which sip-server already sends, so one server can host both. +# echo_guard = "auto" # auto | always | off. "always" is for raw-path clients that send no hint +# echo_guard_gain = 0.6 # Echo cannot exceed this fraction of what produced it +# echo_guard_margin = 1.8 # How far inbound must clear the echo bound to count as the caller +# echo_guard_window_ms = 400 # How long sent audio stays in the reference window + # Provider selection # Speech-to-speech mode. Setting a provider here replaces [stt], [llm], and diff --git a/docs/capabilities.md b/docs/capabilities.md index 9bcc693..8ea23bb 100644 --- a/docs/capabilities.md +++ b/docs/capabilities.md @@ -80,6 +80,7 @@ StreamCore can run a complete speech-to-agent-to-speech pipeline, but that is on - Opus decode → PCM → pipeline → PCM → Opus encode → RTP - Energy-based VAD with configurable onset/offset frame counts, adapting to each call's noise floor so a quiet caller on a clean line and a caller beside a road both register - Barge-in on a faster VAD profile: agent audio ducks while the caller talks over it and recovers if the interruption turns out to be a backchannel +- Echo reference for paths with no AEC, such as SIP/PCMU telephony: the barge-in threshold is bounded by the RMS the server just sent, so the agent stops interrupting itself on its own returning voice. Applied per session from a client hint, so one instance serves browsers and phone calls without either compromising the other - Turn debounce that merges consecutive final transcripts, so "I want to… um… book a table" is answered once, not twice - Sentence-boundary chunking so TTS starts before the LLM finishes, and chunk-level streaming so audio plays before a sentence is fully synthesized - Optional per-utterance delivery tags — the model may prefix a sentence with `[warm]`, `[empathetic]`, `[calm]`, or `[excited]`, which map to provider voice controls and are never spoken aloud diff --git a/docs/capabilities.zh-CN.md b/docs/capabilities.zh-CN.md index 81faf64..6429c5f 100644 --- a/docs/capabilities.zh-CN.md +++ b/docs/capabilities.zh-CN.md @@ -80,6 +80,7 @@ StreamCore 可以跑通一条完整的「语音 → 智能体 → 语音」链 - Opus 解码 → PCM → 处理流水线 → PCM → Opus 编码 → RTP - 基于能量的 VAD，起止帧数可配置，并自适应每通电话的噪声基线，因此安静线路上的轻声用户与路边嘈杂环境中的用户都能被正确识别 - 使用更快 VAD 配置的 barge-in：用户抢话时智能体音量随即压低，若判定只是回应词则恢复 +- 面向没有 AEC 的链路（如 SIP/PCMU 电话）的回声参考：用服务器刚发出音频的 RMS 给打断阈值加下限，智能体不会再被自己绕回来的声音打断。该判定按会话依据客户端提示生效，因此同一个实例可以同时服务浏览器和电话，互不影响 - 轮次去抖，把连续的 final 转写合并，让「我想…嗯…订个位」只被回答一次，而不是两次 - 按句边界分块，使 TTS 在 LLM 生成结束前就开始；再按 chunk 级流式播放，使一句话尚未合成完就已出声 - 可选的逐句表达标签 —— 模型可以在句首加上 `[warm]`、`[empathetic]`、`[calm]` 或 `[excited]`，它们会映射到服务商的音色控制参数，并且永远不会被读出来 diff --git a/docs/configuration.md b/docs/configuration.md index b77964d..9c9f188 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -32,6 +32,10 @@ user_speech_quiet_ms = 600 # Quiet period after the caller stops befor turn_merge_ms = 350 # Debounce window for merging finals into one turn # rag_prefetch = false # Start retrieval during the merge window instead of after it # readback_bargein_guard_enabled = false # Ignore weak barge-ins while the agent reads values back +# echo_guard = "auto" # auto | always | off. Auto follows each client's aec hint +# echo_guard_gain = 0.6 # Echo cannot exceed this fraction of what produced it +# echo_guard_margin = 1.8 # How far inbound must clear the echo bound to count as the caller +# echo_guard_window_ms = 400 # How long sent audio stays in the reference window # Speech-to-speech. When set, replaces [stt], [llm], and [tts] entirely. [realtime] @@ -160,6 +164,18 @@ Notes: - `pipeline.user_speech_quiet_ms` is how long the caller must be quiet before the agent starts speaking. - `pipeline.rag_prefetch` overlaps retrieval with the turn-merge window. Off by default; it issues a speculative embedding + search that is discarded if the turn text changes. - `pipeline.readback_bargein_guard_enabled` keeps weak corrections and backchannels from cutting off a confirmation readback. Only explicit commands (stop, cancel, hang up) interrupt. Off by default. +- `pipeline.echo_guard` stops the agent barging in on its own voice. A browser runs AEC before audio reaches the server, so the VAD never sees the agent's output come back; over a carrier there is no AEC anywhere in the path, the returning audio is attenuated but structurally identical to speech, and an energy test cannot tell it from a caller. The guard keeps a rolling window of the RMS the server actually sent and requires inbound to clear `sent_rms x echo_guard_gain x echo_guard_margin` before it counts as an interruption. While the agent is silent that bound is zero and the ordinary adaptive threshold governs, so a quiet caller on a clean line is unaffected. + + The decision is per session, because one instance usually serves browsers and SIP calls at once and the two need opposite answers. A client declares a raw path by adding `aec=none` to the WHIP URL, which `sip-server` sends on every call; browsers send nothing and are read as already cancelled. + + | `echo_guard` | Effect | + | --- | --- | + | `"auto"` (default) | On for peers that sent `aec=none`, off for everyone else | + | `"always"` | On for every peer. For a raw-path client you cannot change to send the hint | + | `"off"` | Never on | + + Leave it on `"auto"` unless you have a client on a path with no AEC that you cannot modify. Setting `"always"` on a server that also hosts browsers makes genuine browser barge-ins clear the agent's own output level first, which is the regression the per-session default exists to avoid. +- `pipeline.echo_guard_gain`, `pipeline.echo_guard_margin`, and `pipeline.echo_guard_window_ms` tune that bound, for the sessions it applies to. The gain is how loud echo can be relative to the audio that produced it, the margin is what separates double-talk from echo, and the window should cover the round trip of the carrier's echo. The defaults (0.6, 1.8, 400ms) were measured on an 8kHz mu-law path; retune only against recordings of your own. The bound follows the barge-in duck on its own, since it is sampled from what goes on the wire after attenuation. - `deepgram.endpointing` and `deepgram.utterance_end_ms` tune when a turn is considered finished upstream; the turn-merge debounce runs on top of them. - `deepgram.tts_model` picks the Aura voice; STT (`model`) and TTS (`tts_model`) share the one API key. Voices are named `[family]-[voice]-[language]` — see [Deepgram's voice list](https://developers.deepgram.com/docs/tts-models). - `cartesia.max_concurrency` should match your plan's TTS concurrency limit — Cartesia counts active generations, not calls, and returns 429 past the limit. diff --git a/docs/configuration.zh-CN.md b/docs/configuration.zh-CN.md index be23237..60d0f58 100644 --- a/docs/configuration.zh-CN.md +++ b/docs/configuration.zh-CN.md @@ -26,6 +26,10 @@ user_speech_quiet_ms = 600 # Quiet period after the caller stops befor turn_merge_ms = 350 # Debounce window for merging finals into one turn # rag_prefetch = false # Start retrieval during the merge window instead of after it # readback_bargein_guard_enabled = false # Ignore weak barge-ins while the agent reads values back +# echo_guard = "auto" # auto | always | off. Auto follows each client's aec hint +# echo_guard_gain = 0.6 # Echo cannot exceed this fraction of what produced it +# echo_guard_margin = 1.8 # How far inbound must clear the echo bound to count as the caller +# echo_guard_window_ms = 400 # How long sent audio stays in the reference window # Speech-to-speech. When set, replaces [stt], [llm], and [tts] entirely. [realtime] @@ -152,6 +156,18 @@ voice = "en-Emma_woman" - `pipeline.user_speech_quiet_ms` 是用户需要安静多久，智能体才开始说话。 - `pipeline.rag_prefetch` 让检索与轮次合并窗口重叠。默认关闭；它会发出一次推测性的 embedding + 检索，若该轮文本发生变化则丢弃。 - `pipeline.readback_bargein_guard_enabled` 可避免弱纠正与回应词打断智能体的确认复述。只有明确的命令（stop、cancel、hang up）才会打断。默认关闭。 +- `pipeline.echo_guard` 防止智能体被自己的声音打断。浏览器会在音频到达服务器之前先做 AEC，因此 VAD 根本看不到智能体自己的输出绕回来；而在电话线路上整条链路没有任何 AEC，回声虽然衰减了，但结构上和语音完全一样，单靠能量判据无法与真人区分。该开关会维护一个滚动窗口，记录服务器实际发出音频的 RMS，只有当上行音频超过 `已发送 RMS x echo_guard_gain x echo_guard_margin` 时才算作打断。智能体沉默时该下限为零，判定重新交回自适应阈值，因此干净线路上说话轻的来电者不受影响。 + + 这个判断是按会话而不是按服务器做的：同一个实例通常同时服务浏览器和 SIP 通话，而两者需要相反的答案。客户端通过在 WHIP URL 上加 `aec=none` 来声明自己处在没有 AEC 的链路上，`sip-server` 每通电话都会带上它；浏览器什么都不发，会被视为已经做过回声消除。 + + | `echo_guard` | 行为 | + | --- | --- | + | `"auto"`（默认） | 对发送了 `aec=none` 的会话开启，其余关闭 | + | `"always"` | 对所有会话开启。用于无法改动、又发不出该提示的裸链路客户端 | + | `"off"` | 始终关闭 | + + 除非你有一个无法修改、又跑在无 AEC 链路上的客户端，否则请保持 `"auto"`。在同时服务浏览器的服务器上设为 `"always"`，会让浏览器端真实的插话必须先盖过智能体自身的输出电平，而按会话判断的默认值正是为了避免这种回退。 +- `pipeline.echo_guard_gain`、`pipeline.echo_guard_margin` 和 `pipeline.echo_guard_window_ms` 用于调节这个下限（仅对开启了该判定的会话生效）。gain 是回声相对于产生它的音频最多能有多响，margin 用于区分双讲与回声，window 应覆盖线路回声的往返时间。默认值（0.6、1.8、400ms）是在 8kHz µ-law 链路上实测得到的；只有拿到你自己链路的录音再去重新调参。该下限会自动跟随打断时的音量压低，因为它采样的是经过衰减后真正发到线路上的信号。 - `deepgram.endpointing` 与 `deepgram.utterance_end_ms` 调节上游认定一轮结束的时机；轮次合并去抖运行在它们之上。 - `deepgram.tts_model` 选择 Aura 音色；STT（`model`）与 TTS（`tts_model`）共用同一个 API key。音色命名规则为 `[family]-[voice]-[language]` —— 见 [Deepgram 音色列表](https://developers.deepgram.com/docs/tts-models)。 - `cartesia.max_concurrency` 应与你套餐的 TTS 并发上限一致 —— Cartesia 统计的是进行中的生成数而不是通话数，超限会返回 429。 diff --git a/docs/protocol.md b/docs/protocol.md index 966bf0b..6785249 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -20,6 +20,16 @@ The client creates an SDP offer, gathers ICE candidates, and `POST`s it to `/whi This implementation aligns with the core WHIP flow: `POST` with `application/sdp`, `201 Created` with the answer, `Location` for the session URL, `ETag` for the ICE session, `PATCH` for ICE restart, `DELETE` for teardown, `OPTIONS` with `Accept-Post`, and full ICE gathering on both sides. Audio is `sendrecv`, with a DataChannel for bidirectional events. +### Optional query parameters on `POST /whip` + +| Parameter | Values | Meaning | +|---|---|---| +| `resume` | a resume token | Reattach this offer to an existing conversation — see [Session resume](#session-resume) | +| `direction` | `inbound`, `outbound` | Which way a telephony call was placed. Selects `pipeline.greeting_outgoing` for outbound calls | +| `aec` | `none` | Nothing upstream of the server cancels echo on this path, so the agent's own voice comes back to it. Turns on the barge-in echo bound for this session under the default `pipeline.echo_guard = "auto"` | + +All are optional and unknown values are ignored. Omitting `aec` means echo cancellation ran upstream, which is true of every browser, so a client only sends it when running on a raw path — `sip-server` sends it on every call. + ### ICE restart A transient network event — a phone moving between Wi-Fi and cellular, a laptop changing networks, a NAT rebinding after an idle gap — breaks connectivity without ending the call. Recovering by `POST`ing a fresh offer would allocate a new session, a new pipeline, and a new LLM client, so the conversation history and the rolling summary would be gone and the greeting would replay. `PATCH` recovers the *same* connection instead: new ICE credentials and candidates, but the same `PeerConnection`, the same DTLS association, the same tracks, and the same running pipeline. diff --git a/docs/protocol.zh-CN.md b/docs/protocol.zh-CN.md index 4ea202f..dc49b4d 100644 --- a/docs/protocol.zh-CN.md +++ b/docs/protocol.zh-CN.md @@ -20,6 +20,16 @@ 本实现与 WHIP 的核心流程一致：以 `application/sdp` 发起 `POST`，用 `201 Created` 返回 answer，用 `Location` 给出会话 URL，用 `ETag` 标识 ICE 会话，用 `PATCH` 做 ICE 重启，用 `DELETE` 销毁，用 `OPTIONS` 返回 `Accept-Post`，并在双端做完整 ICE 收集。音频为 `sendrecv`，并带一个用于双向事件的 DataChannel。 +### `POST /whip` 的可选查询参数 + +| 参数 | 取值 | 含义 | +|---|---|---| +| `resume` | 恢复令牌 | 把本次 offer 重新挂到既有会话上，见[会话恢复](#会话恢复session-resume) | +| `direction` | `inbound`、`outbound` | 电话呼叫的方向。呼出通话会据此选用 `pipeline.greeting_outgoing` | +| `aec` | `none` | 服务器上游没有任何环节做回声消除，智能体自己的声音会绕回来。在默认的 `pipeline.echo_guard = "auto"` 下，该会话会启用打断回声下限 | + +以上都是可选的，无法识别的取值会被忽略。不带 `aec` 表示上游已经做过回声消除，浏览器都属于这种情况，因此只有跑在裸链路上的客户端才需要发送它 —— `sip-server` 每通电话都会带上。 + ### ICE 重启 短暂的网络事件 —— 手机在 Wi-Fi 与蜂窝之间切换、笔记本更换网络、空闲后 NAT 重新绑定 —— 会中断连通性，但通话本身并未结束。若用重新 `POST` offer 的方式恢复，会分配新的会话、新的流水线和新的 LLM 客户端，对话历史与滚动摘要随之丢失，开场白也会重播。`PATCH` 恢复的是*同一条*连接：ICE 凭据与候选是新的，但 `PeerConnection`、DTLS 关联、媒体轨道以及正在运行的流水线都保持不变。 diff --git a/internal/config/config.go b/internal/config/config.go index 39c4a59..3053a3b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -80,6 +80,36 @@ type PipelineConfig struct { // barge-in behaviour. ReadbackBargeInGuardEnabled bool `toml:"readback_bargein_guard_enabled"` + // EchoGuard keeps the barge-in VAD from hearing the agent's own voice + // coming back. It bounds the interrupt threshold by the RMS of what the + // server just sent, which is the only echo reference available when + // nothing in the path runs AEC. + // + // The choice is per session, not per server: one instance commonly serves + // browsers and SIP calls at once, and the two need opposite answers. A + // browser cancels echo in getUserMedia, so the bound would block nothing + // it was going to catch while making a genuine barge-in clear the agent's + // own level first. A carrier leg has no AEC anywhere. + // + // "auto" — on for peers that arrive with aec=none, off otherwise (default) + // "always" — on for every peer, for clients that cannot send the hint + // "off" — never + EchoGuard string `toml:"echo_guard"` + + // EchoGuardGain bounds echo against the audio that produced it (echo + // cannot be louder than its source), and EchoGuardMargin is how far + // inbound must clear that bound to count as the caller talking rather + // than the agent's tail. Default 0.6 and 1.8; retune only against + // recordings of the actual path. + EchoGuardGain float64 `toml:"echo_guard_gain"` + EchoGuardMargin float64 `toml:"echo_guard_margin"` + + // EchoGuardWindowMs is how long outbound audio stays in the reference + // window, sized to cover the round trip of the carrier's echo. Default + // 400. Too short and the tail of a phrase escapes the bound; too long + // and the agent stays hard to interrupt after it has stopped talking. + EchoGuardWindowMs int `toml:"echo_guard_window_ms"` + // RAGPrefetch starts retrieval speculatively during the turn-merge // window so embedding and vector search overlap the debounce instead of // adding to it. @@ -450,6 +480,18 @@ func Load(path string) (*Config, error) { cfg.Pipeline.TurnMergeMs = 350 } + // Echo-reference tuning. Only consulted for sessions the guard is on for. + setDefault(&cfg.Pipeline.EchoGuard, EchoGuardAuto) + if cfg.Pipeline.EchoGuardGain == 0 { + cfg.Pipeline.EchoGuardGain = 0.6 + } + if cfg.Pipeline.EchoGuardMargin == 0 { + cfg.Pipeline.EchoGuardMargin = 1.8 + } + if cfg.Pipeline.EchoGuardWindowMs == 0 { + cfg.Pipeline.EchoGuardWindowMs = 400 + } + // Default barge-in to true if not explicitly set if cfg.Pipeline.BargeIn == nil { t := true @@ -466,6 +508,10 @@ func Load(path string) (*Config, error) { cfg.Grok.Transcription = &t } + if err := cfg.validateEchoGuard(); err != nil { + return nil, err + } + if err := cfg.validateRealtime(); err != nil { return nil, err } @@ -483,6 +529,36 @@ func (c *Config) RealtimeEnabled() bool { // Without this a typo'd provider or a missing key only surfaces when the // first caller connects, which reads as a broken deployment rather than a // misconfigured one. +// Echo-guard modes for pipeline.echo_guard. +const ( + EchoGuardAuto = "auto" + EchoGuardAlways = "always" + EchoGuardOff = "off" +) + +// EchoGuardFor reports whether the barge-in echo bound applies to a peer, +// given whether that peer declared it has no acoustic echo cancellation. +func (c *Config) EchoGuardFor(aecAbsent bool) bool { + switch c.Pipeline.EchoGuard { + case EchoGuardAlways: + return true + case EchoGuardAuto: + return aecAbsent + default: + return false + } +} + +func (c *Config) validateEchoGuard() error { + switch c.Pipeline.EchoGuard { + case EchoGuardAuto, EchoGuardAlways, EchoGuardOff: + return nil + default: + return fmt.Errorf("[pipeline] echo_guard = %q must be %q, %q, or %q", + c.Pipeline.EchoGuard, EchoGuardAuto, EchoGuardAlways, EchoGuardOff) + } +} + func (c *Config) validateRealtime() error { if !c.RealtimeEnabled() { return nil diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 4baa7de..76a2d04 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -432,3 +432,74 @@ func TestValidateRealtimeRejectsTooManyKeyterms(t *testing.T) { t.Error("101 keyterms accepted; the documented maximum is 100") } } + +// A single instance commonly serves browsers and SIP calls at once, so the +// echo bound has to be decided per peer rather than per server. +func TestEchoGuardFor(t *testing.T) { + cases := []struct { + mode string + aecAbsent bool + want bool + }{ + {EchoGuardAuto, true, true}, // SIP leg declares aec=none + {EchoGuardAuto, false, false}, // browser, AEC ran in getUserMedia + {EchoGuardAlways, false, true}, // client that cannot send the hint + {EchoGuardAlways, true, true}, + {EchoGuardOff, true, false}, + {EchoGuardOff, false, false}, + } + for _, c := range cases { + cfg := &Config{} + cfg.Pipeline.EchoGuard = c.mode + if got := cfg.EchoGuardFor(c.aecAbsent); got != c.want { + t.Errorf("mode %q aecAbsent=%v: got %v, want %v", c.mode, c.aecAbsent, got, c.want) + } + } +} + +// An unset echo_guard must behave as "auto", so an upgrade fixes SIP without +// touching browser sessions and without anyone editing a config. +func TestEchoGuardDefaultsToAuto(t *testing.T) { + cfg := &Config{} + if cfg.EchoGuardFor(true) { + t.Error("zero-value config must not enable the guard before defaults are applied") + } + cfg.Pipeline.EchoGuard = EchoGuardAuto + if !cfg.EchoGuardFor(true) || cfg.EchoGuardFor(false) { + t.Error("auto must follow the peer's aec hint") + } +} + +func TestEchoGuardRejectsUnknownMode(t *testing.T) { + cfg := &Config{} + cfg.Pipeline.EchoGuard = "yes" + if err := cfg.validateEchoGuard(); err == nil { + t.Error("expected an error for an unknown echo_guard mode") + } +} + +func TestLoadDefaultsEchoGuardToAuto(t *testing.T) { + path := writeConfig(t, "[pipeline]\nbarge_in = true\n") + + cfg, err := Load(path) + if err != nil { + t.Fatalf("config rejected: %v", err) + } + if cfg.Pipeline.EchoGuard != EchoGuardAuto { + t.Errorf("echo_guard = %q, want the default %q", cfg.Pipeline.EchoGuard, EchoGuardAuto) + } + if !cfg.EchoGuardFor(true) { + t.Error("a SIP peer declaring aec=none should get the guard by default") + } + if cfg.EchoGuardFor(false) { + t.Error("a browser peer must not get the guard by default") + } +} + +func TestLoadRejectsBadEchoGuardMode(t *testing.T) { + path := writeConfig(t, "[pipeline]\necho_guard = \"on\"\n") + + if _, err := Load(path); err == nil { + t.Error("Load accepted an unknown echo_guard mode") + } +} diff --git a/internal/pipeline/outbound.go b/internal/pipeline/outbound.go index fa39a8f..aa42d71 100644 --- a/internal/pipeline/outbound.go +++ b/internal/pipeline/outbound.go @@ -87,6 +87,11 @@ func (p *Pipeline) encodeAndSend(frame PCMFrame) { samples = ducked } + // Recorded here rather than at enqueue: this is the signal that actually + // goes on the wire, duck attenuation and padding included, and it is what + // can come back as echo on a path with no AEC. No-op when the guard is off. + p.echoGuard.Observe(samples) + opusData, err := p.encoder.Encode(samples) if err != nil { log.Printf("[sender] encode error: %v", err) diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index face30c..06d1ea9 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -10,6 +10,7 @@ import ( "strings" "sync" "sync/atomic" + "time" "github.com/pion/webrtc/v4" "github.com/streamcoreai/streamcore-server/internal/audio" @@ -75,6 +76,10 @@ type Pipeline struct { // VAD vad *vad.Detector bargeInVAD *vad.Detector + // echoGuard feeds the barge-in VAD the RMS of what the sender just put + // on the wire, so returning agent audio on a path with no AEC does not + // read as the caller interrupting. Nil unless pipeline.echo_guard is on. + echoGuard *vad.EchoGuard // Bounded channels inPCMCh chan PCMFrame @@ -180,7 +185,7 @@ func New( sendEvent func(interface{}) error, pluginMgr *plugin.Manager, ragClient rag.Client, - direction string, + opts PeerOptions, conv *ConversationState, resumed bool, ) (*Pipeline, error) { @@ -253,11 +258,25 @@ func New( // most audible way to tell them the agent forgot. suppressGreeting: resumed, sendEvent: sendEvent, - direction: direction, + direction: opts.Direction, ssrc: 12345678, markerNext: true, } + // Per session, not per server: one instance serves browsers and SIP calls + // at once, and a browser has already cancelled its echo upstream. + if cfg.EchoGuardFor(opts.AECAbsent) { + p.echoGuard = vad.NewEchoGuard( + time.Duration(cfg.Pipeline.EchoGuardWindowMs)*time.Millisecond, + cfg.Pipeline.EchoGuardGain, + cfg.Pipeline.EchoGuardMargin, + ) + p.bargeInVAD.SetEchoReference(p.echoGuard) + log.Printf("[pipeline] echo guard on (mode %s, window %dms, gain %.2f, margin %.2f)", + cfg.Pipeline.EchoGuard, cfg.Pipeline.EchoGuardWindowMs, + cfg.Pipeline.EchoGuardGain, cfg.Pipeline.EchoGuardMargin) + } + // Realtime mode wires its own tools and instructions when the // speech-to-speech session opens; the rest of this function configures // the LLM client, which does not exist in that mode. diff --git a/internal/pipeline/types.go b/internal/pipeline/types.go index d6c58e5..2514bf4 100644 --- a/internal/pipeline/types.go +++ b/internal/pipeline/types.go @@ -50,3 +50,16 @@ type stateMsg struct { Type string `json:"type"` State string `json:"state"` } + +// PeerOptions carries what the signalling layer learned about a peer that the +// media pipeline has no way to see for itself. +type PeerOptions struct { + // Direction is "outbound" for an outgoing SIP call, empty otherwise. + Direction string + + // AECAbsent is set when the client says nothing upstream of the server + // removes the agent's own voice from its inbound audio (aec=none on the + // WHIP URL). A browser cancels echo in getUserMedia, so absent means + // present, which is what every client predating the hint is. + AECAbsent bool +} diff --git a/internal/session/session.go b/internal/session/session.go index 50eab8c..8c49e77 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -109,7 +109,11 @@ func NewSession(id string, cfg *config.Config, pluginMgr *plugin.Manager, ragCli // AddPeer creates a new Pion peer and launches a goroutine that waits for // the remote audio track to arrive, then builds and starts the channel-based // pipeline. Event messages are delivered via the peer's DataChannel. -func (s *Session) AddPeer(peerID string, direction string) (*peer.Peer, error) { +// PeerOptions is re-exported so the signalling layer, which already depends on +// this package, does not need to reach into pipeline for it. +type PeerOptions = pipeline.PeerOptions + +func (s *Session) AddPeer(peerID string, opts PeerOptions) (*peer.Peer, error) { s.mu.Lock() defer s.mu.Unlock() @@ -164,7 +168,7 @@ func (s *Session) AddPeer(peerID string, direction string) (*peer.Peer, error) { return } - pl, err = pipeline.New(p.Context(), s.cfg, remoteTrack, p.LocalTrack(), p.SendEvent, s.pluginMgr, s.ragClient, direction, conv, resumed) + pl, err = pipeline.New(p.Context(), s.cfg, remoteTrack, p.LocalTrack(), p.SendEvent, s.pluginMgr, s.ragClient, opts, conv, resumed) if err != nil { log.Printf("[session:%s] pipeline create error: %v", s.ID, err) p.Close() diff --git a/internal/signaling/handler.go b/internal/signaling/handler.go index bc564b0..a6fde2b 100644 --- a/internal/signaling/handler.go +++ b/internal/signaling/handler.go @@ -119,6 +119,19 @@ func NewWHIPHandler(sm *session.Manager) http.HandlerFunc { } } +// peerOptionsFrom reads the optional per-peer metadata a client appends to the +// WHIP URL. aec=none is how a client on a raw path (sip-server on a carrier +// leg) says the agent's own voice comes back to it uncancelled, which the +// barge-in VAD cannot work out for itself. Anything else, including absent, +// means echo cancellation ran upstream, as it does in every browser. +func peerOptionsFrom(r *http.Request) session.PeerOptions { + q := r.URL.Query() + return session.PeerOptions{ + Direction: q.Get("direction"), + AECAbsent: q.Get("aec") == "none", + } +} + // handleWHIPPost implements RFC 9725 §4.2 Ingest Session Setup. // A new sessionId (UUID) is generated for each POST. func handleWHIPPost(w http.ResponseWriter, r *http.Request, sm *session.Manager) { @@ -140,8 +153,7 @@ func handleWHIPPost(w http.ResponseWriter, r *http.Request, sm *session.Manager) return } - // Read optional metadata from query parameters. - direction := r.URL.Query().Get("direction") + peerOpts := peerOptionsFrom(r) // A resume token reattaches this offer to a conversation whose transport // died — the case ICE restart cannot cover, because by the time the client @@ -191,7 +203,7 @@ func handleWHIPPost(w http.ResponseWriter, r *http.Request, sm *session.Manager) sessionID := ses.ID peerID := sessionID - p, err := ses.AddPeer(peerID, direction) + p, err := ses.AddPeer(peerID, peerOpts) if err != nil { log.Printf("[whip] add peer error: %v", err) http.Error(w, "failed to create peer", http.StatusInternalServerError) diff --git a/internal/signaling/handler_test.go b/internal/signaling/handler_test.go index 4e56888..692841b 100644 --- a/internal/signaling/handler_test.go +++ b/internal/signaling/handler_test.go @@ -161,7 +161,7 @@ func TestPatchWildcardIfMatchPassesThePrecondition(t *testing.T) { func TestPatchTrickleOnlyIsDeclined(t *testing.T) { h, sm := testHandler(t) s := sm.GetOrCreate("s1") - if _, err := s.AddPeer("s1", ""); err != nil { + if _, err := s.AddPeer("s1", session.PeerOptions{}); err != nil { t.Fatalf("AddPeer: %v", err) } @@ -183,7 +183,7 @@ func TestPatchTrickleOnlyIsDeclined(t *testing.T) { func TestPatchOnPeerWithoutNegotiation(t *testing.T) { h, sm := testHandler(t) s := sm.GetOrCreate("s1") - if _, err := s.AddPeer("s1", ""); err != nil { + if _, err := s.AddPeer("s1", session.PeerOptions{}); err != nil { t.Fatalf("AddPeer: %v", err) } @@ -367,3 +367,29 @@ func TestResumeRetiresThePreDropETag(t *testing.T) { t.Fatalf("stale ETag after a resume: status = %d, want 412", rec.Code) } } + +// A browser never sends the hint, so the default must be "echo already +// cancelled" — anything else would arm the echo bound against sessions whose +// audio getUserMedia already cleaned. +func TestPeerOptionsFromQuery(t *testing.T) { + cases := []struct { + query string + direction string + aecAbsent bool + }{ + {"", "", false}, + {"?direction=outbound", "outbound", false}, + {"?aec=none", "", true}, + {"?direction=inbound&aec=none", "inbound", true}, + {"?aec=upstream", "", false}, + {"?aec=", "", false}, + } + for _, c := range cases { + r := httptest.NewRequest(http.MethodPost, "/whip"+c.query, nil) + got := peerOptionsFrom(r) + if got.Direction != c.direction || got.AECAbsent != c.aecAbsent { + t.Errorf("%q: got %+v, want direction=%q aecAbsent=%v", + c.query, got, c.direction, c.aecAbsent) + } + } +} diff --git a/internal/vad/echo.go b/internal/vad/echo.go new file mode 100644 index 0000000..808aba3 --- /dev/null +++ b/internal/vad/echo.go @@ -0,0 +1,152 @@ +package vad + +import ( + "sync" + "time" +) + +// Defaults for the echo reference, tuned on an 8kHz µ-law carrier path. +const ( + DefaultEchoWindow = 400 * time.Millisecond + DefaultEchoGain = 0.6 + DefaultEchoMargin = 1.8 +) + +// EchoGuard bounds barge-in detection by what the agent itself just sent. +// +// A browser runs AEC before audio ever reaches the server, so on the WebRTC +// path the detector never sees the agent's own output coming back. Over a +// carrier there is no AEC anywhere in the path. The returning audio is +// attenuated but structurally identical to speech, so an RMS-vs-noise-floor +// test cannot separate the two: adaptiveMinThreshold and noiseFloorMultiplier +// are tuned against background noise, and echo is a copy of speech. +// +// Raising the fixed threshold instead only trades self-barging for being deaf +// to quiet callers. The reference that does separate the two cases is the +// outbound signal. Echo cannot be louder than the audio that produced it, so +// recent outbound RMS scaled by gain bounds it from above, and inbound counts +// as an interruption only once it clears that bound by margin. The margin is +// what tells double-talk from echo. While the agent is silent the bound is +// zero and the ordinary adaptive threshold governs, so sensitivity to a quiet +// caller on a clean line is untouched. +type EchoGuard struct { + mu sync.Mutex + window time.Duration + gain float64 + margin float64 + + // Ring of recent outbound frames, oldest at start, count entries live. + buf []echoSample + start int + count int + + now func() time.Time +} + +type echoSample struct { + at time.Time + rms float64 +} + +// NewEchoGuard builds a guard over the given reference window. Zero or +// negative parameters fall back to the defaults. +func NewEchoGuard(window time.Duration, gain, margin float64) *EchoGuard { + if window <= 0 { + window = DefaultEchoWindow + } + if gain <= 0 { + gain = DefaultEchoGain + } + if margin <= 0 { + margin = DefaultEchoMargin + } + // Sized for frames as short as 5ms so a full window always fits. Nothing + // on the sender path pushes faster than realtime; if something did, the + // oldest entry drops and the window shortens rather than the ring growing. + return &EchoGuard{ + window: window, + gain: gain, + margin: margin, + buf: make([]echoSample, int(window/(5*time.Millisecond))+8), + now: time.Now, + } +} + +// Observe records one frame of outbound audio. +// +// Pass the samples that actually go on the wire, after any duck attenuation. +// Outbound RMS is naturally available at enqueue, but a ducked talkspurt is +// ~12dB quieter by the time it is sent, and a bound built from the un-ducked +// signal would hold the barge-in threshold high through the whole duck, which +// is when the caller most needs to be heard. +func (g *EchoGuard) Observe(samples []int16) { + if g == nil { + return + } + rms := RMSEnergy(samples) + at := g.now() + + g.mu.Lock() + defer g.mu.Unlock() + g.prune(at) + if g.count == len(g.buf) { + g.start = (g.start + 1) % len(g.buf) + g.count-- + } + g.buf[(g.start+g.count)%len(g.buf)] = echoSample{at: at, rms: rms} + g.count++ +} + +// Floor is an upper bound on how loud the agent's echo can be right now: the +// loudest frame sent inside the window, scaled by the gain. It returns to +// zero once the agent has been quiet for a full window. +// +// Loudest rather than mean, because the echo path delay is unknown — anything +// sent inside the window could be what is arriving now. +func (g *EchoGuard) Floor() float64 { + if g == nil { + return 0 + } + g.mu.Lock() + defer g.mu.Unlock() + g.prune(g.now()) + + var peak float64 + for i := 0; i < g.count; i++ { + if v := g.buf[(g.start+i)%len(g.buf)].rms; v > peak { + peak = v + } + } + return peak * g.gain +} + +// Threshold is the inbound RMS a frame must exceed to read as the caller +// talking rather than the agent's own voice returning. Zero while the agent +// is silent, which hands the decision back to the adaptive threshold. +func (g *EchoGuard) Threshold() float64 { + if g == nil { + return 0 + } + return g.Floor() * g.margin +} + +// Reset drops the reference window, for when the audio path changes under the +// detector — a resumed session lands on new tracks and the old outbound RMS +// describes a link that no longer exists. +func (g *EchoGuard) Reset() { + if g == nil { + return + } + g.mu.Lock() + defer g.mu.Unlock() + g.start, g.count = 0, 0 +} + +// prune drops entries that have aged out. Caller holds the lock. +func (g *EchoGuard) prune(now time.Time) { + cutoff := now.Add(-g.window) + for g.count > 0 && g.buf[g.start].at.Before(cutoff) { + g.start = (g.start + 1) % len(g.buf) + g.count-- + } +} diff --git a/internal/vad/echo_test.go b/internal/vad/echo_test.go new file mode 100644 index 0000000..b037ff7 --- /dev/null +++ b/internal/vad/echo_test.go @@ -0,0 +1,163 @@ +package vad + +import ( + "testing" + "time" +) + +// fakeClock drives the guard's rolling window without sleeping. +type fakeClock struct{ t time.Time } + +func (c *fakeClock) now() time.Time { return c.t } +func (c *fakeClock) advance(d time.Duration) { c.t = c.t.Add(d) } + +func newTestGuard() (*EchoGuard, *fakeClock) { + c := &fakeClock{t: time.Unix(0, 0)} + g := NewEchoGuard(DefaultEchoWindow, DefaultEchoGain, DefaultEchoMargin) + g.now = c.now + return g, c +} + +// send feeds n 20ms outbound frames of the given amplitude, advancing the clock. +func send(g *EchoGuard, c *fakeClock, amplitude int16, frames int) { + for i := 0; i < frames; i++ { + g.Observe(frame(amplitude)) + c.advance(20 * time.Millisecond) + } +} + +func TestEchoGuardSilentAgentImposesNoFloor(t *testing.T) { + g, _ := newTestGuard() + if f := g.Floor(); f != 0 { + t.Errorf("Floor with nothing sent = %.0f, want 0", f) + } + if thr := g.Threshold(); thr != 0 { + t.Errorf("Threshold with nothing sent = %.0f, want 0", thr) + } +} + +func TestEchoGuardFloorTracksLoudestRecentFrame(t *testing.T) { + g, c := newTestGuard() + send(g, c, 1000, 3) + send(g, c, 5000, 1) + send(g, c, 1000, 3) + // Echo path delay is unknown, so the peak inside the window governs. + if got, want := g.Floor(), 5000*DefaultEchoGain; got != want { + t.Errorf("Floor = %.0f, want %.0f", got, want) + } +} + +func TestEchoGuardWindowExpires(t *testing.T) { + g, c := newTestGuard() + send(g, c, 5000, 5) + c.advance(DefaultEchoWindow) + if f := g.Floor(); f != 0 { + t.Errorf("Floor a full window after the agent stopped = %.0f, want 0", f) + } +} + +// A nil guard is the WebRTC path: every call must be inert. +func TestNilEchoGuardIsInert(t *testing.T) { + var g *EchoGuard + g.Observe(frame(5000)) + g.Reset() + if g.Floor() != 0 || g.Threshold() != 0 { + t.Error("nil guard reported a non-zero bound") + } +} + +// The bug from the issue: agent speech returning over a carrier at a fraction +// of its original level trips the 2-frame barge-in VAD. +func TestBargeInIgnoresEchoOfOwnVoice(t *testing.T) { + d := NewBargeIn() + g, c := newTestGuard() + d.SetEchoReference(g) + + feed(d, 60, 100) // learn a clean-line noise floor first + + // Agent talking at 6000 RMS; ~30% of it comes back, well above the 900 + // adaptive floor that would otherwise call it speech. + for i := 0; i < 50; i++ { + g.Observe(frame(6000)) + c.advance(20 * time.Millisecond) + if started, _ := d.Process(frame(1800)); started { + t.Fatalf("echo at frame %d read as a caller barge-in", i) + } + } +} + +// Double-talk over the same echo must still get through: the margin is what +// separates the two cases. +func TestBargeInStillFiresOnDoubleTalk(t *testing.T) { + d := NewBargeIn() + g, c := newTestGuard() + d.SetEchoReference(g) + + feed(d, 60, 100) + + var started bool + for i := 0; i < 10; i++ { + g.Observe(frame(6000)) + c.advance(20 * time.Millisecond) + // Caller over the top of the agent: echo (1800) plus their own voice. + s, _ := d.Process(frame(9000)) + started = started || s + } + if !started { + t.Error("caller talking over the agent failed to trigger barge-in") + } +} + +// While the duck is on, the wire carries ~12dB less, so the bound must fall +// with it. Observing post-duck samples is what makes that automatic. +func TestEchoGuardFloorFollowsDuck(t *testing.T) { + d := NewBargeIn() + g, c := newTestGuard() + d.SetEchoReference(g) + feed(d, 60, 100) + + full := frame(6000) + ducked := make([]int16, len(full)) + for i, v := range full { + ducked[i] = v / 4 + } + + // Ducked audio echoing back at the same ratio must not hold the caller out. + var started bool + for i := 0; i < 10; i++ { + g.Observe(ducked) + c.advance(20 * time.Millisecond) + s, _ := d.Process(frame(2500)) + started = started || s + } + if !started { + t.Error("caller inaudible under a ducked echo bound — the duck was not reflected in the floor") + } +} + +// Echo must not be learned as background noise. If it were, the adaptive +// threshold would ratchet up every time the agent spoke and stay there, +// leaving the agent deaf to a quiet caller once the echo bound lifted. +func TestEchoDoesNotPoisonNoiseFloor(t *testing.T) { + d := NewBargeIn() + g, c := newTestGuard() + d.SetEchoReference(g) + + feed(d, 60, 100) + floorBefore := d.noiseFloor + + for i := 0; i < 100; i++ { + g.Observe(frame(6000)) + c.advance(20 * time.Millisecond) + d.Process(frame(1800)) + } + if d.noiseFloor != floorBefore { + t.Errorf("noise floor moved during echo: %.1f → %.1f", floorBefore, d.noiseFloor) + } + + // And once the agent stops, the quiet caller is heard again. + c.advance(DefaultEchoWindow) + if started, _ := feed(d, 1000, 5); !started { + t.Error("quiet caller missed after the agent stopped speaking") + } +} diff --git a/internal/vad/vad.go b/internal/vad/vad.go index fd1129d..df2d375 100644 --- a/internal/vad/vad.go +++ b/internal/vad/vad.go @@ -22,6 +22,11 @@ type Detector struct { adaptive bool noiseFloor float64 // EMA of non-speech frame energy; negative = unset + + // echo, when set, raises the threshold by what the agent just sent, so + // the detector does not hear the agent's own voice returning as speech. + // Nil on paths that already run AEC. See EchoGuard. + echo *EchoGuard } // Adaptive-threshold tuning. The floor adapts over ~1s of silence frames @@ -74,10 +79,17 @@ func NewBargeIn() *Detector { return d } -// effectiveThreshold returns the current decision threshold: the fixed base -// until a noise floor has been learned, then the clamped multiple of the -// floor. -func (d *Detector) effectiveThreshold() float64 { +// SetEchoReference attaches an outbound-audio reference so the detector can +// tell the agent's own voice coming back from a caller talking over it. Only +// needed where nothing in the path runs AEC; passing nil disables the check. +func (d *Detector) SetEchoReference(g *EchoGuard) { + d.echo = g +} + +// noiseThreshold returns the decision threshold from the fixed base and the +// learned noise floor: the base until a floor has been learned, then the +// clamped multiple of the floor. +func (d *Detector) noiseThreshold() float64 { if !d.adaptive || d.noiseFloor < 0 { return d.threshold } @@ -91,6 +103,21 @@ func (d *Detector) effectiveThreshold() float64 { return thr } +// effectiveThreshold is the noise-floor threshold, raised to the echo bound +// whenever the agent's own audio is recent enough to still be arriving back. +// +// The echo bound is not subject to adaptiveMaxFactor: it tracks a signal the +// server generated rather than an estimate of the line, so a loud agent +// legitimately demands a loud caller, and the bound drops to zero on its own +// once the agent stops. +func (d *Detector) effectiveThreshold() float64 { + thr := d.noiseThreshold() + if echo := d.echo.Threshold(); echo > thr { + thr = echo + } + return thr +} + // updateNoiseFloor folds a non-speech frame's energy into the EMA. func (d *Detector) updateNoiseFloor(energy float64) { if !d.adaptive { @@ -106,7 +133,14 @@ func (d *Detector) updateNoiseFloor(energy float64) { // Process evaluates a PCM frame and returns whether speech just started or ended. func (d *Detector) Process(samples []int16) (started, ended bool) { energy := RMSEnergy(samples) - if energy > d.effectiveThreshold() { + echoThr := d.echo.Threshold() + + thr := d.noiseThreshold() + if echoThr > thr { + thr = echoThr + } + + if energy > thr { d.speechCount++ d.silentCount = 0 if !d.isSpeaking && d.speechCount >= d.speechFrames { @@ -116,7 +150,13 @@ func (d *Detector) Process(samples []int16) (started, ended bool) { } else { // Below-threshold frames are what the noise floor is made of — // learning only here keeps speech energy out of the floor estimate. - d.updateNoiseFloor(energy) + // Echo is not background noise: folding it in would ratchet the + // adaptive threshold up every time the agent spoke and leave the + // agent deaf to a quiet caller afterwards, which is the failure the + // echo bound exists to avoid. + if echoThr == 0 { + d.updateNoiseFloor(energy) + } d.silentCount++ d.speechCount = 0 if d.isSpeaking && d.silentCount >= d.silentFrames {