diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5e1328f..4da83e0 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -119,7 +119,8 @@ client + fixtures. Each processor is a `FrameProcessor` (`flowcat-core/src/processor/`). The framework owns the per-processor tokio task, the bounded/priority channels, and -the lifecycle; you write `process_frame` and optionally `start`/`stop`. +the lifecycle; you write `process_frame` and optionally +`start`/`on_interruption`/`stop`. **The one contract that surprises people — lifecycle/system frames bypass `process_frame`** ([`PROCESSOR-DESIGN.md`](PROCESSOR-DESIGN.md) §2.1–§2.3): @@ -129,14 +130,21 @@ the lifecycle; you write `process_frame` and optionally `start`/`stop`. does **not** reach `process_frame`. - `End` / `Stop` / `Cancel` → the framework calls your `async fn stop(&mut self, reason)` (flush + close), then forwards. Also **not** via `process_frame`. -- `Interruption` and other **System** frames ride an unbounded priority channel - and are drained ahead of data/control by a `biased` select; the task loop - handles interruption (draining interruptible queued frames, keeping - uninterruptible ones, cancelling an in-flight interruptible `process_frame`). - -So **`process_frame` only ever sees Data/Control frames.** Do not put socket -open/close in `process_frame` — it will never run for the lifecycle frames that -should trigger it. Other rules: +- `Interruption` (barge-in) → the task loop drains the queued interruptible + frames (keeping uninterruptible ones), calls your `async fn + on_interruption(&mut self)`, then forwards. Also **not** via `process_frame` — + a `Frame::Interruption` arm there is silently dead code. Note what this hook + does *not* buy you: an interruption cannot preempt a `process_frame` that is + already inside an `.await`, so anything needing sub-frame latency (cancelling + an in-flight LLM stream) has to poll an out-of-band flag — see + [`PROCESSOR-DESIGN.md`](PROCESSOR-DESIGN.md) §2.5. +- Other **System** frames ride an unbounded priority channel and are drained + ahead of data/control by a `biased` select. + +So **`process_frame` only ever sees Data/Control frames** (plus non-lifecycle +System frames such as `InputAudio`). Do not put socket open/close in +`process_frame` — it will never run for the lifecycle frames that should trigger +it. Other rules: - **`process_frame` must not block.** Long work (a provider round-trip) is driven by an internally-spawned task that feeds results back as frames — the Gemini diff --git a/PROCESSOR-DESIGN.md b/PROCESSOR-DESIGN.md index c0bd601..704cd59 100644 --- a/PROCESSOR-DESIGN.md +++ b/PROCESSOR-DESIGN.md @@ -330,7 +330,7 @@ impl Link { /// The building block. Each processor runs in **its own tokio task** fed by a /// bounded mpsc channel (§2.2). The framework owns the task loop; an impl only -/// writes `process_frame` (and optional `start`/`stop` hooks). +/// writes `process_frame` (and optional `start`/`on_interruption`/`stop` hooks). /// /// Mirrors pipecat `FrameProcessor` (frame_processor.py:175): `process_frame`, /// prev/next links, system-frame priority, interruption handling — but the per- @@ -354,6 +354,11 @@ pub trait FrameProcessor: Send + 'static { link.push(env.meta, env.frame, env.direction).await; Ok(()) } + /// Called on `Interruption` after the interruptible backlog is drained and + /// before the frame is forwarded — the delivery path for barge-in reactions + /// (§2.5). Default: no-op. **Must not block.** + async fn on_interruption(&mut self) -> Result<()> { Ok(()) } + /// Called on `End`/`Stop`/`Cancel` after the terminal frame is forwarded. /// Flush + close. Default: no-op. async fn stop(&mut self, _reason: StopReason) -> Result<()> { Ok(()) } @@ -390,7 +395,8 @@ async fn run_processor(mut p: Box, mut rx: ProcessorRx, link if let Some(o) = &setup.observer { o.on_process(&link.name, &env, setup.clock.now_ns()); } match &env.frame { Frame::Start(p0) => { p.start(&setup, p0).await.ok(); link.push(env.meta, env.frame, env.direction).await; } - Frame::Interruption => { /* §2.5: drain `normal` of interruptible frames, keep uninterruptible; forward */ } + Frame::Interruption => { /* §2.5: drain `normal` of interruptible frames, keep uninterruptible; + then p.on_interruption().await; then forward */ } Frame::Cancel{..} | Frame::End{..} | Frame::Stop => { let _ = p.stop(reason(&env.frame)).await; link.push(env.meta, env.frame, env.direction).await; if terminal { break } } _ => { if let Err(e) = p.process_frame(env, &link).await { link.push_error(e.to_string(), false).await; } } @@ -485,10 +491,26 @@ shared I/O cost, not framework cost. Barge-in (`Frame::Interruption`) is produced by the turn/VAD start strategy or any processor via `link.broadcast(Frame::Interruption)`. It travels the **system** channel both directions; each processor drains its normal queue of interruptible frames and -cancels in-flight interruptible work, then forwards. The transport-output processor -additionally clears the carrier's playback buffer (today's `transport.send_clear()`, -pipeline.rs:370 → a `process_frame` arm on `TransportOutput`). This is the literal port -of pipecat `broadcast_interruption` (frame_processor.py:704). +cancels in-flight interruptible work, then forwards. This is the literal port of +pipecat `broadcast_interruption` (frame_processor.py:704). + +`Interruption` is a **lifecycle frame**: like `Start`/`End` it is intercepted by the +task loop and never reaches `process_frame`. A processor reacts to barge-in by +overriding **`on_interruption`** — the transport-output sinks clear the carrier's +playback buffer there (`transport.send_clear()`), the text aggregators/filters reset +their buffers, and the assistant context aggregator keeps the partially-spoken reply +and drops the open response span. Writing a `Frame::Interruption` arm in +`process_frame` instead is silently dead code. + +**What the frame path cannot do.** The interruption jumps queues, but it cannot +preempt a `process_frame` that is already inside an `.await` (a TTS synthesis, an LLM +stream): the hop only sees it when that call returns. Anything needing sub-frame +latency — cancelling an in-flight completion, flushing the carrier *now* — needs an +out-of-band signal (a shared atomic polled between streamed chunks, a `Notify`-woken +reactor task) raised at detection time, alongside the broadcast. `VadProcessor`'s +`with_interrupt_flag`/`with_interrupt_notify` are those seams, and +`build_cascaded_call_duplex` wires them; both are opt-in, and the frame path remains +the correctness path. --- @@ -721,7 +743,7 @@ Today's five trait seams (`MediaTransport`, `RealtimeLlm`, `AgentBrain`, `Sessio | Today (seam / inline logic) | becomes | crate | notes | |---|---|---|---| | `MediaTransport::recv` (media.rs:49) | **`TransportInput` processor** — a *source*: reads the transport, emits `Frame::InputAudio`/`UserStartedSpeaking`/lifecycle downstream | `flowcat-transports` (trait stays in core) | pipecat `BaseInputTransport` | -| `MediaTransport::send_audio`/`send_clear` (media.rs:53/57) | **`TransportOutput` processor** — a *sink*: consumes `OutputAudio`/`TtsAudio`, plays to carrier; on `Interruption` clears playback (was pipeline.rs:370) | `flowcat-transports` | pipecat `BaseOutputTransport`; emits `BotStarted/StoppedSpeaking` | +| `MediaTransport::send_audio`/`send_clear` (media.rs:53/57) | **`TransportOutput` processor** — a *sink*: consumes `OutputAudio`/`TtsAudio`, plays to carrier; clears playback from `on_interruption` (was pipeline.rs:370) | `flowcat-transports` | pipecat `BaseOutputTransport`; emits `BotStarted/StoppedSpeaking` | | `RealtimeLlm` (realtime/mod.rs:22) | **`RealtimeLlmService` processor** — consumes `InputAudio`, emits `TtsAudio`(bot)/`Transcription`/`FunctionCallsStarted`/`Interruption`/`Metrics`; the reader-task→mpsc bridge (gemini_live.rs:265) becomes the processor's internal task feeding `link` | `flowcat-services` (`realtime-gemini` feature) | the trait below; Gemini is one impl | | `AgentBrain` (brain.rs:22) | **`BrainProcessor`** — consumes `FunctionCallsStarted`/tool-call frames, emits `UpdateSettings`(new prompt+tools) on transition / `End` on terminal; holds the graph state | the embedder (its glue) | pipecat has no peer; this is the embedder's engine adapter | | `SessionSource` (session.rs:21) | **stays embedder glue, NOT a processor** — a service the `BrainProcessor` + a `FinalizeProcessor` *call*; bootstrap/finalize/artifact-upload is control-plane I/O, not a media frame stage | the embedder | see §6.2 — it leaves flowcat-core for OSS cleanliness | diff --git a/flowcat-core/src/agent/mod.rs b/flowcat-core/src/agent/mod.rs index 6ca48c4..ecc9641 100644 --- a/flowcat-core/src/agent/mod.rs +++ b/flowcat-core/src/agent/mod.rs @@ -42,6 +42,17 @@ pub(crate) mod test_harness { mut proc: Box, frames: Vec, direction: Direction, + ) -> Vec { + drive_ref(proc.as_mut(), frames, direction).await + } + + /// [`drive`] against a borrowed processor, so a test can drive it, poke a + /// lifecycle hook (`on_interruption`), and drive it again — asserting on state + /// carried *across* turns. + pub(crate) async fn drive_ref( + proc: &mut dyn FrameProcessor, + frames: Vec, + direction: Direction, ) -> Vec { // One capture channel acts as both the downstream and upstream neighbour. let (cap_tx, mut cap_rx) = channel(Arc::from("capture"), NORMAL_CHAN_CAP); diff --git a/flowcat-core/src/agent/text_filter.rs b/flowcat-core/src/agent/text_filter.rs index 0ed7521..412114d 100644 --- a/flowcat-core/src/agent/text_filter.rs +++ b/flowcat-core/src/agent/text_filter.rs @@ -527,6 +527,12 @@ impl FrameProcessor for TextAggregatorProcessor self.name } + /// Barge-in: drop the half-built sentence so the next turn doesn't inherit it. + async fn on_interruption(&mut self) -> Result<()> { + self.agg.reset(); + Ok(()) + } + async fn process_frame(&mut self, env: Envelope, link: &Link) -> Result<()> { match &env.frame { Frame::LlmText(t) | Frame::Text(t) => { @@ -550,10 +556,6 @@ impl FrameProcessor for TextAggregatorProcessor } link.push(env.meta, env.frame, env.direction).await; } - Frame::Interruption => { - self.agg.reset(); - link.push(env.meta, env.frame, env.direction).await; - } _ => { link.push(env.meta, env.frame, env.direction).await; } @@ -583,6 +585,12 @@ impl FrameProcessor for TextFilterProcessor { self.name } + /// Barge-in: reset the filter's carry-over state for the next turn. + async fn on_interruption(&mut self) -> Result<()> { + self.filter.reset(); + Ok(()) + } + async fn process_frame(&mut self, env: Envelope, link: &Link) -> Result<()> { match env.frame { Frame::LlmText(t) => { @@ -599,11 +607,6 @@ impl FrameProcessor for TextFilterProcessor { .await; } } - Frame::Interruption => { - self.filter.reset(); - link.push(env.meta, Frame::Interruption, env.direction) - .await; - } other => { link.push(env.meta, other, env.direction).await; } @@ -615,7 +618,7 @@ impl FrameProcessor for TextFilterProcessor { #[cfg(test)] mod tests { use super::*; - use crate::agent::test_harness::drive; + use crate::agent::test_harness::{drive, drive_ref}; use crate::processor::frame::Direction; #[test] @@ -724,6 +727,66 @@ mod tests { assert_eq!(texts, vec!["Hi there.", "Bye now."]); } + /// Barge-in resets the half-built sentence. `Frame::Interruption` is a + /// lifecycle frame the runtime intercepts, so it never reaches + /// `process_frame` — the reset has to hang off `on_interruption`, or the next + /// turn inherits the abandoned fragment. + #[tokio::test] + async fn aggregator_processor_resets_the_buffer_on_interruption() { + let mut proc = TextAggregatorProcessor::new("agg", SimpleTextAggregator::new()); + // A fragment with no sentence end stays buffered. + let out = drive_ref( + &mut proc, + vec![Frame::LlmText("half a sen".into())], + Direction::Downstream, + ) + .await; + assert!( + out.is_empty(), + "an unterminated fragment must not be emitted" + ); + + proc.on_interruption().await.expect("hook ok"); + + // After the reset the next turn starts clean — the abandoned fragment is + // not prefixed onto it. + let out = drive_ref( + &mut proc, + vec![Frame::LlmText("Brand new sentence. Next".into())], + Direction::Downstream, + ) + .await; + assert!( + matches!(out.first(), Some(Frame::Text(t)) if t == "Brand new sentence."), + "got {out:?}" + ); + } + + /// Same for the filter processor: the hook is the only delivery path for a + /// barge-in reset. + #[tokio::test] + async fn filter_processor_resets_on_interruption() { + let mut proc = TextFilterProcessor::new("md", MarkdownTextFilter::default()); + // An unbalanced fence leaves the filter in "inside a code block" state. + let _ = drive_ref( + &mut proc, + vec![Frame::LlmText("```rust\nlet x = 1;".into())], + Direction::Downstream, + ) + .await; + proc.on_interruption().await.expect("hook ok"); + let out = drive_ref( + &mut proc, + vec![Frame::LlmText("**hello**".into())], + Direction::Downstream, + ) + .await; + assert!( + matches!(out.first(), Some(Frame::LlmText(t)) if t == "hello"), + "a reset filter must treat the next turn as fresh prose; got {out:?}" + ); + } + #[tokio::test] async fn filter_processor_cleans_text_frames() { let proc = TextFilterProcessor::new("md", MarkdownTextFilter::default()); diff --git a/flowcat-core/src/audio/vad.rs b/flowcat-core/src/audio/vad.rs index d120490..7657635 100644 --- a/flowcat-core/src/audio/vad.rs +++ b/flowcat-core/src/audio/vad.rs @@ -338,6 +338,18 @@ pub struct VadProcessor { bot_speaking: bool, /// If true, emit `Interruption` on a rising edge while the bot speaks. interrupt_on_barge_in: bool, + /// Optional barge-in generation counter, bumped synchronously at detection + /// time (before the `Interruption` broadcast). Busy processors (an LLM + /// adapter mid-stream) poll it between chunks for cooperative cancellation — + /// the frame path cannot preempt an in-flight `process_frame`. + interrupt_flag: Option>, + /// Optional out-of-band waker fired at barge-in detection. The frame-path + /// `Interruption` can stall behind any hop that is mid-`await` (observed + /// live: 14 ms vs 2.1 s sink delivery depending on TTS activity); a + /// dedicated reactor task woken here can flush playback immediately. + /// Signalled with `notify_one`, so a barge-in raised while the reactor is + /// still handling the previous one is stored rather than lost. + interrupt_notify: Option>, } impl VadProcessor { @@ -353,9 +365,26 @@ impl VadProcessor { last_state: VadState::Quiet, bot_speaking: false, interrupt_on_barge_in: true, + interrupt_flag: None, + interrupt_notify: None, } } + /// Fire `notify` (out-of-band) on each barge-in detection. + pub fn with_interrupt_notify(mut self, notify: std::sync::Arc) -> Self { + self.interrupt_notify = Some(notify); + self + } + + /// Bump `flag` on each barge-in detection (see the field docs). + pub fn with_interrupt_flag( + mut self, + flag: std::sync::Arc, + ) -> Self { + self.interrupt_flag = Some(flag); + self + } + /// Disable broadcasting `Interruption` on barge-in (the turn controller may own /// interruption via an [`InterruptionStrategy`](crate::audio::strategy::InterruptionStrategy)). pub fn without_barge_in_interruption(mut self) -> Self { @@ -380,6 +409,13 @@ impl VadProcessor { .await; link.push_down(Frame::UserStartedSpeaking).await; if self.bot_speaking && self.interrupt_on_barge_in { + tracing::debug!("barge-in: user speech while bot speaking"); + if let Some(flag) = &self.interrupt_flag { + flag.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + if let Some(n) = &self.interrupt_notify { + n.notify_one(); + } link.broadcast(Frame::Interruption).await; } } diff --git a/flowcat-core/src/pipeline/cascaded.rs b/flowcat-core/src/pipeline/cascaded.rs index fe0b4c7..ad6ba00 100644 --- a/flowcat-core/src/pipeline/cascaded.rs +++ b/flowcat-core/src/pipeline/cascaded.rs @@ -274,6 +274,251 @@ impl TurnMute { } } +// =========================================================================== +// BotSpeakingNotifier — duplex-mode bot-speaking edges for VAD barge-in. +// =========================================================================== + +/// Emits [`Frame::BotStartedSpeaking`]/[`Frame::BotStoppedSpeaking`] into the +/// pipeline head as the sink plays a reply, so an upstream +/// [`VadProcessor`](crate::audio::VadProcessor) can arm its barge-in gate. The +/// stock cascaded chain emits neither frame (the s2s realtime model does its own +/// barge-in), which leaves VAD barge-in permanently disarmed — this closes that +/// gap for the duplex builder. +#[derive(Clone)] +struct BotSpeakingNotifier(Arc); + +struct BotSpeakingInner { + speaking: std::sync::atomic::AtomicBool, + /// Bumped per playout watchdog so a superseded watcher exits silently. + generation: std::sync::atomic::AtomicU64, + bot_until: Mutex>, + head_tx: tokio::sync::mpsc::UnboundedSender, + carrier_rate: u32, +} + +impl BotSpeakingNotifier { + fn new(head_tx: tokio::sync::mpsc::UnboundedSender, carrier_rate: u32) -> Self { + Self(Arc::new(BotSpeakingInner { + speaking: std::sync::atomic::AtomicBool::new(false), + generation: std::sync::atomic::AtomicU64::new(0), + bot_until: Mutex::new(None), + head_tx, + carrier_rate, + })) + } + + /// Note `samples` of carrier-rate bot audio sent; on the first chunk of a + /// burst emit `BotStartedSpeaking` and spawn a playout watchdog that emits + /// `BotStoppedSpeaking` once the estimate runs dry. + fn note_audio(&self, samples: usize) { + use std::sync::atomic::Ordering; + { + let mut until = self.0.bot_until.lock().unwrap(); + *until = Some(advance_playout( + *until, + Instant::now(), + samples, + self.0.carrier_rate, + )); + } + if !self.0.speaking.swap(true, Ordering::SeqCst) { + let _ = self.0.head_tx.send(Frame::BotStartedSpeaking); + let my_gen = self.0.generation.fetch_add(1, Ordering::SeqCst) + 1; + let inner = self.0.clone(); + tokio::spawn(async move { + loop { + if inner.generation.load(Ordering::SeqCst) != my_gen { + return; // superseded (new burst or interruption) + } + let until = *inner.bot_until.lock().unwrap(); + let now = Instant::now(); + match until { + Some(t) if t > now => { + tokio::time::sleep((t - now).min(std::time::Duration::from_millis(200))) + .await + } + _ => break, + } + } + if inner.generation.load(Ordering::SeqCst) == my_gen + && inner.speaking.swap(false, Ordering::SeqCst) + { + let _ = inner.head_tx.send(Frame::BotStoppedSpeaking); + } + }); + } + } + + /// Barge-in: kill the watchdog, clear the estimate, emit the stopped edge. + fn on_interruption(&self) { + use std::sync::atomic::Ordering; + self.0.generation.fetch_add(1, Ordering::SeqCst); + *self.0.bot_until.lock().unwrap() = None; + if self.0.speaking.swap(false, Ordering::SeqCst) { + let _ = self.0.head_tx.send(Frame::BotStoppedSpeaking); + } + } +} + +// =========================================================================== +// StaleAudioLatch — drops the interrupted reply's in-flight TTS audio. +// =========================================================================== + +/// Guards the window between the out-of-band reactor flushing the carrier and +/// the frame-path `Interruption` reaching the sink. TTS audio synthesized for +/// the interrupted reply can still land inside it, and must be dropped rather +/// than played over the caller who just barged in. +/// +/// Both ends stamp the **barge-in generation** they observed (the counter the +/// VAD bumps at detection) instead of flipping a flag: the two tasks are woken +/// independently and either can be scheduled first, and a bool armed by the +/// reactor *after* the sink already cleared it would suppress bot audio for the +/// rest of the call. Stamping makes the latch order-independent — it is only +/// stale while a generation was armed that the sink has not yet acknowledged. +struct StaleAudioLatch { + /// Shared with [`VadProcessor::with_interrupt_flag`]; bumped at detection. + generation: Arc, + armed: std::sync::atomic::AtomicU64, + cleared: std::sync::atomic::AtomicU64, +} + +impl StaleAudioLatch { + fn new(generation: Arc) -> Self { + Self { + generation, + armed: std::sync::atomic::AtomicU64::new(0), + cleared: std::sync::atomic::AtomicU64::new(0), + } + } + + fn current(&self) -> u64 { + self.generation.load(std::sync::atomic::Ordering::SeqCst) + } + + /// The reactor flushed the carrier for this barge-in — drop what follows + /// until the sink acknowledges it. `fetch_max`, so a late arm from a + /// superseded barge-in can never walk the latch backwards. + fn arm(&self) { + self.armed + .fetch_max(self.current(), std::sync::atomic::Ordering::SeqCst); + } + + /// The frame-path `Interruption` reached the sink: everything arriving from + /// now on belongs to the *next* reply. + fn clear(&self) { + self.cleared + .fetch_max(self.current(), std::sync::atomic::Ordering::SeqCst); + } + + fn is_stale(&self) -> bool { + self.armed.load(std::sync::atomic::Ordering::SeqCst) + > self.cleared.load(std::sync::atomic::Ordering::SeqCst) + } +} + +// =========================================================================== +// SpeechGate — VAD-edged speech segmentation for duplex STT. +// =========================================================================== + +/// Pre-roll retained ahead of a VAD rising edge, so the first phoneme of the +/// utterance isn't clipped by the detector's own attack time. +const SPEECH_GATE_PREROLL_MS: usize = 300; + +/// Sits between the VAD and STT in the duplex chain. Fixed-window batch STT +/// (whisper.cpp) has no endpointing: fed raw duplex audio it transcribes silence +/// (hallucinated turns) and splits utterances at arbitrary window boundaries. +/// The gate forwards `InputAudio` only between a VAD rising edge (plus +/// [`SPEECH_GATE_PREROLL_MS`] of pre-roll) and the falling edge; the +/// `UserStoppedSpeaking` it forwards is what makes [`SttProcessor`] call +/// [`SttService::flush`](crate::service::SttService::flush), so the STT finalizes +/// exactly one utterance per VAD-detected turn. +/// +/// Recording caveat: the gate is upstream of the [`RecorderProcessor`] inbound +/// tap, so the recording's caller leg contains the gated speech only (silence +/// between turns is silence-padded at render time by the wall-clock stamps, but +/// the re-injected pre-roll lands ~[`SPEECH_GATE_PREROLL_MS`] late within its +/// utterance). Transcription quality is the trade this builder makes. +struct SpeechGate { + open: bool, + preroll: std::collections::VecDeque, + preroll_cap: usize, + sample_rate: u32, +} + +impl SpeechGate { + fn new() -> Self { + Self { + open: false, + preroll: std::collections::VecDeque::new(), + // Rescaled to the negotiated input rate in `start()`. + preroll_cap: 16_000 * SPEECH_GATE_PREROLL_MS / 1000, + sample_rate: 16_000, + } + } +} + +#[async_trait] +impl FrameProcessor for SpeechGate { + fn name(&self) -> &str { + "SpeechGate" + } + + async fn start(&mut self, _s: &ProcessorSetup, p: &StartParams) -> Result<()> { + self.sample_rate = p.audio_in_sample_rate; + self.preroll_cap = self.sample_rate as usize * SPEECH_GATE_PREROLL_MS / 1000; + Ok(()) + } + + /// Barge-in: drop the pre-roll ring. On the VAD's own barge-in this is a + /// no-op (its `UserStartedSpeaking` leads the broadcast, so the ring was + /// already drained into the turn); it matters when the interruption comes + /// from elsewhere — an [`InterruptionStrategy`](crate::audio::strategy) — + /// where the ring holds bot echo rather than the caller's next utterance. + async fn on_interruption(&mut self) -> Result<()> { + self.preroll.clear(); + Ok(()) + } + + async fn process_frame(&mut self, env: Envelope, link: &Link) -> Result<()> { + match &env.frame { + Frame::InputAudio(audio) => { + if self.open { + link.push(env.meta, env.frame, env.direction).await; + } else { + // Swallow the audio, keeping only the trailing pre-roll — + // feeding a batch STT the inter-turn silence is what makes it + // hallucinate turns. + for s in &audio.pcm { + if self.preroll.len() == self.preroll_cap { + self.preroll.pop_front(); + } + self.preroll.push_back(*s); + } + } + } + Frame::UserStartedSpeaking => { + self.open = true; + let pre: Vec = self.preroll.drain(..).collect(); + if !pre.is_empty() { + link.push_down(Frame::InputAudio(Arc::new( + crate::processor::frame::AudioFrame::mono(pre, self.sample_rate), + ))) + .await; + } + link.push(env.meta, env.frame, env.direction).await; + } + // Close the gate and forward the edge: `SttProcessor` turns it into a + // `SttService::flush()`, which is what finalizes the utterance. + Frame::UserStoppedSpeaking => { + self.open = false; + link.push(env.meta, env.frame, env.direction).await; + } + _ => link.push(env.meta, env.frame, env.direction).await, + } + Ok(()) + } +} + // =========================================================================== // Context summarizer hook (pipecat `LLMContextSummarizer`). // =========================================================================== @@ -539,6 +784,29 @@ impl FrameProcessor for AssistantContextAggregator { "AssistantContextAggregator" } + /// Barge-in context repair (duplex): commit what streamed so far as the + /// assistant turn — the best available approximation of "what was spoken" for + /// a one-shot TTS with no word timestamps, and better than losing the turn + /// entirely (the model would re-offer what it already said). + /// + /// Taking the buffer is also what stops the interrupted reply being spoken + /// *after* the barge-in: a late `LlmResponseEnd` from the cancelled stream now + /// finds an empty reply and emits no `TtsSpeak`. (Usually it never arrives — + /// `LlmResponseEnd` is interruptible, so the runtime's drain drops it — but + /// one racing in behind the interruption must be inert too.) + async fn on_interruption(&mut self) -> Result<()> { + self.in_response = false; + let partial = std::mem::take(&mut self.buffer); + if !partial.is_empty() { + if let Some(st) = &self.transcript_state { + st.lock().unwrap().transcript.push_bot(&partial); + } + let mut c = self.ctx.lock().unwrap(); + c.push("assistant", &partial); + } + Ok(()) + } + async fn process_frame(&mut self, env: Envelope, link: &Link) -> Result<()> { match &env.frame { Frame::LlmResponseStart => { @@ -908,6 +1176,11 @@ struct CascadedTransportOutput { end_tx: tokio::sync::mpsc::UnboundedSender, /// Turn lock: extend its playout estimate per chunk (drives unmute + the End-drain). turn_mute: TurnMute, + /// Duplex mode: bot-speaking edge notifier for upstream VAD barge-in. + bot_notifier: Option, + /// Duplex mode: drops bot audio belonging to a reply that outran its own + /// interruption (see [`StaleAudioLatch`]). + suppress_stale: Option>, } impl CascadedTransportOutput { @@ -927,9 +1200,23 @@ impl CascadedTransportOutput { state, end_tx, turn_mute, + bot_notifier: None, + suppress_stale: None, } } + /// Duplex mode: emit bot-speaking edges for the upstream VAD (builder-set). + fn with_bot_notifier(mut self, n: BotSpeakingNotifier) -> Self { + self.bot_notifier = Some(n); + self + } + + /// Duplex mode: stale-audio suppression latch shared with the reactor. + fn with_suppress_latch(mut self, latch: Arc) -> Self { + self.suppress_stale = Some(latch); + self + } + /// Terminal transport error (peer gone): record it (de-duped to one log) and end /// the call **once**. The `OutputAudio` guard then drops remaining buffered frames /// instead of re-failing on the dead transport. @@ -952,6 +1239,22 @@ impl FrameProcessor for CascadedTransportOutput "CascadedTransportOutput" } + /// Barge-in: flush the carrier playback and drop the playout estimate. + async fn on_interruption(&mut self) -> Result<()> { + tracing::debug!("barge-in: flushing carrier playback"); + if let Some(latch) = &self.suppress_stale { + latch.clear(); + } + if let Err(e) = self.transport.send_clear().await { + self.fail_transport(e); + } + let _ = self.turn_mute.take_bot_until(); + if let Some(n) = &self.bot_notifier { + n.on_interruption(); + } + Ok(()) + } + async fn start(&mut self, _setup: &ProcessorSetup, _params: &StartParams) -> Result<()> { self.out_resampler = Some(Resampler::new(self.tts_rate, self.carrier_rate)?); Ok(()) @@ -967,6 +1270,14 @@ impl FrameProcessor for CascadedTransportOutput if self.state.lock().unwrap().transport_dead { return Ok(()); } + // Barge-in raced a mid-synthesis TTS: this audio belongs to the + // interrupted reply (the reactor flushed before the frame-path + // Interruption arrived here). Drop it. + if let Some(latch) = &self.suppress_stale { + if latch.is_stale() { + return Ok(()); + } + } let chunk = AudioChunk::from(audio.as_ref()); self.state.lock().unwrap().recorder.push_outbound(&chunk); let resampler = self @@ -982,20 +1293,15 @@ impl FrameProcessor for CascadedTransportOutput } else { // Keep STT muted until this reply finishes playing. self.turn_mute.note_bot_audio(samples, self.carrier_rate); + if let Some(n) = &self.bot_notifier { + n.note_audio(samples); + } } } } Err(e) => self.state.lock().unwrap().record_error(e), } } - // Barge-in: flush the carrier playback. - Frame::Interruption => { - if let Err(e) = self.transport.send_clear().await { - self.fail_transport(e); - } - let _ = self.turn_mute.take_bot_until(); - link.push(env.meta, env.frame, env.direction).await; - } // End-of-call: wait out the final bot utterance still playing at the // carrier before teardown (mirrors the realtime sink — see s2s.rs). Frame::End { .. } => { @@ -1028,6 +1334,8 @@ pub struct CascadedTask { pub task: PipelineTask, /// The transport-pump source reader (aborted on drop / after `run`). pump: SourcePump, + /// Duplex mode: the out-of-band interrupt reactor (aborted after `run`). + reactor: Option>, } impl CascadedTask { @@ -1035,6 +1343,9 @@ impl CascadedTask { pub async fn run(self) -> Result<()> { let res = self.task.run().await; self.pump.abort(); + if let Some(r) = self.reactor { + r.abort(); + } res } } @@ -1297,7 +1608,184 @@ where // path — emits ClientConnected / InputAudio / End at the head). let pump = spawn_transport_pump(shared, task.queue_sender()); - Ok(CascadedTask { task, pump }) + Ok(CascadedTask { + task, + pump, + reactor: None, + }) +} + +/// Full-duplex variant of [`build_cascaded_call_with_observers`]: inserts a +/// [`VadProcessor`](crate::audio::VadProcessor) ahead of STT, arms its barge-in +/// gate via a [`BotSpeakingNotifier`] in the sink, does **not** engage the +/// half-duplex [`TurnMute`] (STT stays open while the bot speaks), and wires a +/// shared barge-in generation counter through the VAD and the +/// [`LlmProcessor`](crate::service::adapters::LlmProcessor) so an in-flight +/// completion cancels cooperatively. +/// +/// Caller provides the VAD analyzer (e.g. +/// [`SileroVad`](crate::audio::SileroVad) behind `vad-ort`) so this builder +/// stays feature-agnostic. Echo discipline is the caller's problem: the client +/// must not loop bot audio back into its mic (hardware/browser AEC, or +/// separated fixtures in tests). +#[allow(clippy::too_many_arguments)] +pub async fn build_cascaded_call_duplex( + transport: Tr, + vad: crate::audio::VadProcessor, + stt: St, + llm: L, + tts: Ts, + brain: B, + session: Se, + run_id: i64, + token: String, + config: CascadedConfig, + observers: Vec>, +) -> Result +where + Tr: MediaTransport + 'static, + St: SttService + 'static, + L: LlmService + 'static, + Ts: TtsService + 'static, + B: AgentBrain + 'static, + Se: SessionSource + 'static, + V: crate::service::VadAnalyzer + 'static, +{ + use crate::service::adapters::LlmProcessor as LlmProc; + + let carrier_rate = transport.carrier_rate(); + let tts_rate = tts.sample_rate(); + let session = Arc::new(session); + let relay = Arc::new(SessionToolRelay::new( + session.clone(), + run_id, + token.clone(), + )); + + let node_id = brain.current_node_id(); + let mcp = relay.node_tools(&node_id).await; + let mcp_names: std::collections::HashSet = mcp.iter().map(|t| t.name.clone()).collect(); + let mut initial_tools = brain.tools(); + initial_tools.extend(mcp); + let initial_tools_json: Vec = initial_tools + .iter() + .map(|t| serde_json::to_value(t).unwrap_or(Value::Null)) + .collect(); + + let ctx: SharedContext = Arc::new(Mutex::new(RollingContext::new( + Some(brain.system_prompt()), + initial_tools_json, + ))); + let state: SharedState = Arc::new(Mutex::new(LiveState::new(carrier_rate))); + let shared = SharedTransport::new(transport); + let (end_tx, mut end_rx) = tokio::sync::mpsc::unbounded_channel::(); + + // TurnMute is retained ONLY for the sink's End-of-call playout drain; it is + // never `begin()`-ed (no aggregator/kickoff wiring), so STT never mutes. + let turn_mute = TurnMute::new(end_tx.clone(), std::time::Duration::from_secs(12)); + + // The shared barge-in generation counter: VAD bumps it at detection time, + // the LLM adapter polls it between streamed chunks. + let interrupt_flag = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let interrupt_notify = Arc::new(tokio::sync::Notify::new()); + let suppress_latch = Arc::new(StaleAudioLatch::new(interrupt_flag.clone())); + let vad = vad + .with_interrupt_flag(interrupt_flag.clone()) + .with_interrupt_notify(interrupt_notify.clone()); + let bot_notifier = BotSpeakingNotifier::new(end_tx.clone(), carrier_rate); + + // Out-of-band interrupt reactor: the frame-path Interruption can stall + // behind a mid-await hop (a TTS synthesis), delaying the audible stop by + // seconds. This task is woken synchronously at VAD detection, flushes the + // carrier immediately, and arms the stale-audio latch (cleared when the + // frame-path Interruption reaches the sink). + let reactor = { + let mut transport = shared.clone(); + let notifier = bot_notifier.clone(); + let latch = suppress_latch.clone(); + let notify = interrupt_notify.clone(); + tokio::spawn(async move { + loop { + notify.notified().await; + tracing::debug!("barge-in reactor: immediate carrier flush"); + latch.arm(); + if let Err(e) = transport.send_clear().await { + tracing::debug!(error = %e, "reactor send_clear failed"); + } + notifier.on_interruption(); + } + }) + }; + + let summarizer = config + .summarizer + .unwrap_or_else(|| Arc::new(NoopSummarizer) as Arc); + + let processors: Vec> = vec![ + Box::new(TransportInput::new()), + Box::new(vad), + Box::new(SpeechGate::new()), + Box::new(SttProcessor::new(stt)), + Box::new( + UserContextAggregator::new(ctx.clone(), summarizer, config.summarizer_cfg) + .with_transcript_state(state.clone()), + ), + Box::new(CascadedKickoffProcessor::new(ctx.clone())), + Box::new(LlmProc::new(llm).with_interrupt_flag(interrupt_flag.clone())), + Box::new(CascadedToolBridge::new(ctx.clone())), + Box::new(BrainProcessor::new( + brain, + relay, + mcp_names, + state.clone(), + end_tx.clone(), + )), + Box::new(AssistantContextAggregator::new(ctx.clone()).with_transcript_state(state.clone())), + Box::new(TtsProcessor::new(tts)), + Box::new( + CascadedTransportOutput::new( + shared.clone(), + tts_rate, + carrier_rate, + state.clone(), + end_tx.clone(), + turn_mute.clone(), + ) + .with_bot_notifier(bot_notifier.clone()) + .with_suppress_latch(suppress_latch.clone()), + ), + Box::new(RecorderProcessor::new(state.clone())), + Box::new(TranscriptProcessor::new(state.clone())), + Box::new(FinalizeProcessor::new( + session, + run_id, + token, + state.clone(), + )), + ]; + let pipeline = Pipeline::new(processors); + + let params = PipelineTaskParams { + idle_timeout: None, + ..config.task_params + }; + let task = PipelineTask::new(pipeline, params, observers); + + let head = task.queue_sender(); + tokio::spawn(async move { + while let Some(f) = end_rx.recv().await { + if head.send(f).is_err() { + break; + } + } + }); + let pump = spawn_transport_pump(shared, task.queue_sender()); + + Ok(CascadedTask { + task, + pump, + reactor: Some(reactor), + }) } // =========================================================================== @@ -2092,6 +2580,205 @@ mod tests { .contains("cardiology")); } + // ---------------------------------------------------------------------- + // Duplex (full-duplex barge-in) pieces. + // ---------------------------------------------------------------------- + + /// Records the `InputAudio` chunk sizes that make it past the gate. + #[derive(Clone, Default)] + struct AudioCapture(Arc>>); + #[async_trait] + impl FrameProcessor for AudioCapture { + fn name(&self) -> &str { + "AudioCapture" + } + async fn process_frame(&mut self, env: Envelope, link: &Link) -> Result<()> { + if let Frame::InputAudio(a) = &env.frame { + self.0.lock().unwrap().push(a.pcm.len()); + } + link.push(env.meta, env.frame, env.direction).await; + Ok(()) + } + } + + async fn run_gate(frames: Vec) -> Vec { + let cap = AudioCapture::default(); + let task = PipelineTask::new( + Pipeline::new(vec![Box::new(SpeechGate::new()), Box::new(cap.clone())]), + PipelineTaskParams::default(), + vec![], + ); + for f in frames { + task.queue_frame(f).await; + } + task.stop_when_done().await; + tokio::time::timeout(Duration::from_secs(5), task.run()) + .await + .expect("speech gate pipeline timed out") + .expect("run ok"); + let out = cap.0.lock().unwrap().clone(); + out + } + + fn audio(n: usize) -> Frame { + Frame::InputAudio(Arc::new(AudioFrame::mono(vec![1i16; n], 16_000))) + } + + /// The gate passes audio only between the VAD edges, and replays the buffered + /// pre-roll at the rising edge so the first phoneme isn't clipped. Everything + /// outside a turn is swallowed — feeding inter-turn silence to a fixed-window + /// batch STT is what makes it hallucinate turns. + #[tokio::test] + async fn speech_gate_passes_only_vad_delimited_audio_with_preroll() { + let out = run_gate(vec![ + audio(100), // before the turn → pre-roll only + Frame::UserStartedSpeaking, + audio(50), // inside the turn → through + Frame::UserStoppedSpeaking, + audio(70), // after the turn → swallowed + ]) + .await; + assert_eq!( + out, + vec![100, 50], + "expected the replayed pre-roll then the in-turn audio" + ); + } + + /// The pre-roll is a bounded ring: only the last `SPEECH_GATE_PREROLL_MS` of + /// pre-speech audio is replayed, however long the silence before the turn was. + #[tokio::test] + async fn speech_gate_preroll_is_bounded_to_the_configured_window() { + let cap = 16_000 * SPEECH_GATE_PREROLL_MS / 1000; + let out = run_gate(vec![ + audio(cap * 3), + Frame::UserStartedSpeaking, + Frame::UserStoppedSpeaking, + ]) + .await; + assert_eq!(out, vec![cap], "pre-roll must be capped at the window"); + } + + /// The gate emits no marker audio of its own at the falling edge — finalizing + /// the utterance is `SttService::flush`'s job, driven by the forwarded + /// `UserStoppedSpeaking`. Guards against re-introducing a magic silent chunk + /// that only an STT in on the convention could recognize. + #[tokio::test] + async fn speech_gate_emits_no_synthetic_flush_audio() { + let out = run_gate(vec![ + Frame::UserStartedSpeaking, + audio(32), + Frame::UserStoppedSpeaking, + ]) + .await; + assert_eq!(out, vec![32]); + } + + /// The bot-speaking edges the cascaded chain never emitted: without them + /// `VadProcessor` keeps `bot_speaking == false` and its barge-in broadcast is + /// permanently disarmed. + #[tokio::test] + async fn bot_speaking_notifier_emits_started_then_stopped_edges() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let n = BotSpeakingNotifier::new(tx, 16_000); + + n.note_audio(160); // 10 ms of carrier audio + assert!(matches!(rx.recv().await, Some(Frame::BotStartedSpeaking))); + + // A second chunk inside the same burst must not re-announce the start. + n.note_audio(160); + + // The watchdog emits the stopped edge once the playout estimate runs dry. + let stopped = tokio::time::timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("watchdog never emitted BotStoppedSpeaking"); + assert!(matches!(stopped, Some(Frame::BotStoppedSpeaking))); + } + + /// Barge-in cuts the burst short: the stopped edge fires immediately (rather + /// than at the end of the flushed-away playout) and exactly once. + #[tokio::test] + async fn bot_speaking_notifier_stops_immediately_on_interruption() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let n = BotSpeakingNotifier::new(tx, 16_000); + + n.note_audio(16_000); // 1 s of audio queued + assert!(matches!(rx.recv().await, Some(Frame::BotStartedSpeaking))); + + n.on_interruption(); + assert!(matches!(rx.recv().await, Some(Frame::BotStoppedSpeaking))); + + // The superseded watchdog must exit silently, not emit a second edge. + n.on_interruption(); + assert!( + tokio::time::timeout(Duration::from_millis(300), rx.recv()) + .await + .is_err(), + "a second BotStoppedSpeaking would re-arm the VAD gate spuriously" + ); + } + + /// The reactor and the frame-path `Interruption` are woken independently, so + /// either can reach the latch first. Regression: a plain bool armed by the + /// reactor *after* the sink had already cleared it stayed set forever, and the + /// sink silently dropped every subsequent reply — the bot went mute for the + /// rest of the call. + #[test] + fn stale_audio_latch_is_order_independent() { + let gen = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let latch = StaleAudioLatch::new(gen.clone()); + assert!(!latch.is_stale(), "idle: nothing to suppress"); + + // Barge-in #1, reactor first (the common case): audio arriving between the + // reactor's flush and the sink's hook belongs to the interrupted reply. + gen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + latch.arm(); + assert!(latch.is_stale()); + latch.clear(); + assert!(!latch.is_stale(), "the sink's hook re-opens the sink"); + + // Barge-in #2, sink first: the late arm must not wedge the latch on. + gen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + latch.clear(); + latch.arm(); + assert!( + !latch.is_stale(), + "an arm for an already-acknowledged barge-in must not mute the sink" + ); + + // A stale arm from a superseded generation can't walk the latch backwards. + gen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + latch.arm(); + latch.clear(); + latch.arm(); + assert!(!latch.is_stale()); + } + + /// Barge-in context repair: the partially streamed reply is kept as the + /// assistant turn (the model must not re-offer what it already said) and the + /// buffer is cleared, so a late `LlmResponseEnd` from the cancelled stream + /// cannot assemble the full reply and speak it after the interruption. + #[tokio::test] + async fn assistant_aggregator_keeps_the_partial_reply_and_drops_the_span() { + let ctx: SharedContext = Arc::new(Mutex::new(RollingContext::new(None, vec![]))); + let mut agg = AssistantContextAggregator::new(ctx.clone()); + agg.in_response = true; + agg.buffer = "the first part of the rep".to_string(); + + agg.on_interruption().await.expect("hook ok"); + + let snap = ctx.lock().unwrap().snapshot(); + assert_eq!(snap.messages.len(), 1); + assert_eq!(snap.messages[0]["role"], "assistant"); + assert_eq!(snap.messages[0]["content"], "the first part of the rep"); + assert!(agg.buffer.is_empty(), "the open span must be dropped"); + assert!(!agg.in_response); + + // Idempotent: a second interruption before any new text adds nothing. + agg.on_interruption().await.expect("hook ok"); + assert_eq!(ctx.lock().unwrap().snapshot().messages.len(), 1); + } + #[test] fn string_tool_result_passes_through_unchanged() { let mut ctx = RollingContext::new(None, vec![]); diff --git a/flowcat-core/src/pipeline/mod.rs b/flowcat-core/src/pipeline/mod.rs index e505ce3..ec4b0a6 100644 --- a/flowcat-core/src/pipeline/mod.rs +++ b/flowcat-core/src/pipeline/mod.rs @@ -42,9 +42,9 @@ use crate::processor::{ }; pub use cascaded::{ - build_cascaded_call_with_observers, build_cascaded_pipeline, build_cascaded_task, - build_cascaded_task_with_observers, CascadedConfig, CascadedTask, ContextSummarizer, - SummarizerConfig, + build_cascaded_call_duplex, build_cascaded_call_with_observers, build_cascaded_pipeline, + build_cascaded_task, build_cascaded_task_with_observers, CascadedConfig, CascadedTask, + ContextSummarizer, SummarizerConfig, }; pub use context_relay::{ ContextCompactor, ContextDigest, ContextRelayConfig, ContextRelayProcessor, LlmCompactor, diff --git a/flowcat-core/src/pipeline/s2s.rs b/flowcat-core/src/pipeline/s2s.rs index a6a3001..8c30d11 100644 --- a/flowcat-core/src/pipeline/s2s.rs +++ b/flowcat-core/src/pipeline/s2s.rs @@ -1056,6 +1056,18 @@ impl FrameProcessor for TransportOutput { Ok(()) } + /// Barge-in (call.rs's `RealtimeEvent::Interrupted` arm): flush the carrier's + /// queued bot audio. The model stops generating service-side, but whatever we + /// already handed the carrier keeps playing until it is cleared. + async fn on_interruption(&mut self) -> Result<()> { + if let Err(e) = self.transport.send_clear().await { + self.state.lock().unwrap().record_error(e); + } + // The carrier dropped its queued bot audio — nothing left to drain. + self.bot_audio_until = None; + Ok(()) + } + async fn process_frame(&mut self, env: Envelope, link: &Link) -> Result<()> { match &env.frame { // Bot audio out (call.rs's RealtimeEvent::AudioOut arm): record at 24k, @@ -1092,16 +1104,6 @@ impl FrameProcessor for TransportOutput { } } } - // Barge-in (call.rs's RealtimeEvent::Interrupted arm): flush the carrier. - Frame::Interruption => { - if let Err(e) = self.transport.send_clear().await { - self.state.lock().unwrap().record_error(e); - } - // The carrier dropped its queued bot audio — nothing left to drain. - self.bot_audio_until = None; - // Forward the interruption in its direction (framework also drains). - link.push(env.meta, env.frame, env.direction).await; - } // End-of-call: let the final bot utterance finish playing before the // carrier WS closes. The model emits the goodbye audio AND `endCall` in // one turn, so by the time `End` reaches us the goodbye is already @@ -1854,6 +1856,73 @@ mod tests { ); } + /// A carrier that records how many times playback was flushed. + struct ClearSpy { + clears: Arc, + rate: u32, + } + #[async_trait::async_trait] + impl MediaTransport for ClearSpy { + async fn recv(&mut self) -> Option { + None + } + async fn send_audio(&mut self, _c: AudioChunk) -> std::result::Result<(), FlowcatError> { + Ok(()) + } + async fn send_clear(&mut self) -> std::result::Result<(), FlowcatError> { + self.clears + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(()) + } + fn carrier_rate(&self) -> u32 { + self.rate + } + } + + /// Regression: barge-in must flush the carrier's queued bot audio. + /// + /// `Frame::Interruption` is a lifecycle frame the runtime intercepts — it never + /// reaches `process_frame`, so the sink's flush has to hang off the + /// `on_interruption` hook. As a `process_frame` arm it was unreachable, and the + /// already-queued reply kept playing over the interrupting caller even though + /// the model had stopped generating service-side. + #[tokio::test] + async fn barge_in_flushes_the_carrier_playback_at_the_sink() { + let clears = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let state: SharedState = Arc::new(Mutex::new(LiveState::new(CARRIER_RATE))); + let sink = TransportOutput::new( + ClearSpy { + clears: clears.clone(), + rate: CARRIER_RATE, + }, + CARRIER_RATE, + state, + ); + + let task = PipelineTask::new( + Pipeline::new(vec![Box::new(sink)]), + PipelineTaskParams::default(), + vec![], + ); + task.queue_frame(Frame::OutputAudio(Arc::new(AudioFrame::mono( + vec![1i16; 480], + GEMINI_OUTPUT_RATE, + )))) + .await; + task.queue_frame(Frame::Interruption).await; + task.stop_when_done().await; + tokio::time::timeout(Duration::from_secs(5), task.run()) + .await + .expect("sink pipeline timed out") + .expect("run ok"); + + assert_eq!( + clears.load(std::sync::atomic::Ordering::SeqCst), + 1, + "the interruption must reach the sink and flush the carrier exactly once" + ); + } + /// Records bot-side frames at their origin (`source == "RealtimeService"`): /// every final bot transcript line + whether a `BotStoppedSpeaking` was emitted. #[derive(Default)] diff --git a/flowcat-core/src/processor/mod.rs b/flowcat-core/src/processor/mod.rs index 87eaf8c..3ca2496 100644 --- a/flowcat-core/src/processor/mod.rs +++ b/flowcat-core/src/processor/mod.rs @@ -294,10 +294,12 @@ impl Link { /// - [`Frame::Start`] → calls [`start`](FrameProcessor::start), then forwards. /// - A **downstream** [`Frame::End`]/[`Frame::Stop`]/[`Frame::Cancel`] → calls /// [`stop`](FrameProcessor::stop), forwards, and (End/Cancel) ends the task. -/// - [`Frame::Interruption`] → drains the interruptible backlog, then forwards. +/// - [`Frame::Interruption`] → drains the interruptible backlog, calls +/// [`on_interruption`](FrameProcessor::on_interruption), then forwards. /// /// So **your `process_frame` never sees these frames** — observe lifecycle via the -/// `start`/`stop` hooks (this is why the internal `Sink` taps from the hooks). An +/// `start`/`on_interruption`/`stop` hooks (this is why the internal `Sink` taps +/// from the hooks). An /// *upstream* End/Stop/Cancel is the exception: it is a "request to end" and DOES /// reach `process_frame` (the default forwards it upstream so the `Source` can /// convert it to a downstream drain — pipecat's `EndTaskFrame` vs `EndFrame`). Keep @@ -325,6 +327,22 @@ pub trait FrameProcessor: Send + 'static { Ok(()) } + /// Called on [`Frame::Interruption`] (barge-in), after this processor's queued + /// interruptible frames were drained and before the interruption is forwarded. + /// Default: no-op. Override to react — flush carrier playback, reset a text + /// aggregator, repair context. + /// + /// This hook exists because `Interruption` is a lifecycle frame the runtime + /// intercepts: like `Start`/`End`, it never reaches + /// [`process_frame`](FrameProcessor::process_frame), so a `Frame::Interruption` + /// arm there is silently dead code. + /// + /// **Must not block** — same contract as `process_frame`. Interruption is the + /// latency-critical path; anything slow here delays the audible stop. + async fn on_interruption(&mut self) -> Result<()> { + Ok(()) + } + /// Called on `End`/`Stop`/`Cancel` after the terminal frame is forwarded. /// Flush + close. Default: no-op. async fn stop(&mut self, _reason: StopReason) -> Result<()> { diff --git a/flowcat-core/src/processor/runtime.rs b/flowcat-core/src/processor/runtime.rs index 74634f6..b603a5c 100644 --- a/flowcat-core/src/processor/runtime.rs +++ b/flowcat-core/src/processor/runtime.rs @@ -11,8 +11,9 @@ //! interruption can never block on a full queue. //! //! The loop biases the system channel, runs the processor's lifecycle hooks on -//! lifecycle frames, drains interruptible frames on [`Frame::Interruption`], and -//! converts a `process_frame` `Err` into an upstream [`Frame::Error`]. +//! lifecycle frames, drains interruptible frames on [`Frame::Interruption`] (then +//! calls [`FrameProcessor::on_interruption`]), and converts a `process_frame` +//! `Err` into an upstream [`Frame::Error`]. use tokio::sync::mpsc; @@ -98,8 +99,8 @@ fn stop_reason(frame: &Frame) -> StopReason { /// - `biased` select drains the **system** channel first (priority); /// - `Start` runs `start()` then forwards; /// - `Interruption` drains the normal queue of *interruptible* frames (keeping -/// uninterruptible ones — End/Stop/FunctionCallResult/UpdateSettings) then -/// forwards; +/// uninterruptible ones — End/Stop/FunctionCallResult/UpdateSettings), calls +/// `on_interruption()`, then forwards; /// - `End`/`Stop`/`Cancel` run `stop()`, forward, and (End/Cancel) break; /// - any other frame goes to `process_frame`; an `Err` becomes an upstream /// `Error{fatal:false}`. @@ -142,6 +143,13 @@ pub async fn run_processor( // uninterruptible ones. A kept *downstream terminal* (End/Stop) is // returned rather than blindly forwarded — see below. let kept_terminal = drain_on_interruption(&mut rx, &link).await; + // Barge-in hook: the processor reacts (flush playback, reset an + // aggregator, repair context) on a queue that is already clean. + // `Interruption` never reaches `process_frame` — this is the only + // delivery path (PROCESSOR-DESIGN §2.5). + if let Err(e) = p.on_interruption().await { + link.push_error(e.to_string(), false).await; + } // Forward the interruption in the direction it arrived. link.push(env.meta, env.frame, env.direction).await; // A downstream End/Stop that was buffered when the interruption hit @@ -308,4 +316,117 @@ mod tests { "the Sink's stop() hook must run for the buffered End (else PipelineTask hangs)" ); } + + /// Records the order in which the runtime ran hooks vs. delivered data frames. + struct HookOrderTap { + log: Arc>>, + } + #[async_trait] + impl FrameProcessor for HookOrderTap { + fn name(&self) -> &str { + "HookOrderTap" + } + async fn on_interruption(&mut self) -> Result<()> { + self.log.lock().unwrap().push("on_interruption"); + Ok(()) + } + async fn process_frame(&mut self, env: Envelope, _link: &Link) -> Result<()> { + self.log.lock().unwrap().push(env.frame.name()); + Ok(()) + } + } + + /// `Frame::Interruption` is intercepted by the runtime and never reaches + /// `process_frame` — `on_interruption` is the only delivery path, and it runs + /// *after* the interruptible backlog is drained (so the hook sees a clean + /// queue). Regression for barge-in reactions silently never firing. + #[tokio::test] + async fn interruption_calls_the_hook_and_never_reaches_process_frame() { + let name: Arc = Arc::from("HookOrderTap"); + let (tx, rx) = channel(name.clone(), 8); + let clock = Clock::new(); + let setup = ProcessorSetup { + clock: clock.clone(), + observer: None, + cancel: tokio_util::sync::CancellationToken::new(), + enable_metrics: false, + enable_usage_metrics: false, + }; + let log = Arc::new(std::sync::Mutex::new(Vec::new())); + let p = Box::new(HookOrderTap { log: log.clone() }); + + // An interruptible data frame is already queued when the interruption lands. + tx.send(Envelope::new( + Frame::Text("stale".into()), + Direction::Downstream, + )) + .await; + tx.send(Envelope::new(Frame::Interruption, Direction::Downstream)) + .await; + drop(tx); + + let h = tokio::spawn(run_processor(p, rx, sink_link(name, clock), setup)); + tokio::time::timeout(std::time::Duration::from_secs(2), h) + .await + .expect("run_processor hung") + .expect("run_processor panicked"); + + let seen = log.lock().unwrap().clone(); + assert!( + !seen.contains(&"Interruption"), + "Interruption must not reach process_frame, got {seen:?}" + ); + assert_eq!( + seen, + vec!["on_interruption"], + "the hook runs exactly once, and the stale interruptible backlog is \ + drained rather than delivered, got {seen:?}" + ); + } + + /// An `Err` from `on_interruption` becomes a non-fatal upstream `Error` + /// instead of killing the task — same contract as `process_frame`. + #[tokio::test] + async fn on_interruption_error_is_reported_and_the_task_survives() { + struct Failing; + #[async_trait] + impl FrameProcessor for Failing { + fn name(&self) -> &str { + "Failing" + } + async fn on_interruption(&mut self) -> Result<()> { + Err(crate::error::FlowcatError::Other("boom".into())) + } + } + + let name: Arc = Arc::from("Failing"); + let (tx, rx) = channel(name.clone(), 8); + let clock = Clock::new(); + let setup = ProcessorSetup { + clock: clock.clone(), + observer: None, + cancel: tokio_util::sync::CancellationToken::new(), + enable_metrics: false, + enable_usage_metrics: false, + }; + tx.send(Envelope::new(Frame::Interruption, Direction::Downstream)) + .await; + tx.send(Envelope::new( + Frame::End { reason: None }, + Direction::Downstream, + )) + .await; + drop(tx); + + let h = tokio::spawn(run_processor( + Box::new(Failing), + rx, + sink_link(name, clock), + setup, + )); + tokio::time::timeout(std::time::Duration::from_secs(2), h) + .await + .expect("a failing on_interruption must not hang the task") + .expect("run_processor panicked"); + } } diff --git a/flowcat-core/src/service/adapters.rs b/flowcat-core/src/service/adapters.rs index a4498a3..64e2862 100644 --- a/flowcat-core/src/service/adapters.rs +++ b/flowcat-core/src/service/adapters.rs @@ -124,6 +124,21 @@ impl FrameProcessor for SttProcessor { Err(e) => link.push_error(format!("stt: {e}"), false).await, } } + // End of a VAD-delimited utterance: give a batch service the chance to + // transcribe its buffered tail now instead of waiting for its next + // fixed window (which would split the turn). No-op for a streaming + // service — `SttService::flush` defaults to returning nothing. + Frame::UserStoppedSpeaking => { + match self.svc.lock().await.flush().await { + Ok(frames) => { + for f in frames { + link.push_down(f).await; + } + } + Err(e) => link.push_error(format!("stt flush: {e}"), false).await, + } + link.push(env.meta, env.frame, env.direction).await; + } // STT mute control (pipecat `STTMuteFrame`). Forward so a downstream // observer/UI still sees it. Frame::SttMute(muted) => { @@ -155,6 +170,12 @@ impl FrameProcessor for SttProcessor { pub struct LlmProcessor { /// The wrapped LLM service. svc: Arc>, + /// Optional barge-in generation counter (see + /// [`VadProcessor::with_interrupt_flag`](crate::audio::VadProcessor)): + /// checked between streamed chunks so an in-flight completion stops pushing + /// promptly when the user barges in. The frame-path `Interruption` cannot do + /// this — `process_frame` is already running when it arrives. + interrupt_flag: Option>, } impl LlmProcessor { @@ -162,9 +183,16 @@ impl LlmProcessor { pub fn new(svc: L) -> Self { Self { svc: Arc::new(Mutex::new(svc)), + interrupt_flag: None, } } + /// Cancel an in-flight stream when `flag` is bumped (duplex barge-in). + pub fn with_interrupt_flag(mut self, flag: Arc) -> Self { + self.interrupt_flag = Some(flag); + self + } + /// Run the LLM over `ctx`, pushing each streamed frame downstream in order. async fn run(&self, ctx: &LlmContext, link: &Link) { tracing::debug!(messages = ctx.messages.len(), "cascaded LLM run"); @@ -173,9 +201,23 @@ impl LlmProcessor { // guard outlives the stream. Push each frame as it arrives (true streaming); // `link.push_down` doesn't touch the guard, so the borrow is fine. let mut pushed = 0usize; + let start_gen = self + .interrupt_flag + .as_ref() + .map(|f| f.load(std::sync::atomic::Ordering::SeqCst)); + let mut cancelled = false; let err = match guard.run_llm(ctx).await { Ok(mut stream) => { while let Some(f) = stream.next().await { + // Cooperative barge-in cancel: stop consuming (and pushing) + // as soon as the barge-in generation moves. Dropping the + // stream aborts the underlying request. + if let (Some(flag), Some(g0)) = (&self.interrupt_flag, start_gen) { + if flag.load(std::sync::atomic::Ordering::SeqCst) != g0 { + cancelled = true; + break; + } + } pushed += 1; link.push_down(f).await; } @@ -183,6 +225,12 @@ impl LlmProcessor { } Err(e) => Some(format!("llm: {e}")), }; + if cancelled { + tracing::debug!(pushed, "cascaded LLM stream cancelled by barge-in"); + // Close the response framing so downstream aggregators can't wedge + // in an open in_response span. + link.push_down(Frame::LlmResponseEnd).await; + } drop(guard); match &err { Some(msg) => tracing::warn!(error = %msg, "cascaded LLM error"), @@ -332,3 +380,241 @@ impl FrameProcessor for TtsProcessor { Ok(()) } } + +// =========================================================================== +// Tests — the barge-in seams the adapters own (offline, no provider). +// =========================================================================== +#[cfg(test)] +mod tests { + use super::*; + use crate::observer::{FrameEvent, FrameObserver}; + use crate::pipeline::{Pipeline, PipelineTask, PipelineTaskParams}; + use crate::processor::frame::AudioFrame; + use crate::service::Tool; + use futures::stream::BoxStream; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::Mutex as StdMutex; + use std::time::Duration; + + /// Records every frame name the pipeline processed, plus the `LlmText` payloads. + #[derive(Default)] + struct Tap { + names: StdMutex>, + llm_text: StdMutex>, + transcripts: StdMutex>, + } + #[async_trait] + impl FrameObserver for Tap { + async fn on_process(&self, e: &FrameEvent<'_>) { + self.names.lock().unwrap().push(e.frame.name()); + match &e.frame { + // One entry per *distinct* emitted frame — `on_process` fires once + // per processor that sees it, so dedupe on the internal Sink hop. + Frame::LlmText(t) if e.processor == "Sink" => { + self.llm_text.lock().unwrap().push(t.clone()) + } + Frame::Transcription { text, .. } if e.processor == "Sink" => { + self.transcripts.lock().unwrap().push(text.clone()) + } + _ => {} + } + } + } + + // ---- STT flush seam --------------------------------------------------- + + /// A batch-style STT: buffers audio and only transcribes on `flush()` — the + /// shape `whisper_local` has, and the reason the seam exists. + struct BufferingStt { + buffered: usize, + } + #[async_trait] + impl SttService for BufferingStt { + fn name(&self) -> &str { + "BufferingStt" + } + async fn start(&mut self, _p: &StartParams) -> Result<()> { + Ok(()) + } + async fn run_stt(&mut self, audio: Arc) -> Result> { + self.buffered += audio.pcm.len(); + Ok(vec![]) + } + async fn flush(&mut self) -> Result> { + if self.buffered == 0 { + return Ok(vec![]); + } + let n = std::mem::take(&mut self.buffered); + Ok(vec![Frame::Transcription { + text: format!("{n} samples"), + user_id: Arc::from("user"), + language: None, + final_: true, + }]) + } + async fn set_muted(&mut self, _muted: bool) {} + } + + /// `UserStoppedSpeaking` (a VAD falling edge) must reach the service as a + /// `flush()` — otherwise a batch STT's last window sits in its buffer until the + /// *next* turn's audio arrives, and two utterances merge into one late + /// transcript. (The edge is a System frame and the transcription a Control one, + /// so their *arrival* order downstream is set by the channel split, not here; + /// what this pins is that the flush happens and its text is emitted.) + #[tokio::test] + async fn user_stopped_speaking_flushes_a_batch_stt_before_forwarding_the_edge() { + let tap = Arc::new(Tap::default()); + let pipeline = Pipeline::new(vec![Box::new(SttProcessor::new(BufferingStt { + buffered: 0, + }))]); + let task = PipelineTask::new( + pipeline, + PipelineTaskParams::default(), + vec![tap.clone() as Arc], + ); + + task.queue_frame(Frame::InputAudio(Arc::new(AudioFrame::mono( + vec![1i16; 40], + 16_000, + )))) + .await; + task.queue_frame(Frame::UserStoppedSpeaking).await; + task.stop_when_done().await; + tokio::time::timeout(Duration::from_secs(5), task.run()) + .await + .expect("stt flush pipeline timed out") + .expect("run ok"); + + assert_eq!( + tap.transcripts.lock().unwrap().clone(), + vec!["40 samples".to_string()], + "the buffered tail must be transcribed at the falling edge" + ); + let names = tap.names.lock().unwrap().clone(); + assert!( + names.contains(&"UserStoppedSpeaking"), + "the edge must still be forwarded after the flush; saw {names:?}" + ); + } + + /// The default `flush()` is a no-op, so a streaming service (which does its own + /// endpointing) is unaffected by the new seam. + #[tokio::test] + async fn default_flush_is_a_no_op_for_a_streaming_service() { + struct Streaming; + #[async_trait] + impl SttService for Streaming { + fn name(&self) -> &str { + "Streaming" + } + async fn start(&mut self, _p: &StartParams) -> Result<()> { + Ok(()) + } + async fn run_stt(&mut self, _a: Arc) -> Result> { + Ok(vec![]) + } + async fn set_muted(&mut self, _muted: bool) {} + } + assert!(Streaming.flush().await.expect("flush ok").is_empty()); + } + + // ---- LLM cooperative barge-in cancel ----------------------------------- + + /// An LLM whose stream raises the shared barge-in flag partway through, i.e. + /// the user barges in while the completion is still streaming. + struct BargeInMidStream { + flag: Arc, + chunks: usize, + /// Zero-based index of the chunk at which the barge-in fires. + at: usize, + } + #[async_trait] + impl LlmService for BargeInMidStream { + fn name(&self) -> &str { + "BargeInMidStream" + } + async fn start(&mut self, _p: &StartParams) -> Result<()> { + Ok(()) + } + async fn run_llm<'a>(&'a mut self, _ctx: &'a LlmContext) -> Result> { + let flag = self.flag.clone(); + let (chunks, at) = (self.chunks, self.at); + Ok(Box::pin(futures::stream::unfold(0usize, move |i| { + let flag = flag.clone(); + async move { + if i >= chunks { + return None; + } + if i == at { + flag.fetch_add(1, Ordering::SeqCst); + } + Some((Frame::LlmText(format!("t{i}")), i + 1)) + } + }))) + } + fn set_tools(&mut self, _tools: Vec) {} + } + + async fn run_llm_turn(llm: LlmProcessor) -> Arc { + let tap = Arc::new(Tap::default()); + let task = PipelineTask::new( + Pipeline::new(vec![Box::new(llm)]), + PipelineTaskParams::default(), + vec![tap.clone() as Arc], + ); + task.queue_frame(Frame::LlmContext(Arc::new(LlmContext { + messages: vec![json!({"role": "user", "content": "hi"})], + tools: vec![], + }))) + .await; + task.stop_when_done().await; + tokio::time::timeout(Duration::from_secs(5), task.run()) + .await + .expect("llm pipeline timed out") + .expect("run ok"); + tap + } + + /// With the flag wired, a barge-in mid-stream stops the adapter pushing further + /// tokens and closes the response framing. Without the cancel the whole reply + /// would still be assembled and spoken *after* the user interrupted — the frame + /// path can't preempt a `process_frame` that is already inside the stream. + #[tokio::test] + async fn barge_in_flag_cancels_an_in_flight_llm_stream() { + let flag = Arc::new(AtomicU64::new(0)); + let tap = run_llm_turn( + LlmProcessor::new(BargeInMidStream { + flag: flag.clone(), + chunks: 6, + at: 2, + }) + .with_interrupt_flag(flag), + ) + .await; + + assert_eq!( + tap.llm_text.lock().unwrap().clone(), + vec!["t0".to_string(), "t1".to_string()], + "tokens from the barge-in chunk onward must not be pushed" + ); + assert!( + tap.names.lock().unwrap().contains(&"LlmResponseEnd"), + "a cancelled stream must still close its response framing" + ); + } + + /// Same LLM, no flag wired (the default / half-duplex path): every token is + /// pushed. Guards against the cancel changing stock behaviour. + #[tokio::test] + async fn without_the_flag_the_whole_stream_is_pushed() { + // Same barge-in mid-stream, but no `with_interrupt_flag` — the adapter has + // nothing to poll, so it streams to completion exactly as it does today. + let tap = run_llm_turn(LlmProcessor::new(BargeInMidStream { + flag: Arc::new(AtomicU64::new(0)), + chunks: 4, + at: 1, + })) + .await; + assert_eq!(tap.llm_text.lock().unwrap().len(), 4); + } +} diff --git a/flowcat-core/src/service/mod.rs b/flowcat-core/src/service/mod.rs index fb5a759..c6dc9f5 100644 --- a/flowcat-core/src/service/mod.rs +++ b/flowcat-core/src/service/mod.rs @@ -43,6 +43,18 @@ pub trait SttService: Send { /// Feed one audio chunk; transcript frames are returned for the processor to /// forward downstream. async fn run_stt(&mut self, audio: Arc) -> Result>; + /// End-of-utterance: transcribe whatever is still buffered and return it. + /// + /// Called by [`SttProcessor`](crate::service::adapters::SttProcessor) on + /// [`Frame::UserStoppedSpeaking`], i.e. only when something upstream (a + /// [`VadProcessor`](crate::audio::VadProcessor)) actually does endpointing. + /// A *fixed-window batch* service (whisper.cpp) needs this: without it the + /// tail of an utterance sits in its buffer until enough later audio arrives + /// to cross the window, which splits turns at arbitrary boundaries. A + /// *streaming* service does its own endpointing — the default no-op is right. + async fn flush(&mut self) -> Result> { + Ok(vec![]) + } async fn set_muted(&mut self, muted: bool); } @@ -90,6 +102,9 @@ impl SttService for Box { async fn run_stt(&mut self, audio: Arc) -> Result> { (**self).run_stt(audio).await } + async fn flush(&mut self) -> Result> { + (**self).flush().await + } async fn set_muted(&mut self, muted: bool) { (**self).set_muted(muted).await } diff --git a/flowcat-services/src/stt/whisper_local.rs b/flowcat-services/src/stt/whisper_local.rs index 2408d1d..90d7fac 100644 --- a/flowcat-services/src/stt/whisper_local.rs +++ b/flowcat-services/src/stt/whisper_local.rs @@ -307,6 +307,23 @@ impl SttService for WhisperLocalStt { Self::transcribe_segment(ctx, self.language.clone(), samples).await } + /// End-of-utterance (a VAD falling edge upstream): transcribe the buffered + /// tail regardless of the segment threshold. Without this the last (partial) + /// window of every turn waits for the *next* turn's audio to push it over the + /// threshold, which merges two utterances into one late transcript. + async fn flush(&mut self) -> Result> { + if self.muted || self.buffer.is_empty() { + return Ok(vec![]); + } + let ctx = self + .ctx + .as_ref() + .ok_or_else(|| FlowcatError::Other("whisper_local: flush before start".into()))? + .clone(); + let samples = self.buffer.drain(); + Self::transcribe_segment(ctx, self.language.clone(), samples).await + } + async fn set_muted(&mut self, muted: bool) { self.muted = muted; }