diff --git a/client-react/src/conversation/PipecatConversationProvider.tsx b/client-react/src/conversation/PipecatConversationProvider.tsx index ff305d4..47d31fd 100644 --- a/client-react/src/conversation/PipecatConversationProvider.tsx +++ b/client-react/src/conversation/PipecatConversationProvider.tsx @@ -31,15 +31,32 @@ export const ConversationContext = export const PipecatConversationProvider: React.FC = ({ children, }) => { - useConversationEventWiring(); + const { finalizeLastAssistantMessageIfPending } = + useConversationEventWiring(); const injectMessage = useAtomCallback( useCallback((get, set, message: { role: "user" | "assistant" | "system"; parts: ConversationMessagePart[]; }) => { + // An injected message is a turn boundary the RTVI events never report: + // text input reaches the bot through `sendText`, so there is no + // UserStartedSpeaking to close the assistant's turn, and the bot was + // likely mid-utterance, so the BotStoppedSpeaking finalize timer is not + // armed either. Without this the next BotOutput reopens the still-open + // message and the following turn is appended to the previous one. + // + // System messages are excluded: `injectMessage` deliberately backdates + // them behind an in-flight assistant message so they don't split it. + // + // Finalizing here mirrors the UserStartedSpeaking path, which likewise + // leaves the speech cursor where it stopped — the turn was interrupted, + // so unspoken text must stay unspoken. + if (message.role !== "system") { + finalizeLastAssistantMessageIfPending(); + } injectMessageAction(get, set, message); - }, []) + }, [finalizeLastAssistantMessageIfPending]) ); const botOutputSupported = useAtomValue(botOutputSupportedAtom); diff --git a/client-react/src/conversation/useConversationEventWiring.ts b/client-react/src/conversation/useConversationEventWiring.ts index bca5505..bc694a3 100644 --- a/client-react/src/conversation/useConversationEventWiring.ts +++ b/client-react/src/conversation/useConversationEventWiring.ts @@ -437,4 +437,9 @@ export function useConversationEventWiring() { }, []) ) ); + + // Exposed so a caller-driven turn boundary — a message injected into the + // conversation, which no RTVI event announces — can end the assistant turn + // the same way UserStartedSpeaking does. + return { finalizeLastAssistantMessageIfPending }; } diff --git a/client-react/tests/conversation/integration/eventWiring.test.tsx b/client-react/tests/conversation/integration/eventWiring.test.tsx index 5e90f0e..7d333a3 100644 --- a/client-react/tests/conversation/integration/eventWiring.test.tsx +++ b/client-react/tests/conversation/integration/eventWiring.test.tsx @@ -16,9 +16,18 @@ import { RTVIEvent } from "@pipecat-ai/client-js"; import { act, render } from "@testing-library/react"; import { createStore, Provider } from "jotai"; -import { messagesAtom } from "@/conversation/conversationAtoms"; -import { PipecatConversationProvider } from "@/conversation/PipecatConversationProvider"; -import type { ConversationMessage } from "@/conversation/types"; +import { + botOutputMessageStateAtom, + messagesAtom, +} from "@/conversation/conversationAtoms"; +import { + PipecatConversationProvider, + useConversationContext, +} from "@/conversation/PipecatConversationProvider"; +import type { + ConversationMessage, + ConversationMessagePart, +} from "@/conversation/types"; import { RTVIEventContext } from "@/RTVIEventContext"; /** @@ -41,6 +50,15 @@ function renderWiring() { handlers.get(event)?.delete(handler); }; + let injectMessage: ReturnType< + typeof useConversationContext + >["injectMessage"]; + + const CaptureContext = () => { + injectMessage = useConversationContext().injectMessage; + return null; + }; + render( - {null} + + + ); @@ -69,14 +89,31 @@ function renderWiring() { }); }; + const getMessages = () => store.get(messagesAtom); + + const inject = ( + role: "user" | "assistant" | "system", + parts: ConversationMessagePart[] + ) => { + act(() => { + injectMessage({ role, parts }); + }); + }; + return { emit, advance, - getMessages: () => store.get(messagesAtom), + inject, + getMessages, getAssistantMessages: () => - store - .get(messagesAtom) - .filter((m: ConversationMessage) => m.role === "assistant"), + getMessages().filter((m: ConversationMessage) => m.role === "assistant"), + getLastAssistantCursor: () => { + const lastAssistant = [...getMessages()] + .reverse() + .find((m: ConversationMessage) => m.role === "assistant"); + if (!lastAssistant) return undefined; + return store.get(botOutputMessageStateAtom).get(lastAssistant.createdAt); + }, }; } @@ -208,6 +245,86 @@ describe("useConversationEventWiring", () => { }); }); + describe("RTVI 2.0.0+ text-input turn boundaries", () => { + const userText = (text: string): ConversationMessagePart[] => [ + { text, final: true, createdAt: new Date().toISOString() }, + ]; + + /** + * Text input reaches the bot through `sendText`, which produces no + * UserStartedSpeaking, and the bot is mid-utterance, so no finalize timer + * is armed. Injecting the user's message is the only turn boundary the + * conversation ever sees. + */ + function startInterruptedV2Turn() { + const w = renderWiring(); + w.emit(RTVIEvent.BotReady, { version: "2.1.0" }); + w.emit(RTVIEvent.BotStartedSpeaking); + w.emit( + RTVIEvent.BotOutput, + sentence("Hi there, how can I help you today?", 1, { + spoken_status: "new", + }) + ); + w.emit( + RTVIEvent.BotOutput, + sentence("Hi there, how can I help you today?", 1, { + spoken_status: "in-progress", + spoken_progress: { + accumulated_text: "Hi there,", + remaining_text: " how can I help you today?", + }, + }) + ); + return w; + } + + it("finalizes the open turn when a user message is injected", () => { + const w = startInterruptedV2Turn(); + expect(w.getAssistantMessages()[0].final).toBeFalsy(); + + w.inject("user", userText("actually, never mind")); + + expect(w.getAssistantMessages()[0].final).toBe(true); + }); + + it("opens a new message for the reply to injected text", () => { + const w = startInterruptedV2Turn(); + w.inject("user", userText("actually, never mind")); + + w.emit(RTVIEvent.BotStartedSpeaking); + w.emit( + RTVIEvent.BotOutput, + sentence("No problem.", 2, { + spoken_status: "new", + }) + ); + + const assistant = w.getAssistantMessages(); + expect(assistant).toHaveLength(2); + expect(assistant[1].parts.map((p) => p.text)).toEqual(["No problem."]); + }); + + it("leaves the speech cursor where the interruption stopped it", () => { + const w = startInterruptedV2Turn(); + w.inject("user", userText("actually, never mind")); + + // Finalizing must not snap the cursor to the end: the turn was cut off, + // so the unspoken tail stays unspoken. + expect(w.getLastAssistantCursor()!.currentCharIndex).toBe( + "Hi there,".length + ); + }); + + it("does not finalize the turn for an injected system message", () => { + const w = startInterruptedV2Turn(); + w.inject("system", userText("connection is unstable")); + + expect(w.getAssistantMessages()[0].final).toBeFalsy(); + expect(w.getAssistantMessages()).toHaveLength(1); + }); + }); + describe("legacy 1.4.x path", () => { it("still finalizes per sentence", () => { const w = renderWiring();