diff --git a/README.md b/README.md
index 16aecfb..ee3431e 100644
--- a/README.md
+++ b/README.md
@@ -180,6 +180,7 @@ Each entry in `phases` represents one user message and the assertions to run aga
 | `send` | `str` | required | The user utterance to inject into the pipeline. |
 | `expect_reply_contains` | `list[str]` | `[]` | Every string in this list must appear (case-insensitive regex) in the bot's reply. |
 | `expect_llm_judge` | `list[str] \| null` | `null` | List of natural language expectations evaluated dynamically by an LLM. See [`llm_judge_function`](#llm_judge_function). |
+| `barge_in` | `bool \| object` | `false` | Interrupt the **previous** phase's bot reply mid-stream, then send this phase's `send`. See [Barge-in](#barge-in). |
 | `expect_tools` | `list \| null` | `null` | Exact set of tool/function names the LLM must call during this turn. Entries can be plain names or objects that also assert the call **arguments** — see [below](#asserting-tool-call-arguments). `null` skips the assertion. |
 | `expect_flow_state` | `str \| null` | `null` | Expected pipecat-flows node name after this phase. Requires [flows integration](#pipecat-flows-integration). |
 
@@ -231,6 +232,36 @@ Semantics:
   actual call arguments — keys you don't list are ignored, and each listed value
   must compare equal to the actual argument.
 
+#### Barge-in
+
+> **Experimental:** the syntax and behaviour may change in a future release.
+
+Set `barge_in` on a phase to make the user interrupt the bot while it is still answering the **previous** phase. Use it to check that the bot drops the old answer and handles the new message.
+
+```yaml
+phases:
+  - send: "Read me the full cancellation policy."
+    expect_reply_contains:
+      - "cancel"                 # checked against the part of the reply before the interruption
+
+  - send: "Stop, just tell me the fee."
+    barge_in:
+      after: "within 24 hours"   # interrupt right after this regex matches...
+      # after_words: 8           # ...or after the 8th word
+    expect_reply_contains:
+      - "fee"
+```
+
+Semantics:
+
+- `barge_in: true` interrupts after the first word. Otherwise set exactly one of `after` (a case-insensitive regex) or `after_words`.
+- The bot's reply is cut at that point: it only "said" what came before, and that is all it remembers.
+- The test fails if the reply ends before the trigger is reached (regex not found, fewer words than `after_words`).
+- Only the first reply to a phase is interrupted, so [auto-confirm](#autoconfirm_response) is skipped for it.
+- Invalid settings (both triggers, a bad regex, `barge_in` on the first phase) are reported when the file is collected.
+
+> **Note:** the interruption is triggered by a position in the reply text, not by real audio. Voice activity detection, interruption strategies and TTS timing are not tested; use an end-to-end test for those.
+
 #### Mocking and Setup (Custom Fixtures)
 
 If your test requires specific database states or mocked tool responses, you can inject standard pytest fixtures defined in your `conftest.py` directly into the YAML using the `fixtures` key.
diff --git a/src/pytest_catnip/harness.py b/src/pytest_catnip/harness.py
index 5a073b4..2a76c89 100644
--- a/src/pytest_catnip/harness.py
+++ b/src/pytest_catnip/harness.py
@@ -1,6 +1,9 @@
 import asyncio
+import contextlib
 import logging
+import re
 import time
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from enum import StrEnum
 from typing import Any, override
@@ -9,13 +12,17 @@
     Frame,
     FunctionCallResultFrame,
     FunctionCallsStartedFrame,
+    InterruptionFrame,
     LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
     LLMTextFrame,
     TranscriptionFrame,
     UserStoppedSpeakingFrame,
 )
 from pipecat.pipeline.worker import PipelineWorker
+from pipecat.processors.aggregators.llm_response_universal import LLMAssistantAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pytest_catnip.models import BargeIn
 
 logger = logging.getLogger(__name__)
 
@@ -34,6 +41,53 @@ class TurnLogKind(StrEnum):
     TOOL = "TOOL"
 
 
+def _last_assistant_text(aggregator: LLMAssistantAggregator) -> str | None:
+    """Whitespace-normalised text of the last assistant message in the aggregator's context."""
+    for message in reversed(aggregator.context.get_messages()):
+        if message.get("role") != "assistant":
+            continue
+        content = message.get("content")
+        if isinstance(content, list):
+            content = " ".join(str(part.get("text", "")) for part in content if isinstance(part, dict))
+        return " ".join(str(content or "").split())
+    return None
+
+
+class BargeInMissedError(AssertionError):
+    """The bot's reply ended before the barge-in trigger was reached."""
+
+    def __init__(self, reply: str, barge_in: BargeIn, reason: str) -> None:
+        super().__init__(reason)
+        self.reply = reply
+        self.barge_in = barge_in
+        self.reason = reason
+
+
+@dataclass
+class _TurnResult:
+    text: str
+    interrupted: bool = False
+    barge_in_miss: str | None = None
+
+
+def _find_cut(barge_in: BargeIn, text: str, *, at_end: bool) -> int | None:
+    """Return the index in *text* right after the barge-in trigger, or ``None`` if not reached yet.
+
+    Mid-stream, the last word is only counted once whitespace follows it (it may still be growing).
+    At the end of the reply every word is complete.
+    """
+    if barge_in.after is not None:
+        match = re.search(barge_in.after, text, re.IGNORECASE)
+        return match.end() if match else None
+    words = list(re.finditer(r"\S+", text))
+    if barge_in.after_words is None or len(words) < barge_in.after_words:
+        return None
+    last = words[barge_in.after_words - 1]
+    if last.end() == len(text) and not at_end:
+        return None
+    return last.end()
+
+
 class CatnipTurnTracker(FrameProcessor):
     """Frame processor that signals when a complete LLM turn is ready.
 
@@ -44,16 +98,119 @@ class CatnipTurnTracker(FrameProcessor):
 
     def __init__(self) -> None:
         super().__init__()
-        self._reply_queue: asyncio.Queue[str] = asyncio.Queue()
+        self._interrupter: Callable[[], Awaitable[None]] | None = None
+        self._interrupted_response_ended = asyncio.Event()
+        self._reply_queue: asyncio.Queue[_TurnResult] = asyncio.Queue()
         self._inflight: int = 0
         self._chunks: list[str] = []
         self.tool_calls: list[_ToolCall] = []
+        self._barge_in: BargeIn | None = None
+        self._swallowing: bool = False
+        self._assistant_aggregator: LLMAssistantAggregator | None = None
+
+    def watch_assistant_aggregator(self, aggregator: LLMAssistantAggregator | None) -> None:
+        """Let barge-in wait for *aggregator* to consume the "spoken" text before interrupting it.
+
+        System frames such as ``InterruptionFrame`` overtake queued data frames, so without this the
+        aggregator would be interrupted before it has seen the text that was supposedly spoken.
+        """
+        self._assistant_aggregator = aggregator
+
+    async def _wait_until_aggregated(self, spoken: str, timeout: float = 2.0) -> None:
+        if self._assistant_aggregator is None:
+            return
+        expected = " ".join(spoken.split())
+        try:
+            async with asyncio.timeout(timeout):
+                while " ".join(self._assistant_aggregator.aggregation_string().split()) != expected:
+                    await asyncio.sleep(0.005)
+        except TimeoutError:
+            logger.warning("[tracker] assistant aggregator never consumed the spoken text %r", expected)
+
+    def set_interrupter(self, interrupter: Callable[[], Awaitable[None]] | None) -> None:
+        """Set how a barge-in enters the pipeline.
+
+        Real interruptions start at the source (the transport) and flow downstream through every
+        processor. Without an *interrupter* the tracker pushes the frame both ways from its own position.
+        """
+        self._interrupter = interrupter
+
+    def arm_barge_in(self, barge_in: BargeIn | None) -> None:
+        """Interrupt the next bot reply once *barge_in*'s trigger is reached.
+
+        Streamed text is simulated as "spoken" up to the trigger point; anything after it is
+        dropped and an interruption is broadcast through the pipeline. Call before ``prepare_for_input``.
+        """
+        self._barge_in = barge_in
 
     def prepare_for_input(self) -> None:
         """Reset per-turn accumulation state before sending a new user message."""
         self._chunks = []
         self._inflight = 0
 
+    async def _interrupt_at(self, cut: int, *, response_ended: bool = False) -> None:
+        """Finish the current reply at *cut*, then interrupt the rest of the pipeline."""
+        self._interrupted_response_ended.clear()
+        if response_ended:
+            self._interrupted_response_ended.set()
+        text = "".join(self._chunks)[:cut].strip()
+        self._chunks = []
+        self._barge_in = None
+        self._swallowing = True
+        logger.debug("[tracker] barge-in fired, reply cut at %d chars", len(text))
+        # Run apart from this frame's processing: the interruption cancels the tracker's own process task.
+        self.create_task(self._deliver_interruption(text), name="catnip-barge-in")
+
+    async def _deliver_interruption(self, spoken: str) -> None:
+        """Interrupt the pipeline; the turn's result is only released once everything has settled.
+
+        Releasing it last keeps the next user message from racing the tail of this interruption.
+        """
+        try:
+            await self._wait_until_aggregated(spoken)
+            if self._interrupter is not None:
+                await self._interrupter()
+            else:
+                await self.push_frame(InterruptionFrame(), FrameDirection.UPSTREAM)
+                await self.push_frame(InterruptionFrame(), FrameDirection.DOWNSTREAM)
+            # A cancelled LLM still ends its response, sometimes late. If that End frame crossed the next
+            # response's Start it would be taken for that response's end, so wait for it (bounded: a
+            # service may not send one).
+            with contextlib.suppress(TimeoutError):
+                await asyncio.wait_for(self._interrupted_response_ended.wait(), timeout=2.0)
+        finally:
+            await self._reply_queue.put(_TurnResult(spoken, interrupted=True))
+
+    async def _gate_text(self, frame: LLMTextFrame, barge_in: BargeIn) -> None:
+        """Forward only the part of *frame* that comes before the barge-in trigger."""
+        before = "".join(self._chunks)
+        self._chunks.append(frame.text)
+        cut = _find_cut(barge_in, "".join(self._chunks), at_end=False)
+        if cut is None:
+            await self.push_frame(frame, FrameDirection.DOWNSTREAM)
+            return
+        head = frame.text[: max(cut - len(before), 0)]
+        if head:
+            await self.push_frame(LLMTextFrame(text=head), FrameDirection.DOWNSTREAM)
+        await self._interrupt_at(cut)
+
+    async def _gate_end(self, frame: LLMFullResponseEndFrame, barge_in: BargeIn) -> None:
+        text = "".join(self._chunks)
+        cut = _find_cut(barge_in, text, at_end=True)
+        if cut is not None:
+            await self._interrupt_at(cut, response_ended=True)
+            return
+        self._barge_in = None
+        self._chunks = []
+        if not text.strip():
+            reason = "the bot streamed no text to interrupt"
+        elif barge_in.after is not None:
+            reason = f"barge_in.after {barge_in.after!r} not found in the bot reply"
+        else:
+            reason = f"barge_in.after_words is {barge_in.after_words} but the bot reply had {len(text.split())} words"
+        await self._reply_queue.put(_TurnResult(text.strip(), barge_in_miss=reason))
+        await self.push_frame(frame, FrameDirection.DOWNSTREAM)
+
     @override
     async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:  # noqa: C901
         await super().process_frame(frame, direction)
@@ -76,15 +233,33 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
                 self._inflight,
             )
 
+        elif self._swallowing and isinstance(frame, LLMFullResponseStartFrame):
+            # A new response begins: the interrupted one is over.
+            self._swallowing = False
+
+        elif self._swallowing and isinstance(frame, LLMTextFrame | LLMFullResponseEndFrame):
+            # The rest of an interrupted reply is never "spoken". Its end frame may arrive after the
+            # next user message was sent, so keep swallowing until it (or the next response) shows up.
+            self._swallowing = not isinstance(frame, LLMFullResponseEndFrame)
+            if isinstance(frame, LLMFullResponseEndFrame):
+                self._interrupted_response_ended.set()
+            return
+
         elif isinstance(frame, LLMTextFrame):
+            if self._barge_in is not None:
+                await self._gate_text(frame, self._barge_in)
+                return
             self._chunks.append(frame.text)
 
         elif isinstance(frame, LLMFullResponseEndFrame):
             if self._inflight <= 0:
                 # All tool calls resolved — this is the final response for this turn.
+                if self._barge_in is not None:
+                    await self._gate_end(frame, self._barge_in)
+                    return
                 text = "".join(self._chunks).strip()
                 self._chunks = []
-                await self._reply_queue.put(text)
+                await self._reply_queue.put(_TurnResult(text))
                 logger.debug("[tracker] turn complete, queued reply (len=%d)", len(text))
             else:
                 # LLM emitted an intermediate response before tool results arrive.
@@ -94,12 +269,19 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
 
         await self.push_frame(frame, direction)
 
+    async def next_turn(self, timeout: float = DEFAULT_TURN_TIMEOUT) -> _TurnResult:
+        """Wait for the next completed (or interrupted) turn.
+
+        Raises ``TimeoutError`` if no reply arrives within *timeout* seconds.
+        """
+        return await asyncio.wait_for(self._reply_queue.get(), timeout=timeout)
+
     async def next_reply(self, timeout: float = DEFAULT_TURN_TIMEOUT) -> str:
         """Wait for the next completed turn and return the reply text.
 
         Raises ``TimeoutError`` if no reply arrives within *timeout* seconds.
         """
-        return await asyncio.wait_for(self._reply_queue.get(), timeout=timeout)
+        return (await self.next_turn(timeout=timeout)).text
 
 
 class CatnipSession:
@@ -109,10 +291,17 @@ class CatnipSession:
     transcript of the exchange is accumulated in ``transcript``.
     """
 
-    def __init__(self, worker: PipelineWorker, tracker: CatnipTurnTracker) -> None:
+    def __init__(
+        self,
+        worker: PipelineWorker,
+        tracker: CatnipTurnTracker,
+        assistant_aggregator: LLMAssistantAggregator | None = None,
+    ) -> None:
         self._worker = worker
         self._tracker = tracker
+        self._assistant_aggregator = assistant_aggregator
         self.transcript: list[tuple[TurnLogKind, str]] = []
+        self.last_reply_interrupted: bool = False
 
     @property
     def log(self) -> list[tuple[TurnLogKind, str]]:
@@ -124,14 +313,46 @@ def formatted_log(self) -> str:
         """Return a human-readable formatted transcript."""
         return "\n".join(f"[{kind.name}] {text}" for kind, text in self.transcript)
 
-    async def send(self, text: str, timeout: float = DEFAULT_TURN_TIMEOUT) -> str:
+    async def verify_interrupted_context(self, spoken: str, timeout: float = 2.0) -> str | None:
+        """Check the LLM context ends with exactly the *spoken* part of an interrupted reply.
+
+        Returns ``None`` when it does (or there is no assistant aggregator to inspect), otherwise
+        a description of what the context holds instead. The aggregator commits on the
+        interruption asynchronously, so this polls briefly.
+        """
+        if self._assistant_aggregator is None:
+            return None
+        expected = " ".join(spoken.split())
+        found: str | None = None
+        try:
+            async with asyncio.timeout(timeout):
+                while True:
+                    found = _last_assistant_text(self._assistant_aggregator)
+                    if found == expected:
+                        return None
+                    await asyncio.sleep(0.02)
+        except TimeoutError:
+            return f"the last assistant message in the LLM context is {found!r}, expected {expected!r}"
+
+    async def send(
+        self,
+        text: str,
+        timeout: float = DEFAULT_TURN_TIMEOUT,
+        *,
+        interrupt_reply_with: BargeIn | None = None,
+    ) -> str:
         """Inject *text* as a user utterance and block until the bot replies.
 
-        Returns the bot's reply text.
+        If *interrupt_reply_with* is given, the bot's reply is cut at that trigger (see
+        ``CatnipTurnTracker.arm_barge_in``) and ``last_reply_interrupted`` is set. Raises
+        ``BargeInMissedError`` if the reply ended before the trigger.
+
+        Returns the bot's reply text (the truncated text when interrupted).
         """
         t0 = time.monotonic()
         logger.debug("[session] USER -> %r", text)
         self.transcript.append((TurnLogKind.USER, text))
+        self._tracker.arm_barge_in(interrupt_reply_with)
         self._tracker.prepare_for_input()
         tools_before = len(self._tracker.tool_calls)
         await self._worker.queue_frames(
@@ -140,11 +361,15 @@ async def send(self, text: str, timeout: float = DEFAULT_TURN_TIMEOUT) -> str:
                 UserStoppedSpeakingFrame(),
             ]
         )
-        reply = await self._tracker.next_reply(timeout=timeout)
+        turn = await self._tracker.next_turn(timeout=timeout)
+        reply = turn.text
+        self.last_reply_interrupted = turn.interrupted
         elapsed = time.monotonic() - t0
         logger.debug("[session] BOT  <- (%.1fs) %r", elapsed, reply)
-        self.transcript.append((TurnLogKind.BOT, reply))
+        self.transcript.append((TurnLogKind.BOT, f"{reply} [interrupted by user]" if turn.interrupted else reply))
         for call in self._tracker.tool_calls[tools_before:]:
             entry = f"{call.name} {call.arguments}" if call.arguments else call.name
             self.transcript.append((TurnLogKind.TOOL, entry))
+        if turn.barge_in_miss is not None and interrupt_reply_with is not None:
+            raise BargeInMissedError(reply, interrupt_reply_with, turn.barge_in_miss)
         return reply
diff --git a/src/pytest_catnip/models.py b/src/pytest_catnip/models.py
index e7dc72f..63f7223 100644
--- a/src/pytest_catnip/models.py
+++ b/src/pytest_catnip/models.py
@@ -1,7 +1,8 @@
 import pathlib
+import re
 from typing import Annotated, Any
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 
 class ToolExpectation(BaseModel):
@@ -29,15 +30,56 @@ def _coerce_shorthand(cls, value: Any) -> Any:  # noqa: ANN401
         return value
 
 
+class BargeIn(BaseModel):
+    """Interrupt the bot's previous reply mid-stream, then send this phase's utterance.
+
+    Exactly one trigger must be set. ``after`` fires at the first streamed chunk where the
+    accumulated bot text matches the regex (case-insensitive); ``after_words`` fires once the
+    reply reaches that many words. The reply is cut right after the trigger.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    after: str | None = None
+    after_words: Annotated[int, Field(ge=1)] | None = None
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_shorthand(cls, value: Any) -> Any:  # noqa: ANN401
+        # ``barge_in: true`` interrupts after the first word.
+        if value is True:
+            return {"after_words": 1}
+        return value
+
+    @model_validator(mode="after")
+    def _validate_trigger(self) -> "BargeIn":
+        if (self.after is None) == (self.after_words is None):
+            raise ValueError("barge_in needs exactly one of 'after' or 'after_words' (or use 'barge_in: true')")
+        if self.after is not None:
+            if not self.after.strip():
+                raise ValueError("barge_in.after must not be empty")
+            try:
+                re.compile(self.after)
+            except re.error as e:
+                raise ValueError(f"barge_in.after is not a valid regex ({self.after!r}): {e}") from None
+        return self
+
+
 class CatnipTestPhaseData(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     send: str
+    barge_in: BargeIn | None = None
     expect_tools: list[ToolExpectation] | None = None
     expect_reply_contains: list[str] = []
     expect_llm_judge: list[str] | None = None
     expect_flow_state: str | None = None
 
+    @field_validator("barge_in", mode="before")
+    @classmethod
+    def _barge_in_false_is_none(cls, value: Any) -> Any:  # noqa: ANN401
+        return None if value is False else value
+
 
 class ReliabilityConfig(BaseModel):
     model_config = ConfigDict(extra="forbid")
@@ -65,3 +107,9 @@ class CatnipTestCaseData(BaseModel):
     # Expectations
     expect_flow_nodes: list[str] | None = None
     expect_llm_judge: list[str] | None = None
+
+    @model_validator(mode="after")
+    def _validate_barge_in_position(self) -> "CatnipTestCaseData":
+        if self.phases and self.phases[0].barge_in is not None:
+            raise ValueError("phases -> 0 -> barge_in: the first phase has no previous bot reply to interrupt")
+        return self
diff --git a/src/pytest_catnip/runner.py b/src/pytest_catnip/runner.py
index 1418595..70d5341 100644
--- a/src/pytest_catnip/runner.py
+++ b/src/pytest_catnip/runner.py
@@ -8,11 +8,13 @@
 from typing import Any
 
 import pytest
+from pipecat.frames.frames import InterruptionFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.worker import PipelineWorker
+from pipecat.processors.aggregators.llm_response_universal import LLMAssistantAggregator
 from pipecat.workers.runner import WorkerRunner
 from pytest_catnip.flows import CatnipFlowBundle, CatnipFlowTracker
-from pytest_catnip.harness import CatnipSession, CatnipTurnTracker, TurnLogKind, _ToolCall
+from pytest_catnip.harness import BargeInMissedError, CatnipSession, CatnipTurnTracker, TurnLogKind, _ToolCall
 from pytest_catnip.models import CatnipTestCaseData, CatnipTestPhaseData, ToolExpectation
 
 logger = logging.getLogger(__name__)
@@ -123,7 +125,10 @@ async def initialize_test(self, case_data: CatnipTestCaseData) -> _TestRunnerEle
         )
 
         worker = PipelineWorker(pipeline)
-        session = CatnipSession(worker=worker, tracker=turn_tracker)
+        assistant_aggregator = next((p for p in pipeline.processors if isinstance(p, LLMAssistantAggregator)), None)
+        turn_tracker.watch_assistant_aggregator(assistant_aggregator)
+        turn_tracker.set_interrupter(lambda: worker.queue_frame(InterruptionFrame()))
+        session = CatnipSession(worker=worker, tracker=turn_tracker, assistant_aggregator=assistant_aggregator)
 
         worker_runner = WorkerRunner(handle_sigint=False)
         await worker_runner.add_workers(worker)
@@ -327,6 +332,7 @@ async def run_once(
                     reply, triggered_calls = await self._execute_phase_exchanges(
                         phase=phase,
                         phase_idx=phase_idx,
+                        next_phase=case_data.phases[phase_idx + 1] if phase_idx + 1 < len(case_data.phases) else None,
                         case_data=case_data,
                         session=runner_elems.session,
                         tracker=runner_elems.turn_tracker,
@@ -420,20 +426,45 @@ async def _execute_phase_exchanges(
         *,
         phase: CatnipTestPhaseData,
         phase_idx: int,
+        next_phase: CatnipTestPhaseData | None,
         case_data: CatnipTestCaseData,
         session: CatnipSession,
         tracker: CatnipTurnTracker,
     ) -> tuple[str, list[_ToolCall]]:
         """Send the phase utterance, auto-confirming bot questions up to the budget.
 
+        If the next phase has a ``barge_in``, the bot's first reply is interrupted at its trigger
+        and auto-confirm is skipped.
+
         Returns ``(final_reply, triggered_tool_calls)``.
         """
         tools_snapshot = len(tracker.tool_calls)
-        reply = await session.send(phase.send, timeout=case_data.turn_timeout)
+        barge_in = next_phase.barge_in if next_phase is not None else None
+        try:
+            reply = await session.send(phase.send, timeout=case_data.turn_timeout, interrupt_reply_with=barge_in)
+        except BargeInMissedError as e:
+            raise AssertionError(
+                f"\n[{case_data.source_path}] Phase {phase_idx + 1}: barge_in could not interrupt phase {phase_idx}'s reply.\n"
+                f"  Reason:    {e.reason}\n"
+                f"  Bot reply: {e.reply}\n"
+                f"  Hint: the reply must run past the trigger point; adjust barge_in or the prompt."
+            ) from None
         logger.debug("[%s] phase=%d BOT: %s", case_data.name, phase_idx, reply)
 
+        if session.last_reply_interrupted:
+            problem = await session.verify_interrupted_context(reply)
+            if problem is not None:
+                raise AssertionError(
+                    f"\n[{case_data.source_path}] Phase {phase_idx}: the interruption did not truncate the LLM context.\n"
+                    f"  Reason: {problem}"
+                )
+
         confirm_count = 0
-        while confirm_count < case_data.max_auto_confirm and self.is_confirmation_question(reply):
+        while (
+            not session.last_reply_interrupted
+            and confirm_count < case_data.max_auto_confirm
+            and self.is_confirmation_question(reply)
+        ):
             confirm_count += 1
             logger.debug(
                 "[%s] phase=%d auto-confirm %d/%d -> %r",
diff --git a/tests/cases/barge_in/__init__.py b/tests/cases/barge_in/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/cases/barge_in/conftest.py b/tests/cases/barge_in/conftest.py
new file mode 100644
index 0000000..0ae80df
--- /dev/null
+++ b/tests/cases/barge_in/conftest.py
@@ -0,0 +1,67 @@
+"""Pipeline and judge fixtures for barge-in test cases.
+
+Scoped to tests/cases/barge_in/. The bot is told to give long, enumerated answers so there is
+always something left to interrupt.
+"""
+
+import os
+from collections.abc import Awaitable, Callable
+
+import pytest
+from dotenv import load_dotenv
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.services.llm_service import FunctionCallParams
+from pytest_catnip import CatnipTurnTracker
+from pytest_catnip.judges import gemini_judge
+
+load_dotenv()
+
+
+@pytest.fixture
+def catnip_pipeline() -> Callable[[], Pipeline]:
+    def factory() -> Pipeline:
+        price_tool = FunctionSchema(
+            name="lookup_price",
+            description="Look up the price of a pizza size.",
+            properties={"size": {"type": "string", "enum": ["small", "medium", "large"]}},
+            required=["size"],
+        )
+        context = LLMContext(
+            messages=[
+                {
+                    "role": "system",
+                    "content": (
+                        "You are a verbose assistant. When asked to explain or list something, answer with "
+                        "at least six full sentences, numbering the points 'Point one', 'Point two', and so on. "
+                        "If a message of yours ends abruptly mid-sentence, the user interrupted you: never resume, "
+                        "repeat or summarise that earlier answer, and answer only the user's new message, briefly. "
+                        "When asked for a price, call lookup_price."
+                    ),
+                }
+            ],
+            tools=ToolsSchema(standard_tools=[price_tool]),
+        )
+        llm = GoogleLLMService(
+            api_key=os.environ["GOOGLE_API_KEY"],
+            settings=GoogleLLMService.Settings(model="gemini-2.5-flash"),
+        )
+
+        async def handle_lookup_price(params: FunctionCallParams) -> None:
+            await params.result_callback({"price_eur": 12})
+
+        llm.register_function("lookup_price", handle_lookup_price)
+
+        ctx_agg = LLMContextAggregatorPair(context)
+        return Pipeline([ctx_agg.user(), llm, CatnipTurnTracker(), ctx_agg.assistant()])
+
+    return factory
+
+
+@pytest.fixture(scope="session")
+def llm_judge_function() -> Callable[[str, str], Awaitable[tuple[bool, str]]]:
+    return gemini_judge(api_key=os.getenv("GOOGLE_API_KEY", ""), model="gemini-2.5-flash")
diff --git a/tests/cases/barge_in/test_barge_in_after_regex.yaml b/tests/cases/barge_in/test_barge_in_after_regex.yaml
new file mode 100644
index 0000000..695d705
--- /dev/null
+++ b/tests/cases/barge_in/test_barge_in_after_regex.yaml
@@ -0,0 +1,19 @@
+name: barge-in-after-regex
+description: >
+  The user interrupts once the bot has started its second point. The first reply
+  keeps everything up to and including the match and nothing after it.
+
+phases:
+  - send: "List the benefits of exercise."
+    expect_reply_contains:
+      - "point one"
+      - "point two$"
+
+  - send: "Sorry, quick question: what colour is the sky on a clear day?"
+    barge_in:
+      after: "point two"
+    expect_reply_contains:
+      - "blue"
+
+expect_llm_judge:
+  - "The bot never gets to 'Point three' in its first reply."
diff --git a/tests/cases/barge_in/test_barge_in_after_words.yaml b/tests/cases/barge_in/test_barge_in_after_words.yaml
new file mode 100644
index 0000000..fe0697a
--- /dev/null
+++ b/tests/cases/barge_in/test_barge_in_after_words.yaml
@@ -0,0 +1,18 @@
+name: barge-in-after-words
+description: >
+  The user interrupts a long explanation after a few words and asks something
+  unrelated. The bot must drop the explanation and answer the new question.
+
+phases:
+  - send: "Explain in detail how photosynthesis works."
+
+  - send: "Stop. What is 2 + 2?"
+    barge_in:
+      after_words: 8
+    expect_reply_contains:
+      - "4|four"
+    expect_llm_judge:
+      - "The bot answers the arithmetic question instead of continuing the photosynthesis explanation."
+
+expect_llm_judge:
+  - "The first bot reply is cut off mid-explanation and never reaches 'Point three'."
diff --git a/tests/cases/barge_in/test_barge_in_context_truncated.yaml b/tests/cases/barge_in/test_barge_in_context_truncated.yaml
new file mode 100644
index 0000000..4396177
--- /dev/null
+++ b/tests/cases/barge_in/test_barge_in_context_truncated.yaml
@@ -0,0 +1,14 @@
+name: barge-in-context-truncated
+description: >
+  The interrupted reply must be the last thing the bot said, as far as its own
+  context goes: the harness verifies the assistant message holds only the spoken
+  part, and the bot must be able to refer to it.
+
+phases:
+  - send: "Count from one to twenty in words, separated by commas."
+
+  - send: "Stop. What was the last number you said?"
+    barge_in:
+      after: "five"
+    expect_llm_judge:
+      - "The bot says the last number it said was five, and not a larger number."
diff --git a/tests/cases/barge_in/test_barge_in_no_resume.yaml b/tests/cases/barge_in/test_barge_in_no_resume.yaml
new file mode 100644
index 0000000..0c7e85e
--- /dev/null
+++ b/tests/cases/barge_in/test_barge_in_no_resume.yaml
@@ -0,0 +1,16 @@
+name: barge-in-no-resume
+description: >
+  After being interrupted the bot must not resume or repeat what it was saying.
+  Judged on the whole conversation, which marks the cut reply as interrupted.
+
+phases:
+  - send: "Give me a detailed overview of the Roman Empire."
+
+  - send: "Hold on, what's the capital of France?"
+    barge_in:
+      after: "point two"
+    expect_reply_contains:
+      - "paris"
+
+expect_llm_judge:
+  - "After the interruption the bot answers about France and does not resume or repeat the Roman Empire overview."
diff --git a/tests/cases/barge_in/test_barge_in_then_tool.yaml b/tests/cases/barge_in/test_barge_in_then_tool.yaml
new file mode 100644
index 0000000..d6849a2
--- /dev/null
+++ b/tests/cases/barge_in/test_barge_in_then_tool.yaml
@@ -0,0 +1,17 @@
+name: barge-in-then-tool
+description: >
+  The user cuts the bot off mid-answer and immediately asks for something that needs
+  a tool. The bot must call the tool for the new request.
+
+phases:
+  - send: "Tell me all about how pizza dough is made."
+
+  - send: "Actually, how much is a large pizza?"
+    barge_in:
+      after_words: 6
+    expect_tools:
+      - name: lookup_price
+        args:
+          size: large
+    expect_reply_contains:
+      - "12"
diff --git a/tests/cases/barge_in/test_barge_in_true.yaml b/tests/cases/barge_in/test_barge_in_true.yaml
new file mode 100644
index 0000000..f305ae6
--- /dev/null
+++ b/tests/cases/barge_in/test_barge_in_true.yaml
@@ -0,0 +1,12 @@
+name: barge-in-true
+description: The shorthand `barge_in true` interrupts right after the first word.
+
+phases:
+  - send: "Tell me about the moon."
+    expect_reply_contains:
+      - "^\\S+$"
+
+  - send: "Never mind. What is 3 + 5?"
+    barge_in: true
+    expect_reply_contains:
+      - "8|eight"
diff --git a/tests/cases/llm_judge/test_llm_judge_conversation.yaml b/tests/cases/llm_judge/test_llm_judge_conversation.yaml
index a10cf29..5c3c883 100644
--- a/tests/cases/llm_judge/test_llm_judge_conversation.yaml
+++ b/tests/cases/llm_judge/test_llm_judge_conversation.yaml
@@ -3,8 +3,8 @@ description: Tests the integration of LLM-based judge assertions for whole conve
 skip: false
 
 expect_llm_judge:
-  - "The bot should provide interesting facts about dogs"
-  - "The bot should provide interesting facts about cats"
+  - "The bot should provide interesting facts about dogs at some point"
+  - "The bot should provide interesting facts about cats at some point"
   - "The bot never provides any information about dungeons and dragons"
 
 phases:
diff --git a/tests/test_barge_in.py b/tests/test_barge_in.py
new file mode 100644
index 0000000..79e5ac4
--- /dev/null
+++ b/tests/test_barge_in.py
@@ -0,0 +1,137 @@
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+from pytest_catnip.harness import CatnipSession, _find_cut
+from pytest_catnip.models import BargeIn, CatnipTestCaseData, CatnipTestPhaseData
+
+# ---------------------------------------------------------------- models
+
+
+def _case(*phases: dict[str, Any]) -> CatnipTestCaseData:
+    return CatnipTestCaseData(name="x", phases=list(phases), source_path="x.yaml")  # type: ignore[arg-type]
+
+
+def test_barge_in_after_regex() -> None:
+    assert BargeIn(after="foo|bar").after == "foo|bar"
+
+
+def test_barge_in_after_words() -> None:
+    assert BargeIn(after_words=3).after_words == 3
+
+
+def test_barge_in_true_means_first_word() -> None:
+    phase = CatnipTestPhaseData(send="hi", barge_in=True)  # type: ignore[arg-type]
+    assert phase.barge_in == BargeIn(after_words=1)
+
+
+def test_barge_in_false_and_missing_are_none() -> None:
+    assert CatnipTestPhaseData(send="hi", barge_in=False).barge_in is None  # type: ignore[arg-type]
+    assert CatnipTestPhaseData(send="hi").barge_in is None
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "message"),
+    [
+        ({"after": "x", "after_words": 2}, "exactly one"),
+        ({}, "exactly one"),
+        ({"after_words": 0}, "greater than or equal to 1"),
+        ({"after_words": -3}, "greater than or equal to 1"),
+        ({"after": "("}, "not a valid regex"),
+        ({"after": "   "}, "must not be empty"),
+        ({"after": ""}, "must not be empty"),
+    ],
+)
+def test_barge_in_invalid(kwargs: dict[str, Any], message: str) -> None:
+    with pytest.raises(ValidationError, match=message):
+        BargeIn(**kwargs)
+
+
+def test_barge_in_unknown_key() -> None:
+    with pytest.raises(ValidationError, match="Extra inputs"):
+        BargeIn(after_word=3)  # type: ignore[call-arg]
+
+
+def test_barge_in_on_first_phase_rejected() -> None:
+    with pytest.raises(ValidationError, match="first phase"):
+        _case({"send": "a", "barge_in": True})
+
+
+def test_barge_in_on_later_phase_accepted() -> None:
+    case = _case({"send": "a"}, {"send": "b", "barge_in": {"after_words": 2}})
+    assert case.phases[1].barge_in == BargeIn(after_words=2)
+
+
+# ---------------------------------------------------------------- _find_cut
+
+
+def test_find_cut_regex_is_case_insensitive_and_cuts_at_match_end() -> None:
+    assert _find_cut(BargeIn(after="hello"), "Oh HELLO there", at_end=False) == len("Oh HELLO")
+
+
+def test_find_cut_regex_not_found() -> None:
+    assert _find_cut(BargeIn(after="zzz"), "Oh hello there", at_end=True) is None
+
+
+def test_find_cut_words_waits_for_whitespace_mid_stream() -> None:
+    barge_in = BargeIn(after_words=2)
+    assert _find_cut(barge_in, "one tw", at_end=False) is None
+    assert _find_cut(barge_in, "one two", at_end=False) is None  # "two" could still grow
+    assert _find_cut(barge_in, "one two ", at_end=False) == len("one two")
+
+
+def test_find_cut_words_at_end_counts_last_word() -> None:
+    assert _find_cut(BargeIn(after_words=2), "one two", at_end=True) == len("one two")
+
+
+def test_find_cut_words_too_short() -> None:
+    assert _find_cut(BargeIn(after_words=5), "one two", at_end=True) is None
+
+
+class _StubContext:
+    def __init__(self, messages: list[dict[str, Any]]) -> None:
+        self._messages = messages
+
+    def get_messages(self) -> list[dict[str, Any]]:
+        return self._messages
+
+
+class _StubAggregator:
+    def __init__(self, messages: list[dict[str, Any]]) -> None:
+        self.context = _StubContext(messages)
+
+
+def _session_with_context(messages: list[dict[str, Any]]) -> CatnipSession:
+    return CatnipSession(
+        worker=None,  # type: ignore[arg-type]
+        tracker=None,  # type: ignore[arg-type]
+        assistant_aggregator=_StubAggregator(messages),  # type: ignore[arg-type]
+    )
+
+
+@pytest.mark.asyncio
+async def test_verify_interrupted_context_accepts_spoken_text() -> None:
+    session = _session_with_context([{"role": "assistant", "content": "one,  two\nthree"}])
+    assert await session.verify_interrupted_context("one, two three", timeout=0.1) is None
+
+
+@pytest.mark.asyncio
+async def test_verify_interrupted_context_accepts_content_parts() -> None:
+    session = _session_with_context([{"role": "assistant", "content": [{"type": "text", "text": "one two"}]}])
+    assert await session.verify_interrupted_context("one two", timeout=0.1) is None
+
+
+@pytest.mark.asyncio
+async def test_verify_interrupted_context_reports_missing_assistant_message() -> None:
+    session = _session_with_context([{"role": "user", "content": "hi"}])
+    problem = await session.verify_interrupted_context("one two", timeout=0.1)
+    assert problem is not None
+    assert "None" in problem
+
+
+@pytest.mark.asyncio
+async def test_verify_interrupted_context_reports_untruncated_text() -> None:
+    session = _session_with_context([{"role": "assistant", "content": "one two three four"}])
+    problem = await session.verify_interrupted_context("one two", timeout=0.1)
+    assert problem is not None
+    assert "one two three four" in problem
diff --git a/tests/test_barge_in_processors.py b/tests/test_barge_in_processors.py
new file mode 100644
index 0000000..53609a1
--- /dev/null
+++ b/tests/test_barge_in_processors.py
@@ -0,0 +1,85 @@
+"""Barge-in as seen by custom frame processors, run through catnip's own runner with the real LLM pipeline."""
+
+import os
+import pathlib
+from collections.abc import Callable
+from typing import override
+
+import pytest
+from dotenv import load_dotenv
+from pipecat.frames.frames import Frame, InterruptionFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.google.llm import GoogleLLMService
+from pytest_catnip import CatnipTurnTracker
+from pytest_catnip.models import CatnipTestCaseData
+from pytest_catnip.runner import CatnipAssertionManager, CatnipTestInitializer, CatnipTestRunner
+
+load_dotenv()
+
+_CASE = CatnipTestCaseData(
+    name="barge-in-processors",
+    source_path=pathlib.Path("barge-in-processors.yaml"),
+    phases=[
+        {"send": "Explain in detail how volcanoes form."},  # type: ignore[list-item]
+        {"send": "Stop. What is 2 + 2?", "barge_in": {"after_words": 5}},  # type: ignore[list-item]
+    ],
+)
+
+
+class FrameRecorder(FrameProcessor):
+    """Records every frame of interest with the direction it travelled."""
+
+    def __init__(self, log: list[tuple[str, FrameDirection]]) -> None:
+        super().__init__()
+        self._log = log
+
+    @override
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+        if isinstance(frame, InterruptionFrame):
+            self._log.append((type(frame).__name__, direction))
+        await self.push_frame(frame, direction)
+
+
+def _pipeline_factory(before_llm: FrameProcessor) -> Callable[[], Pipeline]:
+    def factory() -> Pipeline:
+        context = LLMContext(
+            messages=[{"role": "system", "content": "You are a verbose assistant. Explain things in 6+ sentences."}]
+        )
+        llm = GoogleLLMService(
+            api_key=os.environ["GOOGLE_API_KEY"],
+            settings=GoogleLLMService.Settings(model="gemini-2.5-flash"),
+        )
+        user_agg, assistant_agg = LLMContextAggregatorPair(context)
+        tracker = CatnipTurnTracker()
+        return Pipeline([user_agg, before_llm, llm, tracker, assistant_agg])
+
+    return factory
+
+
+async def _run(factory: Callable[[], Pipeline]) -> None:
+    async def judge(_: str, __: str) -> tuple[bool, str]:
+        return True, ""
+
+    runner = CatnipTestRunner(
+        test_initializer=CatnipTestInitializer(pipeline_factory=factory),
+        assertion_manager=CatnipAssertionManager(judge),
+        timeout=60,
+        autoconfirm_response="Yes.",
+        is_confirmation_question=lambda _: False,
+    )
+    _, exc = await runner.run_once(_CASE)
+    if exc is not None:
+        raise exc
+
+
+@pytest.mark.asyncio
+async def test_interruption_enters_at_the_source() -> None:
+    before_llm: list[tuple[str, FrameDirection]] = []
+    await _run(_pipeline_factory(FrameRecorder(before_llm)))
+
+    # A processor in front of the LLM sees the interruption flowing downstream, like from a real transport.
+    assert ("InterruptionFrame", FrameDirection.DOWNSTREAM) in before_llm
diff --git a/tests/test_collector.py b/tests/test_collector.py
index 5dca4f8..e111bf7 100644
--- a/tests/test_collector.py
+++ b/tests/test_collector.py
@@ -100,3 +100,39 @@ def catnip_pipeline():
     result = pytester.runpytest()
     assert result.ret == pytest.ExitCode.TESTS_FAILED
     result.stdout.fnmatch_lines(["*CatnipTurnTracker*"])
+
+
+@pytest.mark.parametrize(
+    ("barge_in", "message"),
+    [
+        ("{after: foo, after_words: 3}", "exactly one"),
+        ("{}", "exactly one"),
+        ("{after_words: 0}", "greater than or equal to 1"),
+        ("{after: '('}", "not a valid regex"),
+        ("{after_word: 3}", "Unknown field"),
+    ],
+)
+def test_invalid_barge_in_reports_usage_error(
+    cases_dir: pathlib.Path, pytester: pytest.Pytester, barge_in: str, message: str
+) -> None:
+    (cases_dir / "test_bad.yaml").write_text(
+        f"name: bad\nphases:\n  - send: hi\n  - send: stop\n    barge_in: {barge_in}\n"
+    )
+    result = pytester.runpytest("--collect-only")
+    assert result.ret == pytest.ExitCode.INTERRUPTED
+    result.stdout.fnmatch_lines([f"*phases -> 1 -> barge_in*{message}*"])
+
+
+def test_barge_in_on_first_phase_reports_usage_error(cases_dir: pathlib.Path, pytester: pytest.Pytester) -> None:
+    (cases_dir / "test_bad.yaml").write_text("name: bad\nphases:\n  - send: hi\n    barge_in: true\n")
+    result = pytester.runpytest("--collect-only")
+    assert result.ret == pytest.ExitCode.INTERRUPTED
+    result.stdout.fnmatch_lines(["*first phase*"])
+
+
+def test_valid_barge_in_collects(cases_dir: pathlib.Path, pytester: pytest.Pytester) -> None:
+    (cases_dir / "test_ok.yaml").write_text(
+        "name: ok\nphases:\n  - send: hi\n  - send: stop\n    barge_in:\n      after_words: 4\n"
+    )
+    result = pytester.runpytest("--collect-only")
+    assert result.ret == pytest.ExitCode.OK