flowcat-core: opt-in full-duplex (barge-in) support for the cascaded pipeline

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

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<dyn FrameProcessor>, 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 |

flowcat-core/src/agent/mod.rs

@@ -42,6 +42,17 @@ pub(crate) mod test_harness {
         mut proc: Box<dyn FrameProcessor>,
         frames: Vec<Frame>,
         direction: Direction,
+    ) -> Vec<Frame> {
+        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<Frame>,
+        direction: Direction,
     ) -> Vec<Frame> {
         // One capture channel acts as both the downstream and upstream neighbour.
         let (cap_tx, mut cap_rx) = channel(Arc::from("capture"), NORMAL_CHAN_CAP);

flowcat-core/src/agent/text_filter.rs

@@ -527,6 +527,12 @@ impl<A: TextAggregator + 'static> FrameProcessor for TextAggregatorProcessor<A>
         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<A: TextAggregator + 'static> FrameProcessor for TextAggregatorProcessor<A>
                 }
                 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<F: TextFilter + 'static> FrameProcessor for TextFilterProcessor<F> {
         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<F: TextFilter + 'static> FrameProcessor for TextFilterProcessor<F> {
                         .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<F: TextFilter + 'static> FrameProcessor for TextFilterProcessor<F> {
 #[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());

flowcat-core/src/audio/vad.rs

@@ -338,6 +338,18 @@ pub struct VadProcessor<V: VadAnalyzer> {
     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<std::sync::Arc<std::sync::atomic::AtomicU64>>,
+    /// 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<std::sync::Arc<tokio::sync::Notify>>,
 }
 
 impl<V: VadAnalyzer> VadProcessor<V> {
@@ -353,9 +365,26 @@ impl<V: VadAnalyzer> VadProcessor<V> {
             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<tokio::sync::Notify>) -> 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<std::sync::atomic::AtomicU64>,
+    ) -> 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<V: VadAnalyzer> VadProcessor<V> {
                     .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;
                 }
             }