From 7bd117fdb552366d6a3c8f7299e3104ad3ede8dd Mon Sep 17 00:00:00 2001 From: "luoquan.wl" Date: Sat, 15 Aug 2026 23:45:53 +0800 Subject: [PATCH] feat(voice): add realtime voice sessions --- .env.example | 1 + .gitignore | 1 + README.md | 5 + client/src/App.tsx | 61 +++- client/src/forwardApi.ts | 10 +- client/src/voice/VoiceEntryButton.tsx | 4 + client/src/voice/VoiceSessionView.test.tsx | 104 ++++++ client/src/voice/VoiceSessionView.tsx | 315 ++++++++++++++++++ client/src/voice/useVoiceAvailability.test.ts | 10 + client/src/voice/useVoiceAvailability.ts | 30 ++ client/src/voice/useVoiceSession.ts | 258 ++++++++++++++ client/src/voice/voiceApi.test.ts | 64 ++++ client/src/voice/voiceApi.ts | 57 ++++ client/src/voice/voiceAudio.test.ts | 78 +++++ client/src/voice/voiceAudio.ts | 159 +++++++++ client/src/voice/voiceConnection.test.ts | 132 ++++++++ client/src/voice/voiceConnection.ts | 249 ++++++++++++++ client/src/voice/voiceSession.test.ts | 12 + client/src/voice/voiceSession.ts | 12 + client/src/voice/voiceTimeline.test.ts | 99 ++++++ client/src/voice/voiceTimeline.ts | 233 +++++++++++++ client/tsconfig.app.json | 2 +- client/vite.config.ts | 3 +- package-lock.json | 75 ++++- server/package.json | 5 +- server/src/index.ts | 10 +- server/src/voiceProxy.test.ts | 73 ++++ server/src/voiceProxy.ts | 138 ++++++++ 28 files changed, 2184 insertions(+), 16 deletions(-) create mode 100644 client/src/voice/VoiceEntryButton.tsx create mode 100644 client/src/voice/VoiceSessionView.test.tsx create mode 100644 client/src/voice/VoiceSessionView.tsx create mode 100644 client/src/voice/useVoiceAvailability.test.ts create mode 100644 client/src/voice/useVoiceAvailability.ts create mode 100644 client/src/voice/useVoiceSession.ts create mode 100644 client/src/voice/voiceApi.test.ts create mode 100644 client/src/voice/voiceApi.ts create mode 100644 client/src/voice/voiceAudio.test.ts create mode 100644 client/src/voice/voiceAudio.ts create mode 100644 client/src/voice/voiceConnection.test.ts create mode 100644 client/src/voice/voiceConnection.ts create mode 100644 client/src/voice/voiceSession.test.ts create mode 100644 client/src/voice/voiceSession.ts create mode 100644 client/src/voice/voiceTimeline.test.ts create mode 100644 client/src/voice/voiceTimeline.ts create mode 100644 server/src/voiceProxy.test.ts create mode 100644 server/src/voiceProxy.ts diff --git a/.env.example b/.env.example index 01cc25f..9046f27 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,5 @@ # Optional overrides. Leave empty to use the production defaults in server/src/index.ts. +# Voice Realtime derives its CN/Global WebSocket target from the matching Forward base URL. CN_PROD_FORWARD_API_BASE_URL=https://api.qoder.com.cn/api/v1/forward GLOBAL_PROD_FORWARD_API_BASE_URL=https://api.qoder.com/api/v1/forward CN_PROD_CLOUD_API_BASE_URL=https://api.qoder.com.cn/api/v1/cloud diff --git a/.gitignore b/.gitignore index a7a8511..aa8fa81 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ worktrees/ server/logs/ .vercel .env* +.superpowers/ diff --git a/README.md b/README.md index e622b7a..d9530e7 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ Forward Quickstart 是一个用于体验 Qoder Cloud Agents Forward API 的示 - **密钥管理**：在 Vault 中创建和删除密钥，支持 Bearer Token、OAuth Token 和环境变量类型。 - **会话执行**：基于 Identity 和 Template 创建 Session，发送 `user.message`，并支持取消当前 Turn；遇到需要确认的工具调用或 `AskUserQuestion` 时，由用户显式允许、拒绝、作答或跳过后继续执行。已有任务执行时仍可点击「新建对话」另起新会话，后台任务继续执行且不会干扰当前视图。提问时可点击发送按钮旁的回形针图标添加本地文本类文件（单个 ≤5MB）作为对话附件——文件上传后挂载到 Agent 工作目录（新会话随创建挂载，进行中的会话动态追加挂载），消息中自动标注挂载路径，Agent 可直接读取附件内容作答；附件在消息气泡中以文件卡片展示，刷新后依然可见。 - **实时事件**：通过 SSE 接收 Agent 状态、消息、思考过程、工具调用和工具结果，支持打字机流式输出并实时渲染 Markdown；Agent 通过 DeliverArtifacts 交付的图片文件在对话中内联预览，支持点击放大和下载原图；流式过程中不完整的表格/标题片段也能安全渲染，Session 运行失败（如模型过载）会在对话中显示错误提示；消息按服务端时间戳排序展示，多轮追问时新提问始终显示在对话最底部。多 Agent 协作场景下，子线程的创建、委派、运行和完成等状态事件以轻量标签形式在对话中展示，子线程完成不会提前终止 SSE 流，确保协调者能正确接收子线程结果；单 Agent 会话中自动过滤从其他会话 SSE 流泄漏的多 Agent 状态事件（如「子线程已完成」），避免显示无关标签；协调者（Coordinator）输出的重复回复（如先输出草稿再输出验收版）会通过字符二元组 Jaccard 相似度自动去重，仅保留最终版本。发送按钮旁的设置图标可开关「显示思考过程 / 显示工具调用过程」（选择持久化到本地）；点击历史会话加载事件时显示加载动画，不会闪现欢迎页；历史会话列表支持置顶——悬浮某条记录时显示图钉图标，点击后该会话固定到列表顶部的「置顶」分组（再次点击取消置顶，置顶状态持久化到本地）。 +- **实时语音（本地）**：Template 开启 Realtime 后，可从聊天输入框单击麦克风直接启动独立 Voice Session；页面展示实时字幕、Agent 音频、Work 任务进度、文字混输、静音与结束控制。语音 Session 在历史列表带有“语音”标签，并通过后端 metadata 中的 `conversation_id` 恢复时间线。CN 与 Global 使用相同交互；WebSocket URL 只携带短时单次 connection key，本地 Node 代理负责携带用户在页面中输入的 PAT 连接上游 WebSocket。 - **模板快速切换**：在对话列表顶部直接切换当前会话使用的 Template，无需离开对话界面。 - **权限模式**：内置「开发者模式 / 用户模式」开关（默认用户模式，选择持久化到本地）。开发者模式解锁模板及模板资源（技能、文件、环境、密钥）的新建、编辑和删除权限，并显示对应的「模板资源」菜单；用户模式仅能查看和使用模板。切换到开发者模式时会弹出风险确认提示。 - **会话历史与用量**：查看历史 Session、事件历史、执行状态和会话时长统计。 @@ -113,6 +114,9 @@ npm run dev 启动成功后，终端会同时显示前端 Vite 服务和 Express 本地代理的日志。打开前端地址后，在登录页面选择 API 环境并输入对应的 Forward PAT 即可开始体验。 +Voice 仅在本地开发模式可用。首次启动会请求麦克风权限；若麦克风不可用，仍可在 Voice 页面用文字继续对话。Voice 按钮不可用时，请确认当前 Template 已开启 Realtime。 +Voice 中继只接受来自 `localhost`、`127.0.0.1` 或 `::1` 页面的一次性连接，并仅用于本机开发；请勿将本地代理端口 `3001` 暴露到公网。 + 如果端口被占用，可以先停止占用 `5173` 或 `3001` 的本地进程，再重新执行 `npm run dev`。 ## Vercel 部署 @@ -121,6 +125,7 @@ npm run dev - 本地运行继续使用 `npm run dev`，Vite 会将 `/api` 请求代理到本地 Express 服务。 - Vercel 会构建 `client/dist` 并将现有 Express API 作为 Serverless Functions 部署；前端和 API 使用同一域名。 +- 当前 Vercel 在线 Demo 不提供 Voice WebSocket 中继；普通 HTTP、SSE 和其他现有能力不受影响。 建议先 Fork 本仓库到自己的 GitHub 账号，再在 Vercel 导入 Fork 后的仓库。Vercel 检测到根目录的 `vercel.json` 后会自动使用正确的构建命令和 API 函数配置。 diff --git a/client/src/App.tsx b/client/src/App.tsx index 5ffb157..2f2b3ff 100644 --- a/client/src/App.tsx +++ b/client/src/App.tsx @@ -82,6 +82,10 @@ import { ChatImage } from './chatImage'; import { isImageFile } from './imageUtils'; import { BatchPanel } from './batchPanel'; import { UsagePanel } from './usagePanel'; +import { VoiceEntryButton } from './voice/VoiceEntryButton'; +import { VoiceSessionView } from './voice/VoiceSessionView'; +import { useVoiceAvailability } from './voice/useVoiceAvailability'; +import { isVoiceSession } from './voice/voiceSession'; import { LayerQuizButton } from './layerQuiz'; import { PRODUCT_NAME } from './config/product'; import { @@ -2135,6 +2139,9 @@ export default function App() { const [resourceOptionsByType, setResourceOptionsByType] = useState>(emptyResourceOptions); const [sessions, setSessions] = useState([]); const [currentSessionId, setCurrentSessionId] = useState(''); + const [voiceViewOpen, setVoiceViewOpen] = useState(false); + const [currentVoiceConversationId, setCurrentVoiceConversationId] = useState(null); + const [voiceLaunchKey, setVoiceLaunchKey] = useState(0); const [events, setEvents] = useState([]); const [input, setInput] = useState(''); const [showThinking, setShowThinking] = useState(() => { @@ -2325,6 +2332,8 @@ export default function App() { () => (pat.trim() ? { pat: pat.trim(), environment: apiEnvironment } : null), [apiEnvironment, pat], ); + const voiceAvailability = useVoiceAvailability(ctx, templateId); + useEffect(() => { setVoiceViewOpen(false); setCurrentVoiceConversationId(null); }, [templateId]); // Attachments picked in the composer. They upload immediately on selection // (ChatGPT-style progress chips), then get mounted into the session @@ -2495,12 +2504,13 @@ export default function App() { }, [currentSessionId]); const refreshSessions = useCallback(async (nextIdentity = identity, nextTemplateId?: string) => { - if (!ctx || !nextIdentity) return; + if (!ctx || !nextIdentity) return []; const tplId = nextTemplateId ?? templateIdRef.current; const page = await listSessions(ctx, nextIdentity.id, tplId || undefined); setSessions(page.data); // Never auto-select a session here: background polling calls this while the // user may be sitting on the new-conversation screen (currentSessionId === ''). + return page.data; }, [ctx, identity]); const loadSessionEvents = useCallback(async (sessionId: string) => { @@ -3449,6 +3459,16 @@ export default function App() { setError(''); setSessionLoading(true); streamAbort.current?.abort(); + const selected = sessions.find((session) => session.id === sessionId); + if (selected && isVoiceSession(selected)) { + setCurrentVoiceConversationId(selected.metadata.conversation_id); + setVoiceViewOpen(true); + setVoiceLaunchKey((value) => value + 1); + setSessionLoading(false); + return; + } + setVoiceViewOpen(false); + setCurrentVoiceConversationId(null); try { const page = await listEvents(ctx!, sessionId); // Events are returned in descending order (newest first), reverse to chronological order @@ -4975,7 +4995,7 @@ export default function App() { )} ; +} diff --git a/client/src/voice/VoiceSessionView.test.tsx b/client/src/voice/VoiceSessionView.test.tsx new file mode 100644 index 0000000..5b61607 --- /dev/null +++ b/client/src/voice/VoiceSessionView.test.tsx @@ -0,0 +1,104 @@ +import { renderToStaticMarkup } from 'react-dom/server'; +import { beforeEach, describe, expect, test, vi } from 'vitest'; +import { VoiceSessionView } from './VoiceSessionView'; + +const mocks = vi.hoisted(() => ({ + voice: {} as Record, +})); + +vi.mock('./useVoiceSession', () => ({ + useVoiceSession: () => mocks.voice, +})); + +const props = { + ctx: { environment: 'cn-prod', pat: 'pat' } as never, + identityId: 'identity-1', + templateId: 'template-1', + templateName: '测试模板', + initialConversationId: null, + autoStart: false, + launchKey: 1, + onConversationCreated: vi.fn(), + onStartFailed: vi.fn(), + onNewConversation: vi.fn(), +}; + +beforeEach(() => { + Object.assign(mocks.voice, { + conversationId: 'conv-1', + stage: 'listening', + timeline: [], + muted: false, + error: null, + microphoneWarning: null, + sendText: vi.fn(() => true), + setMuted: vi.fn(), + end: vi.fn(), + continueConversation: vi.fn(), + }); +}); + +describe('VoiceSessionView work timeline', () => { + test('keeps active work open while collapsing tool input and output independently', () => { + mocks.voice.timeline = [{ + kind: 'work', + id: 'work-1', + workId: 'work-1', + objective: '下载并分析 SDK', + status: 'running', + steps: [ + { kind: 'tool_use', title: 'Bash', detail: '{"command":"git clone repo"}' }, + { kind: 'tool_result', title: 'Bash', detail: 'fatal: authentication failed', isError: true }, + ], + }]; + + const html = renderToStaticMarkup(); + + expect(html).toContain('aria-expanded="true"'); + expect(html).toContain('任务处理中'); + expect(html.match(/
{ + mocks.voice.timeline = [{ + kind: 'work', + id: 'work-2', + workId: 'work-2', + objective: '整理结果', + status: 'completed', + steps: [{ kind: 'tool_result', title: 'Bash', detail: 'very verbose raw output' }], + result: '任务完成', + }]; + + const html = renderToStaticMarkup(); + + expect(html).toContain('aria-expanded="false"'); + expect(html).toContain('已完成'); + expect(html).not.toContain('very verbose raw output'); + expect(html).not.toContain('任务完成'); + }); +}); + +describe('VoiceSessionView call controls', () => { + test('uses an icon-only microphone control with an explicit action label', () => { + const html = renderToStaticMarkup(); + + expect(html).toContain('aria-label="静音"'); + expect(html).toContain('麦<'); + }); + + test('describes the inverse action while muted', () => { + mocks.voice.muted = true; + + const html = renderToStaticMarkup(); + + expect(html).toContain('aria-label="取消静音"'); + expect(html).not.toContain('>静<'); + }); +}); diff --git a/client/src/voice/VoiceSessionView.tsx b/client/src/voice/VoiceSessionView.tsx new file mode 100644 index 0000000..28c875b --- /dev/null +++ b/client/src/voice/VoiceSessionView.tsx @@ -0,0 +1,315 @@ +import { useState } from 'react'; +import type { ForwardContext } from '../forwardApi'; +import { useVoiceSession, type VoiceStage } from './useVoiceSession'; +import type { TimelineEntry, WorkEntry, WorkStep } from './voiceTimeline'; + +const stageCopy: Record = { + idle: '准备中', + 'loading-history': '正在加载历史', + connecting: '正在连接', + listening: '我在听', + thinking: '正在处理任务', + speaking: '正在回答', + reconnecting: '正在恢复连接', + ending: '正在结束', + ended: '语音已结束', + error: '连接异常', +}; + +const workStatusCopy: Record = { + accepted: '已受理', + queued: '排队中', + running: '任务处理中', + completed: '已完成', + cancelled: '已取消', + failed: '执行失败', +}; + +const settledWorkStatuses = new Set(['completed', 'cancelled', 'failed']); +const activeVoiceStages = new Set(['listening', 'thinking', 'speaking', 'reconnecting']); + +interface VoiceSessionViewProps { + ctx: ForwardContext; + identityId: string; + templateId: string; + templateName: string; + initialConversationId: string | null; + autoStart: boolean; + launchKey: number; + onConversationCreated: (id: string) => void; + onStartFailed: (message: string) => void; + onNewConversation: () => void; + onEnded?: () => void; +} + +function ChevronIcon({ expanded }: { expanded: boolean }) { + return ( + + ); +} + +function ToolStep({ step }: { step: WorkStep }) { + const result = step.kind === 'tool_result'; + const tone = result ? 'bg-emerald-50 text-emerald-700' : 'bg-orange-50 text-orange-600'; + const payloadTone = result + ? 'border-emerald-200 bg-emerald-50/60' + : 'border-black/10 bg-[#FAFAFA]'; + + return ( +
+ + + {result ? '结果' : '调用'} + {step.title || '工具'} + › + ⌄ + +
+
{result ? '出参' : '入参'}
+ {step.detail ? ( + 
+            {step.detail}
+
+ ) : ( +
暂无详情
+ )} +
+
+ ); +} + +function WorkStepRow({ step }: { step: WorkStep }) { + if (step.kind === 'tool_use' || step.kind === 'tool_result') return ; + + const label = step.kind === 'thinking' + ? '思考' + : step.kind === 'milestone' + ? '进展' + : step.kind === 'message' + ? '输出' + : '进展'; + + return ( +
+ {label} + {step.detail || step.title || '处理中'} +
+ ); +} + +function WorkCard({ entry }: { entry: WorkEntry }) { + const settled = settledWorkStatuses.has(entry.status) || Boolean(entry.result); + const [detailsToggled, setDetailsToggled] = useState(null); + const detailsOpen = detailsToggled ?? !settled; + const failed = entry.status === 'failed'; + const detailCount = entry.steps.length + (entry.result ? 1 : 0); + const statusTone = failed + ? 'bg-red-50 text-red-600' + : settled + ? 'bg-black/[0.04] text-black/45' + : 'bg-[#EEF1FF] text-[#3550FF]'; + + return ( +
+
+ {entry.objective} + + {!settled && } + {workStatusCopy[entry.status] || entry.status} + +
+ {detailCount > 0 && ( +
+ + {detailsOpen && ( +
+ {entry.steps.map((step, index) => )} + {entry.result && ( +
+ {entry.result} +
+ )} +
+ )} +
+ )} +
+ ); +} + +function TimelineItem({ entry }: { entry: TimelineEntry }) { + if (entry.kind === 'work') return ; + + return ( +
+
+ {entry.text} + {entry.pending && …} +
+
+ ); +} + +function MicrophoneIcon({ muted }: { muted: boolean }) { + return muted ? ( + + ) : ( + + ); +} + +function CallControls({ + stage, + muted, + onMutedChange, + onEnd, +}: { + stage: VoiceStage; + muted: boolean; + onMutedChange: (muted: boolean) => void; + onEnd: () => void; +}) { + if (!activeVoiceStages.has(stage)) return null; + + return ( + <> + + + + ); +} + +export function VoiceSessionView(props: VoiceSessionViewProps) { + const voice = useVoiceSession(props); + const [text, setText] = useState(''); + const disabled = !['listening', 'thinking', 'speaking'].includes(voice.stage); + + const send = () => { + if (voice.sendText(text)) setText(''); + }; + + return ( +
+
+
+ + {stageCopy[voice.stage]} + · {props.templateName} +
+
+ {voice.conversationId || 'creating...'} + +
+
+ + {(voice.error || voice.microphoneWarning || voice.stage === 'reconnecting') && ( +
+ {voice.error || voice.microphoneWarning || '正在恢复连接，期间语音不会被处理'} +
+ )} + +
+
+ {voice.timeline.length === 0 &&
直接说话，或在下方输入问题
} + {voice.timeline.map((entry) => )} +
+
+ +
+
+
+ {['介绍一下你能做什么', '帮我看看有哪些文件'].map((item) => ( + + ))} +
+
+ void voice.end().then(() => props.onEnded?.())} + /> + {voice.stage === 'ended' ? ( + + ) : ( + <> + setText(event.target.value)} + onKeyDown={(event) => { + const nativeEvent = event.nativeEvent as KeyboardEvent & { isComposing?: boolean }; + if (event.key === 'Enter' && !nativeEvent.isComposing && nativeEvent.keyCode !== 229) send(); + }} + disabled={disabled} + placeholder="也可以打字说……" + className="h-10 min-w-0 flex-1 rounded-xl border border-black/15 px-4 text-sm outline-none transition focus:border-[#3550FF] disabled:bg-black/[0.02]" + /> + + + )} +
+
+
+
+ ); +} diff --git a/client/src/voice/useVoiceAvailability.test.ts b/client/src/voice/useVoiceAvailability.test.ts new file mode 100644 index 0000000..f928b95 --- /dev/null +++ b/client/src/voice/useVoiceAvailability.test.ts @@ -0,0 +1,10 @@ +import { describe, expect, test } from 'vitest'; +import { voiceAvailabilityFrom } from './useVoiceAvailability'; + +describe('voice availability', () => { + test('maps template config to an enabled or disabled entry', () => { + expect(voiceAvailabilityFrom(true)).toEqual({ status: 'enabled', enabled: true, reason: '' }); + expect(voiceAvailabilityFrom(false)).toMatchObject({ status: 'disabled', enabled: false }); + expect(voiceAvailabilityFrom(true, false)).toEqual({ status: 'disabled', enabled: false, reason: 'Voice 仅支持本地运行' }); + }); +}); diff --git a/client/src/voice/useVoiceAvailability.ts b/client/src/voice/useVoiceAvailability.ts new file mode 100644 index 0000000..dd1d813 --- /dev/null +++ b/client/src/voice/useVoiceAvailability.ts @@ -0,0 +1,30 @@ +import { useEffect, useState } from 'react'; +import type { ForwardContext } from '../forwardApi'; +import { getTemplateRealtimeConfig, getVoiceProxyCapability } from './voiceApi'; + +export type VoiceAvailability = { status: 'loading' | 'enabled' | 'disabled'; enabled: boolean; reason: string }; +export function voiceAvailabilityFrom(enabled: boolean, proxyEnabled = true): VoiceAvailability { + if (!proxyEnabled) return { status: 'disabled', enabled: false, reason: 'Voice 仅支持本地运行' }; + return enabled ? { status: 'enabled', enabled: true, reason: '' } : { status: 'disabled', enabled: false, reason: '当前 Template 未开启 Voice' }; +} +export function useVoiceAvailability(ctx: ForwardContext | null, templateId: string): VoiceAvailability { + const key = ctx && templateId ? `${ctx.environment}:${templateId}` : ''; + const [loaded, setLoaded] = useState<{ key: string; value: VoiceAvailability } | null>(null); + useEffect(() => { + let active = true; + if (!ctx || !templateId) return () => { active = false; }; + void Promise.resolve().then(async () => { + const proxyEnabled = await getVoiceProxyCapability(); + if (!active) return; + if (!proxyEnabled) { + setLoaded({ key, value: voiceAvailabilityFrom(false, false) }); + return; + } + const config = await getTemplateRealtimeConfig(ctx, templateId); + if (active) setLoaded({ key, value: voiceAvailabilityFrom(config.enabled) }); + }).catch(() => { if (active) setLoaded({ key, value: { status: 'disabled', enabled: false, reason: 'Voice 配置不可用' } }); }); + return () => { active = false; }; + }, [ctx, key, templateId]); + if (!key) return { status: 'disabled', enabled: false, reason: '请先选择 Template' }; + return loaded?.key === key ? loaded.value : { status: 'loading', enabled: false, reason: '正在检查 Voice 配置' }; +} diff --git a/client/src/voice/useVoiceSession.ts b/client/src/voice/useVoiceSession.ts new file mode 100644 index 0000000..e895953 --- /dev/null +++ b/client/src/voice/useVoiceSession.ts @@ -0,0 +1,258 @@ +import { useCallback, useEffect, useRef, useState } from 'react'; +import type { ForwardContext } from '../forwardApi'; +import { AudioPlayback, handleVoicePlaybackEvent, MicrophoneCapture, type VoicePlaybackRuntime } from './voiceAudio'; +import { createRealtimeConversation, getCompleteRealtimeConversationHistory, requestVoiceConnectionKey } from './voiceApi'; +import { VoiceConnection, type VoiceServerEvent } from './voiceConnection'; +import { + applyVoiceTimelineEvent, + createVoiceTimelineState, + createVoiceTimelineStateFromEntries, + projectVoiceHistory, + removeTimelineEntry, + selectTimelineEntries, + upsertLocalUserTextDraft, + type TimelineEntry, + type VoiceTimelineState, +} from './voiceTimeline'; + +export type VoiceStage = 'idle' | 'loading-history' | 'connecting' | 'listening' | 'thinking' | 'speaking' | 'reconnecting' | 'ending' | 'ended' | 'error'; + +interface VoiceSessionOptions { + ctx: ForwardContext; + identityId: string; + templateId: string; + initialConversationId: string | null; + autoStart: boolean; + launchKey: number; + onConversationCreated: (id: string) => void; + onStartFailed: (message: string) => void; +} + +const activeWorkTypes = new Set(['work.accepted', 'work.queued', 'work.started', 'work.running', 'work.progress', 'work.milestone']); +const terminalWorkTypes = new Set(['work.completed', 'work.failed', 'work.cancelled']); +const hiddenErrorCodes = new Set(['service_restarting', 'provider_unavailable', 'invalid_playback_receipt']); +const MAX_CLIENT_TEXT_BYTES = 16 * 1024; + +export function useVoiceSession(options: VoiceSessionOptions) { + const [conversationId, setConversationId] = useState(options.initialConversationId); + const [stage, setStage] = useState(options.initialConversationId ? 'loading-history' : 'idle'); + const [timelineState, setTimelineState] = useState(createVoiceTimelineState()); + const [muted, setMuted] = useState(false); + const [error, setError] = useState(null); + const [microphoneWarning, setMicrophoneWarning] = useState(null); + const mutedRef = useRef(muted); + const stageRef = useRef(stage); + const connection = useRef(null); + const microphone = useRef(new MicrophoneCapture()); + const playback = useRef(null); + const generation = useRef(0); + const createIdempotencyKey = useRef(crypto.randomUUID()); + + useEffect(() => { mutedRef.current = muted; stageRef.current = stage; }, [muted, stage]); + + const loadHistory = useCallback(async (id: string) => { + const history = await getCompleteRealtimeConversationHistory(options.ctx, id, { limit: 100, types: 'message,work' }); + setTimelineState(createVoiceTimelineStateFromEntries(projectVoiceHistory(history.events))); + return history; + }, [options.ctx]); + + const createPlayback = useCallback(() => new AudioPlayback({ + onReceipt: (type, identity) => connection.current?.send(type, identity), + onState: (value) => { + if (value === 'idle') { + setStage((current) => current === 'speaking' ? 'listening' : current); + } + }, + }), []); + + const handleEvent = useCallback((event: VoiceServerEvent) => { + const runtime: VoicePlaybackRuntime = { + playback: playback.current, + createPlayback, + onStage: (nextStage) => setStage((current) => ['ending', 'ended', 'error'].includes(current) ? current : nextStage), + }; + if (handleVoicePlaybackEvent(event, runtime)) { + playback.current = runtime.playback; + return; + } + if (event.type === 'voice.ready') { + setError(null); + setStage('listening'); + return; + } + if (event.type === 'voice.replaced') { + setError('该语音会话已被其他连接接管'); + setStage('error'); + void microphone.current.stop(); + void playback.current?.cancel(); + connection.current?.disconnect(); + return; + } + if (event.type === 'error') { + const code = typeof event.payload.code === 'string' ? event.payload.code : 'unknown'; + if (!hiddenErrorCodes.has(code)) setError(`${String(event.payload.message || '语音服务异常')}（${code}）`); + return; + } + if (event.type.startsWith('transcript.') || event.type.startsWith('work.')) { + setTimelineState((state) => applyVoiceTimelineEvent(state, event)); + if (activeWorkTypes.has(event.type)) setStage((current) => current === 'speaking' ? current : 'thinking'); + if (terminalWorkTypes.has(event.type)) setStage((current) => current === 'speaking' ? current : 'listening'); + } + }, [createPlayback]); + + const connect = useCallback(async (id: string) => { + connection.current?.disconnect(); + const next = new VoiceConnection({ + conversationId: id, + getConnectionKey: async () => (await requestVoiceConnectionKey(options.ctx, id)).connection_key, + beforeReconnect: async () => { + setStage('reconnecting'); + await playback.current?.cancel(); + await loadHistory(id); + }, + onEvent: handleEvent, + onState: (value) => { + if (value === 'reconnecting') setStage('reconnecting'); + if (value === 'replaced') { + setStage('error'); + setError('该语音会话已被其他连接接管'); + } + if (value === 'disconnected' && !['ending', 'ended', 'error'].includes(stageRef.current)) { + setStage('error'); + setError('语音连接已断开'); + void microphone.current.stop(); + void playback.current?.cancel(); + } + }, + onError: (value) => { setStage('error'); setError(value.message); }, + }); + connection.current = next; + setStage('connecting'); + await next.connect(); + }, [handleEvent, loadHistory, options.ctx]); + + const startMicrophone = useCallback(async () => { + setMicrophoneWarning(null); + try { + await microphone.current.start((audio) => { + if (!mutedRef.current && connection.current?.ready && stageRef.current !== 'reconnecting') { + connection.current.send('audio.append', { audio }); + } + }); + } catch { + setMicrophoneWarning('麦克风不可用，仍可使用文字对话'); + } + }, []); + + const startNew = useCallback(async () => { + const run = ++generation.current; + setError(null); + setConversationId(null); + setTimelineState(createVoiceTimelineState()); + await startMicrophone(); + if (run !== generation.current) return; + try { + const created = await createRealtimeConversation(options.ctx, { + templateId: options.templateId, + identityId: options.identityId, + title: 'Voice Session', + idempotencyKey: createIdempotencyKey.current, + }); + if (run !== generation.current) return; + setConversationId(created.id); + options.onConversationCreated(created.id); + await connect(created.id); + } catch (value) { + if (run !== generation.current) return; + await microphone.current.stop(); + const message = value instanceof Error ? value.message : String(value); + setStage('error'); + setError(message); + options.onStartFailed(message); + } + }, [connect, options, startMicrophone]); + + useEffect(() => { + const mic = microphone.current; + const id = options.initialConversationId; + void Promise.resolve().then(async () => { + if (id) { + setConversationId(id); + setStage('loading-history'); + await loadHistory(id).then(() => setStage('ended')).catch((value) => { + setStage('error'); + setError(value instanceof Error ? value.message : String(value)); + }); + } else if (options.autoStart) { + await startNew(); + } + }); + return () => { + generation.current += 1; + connection.current?.disconnect(); + void mic.stop(); + void playback.current?.cancel(); + }; + // launchKey remounts one Voice lifecycle; other values are captured for that lifecycle. + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [options.launchKey]); + + const sendText = useCallback((text: string) => { + const value = text.trim(); + const activeConnection = connection.current; + if (!value || !activeConnection?.ready) return false; + if (new TextEncoder().encode(value).byteLength > MAX_CLIENT_TEXT_BYTES) { + setError('文字消息过长，请缩短后重试'); + return false; + } + setError(null); + const id = `local-${crypto.randomUUID()}`; + setTimelineState((state) => upsertLocalUserTextDraft(state, id, value, true)); + activeConnection.send('interrupt', { reason: 'text_message' }); + void playback.current?.cancel(); + const sent = activeConnection.send('text.message', { text: value }); + setTimelineState((state) => sent + ? upsertLocalUserTextDraft(state, id, value, false) + : removeTimelineEntry(state, id)); + if (!sent) setError('消息发送失败，请稍后重试'); + return sent; + }, []); + + const end = useCallback(async () => { + const activeConnection = connection.current; + setStage('ending'); + await Promise.allSettled([microphone.current.stop(), playback.current?.cancel()]); + try { + await activeConnection?.closeGracefully(); + if (conversationId) await loadHistory(conversationId); + } catch (value) { + setError(value instanceof Error ? value.message : '语音连接未能安全结束'); + } finally { + if (connection.current === activeConnection) connection.current = null; + activeConnection?.disconnect(); + playback.current = null; + setStage('ended'); + } + }, [conversationId, loadHistory]); + + const continueConversation = useCallback(async () => { + if (!conversationId) return; + setError(null); + await startMicrophone(); + await connect(conversationId); + }, [connect, conversationId, startMicrophone]); + + return { + conversationId, + stage, + timeline: selectTimelineEntries(timelineState) as TimelineEntry[], + muted, + setMuted, + error, + microphoneWarning, + startNew, + continueConversation, + sendText, + end, + }; +} diff --git a/client/src/voice/voiceApi.test.ts b/client/src/voice/voiceApi.test.ts new file mode 100644 index 0000000..6b92b8d --- /dev/null +++ b/client/src/voice/voiceApi.test.ts @@ -0,0 +1,64 @@ +import { afterEach, describe, expect, test, vi } from 'vitest'; +import type { ForwardContext } from '../forwardApi'; +import { createRealtimeConversation, getCompleteRealtimeConversationHistory, getRealtimeConversationHistory, getTemplateRealtimeConfig, getVoiceProxyCapability, requestVoiceConnectionKey } from './voiceApi'; + +const ctx: ForwardContext = { pat: 'pat_secret', environment: 'global-prod' }; + +describe('voice API', () => { + afterEach(() => vi.unstubAllGlobals()); + + test('uses Forward-relative realtime HTTP paths and stable idempotency', async () => { + const bodies: Array> = []; + vi.stubGlobal('fetch', vi.fn(async (_url: string, init?: RequestInit) => { + bodies.push(JSON.parse(String(init?.body))); + return new Response(JSON.stringify({ id: 'conv_1', object: 'voice.conversation', status: 'ready', events: [], page: { has_more: false, next_before: null } }), { status: 200 }); + })); + await getTemplateRealtimeConfig(ctx, 'tmpl_1'); + await createRealtimeConversation(ctx, { templateId: 'tmpl_1', identityId: 'idn_1', title: 'Voice Session', idempotencyKey: 'voice-create-1' }); + await getRealtimeConversationHistory(ctx, 'conv_1', { limit: 100, types: 'message,work' }); + expect(bodies[0]).toMatchObject({ environment: 'global-prod', method: 'GET', path: '/realtime/templates/tmpl_1' }); + expect(bodies[1]).toMatchObject({ method: 'POST', path: '/realtime/conversations', body: { identity_id: 'idn_1', template_id: 'tmpl_1', title: 'Voice Session' }, idempotencyKey: 'voice-create-1' }); + expect(bodies[2]).toMatchObject({ method: 'GET', path: '/realtime/conversations/conv_1/history', query: { limit: 100, types: 'message,work' } }); + }); + + test('requests a local one-time websocket key', async () => { + let body: Record = {}; + vi.stubGlobal('fetch', vi.fn(async (url: string, init?: RequestInit) => { + expect(url).toBe('/api/voice/connect'); + body = JSON.parse(String(init?.body)); + return new Response(JSON.stringify({ connection_key: 'key_1', expires_in_ms: 30000 }), { status: 200 }); + })); + expect(await requestVoiceConnectionKey(ctx, 'conv_1')).toEqual({ connection_key: 'key_1', expires_in_ms: 30000 }); + expect(body).toEqual({ pat: 'pat_secret', environment: 'global-prod', conversation_id: 'conv_1' }); + }); + + test('reads the local voice proxy capability from health', async () => { + vi.stubGlobal('fetch', vi.fn(async () => new Response(JSON.stringify({ + status: 'ok', + voiceRealtimeProxy: { enabled: false, localOnly: true }, + }), { status: 200 }))); + + await expect(getVoiceProxyCapability()).resolves.toBe(false); + }); + + test('loads every realtime history page using next_before', async () => { + const bodies: Array> = []; + vi.stubGlobal('fetch', vi.fn(async (_url: string, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as Record; + bodies.push(body); + const before = (body.query as Record).before; + return new Response(JSON.stringify({ + conversation: { id: 'conv_1', initialization_status: 'ready' }, + events: [{ id: before ? 'evt_1' : 'evt_2', type: 'voice.user_message.completed', role: 'user', status: 'completed', text: before ? 'first' : 'second', occurred_at: before ? '2026-08-15T00:00:00Z' : '2026-08-15T00:00:01Z' }], + page: before ? { has_more: false, next_before: null } : { has_more: true, next_before: 'cursor-1' }, + }), { status: 200 }); + })); + + const history = await getCompleteRealtimeConversationHistory(ctx, 'conv_1', { limit: 100, types: 'message,work' }); + + expect(history.events.map((event) => event.id)).toEqual(['evt_2', 'evt_1']); + expect(bodies).toHaveLength(2); + expect(bodies[0].query).toEqual({ limit: 100, types: 'message,work' }); + expect(bodies[1].query).toEqual({ limit: 100, types: 'message,work', before: 'cursor-1' }); + }); +}); diff --git a/client/src/voice/voiceApi.ts b/client/src/voice/voiceApi.ts new file mode 100644 index 0000000..c905294 --- /dev/null +++ b/client/src/voice/voiceApi.ts @@ -0,0 +1,57 @@ +import { ForwardApiError, forwardRequest, type ForwardContext } from '../forwardApi'; + +export interface TemplateRealtimeConfig { type: 'template_realtime_config'; template_id: string; enabled: boolean } +export interface RealtimeConversation { id: string; object: 'voice.conversation'; status: 'initializing' | 'ready' | 'failed'; title?: string | null; metadata?: Record; created_at?: string; updated_at?: string } +export interface RealtimeHistoryEvent { id: string; type: string; role?: 'user' | 'assistant'; status: string; text?: string; work_id?: string; objective?: string; result?: string; error?: { code: string }; occurred_at: string; turn_id?: string; user_message_event_id?: string } +export interface RealtimeConversationHistory { conversation: { id: string; title?: string | null; initialization_status: 'initializing' | 'ready' | 'failed'; metadata?: Record; created_at?: string; updated_at?: string }; events: RealtimeHistoryEvent[]; page: { next_before: string | null; has_more: boolean } } +export interface RealtimeHistoryOptions { limit?: number; before?: string; types?: 'message,work' } + +export async function getVoiceProxyCapability() { + const response = await fetch('/api/health'); + const data = await response.json().catch(() => null) as { voiceRealtimeProxy?: { enabled?: boolean } } | null; + if (!response.ok) throw new ForwardApiError(response.status, 'Voice proxy capability check failed'); + return data?.voiceRealtimeProxy?.enabled === true; +} + +export function getTemplateRealtimeConfig(ctx: ForwardContext, templateId: string) { + return forwardRequest(ctx, 'GET', `/realtime/templates/${encodeURIComponent(templateId)}`); +} + +export function createRealtimeConversation(ctx: ForwardContext, input: { templateId: string; identityId: string; title?: string; idempotencyKey: string }) { + return forwardRequest(ctx, 'POST', '/realtime/conversations', { + template_id: input.templateId, + identity_id: input.identityId, + title: input.title || 'Voice Session', + }, undefined, { idempotencyKey: input.idempotencyKey }); +} + +export function getRealtimeConversationHistory(ctx: ForwardContext, conversationId: string, options: RealtimeHistoryOptions = {}) { + return forwardRequest(ctx, 'GET', `/realtime/conversations/${encodeURIComponent(conversationId)}/history`, undefined, { ...options }); +} + +export async function getCompleteRealtimeConversationHistory(ctx: ForwardContext, conversationId: string, options: RealtimeHistoryOptions = {}) { + let before = options.before; + const seenCursors = new Set(before ? [before] : []); + let conversation: RealtimeConversationHistory['conversation'] | undefined; + const events: RealtimeHistoryEvent[] = []; + while (true) { + const page = await getRealtimeConversationHistory(ctx, conversationId, { ...options, ...(before ? { before } : {}) }); + conversation ??= page.conversation; + events.push(...page.events); + const next = page.page.has_more ? page.page.next_before : null; + if (!next || seenCursors.has(next)) return { conversation, events, page: page.page }; + seenCursors.add(next); + before = next; + } +} + +export async function requestVoiceConnectionKey(ctx: ForwardContext, conversationId: string) { + const response = await fetch('/api/voice/connect', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ pat: ctx.pat, environment: ctx.environment, conversation_id: conversationId }), + }); + const data = await response.json().catch(() => null) as { connection_key?: string; expires_in_ms?: number; error?: { message?: string } } | null; + if (!response.ok || !data?.connection_key) throw new ForwardApiError(response.status, data?.error?.message || 'Voice connection key request failed'); + return { connection_key: data.connection_key, expires_in_ms: Number(data.expires_in_ms || 0) }; +} diff --git a/client/src/voice/voiceAudio.test.ts b/client/src/voice/voiceAudio.test.ts new file mode 100644 index 0000000..2698767 --- /dev/null +++ b/client/src/voice/voiceAudio.test.ts @@ -0,0 +1,78 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { AudioPlayback, base64Pcm16ToFloat, floatToPcm16, handleVoicePlaybackEvent, pcm16Frames, type VoicePlaybackRuntime } from './voiceAudio'; + +class FakeSource { + buffer: { duration: number } | null = null; + onended: (() => void) | null = null; + connect = vi.fn(); + start = vi.fn(); + stop = vi.fn(); +} + +class FakeAudioContext { + static instances: FakeAudioContext[] = []; + currentTime = 0; + state = 'running'; + destination = {}; + sources: FakeSource[] = []; + resume = vi.fn(); + close = vi.fn(async () => undefined); + createBuffer = vi.fn(() => ({ duration: 1 / 24_000, copyToChannel: vi.fn() })); + createBufferSource = vi.fn(() => { + const source = new FakeSource(); + this.sources.push(source); + return source; + }); + constructor() { FakeAudioContext.instances.push(this); } +} + +describe('voice audio conversion', () => { + beforeEach(() => { + FakeAudioContext.instances = []; + vi.stubGlobal('AudioContext', FakeAudioContext); + }); + + afterEach(() => vi.unstubAllGlobals()); + + test('clamps float samples to PCM16', () => { + expect(Array.from(floatToPcm16(new Float32Array([-2, -1, 0, 1, 2]), 16_000))).toEqual([-32768, -32768, 0, 32767, 32767]); + }); + test('splits and decodes frames', () => { + const frames = pcm16Frames(new Float32Array(5000).fill(0.5), 16_000); + expect(frames).toHaveLength(2); + expect(base64Pcm16ToFloat(frames[0])).toHaveLength(4096); + expect(() => base64Pcm16ToFloat('AQ==')).toThrow('Invalid PCM16 audio'); + }); + + test('stops queued browser audio when the gateway interrupts playback', async () => { + const receipts: string[] = []; + const stages: string[] = []; + const runtime: VoicePlaybackRuntime = { + playback: null, + createPlayback: () => new AudioPlayback({ onReceipt: (type) => receipts.push(type) }), + onStage: (stage) => stages.push(stage), + }; + + handleVoicePlaybackEvent({ type: 'audio.delta', work_id: 'work-1', payload: { audio: 'AAA=' } }, runtime); + expect(FakeAudioContext.instances[0].sources[0].stop).not.toHaveBeenCalled(); + handleVoicePlaybackEvent({ type: 'playback.interrupt', payload: { reason: 'speech_started' } }, runtime); + await vi.waitFor(() => expect(FakeAudioContext.instances[0].sources[0].stop).toHaveBeenCalledOnce()); + + expect(receipts).toEqual(['playback.started', 'playback.cancelled']); + expect(stages).toEqual(['speaking', 'listening']); + }); + + test('does not replace an active announcement with mismatched audio', () => { + const runtime: VoicePlaybackRuntime = { + playback: null, + createPlayback: () => new AudioPlayback(), + onStage: vi.fn(), + }; + + handleVoicePlaybackEvent({ type: 'audio.delta', work_id: 'work-1', announcement_id: 'ann-1', payload: { audio: 'AAA=' } }, runtime); + handleVoicePlaybackEvent({ type: 'audio.delta', work_id: 'work-1', announcement_id: 'ann-2', payload: { audio: 'AAA=' } }, runtime); + + expect(FakeAudioContext.instances).toHaveLength(1); + expect(FakeAudioContext.instances[0].createBufferSource).toHaveBeenCalledOnce(); + }); +}); diff --git a/client/src/voice/voiceAudio.ts b/client/src/voice/voiceAudio.ts new file mode 100644 index 0000000..5440a5f --- /dev/null +++ b/client/src/voice/voiceAudio.ts @@ -0,0 +1,159 @@ +const INPUT_RATE = 16_000; +const OUTPUT_RATE = 24_000; +const MAX_SAMPLES = 4096; + +function toBase64(bytes: Uint8Array) { + let binary = ''; + for (const byte of bytes) binary += String.fromCharCode(byte); + return btoa(binary); +} + +export function floatToPcm16(samples: Float32Array, sourceRate: number, targetRate = INPUT_RATE): Int16Array { + const ratio = sourceRate / targetRate; + const result = new Int16Array(Math.max(1, Math.floor(samples.length / ratio))); + for (let i = 0; i < result.length; i += 1) { + const start = Math.floor(i * ratio); + const end = Math.max(start + 1, Math.floor((i + 1) * ratio)); + let sum = 0; + for (let j = start; j < end && j < samples.length; j += 1) sum += samples[j]; + const value = Math.max(-1, Math.min(1, sum / Math.max(1, end - start))); + result[i] = value < 0 ? value * 0x8000 : value * 0x7fff; + } + return result; +} + +export function pcm16Frames(samples: Float32Array, sourceRate: number): string[] { + const pcm = floatToPcm16(samples, sourceRate); + const result: string[] = []; + for (let offset = 0; offset < pcm.length; offset += MAX_SAMPLES) { + const frame = pcm.subarray(offset, offset + MAX_SAMPLES); + result.push(toBase64(new Uint8Array(frame.buffer, frame.byteOffset, frame.byteLength))); + } + return result; +} + +export function base64Pcm16ToFloat(audio: string): Float32Array { + const binary = atob(audio); + if (binary.length % 2 !== 0) throw new Error('Invalid PCM16 audio'); + const bytes = new Uint8Array(binary.length); + for (let i = 0; i < binary.length; i += 1) bytes[i] = binary.charCodeAt(i); + const view = new DataView(bytes.buffer); + const output = new Float32Array(new ArrayBuffer(binary.length * 2)); + for (let i = 0; i < output.length; i += 1) output[i] = view.getInt16(i * 2, true) / 0x8000; + return output; +} + +export class MicrophoneCapture { + private generation = 0; + private stream: MediaStream | null = null; + private context: AudioContext | null = null; + private source: MediaStreamAudioSourceNode | null = null; + private processor: ScriptProcessorNode | null = null; + + async start(onFrame: (frame: string) => void): Promise { + if (this.stream) return; + const generation = ++this.generation; + const stream = await navigator.mediaDevices.getUserMedia({ audio: { channelCount: 1, echoCancellation: true, noiseSuppression: true, autoGainControl: true }, video: false }); + if (generation !== this.generation) { stream.getTracks().forEach((track) => track.stop()); return; } + const context = new AudioContext(); + const source = context.createMediaStreamSource(stream); + const processor = context.createScriptProcessor(2048, 1, 1); + processor.onaudioprocess = (event) => pcm16Frames(event.inputBuffer.getChannelData(0), context.sampleRate).forEach(onFrame); + source.connect(processor); + processor.connect(context.destination); + this.stream = stream; this.context = context; this.source = source; this.processor = processor; + } + + async stop(): Promise { + this.generation += 1; + const { stream, context, source, processor } = this; + this.stream = null; this.context = null; this.source = null; this.processor = null; + if (processor) { processor.onaudioprocess = null; processor.disconnect(); } + source?.disconnect(); + stream?.getTracks().forEach((track) => track.stop()); + await context?.close(); + } +} + +export type PlaybackIdentity = { work_id?: string; announcement_id?: string }; +export class AudioPlayback { + private context: AudioContext | null = null; + private cursor = 0; + private sources = new Set(); + private started = false; + private finished = false; + private identity: PlaybackIdentity = {}; + private cancelPromise: Promise | null = null; + private handlers: { onReceipt?: (type: 'playback.started' | 'playback.ended' | 'playback.cancelled', identity: PlaybackIdentity) => void; onState?: (state: 'playing' | 'idle') => void }; + constructor(handlers: { onReceipt?: (type: 'playback.started' | 'playback.ended' | 'playback.cancelled', identity: PlaybackIdentity) => void; onState?: (state: 'playing' | 'idle') => void } = {}) { this.handlers = handlers; } + get active() { return this.context !== null; } + begin(identity: PlaybackIdentity = {}) { if (this.context) return; this.context = new AudioContext({ sampleRate: OUTPUT_RATE }); if (this.context.state === 'suspended') void this.context.resume(); this.cursor = this.context.currentTime; this.identity = { ...identity }; this.started = false; this.finished = false; } + matches(identity: PlaybackIdentity) { return this.identity.work_id === identity.work_id && this.identity.announcement_id === identity.announcement_id; } + append(audio: string) { + if (!this.context) this.begin(); + const context = this.context!; + const samples = base64Pcm16ToFloat(audio); + const buffer = context.createBuffer(1, samples.length, OUTPUT_RATE); buffer.copyToChannel(samples, 0); + const source = context.createBufferSource(); source.buffer = buffer; source.connect(context.destination); + this.cursor = Math.max(this.cursor, context.currentTime); source.start(this.cursor); this.cursor += buffer.duration; this.sources.add(source); + if (!this.started) { this.started = true; this.handlers.onState?.('playing'); this.handlers.onReceipt?.('playback.started', this.identity); } + source.onended = () => { this.sources.delete(source); void this.complete(); }; + } + finish() { this.finished = true; void this.complete(); } + async cancel() { + if (this.cancelPromise) return this.cancelPromise; + this.cancelPromise = (async () => { + if (this.started) this.handlers.onReceipt?.('playback.cancelled', { ...this.identity }); + this.sources.forEach((source) => { source.onended = null; try { source.stop(); } catch { /* ended */ } }); + this.sources.clear(); + await this.reset(); + })(); + try { await this.cancelPromise; } finally { this.cancelPromise = null; } + } + private async complete() { if (!this.finished || this.sources.size) return; if (this.started) this.handlers.onReceipt?.('playback.ended', this.identity); await this.reset(); } + private async reset() { const context = this.context; this.context = null; this.identity = {}; this.started = false; this.finished = false; this.handlers.onState?.('idle'); await context?.close(); } +} + +export interface VoicePlaybackEvent { + type: string; + work_id?: string; + announcement_id?: string; + payload: Record; +} + +export interface VoicePlaybackRuntime { + playback: AudioPlayback | null; + createPlayback: () => AudioPlayback; + onStage: (stage: 'speaking' | 'listening') => void; +} + +export function handleVoicePlaybackEvent(event: VoicePlaybackEvent, runtime: VoicePlaybackRuntime): boolean { + if (event.type === 'playback.interrupt') { + void runtime.playback?.cancel(); + runtime.onStage('listening'); + return true; + } + if (event.type === 'voice.state') { + if (event.payload.state === 'interrupted' || event.payload.state === 'idle') runtime.onStage('listening'); + return true; + } + const identity: PlaybackIdentity = { + ...(event.work_id ? { work_id: event.work_id } : {}), + ...(event.announcement_id ? { announcement_id: event.announcement_id } : {}), + }; + if (event.type === 'audio.delta') { + if (typeof event.payload.audio !== 'string') return true; + const playback = runtime.playback ?? runtime.createPlayback(); + runtime.playback = playback; + if (!playback.active) playback.begin(identity); + if (!playback.matches(identity)) return true; + playback.append(event.payload.audio); + runtime.onStage('speaking'); + return true; + } + if (event.type === 'audio.done') { + if (runtime.playback?.active && runtime.playback.matches(identity)) runtime.playback.finish(); + return true; + } + return false; +} diff --git a/client/src/voice/voiceConnection.test.ts b/client/src/voice/voiceConnection.test.ts new file mode 100644 index 0000000..f092795 --- /dev/null +++ b/client/src/voice/voiceConnection.test.ts @@ -0,0 +1,132 @@ +import { afterEach, describe, expect, test, vi } from 'vitest'; +import { buildLocalVoiceSocketUrl, isValidVoiceServerEvent, VoiceConnection } from './voiceConnection'; + +class FakeWebSocket { + static readonly OPEN = 1; + static readonly CLOSING = 2; + static instances: FakeWebSocket[] = []; + readyState = FakeWebSocket.OPEN; + send = vi.fn(); + close = vi.fn(); + private listeners = new Map) => void>>(); + + constructor() { FakeWebSocket.instances.push(this); } + addEventListener(type: string, listener: (event: Record) => void) { + this.listeners.set(type, [...(this.listeners.get(type) || []), listener]); + } + emit(type: string, event: Record = {}) { + for (const listener of this.listeners.get(type) || []) listener(event); + } +} + +function installBrowser() { + vi.stubGlobal('WebSocket', FakeWebSocket); + vi.stubGlobal('window', { + location: { href: 'http://localhost:5173/' }, + setTimeout: globalThis.setTimeout, + clearTimeout: globalThis.clearTimeout, + setInterval: globalThis.setInterval, + clearInterval: globalThis.clearInterval, + addEventListener: vi.fn(), + removeEventListener: vi.fn(), + }); +} + +describe('voice connection contract', () => { + afterEach(() => { + FakeWebSocket.instances = []; + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + test('keeps credentials and conversation id out of the browser websocket URL', () => { + const url = buildLocalVoiceSocketUrl('key only', 'http://localhost:5173/'); + expect(url).toBe('ws://localhost:5173/api/voice/socket?key=key+only'); + expect(url).not.toContain('conv_'); + expect(url).not.toContain('pat_'); + }); + test('validates the versioned server envelope and conversation', () => { + const event = { version: 'voice.realtime.v1', type: 'voice.ready', event_id: 'evt_1', sequence: 1, conversation_id: 'conv_1', timestamp: '2026-08-15T00:00:00Z', payload: {} }; + expect(isValidVoiceServerEvent(event, 'conv_1')).toBe(true); + expect(isValidVoiceServerEvent({ ...event, conversation_id: 'conv_2' }, 'conv_1')).toBe(false); + expect(isValidVoiceServerEvent({ ...event, version: 'v0' }, 'conv_1')).toBe(false); + expect(isValidVoiceServerEvent({ ...event, work_id: 42 }, 'conv_1')).toBe(false); + expect(isValidVoiceServerEvent({ ...event, announcement_id: '' }, 'conv_1')).toBe(false); + }); + + test('waits for the matching gateway acknowledgement before graceful close completes', async () => { + installBrowser(); + vi.stubGlobal('crypto', { randomUUID: () => 'close-request-1' }); + const connection = new VoiceConnection({ + conversationId: 'conv_1', + getConnectionKey: async () => 'key-1', + webSocketFactory: () => new FakeWebSocket() as unknown as WebSocket, + }); + await connection.connect(); + const socket = FakeWebSocket.instances[0]; + socket.emit('message', { data: JSON.stringify({ + version: 'voice.realtime.v1', type: 'voice.ready', event_id: 'ready-1', sequence: 1, + conversation_id: 'conv_1', timestamp: '2026-08-15T00:00:00Z', + payload: { capabilities: { graceful_close: true } }, + }) }); + + let completed = false; + const closing = connection.closeGracefully().then((result) => { completed = true; return result; }); + expect(completed).toBe(false); + expect(socket.send).toHaveBeenLastCalledWith(JSON.stringify({ + version: 'voice.realtime.v1', type: 'connection.close', payload: { request_id: 'close-request-1' }, + })); + + socket.emit('message', { data: JSON.stringify({ + version: 'voice.realtime.v1', type: 'connection.closed', event_id: 'closed-1', sequence: 2, + conversation_id: 'conv_1', timestamp: '2026-08-15T00:00:01Z', + payload: { request_id: 'close-request-1', outcome: 'saved_interrupted' }, + }) }); + await expect(closing).resolves.toEqual({ outcome: 'saved_interrupted' }); + }); + + test('abandons graceful close when the gateway never acknowledges it', async () => { + vi.useFakeTimers(); + installBrowser(); + vi.stubGlobal('crypto', { randomUUID: () => 'close-request-timeout' }); + const connection = new VoiceConnection({ + conversationId: 'conv_1', + getConnectionKey: async () => 'key-1', + webSocketFactory: () => new FakeWebSocket() as unknown as WebSocket, + }); + await connection.connect(); + const socket = FakeWebSocket.instances[0]; + socket.emit('message', { data: JSON.stringify({ + version: 'voice.realtime.v1', type: 'voice.ready', event_id: 'ready-timeout', sequence: 1, + conversation_id: 'conv_1', timestamp: '2026-08-15T00:00:00Z', + payload: { capabilities: { graceful_close: true } }, + }) }); + + let rejected = false; + void connection.closeGracefully().catch(() => { rejected = true; }); + await vi.advanceTimersByTimeAsync(5_000); + + expect(rejected).toBe(true); + expect(socket.close).toHaveBeenCalled(); + }); + + test('reconnects after a retryable server error even when the socket closes normally', async () => { + vi.useFakeTimers(); + installBrowser(); + const connection = new VoiceConnection({ + conversationId: 'conv_1', + getConnectionKey: async () => 'key-1', + webSocketFactory: () => new FakeWebSocket() as unknown as WebSocket, + }); + await connection.connect(); + FakeWebSocket.instances[0].emit('message', { data: JSON.stringify({ + version: 'voice.realtime.v1', type: 'error', event_id: 'error-1', sequence: 1, + conversation_id: 'conv_1', timestamp: '2026-08-15T00:00:00Z', + payload: { code: 'service_restarting', retryable: true, retry_after_ms: 500 }, + }) }); + FakeWebSocket.instances[0].emit('close', { code: 1000 }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(FakeWebSocket.instances).toHaveLength(2); + }); +}); diff --git a/client/src/voice/voiceConnection.ts b/client/src/voice/voiceConnection.ts new file mode 100644 index 0000000..718e102 --- /dev/null +++ b/client/src/voice/voiceConnection.ts @@ -0,0 +1,249 @@ +const VERSION = 'voice.realtime.v1'; +const RETRY_DELAYS = [1000, 2000, 4000]; +const RETRY_CODES = new Set([1006, 1011, 1012, 1013]); +const HEARTBEAT_INTERVAL_MS = 25_000; +const GRACEFUL_CLOSE_TIMEOUT_MS = 5_000; + +export interface VoiceServerEvent { version: string; type: string; event_id: string; sequence: number; conversation_id: string; timestamp: string; work_id?: string; announcement_id?: string; payload: Record } +export type VoiceConnectionState = 'connecting' | 'reconnecting' | 'disconnected' | 'replaced'; +export interface VoiceCloseResult { outcome: string } +export class VoiceClientError extends Error { code: string; constructor(code: string, message: string) { super(message); this.code = code; } } + +export function buildLocalVoiceSocketUrl(key: string, base = window.location.href) { + const url = new URL('/api/voice/socket', base); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + url.searchParams.set('key', key); + return url.toString(); +} + +export function isValidVoiceServerEvent(value: unknown, conversationId: string): value is VoiceServerEvent { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const event = value as Partial; + const validWorkId = event.work_id === undefined || (typeof event.work_id === 'string' && !!event.work_id); + const validAnnouncementId = event.announcement_id === undefined || (typeof event.announcement_id === 'string' && !!event.announcement_id); + return event.version === VERSION && typeof event.type === 'string' && !!event.type && typeof event.event_id === 'string' && !!event.event_id && Number.isInteger(event.sequence) && Number(event.sequence) > 0 && event.conversation_id === conversationId && typeof event.timestamp === 'string' && !!event.timestamp && validWorkId && validAnnouncementId && !!event.payload && typeof event.payload === 'object' && !Array.isArray(event.payload); +} + +interface VoiceConnectionOptions { + conversationId: string; + getConnectionKey: () => Promise; + beforeReconnect?: () => Promise; + onEvent?: (event: VoiceServerEvent) => void; + onState?: (state: VoiceConnectionState) => void; + onReconnect?: (attempt: number, maxAttempts: number) => void; + onError?: (error: VoiceClientError) => void; + webSocketFactory?: (url: string) => WebSocket; +} + +export class VoiceConnection { + ready = false; + private options: VoiceConnectionOptions; + private socket: WebSocket | null = null; + private generation = 0; + private attemptActive = false; + private reconnectAttempt = 0; + private reconnectTimer: number | null = null; + private heartbeat: number | null = null; + private manual = false; + private replaced = false; + private retryOnClose = false; + private requestedRetryAfterMs = 0; + private lastSequence = 0; + private seen = new Set(); + private graceful = false; + private pendingClose: { + requestId: string; + promise: Promise; + resolve: (result: VoiceCloseResult) => void; + reject: (error: VoiceClientError) => void; + timeout: number; + } | null = null; + + constructor(options: VoiceConnectionOptions) { this.options = options; } + + private handlePageHide = () => this.disconnect(); + + async connect() { + if (this.attemptActive) throw new VoiceClientError('connection_in_progress', '连接正在建立中'); + if (this.socket) this.disconnect(); + this.manual = false; + this.replaced = false; + this.reconnectAttempt = 0; + this.graceful = false; + this.seen.clear(); + this.clearReconnectTimer(); + this.stopHeartbeat(); + window.addEventListener('pagehide', this.handlePageHide); + await this.open(false); + } + + private async open(reconnecting: boolean) { + const generation = ++this.generation; + this.attemptActive = true; + this.lastSequence = 0; + this.retryOnClose = false; + this.requestedRetryAfterMs = 0; + this.stopHeartbeat(); + this.options.onState?.(reconnecting ? 'reconnecting' : 'connecting'); + try { + const key = await this.options.getConnectionKey(); + if (generation !== this.generation || this.manual) return; + const socket = (this.options.webSocketFactory || ((url) => new WebSocket(url)))(buildLocalVoiceSocketUrl(key)); + this.socket = socket; + socket.addEventListener('message', (message) => this.handleMessage(generation, message)); + socket.addEventListener('close', (event) => this.handleClose(generation, event)); + socket.addEventListener('error', () => { + if (generation !== this.generation) return; + this.ready = false; + this.attemptActive = false; + this.stopHeartbeat(); + if (!this.manual) this.scheduleReconnect(); + }); + } catch { + this.attemptActive = false; + if (!this.manual) this.scheduleReconnect(); + } + } + + private handleMessage(generation: number, message: MessageEvent) { + if (generation !== this.generation) return; + let event: unknown; + try { event = JSON.parse(String(message.data)); } catch { + this.options.onError?.(new VoiceClientError('invalid_server_event', '收到无效的服务端事件')); + return; + } + if (!isValidVoiceServerEvent(event, this.options.conversationId) || event.sequence <= this.lastSequence || this.seen.has(event.event_id)) return; + this.lastSequence = event.sequence; + this.seen.add(event.event_id); + if (this.seen.size > 512) this.seen.delete(this.seen.values().next().value!); + if (event.type === 'voice.ready') { + this.ready = true; + this.attemptActive = false; + this.reconnectAttempt = 0; + this.graceful = (event.payload.capabilities as Record | undefined)?.graceful_close === true; + this.startHeartbeat(); + } + if (event.type === 'voice.replaced') { this.replaced = true; this.ready = false; } + if (event.type === 'error' && (event.payload.retryable === true || event.payload.code === 'service_restarting')) { + this.retryOnClose = true; + const retryAfterMs = Number(event.payload.retry_after_ms); + this.requestedRetryAfterMs = Number.isFinite(retryAfterMs) && retryAfterMs > 0 ? retryAfterMs : 0; + } + this.options.onEvent?.(event); + if (event.type === 'connection.closed' && this.pendingClose) { + const requestId = event.payload.request_id; + const outcome = event.payload.outcome; + if (requestId === this.pendingClose.requestId && typeof outcome === 'string' && outcome) { + const pending = this.pendingClose; + this.pendingClose = null; + window.clearTimeout(pending.timeout); + this.manual = true; + this.ready = false; + this.stopHeartbeat(); + pending.resolve({ outcome }); + } + } + } + + private handleClose(generation: number, event: CloseEvent) { + if (generation !== this.generation) return; + this.ready = false; + this.attemptActive = false; + this.socket = null; + this.stopHeartbeat(); + if (this.pendingClose) { + const pending = this.pendingClose; + this.pendingClose = null; + window.clearTimeout(pending.timeout); + pending.reject(new VoiceClientError('graceful_close_failed', '语音连接未能安全结束')); + } + if (this.replaced) this.options.onState?.('replaced'); + else if (!this.manual && (this.retryOnClose || RETRY_CODES.has(event.code))) this.scheduleReconnect(); + else this.options.onState?.('disconnected'); + } + + private scheduleReconnect() { + if (this.manual || this.replaced || this.reconnectTimer !== null) return; + if (this.reconnectAttempt >= RETRY_DELAYS.length) { + this.options.onError?.(new VoiceClientError('realtime_connection_failed', '实时连接恢复失败')); + this.options.onState?.('disconnected'); + return; + } + const attempt = ++this.reconnectAttempt; + const delay = Math.max(RETRY_DELAYS[attempt - 1], this.requestedRetryAfterMs); + this.options.onReconnect?.(attempt, RETRY_DELAYS.length); + this.options.onState?.('reconnecting'); + const generation = this.generation; + this.reconnectTimer = window.setTimeout(() => { + this.reconnectTimer = null; + void this.prepareReconnect(generation); + }, delay); + } + + private async prepareReconnect(generation: number) { + if (this.manual || this.replaced || generation !== this.generation) return; + try { await this.options.beforeReconnect?.(); } catch { + if (!this.manual && !this.replaced && generation === this.generation) this.scheduleReconnect(); + return; + } + if (!this.manual && !this.replaced && generation === this.generation) await this.open(true); + } + + send(type: string, payload: Record = {}) { + if ((type === 'audio.append' || type === 'text.message') && !this.ready) return false; + if (!this.socket || this.socket.readyState !== WebSocket.OPEN) return false; + this.socket.send(JSON.stringify({ version: VERSION, type, payload })); + return true; + } + + closeGracefully(): Promise { + if (this.pendingClose) return this.pendingClose.promise; + if (!this.graceful || !this.ready) { + this.disconnect(); + return Promise.resolve({ outcome: 'unsupported' }); + } + const requestId = crypto.randomUUID(); + let resolveClose!: (result: VoiceCloseResult) => void; + let rejectClose!: (error: VoiceClientError) => void; + const promise = new Promise((resolve, reject) => { resolveClose = resolve; rejectClose = reject; }); + const timeout = window.setTimeout(() => { + const pending = this.pendingClose; + if (!pending || pending.requestId !== requestId) return; + this.pendingClose = null; + pending.reject(new VoiceClientError('graceful_close_timeout', '语音连接结束确认超时')); + this.disconnect(); + }, GRACEFUL_CLOSE_TIMEOUT_MS); + this.pendingClose = { requestId, promise, resolve: resolveClose, reject: rejectClose, timeout }; + this.manual = true; + this.stopHeartbeat(); + if (!this.send('connection.close', { request_id: requestId })) { + window.clearTimeout(timeout); + this.pendingClose = null; + this.disconnect(); + rejectClose(new VoiceClientError('graceful_close_failed', '语音连接未能安全结束')); + } + return promise; + } + + disconnect() { + this.manual = true; + this.generation += 1; + this.ready = false; + this.attemptActive = false; + this.clearReconnectTimer(); + this.stopHeartbeat(); + if (this.pendingClose) { + const pending = this.pendingClose; + this.pendingClose = null; + window.clearTimeout(pending.timeout); + pending.reject(new VoiceClientError('graceful_close_cancelled', '语音连接已终止')); + } + window.removeEventListener('pagehide', this.handlePageHide); + if (this.socket && this.socket.readyState < WebSocket.CLOSING) this.socket.close(1000, 'voice disconnect'); + this.socket = null; + } + + private clearReconnectTimer() { if (this.reconnectTimer !== null) window.clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } + private startHeartbeat() { this.stopHeartbeat(); this.heartbeat = window.setInterval(() => this.send('ping'), HEARTBEAT_INTERVAL_MS); } + private stopHeartbeat() { if (this.heartbeat !== null) window.clearInterval(this.heartbeat); this.heartbeat = null; } +} diff --git a/client/src/voice/voiceSession.test.ts b/client/src/voice/voiceSession.test.ts new file mode 100644 index 0000000..0ca3111 --- /dev/null +++ b/client/src/voice/voiceSession.test.ts @@ -0,0 +1,12 @@ +import { describe, expect, test } from 'vitest'; +import type { ForwardSession } from '../forwardApi'; +import { isVoiceSession } from './voiceSession'; + +describe('isVoiceSession', () => { + test('narrows voice-gateway metadata to a required conversation id', () => { + const voice = { id: 'sess_voice', metadata: { source: 'voice-gateway', conversation_id: 'conv_1' } } as ForwardSession; + expect(isVoiceSession(voice)).toBe(true); + if (isVoiceSession(voice)) expect(voice.metadata.conversation_id).toBe('conv_1'); + expect(isVoiceSession({ ...voice, metadata: { source: 'forward-quickstart' } })).toBe(false); + }); +}); diff --git a/client/src/voice/voiceSession.ts b/client/src/voice/voiceSession.ts new file mode 100644 index 0000000..a6e903e --- /dev/null +++ b/client/src/voice/voiceSession.ts @@ -0,0 +1,12 @@ +import type { ForwardSession } from '../forwardApi'; + +export type VoiceSession = ForwardSession & { + metadata: Record & { + source: 'voice-gateway'; + conversation_id: string; + }; +}; + +export function isVoiceSession(session: ForwardSession): session is VoiceSession { + return session.metadata?.source === 'voice-gateway'; +} diff --git a/client/src/voice/voiceTimeline.test.ts b/client/src/voice/voiceTimeline.test.ts new file mode 100644 index 0000000..ce4cdcc --- /dev/null +++ b/client/src/voice/voiceTimeline.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, test } from 'vitest'; +import { applyVoiceTimelineEvent, createVoiceTimelineState, projectVoiceHistory, selectTimelineEntries, upsertLocalUserTextDraft } from './voiceTimeline'; + +function transcript(type: 'transcript.delta' | 'transcript.final', eventId: string, text: string, itemId = 'item-1') { + return { + version: 'voice.realtime.v1', type, event_id: eventId, sequence: 1, + conversation_id: 'conv_1', timestamp: '2026-08-15T00:00:00Z', + payload: { role: 'user' as const, text, item_id: itemId }, + }; +} + +describe('voice timeline', () => { + test('projects history into captions and one work card', () => { + const events = [ + { id: 'done', type: 'voice.work.completed', status: 'completed', work_id: 'work_1', result: '完成', occurred_at: '2026-01-01T00:00:03Z' }, + { id: 'msg', type: 'voice.user_message.completed', status: 'completed', role: 'user' as const, text: '查一下', turn_id: 'turn_1', occurred_at: '2026-01-01T00:00:01Z' }, + { id: 'run', type: 'voice.work.running', status: 'running', work_id: 'work_1', objective: '查资料', turn_id: 'turn_1', occurred_at: '2026-01-01T00:00:02Z' }, + ]; + const result = projectVoiceHistory(events); + expect(result).toHaveLength(2); + expect(result[1]).toMatchObject({ kind: 'work', workId: 'work_1', objective: '查资料', status: 'completed', result: '完成' }); + }); + test('updates live work in place and does not regress terminal state', () => { + let state = createVoiceTimelineState(); + const event = (type: string, sequence: number, payload: Record = {}) => ({ version: 'voice.realtime.v1', type, event_id: `evt_${sequence}`, sequence, conversation_id: 'conv_1', timestamp: '', work_id: 'work_1', payload }); + state = applyVoiceTimelineEvent(state, event('work.running', 1, { objective: '查资料' })); + state = applyVoiceTimelineEvent(state, event('work.completed', 2, { result: '完成' })); + state = applyVoiceTimelineEvent(state, event('work.progress', 3, { detail: '迟到' })); + expect(selectTimelineEntries(state)).toEqual([{ kind: 'work', id: 'work_1', workId: 'work_1', objective: '查资料', status: 'completed', steps: [], result: '完成' }]); + }); + + test('reconciles the gateway echo with the optimistic user bubble', () => { + let state = upsertLocalUserTextDraft(createVoiceTimelineState(), 'local-1', '帮我看看有哪些文件 ...', false); + state = applyVoiceTimelineEvent(state, { + version: 'voice.realtime.v1', + type: 'transcript.final', + event_id: 'server-1', + sequence: 1, + conversation_id: 'conv_1', + timestamp: '2026-08-15T00:00:00Z', + payload: { role: 'user', text: '帮我看看有哪些文件' }, + }); + + expect(selectTimelineEntries(state)).toEqual([ + { kind: 'caption', id: 'local-1', role: 'user', text: '帮我看看有哪些文件', pending: false }, + ]); + }); + + test('keeps one stream bubble and ignores events after its final', () => { + const event = (type: string, eventId: string, text: string) => ({ + version: 'voice.realtime.v1', type, event_id: eventId, sequence: 1, + conversation_id: 'conv_1', timestamp: '2026-08-15T00:00:00Z', + payload: { role: 'assistant', text, response_id: 'response-1' }, + }); + let state = createVoiceTimelineState(); + state = applyVoiceTimelineEvent(state, event('transcript.delta', 'delta-1', '北')); + state = applyVoiceTimelineEvent(state, event('transcript.delta', 'delta-2', '京')); + state = applyVoiceTimelineEvent(state, event('transcript.final', 'final-1', '北京')); + state = applyVoiceTimelineEvent(state, event('transcript.delta', 'late-1', '迟到内容')); + + expect(selectTimelineEntries(state)).toEqual([ + { kind: 'caption', id: 'delta-1', role: 'assistant', text: '北京', pending: false }, + ]); + }); + + test('replaces user transcript snapshots instead of concatenating them', () => { + let state = createVoiceTimelineState(); + state = applyVoiceTimelineEvent(state, transcript('transcript.delta', 'delta-1', '帮我去 GitHub')); + state = applyVoiceTimelineEvent(state, transcript('transcript.delta', 'delta-2', '帮我去 GitHub 上查询 agent sdk')); + state = applyVoiceTimelineEvent(state, transcript('transcript.delta', 'delta-3', '帮我去 GitHub 上查询 agent sdk，然后下载')); + + expect(selectTimelineEntries(state)).toEqual([ + { kind: 'caption', id: 'delta-1', role: 'user', text: '帮我去 GitHub 上查询 agent sdk，然后下载', pending: true }, + ]); + }); + + test.each([ + { + name: 'keeps the richer snapshot when smart-turn final is only a punctuated prefix', + delta: '请查询一下杭州的天气。', + final: '请查询一下杭州的。', + expected: '请查询一下杭州的天气。', + }, + { + name: 'uses a substantially corrected ASR final', + delta: '在GitHub上搜索ent Bay 满到码下载到沙响本地。', + final: '在GitHub上搜索 agentbase/sdk，并下载到沙箱本地。', + expected: '在GitHub上搜索 agentbase/sdk，并下载到沙箱本地。', + }, + ])('$name', ({ delta, final, expected }) => { + let state = createVoiceTimelineState(); + state = applyVoiceTimelineEvent(state, transcript('transcript.delta', 'delta-1', delta)); + state = applyVoiceTimelineEvent(state, transcript('transcript.final', 'final-1', final)); + + expect(selectTimelineEntries(state)).toEqual([ + { kind: 'caption', id: 'delta-1', role: 'user', text: expected, pending: false }, + ]); + }); +}); diff --git a/client/src/voice/voiceTimeline.ts b/client/src/voice/voiceTimeline.ts new file mode 100644 index 0000000..584a206 --- /dev/null +++ b/client/src/voice/voiceTimeline.ts @@ -0,0 +1,233 @@ +import type { RealtimeHistoryEvent } from './voiceApi'; +import type { VoiceServerEvent } from './voiceConnection'; + +export interface CaptionEntry { kind: 'caption'; id: string; role: 'user' | 'assistant'; text: string; pending: boolean; turnId?: string } +export interface WorkStep { kind: string; title?: string; detail?: string; isError?: boolean } +export interface WorkEntry { kind: 'work'; id: string; workId: string; objective: string; status: string; steps: WorkStep[]; result?: string; turnId?: string } +export type TimelineEntry = CaptionEntry | WorkEntry; + +interface TimelineEntity { entry: TimelineEntry; streamId?: string; terminal: boolean } +export interface VoiceTimelineState { + order: string[]; + entities: Record; + activeCaptions: Partial>; + completedStreams: Record; + sealedStreams: Record; + localUserTextBySignature: Record; +} + +const terminalWorkStatuses = new Set(['completed', 'failed', 'cancelled']); +const MAX_WORK_STEPS = 40; + +export function createVoiceTimelineState(): VoiceTimelineState { + return { order: [], entities: {}, activeCaptions: {}, completedStreams: {}, sealedStreams: {}, localUserTextBySignature: {} }; +} + +export function createVoiceTimelineStateFromEntries(entries: TimelineEntry[]): VoiceTimelineState { + const state = createVoiceTimelineState(); + for (const entry of entries) { + const key = entry.kind === 'work' ? `work:${entry.workId}` : `message:${entry.id}`; + if (state.entities[key]) continue; + state.order.push(key); + state.entities[key] = { entry, terminal: entry.kind === 'caption' ? !entry.pending : terminalWorkStatuses.has(entry.status) }; + } + return state; +} + +export function selectTimelineEntries(state: VoiceTimelineState): TimelineEntry[] { + return state.order.flatMap((key) => state.entities[key] ? [state.entities[key].entry] : []); +} + +function textValue(value: unknown): string | undefined { return typeof value === 'string' && value ? value : undefined; } +function trimmedTextValue(value: unknown): string | undefined { return typeof value === 'string' && value.trim() ? value.trim() : undefined; } +function normalizeTranscript(text: string): string { return text.normalize('NFKC').toLocaleLowerCase().replace(/[\s\p{P}\p{S}]+/gu, ''); } +function transcriptSignature(role: 'user' | 'assistant', text: string): string { const normalized = normalizeTranscript(text); return normalized ? `${role}:${normalized}` : ''; } +function reconcileUserTranscriptFinal(delta: string, final: string): string { + const comparableDelta = delta.normalize('NFKC'); + const finalWithoutTrailingSymbols = final.normalize('NFKC').trim().replace(/[\s\p{P}\p{S}]+$/gu, ''); + return finalWithoutTrailingSymbols && comparableDelta.includes(finalWithoutTrailingSymbols) ? delta : final; +} +function transcriptStreamId(event: VoiceServerEvent, role: 'user' | 'assistant'): string | undefined { return textValue(role === 'assistant' ? event.payload.response_id : event.payload.item_id); } +function streamKey(role: 'user' | 'assistant', streamId: string | undefined): string | undefined { return streamId ? `${role}:${streamId}` : undefined; } + +function sealActiveCaption(state: VoiceTimelineState, role: 'user' | 'assistant'): VoiceTimelineState { + const key = state.activeCaptions[role]; + if (!key) return state; + const current = state.entities[key]; + const activeCaptions = { ...state.activeCaptions }; + delete activeCaptions[role]; + if (!current || current.entry.kind !== 'caption' || current.terminal) return { ...state, activeCaptions }; + const sealedKey = streamKey(role, current.streamId); + return { + ...state, + activeCaptions, + sealedStreams: sealedKey ? { ...state.sealedStreams, [sealedKey]: current.entry.text } : state.sealedStreams, + entities: { ...state.entities, [key]: { ...current, entry: { ...current.entry, pending: false }, terminal: true } }, + }; +} + +function applyTranscript(state: VoiceTimelineState, event: VoiceServerEvent): VoiceTimelineState { + const final = event.type === 'transcript.final'; + if (!final && event.type !== 'transcript.delta') return state; + const role = event.payload.role; + const text = textValue(event.payload.text); + if ((role !== 'user' && role !== 'assistant') || text === undefined) return state; + + const streamId = transcriptStreamId(event, role); + const completedKey = streamKey(role, streamId); + if (completedKey && state.completedStreams[completedKey]) return state; + const sealedText = completedKey ? state.sealedStreams[completedKey] : undefined; + if (completedKey && sealedText !== undefined) { + if (!final) return state; + if (text === sealedText) return { ...state, completedStreams: { ...state.completedStreams, [completedKey]: true } }; + } + + let next = role === 'user' ? sealActiveCaption(state, 'assistant') : state; + const signature = final ? transcriptSignature(role, text) : ''; + const localKey = role === 'user' && signature ? next.localUserTextBySignature[signature] : undefined; + const local = localKey ? next.entities[localKey] : undefined; + if (localKey && local?.entry.kind === 'caption' && local.entry.role === 'user') { + const localUserTextBySignature = { ...next.localUserTextBySignature }; + delete localUserTextBySignature[signature]; + return { + ...next, + activeCaptions: final ? { ...next.activeCaptions, user: undefined } : next.activeCaptions, + completedStreams: final && completedKey ? { ...next.completedStreams, [completedKey]: true } : next.completedStreams, + localUserTextBySignature, + entities: { ...next.entities, [localKey]: { ...local, streamId, entry: { ...local.entry, text, pending: !final }, terminal: final } }, + }; + } + + let key = next.activeCaptions[role]; + let current = key ? next.entities[key] : undefined; + if (current?.streamId && streamId && current.streamId !== streamId) { + next = sealActiveCaption(next, role); + key = undefined; + current = undefined; + } + if (current && key) { + if (current.entry.kind !== 'caption' || current.entry.role !== role || current.terminal) return next; + const nextText = final && role === 'user' + ? reconcileUserTranscriptFinal(current.entry.text, text) + : final || role === 'user' + ? text + : current.entry.text + text; + const entry: CaptionEntry = { ...current.entry, text: nextText, pending: !final }; + return { + ...next, + activeCaptions: final ? { ...next.activeCaptions, [role]: undefined } : next.activeCaptions, + completedStreams: final && completedKey ? { ...next.completedStreams, [completedKey]: true } : next.completedStreams, + entities: { ...next.entities, [key]: { ...current, entry, terminal: final } }, + }; + } + + key = `message:${event.event_id}`; + return { + ...next, + order: [...next.order, key], + activeCaptions: final ? next.activeCaptions : { ...next.activeCaptions, [role]: key }, + completedStreams: final && completedKey ? { ...next.completedStreams, [completedKey]: true } : next.completedStreams, + entities: { ...next.entities, [key]: { entry: { kind: 'caption', id: event.event_id, role, text, pending: !final }, streamId, terminal: final } }, + }; +} + +export function upsertLocalUserTextDraft(state: VoiceTimelineState, id: string, text: string, pending: boolean): VoiceTimelineState { + const key = `message:${id}`; + const current = state.entities[key]; + const previousSignature = current?.entry.kind === 'caption' ? transcriptSignature('user', current.entry.text) : ''; + const nextSignature = transcriptSignature('user', text); + const localUserTextBySignature = { ...state.localUserTextBySignature }; + if (previousSignature) delete localUserTextBySignature[previousSignature]; + if (nextSignature) localUserTextBySignature[nextSignature] = key; + return { + ...state, + order: current ? state.order : [...state.order, key], + localUserTextBySignature, + entities: { ...state.entities, [key]: { entry: { kind: 'caption', id, role: 'user', text, pending }, terminal: !pending } }, + }; +} + +export function removeTimelineEntry(state: VoiceTimelineState, id: string): VoiceTimelineState { + const key = `message:${id}`; + const current = state.entities[key]; + if (!current) return state; + const entities = { ...state.entities }; + delete entities[key]; + const localUserTextBySignature = { ...state.localUserTextBySignature }; + if (current.entry.kind === 'caption') { + const signature = transcriptSignature(current.entry.role, current.entry.text); + if (signature) delete localUserTextBySignature[signature]; + } + return { ...state, order: state.order.filter((entryKey) => entryKey !== key), entities, localUserTextBySignature }; +} + +function workStatus(event: VoiceServerEvent, current: string): string { + if (event.type === 'work.accepted') return 'accepted'; + if (event.type === 'work.started' || event.type === 'work.running' || event.type === 'work.progress') return 'running'; + if (event.type === 'work.queued') return 'queued'; + if (event.type === 'work.completed') return 'completed'; + if (event.type === 'work.cancelled') return 'cancelled'; + if (event.type === 'work.failed') return (event.payload.error as { code?: unknown } | undefined)?.code === 'work_cancelled' ? 'cancelled' : 'failed'; + return current; +} + +function applyWork(state: VoiceTimelineState, event: VoiceServerEvent): VoiceTimelineState { + if (!event.type.startsWith('work.') || !event.work_id) return state; + const key = `work:${event.work_id}`; + const current = state.entities[key]; + if (current && (current.entry.kind !== 'work' || current.terminal)) return state; + const previous = current?.entry as WorkEntry | undefined; + const status = workStatus(event, textValue(event.payload.status) ?? previous?.status ?? 'accepted'); + const steps = previous ? [...previous.steps] : []; + if (event.type === 'work.progress' && steps.length < MAX_WORK_STEPS) { + const title = trimmedTextValue(event.payload.title); + const detail = trimmedTextValue(event.payload.detail) ?? trimmedTextValue(event.payload.text); + if (title || detail) steps.push({ kind: trimmedTextValue(event.payload.kind) ?? 'message', ...(title ? { title } : {}), ...(detail ? { detail } : {}), ...(event.payload.is_error ? { isError: true } : {}) }); + } + if (event.type === 'work.milestone' && steps.length < MAX_WORK_STEPS) { + const detail = trimmedTextValue(event.payload.summary); + if (detail) steps.push({ kind: 'milestone', detail }); + } + const errorCode = (event.payload.error as { code?: unknown } | undefined)?.code; + const result = textValue(event.payload.result) ?? (status === 'failed' ? textValue(errorCode) : undefined) ?? previous?.result; + const entry: WorkEntry = { + kind: 'work', id: event.work_id, workId: event.work_id, + objective: textValue(event.payload.objective) ?? previous?.objective ?? `后台任务 ${event.work_id.slice(-6)}`, + status, steps, ...(result ? { result } : {}), + }; + return { ...state, order: current ? state.order : [...state.order, key], entities: { ...state.entities, [key]: { entry, terminal: terminalWorkStatuses.has(status) } } }; +} + +export function applyVoiceTimelineEvent(state: VoiceTimelineState, event: VoiceServerEvent): VoiceTimelineState { + if (event.type.startsWith('transcript.')) return applyTranscript(state, event); + if (event.type.startsWith('work.')) return applyWork(state, event); + return state; +} + +export function projectVoiceHistory(events: RealtimeHistoryEvent[]): TimelineEntry[] { + const state = createVoiceTimelineState(); + const seen = new Set(); + for (const event of events.map((event, index) => ({ event, index })).sort((a, b) => Date.parse(a.event.occurred_at) - Date.parse(b.event.occurred_at) || a.index - b.index).map(({ event }) => event)) { + if (seen.has(event.id)) continue; + seen.add(event.id); + if (event.role && event.text && ['voice.user_message.completed', 'voice.assistant_message.completed', 'voice.assistant_message.interrupted'].includes(event.type)) { + const key = `message:${event.id}`; + state.order.push(key); + state.entities[key] = { entry: { kind: 'caption', id: event.id, role: event.role, text: event.text, pending: false, ...(event.turn_id ? { turnId: event.turn_id } : {}) }, terminal: true }; + continue; + } + if (!event.work_id || !event.type.startsWith('voice.work.')) continue; + const key = `work:${event.work_id}`; + const current = state.entities[key]?.entry as WorkEntry | undefined; + const status = event.type.endsWith('progress_announced') ? 'running' : event.type.split('.').at(-1)!; + const card: WorkEntry = current || { kind: 'work', id: event.work_id, workId: event.work_id, objective: event.objective || `后台任务 ${event.work_id.slice(-6)}`, status: 'accepted', steps: [], ...(event.turn_id ? { turnId: event.turn_id } : {}) }; + if (!current) state.order.push(key); + if (event.objective) card.objective = event.objective; + card.status = event.error?.code === 'work_cancelled' ? 'cancelled' : status; + if (event.type.endsWith('progress_announced') && event.text) card.steps.push({ kind: 'message', detail: event.text }); + if (event.result) card.result = event.result; + if (event.error?.code && card.status === 'failed') card.result = event.error.code; + state.entities[key] = { entry: card, terminal: terminalWorkStatuses.has(card.status) }; + } + return selectTimelineEntries(state); +} diff --git a/client/tsconfig.app.json b/client/tsconfig.app.json index 4689896..71cc1ad 100644 --- a/client/tsconfig.app.json +++ b/client/tsconfig.app.json @@ -21,6 +21,6 @@ "erasableSyntaxOnly": true, "noFallthroughCasesInSwitch": true }, - "include": ["src/main.tsx", "src/App.tsx", "src/forwardApi.ts", "src/config/product.ts"], + "include": ["src"], "exclude": ["src/**/*.test.ts", "src/**/*.test.tsx"] } diff --git a/client/vite.config.ts b/client/vite.config.ts index 1dd4532..ad5fe39 100644 --- a/client/vite.config.ts +++ b/client/vite.config.ts @@ -8,8 +8,9 @@ export default defineConfig({ host: '0.0.0.0', proxy: { '/api': { - target: 'http://localhost:3001', + target: 'http://127.0.0.1:3001', changeOrigin: true, + ws: true, // Ensure SSE responses are not buffered by the proxy configure: (proxy) => { proxy.on('proxyRes', (proxyRes) => { diff --git a/package-lock.json b/package-lock.json index 1a2d5ac..a7f6929 100644 --- a/package-lock.json +++ b/package-lock.json @@ -68,6 +68,7 @@ "integrity": "sha512-CGOfOJqWjg2qW/Mb6zNsDm+u5vFQ8DxXfbM09z69p5Z6+mE1ikP2jUXw+j42Pf1XTYED2Rni5f95npYeuwMDQA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@babel/code-frame": "^7.29.0", "@babel/generator": "^7.29.0", @@ -294,6 +295,7 @@ "cpu": [ "ppc64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -310,6 +312,7 @@ "cpu": [ "arm" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -326,6 +329,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -342,6 +346,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -358,6 +363,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -374,6 +380,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -390,6 +397,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -406,6 +414,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -422,6 +431,7 @@ "cpu": [ "arm" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -438,6 +448,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -454,6 +465,7 @@ "cpu": [ "ia32" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -470,6 +482,7 @@ "cpu": [ "loong64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -486,6 +499,7 @@ "cpu": [ "mips64el" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -502,6 +516,7 @@ "cpu": [ "ppc64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -518,6 +533,7 @@ "cpu": [ "riscv64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -534,6 +550,7 @@ "cpu": [ "s390x" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -550,6 +567,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -566,6 +584,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -582,6 +601,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -598,6 +618,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -614,6 +635,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -630,6 +652,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -646,6 +669,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -662,6 +686,7 @@ "cpu": [ "arm64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -678,6 +703,7 @@ "cpu": [ "ia32" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -694,6 +720,7 @@ "cpu": [ "x64" ], + "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1603,6 +1630,7 @@ "resolved": "https://registry.npmjs.org/@types/node/-/node-24.12.2.tgz", "integrity": "sha512-A1sre26ke7HDIuY/M23nd9gfB+nrmhtYyMINbjI1zHJxYteKR6qSMX56FsmjMcDb3SMcjJg5BiRRgOCC/yBD0g==", "license": "MIT", + "peer": true, "dependencies": { "undici-types": "~7.16.0" } @@ -1625,6 +1653,7 @@ "integrity": "sha512-ilcTH/UniCkMdtexkoCN0bI7pMcJDvmQFPvuPvmEaYA/NSfFTAgdUSLAoVjaRJm7+6PvcM+q1zYOwS4wTYMF9w==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "csstype": "^3.2.2" } @@ -1658,6 +1687,16 @@ "@types/node": "*" } }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@typescript-eslint/eslint-plugin": { "version": "8.59.1", "resolved": "https://registry.npmjs.org/@typescript-eslint/eslint-plugin/-/eslint-plugin-8.59.1.tgz", @@ -1703,6 +1742,7 @@ "integrity": "sha512-HDQH9O/47Dxi1ceDhBXdaldtf/WV9yRYMjbjCuNk3qnaTD564qwv61Y7+gTxwxRKzSrgO5uhtw584igXVuuZkA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@typescript-eslint/scope-manager": "8.59.1", "@typescript-eslint/types": "8.59.1", @@ -2059,6 +2099,7 @@ "integrity": "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw==", "dev": true, "license": "MIT", + "peer": true, "bin": { "acorn": "bin/acorn" }, @@ -2215,6 +2256,7 @@ } ], "license": "MIT", + "peer": true, "dependencies": { "baseline-browser-mapping": "^2.10.12", "caniuse-lite": "^1.0.30001782", @@ -2664,6 +2706,7 @@ "devOptional": true, "hasInstallScript": true, "license": "MIT", + "peer": true, "bin": { "esbuild": "bin/esbuild" }, @@ -2734,6 +2777,7 @@ "integrity": "sha512-wiyGaKsDgqXvF40P8mDwiUp/KQjE1FdrIEJsM8PZ3XCiniTMXS3OHWWUe5FI5agoCnr8x4xPrTDZuxsBlNHl+Q==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@eslint-community/eslint-utils": "^4.8.0", "@eslint-community/regexpp": "^4.12.2", @@ -4130,6 +4174,7 @@ "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.4.tgz", "integrity": "sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==", "license": "MIT", + "peer": true, "engines": { "node": ">=12" }, @@ -4242,6 +4287,7 @@ "resolved": "https://registry.npmjs.org/react/-/react-19.2.5.tgz", "integrity": "sha512-llUJLzz1zTUBrskt2pwZgLq59AemifIftw4aB7JxOqf1HY2FDaGDxgwpAPVzHU1kdWabH7FauP4i1oEeer2WCA==", "license": "MIT", + "peer": true, "engines": { "node": ">=0.10.0" } @@ -4765,6 +4811,7 @@ "integrity": "sha512-5C1sg4USs1lfG0GFb2RLXsdpXqBSEhAaA/0kPL01wxzpMqLILNxIxIOKiILz+cdg/pLnOUxFYOR5yhHU666wbw==", "devOptional": true, "license": "MIT", + "peer": true, "dependencies": { "esbuild": "~0.27.0", "get-tsconfig": "^4.7.5" @@ -4818,6 +4865,7 @@ "integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==", "dev": true, "license": "Apache-2.0", + "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -4926,6 +4974,7 @@ "resolved": "https://registry.npmjs.org/vite/-/vite-8.0.10.tgz", "integrity": "sha512-rZuUu9j6J5uotLDs+cAA4O5H4K1SfPliUlQwqa6YEwSrWDZzP4rhm00oJR5snMewjxF5V/K3D4kctsUTsIU9Mw==", "license": "MIT", + "peer": true, "dependencies": { "lightningcss": "^1.32.0", "picomatch": "^4.0.4", @@ -5155,6 +5204,27 @@ "integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==", "license": "ISC" }, + "node_modules/ws": { + "version": "8.21.3", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.3.tgz", + "integrity": "sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, "node_modules/y18n": { "version": "5.0.8", "resolved": "https://registry.npmjs.org/y18n/-/y18n-5.0.8.tgz", @@ -5220,6 +5290,7 @@ "integrity": "sha512-a6ENMBBGZBsnlSebQ/eKCguSBeGKSf4O7BPnqVPmYGtpBYI7VSqoVqw+QcB7kPRjbqPwhYTpFbVj/RqNz/CT0Q==", "dev": true, "license": "MIT", + "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" } @@ -5244,12 +5315,14 @@ "cors": "^2.8.5", "dotenv": "^16.5.0", "express": "^5.1.0", - "multer": "^2.2.0" + "multer": "^2.2.0", + "ws": "^8.21.3" }, "devDependencies": { "@types/cors": "^2.8.18", "@types/express": "^5.0.2", "@types/node": "^22.15.3", + "@types/ws": "^8.18.1", "tsx": "^4.19.4", "typescript": "^5.8.3" } diff --git a/server/package.json b/server/package.json index 39ae3f2..f1fc1af 100644 --- a/server/package.json +++ b/server/package.json @@ -5,6 +5,7 @@ "scripts": { "dev": "tsx watch src/index.ts", "build": "tsc", + "test": "tsx --test src/**/*.test.ts", "start": "node dist/index.js" }, "dependencies": { @@ -12,12 +13,14 @@ "cors": "^2.8.5", "dotenv": "^16.5.0", "express": "^5.1.0", - "multer": "^2.2.0" + "multer": "^2.2.0", + "ws": "^8.21.3" }, "devDependencies": { "@types/cors": "^2.8.18", "@types/express": "^5.0.2", "@types/node": "^22.15.3", + "@types/ws": "^8.18.1", "tsx": "^4.19.4", "typescript": "^5.8.3" } diff --git a/server/src/index.ts b/server/src/index.ts index bb9c942..b3c0cc3 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,9 +1,11 @@ import express from 'express'; import cors from 'cors'; import multer from 'multer'; +import { createServer } from 'node:http'; import { appendFileSync, mkdirSync } from 'node:fs'; import { join } from 'node:path'; import './utils/env.js'; +import { createVoiceProxy } from './voiceProxy.js'; // Headroom above the client-side caps (images 10 MB, text files 5 MB) so the // friendly client-side validation is what rejects oversized picks, rather than @@ -21,6 +23,7 @@ const DEFAULT_API_BASE_URLS: Record = { const API_BASE_URLS = Object.fromEntries( Object.entries(DEFAULT_API_BASE_URLS).map(([key, value]) => [key, value.replace(/\/+$/, '')]), ) as Record; +const voiceProxy = createVoiceProxy({ baseUrls: API_BASE_URLS, enabled: !process.env.VERCEL }); const DEFAULT_CLOUD_API_BASE_URLS: Record = { 'cn-prod': process.env.CN_PROD_CLOUD_API_BASE_URL?.trim() || 'https://api.qoder.com.cn/api/v1/cloud', 'global-prod': process.env.GLOBAL_PROD_CLOUD_API_BASE_URL?.trim() || 'https://api.qoder.com/api/v1/cloud', @@ -33,6 +36,7 @@ const LOG_FILE = join(LOG_DIR, 'forward-proxy.log'); app.use(cors()); app.use(express.json({ limit: '50mb' })); +app.post('/api/voice/connect', voiceProxy.issueConnectionKey); function proxyLog(level: 'info' | 'warn', message: string, meta?: Record) { const line = JSON.stringify({ @@ -637,7 +641,7 @@ app.post('/api/cloud/upload', upload.single('file'), makeUploadHandler(CLOUD_API app.post('/api/forward/upload', upload.single('file'), makeUploadHandler(API_BASE_URLS, 'forward')); app.get('/api/health', (_req, res) => { - res.json({ status: 'ok', forwardApiBaseUrls: API_BASE_URLS }); + res.json({ status: 'ok', forwardApiBaseUrls: API_BASE_URLS, voiceRealtimeProxy: { enabled: voiceProxy.enabled, localOnly: true } }); }); export default app; @@ -645,7 +649,9 @@ export default app; // Vercel imports the Express app as a serverless function. Keep the listener // for the existing local development and standalone production commands. if (!process.env.VERCEL) { - app.listen(PORT, () => { + const server = createServer(app); + voiceProxy.attach(server); + server.listen(PORT, () => { console.log(`Forward quickstart server running on http://localhost:${PORT}`); console.log('Forward API targets:', API_BASE_URLS); }); diff --git a/server/src/voiceProxy.test.ts b/server/src/voiceProxy.test.ts new file mode 100644 index 0000000..09808d9 --- /dev/null +++ b/server/src/voiceProxy.test.ts @@ -0,0 +1,73 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { buildRealtimeUrl, createConnectionKeyStore, createVoiceProxy, isAllowedLocalOrigin, relayCloseCode } from './voiceProxy.js'; + +test('connection keys expire and are consumed once', () => { + let now = 1_000; + let serial = 0; + const store = createConnectionKeyStore({ ttlMs: 30_000, now: () => now, randomKey: () => `key_${++serial}` }); + const payload = { pat: 'pat_secret', environment: 'cn-prod' as const, conversationId: 'conv_1' }; + const key = store.issue(payload); + assert.equal(store.consume(key)?.conversationId, 'conv_1'); + assert.equal(store.consume(key), null); + const expired = store.issue(payload); + now += 30_001; + assert.equal(store.consume(expired), null); +}); + +test('unconsumed connection keys are removed after their ttl', () => { + let expire: (() => void) | undefined; + const store = createConnectionKeyStore({ + ttlMs: 30_000, + randomKey: () => 'key_unconsumed', + scheduleExpiry: (callback) => { expire = callback; }, + }); + const key = store.issue({ pat: 'pat_secret', environment: 'cn-prod', conversationId: 'conv_1' }); + + expire?.(); + + assert.equal(store.consume(key), null); +}); + +test('builds CN and Global forward realtime urls', () => { + assert.equal(buildRealtimeUrl('https://api.qoder.com.cn/api/v1/forward', 'conv a').toString(), 'wss://api.qoder.com.cn/api/v1/forward/realtime?conversation_id=conv+a'); + assert.equal(buildRealtimeUrl('https://api.qoder.com/api/v1/forward/', 'conv_2').toString(), 'wss://api.qoder.com/api/v1/forward/realtime?conversation_id=conv_2'); +}); + +test('accepts loopback origins only', () => { + assert.equal(isAllowedLocalOrigin('http://localhost:5173'), true); + assert.equal(isAllowedLocalOrigin('http://127.0.0.1:5173'), true); + assert.equal(isAllowedLocalOrigin('https://attacker.example'), false); + assert.equal(isAllowedLocalOrigin(undefined), false); +}); + +test('replaces reserved websocket close codes before relaying them', () => { + assert.equal(relayCloseCode(1000, 1011), 1000); + assert.equal(relayCloseCode(1001, 1011), 1001); + assert.equal(relayCloseCode(1006, 1011), 1011); + assert.equal(relayCloseCode(1015, 1000), 1000); +}); + +test('connection key endpoint rejects non-loopback browser origins', () => { + const proxy = createVoiceProxy({ + baseUrls: { + 'cn-prod': 'https://api.qoder.com.cn/api/v1/forward', + 'global-prod': 'https://api.qoder.com/api/v1/forward', + }, + }); + let status = 200; + let response: unknown; + const request = { + headers: { origin: 'https://attacker.example' }, + body: { pat: 'pat_secret', environment: 'cn-prod', conversation_id: 'conv_1' }, + }; + const result = { + status(value: number) { status = value; return this; }, + json(value: unknown) { response = value; return this; }, + }; + + proxy.issueConnectionKey(request as never, result as never, () => undefined); + + assert.equal(status, 403); + assert.deepEqual(response, { error: { message: 'Voice WebSocket proxy only accepts loopback browser origins' } }); +}); diff --git a/server/src/voiceProxy.ts b/server/src/voiceProxy.ts new file mode 100644 index 0000000..25c4c4b --- /dev/null +++ b/server/src/voiceProxy.ts @@ -0,0 +1,138 @@ +import { randomBytes } from 'node:crypto'; +import type { Server } from 'node:http'; +import type { RequestHandler } from 'express'; +import { WebSocket, WebSocketServer } from 'ws'; + +export type VoiceEnvironment = 'cn-prod' | 'global-prod'; +export interface VoiceConnectionPayload { pat: string; environment: VoiceEnvironment; conversationId: string } +export interface ConnectionKeyStore { issue(payload: VoiceConnectionPayload): string; consume(key: string): VoiceConnectionPayload | null } + +interface ConnectionKeyStoreOptions { + ttlMs?: number; + now?: () => number; + randomKey?: () => string; + scheduleExpiry?: (callback: () => void, delayMs: number) => void; +} + +function scheduleExpiry(callback: () => void, delayMs: number) { + setTimeout(callback, delayMs).unref(); +} + +export function createConnectionKeyStore({ ttlMs = 30_000, now = Date.now, randomKey = () => randomBytes(32).toString('base64url'), scheduleExpiry: schedule = scheduleExpiry }: ConnectionKeyStoreOptions = {}): ConnectionKeyStore { + const entries = new Map(); + return { + issue(payload) { + const key = randomKey(); + const entry = { payload, expiresAt: now() + ttlMs }; + entries.set(key, entry); + schedule(() => { + if (entries.get(key) === entry) entries.delete(key); + }, ttlMs); + return key; + }, + consume(key) { + const entry = entries.get(key); + entries.delete(key); + return entry && entry.expiresAt >= now() ? entry.payload : null; + }, + }; +} + +export function buildRealtimeUrl(baseUrl: string, conversationId: string): URL { + const url = new URL(`${baseUrl.replace(/\/+$/, '')}/realtime`); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + url.searchParams.set('conversation_id', conversationId); + return url; +} + +export function isAllowedLocalOrigin(origin: string | undefined): boolean { + if (!origin) return false; + try { + const hostname = new URL(origin).hostname; + return hostname === 'localhost' || hostname === '127.0.0.1' || hostname === '[::1]'; + } catch { + return false; + } +} + +export function relayCloseCode(code: number, fallback: number): number { + const protocolCode = code >= 1000 && code <= 1014 && ![1004, 1005, 1006].includes(code); + return protocolCode || (code >= 3000 && code <= 4999) ? code : fallback; +} + +function rejectUpgrade(socket: import('node:stream').Duplex, status: number, message: string) { + socket.write(`HTTP/1.1 ${status} ${message}\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`); + socket.destroy(); +} + +export function createVoiceProxy(options: { + baseUrls: Record; + enabled?: boolean; + ttlMs?: number; +}) { + const enabled = options.enabled ?? true; + const ttlMs = options.ttlMs ?? 30_000; + const store = createConnectionKeyStore({ ttlMs }); + const wss = new WebSocketServer({ noServer: true }); + let attached = false; + + const issueConnectionKey: RequestHandler = (req, res) => { + if (!enabled) { + res.status(501).json({ error: { message: 'Voice WebSocket proxy is available in local development only' } }); + return; + } + if (!isAllowedLocalOrigin(req.headers.origin)) { + res.status(403).json({ error: { message: 'Voice WebSocket proxy only accepts loopback browser origins' } }); + return; + } + const pat = String(req.body?.pat ?? '').trim(); + const environment = req.body?.environment; + const conversationId = String(req.body?.conversation_id ?? '').trim(); + if (!pat || !conversationId || (environment !== 'cn-prod' && environment !== 'global-prod')) { + res.status(400).json({ error: { message: 'pat, environment and conversation_id are required' } }); + return; + } + res.json({ connection_key: store.issue({ pat, environment, conversationId }), expires_in_ms: ttlMs }); + }; + + function attach(server: Server) { + if (!enabled || attached) return; + attached = true; + server.on('upgrade', (request, socket, head) => { + const url = new URL(request.url || '/', 'http://localhost'); + if (url.pathname !== '/api/voice/socket') { + rejectUpgrade(socket, 404, 'Not Found'); + return; + } + if (!isAllowedLocalOrigin(request.headers.origin)) { + rejectUpgrade(socket, 403, 'Forbidden'); + return; + } + const payload = store.consume(url.searchParams.get('key') || ''); + if (!payload) { + rejectUpgrade(socket, 401, 'Unauthorized'); + return; + } + wss.handleUpgrade(request, socket, head, (client) => { + const upstream = new WebSocket(buildRealtimeUrl(options.baseUrls[payload.environment], payload.conversationId), { + headers: { Authorization: `Bearer ${payload.pat}` }, + }); + const closePeer = (peer: WebSocket, code = 1011, reason = 'voice proxy closed', fallback = 1011) => { + if (peer.readyState === WebSocket.OPEN || peer.readyState === WebSocket.CONNECTING) peer.close(relayCloseCode(code, fallback), reason); + }; + client.on('message', (data, isBinary) => { + if (upstream.readyState === WebSocket.OPEN) upstream.send(data, { binary: isBinary }); + }); + upstream.on('message', (data, isBinary) => { + if (client.readyState === WebSocket.OPEN) client.send(data, { binary: isBinary }); + }); + client.on('close', (code) => closePeer(upstream, code, 'client closed', 1000)); + upstream.on('close', (code) => closePeer(client, code, 'upstream closed', 1011)); + client.on('error', () => closePeer(upstream)); + upstream.on('error', () => closePeer(client)); + }); + }); + } + + return { enabled, issueConnectionKey, attach }; +}