diff --git a/app/api/session/dispatch/route.ts b/app/api/session/dispatch/route.ts index 982817d87..59661c172 100644 --- a/app/api/session/dispatch/route.ts +++ b/app/api/session/dispatch/route.ts @@ -20,8 +20,8 @@ export async function POST(req: Request) { agent_name?: string; sessionId?: string; session_id?: string; - requireRoomVideoInputReady?: boolean; - require_room_video_input_ready?: boolean; + requireAgentSessionReady?: boolean; + require_agent_session_ready?: boolean; }; try { body = await req.json(); @@ -57,8 +57,8 @@ export async function POST(req: Request) { sessionId, agentName, readiness: { - requireRoomVideoInputReady: - body.requireRoomVideoInputReady === true || body.require_room_video_input_ready === true, + requireAgentSessionReady: + body.requireAgentSessionReady === true || body.require_agent_session_ready === true, }, }); return NextResponse.json({ status: 'dispatched', roomName, agentName, sessionId, dispatch }); diff --git a/app/api/session/session-dispatch-service.ts b/app/api/session/session-dispatch-service.ts index 067468acc..44075ff1d 100644 --- a/app/api/session/session-dispatch-service.ts +++ b/app/api/session/session-dispatch-service.ts @@ -75,7 +75,7 @@ const globalForInFlightDispatches = globalThis as typeof globalThis & { const inFlightDispatches = globalForInFlightDispatches.__liveavatarInFlightDispatches ?? (globalForInFlightDispatches.__liveavatarInFlightDispatches = new Map()); -const DEFAULT_AGENT_DISPATCH_TIMEOUT_MS = 8_000; +const DEFAULT_AGENT_DISPATCH_TIMEOUT_MS = 30_000; const DEFAULT_PREWARM_TOTAL_TIMEOUT_MS = 45_000; export type PrewarmPhase = 'room' | 'worker_readiness' | 'dispatch_readiness'; @@ -302,7 +302,6 @@ export async function prewarmRoomSession( ...request, readiness: { requireAgentSessionReady: true, - requireRoomInputParticipantsReady: true, }, }, { diff --git a/hooks/useChatMessages.ts b/hooks/useChatMessages.ts index e9b75f5d0..7125e826d 100644 --- a/hooks/useChatMessages.ts +++ b/hooks/useChatMessages.ts @@ -1,4 +1,4 @@ -import { useMemo } from 'react'; +import { useEffect, useMemo, useState } from 'react'; import { Room } from 'livekit-client'; import { type ReceivedChatMessage, @@ -9,6 +9,7 @@ import { } from '@livekit/components-react'; import { type AppConfig } from '@/app-config'; import { isRenderableChatMessage } from '@/lib/chat-message-filter'; +import { mergeTranscriptionHistory } from '@/lib/transcription-history'; function transcriptionToChatMessage( textStream: TextStreamData, @@ -76,10 +77,19 @@ export function useChatMessages( const chat = useChat(); const room = useRoomContext(); const transcriptions: TextStreamData[] = useTranscriptions(); + const [transcriptionHistory, setTranscriptionHistory] = useState([]); + + useEffect(() => { + setTranscriptionHistory([]); + }, [room.name]); + + useEffect(() => { + setTranscriptionHistory((previous) => mergeTranscriptionHistory(previous, transcriptions)); + }, [transcriptions]); const mergedTranscriptions = useMemo(() => { // 处理转录消息 - const transcriptionMessages = transcriptions.map((transcription) => + const transcriptionMessages = transcriptionHistory.map((transcription) => transcriptionToChatMessage(transcription, room, config) ); @@ -112,7 +122,7 @@ export function useChatMessages( ].filter(isRenderableChatMessage); return merged.sort((a, b) => a.timestamp - b.timestamp); - }, [transcriptions, chat.chatMessages, room, config]); + }, [transcriptionHistory, chat.chatMessages, room, config]); return mergedTranscriptions; } diff --git a/hooks/useRoom.ts b/hooks/useRoom.ts index d44922f1d..25e023a88 100644 --- a/hooks/useRoom.ts +++ b/hooks/useRoom.ts @@ -6,7 +6,6 @@ import { useBrowserSourceClient } from '@/hooks/useBrowserSourceClient'; import { getVoiceSessionId, resetVoiceSessionId } from '@/lib/browser-room-session'; import { readConnectionDetailsResponse } from '@/lib/connection-details-response'; import { isValidConnectionRoomId } from '@/lib/connection-room-id'; -import { usesServerRoomInputDevice } from '@/lib/input-device-config'; import { FRONTEND_EVENTS, beginFrontendObservabilitySession, @@ -27,12 +26,6 @@ import { waitForAgentSessionStop, } from '@/lib/session-stop-client'; -function requiresRoomVideoInputReady(appConfig: AppConfig) { - return appConfig.visionInputDevice - ? usesServerRoomInputDevice(appConfig.visionInputDevice) - : false; -} - export function useRoom(appConfig: AppConfig) { const aborted = useRef(false); const sessionIdRef = useRef(null); @@ -212,7 +205,6 @@ export function useRoom(appConfig: AppConfig) { await recoverFromStartError(error); }; - setIsSessionActive(true); beginFrontendObservabilitySession(room); const dispatchAgentSession = async () => { @@ -220,7 +212,7 @@ export function useRoom(appConfig: AppConfig) { dispatchSessionId = sessionId; const signal = beginAgentSessionStart(room.name, sessionId); const dispatchPromise = requestAgentSessionDispatch(appConfig.agentName, sessionId, { - requireRoomVideoInputReady: requiresRoomVideoInputReady(appConfig), + requireAgentSessionReady: usesManagedRoomInput, signal, }); registerAgentSessionDispatch(room.name, sessionId, dispatchPromise); @@ -310,6 +302,7 @@ export function useRoom(appConfig: AppConfig) { if (!usesSandboxConcurrentStartup) { await dispatchAgentSession(); } + setIsSessionActive(true); } catch (error) { await handleStartError(error); } diff --git a/lib/session-dispatch-client.ts b/lib/session-dispatch-client.ts index 0d6b7ef1b..06fa7c93d 100644 --- a/lib/session-dispatch-client.ts +++ b/lib/session-dispatch-client.ts @@ -1,6 +1,6 @@ type DispatchOptions = { signal?: AbortSignal; - requireRoomVideoInputReady?: boolean; + requireAgentSessionReady?: boolean; }; export class AgentSessionDispatchCancelledError extends Error { @@ -27,7 +27,7 @@ export async function requestAgentSessionDispatch( body: JSON.stringify({ agentName: normalizedAgentName, sessionId: normalizedSessionId, - ...(options.requireRoomVideoInputReady ? { requireRoomVideoInputReady: true } : {}), + ...(options.requireAgentSessionReady ? { requireAgentSessionReady: true } : {}), }), signal: options.signal, }); diff --git a/lib/session-dispatch-readiness.ts b/lib/session-dispatch-readiness.ts index 2c6026cc3..51cb58e47 100644 --- a/lib/session-dispatch-readiness.ts +++ b/lib/session-dispatch-readiness.ts @@ -16,6 +16,8 @@ export type ReusableAgentParticipantOptions = AgentParticipantMatchOptions & { }; export const AGENT_SESSION_READY_ATTRIBUTE = 'liveavatar.agent.session_ready'; +// livekit-server-sdk maps protobuf attribute keys to camelCase object keys. +const AGENT_SESSION_READY_ATTRIBUTE_CAMEL = 'liveavatarAgentSessionReady'; const ROOM_AUDIO_INPUT_IDENTITY = 'room_audio_input'; const ROOM_VIDEO_INPUT_IDENTITY = 'room_video_input'; @@ -127,7 +129,11 @@ function isExpectedAgentParticipant(participant: ParticipantInfo, agentName: str } function isAgentSessionReady(participant: ParticipantInfo) { - return participant.attributes?.[AGENT_SESSION_READY_ATTRIBUTE] === 'true'; + const attributes = participant.attributes ?? {}; + return ( + attributes[AGENT_SESSION_READY_ATTRIBUTE] === 'true' || + attributes[AGENT_SESSION_READY_ATTRIBUTE_CAMEL] === 'true' + ); } function isAnonymousLiveKitAgentParticipant(participant: ParticipantInfo) { diff --git a/lib/transcription-history.ts b/lib/transcription-history.ts new file mode 100644 index 000000000..2a16a8da8 --- /dev/null +++ b/lib/transcription-history.ts @@ -0,0 +1,23 @@ +import { type TextStreamData } from '@livekit/components-react'; + +const DEFAULT_TRANSCRIPTION_HISTORY_SIZE = 100; + +/** + * Keep completed text streams even when LiveKit replaces the current entry for + * the same speech segment. A tool preamble and its final answer can share one + * segment id while still arriving as distinct text streams. + */ +export function mergeTranscriptionHistory( + previous: TextStreamData[], + current: TextStreamData[], + maxEntries = DEFAULT_TRANSCRIPTION_HISTORY_SIZE +): TextStreamData[] { + if (current.length === 0) return previous; + + const byStreamId = new Map(previous.map((entry) => [entry.streamInfo.id, entry])); + current.forEach((entry) => byStreamId.set(entry.streamInfo.id, entry)); + + return Array.from(byStreamId.values()) + .sort((a, b) => a.streamInfo.timestamp - b.streamInfo.timestamp) + .slice(-maxEntries); +} diff --git a/tests/chat-message-filter.test.mjs b/tests/chat-message-filter.test.mjs index 9f3074b99..632df3fcd 100644 --- a/tests/chat-message-filter.test.mjs +++ b/tests/chat-message-filter.test.mjs @@ -3,6 +3,19 @@ import { readFile } from 'node:fs/promises'; import { test } from 'node:test'; const { isRenderableChatMessage } = await import('../lib/chat-message-filter.ts'); +const { mergeTranscriptionHistory } = await import('../lib/transcription-history.ts'); + +function transcription(id, segmentId, timestamp, text) { + return { + text, + participantInfo: { identity: 'frontdesk-agent' }, + streamInfo: { + id, + timestamp, + attributes: { 'lk.segment_id': segmentId }, + }, + }; +} test('chat message filter hides empty transcription messages', () => { assert.equal(isRenderableChatMessage({ message: '' }), false); @@ -19,3 +32,33 @@ test('chat message hook filters empty merged messages before sorting', async () assert.match(source, /isRenderableChatMessage/); assert.match(source, /\.filter\(isRenderableChatMessage\)/); }); + +test('transcription history preserves a tool preamble replaced by the final stream', () => { + const preamble = transcription('stream-preamble', 'speech-1', 100, '我查一下。'); + const final = transcription('stream-final', 'speech-1', 200, '已经设置好了。'); + + const history = mergeTranscriptionHistory([preamble], [final]); + + assert.deepEqual( + history.map(({ text }) => text), + ['我查一下。', '已经设置好了。'] + ); +}); + +test('transcription history updates partial text without duplicating one stream', () => { + const partial = transcription('stream-1', 'speech-1', 100, '我查'); + const completed = transcription('stream-1', 'speech-1', 100, '我查一下。'); + + const history = mergeTranscriptionHistory([partial], [completed]); + + assert.equal(history.length, 1); + assert.equal(history[0].text, '我查一下。'); +}); + +test('transcription history survives a transient empty snapshot', () => { + const preamble = transcription('stream-1', 'speech-1', 100, '我查一下。'); + + const history = mergeTranscriptionHistory([preamble], []); + + assert.deepEqual(history, [preamble]); +}); diff --git a/tests/session-dispatch-client.test.mjs b/tests/session-dispatch-client.test.mjs index 09bb1e45a..3e0c9a8c2 100644 --- a/tests/session-dispatch-client.test.mjs +++ b/tests/session-dispatch-client.test.mjs @@ -43,7 +43,7 @@ test('agent session dispatch sends only canonical session id to Next API', async } }); -test('agent session dispatch can require room video input readiness', async () => { +test('agent session dispatch can require authoritative session readiness', async () => { const originalFetch = globalThis.fetch; let postedBody; globalThis.fetch = async (_url, init) => { @@ -55,13 +55,13 @@ test('agent session dispatch can require room video input readiness', async () = const { requestAgentSessionDispatch } = await loadSessionDispatchClientModule(); await requestAgentSessionDispatch('agent-a', '11111111-2222-4333-8444-555555555555', { - requireRoomVideoInputReady: true, + requireAgentSessionReady: true, }); assert.deepEqual(postedBody, { agentName: 'agent-a', sessionId: '11111111-2222-4333-8444-555555555555', - requireRoomVideoInputReady: true, + requireAgentSessionReady: true, }); } finally { globalThis.fetch = originalFetch; diff --git a/tests/session-dispatch-readiness.test.mjs b/tests/session-dispatch-readiness.test.mjs index 7977a5d74..93567c1d6 100644 --- a/tests/session-dispatch-readiness.test.mjs +++ b/tests/session-dispatch-readiness.test.mjs @@ -62,6 +62,24 @@ test('dispatch can require room video input readiness before reusing an agent', ); }); +test('dispatch accepts the server SDK camel-cased agent ready attribute', () => { + const agent = participant({ + identity: 'agent-AJ_ready', + kind: ParticipantInfo_Kind.AGENT, + attributes: { + lkAgentName: 'frontdesk-agent', + liveavatarAgentSessionReady: 'true', + }, + }); + + assert.equal( + findReusableAgentParticipant([agent], 'frontdesk-agent', { + requireAgentSessionReady: true, + }), + agent + ); +}); + test('dispatch can reuse an active agent once room video input is publishing', () => { const agent = participant({ identity: 'agent-AJ_running', diff --git a/tests/session-prewarm.test.mjs b/tests/session-prewarm.test.mjs index 9a44fdb32..243ac835e 100644 --- a/tests/session-prewarm.test.mjs +++ b/tests/session-prewarm.test.mjs @@ -331,7 +331,7 @@ test('missing LiveKit configuration fails before registering a room session', as } }); -test('regular dispatch keeps its 8s timeout while prewarm gets the default 45s total budget', async () => { +test('regular dispatch keeps its 30s timeout while prewarm gets the default 45s total budget', async () => { const originalNow = Date.now; const originalTimeout = process.env.AGENT_DISPATCH_TIMEOUT_MS; const originalPrewarmTimeout = process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; @@ -384,7 +384,7 @@ test('regular dispatch keeps its 8s timeout while prewarm gets the default 45s t ), /agent dispatch failed/ ); - assert.equal(now - regularStartedAt, 8_000); + assert.equal(now - regularStartedAt, 30_000); const prewarmStartedAt = now; await assert.rejects( @@ -1322,7 +1322,7 @@ test('shared dispatch token stays active through per-caller readiness waits', as assert.doesNotMatch(readinessSource, /beginRoomSessionDispatch|finishRoomSessionDispatch/); }); -test('prewarm waits for the agent session and both room input participants', async () => { +test('prewarm completes when the agent session is ready without waiting for optional video input', async () => { const agentName = 'frontdesk-browser-agent-readiness'; let roomCreated = false; let workerReady = false; @@ -1395,7 +1395,7 @@ test('prewarm waits for the agent session and both room input participants', asy assert.deepEqual(result.readiness, { agentSessionReady: true, audioParticipantReady: true, - visionParticipantReady: true, + visionParticipantReady: false, }); }); diff --git a/tests/session-start-dispatch.test.mjs b/tests/session-start-dispatch.test.mjs index dca58fa7e..eba6d783f 100644 --- a/tests/session-start-dispatch.test.mjs +++ b/tests/session-start-dispatch.test.mjs @@ -175,8 +175,8 @@ test('session dispatch route only accepts anonymous LiveKit agent fallback after serviceSource, /const alreadyJoined = await findReusableAgentParticipant\(\s*roomClient,\s*roomName,\s*agentName,\s*reusableAgentOptions\s*\);/ ); - assert.match(routeSource, /requireRoomVideoInputReady/); - assert.match(routeSource, /require_room_video_input_ready/); + assert.match(routeSource, /requireAgentSessionReady/); + assert.match(routeSource, /require_agent_session_ready/); assert.match(readinessSource, /type AgentParticipantMatchOptions/); assert.match(readinessSource, /type ReusableAgentParticipantOptions/); assert.match(readinessSource, /allowAnonymousLiveKitAgentFallback/); @@ -207,9 +207,12 @@ test('start call dispatches the agent with a cancellable room session id', async assert.match(useRoomSource, /isExpectedStartCancellation/); assert.match(useRoomSource, /waitForAgentSessionStop/); assert.match(useRoomSource, /requestAgentSessionDispatch\(\s*appConfig\.agentName,\s*sessionId,/); - assert.match( - useRoomSource, - /requireRoomVideoInputReady: requiresRoomVideoInputReady\(appConfig\)/ + assert.match(useRoomSource, /requireAgentSessionReady: usesManagedRoomInput/); + assert.match(useRoomSource, /await Promise\.allSettled\(\[/); + assert.ok( + useRoomSource.lastIndexOf('setIsSessionActive(true)') > + useRoomSource.lastIndexOf('await dispatchAgentSession()'), + 'session view must become active only after dispatch readiness completes' ); assert.doesNotMatch(useRoomSource, /requestAgentSessionDispatch\(\s*room\.name,/); });