codev / src /services /voiceStreamSTT.ts

init: verspercode v0.1.0

064bfd6 7 months ago

21.4 kB

// Anthropic voice_stream speech-to-text client for push-to-talk.
//
// Only reachable in ant builds (gated by feature('VOICE_MODE') in useVoice.ts import).
//
// Connects to Anthropic's voice_stream WebSocket endpoint using the same
// OAuth credentials as Claude Code.  The endpoint uses conversation_engine
// backed models for speech-to-text.  Designed for hold-to-talk: hold the
// keybinding to record, release to stop and submit.
//
// The wire protocol uses JSON control messages (KeepAlive, CloseStream) and
// binary audio frames.  The server responds with TranscriptText and
// TranscriptEndpoint JSON messages.

import type { ClientRequest, IncomingMessage } from 'http'
import WebSocket from 'ws'
import { getOauthConfig } from '../constants/oauth.js'
import {
  checkAndRefreshOAuthTokenIfNeeded,
  getClaudeAIOAuthTokens,
  isAnthropicAuthEnabled,
} from '../utils/auth.js'
import { logForDebugging } from '../utils/debug.js'
import { getUserAgent } from '../utils/http.js'
import { logError } from '../utils/log.js'
import { getWebSocketTLSOptions } from '../utils/mtls.js'
import { getWebSocketProxyAgent, getWebSocketProxyUrl } from '../utils/proxy.js'
import { jsonParse, jsonStringify } from '../utils/slowOperations.js'

const KEEPALIVE_MSG = '{"type":"KeepAlive"}'
const CLOSE_STREAM_MSG = '{"type":"CloseStream"}'

import { getFeatureValue_CACHED_MAY_BE_STALE } from './analytics/growthbook.js'

// ─── Constants ───────────────────────────────────────────────────────

const VOICE_STREAM_PATH = '/api/ws/speech_to_text/voice_stream'

const KEEPALIVE_INTERVAL_MS = 8_000

// finalize() resolution timers. `noData` fires when no TranscriptText
// arrives post-CloseStream — the server has nothing; don't wait out the
// full ~3-5s WS teardown to confirm emptiness. `safety` is the last-
// resort cap if the WS hangs. Exported so tests can shorten them.
export const FINALIZE_TIMEOUTS_MS = {
  safety: 5_000,
  noData: 1_500,
}

// ─── Types ──────────────────────────────────────────────────────────

export type VoiceStreamCallbacks = {
  onTranscript: (text: string, isFinal: boolean) => void
  onError: (error: string, opts?: { fatal?: boolean }) => void
  onClose: () => void
  onReady: (connection: VoiceStreamConnection) => void
}

// How finalize() resolved. `no_data_timeout` means zero server messages
// after CloseStream — the silent-drop signature (anthropics/anthropic#287008).
export type FinalizeSource =
  | 'post_closestream_endpoint'
  | 'no_data_timeout'
  | 'safety_timeout'
  | 'ws_close'
  | 'ws_already_closed'

export type VoiceStreamConnection = {
  send: (audioChunk: Buffer) => void
  finalize: () => Promise<FinalizeSource>
  close: () => void
  isConnected: () => boolean
}

// The voice_stream endpoint returns transcript chunks and endpoint markers.
type VoiceStreamTranscriptText = {
  type: 'TranscriptText'
  data: string
}

type VoiceStreamTranscriptEndpoint = {
  type: 'TranscriptEndpoint'
}

type VoiceStreamTranscriptError = {
  type: 'TranscriptError'
  error_code?: string
  description?: string
}

type VoiceStreamMessage =
  | VoiceStreamTranscriptText
  | VoiceStreamTranscriptEndpoint
  | VoiceStreamTranscriptError
  | { type: 'error'; message?: string }

// ─── Availability ──────────────────────────────────────────────────────

export function isVoiceStreamAvailable(): boolean {
  // voice_stream uses the same OAuth as Claude Code — available when the
  // user is authenticated with Anthropic (Claude.ai subscriber or has
  // valid OAuth tokens).
  if (!isAnthropicAuthEnabled()) {
    return false
  }
  const tokens = getClaudeAIOAuthTokens()
  return tokens !== null && tokens.accessToken !== null
}

// ─── Connection ────────────────────────────────────────────────────────

export async function connectVoiceStream(
  callbacks: VoiceStreamCallbacks,
  options?: { language?: string; keyterms?: string[] },
): Promise<VoiceStreamConnection | null> {
  // Ensure OAuth token is fresh before connecting
  await checkAndRefreshOAuthTokenIfNeeded()

  const tokens = getClaudeAIOAuthTokens()
  if (!tokens?.accessToken) {
    logForDebugging('[voice_stream] No OAuth token available')
    return null
  }

  // voice_stream is a private_api route, but /api/ws/ is also exposed on
  // the api.anthropic.com listener (service_definitions.yaml private-api:
  // visibility.external: true). We target that host instead of claude.ai
  // because the claude.ai CF zone uses TLS fingerprinting and challenges
  // non-browser clients (anthropics/claude-code#34094). Same private-api
  // pod, same OAuth Bearer auth — just a CF zone that doesn't block us.
  // Desktop dictation still uses claude.ai (Swift URLSession has a
  // browser-class JA3 fingerprint, so CF lets it through).
  const wsBaseUrl =
    process.env.VOICE_STREAM_BASE_URL ||
    getOauthConfig()
      .BASE_API_URL.replace('https://', 'wss://')
      .replace('http://', 'ws://')

  if (process.env.VOICE_STREAM_BASE_URL) {
    logForDebugging(
      `[voice_stream] Using VOICE_STREAM_BASE_URL override: ${process.env.VOICE_STREAM_BASE_URL}`,
    )
  }

  const params = new URLSearchParams({
    encoding: 'linear16',
    sample_rate: '16000',
    channels: '1',
    endpointing_ms: '300',
    utterance_end_ms: '1000',
    language: options?.language ?? 'en',
  })

  // Route through conversation-engine with Deepgram Nova 3 (bypassing
  // the server's project_bell_v2_config GrowthBook gate). The server
  // side is anthropics/anthropic#278327 + #281372; this lets us ramp
  // clients independently.
  const isNova3 = getFeatureValue_CACHED_MAY_BE_STALE(
    'tengu_cobalt_frost',
    false,
  )
  if (isNova3) {
    params.set('use_conversation_engine', 'true')
    params.set('stt_provider', 'deepgram-nova3')
    logForDebugging('[voice_stream] Nova 3 gate enabled (tengu_cobalt_frost)')
  }

  // Append keyterms as query params — the voice_stream proxy forwards
  // these to the STT service which applies appropriate boosting.
  if (options?.keyterms?.length) {
    for (const term of options.keyterms) {
      params.append('keyterms', term)
    }
  }

  const url = `${wsBaseUrl}${VOICE_STREAM_PATH}?${params.toString()}`

  logForDebugging(`[voice_stream] Connecting to ${url}`)

  const headers: Record<string, string> = {
    Authorization: `Bearer ${tokens.accessToken}`,
    'User-Agent': getUserAgent(),
    'x-app': 'cli',
  }

  const tlsOptions = getWebSocketTLSOptions()
  const wsOptions =
    typeof Bun !== 'undefined'
      ? {
          headers,
          proxy: getWebSocketProxyUrl(url),
          tls: tlsOptions || undefined,
        }
      : { headers, agent: getWebSocketProxyAgent(url), ...tlsOptions }

  const ws = new WebSocket(url, wsOptions)

  let keepaliveTimer: ReturnType<typeof setInterval> | null = null
  let connected = false
  // Set to true once CloseStream has been sent (or the ws is closed).
  // After this, further audio sends are dropped.
  let finalized = false
  // Set to true when finalize() is first called, to prevent double-fire.
  let finalizing = false
  // Set when the HTTP upgrade was rejected (unexpected-response). The
  // close event that follows (1006 from our req.destroy()) is just
  // mechanical teardown; the upgrade handler already reported the error.
  let upgradeRejected = false
  // Resolves finalize(). Four triggers: TranscriptEndpoint post-CloseStream
  // (~300ms); no-data timer (1.5s); WS close (~3-5s); safety timer (5s).
  let resolveFinalize: ((source: FinalizeSource) => void) | null = null
  let cancelNoDataTimer: (() => void) | null = null

  // Define the connection object before event handlers so it can be passed
  // to onReady when the WebSocket opens.
  const connection: VoiceStreamConnection = {
    send(audioChunk: Buffer): void {
      if (ws.readyState !== WebSocket.OPEN) {
        return
      }
      if (finalized) {
        // After CloseStream has been sent, the server rejects further audio.
        // Drop the chunk to avoid a protocol error.
        logForDebugging(
          `[voice_stream] Dropping audio chunk after CloseStream: ${String(audioChunk.length)} bytes`,
        )
        return
      }
      logForDebugging(
        `[voice_stream] Sending audio chunk: ${String(audioChunk.length)} bytes`,
      )
      // Copy the buffer before sending: NAPI Buffer objects from native
      // modules may share a pooled ArrayBuffer.  Creating a view with
      // `new Uint8Array(buf.buffer, offset, len)` can reference stale or
      // overlapping memory by the time the ws library reads it.
      // `Buffer.from()` makes an owned copy that the ws library can safely
      // consume as a binary WebSocket frame.
      ws.send(Buffer.from(audioChunk))
    },
    finalize(): Promise<FinalizeSource> {
      if (finalizing || finalized) {
        // Already finalized or WebSocket already closed — resolve immediately.
        return Promise.resolve('ws_already_closed')
      }
      finalizing = true

      return new Promise<FinalizeSource>(resolve => {
        const safetyTimer = setTimeout(
          () => resolveFinalize?.('safety_timeout'),
          FINALIZE_TIMEOUTS_MS.safety,
        )
        const noDataTimer = setTimeout(
          () => resolveFinalize?.('no_data_timeout'),
          FINALIZE_TIMEOUTS_MS.noData,
        )
        cancelNoDataTimer = () => {
          clearTimeout(noDataTimer)
          cancelNoDataTimer = null
        }

        resolveFinalize = (source: FinalizeSource) => {
          clearTimeout(safetyTimer)
          clearTimeout(noDataTimer)
          resolveFinalize = null
          cancelNoDataTimer = null
          // Legacy Deepgram can leave an interim in lastTranscriptText
          // with no TranscriptEndpoint (websocket_manager.py sends
          // TranscriptChunk and TranscriptEndpoint as independent
          // channel items). All resolve triggers must promote it;
          // centralize here. No-op when the close handler already did.
          if (lastTranscriptText) {
            logForDebugging(
              `[voice_stream] Promoting unreported interim before ${source} resolve`,
            )
            const t = lastTranscriptText
            lastTranscriptText = ''
            callbacks.onTranscript(t, true)
          }
          logForDebugging(`[voice_stream] Finalize resolved via ${source}`)
          resolve(source)
        }

        // If the WebSocket is already closed, resolve immediately.
        if (
          ws.readyState === WebSocket.CLOSED ||
          ws.readyState === WebSocket.CLOSING
        ) {
          resolveFinalize('ws_already_closed')
          return
        }

        // Defer CloseStream to the next event-loop iteration so any audio
        // callbacks already queued by the native recording module are flushed
        // to the WebSocket before the server is told to stop accepting audio.
        // Without this, stopRecording() can return synchronously while the
        // native module still has a pending onData callback in the event queue,
        // causing audio to arrive after CloseStream.
        setTimeout(() => {
          finalized = true
          if (ws.readyState === WebSocket.OPEN) {
            logForDebugging('[voice_stream] Sending CloseStream (finalize)')
            ws.send(CLOSE_STREAM_MSG)
          }
        }, 0)
      })
    },
    close(): void {
      finalized = true
      if (keepaliveTimer) {
        clearInterval(keepaliveTimer)
        keepaliveTimer = null
      }
      connected = false
      if (ws.readyState === WebSocket.OPEN) {
        ws.close()
      }
    },
    isConnected(): boolean {
      return connected && ws.readyState === WebSocket.OPEN
    },
  }

  ws.on('open', () => {
    logForDebugging('[voice_stream] WebSocket connected')
    connected = true

    // Send an immediate KeepAlive so the server knows the client is active.
    // Audio hardware initialisation can take >1s, so this prevents the
    // server from closing the connection before audio capture starts.
    logForDebugging('[voice_stream] Sending initial KeepAlive')
    ws.send(KEEPALIVE_MSG)

    // Send periodic keepalive to prevent idle timeout
    keepaliveTimer = setInterval(
      ws => {
        if (ws.readyState === WebSocket.OPEN) {
          logForDebugging('[voice_stream] Sending periodic KeepAlive')
          ws.send(KEEPALIVE_MSG)
        }
      },
      KEEPALIVE_INTERVAL_MS,
      ws,
    )

    // Pass the connection to the caller so it can start sending audio.
    // This fires only after the WebSocket is truly open, guaranteeing
    // that send() calls will not be silently dropped.
    callbacks.onReady(connection)
  })

  // Track the last TranscriptText so that when TranscriptEndpoint arrives
  // we can emit it as the final transcript.  The server sometimes sends
  // multiple non-cumulative TranscriptText messages without endpoints
  // between them; the TranscriptText handler auto-finalizes previous
  // segments when it detects the text has changed non-cumulatively.
  let lastTranscriptText = ''

  ws.on('message', (raw: Buffer | string) => {
    const text = raw.toString()
    logForDebugging(
      `[voice_stream] Message received (${String(text.length)} chars): ${text.slice(0, 200)}`,
    )
    let msg: VoiceStreamMessage
    try {
      msg = jsonParse(text) as VoiceStreamMessage
    } catch {
      return
    }

    switch (msg.type) {
      case 'TranscriptText': {
        const transcript = msg.data
        logForDebugging(`[voice_stream] TranscriptText: "${transcript ?? ''}"`)
        // Data arrived after CloseStream — disarm the no-data timer so
        // a slow-but-real flush isn't cut off. Only disarm once finalized
        // (CloseStream sent); pre-CloseStream data racing the deferred
        // send would cancel the timer prematurely, falling back to the
        // slower 5s safety timeout instead of the 1.5s no-data timer.
        if (finalized) {
          cancelNoDataTimer?.()
        }
        if (transcript) {
          // Detect when the server has moved to a new speech segment.
          // Progressive refinements extend or shorten the previous text
          // (e.g., "hello" → "hello world", or "hello wor" → "hello wo").
          // A new segment starts with completely different text (neither
          // is a prefix of the other). When detected, emit the previous
          // text as final so the caller can accumulate it, preventing
          // the new segment from overwriting and losing the old one.
          //
          // Nova 3's interims are cumulative across segments AND can
          // revise earlier text ("Hello?" → "Hello."). Revision breaks
          // the prefix check, causing false auto-finalize → the same
          // text committed once AND re-appearing in the cumulative
          // interim = duplication. Nova 3 only endpoints on the final
          // flush, so auto-finalize is never correct for it.
          if (!isNova3 && lastTranscriptText) {
            const prev = lastTranscriptText.trimStart()
            const next = transcript.trimStart()
            if (
              prev &&
              next &&
              !next.startsWith(prev) &&
              !prev.startsWith(next)
            ) {
              logForDebugging(
                `[voice_stream] Auto-finalizing previous segment (new segment detected): "${lastTranscriptText}"`,
              )
              callbacks.onTranscript(lastTranscriptText, true)
            }
          }
          lastTranscriptText = transcript
          // Emit as interim so the caller can show a live preview.
          callbacks.onTranscript(transcript, false)
        }
        break
      }
      case 'TranscriptEndpoint': {
        logForDebugging(
          `[voice_stream] TranscriptEndpoint received, lastTranscriptText="${lastTranscriptText}"`,
        )
        // The server signals the end of an utterance.  Emit the last
        // TranscriptText as a final transcript so the caller can commit it.
        const finalText = lastTranscriptText
        lastTranscriptText = ''
        if (finalText) {
          callbacks.onTranscript(finalText, true)
        }
        // When TranscriptEndpoint arrives after CloseStream was sent,
        // the server has flushed its final transcript — nothing more is
        // coming.  Resolve finalize now so the caller reads the
        // accumulated buffer immediately (~300ms) instead of waiting
        // for the WebSocket close event (~3-5s of server teardown).
        // `finalized` (not `finalizing`) is the right gate: it flips
        // inside the setTimeout(0) that actually sends CloseStream, so
        // a TranscriptEndpoint that races the deferred send still waits.
        if (finalized) {
          resolveFinalize?.('post_closestream_endpoint')
        }
        break
      }
      case 'TranscriptError': {
        const desc =
          msg.description ?? msg.error_code ?? 'unknown transcription error'
        logForDebugging(`[voice_stream] TranscriptError: ${desc}`)
        if (!finalizing) {
          callbacks.onError(desc)
        }
        break
      }
      case 'error': {
        const errorDetail = msg.message ?? jsonStringify(msg)
        logForDebugging(`[voice_stream] Server error: ${errorDetail}`)
        if (!finalizing) {
          callbacks.onError(errorDetail)
        }
        break
      }
      default:
        break
    }
  })

  ws.on('close', (code, reason) => {
    const reasonStr = reason?.toString() ?? ''
    logForDebugging(
      `[voice_stream] WebSocket closed: code=${String(code)} reason="${reasonStr}"`,
    )
    connected = false
    if (keepaliveTimer) {
      clearInterval(keepaliveTimer)
      keepaliveTimer = null
    }
    // If the server closed the connection before sending TranscriptEndpoint,
    // promote the last interim transcript to final so no text is lost.
    if (lastTranscriptText) {
      logForDebugging(
        '[voice_stream] Promoting unreported interim transcript to final on close',
      )
      const finalText = lastTranscriptText
      lastTranscriptText = ''
      callbacks.onTranscript(finalText, true)
    }
    // During finalize, suppress onError — the session already delivered
    // whatever it had. useVoice's onError path wipes accumulatedRef,
    // which would destroy the transcript before the finalize .then()
    // reads it. `finalizing` (not resolveFinalize) is the gate: set once
    // at finalize() entry, never cleared, so it stays accurate after the
    // fast path or a timer already resolved.
    resolveFinalize?.('ws_close')
    if (!finalizing && !upgradeRejected && code !== 1000 && code !== 1005) {
      callbacks.onError(
        `Connection closed: code ${String(code)}${reasonStr ? ` — ${reasonStr}` : ''}`,
      )
    }
    callbacks.onClose()
  })

  // The ws library fires 'unexpected-response' when the HTTP upgrade
  // returns a non-101 status. Listening lets us surface the actual status
  // and flag 4xx as fatal (same token/TLS fingerprint won't change on
  // retry). With a listener registered, ws does NOT abort on our behalf —
  // we destroy the request; 'error' does not fire, 'close' does (suppressed
  // via upgradeRejected above).
  //
  // Bun's ws shim historically didn't implement this event (a warning
  // is logged once at registration). Under Bun a non-101 upgrade falls
  // through to the generic 'error' + 'close' 1002 path with no recoverable
  // status; the attemptGenRef guard in useVoice.ts still surfaces the
  // retry-attempt failure, the user just sees "Expected 101 status code"
  // instead of "HTTP 503". No harm — the gen fix is the load-bearing part.
  ws.on('unexpected-response', (req: ClientRequest, res: IncomingMessage) => {
    const status = res.statusCode ?? 0
    // Bun's ws implementation on Windows can fire this event for a
    // successful 101 Switching Protocols response (anthropics/claude-code#40510).
    // 101 is never a rejection — bail before we destroy a working upgrade.
    if (status === 101) {
      logForDebugging(
        '[voice_stream] unexpected-response fired with 101; ignoring',
      )
      return
    }
    logForDebugging(
      `[voice_stream] Upgrade rejected: status=${String(status)} cf-mitigated=${String(res.headers['cf-mitigated'])} cf-ray=${String(res.headers['cf-ray'])}`,
    )
    upgradeRejected = true
    res.resume()
    req.destroy()
    if (finalizing) return
    callbacks.onError(
      `WebSocket upgrade rejected with HTTP ${String(status)}`,
      { fatal: status >= 400 && status < 500 },
    )
  })

  ws.on('error', (err: Error) => {
    logError(err)
    logForDebugging(`[voice_stream] WebSocket error: ${err.message}`)
    if (!finalizing) {
      callbacks.onError(`Voice stream connection error: ${err.message}`)
    }
  })

  return connection
}

	// Anthropic voice_stream speech-to-text client for push-to-talk.
	//
	// Only reachable in ant builds (gated by feature('VOICE_MODE') in useVoice.ts import).
	//
	// Connects to Anthropic's voice_stream WebSocket endpoint using the same
	// OAuth credentials as Claude Code. The endpoint uses conversation_engine
	// backed models for speech-to-text. Designed for hold-to-talk: hold the
	// keybinding to record, release to stop and submit.
	//
	// The wire protocol uses JSON control messages (KeepAlive, CloseStream) and
	// binary audio frames. The server responds with TranscriptText and
	// TranscriptEndpoint JSON messages.

	import type { ClientRequest, IncomingMessage } from 'http'
	import WebSocket from 'ws'
	import { getOauthConfig } from '../constants/oauth.js'
	import {
	checkAndRefreshOAuthTokenIfNeeded,
	getClaudeAIOAuthTokens,
	isAnthropicAuthEnabled,
	} from '../utils/auth.js'
	import { logForDebugging } from '../utils/debug.js'
	import { getUserAgent } from '../utils/http.js'
	import { logError } from '../utils/log.js'
	import { getWebSocketTLSOptions } from '../utils/mtls.js'
	import { getWebSocketProxyAgent, getWebSocketProxyUrl } from '../utils/proxy.js'
	import { jsonParse, jsonStringify } from '../utils/slowOperations.js'

	const KEEPALIVE_MSG = '{"type":"KeepAlive"}'
	const CLOSE_STREAM_MSG = '{"type":"CloseStream"}'

	import { getFeatureValue_CACHED_MAY_BE_STALE } from './analytics/growthbook.js'

	// ─── Constants ───────────────────────────────────────────────────────

	const VOICE_STREAM_PATH = '/api/ws/speech_to_text/voice_stream'

	const KEEPALIVE_INTERVAL_MS = 8_000

	// finalize() resolution timers. `noData` fires when no TranscriptText
	// arrives post-CloseStream — the server has nothing; don't wait out the
	// full ~3-5s WS teardown to confirm emptiness. `safety` is the last-
	// resort cap if the WS hangs. Exported so tests can shorten them.
	export const FINALIZE_TIMEOUTS_MS = {
	safety: 5_000,
	noData: 1_500,
	}

	// ─── Types ──────────────────────────────────────────────────────────

	export type VoiceStreamCallbacks = {
	onTranscript: (text: string, isFinal: boolean) => void
	onError: (error: string, opts?: { fatal?: boolean }) => void
	onClose: () => void
	onReady: (connection: VoiceStreamConnection) => void
	}

	// How finalize() resolved. `no_data_timeout` means zero server messages
	// after CloseStream — the silent-drop signature (anthropics/anthropic#287008).
	export type FinalizeSource =
	\| 'post_closestream_endpoint'
	\| 'no_data_timeout'
	\| 'safety_timeout'
	\| 'ws_close'
	\| 'ws_already_closed'

	export type VoiceStreamConnection = {
	send: (audioChunk: Buffer) => void
	finalize: () => Promise<FinalizeSource>
	close: () => void
	isConnected: () => boolean
	}

	// The voice_stream endpoint returns transcript chunks and endpoint markers.
	type VoiceStreamTranscriptText = {
	type: 'TranscriptText'
	data: string
	}

	type VoiceStreamTranscriptEndpoint = {
	type: 'TranscriptEndpoint'
	}

	type VoiceStreamTranscriptError = {
	type: 'TranscriptError'
	error_code?: string
	description?: string
	}

	type VoiceStreamMessage =
	\| VoiceStreamTranscriptText
	\| VoiceStreamTranscriptEndpoint
	\| VoiceStreamTranscriptError
	\| { type: 'error'; message?: string }

	// ─── Availability ──────────────────────────────────────────────────────

	export function isVoiceStreamAvailable(): boolean {
	// voice_stream uses the same OAuth as Claude Code — available when the
	// user is authenticated with Anthropic (Claude.ai subscriber or has
	// valid OAuth tokens).
	if (!isAnthropicAuthEnabled()) {
	return false
	}
	const tokens = getClaudeAIOAuthTokens()
	return tokens !== null && tokens.accessToken !== null
	}

	// ─── Connection ────────────────────────────────────────────────────────

	export async function connectVoiceStream(
	callbacks: VoiceStreamCallbacks,
	options?: { language?: string; keyterms?: string[] },
	): Promise<VoiceStreamConnection \| null> {
	// Ensure OAuth token is fresh before connecting
	await checkAndRefreshOAuthTokenIfNeeded()

	const tokens = getClaudeAIOAuthTokens()
	if (!tokens?.accessToken) {
	logForDebugging('[voice_stream] No OAuth token available')
	return null
	}

	// voice_stream is a private_api route, but /api/ws/ is also exposed on
	// the api.anthropic.com listener (service_definitions.yaml private-api:
	// visibility.external: true). We target that host instead of claude.ai
	// because the claude.ai CF zone uses TLS fingerprinting and challenges
	// non-browser clients (anthropics/claude-code#34094). Same private-api
	// pod, same OAuth Bearer auth — just a CF zone that doesn't block us.
	// Desktop dictation still uses claude.ai (Swift URLSession has a
	// browser-class JA3 fingerprint, so CF lets it through).
	const wsBaseUrl =
	process.env.VOICE_STREAM_BASE_URL \|\|
	getOauthConfig()
	.BASE_API_URL.replace('https://', 'wss://')
	.replace('http://', 'ws://')

	if (process.env.VOICE_STREAM_BASE_URL) {
	logForDebugging(
	`[voice_stream] Using VOICE_STREAM_BASE_URL override: ${process.env.VOICE_STREAM_BASE_URL}`,
	)
	}

	const params = new URLSearchParams({
	encoding: 'linear16',
	sample_rate: '16000',
	channels: '1',
	endpointing_ms: '300',
	utterance_end_ms: '1000',
	language: options?.language ?? 'en',
	})

	// Route through conversation-engine with Deepgram Nova 3 (bypassing
	// the server's project_bell_v2_config GrowthBook gate). The server
	// side is anthropics/anthropic#278327 + #281372; this lets us ramp
	// clients independently.
	const isNova3 = getFeatureValue_CACHED_MAY_BE_STALE(
	'tengu_cobalt_frost',
	false,
	)
	if (isNova3) {
	params.set('use_conversation_engine', 'true')
	params.set('stt_provider', 'deepgram-nova3')
	logForDebugging('[voice_stream] Nova 3 gate enabled (tengu_cobalt_frost)')
	}

	// Append keyterms as query params — the voice_stream proxy forwards
	// these to the STT service which applies appropriate boosting.
	if (options?.keyterms?.length) {
	for (const term of options.keyterms) {
	params.append('keyterms', term)
	}
	}

	const url = `${wsBaseUrl}${VOICE_STREAM_PATH}?${params.toString()}`

	logForDebugging(`[voice_stream] Connecting to ${url}`)

	const headers: Record<string, string> = {
	Authorization: `Bearer ${tokens.accessToken}`,
	'User-Agent': getUserAgent(),
	'x-app': 'cli',
	}

	const tlsOptions = getWebSocketTLSOptions()
	const wsOptions =
	typeof Bun !== 'undefined'
	? {
	headers,
	proxy: getWebSocketProxyUrl(url),
	tls: tlsOptions \|\| undefined,
	}
	: { headers, agent: getWebSocketProxyAgent(url), ...tlsOptions }

	const ws = new WebSocket(url, wsOptions)

	let keepaliveTimer: ReturnType<typeof setInterval> \| null = null
	let connected = false
	// Set to true once CloseStream has been sent (or the ws is closed).
	// After this, further audio sends are dropped.
	let finalized = false
	// Set to true when finalize() is first called, to prevent double-fire.
	let finalizing = false
	// Set when the HTTP upgrade was rejected (unexpected-response). The
	// close event that follows (1006 from our req.destroy()) is just
	// mechanical teardown; the upgrade handler already reported the error.
	let upgradeRejected = false
	// Resolves finalize(). Four triggers: TranscriptEndpoint post-CloseStream
	// (~300ms); no-data timer (1.5s); WS close (~3-5s); safety timer (5s).
	let resolveFinalize: ((source: FinalizeSource) => void) \| null = null
	let cancelNoDataTimer: (() => void) \| null = null

	// Define the connection object before event handlers so it can be passed
	// to onReady when the WebSocket opens.
	const connection: VoiceStreamConnection = {
	send(audioChunk: Buffer): void {
	if (ws.readyState !== WebSocket.OPEN) {
	return
	}
	if (finalized) {
	// After CloseStream has been sent, the server rejects further audio.
	// Drop the chunk to avoid a protocol error.
	logForDebugging(
	`[voice_stream] Dropping audio chunk after CloseStream: ${String(audioChunk.length)} bytes`,
	)
	return
	}
	logForDebugging(
	`[voice_stream] Sending audio chunk: ${String(audioChunk.length)} bytes`,
	)
	// Copy the buffer before sending: NAPI Buffer objects from native
	// modules may share a pooled ArrayBuffer. Creating a view with
	// `new Uint8Array(buf.buffer, offset, len)` can reference stale or
	// overlapping memory by the time the ws library reads it.
	// `Buffer.from()` makes an owned copy that the ws library can safely
	// consume as a binary WebSocket frame.
	ws.send(Buffer.from(audioChunk))
	},
	finalize(): Promise<FinalizeSource> {
	if (finalizing \|\| finalized) {
	// Already finalized or WebSocket already closed — resolve immediately.
	return Promise.resolve('ws_already_closed')
	}
	finalizing = true

	return new Promise<FinalizeSource>(resolve => {
	const safetyTimer = setTimeout(
	() => resolveFinalize?.('safety_timeout'),
	FINALIZE_TIMEOUTS_MS.safety,
	)
	const noDataTimer = setTimeout(
	() => resolveFinalize?.('no_data_timeout'),
	FINALIZE_TIMEOUTS_MS.noData,
	)
	cancelNoDataTimer = () => {
	clearTimeout(noDataTimer)
	cancelNoDataTimer = null
	}

	resolveFinalize = (source: FinalizeSource) => {
	clearTimeout(safetyTimer)
	clearTimeout(noDataTimer)
	resolveFinalize = null
	cancelNoDataTimer = null
	// Legacy Deepgram can leave an interim in lastTranscriptText
	// with no TranscriptEndpoint (websocket_manager.py sends
	// TranscriptChunk and TranscriptEndpoint as independent
	// channel items). All resolve triggers must promote it;
	// centralize here. No-op when the close handler already did.
	if (lastTranscriptText) {
	logForDebugging(
	`[voice_stream] Promoting unreported interim before ${source} resolve`,
	)
	const t = lastTranscriptText
	lastTranscriptText = ''
	callbacks.onTranscript(t, true)
	}
	logForDebugging(`[voice_stream] Finalize resolved via ${source}`)
	resolve(source)
	}

	// If the WebSocket is already closed, resolve immediately.
	if (
	ws.readyState === WebSocket.CLOSED \|\|
	ws.readyState === WebSocket.CLOSING
	) {
	resolveFinalize('ws_already_closed')
	return
	}

	// Defer CloseStream to the next event-loop iteration so any audio
	// callbacks already queued by the native recording module are flushed
	// to the WebSocket before the server is told to stop accepting audio.
	// Without this, stopRecording() can return synchronously while the
	// native module still has a pending onData callback in the event queue,
	// causing audio to arrive after CloseStream.
	setTimeout(() => {
	finalized = true
	if (ws.readyState === WebSocket.OPEN) {
	logForDebugging('[voice_stream] Sending CloseStream (finalize)')
	ws.send(CLOSE_STREAM_MSG)
	}
	}, 0)
	})
	},
	close(): void {
	finalized = true
	if (keepaliveTimer) {
	clearInterval(keepaliveTimer)
	keepaliveTimer = null
	}
	connected = false
	if (ws.readyState === WebSocket.OPEN) {
	ws.close()
	}
	},
	isConnected(): boolean {
	return connected && ws.readyState === WebSocket.OPEN
	},
	}

	ws.on('open', () => {
	logForDebugging('[voice_stream] WebSocket connected')
	connected = true

	// Send an immediate KeepAlive so the server knows the client is active.
	// Audio hardware initialisation can take >1s, so this prevents the
	// server from closing the connection before audio capture starts.
	logForDebugging('[voice_stream] Sending initial KeepAlive')
	ws.send(KEEPALIVE_MSG)

	// Send periodic keepalive to prevent idle timeout
	keepaliveTimer = setInterval(
	ws => {
	if (ws.readyState === WebSocket.OPEN) {
	logForDebugging('[voice_stream] Sending periodic KeepAlive')
	ws.send(KEEPALIVE_MSG)
	}
	},
	KEEPALIVE_INTERVAL_MS,
	ws,
	)

	// Pass the connection to the caller so it can start sending audio.
	// This fires only after the WebSocket is truly open, guaranteeing
	// that send() calls will not be silently dropped.
	callbacks.onReady(connection)
	})

	// Track the last TranscriptText so that when TranscriptEndpoint arrives
	// we can emit it as the final transcript. The server sometimes sends
	// multiple non-cumulative TranscriptText messages without endpoints
	// between them; the TranscriptText handler auto-finalizes previous
	// segments when it detects the text has changed non-cumulatively.
	let lastTranscriptText = ''

	ws.on('message', (raw: Buffer \| string) => {
	const text = raw.toString()
	logForDebugging(
	`[voice_stream] Message received (${String(text.length)} chars): ${text.slice(0, 200)}`,
	)
	let msg: VoiceStreamMessage
	try {
	msg = jsonParse(text) as VoiceStreamMessage
	} catch {
	return
	}

	switch (msg.type) {
	case 'TranscriptText': {
	const transcript = msg.data
	logForDebugging(`[voice_stream] TranscriptText: "${transcript ?? ''}"`)
	// Data arrived after CloseStream — disarm the no-data timer so
	// a slow-but-real flush isn't cut off. Only disarm once finalized
	// (CloseStream sent); pre-CloseStream data racing the deferred
	// send would cancel the timer prematurely, falling back to the
	// slower 5s safety timeout instead of the 1.5s no-data timer.
	if (finalized) {
	cancelNoDataTimer?.()
	}
	if (transcript) {
	// Detect when the server has moved to a new speech segment.
	// Progressive refinements extend or shorten the previous text
	// (e.g., "hello" → "hello world", or "hello wor" → "hello wo").
	// A new segment starts with completely different text (neither
	// is a prefix of the other). When detected, emit the previous
	// text as final so the caller can accumulate it, preventing
	// the new segment from overwriting and losing the old one.
	//
	// Nova 3's interims are cumulative across segments AND can
	// revise earlier text ("Hello?" → "Hello."). Revision breaks
	// the prefix check, causing false auto-finalize → the same
	// text committed once AND re-appearing in the cumulative
	// interim = duplication. Nova 3 only endpoints on the final
	// flush, so auto-finalize is never correct for it.
	if (!isNova3 && lastTranscriptText) {
	const prev = lastTranscriptText.trimStart()
	const next = transcript.trimStart()
	if (
	prev &&
	next &&
	!next.startsWith(prev) &&
	!prev.startsWith(next)
	) {
	logForDebugging(
	`[voice_stream] Auto-finalizing previous segment (new segment detected): "${lastTranscriptText}"`,
	)
	callbacks.onTranscript(lastTranscriptText, true)
	}
	}
	lastTranscriptText = transcript
	// Emit as interim so the caller can show a live preview.
	callbacks.onTranscript(transcript, false)
	}
	break
	}
	case 'TranscriptEndpoint': {
	logForDebugging(
	`[voice_stream] TranscriptEndpoint received, lastTranscriptText="${lastTranscriptText}"`,
	)
	// The server signals the end of an utterance. Emit the last
	// TranscriptText as a final transcript so the caller can commit it.
	const finalText = lastTranscriptText
	lastTranscriptText = ''
	if (finalText) {
	callbacks.onTranscript(finalText, true)
	}
	// When TranscriptEndpoint arrives after CloseStream was sent,
	// the server has flushed its final transcript — nothing more is
	// coming. Resolve finalize now so the caller reads the
	// accumulated buffer immediately (~300ms) instead of waiting
	// for the WebSocket close event (~3-5s of server teardown).
	// `finalized` (not `finalizing`) is the right gate: it flips
	// inside the setTimeout(0) that actually sends CloseStream, so
	// a TranscriptEndpoint that races the deferred send still waits.
	if (finalized) {
	resolveFinalize?.('post_closestream_endpoint')
	}
	break
	}
	case 'TranscriptError': {
	const desc =
	msg.description ?? msg.error_code ?? 'unknown transcription error'
	logForDebugging(`[voice_stream] TranscriptError: ${desc}`)
	if (!finalizing) {
	callbacks.onError(desc)
	}
	break
	}
	case 'error': {
	const errorDetail = msg.message ?? jsonStringify(msg)
	logForDebugging(`[voice_stream] Server error: ${errorDetail}`)
	if (!finalizing) {
	callbacks.onError(errorDetail)
	}
	break
	}
	default:
	break
	}
	})

	ws.on('close', (code, reason) => {
	const reasonStr = reason?.toString() ?? ''
	logForDebugging(
	`[voice_stream] WebSocket closed: code=${String(code)} reason="${reasonStr}"`,
	)
	connected = false
	if (keepaliveTimer) {
	clearInterval(keepaliveTimer)
	keepaliveTimer = null
	}
	// If the server closed the connection before sending TranscriptEndpoint,
	// promote the last interim transcript to final so no text is lost.
	if (lastTranscriptText) {
	logForDebugging(
	'[voice_stream] Promoting unreported interim transcript to final on close',
	)
	const finalText = lastTranscriptText
	lastTranscriptText = ''
	callbacks.onTranscript(finalText, true)
	}
	// During finalize, suppress onError — the session already delivered
	// whatever it had. useVoice's onError path wipes accumulatedRef,
	// which would destroy the transcript before the finalize .then()
	// reads it. `finalizing` (not resolveFinalize) is the gate: set once
	// at finalize() entry, never cleared, so it stays accurate after the
	// fast path or a timer already resolved.
	resolveFinalize?.('ws_close')
	if (!finalizing && !upgradeRejected && code !== 1000 && code !== 1005) {
	callbacks.onError(
	`Connection closed: code ${String(code)}${reasonStr ? ` — ${reasonStr}` : ''}`,
	)
	}
	callbacks.onClose()
	})

	// The ws library fires 'unexpected-response' when the HTTP upgrade
	// returns a non-101 status. Listening lets us surface the actual status
	// and flag 4xx as fatal (same token/TLS fingerprint won't change on
	// retry). With a listener registered, ws does NOT abort on our behalf —
	// we destroy the request; 'error' does not fire, 'close' does (suppressed
	// via upgradeRejected above).
	//
	// Bun's ws shim historically didn't implement this event (a warning
	// is logged once at registration). Under Bun a non-101 upgrade falls
	// through to the generic 'error' + 'close' 1002 path with no recoverable
	// status; the attemptGenRef guard in useVoice.ts still surfaces the
	// retry-attempt failure, the user just sees "Expected 101 status code"
	// instead of "HTTP 503". No harm — the gen fix is the load-bearing part.
	ws.on('unexpected-response', (req: ClientRequest, res: IncomingMessage) => {
	const status = res.statusCode ?? 0
	// Bun's ws implementation on Windows can fire this event for a
	// successful 101 Switching Protocols response (anthropics/claude-code#40510).
	// 101 is never a rejection — bail before we destroy a working upgrade.
	if (status === 101) {
	logForDebugging(
	'[voice_stream] unexpected-response fired with 101; ignoring',
	)
	return
	}
	logForDebugging(
	`[voice_stream] Upgrade rejected: status=${String(status)} cf-mitigated=${String(res.headers['cf-mitigated'])} cf-ray=${String(res.headers['cf-ray'])}`,
	)
	upgradeRejected = true
	res.resume()
	req.destroy()
	if (finalizing) return
	callbacks.onError(
	`WebSocket upgrade rejected with HTTP ${String(status)}`,
	{ fatal: status >= 400 && status < 500 },
	)
	})

	ws.on('error', (err: Error) => {
	logError(err)
	logForDebugging(`[voice_stream] WebSocket error: ${err.message}`)
	if (!finalizing) {
	callbacks.onError(`Voice stream connection error: ${err.message}`)
	}
	})

	return connection
	}