chenjintao/lucky-cat
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(multimodal): recover stalled downstream sessions

Browse Source
This commit is contained in:
chenjintao
2026-08-12 16:32:25 +08:00
parent 4e7fadde56
commit a94df9545d
9 changed files with 272 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/control_plane/internal/control_plane_internal.h
+4 -1
View File
@@ -56,7 +56,8 @@ typedef enum {
CONTROL_PLANE_MULTIMODAL_EVENT_SPEECH_CONTENT,
CONTROL_PLANE_MULTIMODAL_EVENT_SPEECH_ENDED,
CONTROL_PLANE_MULTIMODAL_EVENT_OUTPUT_ERROR,
CONTROL_PLANE_MULTIMODAL_EVENT_DOWNSTREAM_AUDIO
CONTROL_PLANE_MULTIMODAL_EVENT_DOWNSTREAM_AUDIO,
CONTROL_PLANE_MULTIMODAL_EVENT_DOWNSTREAM_STALLED,
} control_plane_multimodal_event_t;
typedef void (*control_plane_multimodal_event_callback_t)(control_plane_multimodal_event_t event,
@@ -90,6 +91,7 @@ typedef enum {
CONTROL_PLANE_TRANSPORT_LOSS_SEND_BINARY_FAILURE,
CONTROL_PLANE_TRANSPORT_LOSS_LISTENING_TIMEOUT,
CONTROL_PLANE_TRANSPORT_LOSS_NO_RESPONSE_TIMEOUT,
CONTROL_PLANE_TRANSPORT_LOSS_DOWNSTREAM_STALL,
CONTROL_PLANE_TRANSPORT_LOSS_PROTOCOL_ERROR,
} control_plane_transport_loss_reason_t;
@@ -98,6 +100,7 @@ void control_plane_multimodal_stop(void);
bool control_plane_multimodal_is_connected(void);
esp_err_t control_plane_multimodal_send_start(void);
esp_err_t control_plane_multimodal_send_stop(const char *reason);
esp_err_t control_plane_multimodal_request_downstream_stall_stop(void);
esp_err_t control_plane_multimodal_send_request_to_speak(void);
esp_err_t control_plane_multimodal_send_speech_start(void);
esp_err_t control_plane_multimodal_send_speech_stop(void);
components/control_plane/src/multimodal/control_plane_multimodal_runtime.c
+111
View File
@@ -36,6 +36,7 @@
#endif
#define CONTROL_PLANE_DOWNSTREAM_PLAYBACK_GUARD_MS 220
#define CONTROL_PLANE_DOWNSTREAM_STALL_STOP_GRACE_MS 2000
#define CONTROL_PLANE_IDLE_TRANSPORT_SHUTDOWN_MS 30000
#define CONTROL_PLANE_WATCHDOG_TASK_STACK_SIZE 6144
#define CONTROL_PLANE_SERVER_SUCCESS_PROMPT_FILE "/littlefs/audio_prompts/server_connect_success.opus"
@@ -57,6 +58,9 @@ static bool s_multimodal_watchdog_task_starting;
static bool s_model_command_dispatched;
static bool s_server_success_prompt_played;
static bool s_cloud_chouqian_pending;
static bool s_downstream_stall_recovery_pending;
static bool s_downstream_stall_stop_sent;
static uint32_t s_downstream_stall_recovery_started_ms;
static control_plane_model_command_callback_t s_model_command_cb;
static void *s_model_command_cb_user_data;
static portMUX_TYPE s_multimodal_runtime_lock = portMUX_INITIALIZER_UNLOCKED;
@@ -118,6 +122,55 @@ static void set_multimodal_start_flag(bool *flag, bool value)
portEXIT_CRITICAL(&s_multimodal_runtime_lock);
}
static bool control_plane_begin_downstream_stall_recovery(void)
{
bool started = false;
portENTER_CRITICAL(&s_multimodal_runtime_lock);
if (!s_downstream_stall_recovery_pending) {
s_downstream_stall_recovery_pending = true;
s_downstream_stall_stop_sent = false;
s_downstream_stall_recovery_started_ms = 0U;
started = true;
}
portEXIT_CRITICAL(&s_multimodal_runtime_lock);
return started;
}
static bool control_plane_get_downstream_stall_recovery(bool *out_stop_sent,
uint32_t *out_started_ms)
{
bool pending = false;
portENTER_CRITICAL(&s_multimodal_runtime_lock);
pending = s_downstream_stall_recovery_pending;
if (out_stop_sent != NULL) {
*out_stop_sent = s_downstream_stall_stop_sent;
}
if (out_started_ms != NULL) {
*out_started_ms = s_downstream_stall_recovery_started_ms;
}
portEXIT_CRITICAL(&s_multimodal_runtime_lock);
return pending;
}
static void control_plane_mark_downstream_stall_stop_sent(uint32_t now_ms)
{
portENTER_CRITICAL(&s_multimodal_runtime_lock);
if (s_downstream_stall_recovery_pending) {
s_downstream_stall_stop_sent = true;
s_downstream_stall_recovery_started_ms = now_ms;
}
portEXIT_CRITICAL(&s_multimodal_runtime_lock);
}
static void control_plane_clear_downstream_stall_recovery(void)
{
portENTER_CRITICAL(&s_multimodal_runtime_lock);
s_downstream_stall_recovery_pending = false;
s_downstream_stall_stop_sent = false;
s_downstream_stall_recovery_started_ms = 0U;
portEXIT_CRITICAL(&s_multimodal_runtime_lock);
}
static bool try_begin_multimodal_task_start(TaskHandle_t *slot, bool *start_flag)
{
bool can_start = false;
@@ -454,6 +507,49 @@ static void multimodal_watchdog_task(void *arg)
}
uint32_t now_ms = control_plane_now_ms();
bool stall_stop_sent = false;
uint32_t stall_recovery_started_ms = 0U;
if (control_plane_get_downstream_stall_recovery(&stall_stop_sent,
&stall_recovery_started_ms)) {
if (!state.connected) {
control_plane_clear_downstream_stall_recovery();
} else if (!stall_stop_sent) {
ESP_LOGW(TAG,
"Downstream stall recovery started: request Stop and wait up to %u ms",
(unsigned)CONTROL_PLANE_DOWNSTREAM_STALL_STOP_GRACE_MS);
esp_err_t stop_err =
control_plane_multimodal_request_downstream_stall_stop();
if (stop_err == ESP_OK) {
control_plane_mark_downstream_stall_stop_sent(now_ms);
} else {
ESP_LOGW(TAG,
"Downstream stall Stop request failed: %s; retire transport now",
esp_err_to_name(stop_err));
control_plane_clear_downstream_stall_recovery();
control_plane_multimodal_mark_transport_unavailable(
CONTROL_PLANE_TRANSPORT_LOSS_DOWNSTREAM_STALL,
"downstream_stall_stop_failed");
(void)control_plane_schedule_websocket_wait_task();
}
continue;
} else if (!state.has_active_task) {
ESP_LOGI(TAG,
"Downstream stall recovered by server Stop acknowledgement; keep WebSocket");
control_plane_clear_downstream_stall_recovery();
} else if ((now_ms - stall_recovery_started_ms) >=
CONTROL_PLANE_DOWNSTREAM_STALL_STOP_GRACE_MS) {
ESP_LOGW(TAG,
"Downstream stall Stop acknowledgement timed out after %u ms; retire transport",
(unsigned)CONTROL_PLANE_DOWNSTREAM_STALL_STOP_GRACE_MS);
control_plane_clear_downstream_stall_recovery();
control_plane_multimodal_mark_transport_unavailable(
CONTROL_PLANE_TRANSPORT_LOSS_DOWNSTREAM_STALL,
"downstream_stall_recovery");
(void)control_plane_schedule_websocket_wait_task();
continue;
}
}
if (state.connected && !state.has_active_task) {
if (idle_transport_started_ms == 0U) {
idle_transport_started_ms = now_ms;
@@ -545,6 +641,7 @@ static void control_plane_multimodal_event_bridge(control_plane_multimodal_event
}
case CONTROL_PLANE_MULTIMODAL_EVENT_TRANSPORT_DISCONNECTED:
control_plane_clear_downstream_stall_recovery();
control_plane_clear_cloud_chouqian_pending("transport disconnected");
control_plane_multimodal_stop_uplink();
ESP_LOGI(TAG, "Schedule fresh-auth reconnect after transport disconnect");
@@ -552,6 +649,7 @@ static void control_plane_multimodal_event_bridge(control_plane_multimodal_event
break;
case CONTROL_PLANE_MULTIMODAL_EVENT_TRANSPORT_ERROR:
control_plane_clear_downstream_stall_recovery();
control_plane_clear_cloud_chouqian_pending("transport error");
control_plane_multimodal_stop_uplink();
ESP_LOGI(TAG, "Schedule fresh-auth reconnect after transport error");
@@ -559,6 +657,7 @@ static void control_plane_multimodal_event_bridge(control_plane_multimodal_event
break;
case CONTROL_PLANE_MULTIMODAL_EVENT_TASK_FAILED:
control_plane_clear_downstream_stall_recovery();
control_plane_clear_cloud_chouqian_pending("task failed");
control_plane_request_server_error_expression(DOMAIN_EXPRESSION_TRIGGER_SERVER_TASK_FAILED);
control_plane_multimodal_stop_uplink();
@@ -605,6 +704,15 @@ static void control_plane_multimodal_event_bridge(control_plane_multimodal_event
control_plane_request_server_error_expression(DOMAIN_EXPRESSION_TRIGGER_SERVER_OUTPUT_ERROR);
break;
case CONTROL_PLANE_MULTIMODAL_EVENT_DOWNSTREAM_STALLED:
if (control_plane_begin_downstream_stall_recovery()) {
ESP_LOGW(TAG,
"Downstream stall recovery queued for watchdog worker");
} else {
ESP_LOGI(TAG, "Downstream stall recovery already pending");
}
break;
default:
break;
}
@@ -612,6 +720,8 @@ static void control_plane_multimodal_event_bridge(control_plane_multimodal_event
esp_err_t control_plane_start_multimodal_client(void)
{
control_plane_clear_downstream_stall_recovery();
esp_err_t prepare_err = control_plane_prepare_for_multimodal_start();
if (prepare_err != ESP_OK) {
ESP_LOGW(TAG, "Prepare multimodal start failed: %s", esp_err_to_name(prepare_err));
@@ -666,6 +776,7 @@ esp_err_t control_plane_start_multimodal_client(void)
void control_plane_stop_multimodal_client(void)
{
control_plane_clear_downstream_stall_recovery();
control_plane_clear_cloud_chouqian_pending("multimodal client stopped");
stop_watchdog_task();
control_plane_multimodal_stop();
components/control_plane/src/multimodal/control_plane_multimodal_session_service.c
+136 -10
View File
@@ -93,6 +93,10 @@
#define CONFIG_TQ_LOCAL_TASK_CORE 1
#endif
#ifndef CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
#define CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE 0
#endif
#if CONFIG_TQ_LOCAL_TASK_CORE < 0
#define CONTROL_PLANE_LOCAL_RESPONDING_TASK_CORE tskNO_AFFINITY
#else
@@ -122,6 +126,9 @@ static const char *TAG = "cp_mm_session";
static bool s_started;
static bool s_starting;
static bool s_stopping;
#if CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
static uint32_t s_session_generation;
#endif
static control_plane_multimodal_auth_info_t s_auth;
static char s_task_id[40];
@@ -143,6 +150,7 @@ static bool s_ptt_release_required;
static bool s_local_responding_started;
static bool s_local_responding_ended_sent;
static bool s_downstream_playback_completed;
static bool s_downstream_stall_stop_pending;
static bool s_drop_downstream_audio;
static bool s_received_downstream_audio;
static bool s_current_response_received_audio;
@@ -381,6 +389,7 @@ static void reset_session_state_locked(void)
s_local_responding_started = false;
s_local_responding_ended_sent = false;
s_downstream_playback_completed = false;
s_downstream_stall_stop_pending = false;
s_drop_downstream_audio = false;
s_received_downstream_audio = false;
s_current_response_received_audio = false;
@@ -422,6 +431,8 @@ static const char *transport_loss_reason_name(control_plane_transport_loss_reaso
return "listening_timeout";
case CONTROL_PLANE_TRANSPORT_LOSS_NO_RESPONSE_TIMEOUT:
return "no_response_timeout";
case CONTROL_PLANE_TRANSPORT_LOSS_DOWNSTREAM_STALL:
return "downstream_stall";
case CONTROL_PLANE_TRANSPORT_LOSS_PROTOCOL_ERROR:
return "protocol_error";
default:
@@ -443,10 +454,23 @@ void control_plane_multimodal_mark_transport_unavailable(
control_plane_transport_loss_reason_t reason,
const char *caller)
{
#if CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
uint32_t session_generation = 0U;
if (lock_state()) {
session_generation = s_session_generation;
unlock_state();
}
ESP_LOGW(TAG,
"Transport unavailable: session_generation=%u reason=%s caller=%s",
(unsigned)session_generation,
transport_loss_reason_name(reason),
caller != NULL ? caller : "-");
#else
ESP_LOGW(TAG,
"Transport unavailable: reason=%s caller=%s",
transport_loss_reason_name(reason),
caller != NULL ? caller : "-");
#endif
if (lock_state()) {
s_started = false;
@@ -465,7 +489,8 @@ void control_plane_multimodal_mark_transport_unavailable(
reason == CONTROL_PLANE_TRANSPORT_LOSS_WS_DISCONNECTED ||
reason == CONTROL_PLANE_TRANSPORT_LOSS_SEND_TEXT_FAILURE ||
reason == CONTROL_PLANE_TRANSPORT_LOSS_LISTENING_TIMEOUT ||
reason == CONTROL_PLANE_TRANSPORT_LOSS_NO_RESPONSE_TIMEOUT;
reason == CONTROL_PLANE_TRANSPORT_LOSS_NO_RESPONSE_TIMEOUT ||
reason == CONTROL_PLANE_TRANSPORT_LOSS_DOWNSTREAM_STALL;
if (close_transport) {
ESP_LOGI(TAG,
"Retire unavailable transport: reason=%s; next start requires fresh auth",
@@ -1422,18 +1447,27 @@ static void process_text_json(const char *json_text)
notify_event(CONTROL_PLANE_MULTIMODAL_EVENT_RESPONDING_ENDED);
maybe_send_local_responding_ended_if_ready();
} else if (strcmp(output_event->valuestring, "Stopped") == 0) {
bool downstream_stall_stop_acknowledged = false;
if (lock_state()) {
s_server_listening = false;
s_remote_responding = false;
s_remote_responding_ended = false;
s_request_to_speak_pending = false;
s_client_speech_active = false;
s_ptt_release_required = false;
s_local_responding_started = false;
s_downstream_playback_completed = false;
s_drop_downstream_audio = false;
downstream_stall_stop_acknowledged = s_downstream_stall_stop_pending;
if (downstream_stall_stop_acknowledged) {
reset_session_state_locked();
} else {
s_server_listening = false;
s_remote_responding = false;
s_remote_responding_ended = false;
s_request_to_speak_pending = false;
s_client_speech_active = false;
s_ptt_release_required = false;
s_local_responding_started = false;
s_downstream_playback_completed = false;
s_drop_downstream_audio = false;
}
unlock_state();
}
if (downstream_stall_stop_acknowledged) {
ESP_LOGI(TAG, "Downstream stall Stop acknowledged; transport remains reusable");
}
stop_local_downstream_playback();
notify_event(CONTROL_PLANE_MULTIMODAL_EVENT_SERVER_NOT_LISTENING);
} else if (strcmp(output_event->valuestring, "SpeechStarted") == 0) {
@@ -1622,6 +1656,27 @@ static void downstream_playback_event_bridge(
send_local_responding_ended_after_playback_stop();
break;
case DOMAIN_MULTIMODAL_DOWNSTREAM_PLAYBACK_STALLED: {
bool cloud_response_active = false;
if (lock_state()) {
cloud_response_active = s_task_id[0] != '\0' &&
(s_remote_responding ||
s_current_response_received_audio);
unlock_state();
}
send_local_responding_ended_after_playback_stop();
if (cloud_response_active) {
ESP_LOGW(TAG,
"Cloud downstream playback stalled; request bounded session recovery");
notify_event(CONTROL_PLANE_MULTIMODAL_EVENT_DOWNSTREAM_STALLED);
} else {
ESP_LOGW(TAG,
"Local downstream playback stalled; keep cloud transport unchanged");
}
break;
}
default:
break;
}
@@ -1640,6 +1695,9 @@ esp_err_t control_plane_multimodal_start(void)
return ESP_ERR_NOT_SUPPORTED;
#endif
#if CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
uint32_t session_generation = 0U;
#endif
if (lock_state()) {
bool already_started = s_started || s_auth.token[0] != '\0';
bool starting = s_starting;
@@ -1653,9 +1711,42 @@ esp_err_t control_plane_multimodal_start(void)
return ESP_ERR_INVALID_STATE;
}
s_starting = true;
#if CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
s_session_generation++;
session_generation = s_session_generation;
#endif
unlock_state();
}
#if CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
ESP_LOGI(TAG,
"Multimodal start generation=%u: wait for previous transport cleanup",
(unsigned)session_generation);
#endif
esp_err_t cleanup_err = domain_runtime_multimodal_transport_stop(2000);
if (cleanup_err != ESP_OK) {
#if CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
ESP_LOGW(TAG,
"Multimodal start generation=%u blocked by previous transport cleanup: %s",
(unsigned)session_generation,
esp_err_to_name(cleanup_err));
#else
ESP_LOGW(TAG,
"Multimodal start blocked by previous transport cleanup: %s",
esp_err_to_name(cleanup_err));
#endif
if (lock_state()) {
s_starting = false;
unlock_state();
}
return cleanup_err;
}
#if CONFIG_TQ_MULTIMODAL_DIAGNOSTICS_ENABLE
ESP_LOGI(TAG,
"Multimodal start generation=%u: previous transport cleanup complete; begin auth",
(unsigned)session_generation);
#endif
control_plane_multimodal_auth_info_t auth = {0};
esp_err_t err = control_plane_multimodal_auth_fetch(&auth);
if (err != ESP_OK) {
@@ -1981,6 +2072,41 @@ esp_err_t control_plane_multimodal_send_stop(const char *reason)
return err;
}
esp_err_t control_plane_multimodal_request_downstream_stall_stop(void)
{
control_plane_multimodal_stop_uplink();
bool already_pending = false;
bool cloud_response_active = false;
if (!lock_state()) {
return ESP_ERR_TIMEOUT;
}
already_pending = s_downstream_stall_stop_pending;
cloud_response_active = s_task_id[0] != '\0' &&
(s_remote_responding || s_current_response_received_audio);
if (!already_pending && cloud_response_active) {
s_downstream_stall_stop_pending = true;
s_request_to_speak_pending = false;
s_drop_downstream_audio = true;
}
unlock_state();
if (already_pending) {
return ESP_OK;
}
if (!cloud_response_active) {
return ESP_ERR_INVALID_STATE;
}
ESP_LOGW(TAG, "Request Stop for stalled downstream response");
esp_err_t err = send_directive("finish-task", "Stop", true, false);
if (err != ESP_OK && lock_state()) {
s_downstream_stall_stop_pending = false;
unlock_state();
}
return err;
}
esp_err_t control_plane_multimodal_send_request_to_speak(void)
{
if (!transport_available()) {
components/domain/include/domain.h
+1
View File
@@ -98,6 +98,7 @@ typedef enum {
DOMAIN_MULTIMODAL_DOWNSTREAM_PLAYBACK_STARTED = 0,
DOMAIN_MULTIMODAL_DOWNSTREAM_PLAYBACK_COMPLETED,
DOMAIN_MULTIMODAL_DOWNSTREAM_PLAYBACK_STOPPED,
DOMAIN_MULTIMODAL_DOWNSTREAM_PLAYBACK_STALLED,
} domain_multimodal_downstream_playback_event_type_t;
typedef struct {
components/domain/src/runtime/domain_runtime_adapters.c
+7
View File
@@ -452,6 +452,12 @@ static void domain_downstream_playback_stopped_adapter(void *user_data)
domain_downstream_playback_dispatch(DOMAIN_MULTIMODAL_DOWNSTREAM_PLAYBACK_STOPPED);
}
static void domain_downstream_playback_stalled_adapter(void *user_data)
{
(void)user_data;
domain_downstream_playback_dispatch(DOMAIN_MULTIMODAL_DOWNSTREAM_PLAYBACK_STALLED);
}
static void domain_wifi_set_keep_awake_adapter(bool keep_awake, void *user_data)
{
(void)user_data;
@@ -688,6 +694,7 @@ esp_err_t domain_runtime_multimodal_audio_register_downstream_playback_callback(
.on_started = domain_downstream_playback_started_adapter,
.on_completed = domain_downstream_playback_completed_adapter,
.on_stopped = domain_downstream_playback_stopped_adapter,
.on_stalled = domain_downstream_playback_stalled_adapter,
.user_data = NULL,
};
components/platform/include/platform.h
+1
View File
@@ -46,6 +46,7 @@ typedef struct {
void (*on_started)(void *user_data);
void (*on_completed)(void *user_data);
void (*on_stopped)(void *user_data);
void (*on_stalled)(void *user_data);
void *user_data;
} platform_multimodal_downstream_playback_callbacks_t;
components/platform/internal/multimodal/audio_pipeline.h
+1
View File
@@ -20,6 +20,7 @@ typedef enum {
MULTIMODAL_AUDIO_DOWNSTREAM_PLAYBACK_STARTED = 0,
MULTIMODAL_AUDIO_DOWNSTREAM_PLAYBACK_COMPLETED,
MULTIMODAL_AUDIO_DOWNSTREAM_PLAYBACK_STOPPED,
MULTIMODAL_AUDIO_DOWNSTREAM_PLAYBACK_STALLED,
} multimodal_audio_downstream_playback_event_type_t;
typedef struct {
components/platform/src/multimodal/audio_pipeline/audio_pipeline_runtime_impl.c
+6 -1
View File
@@ -1774,6 +1774,11 @@ static void notify_downstream_playback_stopped(void)
notify_downstream_playback_finished(MULTIMODAL_AUDIO_DOWNSTREAM_PLAYBACK_STOPPED);
}
static void notify_downstream_playback_stalled(void)
{
notify_downstream_playback_finished(MULTIMODAL_AUDIO_DOWNSTREAM_PLAYBACK_STALLED);
}
static void abort_downstream_response_after_active_stall(multimodal_audio_ctx_t *ctx,
uint32_t idle_ms)
{
@@ -1781,7 +1786,7 @@ static void abort_downstream_response_after_active_stall(multimodal_audio_ctx_t
"Downstream protection watchdog: abort stalled response idle_ms=%u",
(unsigned)idle_ms);
(void)advance_downstream_playback_generation();
notify_downstream_playback_stopped();
notify_downstream_playback_stalled();
update_downstream_response_state(false, false);
if (audio_get_downstream_playback_task(ctx) != NULL ||
audio_get_downstream_playback_starting(ctx)) {
components/platform/src/runtime/platform_multimodal_runtime.c
+5
View File
@@ -243,6 +243,11 @@ static void platform_multimodal_downstream_playback_bridge(
callbacks.on_stopped(callbacks.user_data);
}
break;
case MULTIMODAL_AUDIO_DOWNSTREAM_PLAYBACK_STALLED:
if (callbacks.on_stalled != NULL) {
callbacks.on_stalled(callbacks.user_data);
}
break;
default:
break;
}