server/protocol.c

Go to the documentation of this file.

 
#include <math.h>
#include <stdarg.h>
#include <stdatomic.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
 
#include "protocol.h"
#include "client.h"
#include "common.h"
#include "util/endian.h"
#include "util/validation.h"
#include "util/endian.h"
#include "util/bytes.h"
#include "util/image.h"
#include "video/video_frame.h"
#include "audio/audio.h"
#include "video/palette.h"
#include "video/image.h"
#include "network/compression.h"
#include "network/packet_parsing.h"
#include "network/acip/send.h"
#include "network/acip/server.h"
#include "util/format.h"
#include "platform/system.h"
#include "audio/opus_codec.h"
#include "network/packet_parsing.h"
#include "network/logging.h"
#include "crypto/handshake/common.h"
 
extern atomic_bool g_server_should_exit;
 
static void protocol_cleanup_thread_locals(void) {
  // Placeholder for thread-local resources owned by the receive thread.
  // Add cleanup logic here if future protocol changes introduce
  // thread-local allocations that must be released before disconnecting.
}
 
void disconnect_client_for_bad_data(client_info_t *client, const char *format, ...) {
  if (!client) {
    return;
  }
 
  protocol_cleanup_thread_locals();
 
  bool already_requested = atomic_exchange(&client->protocol_disconnect_requested, true);
  if (already_requested) {
    return;
  }
 
  char reason[256] = {0};
  if (format) {
    va_list args;
    va_start(args, format);
    (void)vsnprintf(reason, sizeof(reason), format, args);
    va_end(args);
  } else {
    SAFE_STRNCPY(reason, "Protocol violation", sizeof(reason));
  }
 
  const char *reason_str = reason[0] != '\0' ? reason : "Protocol violation";
  uint32_t client_id = atomic_load(&client->client_id);
 
  socket_t socket_snapshot = INVALID_SOCKET_VALUE;
  const crypto_context_t *crypto_ctx = NULL;
 
  mutex_lock(&client->client_state_mutex);
  if (client->socket != INVALID_SOCKET_VALUE) {
    socket_snapshot = client->socket;
    if (client->crypto_initialized) {
      crypto_ctx = crypto_handshake_get_context(&client->crypto_handshake_ctx);
    }
  }
  mutex_unlock(&client->client_state_mutex);
 
  // NOTE: Disconnecting a client due to the client's own bad behavior isn't an
  // error for us, it's desired behavior for us, so we simply warn and do not
  // have a need for asciichat_errno here.
  log_warn("Disconnecting client %u due to protocol violation: %s", client_id, reason_str);
 
  if (socket_snapshot != INVALID_SOCKET_VALUE) {
    // CRITICAL: Protect socket writes with send_mutex to prevent race with send_thread
    // This receive_thread and send_thread both write to same socket
    mutex_lock(&client->send_mutex);
 
    asciichat_error_t log_result =
        log_network_message(socket_snapshot, (const struct crypto_context_t *)crypto_ctx, LOG_ERROR,
                            REMOTE_LOG_DIRECTION_SERVER_TO_CLIENT, "Protocol violation: %s", reason_str);
    if (log_result != ASCIICHAT_OK) {
      log_warn("Failed to send remote log to client %u: %s", client_id, asciichat_error_string(log_result));
    }
 
    asciichat_error_t send_result = packet_send_error(socket_snapshot, crypto_ctx, ERROR_NETWORK_PROTOCOL, reason_str);
    if (send_result != ASCIICHAT_OK) {
      log_warn("Failed to send error packet to client %u: %s", client_id, asciichat_error_string(send_result));
    }
 
    mutex_unlock(&client->send_mutex);
  }
 
  platform_sleep_ms(500);
 
  atomic_store(&client->active, false);
  atomic_store(&client->shutting_down, true);
  atomic_store(&client->send_thread_running, false);
  atomic_store(&client->video_render_thread_running, false);
  atomic_store(&client->audio_render_thread_running, false);
 
  if (client->audio_queue) {
    packet_queue_shutdown(client->audio_queue);
  }
 
  mutex_lock(&client->client_state_mutex);
  if (client->socket != INVALID_SOCKET_VALUE) {
    socket_shutdown(client->socket, 2);
    socket_close(client->socket);
    client->socket = INVALID_SOCKET_VALUE;
  }
  mutex_unlock(&client->client_state_mutex);
}
 
/* ============================================================================
 * Client Lifecycle Packet Handlers
 * ============================================================================
 */
 
void handle_client_join_packet(client_info_t *client, const void *data, size_t len) {
  VALIDATE_PACKET_SIZE(client, data, len, sizeof(client_info_packet_t), "CLIENT_JOIN");
 
  const client_info_packet_t *join_info = (const client_info_packet_t *)data;
 
  // Validate display name is present and not just whitespace
  if (join_info->display_name[0] == '\0') {
    disconnect_client_for_bad_data(client, "CLIENT_JOIN display_name cannot be empty");
    return;
  }
 
  uint32_t capabilities = NET_TO_HOST_U32(join_info->capabilities);
 
  // Validate at least one capability flag is set
  const uint32_t VALID_CAP_MASK = CLIENT_CAP_VIDEO | CLIENT_CAP_AUDIO | CLIENT_CAP_COLOR | CLIENT_CAP_STRETCH;
  VALIDATE_CAPABILITY_FLAGS(client, capabilities, VALID_CAP_MASK, "CLIENT_JOIN");
 
  // Validate no unknown capability bits are set
  VALIDATE_FLAGS_MASK(client, capabilities, VALID_CAP_MASK, "CLIENT_JOIN");
 
  SAFE_STRNCPY(client->display_name, join_info->display_name, MAX_DISPLAY_NAME_LEN - 1);
 
  client->can_send_video = (capabilities & CLIENT_CAP_VIDEO) != 0;
  client->can_send_audio = (capabilities & CLIENT_CAP_AUDIO) != 0;
  client->wants_stretch = (capabilities & CLIENT_CAP_STRETCH) != 0;
 
  log_info("Client %u joined: %s (video=%d, audio=%d, stretch=%d)", atomic_load(&client->client_id),
           client->display_name, client->can_send_video, client->can_send_audio, client->wants_stretch);
 
  // Notify client of successful join (encrypted channel)
  log_info_client(client, "Joined as '%s' (video=%s, audio=%s)", client->display_name,
                  client->can_send_video ? "yes" : "no", client->can_send_audio ? "yes" : "no");
}
 
void handle_protocol_version_packet(client_info_t *client, const void *data, size_t len) {
  if (!data) {
    disconnect_client_for_bad_data(client, "PROTOCOL_VERSION payload missing");
    return;
  }
 
  if (len != sizeof(protocol_version_packet_t)) {
    disconnect_client_for_bad_data(client, "PROTOCOL_VERSION invalid size: %zu (expected %zu)", len,
                                   sizeof(protocol_version_packet_t));
    return;
  }
 
  const protocol_version_packet_t *version = (const protocol_version_packet_t *)data;
  uint16_t client_major = NET_TO_HOST_U16(version->protocol_version);
  uint16_t client_minor = NET_TO_HOST_U16(version->protocol_revision);
 
  // Validate major version match (minor version can differ for backward compat)
  if (client_major != PROTOCOL_VERSION_MAJOR) {
    log_warn("Client %u protocol version mismatch: client=%u.%u, server=%u.%u", atomic_load(&client->client_id),
             client_major, client_minor, PROTOCOL_VERSION_MAJOR, PROTOCOL_VERSION_MINOR);
    // Note: We don't disconnect on version mismatch for backward compatibility
    // Clients may be older or newer than server
  } else if (client_minor != PROTOCOL_VERSION_MINOR) {
    log_info("Client %u has different protocol revision: client=%u.%u, server=%u.%u", atomic_load(&client->client_id),
             client_major, client_minor, PROTOCOL_VERSION_MAJOR, PROTOCOL_VERSION_MINOR);
  }
 
  // Validate reserved bytes are zero
  for (size_t i = 0; i < sizeof(version->reserved); i++) {
    if (version->reserved[i] != 0) {
      log_warn("Client %u sent non-zero reserved bytes in PROTOCOL_VERSION packet", atomic_load(&client->client_id));
      // Don't disconnect - reserved bytes may be used in future versions
      break;
    }
  }
 
  // Log supported features
  if (version->supports_encryption) {
    log_debug("Client %u supports encryption", atomic_load(&client->client_id));
  }
  if (version->compression_algorithms != 0) {
    log_debug("Client %u supports compression: 0x%02x", atomic_load(&client->client_id),
              version->compression_algorithms);
  }
  if (version->feature_flags != 0) {
    uint16_t feature_flags = NET_TO_HOST_U16(version->feature_flags);
    log_debug("Client %u supports features: 0x%04x", atomic_load(&client->client_id), feature_flags);
  }
}
 
void handle_client_leave_packet(client_info_t *client, const void *data, size_t len) {
  if (!client) {
    return;
  }
 
  uint32_t client_id = atomic_load(&client->client_id);
 
  if (len == 0) {
    // Empty reason - client disconnecting without explanation
    log_info("Client %u sent leave notification (no reason)", client_id);
  } else if (len <= 256) {
    // Reason provided - extract and log it
    if (!data) {
      SET_ERRNO(ERROR_INVALID_STATE, "Client %u sent leave notification with non-zero length but NULL data", client_id);
      return;
    }
 
    char reason[257] = {0};
    memcpy(reason, data, len);
    reason[len] = '\0';
 
    // Validate reason is printable (handle potential non-UTF8 gracefully)
    bool all_printable = true;
    for (size_t i = 0; i < len; i++) {
      uint8_t c = (uint8_t)reason[i];
      if (c < 32 && c != '\t' && c != '\n') {
        all_printable = false;
        break;
      }
    }
 
    if (all_printable) {
      log_info("Client %u sent leave notification: %s", client_id, reason);
    } else {
      log_info("Client %u sent leave notification (reason contains non-printable characters)", client_id);
    }
  } else {
    // Oversized reason - shouldn't happen with validation.h checks
    log_warn("Client %u sent oversized leave reason (%zu bytes, max 256)", client_id, len);
  }
 
  // Deactivate client to stop processing packets
  // Sets client->active = false immediately - triggers client cleanup procedures
  atomic_store(&client->active, false);
 
  // Note: We don't disconnect the client here - that happens when socket closes
  // This is just a clean notification before disconnect
}
 
void handle_stream_start_packet(client_info_t *client, const void *data, size_t len) {
  VALIDATE_PACKET_SIZE(client, data, len, sizeof(uint32_t), "STREAM_START");
 
  uint32_t stream_type_net;
  memcpy(&stream_type_net, data, sizeof(uint32_t));
  uint32_t stream_type = NET_TO_HOST_U32(stream_type_net);
 
  // Validate at least one stream type flag is set
  const uint32_t VALID_STREAM_MASK = STREAM_TYPE_VIDEO | STREAM_TYPE_AUDIO;
  VALIDATE_CAPABILITY_FLAGS(client, stream_type, VALID_STREAM_MASK, "STREAM_START");
 
  // Validate no unknown stream type bits are set
  VALIDATE_FLAGS_MASK(client, stream_type, VALID_STREAM_MASK, "STREAM_START");
 
  if (stream_type & STREAM_TYPE_VIDEO) {
    // Wait for first IMAGE_FRAME before marking sending_video true (avoids race)
  }
  if (stream_type & STREAM_TYPE_AUDIO) {
    atomic_store(&client->is_sending_audio, true);
 
    // Create Opus decoder for this client if not already created
    if (!client->opus_decoder) {
      client->opus_decoder = opus_codec_create_decoder(48000);
      if (client->opus_decoder) {
        log_info("Client %u: Opus decoder created (48kHz)", atomic_load(&client->client_id));
      } else {
        log_error("Client %u: Failed to create Opus decoder", atomic_load(&client->client_id));
      }
    }
  }
 
  if (stream_type & STREAM_TYPE_VIDEO) {
    log_info("Client %u announced video stream (waiting for first frame)", atomic_load(&client->client_id));
  }
  if (stream_type & STREAM_TYPE_AUDIO) {
    log_info("Client %u started audio stream", atomic_load(&client->client_id));
  }
 
  // Notify client of stream start acknowledgment
  const char *streams = (stream_type & STREAM_TYPE_VIDEO) && (stream_type & STREAM_TYPE_AUDIO)
                            ? "video+audio"
                            : ((stream_type & STREAM_TYPE_VIDEO) ? "video" : "audio");
  log_info_client(client, "Stream started: %s", streams);
}
 
void handle_stream_stop_packet(client_info_t *client, const void *data, size_t len) {
  VALIDATE_PACKET_SIZE(client, data, len, sizeof(uint32_t), "STREAM_STOP");
 
  uint32_t stream_type_net;
  memcpy(&stream_type_net, data, sizeof(uint32_t));
  uint32_t stream_type = NET_TO_HOST_U32(stream_type_net);
 
  // Validate at least one stream type flag is set
  const uint32_t VALID_STREAM_MASK = STREAM_TYPE_VIDEO | STREAM_TYPE_AUDIO;
  VALIDATE_CAPABILITY_FLAGS(client, stream_type, VALID_STREAM_MASK, "STREAM_STOP");
 
  // Validate no unknown stream type bits are set
  VALIDATE_FLAGS_MASK(client, stream_type, VALID_STREAM_MASK, "STREAM_STOP");
 
  if (stream_type & STREAM_TYPE_VIDEO) {
    atomic_store(&client->is_sending_video, false);
  }
  if (stream_type & STREAM_TYPE_AUDIO) {
    atomic_store(&client->is_sending_audio, false);
  }
 
  if (stream_type & STREAM_TYPE_VIDEO) {
    log_info("Client %u stopped video stream", atomic_load(&client->client_id));
  }
  if (stream_type & STREAM_TYPE_AUDIO) {
    log_info("Client %u stopped audio stream", atomic_load(&client->client_id));
  }
 
  // Notify client of stream stop acknowledgment
  const char *streams = (stream_type & STREAM_TYPE_VIDEO) && (stream_type & STREAM_TYPE_AUDIO)
                            ? "video+audio"
                            : ((stream_type & STREAM_TYPE_VIDEO) ? "video" : "audio");
  log_info_client(client, "Stream stopped: %s", streams);
}
 
/* ============================================================================
 * Media Data Packet Handlers
 * ============================================================================
 */
 
void handle_image_frame_packet(client_info_t *client, void *data, size_t len) {
  // Handle incoming image data from client
  // New format: [width:4][height:4][compressed_flag:4][data_size:4][rgb_data:data_size]
  // Old format: [width:4][height:4][rgb_data:w*h*3] (for backward compatibility)
  // Use atomic compare-and-swap to avoid race condition - ensures thread-safe auto-enabling of video stream
  if (!data || len < sizeof(uint32_t) * 2) {
    disconnect_client_for_bad_data(client, "IMAGE_FRAME payload too small: %zu bytes", len);
    return;
  }
  bool was_sending_video = atomic_load(&client->is_sending_video);
  if (!was_sending_video) {
    // Try to atomically enable video sending
    // Use atomic_compare_exchange_strong to avoid spurious failures
    if (atomic_compare_exchange_strong(&client->is_sending_video, &was_sending_video, true)) {
      log_info("Client %u auto-enabled video stream (received IMAGE_FRAME)", atomic_load(&client->client_id));
      // Notify client that their first video frame was received
      log_info_client(client, "First video frame received - streaming active");
    }
  } else {
    // Log periodically to confirm we're receiving frames
    // Use per-client counter protected by client_state_mutex to avoid race conditions
    mutex_lock(&client->client_state_mutex);
    client->frames_received_logged++;
    if (client->frames_received_logged % 25000 == 0) {
      char pretty[64];
      format_bytes_pretty(len, pretty, sizeof(pretty));
      log_debug("Client %u has sent %u IMAGE_FRAME packets (%s)", atomic_load(&client->client_id),
                client->frames_received_logged, pretty);
    }
    mutex_unlock(&client->client_state_mutex);
  }
 
  // Parse image dimensions (use memcpy to avoid unaligned access)
  uint32_t img_width_net, img_height_net;
  memcpy(&img_width_net, data, sizeof(uint32_t));
  memcpy(&img_height_net, (char *)data + sizeof(uint32_t), sizeof(uint32_t));
  uint32_t img_width = NET_TO_HOST_U32(img_width_net);
  uint32_t img_height = NET_TO_HOST_U32(img_height_net);
 
  log_debug("IMAGE_FRAME packet: width=%u, height=%u, payload_len=%zu", img_width, img_height, len);
 
  // Validate dimensions using image utility functions
  if (image_validate_dimensions((size_t)img_width, (size_t)img_height) != ASCIICHAT_OK) {
    log_error("IMAGE_FRAME validation failed for dimensions: %u x %u", img_width, img_height);
    disconnect_client_for_bad_data(client, "IMAGE_FRAME invalid dimensions");
    return;
  }
 
  // Calculate RGB buffer size with overflow checking
  size_t rgb_size = 0;
  if (image_calc_rgb_size((size_t)img_width, (size_t)img_height, &rgb_size) != ASCIICHAT_OK) {
    disconnect_client_for_bad_data(client, "IMAGE_FRAME buffer size calculation failed");
    return;
  }
 
  // Validate final buffer size against maximum
  if (image_validate_buffer_size(rgb_size) != ASCIICHAT_OK) {
    disconnect_client_for_bad_data(client, "IMAGE_FRAME buffer size exceeds maximum");
    return;
  }
 
  // Check if this is the new compressed format (has 4 fields) or old format (has 2 fields)
  const size_t legacy_header_size = sizeof(uint32_t) * 2;
  if (rgb_size > SIZE_MAX - legacy_header_size) {
    char size_str[32];
    format_bytes_pretty(rgb_size, size_str, sizeof(size_str));
    disconnect_client_for_bad_data(client, "IMAGE_FRAME legacy packet size overflow: %s", size_str);
    return;
  }
  size_t old_format_size = legacy_header_size + rgb_size;
  bool is_new_format = (len != old_format_size) && (len > sizeof(uint32_t) * 4);
 
  void *rgb_data = NULL;
  size_t rgb_data_size = 0;
  bool needs_free = false;
 
  if (is_new_format) {
    // New format: [width:4][height:4][compressed_flag:4][data_size:4][data:data_size]
    if (len < sizeof(uint32_t) * 4) {
      disconnect_client_for_bad_data(client, "IMAGE_FRAME new-format header too small (%zu bytes)", len);
      return;
    }
 
    // Use memcpy to avoid unaligned access for compressed_flag and data_size
    uint32_t compressed_flag_net, data_size_net;
    memcpy(&compressed_flag_net, (char *)data + sizeof(uint32_t) * 2, sizeof(uint32_t));
    memcpy(&data_size_net, (char *)data + sizeof(uint32_t) * 3, sizeof(uint32_t));
    uint32_t compressed_flag = NET_TO_HOST_U32(compressed_flag_net);
    uint32_t data_size = NET_TO_HOST_U32(data_size_net);
    void *frame_data = (char *)data + sizeof(uint32_t) * 4;
 
    const size_t new_header_size = sizeof(uint32_t) * 4;
    size_t data_size_sz = (size_t)data_size;
    if (data_size_sz > IMAGE_MAX_PIXELS_SIZE) {
      char size_str[32];
      format_bytes_pretty(data_size_sz, size_str, sizeof(size_str));
      disconnect_client_for_bad_data(client, "IMAGE_FRAME compressed data too large: %s", size_str);
      return;
    }
    if (data_size_sz > SIZE_MAX - new_header_size) {
      disconnect_client_for_bad_data(client, "IMAGE_FRAME new-format packet size overflow");
      return;
    }
    size_t expected_total = new_header_size + data_size_sz;
    if (len != expected_total) {
      disconnect_client_for_bad_data(client, "IMAGE_FRAME new-format length mismatch: expected %zu got %zu",
                                     expected_total, len);
      return;
    }
 
    if (compressed_flag) {
      // Decompress the data
      rgb_data = SAFE_MALLOC(rgb_size, void *);
      if (!rgb_data) {
        SET_ERRNO(ERROR_MEMORY, "Failed to allocate decompression buffer for client %u",
                  atomic_load(&client->client_id));
        return;
      }
 
      asciichat_error_t decompress_result = decompress_data(frame_data, data_size, rgb_data, rgb_size);
      if (decompress_result != ASCIICHAT_OK) {
        SAFE_FREE(rgb_data);
        disconnect_client_for_bad_data(client, "IMAGE_FRAME decompression failure for %zu bytes: %s", data_size_sz,
                                       asciichat_error_string(decompress_result));
        return;
      }
 
      rgb_data_size = rgb_size;
      needs_free = true;
    } else {
      // Uncompressed data
      rgb_data = frame_data;
      rgb_data_size = data_size;
      if (rgb_data_size != rgb_size) {
        disconnect_client_for_bad_data(client, "IMAGE_FRAME uncompressed size mismatch: expected %zu got %zu", rgb_size,
                                       rgb_data_size);
        return;
      }
    }
  } else {
    // Old format: [width:4][height:4][rgb_data:w*h*3]
    if (len != old_format_size) {
      disconnect_client_for_bad_data(client, "IMAGE_FRAME legacy length mismatch: expected %zu got %zu",
                                     old_format_size, len);
      return;
    }
    rgb_data = (char *)data + sizeof(uint32_t) * 2;
    rgb_data_size = rgb_size;
  }
 
  if (client->incoming_video_buffer) {
    // Get the write buffer
    video_frame_t *frame = video_frame_begin_write(client->incoming_video_buffer);
 
    if (frame && frame->data) {
      // Build the packet in the old format for internal storage: [width:4][height:4][rgb_data:w*h*3]
      if (rgb_data_size > SIZE_MAX - legacy_header_size) {
        if (needs_free && rgb_data) {
          SAFE_FREE(rgb_data);
        }
        char size_str[32];
        format_bytes_pretty(rgb_data_size, size_str, sizeof(size_str));
        disconnect_client_for_bad_data(client, "IMAGE_FRAME size overflow while repacking: rgb_data_size=%s", size_str);
        return;
      }
      size_t old_packet_size = legacy_header_size + rgb_data_size;
 
      if (old_packet_size <= 2 * 1024 * 1024) { // Max 2MB frame size
        uint32_t width_net = HOST_TO_NET_U32(img_width);
        uint32_t height_net = HOST_TO_NET_U32(img_height);
 
        // Pack in old format for internal consistency
        memcpy(frame->data, &width_net, sizeof(uint32_t));
        memcpy((char *)frame->data + sizeof(uint32_t), &height_net, sizeof(uint32_t));
        memcpy((char *)frame->data + sizeof(uint32_t) * 2, rgb_data, rgb_data_size);
 
        frame->size = old_packet_size;
        frame->width = img_width;
        frame->height = img_height;
        frame->capture_timestamp_us = (uint64_t)time(NULL) * 1000000;
        frame->sequence_number = ++client->frames_received;
        video_frame_commit(client->incoming_video_buffer);
      } else {
        if (needs_free && rgb_data) {
          SAFE_FREE(rgb_data);
        }
        disconnect_client_for_bad_data(client, "IMAGE_FRAME repacked frame too large (%zu bytes)", old_packet_size);
        return;
      }
    } else {
      log_warn("Failed to get write buffer for client %u (frame=%p, frame->data=%p)", atomic_load(&client->client_id),
               (void *)frame, frame ? frame->data : NULL);
    }
  } else {
    // During shutdown, this is expected - don't spam error logs
    if (!atomic_load(&g_server_should_exit)) {
      SET_ERRNO(ERROR_INVALID_STATE, "Client %u has no incoming video buffer!", atomic_load(&client->client_id));
    } else {
      log_debug("Client %u: ignoring video packet during shutdown", atomic_load(&client->client_id));
    }
  }
 
  // Clean up decompressed data if allocated
  if (needs_free && rgb_data) {
    SAFE_FREE(rgb_data);
  }
}
 
void handle_audio_packet(client_info_t *client, const void *data, size_t len) {
  VALIDATE_NOTNULL_DATA(client, data, "AUDIO");
  VALIDATE_AUDIO_ALIGNMENT(client, len, sizeof(float), "AUDIO");
  VALIDATE_AUDIO_STREAM_ENABLED(client, "AUDIO");
 
  int num_samples = (int)(len / sizeof(float));
  VALIDATE_AUDIO_SAMPLE_COUNT(client, num_samples, AUDIO_SAMPLES_PER_PACKET, "AUDIO");
  VALIDATE_RESOURCE_INITIALIZED(client, client->incoming_audio_buffer, "audio buffer");
 
  const float *samples = (const float *)data;
  asciichat_error_t result = audio_ring_buffer_write(client->incoming_audio_buffer, samples, num_samples);
  if (result != ASCIICHAT_OK) {
    log_error("Failed to write audio samples to buffer: %s", asciichat_error_string(result));
  }
}
 
void handle_remote_log_packet_from_client(client_info_t *client, const void *data, size_t len) {
  if (!client) {
    return;
  }
 
  log_level_t remote_level = LOG_INFO;
  remote_log_direction_t direction = REMOTE_LOG_DIRECTION_UNKNOWN;
  uint16_t flags = 0;
  char message[MAX_REMOTE_LOG_MESSAGE_LENGTH + 1] = {0};
 
  asciichat_error_t parse_result =
      packet_parse_remote_log(data, len, &remote_level, &direction, &flags, message, sizeof(message), NULL);
  if (parse_result != ASCIICHAT_OK) {
    disconnect_client_for_bad_data(client, "Invalid REMOTE_LOG packet: %s", asciichat_error_string(parse_result));
    return;
  }
 
  if (direction != REMOTE_LOG_DIRECTION_CLIENT_TO_SERVER) {
    disconnect_client_for_bad_data(client, "REMOTE_LOG direction mismatch: %u", direction);
    return;
  }
 
  const bool truncated = (flags & REMOTE_LOG_FLAG_TRUNCATED) != 0;
  const char *display_name = client->display_name[0] ? client->display_name : "(unnamed)";
  uint32_t client_id = atomic_load(&client->client_id);
 
  if (truncated) {
    log_msg(remote_level, __FILE__, __LINE__, __func__, "[REMOTE CLIENT %u \"%s\"] %s [message truncated]", client_id,
            display_name, message);
  } else {
    log_msg(remote_level, __FILE__, __LINE__, __func__, "[REMOTE CLIENT %u \"%s\"] %s", client_id, display_name,
            message);
  }
}
 
void handle_audio_batch_packet(client_info_t *client, const void *data, size_t len) {
  // Log every audio batch packet reception
  log_debug_every(LOG_RATE_DEFAULT, "Received audio batch packet from client %u (len=%zu, is_sending_audio=%d)",
                  atomic_load(&client->client_id), len, atomic_load(&client->is_sending_audio));
 
  VALIDATE_NOTNULL_DATA(client, data, "AUDIO_BATCH");
  VALIDATE_MIN_SIZE(client, len, sizeof(audio_batch_packet_t), "AUDIO_BATCH");
  VALIDATE_AUDIO_STREAM_ENABLED(client, "AUDIO_BATCH");
 
  // Parse batch header using utility function
  audio_batch_info_t batch_info;
  asciichat_error_t parse_result = audio_parse_batch_header(data, len, &batch_info);
  if (parse_result != ASCIICHAT_OK) {
    disconnect_client_for_bad_data(client, "Failed to parse audio batch header");
    return;
  }
 
  uint32_t packet_batch_count = batch_info.batch_count;
  uint32_t total_samples = batch_info.total_samples;
  uint32_t sample_rate = batch_info.sample_rate;
 
  (void)packet_batch_count;
  (void)sample_rate;
 
  VALIDATE_NONZERO(client, packet_batch_count, "batch_count", "AUDIO_BATCH");
  VALIDATE_NONZERO(client, total_samples, "total_samples", "AUDIO_BATCH");
 
  size_t samples_bytes = 0;
  if (safe_size_mul(total_samples, sizeof(uint32_t), &samples_bytes)) {
    disconnect_client_for_bad_data(client, "AUDIO_BATCH sample size overflow (samples=%u)", total_samples);
    return;
  }
 
  size_t expected_size = sizeof(audio_batch_packet_t) + samples_bytes;
  if (len != expected_size) {
    disconnect_client_for_bad_data(client, "AUDIO_BATCH length mismatch: got %zu expected %zu", len, expected_size);
    return;
  }
 
  // Bounds check to prevent integer overflow on allocation
  // Maximum allowed samples: AUDIO_BATCH_SAMPLES * 2 (2048 samples)
  // This prevents total_samples * sizeof(float) from exceeding 8KB
  const uint32_t MAX_AUDIO_SAMPLES = AUDIO_BATCH_SAMPLES * 2;
  if (total_samples > MAX_AUDIO_SAMPLES) {
    disconnect_client_for_bad_data(client, "AUDIO_BATCH too many samples: %u (max: %u)", total_samples,
                                   MAX_AUDIO_SAMPLES);
    return;
  }
 
  const uint8_t *samples_ptr = (const uint8_t *)data + sizeof(audio_batch_packet_t);
 
  // Safe allocation: total_samples is bounded above, so multiplication won't overflow
  size_t alloc_size = (size_t)total_samples * sizeof(float);
  float *samples = SAFE_MALLOC(alloc_size, float *);
  if (!samples) {
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate memory for audio sample conversion");
    return;
  }
 
  // Use helper function to dequantize samples
  asciichat_error_t dq_result = audio_dequantize_samples(samples_ptr, total_samples, samples);
  if (dq_result != ASCIICHAT_OK) {
    SAFE_FREE(samples);
    return;
  }
 
#ifndef NDEBUG
  static int recv_count = 0;
  recv_count++;
  if (recv_count % 100 == 0) {
    uint32_t raw0 = bytes_read_u32_unaligned(samples_ptr + 0 * sizeof(uint32_t));
    uint32_t raw1 = bytes_read_u32_unaligned(samples_ptr + 1 * sizeof(uint32_t));
    uint32_t raw2 = bytes_read_u32_unaligned(samples_ptr + 2 * sizeof(uint32_t));
    int32_t scaled0 = (int32_t)NET_TO_HOST_U32(raw0);
    int32_t scaled1 = (int32_t)NET_TO_HOST_U32(raw1);
    int32_t scaled2 = (int32_t)NET_TO_HOST_U32(raw2);
    log_info("RECV: network[0]=0x%08x, network[1]=0x%08x, network[2]=0x%08x", raw0, raw1, raw2);
    log_info("RECV: scaled[0]=%d, scaled[1]=%d, scaled[2]=%d", scaled0, scaled1, scaled2);
    log_info("RECV: samples[0]=%.6f, samples[1]=%.6f, samples[2]=%.6f", samples[0], samples[1], samples[2]);
  }
#endif
 
  if (client->incoming_audio_buffer) {
    asciichat_error_t write_result = audio_ring_buffer_write(client->incoming_audio_buffer, samples, total_samples);
    if (write_result != ASCIICHAT_OK) {
      log_error("Failed to write decoded audio batch to buffer: %s", asciichat_error_string(write_result));
    }
  }
 
  SAFE_FREE(samples);
}
 
void handle_audio_opus_batch_packet(client_info_t *client, const void *data, size_t len) {
  log_debug_every(LOG_RATE_SLOW, "Received Opus audio batch from client %u (len=%zu)", atomic_load(&client->client_id),
                  len);
 
  VALIDATE_NOTNULL_DATA(client, data, "AUDIO_OPUS_BATCH");
  VALIDATE_AUDIO_STREAM_ENABLED(client, "AUDIO_OPUS_BATCH");
  VALIDATE_RESOURCE_INITIALIZED(client, client->opus_decoder, "Opus decoder");
 
  // Parse Opus batch packet
  const uint8_t *opus_data = NULL;
  size_t opus_size = 0;
  const uint16_t *frame_sizes = NULL;
  int sample_rate = 0;
  int frame_duration = 0;
  int frame_count = 0;
 
  asciichat_error_t result = packet_parse_opus_batch(data, len, &opus_data, &opus_size, &frame_sizes, &sample_rate,
                                                     &frame_duration, &frame_count);
 
  if (result != ASCIICHAT_OK) {
    disconnect_client_for_bad_data(client, "Failed to parse AUDIO_OPUS_BATCH packet");
    return;
  }
 
  VALIDATE_NONZERO(client, frame_count, "frame_count", "AUDIO_OPUS_BATCH");
  VALIDATE_NONZERO(client, opus_size, "opus_size", "AUDIO_OPUS_BATCH");
 
  // Calculate samples per frame (20ms @ 48kHz = 960 samples)
  int samples_per_frame = (sample_rate * frame_duration) / 1000;
  VALIDATE_RANGE(client, samples_per_frame, 1, 4096, "samples_per_frame", "AUDIO_OPUS_BATCH");
 
  // Use static buffer for common case to avoid malloc in hot path
  // Typical batches: 1-32 frames of 960 samples = up to 30,720 samples
  // Static buffer holds 32 frames @ 48kHz 20ms = 30,720 samples (120KB)
#define OPUS_DECODE_STATIC_MAX_SAMPLES (32 * 960)
  static float static_decode_buffer[OPUS_DECODE_STATIC_MAX_SAMPLES];
 
  size_t total_samples = (size_t)samples_per_frame * (size_t)frame_count;
  float *decoded_samples;
  bool used_malloc = false;
 
  if (total_samples <= OPUS_DECODE_STATIC_MAX_SAMPLES) {
    decoded_samples = static_decode_buffer;
  } else {
    // Unusual large batch - fall back to malloc
    log_warn("Client %u: Large audio batch requires malloc (%zu samples)", atomic_load(&client->client_id),
             total_samples);
    decoded_samples = SAFE_MALLOC(total_samples * sizeof(float), float *);
    if (!decoded_samples) {
      SET_ERRNO(ERROR_MEMORY, "Failed to allocate buffer for Opus decoded samples");
      return;
    }
    used_malloc = true;
  }
 
  // Decode each Opus frame using frame_sizes array
  int total_decoded = 0;
  size_t opus_offset = 0;
 
  for (int i = 0; i < frame_count; i++) {
    // Get exact frame size from frame_sizes array (convert from network byte order)
    size_t frame_size = (size_t)NET_TO_HOST_U16(frame_sizes[i]);
 
    // DEBUG: Log the actual bytes of each Opus frame
    if (frame_size > 0) {
      log_debug_every(LOG_RATE_DEFAULT, "Client %u: Opus frame %d: size=%zu, first_bytes=[0x%02x,0x%02x,0x%02x,0x%02x]",
                      atomic_load(&client->client_id), i, frame_size, opus_data[opus_offset] & 0xFF,
                      frame_size > 1 ? (opus_data[opus_offset + 1] & 0xFF) : 0,
                      frame_size > 2 ? (opus_data[opus_offset + 2] & 0xFF) : 0,
                      frame_size > 3 ? (opus_data[opus_offset + 3] & 0xFF) : 0);
    }
 
    if (opus_offset + frame_size > opus_size) {
      log_error("Client %u: Frame %d size overflow (offset=%zu, frame_size=%zu, total=%zu)",
                atomic_load(&client->client_id), i + 1, opus_offset, frame_size, opus_size);
      if (used_malloc) {
        SAFE_FREE(decoded_samples);
      }
      return;
    }
 
    // SECURITY: Bounds check before writing decoded samples to prevent buffer overflow
    // An attacker could send malicious Opus frames that decode to more samples than expected
    if ((size_t)total_decoded + (size_t)samples_per_frame > total_samples) {
      log_error("Client %u: Opus decode would overflow buffer (decoded=%d, frame_samples=%d, max=%zu)",
                atomic_load(&client->client_id), total_decoded, samples_per_frame, total_samples);
      if (used_malloc) {
        SAFE_FREE(decoded_samples);
      }
      return;
    }
 
    int decoded_count = opus_codec_decode((opus_codec_t *)client->opus_decoder, &opus_data[opus_offset], frame_size,
                                          &decoded_samples[total_decoded], samples_per_frame);
 
    if (decoded_count < 0) {
      log_error("Client %u: Opus decoding failed for frame %d/%d (size=%zu)", atomic_load(&client->client_id), i + 1,
                frame_count, frame_size);
      if (used_malloc) {
        SAFE_FREE(decoded_samples);
      }
      return;
    }
 
    total_decoded += decoded_count;
    opus_offset += frame_size;
  }
 
  log_debug_every(LOG_RATE_DEFAULT, "Client %u: Decoded %d Opus frames -> %d samples", atomic_load(&client->client_id),
                  frame_count, total_decoded);
 
  // DEBUG: Log sample values to detect all-zero issue
  static int server_decode_count = 0;
  server_decode_count++;
  if (total_decoded > 0 && (server_decode_count <= 10 || server_decode_count % 100 == 0)) {
    float peak = 0.0f, rms = 0.0f;
    for (int i = 0; i < total_decoded && i < 100; i++) {
      float abs_val = fabsf(decoded_samples[i]);
      if (abs_val > peak)
        peak = abs_val;
      rms += decoded_samples[i] * decoded_samples[i];
    }
    rms = sqrtf(rms / (total_decoded > 100 ? 100 : total_decoded));
    // Log first 4 bytes of Opus data to compare with client encode
    log_info("SERVER OPUS DECODE #%d from client %u: decoded_rms=%.6f, opus_first4=[0x%02x,0x%02x,0x%02x,0x%02x]",
             server_decode_count, atomic_load(&client->client_id), rms, opus_size > 0 ? opus_data[0] : 0,
             opus_size > 1 ? opus_data[1] : 0, opus_size > 2 ? opus_data[2] : 0, opus_size > 3 ? opus_data[3] : 0);
  }
 
  // Write decoded samples to client's incoming audio buffer
  // Note: audio_ring_buffer_write returns error code, not sample count
  // Buffer overflow warnings are logged inside audio_ring_buffer_write if buffer is full
  if (client->incoming_audio_buffer && total_decoded > 0) {
    asciichat_error_t result = audio_ring_buffer_write(client->incoming_audio_buffer, decoded_samples, total_decoded);
    if (result != ASCIICHAT_OK) {
      log_error("Client %u: Failed to write decoded audio to buffer: %d", atomic_load(&client->client_id), result);
    }
  }
 
  if (used_malloc) {
    SAFE_FREE(decoded_samples);
  }
}
 
void handle_audio_opus_packet(client_info_t *client, const void *data, size_t len) {
  log_debug_every(LOG_RATE_DEFAULT, "Received Opus audio from client %u (len=%zu)", atomic_load(&client->client_id),
                  len);
 
  if (VALIDATE_PACKET_NOT_NULL(client, data, "AUDIO_OPUS")) {
    return;
  }
 
  // Minimum size: 16-byte header + at least 1 byte of Opus data
  if (len < 17) {
    disconnect_client_for_bad_data(client, "AUDIO_OPUS packet too small: %zu bytes", len);
    return;
  }
 
  if (!atomic_load(&client->is_sending_audio)) {
    disconnect_client_for_bad_data(client, "AUDIO_OPUS received before audio stream enabled");
    return;
  }
 
  if (!client->opus_decoder) {
    disconnect_client_for_bad_data(client, "Opus decoder not initialized");
    return;
  }
 
  // Parse header (16 bytes) - convert from network byte order
  const uint8_t *buf = (const uint8_t *)data;
  uint32_t sample_rate_net, frame_duration_net;
  memcpy(&sample_rate_net, buf, 4);
  memcpy(&frame_duration_net, buf + 4, 4);
  uint32_t sample_rate = NET_TO_HOST_U32(sample_rate_net);
  uint32_t frame_duration = NET_TO_HOST_U32(frame_duration_net);
 
  // Extract Opus data (after 16-byte header)
  const uint8_t *opus_data = buf + 16;
  size_t opus_size = len - 16;
 
  // Validate parameters
  if (sample_rate == 0 || sample_rate > 192000) {
    disconnect_client_for_bad_data(client, "AUDIO_OPUS invalid sample_rate: %u", sample_rate);
    return;
  }
 
  if (frame_duration == 0 || frame_duration > 120) {
    disconnect_client_for_bad_data(client, "AUDIO_OPUS invalid frame_duration: %u ms", frame_duration);
    return;
  }
 
  // Calculate expected samples per frame
  int samples_per_frame = (int)((sample_rate * frame_duration) / 1000);
  if (samples_per_frame <= 0 || samples_per_frame > 5760) { // Max 120ms @ 48kHz
    disconnect_client_for_bad_data(client, "AUDIO_OPUS invalid samples_per_frame: %d", samples_per_frame);
    return;
  }
 
  // Decode Opus frame
  float decoded_samples[5760]; // Max Opus frame size (120ms @ 48kHz)
  int decoded_count =
      opus_codec_decode((opus_codec_t *)client->opus_decoder, opus_data, opus_size, decoded_samples, samples_per_frame);
 
  if (decoded_count < 0) {
    log_error("Client %u: Opus decoding failed (size=%zu)", atomic_load(&client->client_id), opus_size);
    return;
  }
 
  log_debug_every(LOG_RATE_VERY_FAST, "Client %u: Decoded Opus frame -> %d samples", atomic_load(&client->client_id),
                  decoded_count);
 
  // Write decoded samples to client's incoming audio buffer
  if (client->incoming_audio_buffer && decoded_count > 0) {
    asciichat_error_t write_result =
        audio_ring_buffer_write(client->incoming_audio_buffer, decoded_samples, decoded_count);
    if (write_result != ASCIICHAT_OK) {
      log_error("Failed to write decoded Opus samples to buffer: %s", asciichat_error_string(write_result));
    }
  }
}
 
/* ============================================================================
 * Client Configuration Packet Handlers
 * ============================================================================
 */
 
void handle_client_capabilities_packet(client_info_t *client, const void *data, size_t len) {
  VALIDATE_PACKET_SIZE(client, data, len, sizeof(terminal_capabilities_packet_t), "CLIENT_CAPABILITIES");
 
  const terminal_capabilities_packet_t *caps = (const terminal_capabilities_packet_t *)data;
 
  // Extract and validate dimensions
  uint16_t width = NET_TO_HOST_U16(caps->width);
  uint16_t height = NET_TO_HOST_U16(caps->height);
 
  VALIDATE_NONZERO(client, width, "width", "CLIENT_CAPABILITIES");
  VALIDATE_NONZERO(client, height, "height", "CLIENT_CAPABILITIES");
  VALIDATE_RANGE(client, width, 1, 4096, "width", "CLIENT_CAPABILITIES");
  VALIDATE_RANGE(client, height, 1, 4096, "height", "CLIENT_CAPABILITIES");
 
  // Extract and validate color level (0=none, 1=16, 2=256, 3=truecolor)
  uint32_t color_level = NET_TO_HOST_U32(caps->color_level);
  VALIDATE_RANGE(client, color_level, 0, 3, "color_level", "CLIENT_CAPABILITIES");
 
  // Extract and validate render mode (0=foreground, 1=background, 2=half-block)
  uint32_t render_mode = NET_TO_HOST_U32(caps->render_mode);
  VALIDATE_RANGE(client, render_mode, 0, 2, "render_mode", "CLIENT_CAPABILITIES");
 
  // Extract and validate palette type (0-5 are valid, 5=PALETTE_CUSTOM)
  uint32_t palette_type = NET_TO_HOST_U32(caps->palette_type);
  VALIDATE_RANGE(client, palette_type, 0, 5, "palette_type", "CLIENT_CAPABILITIES");
 
  // Validate desired FPS (1-144)
  VALIDATE_RANGE(client, caps->desired_fps, 1, 144, "desired_fps", "CLIENT_CAPABILITIES");
 
  mutex_lock(&client->client_state_mutex);
 
  atomic_store(&client->width, width);
  atomic_store(&client->height, height);
 
  log_debug("Client %u dimensions: %ux%u, desired_fps=%u", atomic_load(&client->client_id), client->width,
            client->height, caps->desired_fps);
 
  client->terminal_caps.capabilities = NET_TO_HOST_U32(caps->capabilities);
  client->terminal_caps.color_level = color_level;
  client->terminal_caps.color_count = NET_TO_HOST_U32(caps->color_count);
  client->terminal_caps.render_mode = render_mode;
  client->terminal_caps.detection_reliable = caps->detection_reliable;
  client->terminal_caps.wants_background = (render_mode == RENDER_MODE_BACKGROUND);
 
  SAFE_STRNCPY(client->terminal_caps.term_type, caps->term_type, sizeof(client->terminal_caps.term_type));
  SAFE_STRNCPY(client->terminal_caps.colorterm, caps->colorterm, sizeof(client->terminal_caps.colorterm));
 
  client->terminal_caps.utf8_support = NET_TO_HOST_U32(caps->utf8_support);
  client->terminal_caps.palette_type = palette_type;
  SAFE_STRNCPY(client->terminal_caps.palette_custom, caps->palette_custom,
               sizeof(client->terminal_caps.palette_custom));
 
  client->terminal_caps.desired_fps = caps->desired_fps;
 
  const char *custom_chars =
      (client->terminal_caps.palette_type == PALETTE_CUSTOM && client->terminal_caps.palette_custom[0])
          ? client->terminal_caps.palette_custom
          : NULL;
 
  if (initialize_client_palette((palette_type_t)client->terminal_caps.palette_type, custom_chars,
                                client->client_palette_chars, &client->client_palette_len,
                                client->client_luminance_palette) == 0) {
    client->client_palette_type = (palette_type_t)client->terminal_caps.palette_type;
    client->client_palette_initialized = true;
    log_info("Client %d palette initialized: type=%u, %zu chars, utf8=%u", atomic_load(&client->client_id),
             client->terminal_caps.palette_type, client->client_palette_len, client->terminal_caps.utf8_support);
  } else {
    SET_ERRNO(ERROR_INVALID_STATE, "Failed to initialize palette for client %d", atomic_load(&client->client_id));
    client->client_palette_initialized = false;
  }
 
  client->has_terminal_caps = true;
 
  log_info("Client %u capabilities: %ux%u, color_level=%s (%u colors), caps=0x%x, term=%s, colorterm=%s, "
           "render_mode=%s, reliable=%s, fps=%u",
           atomic_load(&client->client_id), client->width, client->height,
           terminal_color_level_name(client->terminal_caps.color_level), client->terminal_caps.color_count,
           client->terminal_caps.capabilities, client->terminal_caps.term_type, client->terminal_caps.colorterm,
           (client->terminal_caps.render_mode == RENDER_MODE_HALF_BLOCK
                ? "half-block"
                : (client->terminal_caps.render_mode == RENDER_MODE_BACKGROUND ? "background" : "foreground")),
           client->terminal_caps.detection_reliable ? "yes" : "no", client->terminal_caps.desired_fps);
 
  // Send capabilities acknowledgment to client
  log_info_client(client, "Terminal configured: %ux%u, %s, %s mode, %u fps", client->width, client->height,
                  terminal_color_level_name(client->terminal_caps.color_level),
                  (client->terminal_caps.render_mode == RENDER_MODE_HALF_BLOCK
                       ? "half-block"
                       : (client->terminal_caps.render_mode == RENDER_MODE_BACKGROUND ? "background" : "foreground")),
                  client->terminal_caps.desired_fps);
 
  mutex_unlock(&client->client_state_mutex);
}
 
void handle_size_packet(client_info_t *client, const void *data, size_t len) {
  VALIDATE_PACKET_SIZE(client, data, len, sizeof(size_packet_t), "SIZE");
 
  const size_packet_t *size_pkt = (const size_packet_t *)data;
 
  // Extract and validate new dimensions
  uint16_t width = NET_TO_HOST_U16(size_pkt->width);
  uint16_t height = NET_TO_HOST_U16(size_pkt->height);
 
  VALIDATE_NONZERO(client, width, "width", "SIZE");
  VALIDATE_NONZERO(client, height, "height", "SIZE");
  VALIDATE_RANGE(client, width, 1, 4096, "width", "SIZE");
  VALIDATE_RANGE(client, height, 1, 4096, "height", "SIZE");
 
  mutex_lock(&client->client_state_mutex);
  client->width = width;
  client->height = height;
  mutex_unlock(&client->client_state_mutex);
 
  log_info("Client %u updated terminal size: %ux%u", atomic_load(&client->client_id), width, height);
}
 
/* ============================================================================
 * Protocol Control Packet Handlers
 * ============================================================================
 */
 
void handle_ping_packet(client_info_t *client) {
  // PONG responses should be handled directly via socket in send thread
  // For now, just log the ping
  (void)client;
}
 
/* ============================================================================
 * Protocol Utility Functions
 * ============================================================================
 */
 
int send_server_state_to_client(client_info_t *client) {
  if (!client) {
    return -1;
  }
 
  // Count active clients - LOCK OPTIMIZATION: Use atomic reads, no rwlock needed
  int active_count = 0;
  for (int i = 0; i < MAX_CLIENTS; i++) {
    if (atomic_load(&g_client_manager.clients[i].active)) {
      active_count++;
    }
  }
 
  // Prepare server state packet
  server_state_packet_t state;
  state.connected_client_count = active_count;
  state.active_client_count = active_count; // For now, all connected are active
  memset(state.reserved, 0, sizeof(state.reserved));
 
  // Convert to network byte order
  server_state_packet_t net_state;
  net_state.connected_client_count = HOST_TO_NET_U32(state.connected_client_count);
  net_state.active_client_count = HOST_TO_NET_U32(state.active_client_count);
  memset(net_state.reserved, 0, sizeof(net_state.reserved));
 
  // Send server state via ACIP transport
  // CRITICAL: Protect socket writes with send_mutex to prevent race with send_thread
  mutex_lock(&client->send_mutex);
  asciichat_error_t result = acip_send_server_state(client->transport, &net_state);
  mutex_unlock(&client->send_mutex);
 
  if (result != ASCIICHAT_OK) {
    SET_ERRNO(ERROR_NETWORK, "Failed to send server state to client %u: %s", client->client_id,
              asciichat_error_string(result));
    return -1;
  }
 
  log_debug("Sent server state to client %u: %u connected, %u active", client->client_id, state.connected_client_count,
            state.active_client_count);
  return 0;
}
 
// NOTE: This function is no longer used - CLEAR_CONSOLE is now sent directly
// from each client's render thread when it detects a grid layout change.
// Keeping this for reference but it should not be called.
void broadcast_clear_console_to_all_clients(void) {
  SET_ERRNO(ERROR_INVALID_STATE, "broadcast_clear_console_to_all_clients() called - unexpected usage");
  log_warn("CLEAR_CONSOLE is now sent from render threads, not broadcast");
}