diff --git a/app/api/session/session-dispatch-service.ts b/app/api/session/session-dispatch-service.ts index 4c96a565f..a9476a041 100644 --- a/app/api/session/session-dispatch-service.ts +++ b/app/api/session/session-dispatch-service.ts @@ -85,7 +85,8 @@ const globalForInFlightDispatches = globalThis as typeof globalThis & { const inFlightDispatches = globalForInFlightDispatches.__liveavatarInFlightDispatches ?? (globalForInFlightDispatches.__liveavatarInFlightDispatches = new Map()); -const DEFAULT_AGENT_DISPATCH_TIMEOUT_MS = 30_000; +const DEFAULT_AGENT_DISPATCH_TIMEOUT_MS = 45_000; +const DEFAULT_AGENT_DISPATCH_ATTEMPT_TIMEOUT_MS = 5_000; const DEFAULT_PREWARM_TOTAL_TIMEOUT_MS = 45_000; export type PrewarmPhase = 'room' | 'worker_readiness' | 'dispatch_readiness'; @@ -521,6 +522,10 @@ async function createAgentDispatchWithRetry( const getDeadline = options.getDeadline || (() => fixedDeadline); const retryMs = options.retryMs || readPositiveIntEnv('AGENT_DISPATCH_RETRY_MS', 500); const pollMs = options.pollMs || readPositiveIntEnv('AGENT_DISPATCH_POLL_MS', 200); + const attemptTimeoutMs = readPositiveIntEnv( + 'AGENT_DISPATCH_ATTEMPT_TIMEOUT_MS', + DEFAULT_AGENT_DISPATCH_ATTEMPT_TIMEOUT_MS + ); const sleepFn = options.sleep || sleep; let lastError: unknown; let attempts = 0; @@ -561,12 +566,13 @@ async function createAgentDispatchWithRetry( throw new RoomSessionCancelledError(session); } + const attemptDeadline = Math.min(getDeadline(), Date.now() + attemptTimeoutMs); const agentParticipant = await waitForReusableAgentParticipant( roomClient, roomName, agentName, reusableAgentOptions, - getDeadline, + () => Math.min(getDeadline(), attemptDeadline), pollMs, session, sleepFn @@ -581,6 +587,15 @@ async function createAgentDispatchWithRetry( } lastError = new Error('agent and required room inputs did not become ready'); + await deleteDispatchQuietly(dispatchClient, dispatchId, roomName); + dispatchId = ''; + const waitMs = Math.min( + calculateDispatchRetryDelay(attempts, retryMs), + remainingDispatchTime(getDeadline()) + ); + if (waitMs > 0) { + await sleepFn(waitMs); + } } catch (error) { if (error instanceof RoomSessionCancelledError) { throw error; diff --git a/components/livekit/scroll-area/hooks/useAutoScroll.ts b/components/livekit/scroll-area/hooks/useAutoScroll.ts index 255f63b77..e5ceadabf 100644 --- a/components/livekit/scroll-area/hooks/useAutoScroll.ts +++ b/components/livekit/scroll-area/hooks/useAutoScroll.ts @@ -1,13 +1,15 @@ -import { useEffect, useRef } from 'react'; +import { type RefObject, useEffect, useRef } from 'react'; const AUTO_SCROLL_THRESHOLD_PX = 100; -export function useAutoScroll(scrollContentContainer?: Element | null) { +export function useAutoScroll(scrollContentRef: RefObject) { const isUserScrollingRef = useRef(false); const scrollTimeoutRef = useRef(null); const hasUserScrollIntentRef = useRef(false); useEffect(() => { + const scrollContentContainer = scrollContentRef.current; + function scrollToBottom() { if (!scrollContentContainer || isUserScrollingRef.current) return; @@ -68,5 +70,5 @@ export function useAutoScroll(scrollContentContainer?: Element | null) { } }; } - }, [scrollContentContainer]); + }, [scrollContentRef]); } diff --git a/components/livekit/scroll-area/scroll-area.tsx b/components/livekit/scroll-area/scroll-area.tsx index 8868d06e3..8a693ee2b 100644 --- a/components/livekit/scroll-area/scroll-area.tsx +++ b/components/livekit/scroll-area/scroll-area.tsx @@ -14,7 +14,7 @@ export function ScrollArea({ }: ScrollAreaProps & React.HTMLAttributes) { const scrollContentRef = useRef(null); - useAutoScroll(scrollContentRef.current); + useAutoScroll(scrollContentRef); return (
diff --git a/lib/transcription-history.ts b/lib/transcription-history.ts index 1dbe2384f..b6242ac47 100644 --- a/lib/transcription-history.ts +++ b/lib/transcription-history.ts @@ -23,6 +23,19 @@ export function mergeTranscriptionHistory( finalValue === false || finalValue === 'true' || finalValue === 'false'; + const entryIsPartial = finalValue === false || finalValue === 'false'; + + if (segmentId && entryIsPartial) { + const finalAlreadyExists = Array.from(byStreamId.values()).some((existing) => { + const sameSegment = existing.streamInfo.attributes?.['lk.segment_id'] === segmentId; + const sameParticipant = + existing.participantInfo.identity === entry.participantInfo.identity; + const existingFinal: unknown = existing.streamInfo.attributes?.['lk.transcription_final']; + const existingIsFinal = existingFinal === true || existingFinal === 'true'; + return sameSegment && sameParticipant && existingIsFinal; + }); + if (finalAlreadyExists) return; + } if (segmentId && hasFinalState) { for (const [streamId, existing] of byStreamId) { diff --git a/tests/chat-message-filter.test.mjs b/tests/chat-message-filter.test.mjs index 3fb61d2fb..88834c613 100644 --- a/tests/chat-message-filter.test.mjs +++ b/tests/chat-message-filter.test.mjs @@ -93,3 +93,25 @@ test('transcription history survives a transient empty snapshot', () => { assert.deepEqual(history, [preamble]); }); + +test('a late one-character partial cannot reappear after its final segment', () => { + const partial = transcription('agent-partial', 'speech-agent-1', 100, '请', false); + const completed = transcription('agent-final', 'speech-agent-1', 110, '请稍等，我查一下。', true); + const nextUserTurn = transcription( + 'user-final', + 'speech-user-2', + 200, + '帮我预订中会议室。', + true + ); + + const afterFinal = mergeTranscriptionHistory([], [completed]); + const afterLatePartial = mergeTranscriptionHistory(afterFinal, [partial]); + const afterNextTurn = mergeTranscriptionHistory(afterLatePartial, [nextUserTurn]); + + assert.deepEqual( + afterNextTurn.map(({ text }) => text), + ['请稍等，我查一下。', '帮我预订中会议室。'] + ); + assert.ok(afterNextTurn.every(({ text }) => text !== '请')); +}); diff --git a/tests/scroll-area.test.mjs b/tests/scroll-area.test.mjs new file mode 100644 index 000000000..2c9db2233 --- /dev/null +++ b/tests/scroll-area.test.mjs @@ -0,0 +1,12 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import { test } from 'node:test'; + +test('auto-scroll receives the mounted element ref instead of its initial null value', async () => { + const component = await readFile('components/livekit/scroll-area/scroll-area.tsx', 'utf8'); + const hook = await readFile('components/livekit/scroll-area/hooks/useAutoScroll.ts', 'utf8'); + + assert.match(component, /useAutoScroll\(scrollContentRef\)/); + assert.doesNotMatch(component, /useAutoScroll\(scrollContentRef\.current\)/); + assert.match(hook, /const scrollContentContainer = scrollContentRef\.current/); +}); diff --git a/tests/session-prewarm.test.mjs b/tests/session-prewarm.test.mjs index f78cbd026..39e0d9ca0 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 30s timeout while prewarm gets the default 45s total budget', async () => { +test('regular dispatch and prewarm both allow the default 45s total startup 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 30s timeout while prewarm gets the default 45s ), /agent dispatch failed/ ); - assert.equal(now - regularStartedAt, 30_000); + assert.equal(now - regularStartedAt, 45_000); const prewarmStartedAt = now; await assert.rejects( @@ -414,6 +414,63 @@ test('regular dispatch keeps its 30s timeout while prewarm gets the default 45s } }); +test('dispatch replaces an unassigned dispatch instead of waiting on it for the full deadline', async () => { + const originalNow = Date.now; + const originalAttemptTimeout = process.env.AGENT_DISPATCH_ATTEMPT_TIMEOUT_MS; + let now = 1_000; + let dispatchCount = 0; + const deletedDispatchIds = []; + const agentName = 'frontdesk-browser-agent-recovered-worker'; + Date.now = () => now; + process.env.AGENT_DISPATCH_ATTEMPT_TIMEOUT_MS = '1000'; + + try { + const result = await dispatchRoomSession( + { + roomName: 'voice_assistant_room_recovered_worker', + sessionId: 'recovered-worker', + agentName, + readiness: { requireAgentSessionReady: true }, + }, + { + dispatchClient: { + async createDispatch() { + dispatchCount += 1; + return { id: `dispatch-recovered-worker-${dispatchCount}` }; + }, + async deleteDispatch(dispatchId) { + deletedDispatchIds.push(dispatchId); + }, + }, + roomClient: { + async listParticipants() { + return dispatchCount >= 2 ? readyParticipants(agentName) : []; + }, + async deleteRoom() {}, + }, + dispatchTimeoutMs: 5_000, + dispatchPollMs: 500, + dispatchRetryMs: 500, + sleep: async (ms) => { + now += ms; + }, + } + ); + + assert.equal(result.attempts, 2); + assert.equal(result.dispatchId, 'dispatch-recovered-worker-2'); + assert.deepEqual(deletedDispatchIds, ['dispatch-recovered-worker-1']); + assert.equal(now, 2_500); + } finally { + Date.now = originalNow; + if (originalAttemptTimeout === undefined) { + delete process.env.AGENT_DISPATCH_ATTEMPT_TIMEOUT_MS; + } else { + process.env.AGENT_DISPATCH_ATTEMPT_TIMEOUT_MS = originalAttemptTimeout; + } + } +}); + test('prewarm shares its 45s total budget across worker readiness and dispatch', async () => { const originalNow = Date.now; const originalPrewarmTimeout = process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; @@ -1611,7 +1668,7 @@ test('a concurrent prewarm budget extends the shared in-flight dispatch', async }); const results = await Promise.allSettled([regularDispatch, prewarmDispatch]); - assert.equal(dispatchCalls, 1); + assert.equal(dispatchCalls, 3); assert.equal(now - startedAt, 20_000); for (const result of results) { assert.equal(result.status, 'rejected'); @@ -1684,7 +1741,7 @@ test('a prewarm budget can extend the dispatch during the old deadline check', a releaseDeadlineCheck(); const results = await resultsPromise; - assert.equal(dispatchCalls, 1); + assert.equal(dispatchCalls, 4); assert.equal(now, 28_000); assert.equal( results.every((result) => result.status === 'rejected'),