fix(audio): isolate conversation timeouts by round

ql-application/threadx/interface/custom_app/vad_det.c

@@ -43,7 +43,9 @@ static uint8_t vad_leave_message_flag = 0;//留言模式vad只做单次
 static Fvad *fvad_inst = NULL;
 extern struct ringbuffer *ring_sender;
 extern ql_sem_t sender_sem;
-extern unsigned char wss_connect_start; //表示vad拾音结束 开始等待后台返回 如果超过5s没有返回 则报超时语音
+extern void conversation_no_speech(void);
+extern void conversation_record_begin(void);
+extern void conversation_cancel_wait(void);
 
 // 配置
 static int sample_rate = 16000;             // FVAD 支持的采样率：8k/16k/32k/48k
@@ -89,6 +91,12 @@ void vad_tts_playback_finished()
 	vad_tts_playback_finished_status = 1;
 }
 
+void vad_reset_conversation_flags(void)
+{
+    no_speech_timeout = 0;
+    vad_tts_playback_finished_status = 0;
+}
+
 void vad_set_speech_timeout_flag(uint8_t value)
 {
 	no_speech_timeout = value;
@@ -216,7 +224,7 @@ int vad_audio_stop(void) {
 			 
 	ql_audio_record_close(record_hd, 1);
 
-	wss_connect_start = 1;//用于判断后台响应超时
+	/* Reply timeout starts in the sender only after END is submitted. */
 
     key_start_leave = 0;
 	//ql_rtos_semaphore_release(sender_sem); 
@@ -226,6 +234,7 @@ int vad_audio_stop(void) {
 
 // 停止录音
 int vad_audio_stop_no_wait_resp(void) {
+    conversation_cancel_wait();
 
     printf("Audio recording stopped no_wait_resp, frames: %d, speech: %d", 
              frame_count, speech_frame_count);
@@ -234,7 +243,7 @@ int vad_audio_stop_no_wait_resp(void) {
 			 
 	ql_audio_record_close(record_hd, 1);
 
-	wss_connect_start = 0;
+
 			 
     return 0;
 }
@@ -242,6 +251,7 @@ int vad_audio_stop_no_wait_resp(void) {
 
 // 触发录音（按键触发）
 void vad_audio_trigger_record(void) {
+    conversation_record_begin();
     g_triggered = true;
     printf("Record triggered");
 }
@@ -449,6 +459,7 @@ void vad_audio_process(void) {
                     if ((now - no_sound_start_time) > VAD_NO_SOUND_TIMEOUT_MS) {
                         printf("No speech for timeout, stopping");
                         no_speech_timeout = 1;
+                        conversation_no_speech();
                         set_state(STATE_WAITING_COMMAND);
                         vad_audio_stop();
                         buffer_offset = 0;

ql-application/threadx/interface/custom_app/ws_talkingq.c

@@ -52,7 +52,57 @@ static int16 *ppbuff=mp3_buff;
 
 static int onmessage_status = -1;
 volatile int tts_playing = 0;
-unsigned char wss_connect_start = 0;
+/* Reply waits and no-speech prompts belong to one listening round. */
+static ql_mutex_t conversation_mutex = NULL;
+static volatile unsigned int conversation_round = 0;
+static unsigned int wait_round = 0;
+static unsigned int sender_round = 0;
+static uint64_t reply_wait_start_ms = 0;
+enum { WAIT_NONE, WAIT_NO_SPEECH, WAIT_REPLY, WAIT_SEND_FAILURE };
+static int conversation_wait = WAIT_NONE;
+static int timeout_action_busy = 0;
+static int round_response_received = 0;
+
+static void conversation_lock(void)
+{
+    ql_rtos_mutex_lock(conversation_mutex, 0xFFFFFFFF);
+}
+
+static void conversation_unlock(void)
+{
+    ql_rtos_mutex_unlock(conversation_mutex);
+}
+
+/* Caller holds the mutex; ignored recording keys must not call this. */
+static void conversation_begin_round(void)
+{
+    conversation_round++;
+    conversation_wait = WAIT_NONE;
+    round_response_received = 0;
+}
+
+void conversation_record_begin(void)
+{
+    conversation_lock();
+    if (!g_mic_working) conversation_begin_round();
+    g_mic_working = 1;
+    conversation_unlock();
+}
+
+void conversation_cancel_wait(void)
+{
+    conversation_lock();
+    conversation_wait = WAIT_NONE;
+    conversation_unlock();
+}
+
+void conversation_no_speech(void)
+{
+    conversation_lock();
+    wait_round = conversation_round;
+    conversation_wait = WAIT_NO_SPEECH;
+    conversation_unlock();
+}
 static volatile int player_mq_couter = 0;
 /* Set while a barge-in prompt bridges the old and new sessions.  The old
  * server response can still deliver URLs after its player queue was cleared. */
@@ -73,6 +123,7 @@ extern u8 power_down_status;
 extern int aofei_mp3_player_get_result(void);
 extern int vad_leave_message_is_active(void);
 extern int vad_cancel_leave_message(void);
+extern void vad_reset_conversation_flags(void);
 
 void sendAuthentication(void);
 void vad_start_func(void);
@@ -83,7 +134,8 @@ int audio_packet_type = 0;
 
 typedef struct ice_work_item_s
 {
-    int type;  //0-tts_out 1-session_continue 2-session_end 3-云端超时提示（失败时播放本地兜底）
+    int type;  //0-tts_out 1-session_continue 2-session_end 3-云端超时提示（失败时播放本地兜底） 4-本地超时兜底
+    unsigned int timeout_round; /* Timeout items (type 3/4) use this cancellation token. */
     int error;
 	int del_url;
     void *data; 
@@ -291,13 +343,14 @@ void add_file_to_play_player_queue(char *path, int type)
 	}
 				
 	item->type = type;
+	item->timeout_round = conversation_round;
 
 	// 安全修复：使用strncpy替代sprintf防止溢出
 	snprintf((char *)item->data, strlen(path)+1, "%s", path);
 
 	printf("play file:%s\r\n", (char *)item->data);
 				
-	if (ql_rtos_queue_release(player_queue, sizeof(item), &item, 0xFFFFFFFF) != 0)
+	if (ql_rtos_queue_release(player_queue, sizeof(item), &item, (type == 3 || type == 4) ? 0 : 0xFFFFFFFF) != 0)
 	{
 		printf("player_mq rt_mq_send ERR\n");
 		free(item->data);
@@ -348,7 +401,6 @@ static void onmessage_text(rws_socket socket, const char *text, const unsigned i
 		}else if(strstr(text, "END_ACK")||strstr(text, "START_ACK")){
 			
 		}else{//TTS_URL:audio_url
-			wss_connect_start = 0;
 			char *ret_str1 = strstr(text, "https://");
 			char *ret_str2 = NULL;
 			size_t url_len = 0;
@@ -402,7 +454,12 @@ static void onmessage_text(rws_socket socket, const char *text, const unsigned i
 					tts_play_status = 2;
 				}
 
-				if(is_tts_error_url == 1 || is_message_ok_url){
+				conversation_lock();
+                    conversation_wait = WAIT_NONE;
+                    round_response_received = 1;
+                    conversation_unlock();
+
+                    if(is_tts_error_url == 1 || is_message_ok_url){
 					printf("tts_error or message_ok\n");
 					vad_set_speech_timeout_flag(1);
 				}
@@ -583,12 +640,10 @@ int talkingq_ws_send_bin(uint8_t *senddata, int len)
     
  //   LOG_D("talkingq_wss_send_bin, len = %d", len); 
 
-    rws_socket_send_bin(app->socket, senddata, len,2,rws_true); 
+    return rws_socket_send_bin(app->socket, senddata, len,2,rws_true) ? 0 : -1;
-    
-    return 0; 
 }
 
-void sendRecordingAudioPacket(AudioPacket *packet);
+int sendRecordingAudioPacket(AudioPacket *packet);
 
 /*串口AT指令做设置*/
 void generateDeviceId(void) {
@@ -806,7 +861,16 @@ void sendEndPacket(void) {
     endPacket.packetType = 2;
     endPacket.dataSize = 0;
     
-    sendRecordingAudioPacket(&endPacket);
+    conversation_lock();
+    if (sender_round == conversation_round) {
+        int end_ret = sendRecordingAudioPacket(&endPacket);
+        if (!round_response_received) {
+            wait_round = sender_round;
+            reply_wait_start_ms = hal_os_up_time_ms();
+            conversation_wait = end_ret == 0 ? WAIT_REPLY : WAIT_SEND_FAILURE;
+        }
+    }
+    conversation_unlock();
   //  rt_kprintf("sendEndPacket, seq=%u\r\n", endPacket.sequenceNumber);
 }
 
@@ -862,13 +926,13 @@ void sendAuthentication()
 	cJSON_Delete(sendAuth);
 }
 
-void sendRecordingAudioPacket(AudioPacket *packet) {
+int sendRecordingAudioPacket(AudioPacket *packet) {
     uint8_t *sendBuffer=NULL;
     
     sendBuffer = malloc(DEVICE_ID_SIZE + SESSION_ID_SIZE + sizeof(uint32_t) * 2 + sizeof(uint32_t) + packet->dataSize);
     if(sendBuffer==NULL){
         RLOGE("sendBuffer NULL");
-        return;
+        return -1;
     }
 
     size_t headerSize = DEVICE_ID_SIZE + SESSION_ID_SIZE + sizeof(packet->sequenceNumber) 
@@ -897,8 +961,9 @@ void sendRecordingAudioPacket(AudioPacket *packet) {
    // memcpy(recorder->save_buf, (void *)sendBuffer, recorder->save_len);
     //record_msg_send(recorder, RECORD_MSG_DATA, recorder->save_len);
 
-	talkingq_ws_send_bin((uint8_t*)sendBuffer, totalSize);
+	int send_ret = talkingq_ws_send_bin((uint8_t*)sendBuffer, totalSize);
     free(sendBuffer);
+    return send_ret;
  //   rt_kprintf("sendRecordingAudioPacket - Device ID: %s, Session ID: %s\n", deviceId, sessionId);
 }
 #endif
@@ -1005,6 +1070,17 @@ void app_player_thread(void *arvg)
       ql_rtos_queue_wait(player_queue, (u8*)(&player_item), sizeof(ice_work_item *), 0xFFFFFFFF);
 
 		if(player_item!=NULL){
+                int stale_timeout;
+                conversation_lock();
+                stale_timeout = (player_item->type == 3 || player_item->type == 4) &&
+                                player_item->timeout_round != conversation_round;
+                conversation_unlock();
+                if (stale_timeout) {
+                    if (player_mq_couter > 0) player_mq_couter--;
+                    free(player_item->data);
+                    free(player_item);
+                    continue;
+                }
 		 // if(player_item->type==0){//local mp3 file
 			  if(player_item->data){
 			  	 // printf("xiaoice player! %s\n",player_item->data);		  
@@ -1020,17 +1096,22 @@ void app_player_thread(void *arvg)
 				   * type=3 只用于对话超时提示。云端连接、HTTP、读取或播放器初始化
 				   * 失败时才排入本地兜底；如果用户已经开始下一轮录音，则丢弃旧提示。
 				   */
+                      conversation_lock();
 				  if(play_ret != 0 && player_item->type == 3 &&
-				     !vad_audio_is_recording() && tts_playing_ret() != 2){
+				     !vad_audio_is_recording() && !g_mic_working &&
+                         player_item->timeout_round == conversation_round &&
+                         tts_playing_ret() != 2){
 					  printf("cloud tts error play failed:%d, fallback:%s\r\n",
 					         play_ret, LOCAL_TTS_ERROR_PATH);
 					  if(ql_access(LOCAL_TTS_ERROR_PATH, 0) == 0){
-						  add_file_to_play_player_queue(LOCAL_TTS_ERROR_PATH, 0);
+						  add_file_to_play_player_queue(LOCAL_TTS_ERROR_PATH, 4);
 					  }else{
 						  printf("local tts error fallback missing:%s\r\n", LOCAL_TTS_ERROR_PATH);
 					  }
 				  }
 
+                      conversation_unlock();
+
 				  if(player_mq_couter==0){
 					  printf("player_mq_couter==0\r\n");
 
@@ -1069,6 +1150,21 @@ void app_player_thread(void *arvg)
     }
 }
 
+static int conversation_try_begin(void)
+{
+	conversation_lock();
+    if(vad_audio_is_recording() || g_mic_working || shine_init || timeout_action_busy) {
+        conversation_unlock();
+        return -1;
+    }
+    conversation_begin_round();
+    vad_reset_conversation_flags();
+    g_mic_working = 1;
+    conversation_unlock();
+
+    return 0;
+}
+
 int tts_play_interrupt_func(void)
 {
 	int playback_interrupted = 0;
@@ -1097,9 +1193,9 @@ int tts_play_interrupt_func(void)
 	}
 
 	//如果在vad状态，则直接返回
-	if(vad_audio_is_recording()||g_mic_working) return -1;
+    if (conversation_try_begin() != 0) return -1;
 
-	g_mic_working = 1;//等待转为vad状态就清0 或者拾音超时 或者ws没有正常
+	/* g_mic_working reserves startup until VAD enters recording. */
 
 	/* Remember whether this key press is interrupting TTS.  tts_playing == 2
 	 * is only a cancellation signal for the old player; it must not leak into
@@ -1153,9 +1249,54 @@ void vad_start_func(void)
     vad_audio_trigger_record();
 }
 
+static void conversation_check_timeout(void)
+{
+    int expired = WAIT_NONE;
+    unsigned int expired_round = 0;
+    conversation_lock();
+    if (conversation_wait != WAIT_NONE &&
+        wait_round == conversation_round &&
+        !vad_audio_is_recording() && !g_mic_working &&
+        !timeout_action_busy && fota_start != 1 &&
+        (conversation_wait == WAIT_NO_SPEECH ||
+         conversation_wait == WAIT_SEND_FAILURE ||
+         hal_os_up_time_ms() - reply_wait_start_ms > 5000)) {
+        expired = conversation_wait;
+        expired_round = wait_round;
+        conversation_wait = WAIT_NONE;
+        timeout_action_busy = 1;
+    }
+    conversation_unlock();
+
+    if (expired != WAIT_NONE) {
+        char cloud_tts_error_url[128] = {0};
+        /* Reserve the transition, without holding a mutex while
+         * disconnect joins a socket thread/calls callbacks. */
+        if (expired != WAIT_NO_SPEECH) talkingq_ws_disconnect();
+        conversation_lock();
+        if (expired_round == conversation_round) {
+            printf("conversation timeout:round=%u reason=%s\n",
+                   expired_round,
+                   expired == WAIT_NO_SPEECH ? "no_speech" :
+                   (expired == WAIT_REPLY ? "reply" : "send_failure"));
+            tts_playing = 0;
+            tts_play_status = 2;
+            audio_packet_type = 0;
+            vad_set_speech_timeout_flag(1);
+            if (build_random_tts_error_url(cloud_tts_error_url,
+                                          sizeof(cloud_tts_error_url)) == 1) {
+                add_file_to_play_player_queue(cloud_tts_error_url, 3);
+            } else {
+                app_play_tips_file(9);
+            }
+        }
+        timeout_action_busy = 0;
+        conversation_unlock();
+    }
+}
+
 static void check_connect_thread_entry(void *parameter)
 {	
-	uint32 time_counter = 0;
 	uint32 cfun_recorder = 0;
 	int sim_status = 0;
 	int per_socket_status = 0;
@@ -1247,34 +1388,7 @@ static void check_connect_thread_entry(void *parameter)
 	//}
 #endif
 #if 1
-		//如果后台10s以上没有反馈 则进行重连
+            conversation_check_timeout();
-		if(wss_connect_start==1){
-			wss_connect_start = 2;
-			time_counter = ql_rtos_get_systicks();
-		}else if(wss_connect_start==2 && (ql_rtos_get_systicks()-time_counter)>5000/5){
-			wss_connect_start = 0;
-			tts_playing = 0;
-			audio_packet_type = 0;
-			
-			talkingq_ws_disconnect();
-						
-			printf("tts_error_set_speech1\n");
-			vad_set_speech_timeout_flag(1);
-			
-			/*
-			 * 后台没有返回任何音频 URL 时，也先由设备生成随机云端提示音。
-			 * type=3 会在 HTTP 下载或播放初始化失败时自动切换到本地音频。
-			 */
-			{
-				char cloud_tts_error_url[128] = {0};
-				if(build_random_tts_error_url(cloud_tts_error_url,
-				                              sizeof(cloud_tts_error_url)) == 1){
-					add_file_to_play_player_queue(cloud_tts_error_url, 3);
-				}else{
-					app_play_tips_file(9);
-				}
-			}
-		}
 	#endif
 	
 		ql_rtos_task_sleep_ms(3000);
@@ -1308,9 +1422,9 @@ void xiaoice_close_all_audio(void)
 			
 #if 1
 	if(tts_playing){
-		tts_playing = 2;
 		sendInterruptPacket();
 	}
+    if (tts_playing || player_mq_couter > 0) tts_playing = 2;
 		
 	if(player_mq_couter>0){
 		
@@ -1385,6 +1499,9 @@ sender_loop:
 				sh_config.mpeg.bitr=16; //64kbps for 16k //32kbps for 8k samplerate
 				sh_shine=shine_initialise(&sh_config);
 				shine_init = 1;
+                    conversation_lock();
+                    sender_round = conversation_round;
+                    conversation_unlock();
 				talkingq_ws_connect();
 				sendStartPacket();	
 			}
@@ -1479,6 +1596,7 @@ void app_check_connect_sem_release(void)
 
 int app_ws_talkingq_init()
 {
+	if (ql_rtos_mutex_create(&conversation_mutex) != 0) return -1;
 	//one second   = 640*50  = 32000  
 	ring_sender = hal_os_rb_create(32000*5);
 	ql_rtos_semaphore_create(&sender_sem, 0);

tools/test_conversation_timeout.py

@@ -0,0 +1,105 @@
+"""Run firmware timeout/round functions on host with scripted RTOS and transport."""
+from pathlib import Path
+import subprocess
+import tempfile
+
+root = Path(__file__).resolve().parents[1]
+source = (root / 'ql-application/threadx/interface/custom_app/ws_talkingq.c').read_text()
+
+def function(name):
+    start = source.index(name + '(')
+    start = source.rfind('\n', 0, start) + 1
+    opening = source.index('{', start)
+    depth = 1
+    end = opening + 1
+    while depth:
+        depth += (source[end] == '{') - (source[end] == '}')
+        end += 1
+    return source[start:end] + '\n'
+
+state_start = source.index('static ql_mutex_t conversation_mutex')
+state_end = source.index('static volatile int player_mq_couter')
+harness = r'''
+#include <assert.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stddef.h>
+typedef void *ql_mutex_t;
+static int g_mic_working, recording, shine_init, fota_start;
+static int tts_playing, tts_play_status, audio_packet_type, no_speech_timeout;
+static int queued, disconnected, flags_reset, mutex_locked;
+static uint64_t now;
+static unsigned int gSequenceNumber;
+static int end_result;
+typedef struct { unsigned int sequenceNumber; int packetType, dataSize; } AudioPacket;
+static int ql_rtos_mutex_lock(ql_mutex_t m, unsigned int t) {
+    (void)m; (void)t; assert(!mutex_locked); mutex_locked = 1; return 0;
+}
+static void ql_rtos_mutex_unlock(ql_mutex_t m) { (void)m; assert(mutex_locked); mutex_locked = 0; }
+static int vad_audio_is_recording(void) { return recording; }
+static uint64_t hal_os_up_time_ms(void) { return now; }
+static void vad_set_speech_timeout_flag(int value) { no_speech_timeout = value; }
+static void vad_reset_conversation_flags(void) { no_speech_timeout = 0; flags_reset++; }
+static int build_random_tts_error_url(char *p, size_t n) { (void)p; (void)n; return 1; }
+static void add_file_to_play_player_queue(char *p, int type) { (void)p; assert(type == 3); queued++; }
+static void app_play_tips_file(int n) { (void)n; assert(0); }
+static int conversation_try_begin(void);
+static void talkingq_ws_disconnect(void) {
+    assert(!mutex_locked); disconnected++;
+    /* A key racing a committed timeout cannot start inside disconnect. */
+    assert(conversation_try_begin() == -1);
+}
+static int sendRecordingAudioPacket(AudioPacket *p) { assert(p->packetType == 2); return end_result; }
+'''
+harness += source[state_start:state_end]
+for name in ['conversation_try_begin', 'sendEndPacket', 'conversation_check_timeout']:
+    harness += function(name)
+harness += r'''
+static void reset(void) {
+    g_mic_working = recording = shine_init = fota_start = 0;
+    conversation_round = wait_round = sender_round = 0;
+    conversation_wait = WAIT_NONE; timeout_action_busy = round_response_received = 0;
+    queued = disconnected = flags_reset = no_speech_timeout = mutex_locked = 0;
+    now = 1000; end_result = 0;
+}
+int main(void) {
+    /* No confirmed speech: prompt remains, without disconnect or reply wait. */
+    reset(); assert(conversation_try_begin() == 0); g_mic_working = 0;
+    conversation_no_speech(); conversation_check_timeout();
+    assert(queued == 1 && disconnected == 0 && no_speech_timeout == 1);
+    conversation_check_timeout(); assert(queued == 1);
+    /* New key cancels pending old no-speech prompt. */
+    reset(); conversation_no_speech(); assert(conversation_try_begin() == 0);
+    g_mic_working = 0; now += 20000; conversation_check_timeout();
+    assert(!queued && !disconnected && !no_speech_timeout);
+    /* Repeated keys during either waiting-for-speech or active speech are ignored. */
+    reset(); recording = 1; conversation_wait = WAIT_REPLY; wait_round = 0;
+    assert(conversation_try_begin() == -1);
+    assert(conversation_round == 0 && conversation_wait == WAIT_REPLY && !flags_reset);
+    /* END starts reply deadline; stale round and early URL cannot arm it. */
+    reset(); sendEndPacket(); assert(conversation_wait == WAIT_REPLY);
+    now += 5000; conversation_check_timeout(); assert(!queued);
+    now++; conversation_check_timeout(); assert(queued == 1 && disconnected == 1);
+    reset(); sendEndPacket(); assert(conversation_try_begin() == 0);
+    g_mic_working = 0; now += 20000; conversation_check_timeout(); assert(!queued && !disconnected);
+    reset(); conversation_round = 1; sendEndPacket(); assert(conversation_wait == WAIT_NONE);
+    reset(); round_response_received = 1; sendEndPacket(); assert(conversation_wait == WAIT_NONE);
+    /* END queue failure still produces an error instead of waiting forever. */
+    reset(); end_result = -1; sendEndPacket(); assert(conversation_wait == WAIT_SEND_FAILURE);
+    conversation_check_timeout(); assert(queued == 1 && disconnected == 1);
+    /* Prompt completion preserves the key-reserved round; automatic listening advances. */
+    reset(); assert(conversation_try_begin() == 0); conversation_record_begin(); assert(conversation_round == 1);
+    g_mic_working = 0; conversation_record_begin(); assert(conversation_round == 2);
+    /* Force-stop cancels waits; stale timeout owner cannot affect a new round. */
+    reset(); conversation_no_speech(); conversation_cancel_wait(); conversation_check_timeout(); assert(!queued);
+    reset(); conversation_wait = WAIT_REPLY; wait_round = 9; now += 20000;
+    conversation_check_timeout(); assert(!queued && !disconnected);
+    puts("PASS: no-speech prompt, new-round cancellation, ignored recording keys, END deadline, stale rounds, early reply, send failure, timeout/key race, auto-listening");
+}
+'''
+with tempfile.TemporaryDirectory(prefix='banban-timeout-') as directory:
+    cfile = Path(directory) / 'timeout.c'
+    exe = Path(directory) / 'timeout-test'
+    cfile.write_text(harness)
+    subprocess.run(['cc', '-std=c99', '-Wall', '-Wextra', '-Werror', str(cfile), '-o', str(exe)], check=True)
+    subprocess.run([str(exe)], check=True)