#include "cloud_client.h"

#include <cstdio>
#include <cstring>
#include <string>
#include <atomic>
#include "audio_backend.h"
#include "input_sender_timing.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};
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 unsigned s_rx_parts;

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 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<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_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);
    // Speech mode preference selection is supported and independent of audio
    // playback. Playback stays FROZEN: no speech_playback capability is
    // advertised, so the Gateway never sends speech_start/chunk/end to this
    // firmware and a saved mode cannot re-enable playback.
    const char *capabilities[]={"agent_target","approval","voice_input","settings","catalog","target-select","permissions","native-interaction","session-create","speech-settings","history"};
    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)) {
        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_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);
        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 == 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();
            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 report_control_send_failure(const char *control)
{
    auto *failed=cJSON_Parse(control);
    if (!failed) return;
    const auto *kind=cJSON_GetObjectItemCaseSensitive(failed,"type");
    const bool speech=cJSON_IsString(kind) && strncmp(kind->valuestring,"speech_",7)==0;
    const bool history=cJSON_IsString(kind) && strncmp(kind->valuestring,"history_",8)==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":history?"history_error":cJSON_IsString(kind) && strncmp(kind->valuestring,"voice_",6)==0?"voice_error":"navigator_error"));
    cJSON_AddStringToObject(failed,"message",speech || setting || history ? "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);
}

static void input_sender_task(void *)
{
    auto *frame = new AudioCaptureFrame{};
    auto *encoded = new unsigned char[4705];
    auto *control = new char[4096];
    uint64_t active=0, last_dropped=0;unsigned sequence=0, pcm_bytes=0;
    uint32_t voice_generation=0;
    int64_t max_send_ms=0;bool reported_stall=false;
    while (true) {
        while (true) {
            const auto generation = s_generation.load();
            if (!board_take_control_message(control,4096)) break;
            auto *root=cJSON_Parse(control);
            if (root && !send_json(root, generation)) {
                report_control_send_failure(control);
            }
        }
        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 (!cloud_operation_current(voice_generation,s_generation.load(),s_ready.load()) || !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);
                active=0;continue;
            }
            auto *root=cJSON_CreateObject();
            cJSON_AddStringToObject(root,"voice_id",id);
            if (frame->kind==CaptureKind::START) {
                active=frame->id;sequence=0;pcm_bytes=0;max_send_ms=0;reported_stall=false;
                cJSON_AddStringToObject(root,"type","voice_start");
                cJSON_AddNumberToObject(root,"sample_rate",44100);
                cJSON_AddNumberToObject(root,"bits",16);cJSON_AddNumberToObject(root,"channels",1);
                char context[2048];
                if(board_voice_context(frame->id,context,sizeof(context))) {
                    auto* extra=cJSON_Parse(context);
                    if(extra)cJSON_AddItemToObject(root,"context",extra);
                }
            } else if (frame->kind==CaptureKind::PCM && frame->id==active) {
                size_t count=0;
                mbedtls_base64_encode(encoded,4705,&count,frame->pcm,frame->size);encoded[count]=0;
                cJSON_AddStringToObject(root,"type","voice_chunk");
                cJSON_AddNumberToObject(root,"sequence",sequence++);
                pcm_bytes+=frame->size;
                cJSON_AddStringToObject(root,"audio",reinterpret_cast<char *>(encoded));
            } else if (frame->kind==CaptureKind::END || frame->kind==CaptureKind::CANCEL) {
                cJSON_AddStringToObject(root,"type",frame->kind==CaptureKind::END?"voice_end":"voice_cancel");
                if(frame->kind==CaptureKind::END) {const auto started=audio_backend_record_started_ms(frame->id);cJSON_AddNumberToObject(root,"recorded_duration_ms",started?esp_timer_get_time()/1000-started:0);}
                board_capture_finished(frame->id,frame->kind==CaptureKind::CANCEL);active=0;
            } else {cJSON_Delete(root);continue;}
            const char *wire_type=frame->kind==CaptureKind::START?"voice_start":frame->kind==CaptureKind::PCM?"voice_chunk":frame->kind==CaptureKind::END?"voice_end":"voice_cancel";
            const int64_t send_started=esp_timer_get_time();
            const bool sent=send_json(root, 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=%s id=%s elapsed_ms=%lld internal_free=%u dma_largest=%u",wire_type,id,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;
            }
            if (frame->kind!=CaptureKind::PCM || sequence==1 || !sent)
                ESP_LOGI(TAG,"[VOICE] tx type=%s id=%s sent=%d chunks=%u pcm_bytes=%u max_send_ms=%lld",wire_type,id,sent,sequence,pcm_bytes,static_cast<long long>(max_send_ms));
            if (!sent) {
                board_voice_interrupted(frame->id);
                auto *cancel=cJSON_CreateObject();cJSON_AddStringToObject(cancel,"type","voice_cancel");
                cJSON_AddStringToObject(cancel,"voice_id",id);send_json(cancel,voice_generation);active=0;
            }
        } else vTaskDelay(input_sender_idle_ticks(pdMS_TO_TICKS(5)));
    }
}

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.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,nullptr)!=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)
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddStringToObject(root, "type", "button_event");
    cJSON_AddStringToObject(root, "button", button);
    cJSON_AddStringToObject(root, "action", "click");
    cJSON_AddNumberToObject(root, "timestamp", esp_timer_get_time() / 1000);
    bool sent = send_json(root, s_generation.load());
    ESP_LOGI(TAG, "[CLOUD] %s button_event button=%s", sent ? "sent" : "did not send", button);
    return sent;
}

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.
}