#include "cloud_client.h"

#include <cstdio>
#include <cstring>
#include <string>
#include <algorithm>
#include <atomic>
#include <cstdlib>
#include "voice_delivery.h"
#include "audio_backend.h"
#include "input_sender_timing.h"
#include "voice_capture_limits.h"
#include "ui_button_outbox.h"
#include "cloud_link_state.h"
#include "websocket_text_receiver.h"
#include "mbedtls/base64.h"

#include "board.h"
#include "cJSON.h"
#include "esp_check.h"
#include "esp_crt_bundle.h"
#include "esp_log.h"
#include "esp_heap_caps.h"
#include "esp_mac.h"
#include "esp_timer.h"
#include "esp_wifi.h"
#include "esp_websocket_client.h"
#include "freertos/FreeRTOS.h"
#include "freertos/event_groups.h"
#include "freertos/task.h"
#include "wifi_manager.h"

static const char *TAG = "CLOUD";
static constexpr int CLOUD_RECONNECT_TIMEOUT_MS = 1500;
static constexpr int CLOUD_RX_BUFFER_BYTES = 16 * 1024;
static esp_websocket_client_handle_t s_client;
static std::atomic<bool> s_connected{false};
static std::atomic<bool> s_ready{false};
static std::atomic<uint32_t> s_generation{0};
// ACK handoff from the websocket RX task to the uploader. The websocket task
// only stores the newest acknowledged offset and pokes the uploader; it never
// blocks on the transport or the upload loop.
static std::atomic<uint64_t> s_voice_ack_id{0};
static std::atomic<unsigned> s_voice_ack_next{0};
// Stamped with the capture generation the ack belongs to. An ack that arrives
// after a fresh capture started must never advance the new upload: its sequence
// space is unrelated, and honouring it would skip recorded PCM.
static std::atomic<uint32_t> s_voice_ack_generation{0};
static std::atomic<bool> s_voice_ack_pending{false};
static std::atomic<uint64_t> s_voice_end_ack_id{0};
static std::atomic<uint32_t> s_voice_end_ack_generation{0};
static TaskHandle_t s_input_sender_task;
static void publish_voice_ack(uint64_t voice_id, unsigned next_sequence, uint32_t generation) {
    s_voice_ack_id.store(voice_id);
    s_voice_ack_next.store(next_sequence);
    s_voice_ack_generation.store(generation);
    s_voice_ack_pending.store(true);
    if (s_input_sender_task) xTaskNotifyGive(s_input_sender_task);
}
// An ack may only move the uploader forward to the next unacknowledged chunk.
// In normal operation that is exactly sent+1. A resume handshake may legitimately
// report a breakpoint anywhere in the already-sent range (the gateway's view of
// what it stored), but never beyond it: a skipping ack would drop PCM silently.
static bool accept_ack(unsigned next, unsigned sent, bool) {
    return wearable::voice_ack_valid(next,sent,false,0);
}
static CloudLinkState s_link;
static char s_device_id[24];
static WebSocketTextReceiver s_text_receiver;
static int64_t s_rx_started_us, s_last_rx_trace_us;
static int64_t s_last_text_us, s_last_pong_us;
static unsigned s_rx_parts;
static UiButtonOutbox s_button_outbox;

static const char *link_error_type(esp_websocket_error_type_t type)
{
    switch (type) {
    case WEBSOCKET_ERROR_TYPE_TCP_TRANSPORT: return "TCP";
    case WEBSOCKET_ERROR_TYPE_PONG_TIMEOUT: return "PONG_TIMEOUT";
    case WEBSOCKET_ERROR_TYPE_HANDSHAKE: return "HANDSHAKE";
    case WEBSOCKET_ERROR_TYPE_SERVER_CLOSE: return "SERVER_CLOSE";
    default: return "NONE";
    }
}

static void log_connection(const char *kind, const esp_websocket_event_data_t *event,
                           uint64_t recovery_ms = 0)
{
    wifi_ap_record_t ap = {};
    const int rssi = esp_wifi_sta_get_ap_info(&ap) == ESP_OK ? ap.rssi : -127;
    // 1.8.0 only initializes TLS/socket fields for TCP_TRANSPORT events.
    esp_websocket_error_codes_t error = {};
    if (event) {
        error.error_type = event->error_handle.error_type;
        error.esp_ws_handshake_status_code = event->error_handle.esp_ws_handshake_status_code;
        if (error.error_type == WEBSOCKET_ERROR_TYPE_TCP_TRANSPORT) {
            error.esp_tls_last_esp_err = event->error_handle.esp_tls_last_esp_err;
            error.esp_tls_stack_err = event->error_handle.esp_tls_stack_err;
            error.esp_transport_sock_errno = event->error_handle.esp_transport_sock_errno;
        }
    }
    const uint64_t now = esp_timer_get_time() / 1000;
    ESP_LOGI(TAG, "[CLOUD_LINK] event=%s type=%s esp_err=0x%x tls_err=%d sock_errno=%d close_code=%d http_status=%d generation=%lu attempt=%lu uptime_ms=%llu connection_uptime_ms=%llu recovery_ms=%llu rssi=%d internal_free=%u dma_largest=%u",
             kind, link_error_type(error.error_type), static_cast<unsigned>(error.esp_tls_last_esp_err),
             error.esp_tls_stack_err, error.esp_transport_sock_errno, event ? event->close_status_code : 0,
             error.esp_ws_handshake_status_code, static_cast<unsigned long>(s_link.generation),
             static_cast<unsigned long>(s_link.attempt), static_cast<unsigned long long>(now),
             static_cast<unsigned long long>(s_link.generation ? now-s_link.connected_ms : 0),
             static_cast<unsigned long long>(recovery_ms), rssi,
             static_cast<unsigned>(heap_caps_get_free_size(MALLOC_CAP_INTERNAL)),
             static_cast<unsigned>(heap_caps_get_largest_free_block(MALLOC_CAP_DMA)));
}

static void mark_synchronized()
{
    if (!s_link.ready() || s_ready) return;
    const auto recovery_ms = s_link.synchronized(esp_timer_get_time()/1000);
    s_ready = true;
    board_set_cloud_status(true);
    log_connection("synchronized", nullptr, recovery_ms);
}

static void log_link_health(const char *reason, size_t bytes, unsigned parts, int64_t elapsed_ms)
{
    wifi_ap_record_t ap = {};
    const int rssi = esp_wifi_sta_get_ap_info(&ap) == ESP_OK ? ap.rssi : -127;
    wifi_ps_type_t power_save = WIFI_PS_NONE;
    esp_wifi_get_ps(&power_save);
    ESP_LOGI(TAG, "[CLOUD] link reason=%s bytes=%u parts=%u elapsed_ms=%lld rssi=%d ps=%d last_text_age_ms=%lld last_pong_age_ms=%lld internal_free=%u dma_largest=%u",
             reason, static_cast<unsigned>(bytes), parts, static_cast<long long>(elapsed_ms), rssi,
             static_cast<int>(power_save),
             static_cast<long long>(s_last_text_us ? (esp_timer_get_time()-s_last_text_us)/1000 : -1),
             static_cast<long long>(s_last_pong_us ? (esp_timer_get_time()-s_last_pong_us)/1000 : -1),
             static_cast<unsigned>(heap_caps_get_free_size(MALLOC_CAP_INTERNAL)),
             static_cast<unsigned>(heap_caps_get_largest_free_block(MALLOC_CAP_DMA)));
}

static double json_number(const cJSON *root, const char *name, double fallback = 0)
{
    const cJSON *value = cJSON_GetObjectItemCaseSensitive(root, name);
    return cJSON_IsNumber(value) ? value->valuedouble : fallback;
}

static std::string json_string(cJSON *root)
{
    char *text = cJSON_PrintUnformatted(root);
    std::string result = text == nullptr ? "{}" : text;
    cJSON_free(text);
    cJSON_Delete(root);
    return result;
}

static bool send_json(cJSON *root, uint32_t expected_generation = 0)
{
    std::string text = json_string(root);
    if (!s_connected || s_client == nullptr || (expected_generation &&
        !cloud_operation_current(expected_generation, s_generation.load(), s_ready.load()))) {
        ESP_LOGW(TAG, "[CLOUD] not connected; dropped event");
        return false;
    }
    const int64_t started = esp_timer_get_time();
    int result = esp_websocket_client_send_text(s_client, text.c_str(), text.size(), pdMS_TO_TICKS(VOICE_SEND_TIMEOUT_MS));
    if (result != static_cast<int>(text.size()))
        log_link_health("send-failed", text.size(), 0, (esp_timer_get_time() - started) / 1000);
    return result == static_cast<int>(text.size()) && (!expected_generation ||
        cloud_operation_current(expected_generation, s_generation.load(), s_ready.load()));
}

static void send_pending_button()
{
    UiButtonOutbox::Event event{};
    if(!s_button_outbox.pop_current(event,s_generation.load(),s_ready.load(),esp_timer_get_time()/1000))return;
    auto *root=cJSON_CreateObject();
    cJSON_AddStringToObject(root,"type","button_event");
    const char button[]={event.button,0};
    cJSON_AddStringToObject(root,"button",button);
    cJSON_AddStringToObject(root,"action","click");
    cJSON_AddNumberToObject(root,"timestamp",event.timestamp_ms);
    const auto started=esp_timer_get_time();
    const bool sent=send_json(root,event.generation);
    ESP_LOGI(TAG,"[CLOUD] button_event button=%s sent=%d elapsed_ms=%lld age_ms=%llu",button,sent,
        static_cast<long long>((esp_timer_get_time()-started)/1000),
        static_cast<unsigned long long>(esp_timer_get_time()/1000-event.timestamp_ms));
}

static void send_hello(void)
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddStringToObject(root, "type", "hello");
    cJSON_AddStringToObject(root, "device_id", s_device_id);
    cJSON_AddStringToObject(root, "device_token", CONFIG_DEVICE_LAB_DEVICE_TOKEN);
    cJSON_AddStringToObject(root, "board", board_slug());
    cJSON_AddStringToObject(root, "firmware", DEVICE_LAB_FIRMWARE_VERSION);
    const char *capabilities[]={"agent_target","approval","voice_input","settings","catalog","target-select","permissions","native-interaction","session-create","history","view-pagination","speech-settings","speech_playback","speech-buffer-v1"};
    cJSON_AddItemToObject(root,"capabilities",cJSON_CreateStringArray(capabilities,
        sizeof(capabilities)/sizeof(capabilities[0])));
    if (send_json(root)) {
        ESP_LOGI(TAG, "[CLOUD] sent hello device_id=%s", s_device_id);
    }
}

static void handle_message(const char *payload, size_t length)
{
    cJSON *root = cJSON_ParseWithLength(payload, length);
    if (root == nullptr) {
        ESP_LOGW(TAG, "[CLOUD] ignored invalid JSON");
        return;
    }
    const cJSON *type = cJSON_GetObjectItemCaseSensitive(root, "type");
    if (cJSON_IsString(type) && strcmp(type->valuestring, "hello_ack") == 0) {
        ESP_LOGI(TAG, "[CLOUD] hello acknowledged");
        board_control_message(payload);
        const auto *server_time=cJSON_GetObjectItemCaseSensitive(root,"server_time_ms");
        const auto *timezone_offset=cJSON_GetObjectItemCaseSensitive(root,"timezone_offset_minutes");
        if(cJSON_IsNumber(server_time)) board_sync_utc_time(server_time->valuedouble,
            cJSON_IsNumber(timezone_offset)?int(timezone_offset->valuedouble):0);
        s_link.hello = true;
        mark_synchronized();
    } else if (cJSON_IsString(type) && strcmp(type->valuestring, "approval_request") == 0) {
        board_control_message(payload);
    } else if (cJSON_IsString(type) && strcmp(type->valuestring, "approval_cancelled") == 0) {
        const auto *id = cJSON_GetObjectItemCaseSensitive(root, "approval_id");
        const auto *state = cJSON_GetObjectItemCaseSensitive(root, "state");
        if (cJSON_IsString(id) && cJSON_IsString(state)) board_approval_cancelled(id->valuestring, state->valuestring);
    } else if (cJSON_IsString(type) && strcmp(type->valuestring, "server_snapshot") == 0) {
        const cJSON *hostname = cJSON_GetObjectItemCaseSensitive(root, "hostname");
        board_server_snapshot_t snapshot = {};
        snapshot.hostname = cJSON_IsString(hostname) ? hostname->valuestring : "lighthouse";
        snapshot.cpu_percent = json_number(root, "cpu_percent");
        snapshot.load_1m = json_number(root, "load_1m");
        snapshot.load_5m = json_number(root, "load_5m");
        snapshot.memory_used = json_number(root, "memory_used");
        snapshot.memory_total = json_number(root, "memory_total");
        snapshot.memory_percent = json_number(root, "memory_percent");
        snapshot.swap_percent = json_number(root, "swap_percent");
        snapshot.disk_used = json_number(root, "disk_used");
        snapshot.disk_total = json_number(root, "disk_total");
        snapshot.disk_percent = json_number(root, "disk_percent");
        snapshot.network_rx_bytes = json_number(root, "network_rx_bytes");
        snapshot.network_tx_bytes = json_number(root, "network_tx_bytes");
        snapshot.network_rx_bps = json_number(root, "network_rx_bps");
        snapshot.network_tx_bps = json_number(root, "network_tx_bps");
        snapshot.uptime = json_number(root, "uptime");
        const auto *host_state = cJSON_GetObjectItemCaseSensitive(root, "host_state");
        snapshot.host_state = cJSON_IsString(host_state) ? host_state->valuestring : "";
        const auto *quota_valid = cJSON_GetObjectItemCaseSensitive(root, "codex_quota_valid");
        snapshot.codex_quota_valid = cJSON_IsBool(quota_valid) && cJSON_IsTrue(quota_valid);
        snapshot.codex_remaining_percent = uint8_t(json_number(root, "codex_remaining_percent"));
        snapshot.codex_resets_at = int64_t(json_number(root, "codex_resets_at"));
        snapshot.codex_window_minutes = int32_t(json_number(root, "codex_window_minutes"));
        board_set_server_snapshot(&snapshot);
        ESP_LOGD(TAG, "[CLOUD] server_snapshot host=%s cpu=%.1f", snapshot.hostname,
                 snapshot.cpu_percent);
    } else if (cJSON_IsString(type) && (strcmp(type->valuestring, "target_state") == 0 || strcmp(type->valuestring, "agent_state") == 0)) {
        if(!board_accept_target(payload)) {cJSON_Delete(root);return;}
        const cJSON *target = cJSON_GetObjectItemCaseSensitive(root, "target");
        if (cJSON_IsString(target) && strcmp(target->valuestring, "server") == 0) {
            board_select_server_target();
            ESP_LOGI(TAG, "[CLOUD] target server");
        } else {
            const cJSON *session_id = cJSON_GetObjectItemCaseSensitive(root, "session_id");
            const cJSON *name = cJSON_GetObjectItemCaseSensitive(root, "name");
            const cJSON *cwd = cJSON_GetObjectItemCaseSensitive(root, "cwd");
            const cJSON *state = cJSON_GetObjectItemCaseSensitive(root, "state");
            const cJSON *activity_type = cJSON_GetObjectItemCaseSensitive(root, "activity_type");
            const cJSON *latest_activity = cJSON_GetObjectItemCaseSensitive(root, "latest_activity");
            const cJSON *recent_text = cJSON_GetObjectItemCaseSensitive(root, "recent_text");
            const cJSON *final_text = cJSON_GetObjectItemCaseSensitive(root, "final_text");
            const cJSON *notification = cJSON_GetObjectItemCaseSensitive(root, "notification");
            board_pi_snapshot_t snapshot = {};
            const auto *agent=cJSON_GetObjectItemCaseSensitive(root,"agent");
            const auto *model=cJSON_GetObjectItemCaseSensitive(root,"model");
            const auto *effort=cJSON_GetObjectItemCaseSensitive(root,"effort");
            snapshot.agent=cJSON_IsString(agent)?agent->valuestring:"pi";
            snapshot.model=cJSON_IsString(model)?model->valuestring:"";
            snapshot.effort=cJSON_IsString(effort)?effort->valuestring:"";
            snapshot.session_id = cJSON_IsString(session_id) ? session_id->valuestring : "unknown";
            snapshot.name = cJSON_IsString(name) ? name->valuestring : snapshot.session_id;
            snapshot.cwd = cJSON_IsString(cwd) ? cwd->valuestring : "";
            snapshot.state = cJSON_IsString(state) ? state->valuestring : "OFFLINE";
            snapshot.activity_type = cJSON_IsString(activity_type) ? activity_type->valuestring : "";
            snapshot.latest_activity = cJSON_IsString(latest_activity) ? latest_activity->valuestring : "";
            snapshot.recent_text = cJSON_IsString(recent_text) ? recent_text->valuestring : "";
            snapshot.final_text = cJSON_IsString(final_text) ? final_text->valuestring : nullptr;
            snapshot.elapsed_seconds = json_number(root, "elapsed_seconds");
            board_set_pi_snapshot(&snapshot);
            if (cJSON_IsString(notification)) {
                if (strcmp(notification->valuestring, "done") == 0) {
                    board_notify_pi(false);
                } else if (strcmp(notification->valuestring, "failed") == 0) {
                    board_notify_pi(true);
                }
            }
            ESP_LOGD(TAG, "[CLOUD] Agent state=%s", snapshot.state);
        }
        s_link.target = true;
        mark_synchronized();
    } else if (cJSON_IsString(type) && (strcmp(type->valuestring, "pi_activity") == 0 || strcmp(type->valuestring, "agent_activity") == 0)) {
        const cJSON *activity_type = cJSON_GetObjectItemCaseSensitive(root, "activity_type");
        const cJSON *activity = cJSON_GetObjectItemCaseSensitive(root, "activity");
        if (cJSON_IsString(activity)) {
            const auto *agent=cJSON_GetObjectItemCaseSensitive(root,"agent");
            const auto *sid=cJSON_GetObjectItemCaseSensitive(root,"session_id");
            const auto *recent=cJSON_GetObjectItemCaseSensitive(root,"recent_text");
            if (cJSON_IsString(sid)) board_agent_activity(cJSON_IsString(agent)?agent->valuestring:"pi",sid->valuestring,
                cJSON_IsString(activity_type)?activity_type->valuestring:"text",activity->valuestring,
                cJSON_IsString(recent)?recent->valuestring:nullptr);
        }
    } else if (cJSON_IsString(type) && (strncmp(type->valuestring,"voice_",6)==0 || strncmp(type->valuestring,"settings_",9)==0 || strncmp(type->valuestring,"speech_",7)==0 || strncmp(type->valuestring,"catalog_",8)==0 || strncmp(type->valuestring,"target_",7)==0 || strncmp(type->valuestring,"user_input_",11)==0 || strncmp(type->valuestring,"session_",8)==0 || strncmp(type->valuestring,"creation_",9)==0 || strncmp(type->valuestring,"history_",8)==0 || strcmp(type->valuestring,"navigator_error")==0)) {
        // Acknowledge handoff happens before board dispatch so a resumed upload
        // is unblocked even if the UI ignores the message.
        const auto *end_state=cJSON_GetObjectItemCaseSensitive(root,"state");
        if(wearable::voice_terminal_ack(type->valuestring,cJSON_IsString(end_state)?end_state->valuestring:nullptr)) {
            const auto *id=cJSON_GetObjectItemCaseSensitive(root,"voice_id");
            if(cJSON_IsString(id)) {
                s_voice_end_ack_generation.store(s_generation.load());
                s_voice_end_ack_id.store(strtoull(id->valuestring,nullptr,16));
            }
        }
        if (strcmp(type->valuestring, "voice_ack") == 0) {
            char voice_id[17] = {};
            const auto *ack_id = cJSON_GetObjectItemCaseSensitive(root, "voice_id");
            if (cJSON_IsString(ack_id)) snprintf(voice_id, sizeof(voice_id), "%s", ack_id->valuestring);
            const auto *next = cJSON_GetObjectItemCaseSensitive(root, "next_sequence");
            if (voice_id[0]) publish_voice_ack(strtoull(voice_id, nullptr, 16), cJSON_IsNumber(next) ? unsigned(cJSON_GetNumberValue(next)) : 0U, s_generation.load());
        }
        char *wire=cJSON_PrintUnformatted(root);
        if (wire) {board_control_message(wire);cJSON_free(wire);}
    } else if (cJSON_IsString(type) && strcmp(type->valuestring, "notification_test") == 0) {
        const cJSON *kind = cJSON_GetObjectItemCaseSensitive(root, "kind");
        if (cJSON_IsString(kind) && strcmp(kind->valuestring, "failed") == 0) {
            board_notify_pi(true);
        } else if (cJSON_IsString(kind) && strcmp(kind->valuestring, "done") == 0) {
            board_notify_pi(false);
        }
    } else if (cJSON_IsString(type) && strcmp(type->valuestring, "display_message") == 0) {
        const cJSON *title = cJSON_GetObjectItemCaseSensitive(root, "title");
        const cJSON *body = cJSON_GetObjectItemCaseSensitive(root, "body");
        const char *title_text = cJSON_IsString(title) ? title->valuestring : "Message";
        const char *body_text = cJSON_IsString(body) ? body->valuestring : "";
        board_show_message(title_text, body_text);
        board_vibrate(80);
        ESP_LOGI(TAG, "[CLOUD] displayed display_message");
    }
    cJSON_Delete(root);
}

static void websocket_event(void *, esp_event_base_t, int32_t event_id, void *event_data)
{
    auto *event = static_cast<esp_websocket_event_data_t *>(event_data);
    switch (event_id) {
    case WEBSOCKET_EVENT_BEFORE_CONNECT:
        s_link.connecting();
        log_connection("connecting", event);
        break;
    case WEBSOCKET_EVENT_CONNECTED:
        s_text_receiver.reset();
        s_last_text_us = s_last_pong_us = 0;
        s_ready = false;
        {
            const auto recovery_ms = s_link.connected(esp_timer_get_time()/1000);
            s_generation = s_link.generation;
            log_connection("connected", event, recovery_ms);
        }
        s_connected = true;
        ESP_LOGI(TAG, "[CLOUD] connected");
        log_link_health("connected", 0, 0, 0);
        // A recording interrupted by this disconnect is resumed by asking the
        // gateway where to continue; the retained PSRAM buffer is never resent
        // from the start.
        if (s_input_sender_task) xTaskNotifyGive(s_input_sender_task);
        send_hello();
        break;
    case WEBSOCKET_EVENT_DISCONNECTED:
    case WEBSOCKET_EVENT_CLOSED:
        s_text_receiver.reset();
        s_ready = false;
        s_connected = false;
        log_connection(event_id == WEBSOCKET_EVENT_CLOSED ? "closed" : "disconnected", event);
        if (s_link.disconnected(esp_timer_get_time()/1000)) board_set_cloud_status(false);
        ESP_LOGW(TAG, "[CLOUD] disconnected");
        ESP_LOGI(TAG, "[CLOUD] reconnecting in %d ms", CLOUD_RECONNECT_TIMEOUT_MS);
        break;
    case WEBSOCKET_EVENT_DATA: {
        if (event->op_code == 0xA) {
            s_last_pong_us = esp_timer_get_time();
            log_link_health("pong", event->data_len, 0, 0);
        }
        if (event->op_code == 0x1 && event->payload_offset == 0) {
            s_rx_started_us = esp_timer_get_time();
            s_rx_parts = 0;
        }
        if (event->op_code == 0x1 || event->op_code == 0x0) ++s_rx_parts;
        const auto result = s_text_receiver.receive(event->op_code, event->fin, event->data_ptr,
            event->data_len, event->payload_len, event->payload_offset);
        if (result == WebSocketTextReceiver::Result::COMPLETE) {
            const int64_t now = esp_timer_get_time();
            s_last_text_us = now;
            const int64_t elapsed_ms = (now - s_rx_started_us) / 1000;
            // Trace a prolonged receive at most once per five seconds, never its payload.
            if (elapsed_ms > 250 && now - s_last_rx_trace_us > 5000000) {
                s_last_rx_trace_us = now;
                log_link_health("slow-text", s_text_receiver.size(), s_rx_parts, elapsed_ms);
            }
            handle_message(s_text_receiver.data(), s_text_receiver.size());
        } else if (result == WebSocketTextReceiver::Result::REJECTED)
            ESP_LOGW(TAG, "[CLOUD] rejected text frame opcode=%u offset=%d bytes=%d total=%d",
                     event->op_code, event->payload_offset, event->data_len, event->payload_len);
        break;
    }
    case WEBSOCKET_EVENT_ERROR:
        log_connection("error", event);
        log_link_health("receive-error", s_text_receiver.size(), s_rx_parts,
                        (esp_timer_get_time() - s_rx_started_us) / 1000);
        s_text_receiver.reset();
        ESP_LOGE(TAG, "[CLOUD] WebSocket error");
        break;
    default:
        break;
    }
}

static void input_sender_task(void *)
{
    auto *encoded = new unsigned char[4705];
    auto *control = new char[4096];
    constexpr unsigned kChunkBytes = AUDIO_CAPTURE_BYTES;
    uint64_t active=0;uint64_t last_dropped=0;unsigned sequence=0;size_t uploaded_bytes=0;size_t total_bytes=0;
    uint32_t voice_generation=0;
    int64_t max_send_ms=0;bool reported_stall=false;
    bool ended=false;
    bool handshake=false, need_resume=false;
    unsigned sent_limit=0;
    int64_t handshake_ms=0;
    bool failed_reported=false;
    int64_t upload_started_ms=0, last_progress_ms=0, paused_since_ms=0;
    int64_t retained_since_ms=0;
    bool retry_paused=false;
    wearable::VoiceDelivery delivery;
    wearable::VoiceCleanup cleanup;
    auto cancel_remote=[&](uint64_t id) {cleanup.add(id);};
    auto rearm_upload=[&](int64_t now){
        upload_started_ms=now;retry_paused=false;failed_reported=false;
        handshake=false;need_resume=true;s_voice_ack_pending.store(false);
    };
    int last_reported_percent=0;
    auto reset_voice=[&](){
        delivery.clear();s_voice_end_ack_id.store(0);
        active=0;sequence=0;uploaded_bytes=0;total_bytes=0;ended=false;
        upload_started_ms=0;last_progress_ms=0;paused_since_ms=0;last_reported_percent=0;
        failed_reported=false;handshake=false;need_resume=false;sent_limit=0;
        retained_since_ms=0;retry_paused=false;
        s_voice_ack_pending.store(false);
    };
    // Sends one chunk from the retained PSRAM recording at the current offset
    // and waits for its acknowledgement. Returns 1 progressed, 0 recoverable
    // block, -1 the request ended and must stop.
    auto upload_until_ack=[&](uint64_t id,const char *sid,bool resume_mode)->int {
        size_t recorded=0;
        const uint8_t *pcm=audio_backend_capture_pcm(&recorded);
        if(!board_voice_active(id)) return -1;
        if(need_resume || !cloud_operation_current(voice_generation,s_generation.load(),s_ready.load())) return 0;
        while (wearable::voice_chunk_ready(sequence,recorded,kChunkBytes,ended)) {
            const int64_t now=esp_timer_get_time()/1000;
            if(delivery.expired(now,retained_since_ms,VOICE_RETAIN_AFTER_FAILURE_MS) ||
               now-upload_started_ms>=int64_t(VOICE_UPLOAD_MAX_RETRY_MS)) return 0;
            const size_t offset=size_t(sequence)*kChunkBytes;
            const size_t size=std::min<size_t>(kChunkBytes,recorded-offset);
            auto *chunk=cJSON_CreateObject();
            cJSON_AddStringToObject(chunk,"type","voice_chunk");
            cJSON_AddStringToObject(chunk,"voice_id",sid);
            cJSON_AddNumberToObject(chunk,"sequence",sequence);
            size_t count=0;
            if(mbedtls_base64_encode(encoded,4705,&count,pcm+offset,size)!=0){cJSON_Delete(chunk);return -1;}
            encoded[count]=0;
            cJSON_AddStringToObject(chunk,"audio",reinterpret_cast<char *>(encoded));
            const int64_t send_started=esp_timer_get_time();
            sent_limit=std::max(sent_limit,sequence+1);
            const bool sent=send_json(chunk,voice_generation);
            const int64_t send_ms=(esp_timer_get_time()-send_started)/1000;
            if(send_ms>max_send_ms)max_send_ms=send_ms;
            if(send_ms>250&&!reported_stall){
                ESP_LOGI(TAG,"[VOICE] tx-stall type=voice_chunk id=%s elapsed_ms=%lld internal_free=%u dma_largest=%u",sid,static_cast<long long>(send_ms),
                         static_cast<unsigned>(heap_caps_get_free_size(MALLOC_CAP_INTERNAL)),
                         static_cast<unsigned>(heap_caps_get_largest_free_block(MALLOC_CAP_DMA)));
                reported_stall=true;
            }
            // send_json 已消费 JSON；成功或失败均不可再次释放。
            if(!sent || !cloud_operation_current(voice_generation,s_generation.load(),s_ready.load())){
                return 0;
            }
            // Bounded wait for the acknowledgement of exactly this chunk.
            const int64_t wait_deadline=esp_timer_get_time()/1000+VOICE_SEND_TIMEOUT_MS;
            while(true){
                if(s_voice_ack_pending.exchange(false)){
                    // Only an ack for this recording, from this capture generation,
                    // may move the uploader. A stale ack from a superseded capture
                    // is discarded rather than allowed to skip PCM.
                    if(s_voice_ack_id.load()==id && s_voice_ack_generation.load()==voice_generation){
                        const unsigned acked=s_voice_ack_next.load();
                        if(accept_ack(acked,sequence,resume_mode)){
                            sequence=acked;
                            uploaded_bytes=std::min<size_t>(size_t(sequence)*kChunkBytes,recorded);
                            last_progress_ms=esp_timer_get_time()/1000;paused_since_ms=0;
                            return 1;
                        }
                        ESP_LOGW(TAG,"[VOICE] ack-rejected id=%s acked=%u sent=%u resume=%d",sid,acked,sequence,resume_mode?1:0);
                    } else if(s_voice_ack_id.load()==id) {
                        ESP_LOGW(TAG,"[VOICE] ack-stale-generation id=%s ack_gen=%lu current=%lu",sid,
                                 static_cast<unsigned long>(s_voice_ack_generation.load()),
                                 static_cast<unsigned long>(voice_generation));
                    }
                }
                if(!cloud_operation_current(voice_generation,s_generation.load(),s_ready.load())) return 0;
                if(!board_voice_active(id)) return -1;
                const int64_t now=esp_timer_get_time()/1000;
                if(now>=wait_deadline || delivery.expired(now,retained_since_ms,VOICE_RETAIN_AFTER_FAILURE_MS)) return 0;
                ulTaskNotifyTake(pdTRUE,pdMS_TO_TICKS(20));
            }
            (void)offset;
        }
        return 2;
    };
    while (true) {
        while (true) {
            const auto generation = s_generation.load();
            if (!board_take_control_message(control,4096)) break;
            auto *root=cJSON_Parse(control);
            // 仅当前录音的明确用户重试可重开窗口；普通 ACK 不可续期。
            const auto *kind=cJSON_GetObjectItemCaseSensitive(root,"type");
            const auto *decision=cJSON_GetObjectItemCaseSensitive(root,"decision");
            const auto *voice_id=cJSON_GetObjectItemCaseSensitive(root,"voice_id");
            if(cJSON_IsString(kind) && cJSON_IsString(voice_id) &&
               (strcmp(kind->valuestring,"voice_cancel")==0 ||
                (strcmp(kind->valuestring,"voice_decision")==0 && cJSON_IsString(decision) && strcmp(decision->valuestring,"cancel")==0))) {
                const uint64_t id=strtoull(voice_id->valuestring,nullptr,16);
                cancel_remote(id);
                if(id==active){audio_backend_capture_release(id);reset_voice();}
                cJSON_Delete(root);continue;
            }
            if(cJSON_IsString(kind) && strcmp(kind->valuestring,"voice_decision")==0 &&
               cJSON_IsString(decision) && strcmp(decision->valuestring,"retry")==0 &&
               (!active || delivery.remote_pending || delivery.expired(esp_timer_get_time()/1000,retained_since_ms,VOICE_RETAIN_AFTER_FAILURE_MS))) {
                cJSON_Delete(root);continue;
            }
            if(active && cJSON_IsString(kind) && strcmp(kind->valuestring,"voice_decision")==0 &&
               cJSON_IsString(decision) && strcmp(decision->valuestring,"retry")==0 &&
               cJSON_IsString(voice_id) && strtoull(voice_id->valuestring,nullptr,16)==active &&
               esp_timer_get_time()/1000-retained_since_ms<int64_t(VOICE_RETAIN_AFTER_FAILURE_MS)){
                rearm_upload(esp_timer_get_time()/1000);
            }
            if (root && !send_json(root, generation)) {
                auto *failed=cJSON_Parse(control);
                const auto *kind=cJSON_GetObjectItemCaseSensitive(failed,"type");
                if(cJSON_IsString(kind) && input_sender_best_effort_control(kind->valuestring)) {
                    ESP_LOGW(TAG,"[SPEECH] control-send-failed type=%s generation=%lu",kind->valuestring,static_cast<unsigned long>(generation));
                    cJSON_Delete(failed);continue;
                }
                const bool speech=cJSON_IsString(kind) && strncmp(kind->valuestring,"speech_",7)==0;
                const bool setting=cJSON_IsString(kind) && strncmp(kind->valuestring,"settings_",9)==0;
                cJSON_ReplaceItemInObject(failed,"type",cJSON_CreateString(speech?"speech_error":setting?"settings_error":cJSON_IsString(kind) && strncmp(kind->valuestring,"voice_",6)==0?"voice_error":"navigator_error"));
                cJSON_AddStringToObject(failed,"message",speech || setting ? "Connection failed; try again" : "Voice interrupted; try again");
                char *error=cJSON_PrintUnformatted(failed);
                if (error) {board_control_message(error);cJSON_free(error);}
                cJSON_Delete(failed);
            }
        }
        // Capture control frames arrive from the microphone task. PCM never
        // flows through here, so a stalled upload cannot truncate a recording.
        AudioCaptureFrame frame{};
        if (audio_backend_take_capture(&frame)) {
            char id[17];snprintf(id,sizeof(id),"%016llx",static_cast<unsigned long long>(frame.id));
            if(frame.kind==CaptureKind::START) voice_generation=s_generation.load();
            if(!board_voice_active(frame.id)){
                if(last_dropped!=frame.id){
                    ESP_LOGI(TAG,"[VOICE] tx-drop id=%s reason=%s kind=%d",id,s_connected?"inactive-modal":"cloud-disconnect",static_cast<int>(frame.kind));
                    last_dropped=frame.id;
                }
                if(board_voice_active(frame.id))board_voice_interrupted(frame.id);
                if(active==frame.id){cancel_remote(active);audio_backend_capture_release(active);reset_voice();}
                continue;
            }
            if(frame.kind==CaptureKind::START){
                if(active){cancel_remote(active);audio_backend_capture_release(active);}
                reset_voice();active=frame.id;delivery.id=active;
                // A fresh capture invalidates any ack still in flight for the
                // previous one; its sequence space does not apply here.
                s_voice_ack_pending.store(false);
                auto *root=cJSON_CreateObject();
                cJSON_AddStringToObject(root,"type","voice_start");cJSON_AddStringToObject(root,"voice_id",id);
                cJSON_AddNumberToObject(root,"sample_rate",44100);cJSON_AddNumberToObject(root,"bits",16);cJSON_AddNumberToObject(root,"channels",1);
                cJSON_AddNumberToObject(root,"max_recording_ms",VOICE_MAX_RECORDING_SECONDS*1000);
                char context[2048];
                if(board_voice_context(frame.id,context,sizeof(context))){auto* extra=cJSON_Parse(context);if(extra)cJSON_AddItemToObject(root,"context",extra);}
                const bool sent=send_json(root,voice_generation);
                if(!sent)need_resume=true;
                upload_started_ms=esp_timer_get_time()/1000;last_progress_ms=upload_started_ms;
                retained_since_ms=upload_started_ms;
                ESP_LOGI(TAG,"[VOICE] tx type=voice_start id=%s sent=%d",id,sent);
                continue;
            }
            if(frame.id!=active){continue;}
            if(frame.kind==CaptureKind::CANCEL){
                auto *root=cJSON_CreateObject();cJSON_AddStringToObject(root,"type","voice_cancel");cJSON_AddStringToObject(root,"voice_id",id);
                if(!send_json(root,voice_generation))cancel_remote(frame.id);
                audio_backend_capture_release(frame.id);
                ESP_LOGI(TAG,"[VOICE] tx type=voice_cancel id=%s",id);
                board_capture_finished(frame.id,true);reset_voice();continue;
            }
            total_bytes=frame.size;ended=true;
            last_progress_ms=esp_timer_get_time()/1000;
            ESP_LOGI(TAG,"[VOICE] capture-complete id=%s bytes=%u",id,static_cast<unsigned>(total_bytes));
            continue;
        }
        if(const uint64_t cleanup_id=cleanup.next(s_ready.load())) {
            char id[17];snprintf(id,sizeof(id),"%016llx",static_cast<unsigned long long>(cleanup_id));
            auto *root=cJSON_CreateObject();cJSON_AddStringToObject(root,"type","voice_cancel");cJSON_AddStringToObject(root,"voice_id",id);
            cleanup.sent(cleanup_id,send_json(root,s_generation.load()));
        }
        if(!active){send_pending_button();vTaskDelay(input_sender_idle_ticks(pdMS_TO_TICKS(5)));continue;}
        char sid[17];snprintf(sid,sizeof(sid),"%016llx",static_cast<unsigned long long>(active));
        const int64_t upload_now=esp_timer_get_time()/1000;
        if(!board_voice_active(active)) {
            cancel_remote(active);audio_backend_capture_release(active);reset_voice();continue;
        }
        // 硬截止优先于迟到 ACK；重试和重连不得复活到期请求。
        if(delivery.expired(upload_now,retained_since_ms,VOICE_RETAIN_AFTER_FAILURE_MS)){
            ESP_LOGW(TAG,"[VOICE] upload-expired id=%s uploaded=%u total=%u",sid,
                     static_cast<unsigned>(uploaded_bytes),static_cast<unsigned>(total_bytes));
            cancel_remote(active);audio_backend_capture_release(active);
            board_voice_upload_expired(active);
            reset_voice();continue;
        }
        const auto terminal_id=s_voice_end_ack_id.exchange(0);
        if(ended && s_ready.load() && voice_generation==s_generation.load() && delivery.acknowledge(terminal_id,s_voice_end_ack_generation.load(),voice_generation)) {
            audio_backend_capture_release(active);board_capture_finished(active,false);
        }
        if(delivery.remote_pending) {
            if(s_ready.load() && (delivery.needs_status(s_generation.load(),true) || upload_now-handshake_ms>=VOICE_SEND_TIMEOUT_MS)) {
                auto *status=cJSON_CreateObject();cJSON_AddStringToObject(status,"type","voice_status");cJSON_AddStringToObject(status,"voice_id",sid);
                if(send_json(status,s_generation.load())){delivery.status_generation=s_generation.load();handshake_ms=upload_now;}
            }
            ulTaskNotifyTake(pdTRUE,pdMS_TO_TICKS(100));continue;
        }
        if(s_ready.load() && voice_generation!=s_generation.load()){
            rearm_upload(upload_now);voice_generation=s_generation.load();
        }
        if(!retry_paused && upload_now-upload_started_ms>=int64_t(VOICE_UPLOAD_MAX_RETRY_MS)){
            retry_paused=true;need_resume=true;handshake=false;
            s_voice_ack_pending.store(false);
            ESP_LOGW(TAG,"[VOICE] upload-failed id=%s reason=retry-window-expired",sid);
            board_voice_upload_failed(active);failed_reported=true;
        }
        if(retry_paused){ulTaskNotifyTake(pdTRUE,pdMS_TO_TICKS(200));continue;}
        const bool resume_mode=false;
        const int result=upload_until_ack(active,sid,resume_mode);
        if(result==-1){cancel_remote(active);audio_backend_capture_release(active);reset_voice();continue;}
        if(result==0){
            if(delivery.expired(esp_timer_get_time()/1000,retained_since_ms,VOICE_RETAIN_AFTER_FAILURE_MS))continue;
            // Recoverable block: hold the recording, report paused, keep waiting
            // until the bounded retention window expires. Never a false success.
            const int64_t now=esp_timer_get_time()/1000;
            if(!paused_since_ms)paused_since_ms=now;
            board_voice_upload_progress(active,voice_upload_percent(uploaded_bytes,total_bytes),true);
            // After a reconnect, ask the gateway for the acknowledged offset so
            // the upload resumes at the breakpoint instead of restarting.
            need_resume=true;
            if(now-upload_started_ms>=int64_t(VOICE_UPLOAD_MAX_RETRY_MS)){
                // 下次循环统一关闭重试；截止后不得再发送 status 或接纳迟到 ACK。
                ulTaskNotifyTake(pdTRUE,pdMS_TO_TICKS(20));continue;
            }
            if(s_ready.load() && (!handshake || voice_generation!=s_generation.load() || now-handshake_ms>=VOICE_SEND_TIMEOUT_MS)){
                voice_generation=s_generation.load();
                s_voice_ack_pending.store(false);
                auto *status=cJSON_CreateObject();
                cJSON_AddStringToObject(status,"type","voice_status");
                cJSON_AddStringToObject(status,"voice_id",sid);
                if(send_json(status,voice_generation)){
                    ESP_LOGI(TAG,"[VOICE] tx type=voice_status id=%s resume_from=%u",sid,sequence);
                    handshake=true;handshake_ms=now;
                }
            }
            // voice_start 未送达时 status 可能永无 ACK：仅有界等待，
            // 不盲重发 start，以免覆盖网关已保存的上下文。
            if(s_voice_ack_pending.exchange(false)){
                if(s_voice_ack_id.load()==active && s_voice_ack_generation.load()==voice_generation){
                    const unsigned acked=s_voice_ack_next.load();
                    // Resume ack: the gateway may point anywhere inside the sent
                    // range, but never past it.
                    if(handshake && wearable::voice_ack_valid(acked,sequence,true,sent_limit)){
                        sequence=acked;
                        uploaded_bytes=std::min<size_t>(size_t(sequence)*kChunkBytes,total_bytes);
                        handshake=false;need_resume=false;
                        failed_reported=false;paused_since_ms=0;last_progress_ms=now;
                        // ACK 只恢复偏移，不延长本次重试窗口。
                        ESP_LOGI(TAG,"[VOICE] upload-resumed id=%s from_sequence=%u",sid,sequence);
                    }
                }
            }
            ulTaskNotifyTake(pdTRUE,pdMS_TO_TICKS(200));
            continue;
        }
        if(result==1){
            if(ended) board_voice_upload_progress(active,voice_upload_percent(uploaded_bytes,total_bytes),false);
            continue;
        }
        if(!ended){
            // Still recording: the UI stays in RECORDING. Progress is a property
            // of the transfer only, so it must not repaint the capture phase.
            if(voice_should_report_progress(ended) && uploaded_bytes>0)
                board_voice_upload_progress(active,voice_upload_percent(uploaded_bytes,total_bytes),false);
            ulTaskNotifyTake(pdTRUE,pdMS_TO_TICKS(20));
            continue;
        }
        if(delivery.expired(esp_timer_get_time()/1000,retained_since_ms,VOICE_RETAIN_AFTER_FAILURE_MS))continue;
        // PCM 全部确认后发送 END，但仍须远端终态确认才可关闭上传窗口。
        auto *root=cJSON_CreateObject();
        cJSON_AddStringToObject(root,"type","voice_end");cJSON_AddStringToObject(root,"voice_id",sid);
        const auto duration=audio_backend_record_duration_ms(active);
        cJSON_AddNumberToObject(root,"recorded_duration_ms",static_cast<double>(duration));
        cJSON_AddNumberToObject(root,"total_bytes",static_cast<double>(total_bytes));
        const bool sent=send_json(root,voice_generation);
        ESP_LOGI(TAG,"[VOICE] tx type=voice_end id=%s sent=%d chunks=%u pcm_bytes=%u max_send_ms=%lld",sid,sent,sequence,static_cast<unsigned>(uploaded_bytes),static_cast<long long>(max_send_ms));
        if(sent){handshake_ms=upload_now;need_resume=true;handshake=false;}
        else {need_resume=true;board_voice_upload_progress(active,99,true);}
        ulTaskNotifyTake(pdTRUE,pdMS_TO_TICKS(20));
    }
}

static void cloud_task(void *)
{
    xEventGroupWaitBits(wifi_manager_event_group(), WIFI_CONNECTED_BIT, pdFALSE, pdTRUE, portMAX_DELAY);

    esp_websocket_client_config_t config = {};
    config.uri = CONFIG_DEVICE_LAB_CLOUD_WS_URL;
    config.task_prio = 5; // Keep TLS work below the GPIO/UI polling owner.
    config.network_timeout_ms = 10000;
    config.reconnect_timeout_ms = CLOUD_RECONNECT_TIMEOUT_MS;
    config.ping_interval_sec = 10;
    config.pingpong_timeout_sec = 20;
    config.keep_alive_enable = true;
    config.keep_alive_idle = 30;
    config.keep_alive_interval = 10;
    config.keep_alive_count = 3;
    config.buffer_size = CLOUD_RX_BUFFER_BYTES;
    config.enable_close_reconnect = true;
    if (strncmp(config.uri, "wss://", 6) == 0) {
        config.crt_bundle_attach = esp_crt_bundle_attach;
    }

    ESP_LOGI(TAG, "[CLOUD_LINK] config ping_sec=10 pong_timeout_sec=20 tcp_idle_sec=30 tcp_interval_sec=10 tcp_count=3 reconnect_ms=%d network_timeout_ms=%d",
             CLOUD_RECONNECT_TIMEOUT_MS, config.network_timeout_ms);
    s_client = esp_websocket_client_init(&config);
    if (s_client == nullptr) {
        ESP_LOGE(TAG, "[CLOUD] client allocation failed");
        vTaskDelete(nullptr);
        return;
    }
    esp_websocket_register_events(s_client, WEBSOCKET_EVENT_ANY, websocket_event, nullptr);
    ESP_ERROR_CHECK(esp_websocket_client_start(s_client));

    while (true) {
        vTaskDelay(pdMS_TO_TICKS(CONFIG_DEVICE_LAB_HEARTBEAT_SECONDS * 1000));
        if (s_ready) {
            cJSON *root = cJSON_CreateObject();
            cJSON_AddStringToObject(root, "type", "heartbeat");
            cJSON_AddNumberToObject(root, "uptime", esp_timer_get_time() / 1000000);
            if (send_json(root)) {
                ESP_LOGI(TAG, "[CLOUD] sent heartbeat");
            }
        }
    }
}

esp_err_t cloud_client_start(void)
{
    auto *text_storage = static_cast<char *>(heap_caps_malloc(CLOUD_RX_BUFFER_BYTES + 1,
        MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT));
    if (!text_storage) return ESP_ERR_NO_MEM;
    s_text_receiver.set_storage(text_storage, CLOUD_RX_BUFFER_BYTES);
    uint8_t mac[6] = {};
    ESP_RETURN_ON_ERROR(esp_read_mac(mac, ESP_MAC_WIFI_STA), TAG, "read MAC");
    snprintf(s_device_id, sizeof(s_device_id), "esp32s3-%02x%02x%02x%02x%02x%02x",
             mac[0], mac[1], mac[2], mac[3], mac[4], mac[5]);
    board_set_device_id(s_device_id);
    if (xTaskCreate(input_sender_task,"input_sender",8192,nullptr,5,&s_input_sender_task)!=pdPASS) return ESP_ERR_NO_MEM;
    return xTaskCreate(cloud_task, "cloud", 8192, nullptr, 5, nullptr) == pdPASS
        ? ESP_OK
        : ESP_ERR_NO_MEM;
}

bool cloud_client_send_button(const char *button)
{
    if(!button || (button[0]!='A' && button[0]!='B') || button[1] || !s_ready)return false;
    return s_button_outbox.push(button[0],s_generation.load(),esp_timer_get_time()/1000);
}

bool cloud_client_send_approval(const char *approval_id, bool approve)
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddStringToObject(root, "type", "approval_response");
    cJSON_AddStringToObject(root, "approval_id", approval_id);
    cJSON_AddStringToObject(root, "decision", approve ? "approve" : "reject");
    return send_json(root, s_generation.load()); // No retry queue: stale decisions are never replayed.
}