实时语音网关 深度解析

86export function attachRealtimeGateway(server, {
87  identityManager,
88  memoryStore,
89  coordinator,
90  coordinatorAvailable = async () => true,
91  respondPermission,
92  permissionPolicy,
93}) {
94  const wss = new WebSocketServer({ noServer: true, maxPayload: 2 * 1024 * 1024 })
95  const activeVoiceClients = new ActiveVoiceClients()
96  const voiceConnections = new Map()
97
98  const broadcastVoiceOwnership = ownerId => {
99    const active = activeVoiceClients.active(ownerId)
100    const holder = active?.descriptor || null
101    for (const client of voiceConnections.get(ownerId) || []) {
102      send(client.ws, {
103        type: 'voice.ownership',
104        state: active === client
105          ? 'active'
106          : holder ? 'busy' : 'available',
107        holder,
108      })
109    }
110  }
111
112  server.on('upgrade', (request, socket, head) => {
113    const url = new URL(request.url, 'http://localhost')
114    if (rejectUnsupportedRealtimeUpgrade(socket, url.pathname)) return
115    if (!isAllowedOrigin(request)) {
116      rejectUpgrade(socket, '403 Forbidden', 'origin not allowed')
117      return
118    }
119    const identity = identityManager.resolveUpgrade(request)
120    if (!identity) {
121      rejectUpgrade(socket, '401 Unauthorized', 'identity required')
122      return
123    }
124    wss.handleUpgrade(request, socket, head, ws => {
125      wss.emit('connection', ws, url, identity)
126    })
127  })

276    if (!voiceConnections.has(ownerId)) voiceConnections.set(ownerId, new Set())
277    voiceConnections.get(ownerId).add(voiceClient)
278
279    const activateVoiceClient = ({
280      takeover = false,
281      enableInput = true,
282      enableOutput = true,
283    } = {}) => {
284      const result = activeVoiceClients.activate(
285        ownerId,
286        voiceClient,
287        { takeover },
288      )
289      inputEnabled = result.granted && enableInput
290      outputEnabled = result.granted && enableOutput
291      broadcastVoiceOwnership(ownerId)
292      return result.granted
293    }
294    const releaseVoiceClient = () => {
295      inputEnabled = false
296      outputEnabled = false
297      if (activeVoiceClients.release(ownerId, voiceClient)) {
298        broadcastVoiceOwnership(ownerId)
299      }
300    }
301    const toolCalls = new ToolCallHandler({
302      taskManager,
303      ownerId,
304      sessionId,
305      transcripts,
306      getFrontend: () => frontend,
307      getTurnId: () => committedTurnId,
308      getTurnGeneration: () => committedTurnGeneration,
309      memoryStore,
310      getClientContext: () => clientContext,
311      getConversationContext: () => conversationSync.frontendContext({
312        ownerId,
313        sessionId,
314      }),
315      onMemoryChanged: () => frontend?.updateAgentContext({
316        memories: memoryStore?.list(ownerId) || [],
317      }),
318      coordinator,
319      coordinatorAvailable,
320      respondPermission,
321      permissionPolicy,
322    })

502    const startPlayback = id => {
503      const context = responseContexts.get(id)
504      // A cancelled response remains as a short-lived tombstone so late
505      // provider audio and client receipts cannot resurrect it.
506      if (context?.suppressed) return
507      announcementWindow.startPlayback(id)
508      const playbackTurnId = context?.turnId || playbackTurns.get(id) || turnId
509      send(ws, {
510        type: 'voice.state',
511        state: 'speaking',
512        turnId: playbackTurnId,
513        origin: context?.origin || 'model',
514      })
515      if (!context || context.playbackStarted) return
516      context.playbackStarted = true
517      if (confirmsTaskNotificationOnPlaybackStart(context)) {
518        announcements.confirmMany(contextTaskIds(context))
519      }
520      flushPendingTranscripts(id, context)
521    }
522
523    const cancelQueuedPlayback = (id, { reason = '' } = {}) => {
524      const context = responseContexts.get(id)
525      announcementWindow.finishPlayback(id, {
526        hasFunctionCall: Boolean(context?.hasFunctionCall),
527      })
528      const playbackTurnId = playbackTurns.get(id) || turnId
529      playbackTurns.delete(id)
530      if (context?.origin === 'announcement') {
531        if (reason === 'user_interruption') {
532          announcements.confirmMany(contextTaskIds(context))
533        } else {
534          announcements.retryMany(contextTaskIds(context))
535        }
536      }
537      if (context?.playbackStarted && reason === 'user_interruption') {
538        send(ws, {
539          type: 'response.interrupted',
540          responseId: id,
541          ...publicResponseContext(context),
542        })
543      }
544      if (context) {
545        context.suppressed = true
546        context.playbackEnded = true
547        context.pendingTranscripts = []
548        scheduleResponseContextCleanup(id, context)
549      }
550      send(ws, {
551        type: 'voice.state',
552        state: userSpeaking ? 'listening' : 'idle',
553        turnId: userSpeaking ? turnId : playbackTurnId,
554        origin: context?.origin || 'model',
555      })
556      const timer = setTimeout(
557        () => announcements.flush(),
558        config.announcementQuietMs,
559      )
560      timer.unref?.()
561    }