fix: keep pcm stream continuous

cai

2026-07-07 a62af4b377643d3b76318ee782763b3833b27456

 src/main.rs

@@ -12,7 +12,7 @@
};

use anyhow::{Context, Result, anyhow};
use audio::{AudioDiagnostics, PcmFrame, load_pre_recorded_frames};
use audio::{AudioDiagnostics, load_pre_recorded_frames};
use base64::{Engine as _, engine::general_purpose};
use futures_util::StreamExt;
use libwebrtc::{
@@ -1343,13 +1343,54 @@
        .trim()
        .to_ascii_lowercase();
    let frames = if format == "pcm_s16le" {
        pcm_s16le_payload_to_frames(
            &payload,
            audio_chunk.sample_rate.unwrap_or(TARGET_SAMPLE_RATE_HZ),
            audio_chunk
                .channels
                .unwrap_or(u32::from(TARGET_NUM_CHANNELS)),
        )?
        let sample_rate = audio_chunk.sample_rate.unwrap_or(TARGET_SAMPLE_RATE_HZ);
        let channels = audio_chunk
            .channels
            .unwrap_or(u32::from(TARGET_NUM_CHANNELS));
        if state.pcm_stream_decoder.is_none() {
            state.pcm_stream_decoder = Some(audio::PcmS16leStreamDecoder::new(
                sample_rate,
                channels,
                TARGET_SAMPLE_RATE_HZ,
                TARGET_NUM_CHANNELS,
            )?);
        }
        state.pcm_stream_network_chunk_count =
            state.pcm_stream_network_chunk_count.saturating_add(1);
        let stream_result = state
            .pcm_stream_decoder
            .as_mut()
            .expect("pcm stream decoder initialized")
            .push_bytes(
                &payload,
                sample_rate,
                channels,
                audio_chunk.last.unwrap_or(false),
            )?;
        if stream_result.dropped_tail_bytes > 0 {
            warn!(
                call_id = %call_id,
                trace_id = %trace_id,
                turn_id = %turn.turn_id,
                chunk_seq = audio_chunk.chunk_seq.unwrap_or_default(),
                dropped_tail_bytes = stream_result.dropped_tail_bytes,
                "runtime helper stream_audio_pcm_unaligned_tail_dropped"
            );
        }
        if stream_result.frames.is_empty() {
            warn!(
                call_id = %call_id,
                trace_id = %trace_id,
                turn_id = %turn.turn_id,
                chunk_seq = audio_chunk.chunk_seq.unwrap_or_default(),
                buffered_source_bytes = stream_result.buffered_source_bytes,
                buffered_source_samples = stream_result.buffered_source_samples,
                network_chunk_count = state.pcm_stream_network_chunk_count,
                "runtime helper stream_audio_pcm_waiting_for_20ms_frame"
            );
            return Ok(0);
        }
        stream_result.frames
    } else if matches!(format.as_str(), "mp3" | "mpeg" | "wav") {
        state.encoded_audio_buffer.extend_from_slice(&payload);
        match audio::decode_audio_bytes_to_frames(
@@ -1375,6 +1416,20 @@
                    error = %safe_error(&error.to_string()),
                    "runtime helper stream_audio_chunk_decode_waiting_for_more_data"
                );
                return Ok(0);
            }
            Err(error) if state.first_audio_frame_written => {
                warn!(
                    call_id = %call_id,
                    trace_id = %trace_id,
                    turn_id = %turn.turn_id,
                    chunk_seq = audio_chunk.chunk_seq.unwrap_or_default(),
                    format = %format,
                    buffered_bytes = state.encoded_audio_buffer.len(),
                    error = %safe_error(&error.to_string()),
                    "runtime helper stream_audio_final_chunk_decode_ignored"
                );
                state.encoded_audio_buffer.clear();
                return Ok(0);
            }
            Err(error) => return Err(error).context("failed to decode final stream audio chunk"),
@@ -1427,6 +1482,34 @@
        sleep_until(pacing_started_at + Duration::from_millis(((index + 1) as u64) * 20)).await;
    }
    if audio_chunk.last.unwrap_or(false) {
        let mut debug_source_path = None;
        let mut debug_pcm_wav_path = None;
        let mut debug_pcm_wav_size_bytes = None;
        if format == "pcm_s16le" {
            if let (Some(debug_dump_dir), Some(decoder)) = (
                bridge_config.audio_debug_dump_dir.as_deref(),
                state.pcm_stream_decoder.as_ref(),
            ) {
                match decoder.write_debug_dump(
                    debug_dump_dir,
                    call_id,
                    &format!("stream-reply-{}", turn.turn_id),
                ) {
                    Ok(debug_dump) => {
                        debug_source_path = debug_dump.debug_source_path;
                        debug_pcm_wav_path = debug_dump.debug_pcm_wav_path;
                        debug_pcm_wav_size_bytes = debug_dump.debug_pcm_wav_size_bytes;
                    }
                    Err(error) => warn!(
                        call_id = %call_id,
                        trace_id = %trace_id,
                        turn_id = %turn.turn_id,
                        error = %safe_error(&error.to_string()),
                        "runtime helper stream_audio_pcm_debug_dump_failed"
                    ),
                }
            }
        }
        emit_activity(
            call_id,
            trace_id,
@@ -1439,51 +1522,17 @@
                "replyPlaybackMode": state.reply_playback_mode.as_str(),
                "format": format.as_str(),
                "chunkSeq": audio_chunk.chunk_seq,
                "networkChunkCount": state.pcm_stream_network_chunk_count,
                "debugSourcePath": debug_source_path,
                "debugPcmWavPath": debug_pcm_wav_path,
                "debugPcmWavSizeBytes": debug_pcm_wav_size_bytes,
                "sampleRate": audio_chunk.sample_rate,
                "channels": audio_chunk.channels,
                "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
            }),
        );
    }
    Ok(frames.len())
}

fn pcm_s16le_payload_to_frames(
    payload: &[u8],
    sample_rate: u32,
    channels: u32,
) -> Result<Vec<PcmFrame>> {
    if sample_rate != TARGET_SAMPLE_RATE_HZ || channels != u32::from(TARGET_NUM_CHANNELS) {
        return Err(anyhow!(
            "unsupported pcm_s16le stream format: sample_rate={}, channels={}",
            sample_rate,
            channels
        ));
    }
    if payload.len() % 2 != 0 {
        return Err(anyhow!("pcm_s16le payload has odd byte length"));
    }
    let samples: Vec<i16> = payload
        .chunks_exact(2)
        .map(|chunk| i16::from_le_bytes([chunk[0], chunk[1]]))
        .collect();
    if samples.is_empty() {
        return Ok(Vec::new());
    }
    let samples_per_channel = (sample_rate / 50).max(1);
    let frame_sample_count = samples_per_channel as usize * channels as usize;
    let mut frames = Vec::new();
    for chunk in samples.chunks(frame_sample_count) {
        let chunk_samples_per_channel = (chunk.len() / channels as usize) as u32;
        if chunk_samples_per_channel == 0 {
            continue;
        }
        frames.push(PcmFrame::new(
            chunk.to_vec(),
            sample_rate,
            channels,
            chunk_samples_per_channel,
        ));
    }
    Ok(frames)
}

fn trim_ascii_whitespace(value: &[u8]) -> &[u8] {
@@ -1861,6 +1910,8 @@
    audio_chunk_count: u64,
    device_output_count: u64,
    encoded_audio_buffer: Vec<u8>,
    pcm_stream_decoder: Option<audio::PcmS16leStreamDecoder>,
    pcm_stream_network_chunk_count: u64,
}

impl Default for RuntimeTurnStreamState {
@@ -1874,6 +1925,8 @@
            audio_chunk_count: 0,
            device_output_count: 0,
            encoded_audio_buffer: Vec::new(),
            pcm_stream_decoder: None,
            pcm_stream_network_chunk_count: 0,
        }
    }
}