diff --git a/lib/Makefile.am b/lib/Makefile.am
index c8127f96..eafade5d 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -26,7 +26,9 @@ libuasync_a_SOURCES = \
 	radix.c \
 	radix.h \
 	serialize.c \
-	serialize.h
+	serialize.h \
+	tcp_io.c \
+	tcp_io.h
 
 libuasync_a_CFLAGS = \
 	-D_ISOC99_SOURCE \
diff --git a/lib/ll_queue.c b/lib/ll_queue.c
index e202e568..f362432f 100644
--- a/lib/ll_queue.c
+++ b/lib/ll_queue.c
@@ -46,6 +46,17 @@ static inline void queue_check_thread(struct ll_queue* q) {
 // Предварительные объявления внутренних функций
 static void queue_resume_timeout_cb(void* arg);
 static void check_waiters(struct ll_queue* q);
+static void empty_trampoline(void* arg);
+
+static void empty_trampoline(void* arg) {
+    struct ll_queue* q = (struct ll_queue*)arg;
+    q->empty_call_soon_id = NULL;
+    queue_callback_fn cb = q->empty_callback;
+    void* cb_arg = q->empty_callback_arg;
+    q->empty_callback = NULL;
+    q->empty_callback_arg = NULL;
+    if (q->count == 0 && cb) cb(q, cb_arg);
+}
 
 // ==================== Управление очередью ====================
 
@@ -134,6 +145,12 @@ void queue_free(struct ll_queue* q) {
         uasync_call_soon_cancel(q->ua, q->resume_timeout_id);
         q->resume_timeout_id = NULL;
     }
+
+    void* old_eid = q->empty_call_soon_id;
+    q->empty_call_soon_id = NULL;
+    if (old_eid) uasync_call_soon_cancel(q->ua, old_eid);
+    q->empty_callback = NULL;
+    q->empty_callback_arg = NULL;
     
     u_free(q);
 }
@@ -532,6 +549,9 @@ struct ll_entry* queue_data_get(struct ll_queue* q) {
     // Проверить ожидающие коллбэки
     check_waiters(q);
     
+    if (q->count == 0 && q->empty_callback && !q->empty_call_soon_id)
+        q->empty_call_soon_id = uasync_call_soon(q->ua, q, (timeout_callback_t)empty_trampoline);
+    
 #ifdef QUEUE_DEBUG
     queue_check_consistency(q);// !!!! for debug
 #endif
@@ -613,6 +633,17 @@ void queue_set_waiter_defer(struct ll_queue* q, int enable) {
     q->waiter_defer = enable;
 }
 
+void queue_set_empty_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg) {
+    if (!q) return;
+    void* old = q->empty_call_soon_id;
+    q->empty_call_soon_id = NULL;
+    if (old) uasync_call_soon_cancel(q->ua, old);
+    q->empty_callback = cbk_fn;
+    q->empty_callback_arg = arg;
+    if (cbk_fn && q->count == 0)
+        q->empty_call_soon_id = uasync_call_soon(q->ua, q, (timeout_callback_t)empty_trampoline);
+}
+
 int queue_waiter_wait(struct ll_queue* q, struct queue_waiter_handle* h,
                       queue_threshold_callback_fn callback, void* arg) {
     if (!q || !h) return -1;
diff --git a/lib/ll_queue.h b/lib/ll_queue.h
index be3aeb2d..23f06cd0 100644
--- a/lib/ll_queue.h
+++ b/lib/ll_queue.h
@@ -144,6 +144,10 @@ struct ll_queue {
     size_t threshold_max_bytes;       // Общий порог: макс. объём данных (0 = не проверять)
     int waiter_defer;                 // 0=immediate callback, 1=defer via uasync_call_soon
 
+    queue_callback_fn empty_callback;     // одноразовый коллбэк при опустошении очереди
+    void*             empty_callback_arg;
+    void*             empty_call_soon_id; // handle от uasync_call_soon
+
     struct ll_entry** hash_table;   // Хеш-таблица для поиска по id (если hash_size > 0)
     size_t hash_size;               // Размер хеш-таблицы
     uint16_t index_offset;          // Смещение индекса в data[] (для всех entry очереди)
@@ -228,6 +232,17 @@ void queue_set_threshold(struct ll_queue* q, int max_packets, size_t max_bytes);
 
 void queue_set_waiter_defer(struct ll_queue* q, int enable);
 
+/**
+ * @brief Устанавливает одноразовый коллбэк при опустошении очереди (count==0).
+ * @param q очередь
+ * @param cbk_fn коллбэк (NULL для отмены)
+ * @param arg пользовательский аргумент
+ *
+ * Коллбэк вызывается асинхронно (через uasync_call_soon) когда count становится 0.
+ * После вызова сбрасывается — для повторного использования нужно вызвать снова.
+ */
+void queue_set_empty_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg);
+
 /**
  * @brief Регистрирует ожидание освобождения очереди до общего порога.
  * @param q очередь
diff --git a/lib/tcp_io.c b/lib/tcp_io.c
new file mode 100644
index 00000000..bcac4e38
--- /dev/null
+++ b/lib/tcp_io.c
@@ -0,0 +1,358 @@
+// tcp_io.c — управление TCP-соединением через uasync + ll_queue
+#include "tcp_io.h"
+#include "debug_config.h"
+#include "mem.h"
+#include <stdlib.h>
+#include <string.h>
+#include <errno.h>
+#ifndef _WIN32
+#include <unistd.h>
+#include <sys/socket.h>
+#include <fcntl.h>
+#endif
+#ifndef MSG_NOSIGNAL
+#define MSG_NOSIGNAL 0
+#endif
+
+static void read_cb(socket_t sock, void* arg);
+static void write_cb(socket_t sock, void* arg);
+static void error_cb(socket_t sock, void* arg);
+static void resume_read_cb(struct ll_queue* q, void* arg);
+static void fin_deferred_cb(struct ll_queue* q, void* arg);
+static void write_queue_fetch_cb(struct ll_queue* q, void* arg);
+static void flush_write_buf(struct tcp_conn* tc);
+
+struct tcp_conn* tcp_conn_create(
+    struct UASYNC* ua, socket_t sock,
+    size_t entry_data_size, size_t write_chunk_size,
+    int read_high_water, int read_low_water,
+    void (*on_fin)(struct tcp_conn* tc, void* arg),
+    void (*on_error)(struct tcp_conn* tc, int err, void* arg),
+    void* arg)
+{
+    if (!ua || sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: invalid args ua=%p", ua); return NULL; }
+
+    struct tcp_conn* tc = u_calloc(1, sizeof(struct tcp_conn));
+    if (!tc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: u_calloc failed"); return NULL; }
+
+    tc->ua = ua;
+    tc->sock = sock;
+    tc->entry_data_size = entry_data_size;
+    tc->write_chunk_size = write_chunk_size;
+    tc->read_high_water = read_high_water;
+    tc->read_low_water = read_low_water;
+    tc->on_fin = on_fin;
+    tc->on_error = on_error;
+    tc->arg = arg;
+
+    tc->entry_pool = memory_pool_init(sizeof(struct ll_entry));
+    {
+        size_t ds = entry_data_size > write_chunk_size ? entry_data_size : write_chunk_size;
+        tc->data_pool = memory_pool_init(ds);
+    }
+    if (!tc->entry_pool || !tc->data_pool) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: memory_pool_init failed");
+        if (tc->entry_pool) memory_pool_destroy(tc->entry_pool);
+        if (tc->data_pool)  memory_pool_destroy(tc->data_pool);
+        u_free(tc); return NULL;
+    }
+
+    tc->read_queue = queue_new(ua, 0, 0, 0, "tcp_io_rq");
+    tc->write_queue = queue_new(ua, 0, 0, 0, "tcp_io_wq");
+    if (!tc->read_queue || !tc->write_queue) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: queue_new failed");
+        if (tc->read_queue) queue_free(tc->read_queue);
+        if (tc->write_queue) queue_free(tc->write_queue);
+        memory_pool_destroy(tc->entry_pool);
+        memory_pool_destroy(tc->data_pool);
+        u_free(tc); return NULL;
+    }
+    queue_set_threshold(tc->read_queue, read_low_water, 0);
+    queue_set_callback(tc->write_queue, write_queue_fetch_cb, tc);
+    queue_set_waiter_defer(tc->write_queue, 1);
+
+    tc->socket_id = uasync_add_socket_t(ua, sock, read_cb, write_cb, error_cb, tc);
+    if (!tc->socket_id) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: uasync_add_socket_t failed");
+        queue_free(tc->read_queue);
+        queue_free(tc->write_queue);
+        memory_pool_destroy(tc->entry_pool);
+        memory_pool_destroy(tc->data_pool);
+        u_free(tc); return NULL;
+    }
+    tc->write_monitor = 1;
+
+    {
+        struct sockaddr_storage addr;
+        socklen_t alen = sizeof(addr);
+        if (getpeername(sock, (struct sockaddr*)&addr, &alen) == 0) tc->connected = 1;
+    }
+
+    DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: fd=%d entry=%zu chunk=%zu hw=%d lw=%d connected=%d",
+               (int)sock, entry_data_size, write_chunk_size, read_high_water, read_low_water, tc->connected);
+    return tc;
+}
+
+void tcp_conn_destroy(struct tcp_conn* tc) {
+    if (!tc) return;
+    DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_conn_destroy: fd=%d connected=%d error=%d fin=%d",
+               (int)tc->sock, tc->connected, tc->error, tc->fin);
+
+    if (tc->socket_id) {
+        uasync_remove_socket_t(tc->ua, tc->sock);
+        tc->socket_id = NULL;
+    }
+    queue_waiter_cancel(tc->read_queue, &tc->read_waiter);
+    queue_set_empty_callback(tc->read_queue, NULL, NULL);
+
+    struct ll_entry* e;
+    while ((e = queue_data_get(tc->read_queue)) != NULL) {
+        if (e->dgram) memory_pool_free(tc->data_pool, e->dgram);
+        queue_entry_free(e); queue_resume_callback(tc->read_queue);
+    }
+    while ((e = queue_data_get(tc->write_queue)) != NULL) {
+        if (e->dgram) memory_pool_free(tc->data_pool, e->dgram);
+        queue_entry_free(e); queue_resume_callback(tc->write_queue);
+    }
+    queue_free(tc->read_queue);
+    queue_free(tc->write_queue);
+
+    if (tc->write_buf) memory_pool_free(tc->data_pool, tc->write_buf);
+    memory_pool_destroy(tc->entry_pool);
+    memory_pool_destroy(tc->data_pool);
+    u_free(tc);
+}
+
+// ====================================================================
+// Чтение из сокета
+// ====================================================================
+
+static void read_cb(socket_t sock, void* arg) {
+    (void)sock;
+    struct tcp_conn* tc = (struct tcp_conn*)arg;
+    if (!tc || tc->sock == SOCKET_INVALID) return;
+
+    struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool);
+    uint8_t* buf = memory_pool_alloc(tc->data_pool);
+    if (!e || !buf) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: read_cb alloc failed fd=%d", (int)tc->sock);
+        if (e) queue_entry_free(e);
+        if (buf) memory_pool_free(tc->data_pool, buf);
+        uasync_set_socket_read(tc->ua, tc->socket_id, 0);
+        tc->read_paused = 1;
+        return;
+    }
+
+    ssize_t n = recv(tc->sock, buf, tc->entry_data_size, 0);
+    if (n > 0) {
+        e->dgram = buf;
+        e->len = (uint16_t)n;
+        queue_data_put(tc->read_queue, e);
+        if (tc->read_queue->count >= tc->read_high_water && !tc->read_paused) {
+            uasync_set_socket_read(tc->ua, tc->socket_id, 0);
+            tc->read_paused = 1;
+            queue_waiter_wait(tc->read_queue, &tc->read_waiter, resume_read_cb, tc);
+        }
+    } else if (n == 0) {
+        memory_pool_free(tc->data_pool, buf);
+        queue_entry_free(e);
+        tc->fin = 1;
+        DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_io: FIN fd=%d", (int)tc->sock);
+        uasync_set_socket_read(tc->ua, tc->socket_id, 0);
+        if (tc->read_queue->count == 0) {
+            if (tc->on_fin) tc->on_fin(tc, tc->arg);
+        } else {
+            queue_set_empty_callback(tc->read_queue, fin_deferred_cb, tc);
+        }
+    } else {
+        memory_pool_free(tc->data_pool, buf);
+        queue_entry_free(e);
+        if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return;
+        tc->error = 1;
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: recv error fd=%d errno=%d", (int)tc->sock, errno);
+        if (tc->on_error) tc->on_error(tc, errno, tc->arg);
+    }
+}
+
+static void resume_read_cb(struct ll_queue* q, void* arg) {
+    (void)q;
+    struct tcp_conn* tc = (struct tcp_conn*)arg;
+    if (!tc || tc->sock == SOCKET_INVALID) return;
+    uasync_set_socket_read(tc->ua, tc->socket_id, 1);
+    tc->read_paused = 0;
+}
+
+static void fin_deferred_cb(struct ll_queue* q, void* arg) {
+    (void)q;
+    struct tcp_conn* tc = (struct tcp_conn*)arg;
+    if (tc->on_fin) tc->on_fin(tc, tc->arg);
+}
+
+// ====================================================================
+// Запись в сокет
+// ====================================================================
+
+static void flush_write_buf(struct tcp_conn* tc) {
+    while (tc->write_buf && tc->write_offset < tc->write_len) {
+        ssize_t n = send(tc->sock, tc->write_buf + tc->write_offset, tc->write_len - tc->write_offset, MSG_NOSIGNAL);
+        if (n > 0) {
+            tc->write_offset += (size_t)n;
+            if (tc->write_offset >= tc->write_len) {
+                memory_pool_free(tc->data_pool, tc->write_buf);
+                tc->write_buf = NULL;
+                tc->write_len = 0;
+                tc->write_offset = 0;
+                return;
+            }
+            continue;
+        }
+        if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) return;
+        tc->error = 1;
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: send error fd=%d errno=%d", (int)tc->sock, errno);
+        if (tc->on_error) tc->on_error(tc, errno, tc->arg);
+        return;
+    }
+}
+
+static void write_queue_fetch_cb(struct ll_queue* q, void* arg) {
+    struct tcp_conn* tc = (struct tcp_conn*)arg;
+
+    if (tc->error || tc->fin) return;
+
+    if (tc->write_buf) {
+        flush_write_buf(tc);
+        if (tc->write_buf) return;         // остался остаток, ждём write_cb
+    }
+
+    if (!tc->connected) return;             // ждём connect
+
+    struct ll_entry* e = queue_data_get(q);
+    if (!e) {
+        uasync_set_socket_write(tc->ua, tc->socket_id, 0);
+        tc->write_monitor = 0;
+        queue_resume_callback(q);
+        if (tc->on_flushed) {
+            void (*cb)(struct tcp_conn*, void*) = tc->on_flushed;
+            tc->on_flushed = NULL;
+            cb(tc, tc->arg);
+        }
+        return;
+    }
+
+    ssize_t n = send(tc->sock, e->dgram, e->len, MSG_NOSIGNAL);
+    if (n == (ssize_t)e->len) {
+        memory_pool_free(tc->data_pool, e->dgram);
+        queue_entry_free(e);
+        queue_resume_callback(q);
+    } else if (n > 0) {
+        size_t rem = e->len - (size_t)n;
+        tc->write_buf = memory_pool_alloc(tc->data_pool);
+        if (!tc->write_buf) {
+            DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: write_buf alloc failed fd=%d", (int)tc->sock);
+            tc->error = 1;
+            memory_pool_free(tc->data_pool, e->dgram);
+            queue_entry_free(e);
+            return;
+        }
+        memcpy(tc->write_buf, e->dgram + n, rem);
+        tc->write_len = rem;
+        tc->write_offset = 0;
+        memory_pool_free(tc->data_pool, e->dgram);
+        queue_entry_free(e);
+        uasync_set_socket_write(tc->ua, tc->socket_id, 1);
+        tc->write_monitor = 1;
+    } else if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) {
+        tc->write_buf = memory_pool_alloc(tc->data_pool);
+        if (!tc->write_buf) {
+            DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: write_buf alloc failed fd=%d", (int)tc->sock);
+            tc->error = 1;
+            memory_pool_free(tc->data_pool, e->dgram);
+            queue_entry_free(e);
+            return;
+        }
+        memcpy(tc->write_buf, e->dgram, e->len);
+        tc->write_len = e->len;
+        tc->write_offset = 0;
+        memory_pool_free(tc->data_pool, e->dgram);
+        queue_entry_free(e);
+        uasync_set_socket_write(tc->ua, tc->socket_id, 1);
+        tc->write_monitor = 1;
+    } else {
+        tc->error = 1;
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: send error fd=%d errno=%d", (int)tc->sock, errno);
+        if (tc->on_error) tc->on_error(tc, errno, tc->arg);
+        memory_pool_free(tc->data_pool, e->dgram);
+        queue_entry_free(e);
+    }
+}
+
+static void write_cb(socket_t sock, void* arg) {
+    (void)sock;
+    struct tcp_conn* tc = (struct tcp_conn*)arg;
+    if (!tc || tc->sock == SOCKET_INVALID) return;
+
+    if (!tc->connected) {
+        int err = 0;
+        socklen_t len = sizeof(err);
+        if (getsockopt(tc->sock, SOL_SOCKET, SO_ERROR, &err, &len) == 0 && err == 0) {
+            tc->connected = 1;
+            DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_io: connect ok fd=%d", (int)tc->sock);
+        } else {
+            tc->error = 1;
+            DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: connect fail fd=%d err=%d", (int)tc->sock, err);
+            if (tc->on_error) tc->on_error(tc, err ? err : -1, tc->arg);
+            return;
+        }
+    }
+
+    flush_write_buf(tc);
+    if (!tc->write_buf) queue_resume_callback(tc->write_queue);
+}
+
+// ====================================================================
+// Асинхронная ошибка сокета
+// ====================================================================
+
+static void error_cb(socket_t sock, void* arg) {
+    (void)sock;
+    struct tcp_conn* tc = (struct tcp_conn*)arg;
+    if (!tc || tc->sock == SOCKET_INVALID) return;
+    tc->error = 1;
+    DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: async error fd=%d", (int)tc->sock);
+    if (tc->on_error) tc->on_error(tc, -1, tc->arg);
+}
+
+void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_conn* tc, void* arg)) {
+    if (!tc) return;
+    tc->on_flushed = on_flushed;
+}
+
+// ====================================================================
+// Отправка данных в сокет (внешний интерфейс)
+// ====================================================================
+
+int tcp_conn_push_write(struct tcp_conn* tc, const uint8_t* data, size_t len) {
+    if (!tc || !data || len == 0) return -1;
+    if (tc->error || tc->fin) return -1;
+
+    size_t offset = 0;
+    while (offset < len) {
+        size_t chunk = len - offset;
+        if (chunk > tc->write_chunk_size) chunk = tc->write_chunk_size;
+
+        struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool);
+        uint8_t* buf = memory_pool_alloc(tc->data_pool);
+        if (!e || !buf) {
+            DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: push_write alloc failed fd=%d chunk=%zu", (int)tc->sock, chunk);
+            if (e) queue_entry_free(e);
+            if (buf) memory_pool_free(tc->data_pool, buf);
+            return -1;
+        }
+        memcpy(buf, data + offset, chunk);
+        e->dgram = buf;
+        e->len = (uint16_t)chunk;
+        queue_data_put(tc->write_queue, e);
+        offset += chunk;
+    }
+    return 0;
+}
diff --git a/lib/tcp_io.h b/lib/tcp_io.h
new file mode 100644
index 00000000..ec783bff
--- /dev/null
+++ b/lib/tcp_io.h
@@ -0,0 +1,96 @@
+// tcp_io.h — управление TCP-соединением через uasync + ll_queue
+//
+// Один tcp_conn = одно TCP-соединение. Две очереди: read_queue (сокет → данные) и
+// write_queue (данные → сокет). Обе работают через автозабор ll_queue (deferred).
+//
+// Пул-аллокация:
+//   entry_pool — только struct ll_entry (без inline data)
+//   data_pool  — буферы данных чтения/записи (max(entry_data_size, write_chunk_size))
+// Данные и структуры аллоцируются раздельно — не копируются при recv/send.
+//
+// Чтение:
+//   read_cb → memory_pool_alloc(read_pool) → recv() прямо в буфер → entry в read_queue
+//   read_queue автозабор(deferred) → внешний коллбэк (например, отправка в ETCP)
+//   high_water → пауза чтения (убираем EPOLLIN), low_water/waiter → возобновление
+//   FIN: если read_queue не пуст → empty_callback откладывает on_fin; иначе сразу
+//
+// Запись:
+//   Внешний код: tcp_conn_push_write(data, len) → аллокация через write_pool → entry в write_queue
+//   write_queue автозабор(deferred): fetch_cb → send() в сокет
+//   EAGAIN → write_buf (из data_pool) + EPOLLOUT ON → write_cb досылает → resume автозабора
+//   Всё отправлено → EPOLLOUT OFF + on_flushed (если установлен)
+//
+// Connect:
+//   tcp_conn_create регистрирует сокет с read_cb + write_cb в uasync
+//   После create вызывается connect() (неблокирующий, EINPROGRESS)
+//   write_cb детектит завершение connect через getsockopt(SO_ERROR)
+//   getpeername() сразу после create проверяет pre-connected сокеты (socketpair)
+//   До connect данные копятся в write_queue (автозабор не пытается send на unconnected сокет)
+
+#ifndef TCP_IO_H
+#define TCP_IO_H
+
+#include "u_async.h"
+#include "socket_compat.h"
+#include "ll_queue.h"
+#include "memory_pool.h"
+
+struct tcp_conn {
+    socket_t sock;
+    struct UASYNC* ua;
+    void*    socket_id;
+
+    struct ll_queue* read_queue;      // сокет → данные (блоки до entry_data_size)
+    struct ll_queue* write_queue;     // данные → сокет (блоки до write_chunk_size)
+
+    int      read_high_water;
+    int      read_low_water;
+    uint8_t  read_paused;
+
+    uint8_t  write_monitor;           // 1 = EPOLLOUT активен
+    uint8_t  connected;
+    uint8_t  error;
+    uint8_t  fin;                     // FIN от сокета
+
+    // Частичная отправка (из write_pool, не в очереди — досылается первой)
+    uint8_t* write_buf;
+    size_t   write_len;
+    size_t   write_offset;
+    size_t   write_chunk_size;
+
+    // Пулы памяти
+    struct memory_pool* entry_pool;   // sizeof(struct ll_entry)
+    struct memory_pool* data_pool;    // max(entry_data_size, write_chunk_size)
+    size_t   entry_data_size;
+
+    struct queue_waiter_handle read_waiter;
+
+    // Коллбэки
+    void (*on_fin)(struct tcp_conn* tc, void* arg);
+    void (*on_error)(struct tcp_conn* tc, int err, void* arg);
+    void (*on_flushed)(struct tcp_conn* tc, void* arg);  // все данные записи отправлены
+    void* arg;
+};
+
+struct tcp_conn* tcp_conn_create(
+    struct UASYNC* ua,
+    socket_t sock,
+    size_t entry_data_size,
+    size_t write_chunk_size,
+    int read_high_water,
+    int read_low_water,
+    void (*on_fin)(struct tcp_conn* tc, void* arg),
+    void (*on_error)(struct tcp_conn* tc, int err, void* arg),
+    void* arg);
+
+void tcp_conn_destroy(struct tcp_conn* tc);
+
+// Отправка данных в сокет: режет на чанки по write_chunk_size, кладёт в write_queue.
+// Автозабор write_queue сам отправит когда сокет будет готов.
+int  tcp_conn_push_write(struct tcp_conn* tc, const uint8_t* data, size_t len);
+
+// Одноразовый коллбэк: вызывается когда write_queue + write_buf полностью опустели.
+// После вызова сбрасывается. Установить повторно можно в любой момент.
+void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_conn* tc, void* arg));
+
+#endif
diff --git a/lib/u_async.c b/lib/u_async.c
index 9dbf47dd..48ba795a 100644
--- a/lib/u_async.c
+++ b/lib/u_async.c
@@ -42,17 +42,19 @@ struct timeout_node {
 };
 
 // Socket node with array-based storage
-struct socket_node {
-    int fd;                     // File descriptor (for pipe, file)
-    socket_t sock;              // Socket (for cross-platform sockets)
-    int type;                   // SOCKET_NODE_TYPE_FD or SOCKET_NODE_TYPE_SOCK
-    socket_callback_t read_cbk;       // For FD type
-    socket_callback_t write_cbk;      // For FD type
-    socket_t_callback_t read_cbk_sock;  // For SOCK type
-    socket_t_callback_t write_cbk_sock; // For SOCK type
-    socket_callback_t except_cbk;
-    void* user_data;
-    int active;  // 1 if socket is active, 0 if u_freed (for reuse)
+struct socket_node {
+    int fd;                     // File descriptor (for pipe, file)
+    socket_t sock;              // Socket (for cross-platform sockets)
+    int type;                   // SOCKET_NODE_TYPE_FD or SOCKET_NODE_TYPE_SOCK
+    socket_callback_t read_cbk;       // For FD type
+    socket_callback_t write_cbk;      // For FD type
+    socket_t_callback_t read_cbk_sock;  // For SOCK type
+    socket_t_callback_t write_cbk_sock; // For SOCK type
+    socket_callback_t except_cbk;
+    void* user_data;
+    int active;       // 1 if socket is active, 0 if u_freed (for reuse)
+    int enable_read;  // 1 if read monitoring is enabled
+    int enable_write; // 1 if write monitoring is enabled
 };
 
 // Array-based socket management for O(1) operations
@@ -187,10 +189,12 @@ static int socket_array_add_internal(struct socket_array* sa, int fd, socket_t s
     sa->sockets[index].write_cbk = write_cbk_fd;
     sa->sockets[index].read_cbk_sock = read_cbk_sock;
     sa->sockets[index].write_cbk_sock = write_cbk_sock;
-    sa->sockets[index].except_cbk = except_cbk;
-    sa->sockets[index].user_data = user_data;
-    sa->sockets[index].active = 1;
-    
+    sa->sockets[index].except_cbk = except_cbk;
+    sa->sockets[index].user_data = user_data;
+    sa->sockets[index].active = 1;
+    sa->sockets[index].enable_read = (read_cbk_fd != NULL || read_cbk_sock != NULL) ? 1 : 0;
+    sa->sockets[index].enable_write = (write_cbk_fd != NULL || write_cbk_sock != NULL) ? 1 : 0;
+    
     sa->fd_to_index[fd] = index;
     sa->index_to_fd[index] = fd;
     sa->active_indices[sa->count] = index;  // Add to active list
@@ -240,6 +244,8 @@ static int socket_array_remove(struct socket_array* sa, int fd) {
     sa->sockets[index].write_cbk_sock = NULL;
     sa->sockets[index].except_cbk = NULL;
     sa->sockets[index].user_data = NULL;
+    sa->sockets[index].enable_read = 0;
+    sa->sockets[index].enable_write = 0;
     sa->fd_to_index[fd] = -1;
     sa->index_to_fd[index] = -1;
     
@@ -755,10 +761,84 @@ err_t uasync_remove_socket_t(struct UASYNC* ua, socket_t sock) {
         ua->poll_fds_dirty = 1;
         return ERR_OK;
     }
-    return ERR_FAIL;
-}
-
-// Helper function to rebuild cached pollfd array
+    return ERR_FAIL;
+}
+
+err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable) {
+    if (!ua || !s_id) return ERR_FAIL;
+
+    struct socket_node* node = (struct socket_node*)s_id;
+    if (!node->active || node->fd < 0) return ERR_FAIL;
+
+    int val = enable ? 1 : 0;
+    if (node->enable_read == val) return ERR_OK;
+    node->enable_read = val;
+
+#if HAS_EPOLL
+    if (ua->use_epoll && ua->epoll_fd >= 0) {
+        struct epoll_event ev;
+        ev.events = 0;
+        if (node->type == SOCKET_NODE_TYPE_SOCK) {
+            if (node->read_cbk_sock && node->enable_read) ev.events |= EPOLLIN;
+            if (node->write_cbk_sock && node->enable_write) ev.events |= EPOLLOUT;
+            ev.data.fd = node->sock;
+        } else {
+            if (node->read_cbk && node->enable_read) ev.events |= EPOLLIN;
+            if (node->write_cbk && node->enable_write) ev.events |= EPOLLOUT;
+            ev.data.fd = node->fd;
+        }
+        if (node->except_cbk) ev.events |= EPOLLPRI;
+#ifdef _WIN32
+        int efd = (int)(intptr_t)ev.data.fd;
+#else
+        int efd = ev.data.fd;
+#endif
+        epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, efd, &ev);
+    }
+#endif
+
+    ua->poll_fds_dirty = 1;
+    return ERR_OK;
+}
+
+err_t uasync_set_socket_write(struct UASYNC* ua, void* s_id, int enable) {
+    if (!ua || !s_id) return ERR_FAIL;
+
+    struct socket_node* node = (struct socket_node*)s_id;
+    if (!node->active || node->fd < 0) return ERR_FAIL;
+
+    int val = enable ? 1 : 0;
+    if (node->enable_write == val) return ERR_OK;
+    node->enable_write = val;
+
+#if HAS_EPOLL
+    if (ua->use_epoll && ua->epoll_fd >= 0) {
+        struct epoll_event ev;
+        ev.events = 0;
+        if (node->type == SOCKET_NODE_TYPE_SOCK) {
+            if (node->read_cbk_sock && node->enable_read) ev.events |= EPOLLIN;
+            if (node->write_cbk_sock && node->enable_write) ev.events |= EPOLLOUT;
+            ev.data.fd = node->sock;
+        } else {
+            if (node->read_cbk && node->enable_read) ev.events |= EPOLLIN;
+            if (node->write_cbk && node->enable_write) ev.events |= EPOLLOUT;
+            ev.data.fd = node->fd;
+        }
+        if (node->except_cbk) ev.events |= EPOLLPRI;
+#ifdef _WIN32
+        int efd = (int)(intptr_t)ev.data.fd;
+#else
+        int efd = ev.data.fd;
+#endif
+        epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, efd, &ev);
+    }
+#endif
+
+    ua->poll_fds_dirty = 1;
+    return ERR_OK;
+}
+
+// Helper function to rebuild cached pollfd array
 static void rebuild_poll_fds(struct UASYNC* ua) {
     if (!ua || !ua->sockets) return;
     
@@ -808,14 +888,14 @@ static void rebuild_poll_fds(struct UASYNC* ua) {
         ua->poll_fds[idx].events = 0;
         ua->poll_fds[idx].revents = 0;
         
-        if (cur->type == SOCKET_NODE_TYPE_SOCK) {
-            if (cur->read_cbk_sock) ua->poll_fds[idx].events |= POLLIN;
-            if (cur->write_cbk_sock) ua->poll_fds[idx].events |= POLLOUT;
-        } else {
-            if (cur->read_cbk) ua->poll_fds[idx].events |= POLLIN;
-            if (cur->write_cbk) ua->poll_fds[idx].events |= POLLOUT;
-        }
-        if (cur->write_cbk) ua->poll_fds[idx].events |= POLLOUT;
+        if (cur->type == SOCKET_NODE_TYPE_SOCK) {
+            if (cur->read_cbk_sock && cur->enable_read) ua->poll_fds[idx].events |= POLLIN;
+            if (cur->write_cbk_sock && cur->enable_write) ua->poll_fds[idx].events |= POLLOUT;
+        } else {
+            if (cur->read_cbk && cur->enable_read) ua->poll_fds[idx].events |= POLLIN;
+            if (cur->write_cbk && cur->enable_write) ua->poll_fds[idx].events |= POLLOUT;
+        }
+        if (cur->write_cbk && cur->enable_write) ua->poll_fds[idx].events |= POLLOUT;
         if (cur->except_cbk) ua->poll_fds[idx].events |= POLLPRI;
         
         idx++;
@@ -1012,12 +1092,12 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) {
             s = (SOCKET)node->fd;
         }
         
-        if (node->type == SOCKET_NODE_TYPE_SOCK) {
-            if (node->read_cbk_sock) FD_SET(s, &read_fds);
-            if (node->write_cbk_sock) FD_SET(s, &write_fds);
-        } else {
-            if (node->read_cbk) FD_SET(s, &read_fds);
-            if (node->write_cbk) FD_SET(s, &write_fds);
+        if (node->type == SOCKET_NODE_TYPE_SOCK) {
+            if (node->read_cbk_sock && node->enable_read) FD_SET(s, &read_fds);
+            if (node->write_cbk_sock && node->enable_write) FD_SET(s, &write_fds);
+        } else {
+            if (node->read_cbk && node->enable_read) FD_SET(s, &read_fds);
+            if (node->write_cbk && node->enable_write) FD_SET(s, &write_fds);
         }
         if (node->except_cbk) FD_SET(s, &except_fds);
         
diff --git a/lib/u_async.h b/lib/u_async.h
index 2e7a003a..187e12b7 100644
--- a/lib/u_async.h
+++ b/lib/u_async.h
@@ -102,8 +102,12 @@ err_t uasync_call_soon_cancel(struct UASYNC* ua, void* t_id);
 void* uasync_add_socket(struct UASYNC* ua, int fd, socket_callback_t read_cbk, socket_callback_t write_cbk, socket_callback_t except_cbk, void* user_arg);
 // Sockets - for socket_t (cross-platform sockets)
 void* uasync_add_socket_t(struct UASYNC* ua, socket_t sock, socket_t_callback_t read_cbk, socket_t_callback_t write_cbk, socket_t_callback_t except_cbk, void* user_arg);
-err_t uasync_remove_socket(struct UASYNC* ua, void* s_id);
-err_t uasync_remove_socket_t(struct UASYNC* ua, socket_t sock);
+err_t uasync_remove_socket(struct UASYNC* ua, void* s_id);
+err_t uasync_remove_socket_t(struct UASYNC* ua, socket_t sock);
+
+// Динамическое управление мониторингом чтения/записи для сокета
+err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable);
+err_t uasync_set_socket_write(struct UASYNC* ua, void* s_id, int enable);
 
 // Single iteration of event loop with timeout (timebase units)
 void uasync_poll(struct UASYNC* ua, int timeout_tb);
diff --git a/src/proxy/tcp_proxy_client.c b/src/proxy/tcp_proxy_client.c
index 20378994..bae5ab27 100644
--- a/src/proxy/tcp_proxy_client.c
+++ b/src/proxy/tcp_proxy_client.c
@@ -486,7 +486,7 @@ void tcp_proxy_client_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entr
         if (rc) {
             if (subcmd == TCP_PROXY_SUBCMD_DATA)  { tcp_proxy_server_handle_data(inst, conn, entry, stream_id); return; }
             if (subcmd == TCP_PROXY_SUBCMD_CLOSE) { tcp_proxy_server_handle_close(inst, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return; }
-            if (subcmd == TCP_PROXY_SUBCMD_ERROR) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "RP ERROR recv sid=%08x", stream_id); rc->error = 1; tcp_proxy_server_conn_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return; }
+            if (subcmd == TCP_PROXY_SUBCMD_ERROR) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "RP ERROR recv sid=%08x", stream_id); tcp_proxy_server_conn_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return; }
         }
     }
     if (proxy) {
diff --git a/src/proxy/tcp_proxy_server.c b/src/proxy/tcp_proxy_server.c
index 0d792866..1b5d0b3d 100644
--- a/src/proxy/tcp_proxy_server.c
+++ b/src/proxy/tcp_proxy_server.c
@@ -12,6 +12,7 @@
 #include "../lib/debug_config.h"
 #include "../lib/ll_queue.h"
 #include "../lib/mem.h"
+#include "../lib/tcp_io.h"
 #include <stdlib.h>
 #include <string.h>
 #include <errno.h>
@@ -22,29 +23,35 @@
 #include <arpa/inet.h>
 #include <fcntl.h>
 #endif
-#ifndef MSG_NOSIGNAL
-#define MSG_NOSIGNAL 0
-#endif
 
 static struct tcp_proxy_server* g_tcp_proxy_server_ctx = NULL;
 
-static void tcp_proxy_server_sock_read_cb(socket_t sock, void* arg);
-static void tcp_proxy_server_sock_write_cb(socket_t sock, void* arg);
-static void tcp_proxy_server_sock_error_cb(socket_t sock, void* arg);
-static void tcp_proxy_server_sock_retry_cb(void* arg);
-static void tcp_proxy_server_pause_waiter_cb(struct ll_queue* q, void* arg);
-static void tcp_proxy_server_close_retry_cb(void* arg);
-static void tcp_proxy_server_sock_send(struct tcp_proxy_server_conn* rc, const uint8_t* data, size_t len);
+static void on_fin_cb(struct tcp_conn* tc, void* arg);
+static void on_error_cb(struct tcp_conn* tc, int err, void* arg);
+static void on_flushed_cb(struct tcp_conn* tc, void* arg);
+static void read_queue_drain_cb(struct ll_queue* q, void* arg);
+static void pause_resume_cb(struct ll_queue* q, void* arg);
+static void close_retry_cb(void* arg);
+static int  send_msg(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t subcmd,
+                     uint32_t sid, const uint8_t* data, size_t len);
+static void send_close(struct tcp_proxy_server_conn* rc);
+static void send_error(struct tcp_proxy_server_conn* rc);
 void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc);
 
-// Отправка сообщения через ETCP-маршрут: выделяет queue_entry, заполняет заголовок TCP_PROXY_HDR_SIZE, вызывает etcp_route_send.
-// Вызывается из всех send-функций (close/error/data).
-static int tcp_proxy_server_send_msg(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t subcmd,
-                        uint32_t sid, const uint8_t* data, size_t len) {
+static inline int write_pending(struct tcp_conn* tc) {
+    return tc && (tc->write_buf || tc->write_queue->head);
+}
+
+// ====================================================================
+// Отправка сообщений через ETCP
+// ====================================================================
+
+static int send_msg(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t subcmd,
+                    uint32_t sid, const uint8_t* data, size_t len) {
     struct ll_entry* e = queue_entry_new(0);
-    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_server_send_msg: queue_entry_new failed subcmd=%02x sid=%08x", subcmd, sid); return -1; }
+    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_server: queue_entry_new failed subcmd=%02x sid=%08x", subcmd, sid); return -1; }
     e->dgram = u_malloc(TCP_PROXY_HDR_SIZE + len);
-    if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_server_send_msg: malloc(%zu) failed subcmd=%02x sid=%08x", TCP_PROXY_HDR_SIZE + len, subcmd, sid); queue_entry_free(e); return -1; }
+    if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_server: malloc(%zu) failed", TCP_PROXY_HDR_SIZE + len); queue_entry_free(e); return -1; }
     e->dgram[0] = ETCP_ID_TCP_PROXY;
     e->dgram[1] = subcmd;
     memcpy(e->dgram + 2, &sid, 4);
@@ -53,296 +60,166 @@ static int tcp_proxy_server_send_msg(struct UTUN_INSTANCE* inst, uint64_t dst, u
     return etcp_route_send(inst, dst, e);
 }
 
-// Таймер-коллбэк повтора CLOSE/ERROR с экспоненциальным backoff (50..5000 tb). Запускается из send_close/send_error при неудаче etcp_route_send.
-static void tcp_proxy_server_close_retry_cb(void* arg) {
+static void close_retry_cb(void* arg) {
     struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
     if (!rc) return;
     rc->close_timer = NULL;
     if (!rc->close_pending) return;
     struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
     if (!inst) return;
-    uint8_t subcmd = rc->error ? TCP_PROXY_SUBCMD_ERROR : TCP_PROXY_SUBCMD_CLOSE;
-    if (tcp_proxy_server_send_msg(inst, rc->peer_node_id, subcmd, rc->stream_id, NULL, 0) < 0) {
+    uint8_t subcmd = (rc->tc && rc->tc->error) ? TCP_PROXY_SUBCMD_ERROR : TCP_PROXY_SUBCMD_CLOSE;
+    if (send_msg(inst, rc->peer_node_id, subcmd, rc->stream_id, NULL, 0) < 0) {
         rc->close_backoff = rc->close_backoff < 5000 ? rc->close_backoff * 2 : 5000;
-        rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, tcp_proxy_server_close_retry_cb, "tcp_proxy_server_close_retry");
+        rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, close_retry_cb, "tps_close_retry");
         return;
     }
     rc->close_pending = 0;
 }
 
-// Отправка TCP_PROXY_SUBCMD_CLOSE пиру; при неудаче — close_pending=1 + retry-таймер. Вызывается из sock_read_cb (EOF), sock_retry_cb (завершение send).
-static void tcp_proxy_server_send_close(struct tcp_proxy_server_conn* rc) {
+static void send_close(struct tcp_proxy_server_conn* rc) {
     struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
     if (!inst) return;
-    if (tcp_proxy_server_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_CLOSE, rc->stream_id, NULL, 0) < 0) {
+    if (send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_CLOSE, rc->stream_id, NULL, 0) < 0) {
         rc->close_pending = 1;
         rc->close_backoff = 50;
         if (!rc->close_timer)
-            rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, tcp_proxy_server_close_retry_cb, "tcp_proxy_server_close_retry");
+            rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, close_retry_cb, "tps_close_retry");
     }
 }
 
-// Отправка TCP_PROXY_SUBCMD_ERROR пиру; при неудаче — close_pending=1 + retry-таймер. Вызывается из sock_write_cb (ошибка connect), sock_error_cb.
-static void tcp_proxy_server_send_error(struct tcp_proxy_server_conn* rc) {
+static void send_error(struct tcp_proxy_server_conn* rc) {
     struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
     if (!inst) return;
-    if (tcp_proxy_server_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_ERROR, rc->stream_id, NULL, 0) < 0) {
+    if (send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_ERROR, rc->stream_id, NULL, 0) < 0) {
         rc->close_pending = 1;
         rc->close_backoff = 50;
         if (!rc->close_timer)
-            rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, tcp_proxy_server_close_retry_cb, "tcp_proxy_server_close_retry");
+            rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, close_retry_cb, "tps_close_retry");
     }
 }
 
-// Подсчёт всех коннектов в ctx->conns (для диагностики).
-static int tcp_proxy_server_conn_total(struct tcp_proxy_server_conn* rc) {
-    if (!rc || !rc->ctx) return 0;
-    int n = 0; struct tcp_proxy_server_conn* c;
-    for (c = rc->ctx->conns; c; c = c->next) n++;
-    return n;
-}
-
-// Неблокирующий send в TCP-сокет с обработкой EAGAIN/EWOULDBLOCK/EINTR. Возвращает: 0=done, 1=need_retry, -1=error.
-// Вызывается из sock_send и sock_retry_cb.
 // ====================================================================
-// Неблокирующий send с EAGAIN и таймером повтора
+// Коллбэки tcp_io
 // ====================================================================
-static int tcp_proxy_server_sock_try_send(struct tcp_proxy_server_conn* rc) {
-    if (!rc || rc->sock == SOCKET_INVALID || !rc->out_buf) return 0;
-    while (rc->out_off < rc->out_len) {
-        ssize_t n = send(rc->sock, rc->out_buf + rc->out_off, rc->out_len - rc->out_off, MSG_NOSIGNAL);
-        if (n > 0) {
-            rc->out_off += n;
-            if (rc->out_off >= rc->out_len) {
-                u_free(rc->out_buf); rc->out_buf = NULL; rc->out_len = rc->out_off = 0;
-                return 0;
-            }
-            continue;
-        }
-        if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) return 1;
-        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:SEND_ERR fd=%d sid=%08x err=%s", (int)rc->sock, rc->stream_id, strerror(errno));
-        return -1;
-    }
-    return 0;
-}
 
-// Таймер-коллбэк повтора sock_try_send с backoff (10..5000 tb). При успешном завершении — shutdown(SHUT_WR) если cli_closed, send_close если sock_closed, conn_free если обе стороны закрыты.
-static void tcp_proxy_server_sock_retry_cb(void* arg) {
+static void on_fin_cb(struct tcp_conn* tc, void* arg) {
     struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
-    if (!rc || !rc->out_buf) { rc->out_timer = NULL; return; }
-    int ret = tcp_proxy_server_sock_try_send(rc);
-    if (ret == 0) {
-        rc->out_timer = NULL; rc->out_backoff = 0;
-        if (rc->cli_closed && rc->sock != SOCKET_INVALID && rc->connected == 1) shutdown(rc->sock, SHUT_WR);
-        if (rc->sock_closed) tcp_proxy_server_send_close(rc);
-        if (rc->cli_closed && rc->sock_closed) tcp_proxy_server_conn_free(rc);
-    } else if (ret == 1) {
-        rc->out_backoff = rc->out_backoff < 20 ? 20 : rc->out_backoff * 2;
-        if (rc->out_backoff > 5000) rc->out_backoff = 5000;
-        rc->out_timer = uasync_set_timeout(rc->ua, rc->out_backoff, rc, tcp_proxy_server_sock_retry_cb, "tcp_proxy_server_send");
-    } else {
-        u_free(rc->out_buf); rc->out_buf = NULL; rc->out_len = rc->out_off = 0;
-        rc->out_timer = NULL; rc->out_backoff = 0;
-    }
+    if (write_pending(tc))
+        tcp_conn_set_flushed(tc, on_flushed_cb);
+    else
+        send_close(rc);
 }
 
-// Буферизованная неблокирующая отправка данных в TCP-сокет. Если уже есть out_buf — дописывает (realloc). Иначе sock_try_send, при EAGAIN ставит retry-таймер.
-// Вызывается из sock_write_cb (сброс после connect) и handle_data (приём DATA от клиента).
-static void tcp_proxy_server_sock_send(struct tcp_proxy_server_conn* rc, const uint8_t* data, size_t len) {
-    if (!rc || !data || len == 0) return;
-    if (rc->out_buf) {
-        size_t new_len = rc->out_len + len;
-        uint8_t* new_buf = u_realloc(rc->out_buf, new_len);
-        if (!new_buf) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:SEND realloc(%zu) failed sid=%08x", new_len, rc->stream_id); return; }
-        memcpy(new_buf + rc->out_len, data, len);
-        rc->out_buf = new_buf; rc->out_len = new_len;
-        return;
-    }
-    rc->out_buf = u_malloc(len);
-    if (!rc->out_buf) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:SEND malloc(%zu) failed sid=%08x", len, rc->stream_id); return; }
-    memcpy(rc->out_buf, data, len); rc->out_len = len; rc->out_off = 0;
-    int ret = tcp_proxy_server_sock_try_send(rc);
-    if (ret == 0) return;
-    if (ret == 1) {
-        rc->out_backoff = 10;
-        rc->out_timer = uasync_set_timeout(rc->ua, rc->out_backoff, rc, tcp_proxy_server_sock_retry_cb, "tcp_proxy_server_send");
-    } else {
-        u_free(rc->out_buf); rc->out_buf = NULL; rc->out_len = rc->out_off = 0;
-    }
+static void on_flushed_cb(struct tcp_conn* tc, void* arg) {
+    struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
+    if (rc->cli_closed) { tcp_proxy_server_conn_free(rc); return; }
+    send_close(rc);
 }
 
-// [queue] Коллбэк освобождения очереди ETCP (backpressure relief). Срабатывает когда etcp_router_waiter сигнализирует что очередь освободилась.
-// Отправляет сохранённый в pause_buf буфер через ETCP, затем восстанавливает uasync-чтение из сокета.
-// Цепочка: queue waiter (threshold) → etcp_router → эта функция.
-static void tcp_proxy_server_pause_waiter_cb(struct ll_queue* q, void* arg) {
-    (void)q;
+static void on_error_cb(struct tcp_conn* tc, int err, void* arg) {
+    (void)tc; (void)err;
     struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
-    if (!rc || rc->sock == SOCKET_INVALID) return;
-    struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
-    if (!inst) return;
-    if (rc->pause_buf && rc->pause_len > 0) {
-        DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:PAUSE_FLUSH fd=%d sid=%08x len=%zu", (int)rc->sock, rc->stream_id, rc->pause_len);
-        if (tcp_proxy_server_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->pause_buf, rc->pause_len) < 0) {
-            DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:PAUSE_FLUSH send failed — closing sid=%08x", rc->stream_id);
-            rc->error = 1; u_free(rc->pause_buf); rc->pause_buf = NULL; rc->pause_len = 0; tcp_proxy_server_conn_free(rc); return;
-        }
-        u_free(rc->pause_buf); rc->pause_buf = NULL; rc->pause_len = 0;
-    }
-    rc->read_id = uasync_add_socket_t(rc->ua, rc->sock, tcp_proxy_server_sock_read_cb, tcp_proxy_server_sock_write_cb, tcp_proxy_server_sock_error_cb, rc);
-    if (rc->sock_closed) tcp_proxy_server_send_close(rc);
+    send_error(rc);
+    tcp_proxy_server_conn_free(rc);
 }
 
 // ====================================================================
-// Socket callbacks — вызываются из uasync при событиях на TCP-сокете
+// Дрейн read_queue → ETCP (автозабор + backpressure)
 // ====================================================================
 
-// Сокет-коллбэк чтения. Читает данные из TCP-сокета, отправляет через ETCP пиру (DATA).
-// При ошибке ETCP-send (backpressure) — буферизует в pause_buf, снимает сокет с uasync, регистрирует waiter в очереди.
-// При EOF (n=0) — sock_closed=1, send_close.
-static void tcp_proxy_server_sock_read_cb(socket_t sock, void* arg) {
-    (void)sock; struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
-    if (!rc || rc->sock == SOCKET_INVALID) return;
-    uint8_t buf[8192]; ssize_t n = recv(rc->sock, buf, sizeof(buf), 0);
-    if (n > 0) {
-        if (rc->cli_closed || rc->error) return;
-        struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
-        if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:RECV fd=%d sid=%08x — inst is NULL, drop", (int)rc->sock, rc->stream_id); return; }
-        if (rc->pause_buf) return;
-        int ret = tcp_proxy_server_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, buf, (size_t)n);
-        if (ret == 0) {
-            DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:RECV fd=%d sid=%08x len=%zd total=%d", (int)rc->sock, rc->stream_id, n, tcp_proxy_server_conn_total(rc));
-        } else {
-            DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:PAUSE fd=%d sid=%08x len=%zd — backpressure, paused", (int)rc->sock, rc->stream_id, n);
-            rc->pause_buf = u_malloc(n);
-            if (rc->pause_buf) { memcpy(rc->pause_buf, buf, n); rc->pause_len = n; }
-            else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:PAUSE malloc(%zd) failed sid=%08x", n, rc->stream_id);
-            if (rc->read_id) { uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL; }
-            etcp_router_waiter_register(inst, rc->peer_node_id, &rc->pause_waiter,
-                                         tcp_proxy_server_pause_waiter_cb, rc);
-        }
-    } else if (n == 0) {
-        DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:EOF fd=%d sid=%08x total=%d cli_closed=%d sock_closed=%d",
-                   (int)rc->sock, rc->stream_id, tcp_proxy_server_conn_total(rc), rc->cli_closed, rc->sock_closed);
-        rc->sock_closed = 1;
-        if (!rc->pause_buf && !rc->out_buf) tcp_proxy_server_send_close(rc);
-        if (rc->read_id) { uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL; }
-        if (rc->cli_closed && !rc->out_buf && !rc->pause_buf) tcp_proxy_server_conn_free(rc);
-    } else if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
-        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:ERROR fd=%d sid=%08x errno=%d %s total=%d",
-                    (int)rc->sock, rc->stream_id, errno, strerror(errno), tcp_proxy_server_conn_total(rc));
-        rc->error = 1;
-        struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
-        if (inst) tcp_proxy_server_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_ERROR, rc->stream_id, NULL, 0);
-        else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:ERROR fd=%d sid=%08x — inst is NULL, can't send ERROR", (int)rc->sock, rc->stream_id);
-        tcp_proxy_server_conn_free(rc);
+static void read_queue_drain_cb(struct ll_queue* q, void* arg) {
+    struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
+    struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
+    struct ll_entry* e = queue_data_get(q);
+    if (!e) { queue_resume_callback(q); return; }
+    int ret = send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, e->dgram, e->len);
+    DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:RECV fd=%d sid=%08x len=%u total=%d",
+               (int)rc->tc->sock, rc->stream_id, e->len, rc->ctx ? rc->ctx->conn_count : 0);
+    memory_pool_free(rc->tc->data_pool, e->dgram);
+    queue_entry_free(e);
+    if (ret == 0) {
+        queue_resume_callback(q);
+    } else {
+        etcp_router_waiter_register(inst, rc->peer_node_id, &rc->pause_waiter, pause_resume_cb, rc);
     }
 }
 
-// Сокет-коллбэк записи. Срабатывает при завершении неблокирующего connect(). Проверяет SO_ERROR, при успехе — connected=1, сбрасывает pending_buf в сокет.
-static void tcp_proxy_server_sock_write_cb(socket_t sock, void* arg) {
-    (void)sock; struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
-    if (!rc || rc->sock == SOCKET_INVALID || rc->connected) return;
-    int err = 0; socklen_t len = sizeof(err);
-    if (getsockopt(rc->sock, SOL_SOCKET, SO_ERROR, &err, &len) == 0 && err == 0) {
-        rc->connected = 1;
-        struct sockaddr_in local; socklen_t llen = sizeof(local);
-        uint16_t local_port = 0;
-        if (getsockname(rc->sock, (struct sockaddr*)&local, &llen) == 0) local_port = local.sin_port;
-        DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:CONN fd=%d sid=%08x local=%d dest=%d.%d.%d.%d:%d total=%d",
-                   (int)rc->sock, rc->stream_id, ntohs(local_port),
-                   rc->dest_ip[0], rc->dest_ip[1], rc->dest_ip[2], rc->dest_ip[3], ntohs(rc->dest_port),
-                   tcp_proxy_server_conn_total(rc));
-        if (rc->pending_buf && rc->pending_len > 0) {
-            DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:FLUSH fd=%d sid=%08x len=%zu", (int)rc->sock, rc->stream_id, rc->pending_len);
-            tcp_proxy_server_sock_send(rc, rc->pending_buf, rc->pending_len);
-            u_free(rc->pending_buf); rc->pending_buf = NULL; rc->pending_len = 0;
-        }
-    } else {
-        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:FAIL fd=%d sid=%08x err=%d %s",
-                    (int)rc->sock, rc->stream_id, err, err ? strerror(err) : "unknown");
-        tcp_proxy_server_send_error(rc);
-        tcp_proxy_server_conn_free(rc);
-    }
+static void pause_resume_cb(struct ll_queue* q, void* arg) {
+    (void)q;
+    struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
+    if (!rc->tc || rc->tc->sock == SOCKET_INVALID) return;
+    queue_resume_callback(rc->tc->read_queue);
 }
 
-// Сокет-коллбэк ошибки. Асинхронная ошибка сокета — отправляет ERROR пиру и освобождает коннект через conn_free.
-static void tcp_proxy_server_sock_error_cb(socket_t sock, void* arg) {
-    (void)sock; struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
-    if (!rc || rc->sock == SOCKET_INVALID) return;
-    DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:ERR fd=%d sid=%08x", (int)rc->sock, rc->stream_id);
-    rc->error = 1;
-    tcp_proxy_server_send_error(rc);
-    tcp_proxy_server_conn_free(rc);
+// ====================================================================
+// Управление жизненным циклом коннекта
+// ====================================================================
+
+static int conn_total(struct tcp_proxy_server_conn* rc) {
+    if (!rc || !rc->ctx) return 0;
+    int n = 0; struct tcp_proxy_server_conn* c;
+    for (c = rc->ctx->conns; c; c = c->next) n++;
+    return n;
 }
 
-// Полное освобождение коннекта: удаление из ctx->conns, закрытие сокета, отмена таймеров (out/close), освобождение буферов (pending/out/pause), отмена queue waiter.
-// Вызывается из всех точек завершения/очистки коннекта.
 void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) {
     if (!rc) return;
-    int total = tcp_proxy_server_conn_total(rc);
-    DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:FREE fd=%d sid=%08x total=%d cli_closed=%d sock_closed=%d error=%d",
-               (int)rc->sock, rc->stream_id, total, rc->cli_closed, rc->sock_closed, rc->error);
+    int total = conn_total(rc);
+    DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:FREE fd=%d sid=%08x total=%d cli_closed=%d fin=%d error=%d",
+               rc->tc ? (int)rc->tc->sock : -1, rc->stream_id, total, rc->cli_closed, rc->tc ? rc->tc->fin : 0, rc->tc ? rc->tc->error : 0);
     if (rc->ctx) {
         struct tcp_proxy_server_conn** prev = &rc->ctx->conns;
         while (*prev) { if (*prev == rc) { *prev = rc->next; rc->ctx->conn_count--; break; } prev = &(*prev)->next; }
     }
-    if (rc->sock != SOCKET_INVALID) {
-        if (rc->read_id) { uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL; }
-        socket_close_wrapper(rc->sock); rc->sock = SOCKET_INVALID;
-    }
-    if (rc->pending_buf) { u_free(rc->pending_buf); rc->pending_buf = NULL; }
-    if (rc->out_timer) { uasync_cancel_timeout(rc->ua, rc->out_timer); rc->out_timer = NULL; }
-    if (rc->out_buf) { u_free(rc->out_buf); rc->out_buf = NULL; }
-    if (rc->pause_buf) { u_free(rc->pause_buf); rc->pause_buf = NULL; rc->pause_len = 0; }
+    if (rc->tc) { tcp_conn_destroy(rc->tc); rc->tc = NULL; }
     if (rc->close_timer) { uasync_cancel_timeout(rc->ua, rc->close_timer); rc->close_timer = NULL; }
     if (rc->ctx && rc->ctx->inst) etcp_router_waiter_cancel(rc->ctx->inst, rc->peer_node_id, &rc->pause_waiter);
     u_free(rc);
 }
 
-// ====================================================================
-// Public API — входные точки из ETCP-диспетчера (tcp_proxy_client_etcp_recv_cb)
-// ====================================================================
-
-// Поиск коннекта по stream_id в цепочке ctx->conns.
 struct tcp_proxy_server_conn* tcp_proxy_server_find_conn(struct tcp_proxy_server* ctx, uint32_t stream_id) {
     struct tcp_proxy_server_conn* c;
     for (c = ctx->conns; c; c = c->next) if (c->stream_id == stream_id) return c;
     return NULL;
 }
 
-// Входная точка: обработка CONNECT от клиента. Создаёт коннект, инициирует неблокирующий connect() к dest_ip:dest_port, регистрирует сокет в uasync.
-// Вызывается из ETCP-диспетчера при приёме пакета с subcmd=CONNECT.
-int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* entry,
-                                 uint32_t stream_id, uint64_t src_node_id) {
+// ====================================================================
+// Входные точки из ETCP-диспетчера
+// ====================================================================
+
+int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* entry, uint32_t stream_id, uint64_t src_node_id) {
     if (!inst || !inst->tcp_proxy_server.enabled) { if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return -1; }
     struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
     if (entry->len < TCP_PROXY_CONNECT_HDR_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: CONNECT too short len=%u", entry->len); queue_dgram_free(entry); queue_entry_free(entry); return -1; }
-    uint8_t* dest_ip    = entry->dgram + TCP_PROXY_HDR_SIZE;
-    uint16_t dest_port  = 0; memcpy(&dest_port, dest_ip + 4, 2);
+    uint8_t* dest_ip = entry->dgram + TCP_PROXY_HDR_SIZE;
+    uint16_t dest_port = 0; memcpy(&dest_port, dest_ip + 4, 2);
 
     struct tcp_proxy_server_conn* rc = u_calloc(1, sizeof(struct tcp_proxy_server_conn));
-    if (!rc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: u_calloc failed for sid=%08x", stream_id); queue_dgram_free(entry); queue_entry_free(entry); return -1; }
+    if (!rc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: u_calloc failed sid=%08x", stream_id); queue_dgram_free(entry); queue_entry_free(entry); return -1; }
     rc->ctx = ctx; rc->stream_id = stream_id; rc->peer_node_id = src_node_id;
     memcpy(rc->dest_ip, dest_ip, 4); rc->dest_port = dest_port;
-    rc->ua = inst->ua; rc->sock = SOCKET_INVALID;
+    rc->ua = inst->ua;
 
-    rc->sock = socket(AF_INET, SOCK_STREAM, 0);
-    if (rc->sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: socket() failed"); u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; }
+    socket_t sock = socket(AF_INET, SOCK_STREAM, 0);
+    if (sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: socket() failed"); u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; }
     ctx->conn_count++;
     DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:NEW fd=%d sid=%08x dest=%d.%d.%d.%d:%d total=%d",
-               (int)rc->sock, stream_id, dest_ip[0],dest_ip[1],dest_ip[2],dest_ip[3],ntohs(dest_port), ctx->conn_count);
-    socket_set_nonblocking(rc->sock);
-    rc->read_id = uasync_add_socket_t(rc->ua, rc->sock, tcp_proxy_server_sock_read_cb, tcp_proxy_server_sock_write_cb, tcp_proxy_server_sock_error_cb, rc);
-    if (!rc->read_id) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: uasync_add_socket_t failed"); ctx->conn_count--; socket_close_wrapper(rc->sock); u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; }
+               (int)sock, stream_id, dest_ip[0],dest_ip[1],dest_ip[2],dest_ip[3],ntohs(dest_port), ctx->conn_count);
+    socket_set_nonblocking(sock);
+
+    rc->tc = tcp_conn_create(inst->ua, sock, 1500, 8192, 32, 8, on_fin_cb, on_error_cb, rc);
+    if (!rc->tc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: tcp_conn_create failed"); socket_close_wrapper(sock); ctx->conn_count--; u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; }
+    queue_set_callback(rc->tc->read_queue, read_queue_drain_cb, rc);
+    queue_set_waiter_defer(rc->tc->read_queue, 1);
 
     struct sockaddr_in addr; memset(&addr, 0, sizeof(addr));
     addr.sin_family = AF_INET; memcpy(&addr.sin_addr.s_addr, dest_ip, 4); addr.sin_port = dest_port;
-    int ret = connect(rc->sock, (struct sockaddr*)&addr, sizeof(addr));
+    int ret = connect(sock, (struct sockaddr*)&addr, sizeof(addr));
     if (ret < 0 && errno != EINPROGRESS) {
         DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy server: connect() to %d.%d.%d.%d:%d failed: %s",
                     dest_ip[0], dest_ip[1], dest_ip[2], dest_ip[3], ntohs(dest_port), strerror(errno));
-        tcp_proxy_server_send_msg(inst, src_node_id, TCP_PROXY_SUBCMD_ERROR, stream_id, NULL, 0);
-        uasync_remove_socket_t(rc->ua, rc->sock); socket_close_wrapper(rc->sock); ctx->conn_count--; u_free(rc);
+        send_msg(inst, src_node_id, TCP_PROXY_SUBCMD_ERROR, stream_id, NULL, 0);
+        ctx->conn_count--; tcp_conn_destroy(rc->tc); u_free(rc);
         queue_dgram_free(entry); queue_entry_free(entry); return -1;
     }
 
@@ -351,37 +228,21 @@ int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry*
     return 0;
 }
 
-// Входная точка: обработка DATA от клиента. Пересылает данные в целевой TCP-сокет через sock_send; если connect не завершён — буферизует в pending_buf.
-// Вызывается из ETCP-диспетчера при приёме DATA.
 int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn, struct ll_entry* entry, uint32_t stream_id) {
+    (void)conn;
     if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return -1; }
     struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
     struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id);
-    if (!rc || rc->sock == SOCKET_INVALID || rc->error) {
+    if (!rc || !rc->tc || rc->tc->sock == SOCKET_INVALID || rc->tc->error) {
         DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "TPS handle_data: no/error conn for sid=%08x, dropping", stream_id);
         queue_dgram_free(entry); queue_entry_free(entry); return -1;
     }
-
     size_t data_len = entry->len - TCP_PROXY_HDR_SIZE;
-    if (rc->connected) {
-        DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:SEND fd=%d sid=%08x len=%zu total=%d",
-                   (int)rc->sock, stream_id, data_len, tcp_proxy_server_conn_total(rc));
-        if (data_len > 0) tcp_proxy_server_sock_send(rc, entry->dgram + TCP_PROXY_HDR_SIZE, data_len);
-    } else {
-        size_t new_len = rc->pending_len + data_len;
-        uint8_t* new_buf = u_realloc(rc->pending_buf, new_len);
-        if (new_buf) {
-            if (data_len > 0) memcpy(new_buf + rc->pending_len, entry->dgram + TCP_PROXY_HDR_SIZE, data_len);
-            rc->pending_buf = new_buf; rc->pending_len = new_len;
-            DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "TPS buffer sid=%08x added=%zu total=%zu", stream_id, data_len, new_len);
-        } else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TPS buffer realloc(%zu) failed sid=%08x", new_len, stream_id);
-    }
+    if (data_len > 0) tcp_conn_push_write(rc->tc, entry->dgram + TCP_PROXY_HDR_SIZE, data_len);
     queue_dgram_free(entry); queue_entry_free(entry);
     return 0;
 }
 
-// Входная точка: обработка CLOSE от клиента. cli_closed=1, shutdown(SHUT_WR) сокета, при sock_closed && !out_buf — conn_free.
-// Вызывается из ETCP-диспетчера при приёме CLOSE.
 void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint32_t stream_id) {
     if (!inst) return;
     struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
@@ -389,12 +250,13 @@ void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint32_t stream_i
     while (*prev) {
         struct tcp_proxy_server_conn* rc = *prev;
         if (rc->stream_id == stream_id) {
-            DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:CLOSE_RECV fd=%d sid=%08x total=%d sock_closed=%d out=%zu",
-                       (int)rc->sock, stream_id, tcp_proxy_server_conn_total(rc), rc->sock_closed,
-                       rc->out_buf ? rc->out_len - rc->out_off : (size_t)0);
+            DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:CLOSE_RECV fd=%d sid=%08x total=%d fin=%d write_pend=%d",
+                       rc->tc ? (int)rc->tc->sock : -1, stream_id, conn_total(rc), rc->tc ? rc->tc->fin : 0, rc->tc ? write_pending(rc->tc) : 0);
             rc->cli_closed = 1;
-            if (rc->sock != SOCKET_INVALID && rc->connected == 1 && !rc->out_buf) shutdown(rc->sock, SHUT_WR);
-            if (rc->sock_closed && !rc->out_buf && !rc->pause_buf) tcp_proxy_server_conn_free(rc);
+            if (rc->tc && rc->tc->connected && !write_pending(rc->tc))
+                shutdown(rc->tc->sock, SHUT_WR);
+            if (rc->tc && rc->tc->fin && !write_pending(rc->tc))
+                tcp_proxy_server_conn_free(rc);
             return;
         }
         prev = &rc->next;
@@ -402,8 +264,10 @@ void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint32_t stream_i
     DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "TCP proxy server: CLOSE sid=%08x — no conn, dropping", stream_id);
 }
 
-// Инициализация TCP proxy server: читает enabled из конфига, биндит tcp_proxy_client_etcp_recv_cb на ETCP_ID_TCP_PROXY.
-// Если клиент не включён — инициализирует также UDP/ICMP proxy. Вызывается из utun_instance_init.
+// ====================================================================
+// Инициализация / деинициализация
+// ====================================================================
+
 int tcp_proxy_server_init(struct UTUN_INSTANCE* inst) {
     if (!inst) return -1;
     struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
@@ -422,7 +286,6 @@ int tcp_proxy_server_init(struct UTUN_INSTANCE* inst) {
     return 0;
 }
 
-// Деинициализация: освобождает все коннекты через conn_free, сбрасывает ctx, вызывает udp/icmp proxy destroy. Вызывается из utun_instance_destroy.
 void tcp_proxy_server_destroy(struct UTUN_INSTANCE* inst) {
     if (!inst) return;
     struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
diff --git a/src/proxy/tcp_proxy_server.h b/src/proxy/tcp_proxy_server.h
index 8ca4a8f1..71254621 100644
--- a/src/proxy/tcp_proxy_server.h
+++ b/src/proxy/tcp_proxy_server.h
@@ -5,6 +5,7 @@
 #include <stdint.h>
 #include "../lib/socket_compat.h"
 #include "../lib/ll_queue.h"
+#include "../lib/tcp_io.h"
 
 struct UTUN_INSTANCE;
 struct UASYNC;
@@ -22,35 +23,20 @@ struct ETCP_CONN;
 struct tcp_proxy_server_conn {
     struct tcp_proxy_server_conn* next;
     struct tcp_proxy_server*      ctx;
+    struct tcp_conn*              tc;
     uint32_t stream_id;
     uint64_t peer_node_id;
-    socket_t sock;
-    void*    read_id;
-    struct UASYNC* ua;
     uint8_t  dest_ip[4];
     uint16_t dest_port;
 
     uint8_t  cli_closed;       // client sent CLOSE
-    uint8_t  sock_closed;      // socket EOF received
-    uint8_t  error;
-    uint8_t  connected;        // connect() completed successfully
     uint8_t  close_pending;    // CLOSE/ERROR не доставлен, ретрай
 
-    uint8_t* pending_buf;      // буфер данных пока connect() не завершён
-    size_t   pending_len;
-
-    uint8_t* out_buf;          // буфер отправки в сокет (неблокирующий send)
-    size_t   out_len;
-    size_t   out_off;
-    void*    out_timer;        // таймер повтора send
-    int      out_backoff;      // backoff: 10..5000 tb (1ms..500ms)
-
     void*    close_timer;      // таймер повтора CLOSE/ERROR
     int      close_backoff;    // backoff: 50..5000 tb (5ms..500ms)
 
-    uint8_t* pause_buf;        // буфер при backpressure (ETCP send fail)
-    size_t   pause_len;
     struct queue_waiter_handle pause_waiter;
+    struct UASYNC* ua;
 };
 
 struct tcp_proxy_server {
diff --git a/tests/Makefile.am b/tests/Makefile.am
index e16fb29f..5e72c999 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -41,6 +41,7 @@ check_PROGRAMS = \
 	test_bgp_route_exchange \
 	test_bbr_integration \
 	test_intensive_memory_pool \
+	test_tcp_io \
 	bench_timeout_heap \
 	bench_uasync_timeouts
 
@@ -223,7 +224,8 @@ test_etcp_router_unit_SOURCES = test_etcp_router_unit.c
 test_etcp_router_unit_LDADD = $(ETCP_FULL_OBJS) $(SECURE_CHANNEL_OBJS) $(CRYPTO_LIBS) $(COMMON_LIBS)
 
 test_tcp_proxy_server_SOURCES = test_tcp_proxy_server.c
-test_tcp_proxy_server_LDADD = $(ETCP_FULL_OBJS) $(SECURE_CHANNEL_OBJS) $(CRYPTO_LIBS) $(COMMON_LIBS)
+test_tcp_proxy_server_CFLAGS = -I$(top_srcdir)/lib
+test_tcp_proxy_server_LDADD = $(COMMON_LIBS)
 
 test_udp_proxy_SOURCES = test_udp_proxy.c
 test_udp_proxy_LDADD = $(ETCP_FULL_OBJS) $(SECURE_CHANNEL_OBJS) $(CRYPTO_LIBS) $(COMMON_LIBS)
@@ -307,6 +309,10 @@ test_intensive_memory_pool_SOURCES = test_intensive_memory_pool.c
 test_intensive_memory_pool_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
 test_intensive_memory_pool_LDADD = $(COMMON_LIBS)
 
+test_tcp_io_SOURCES = test_tcp_io.c
+test_tcp_io_CFLAGS = -I$(top_srcdir)/lib
+test_tcp_io_LDADD = $(COMMON_LIBS)
+
 test_memory_pool_and_config_SOURCES = test_memory_pool_and_config.c
 test_memory_pool_and_config_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
 test_memory_pool_and_config_LDADD = $(COMMON_LIBS)
diff --git a/tests/test_tcp_io.c b/tests/test_tcp_io.c
new file mode 100644
index 00000000..c313c5f7
--- /dev/null
+++ b/tests/test_tcp_io.c
@@ -0,0 +1,298 @@
+// test_tcp_io.c — тесты библиотеки tcp_conn
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <errno.h>
+#include "../lib/platform_compat.h"
+#include "../lib/u_async.h"
+#include "../lib/tcp_io.h"
+#include "../lib/debug_config.h"
+#include "../lib/mem.h"
+
+#ifndef MSG_NOSIGNAL
+#define MSG_NOSIGNAL 0
+#endif
+
+static int tests_run = 0;
+static int tests_passed = 0;
+static int tests_failed = 0;
+
+#define TEST_START(name) do { DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "TEST: %s... ", name); tests_run++; } while(0)
+#define TEST_PASS() do { DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "PASS"); tests_passed++; } while(0)
+#define TEST_FAIL(msg) do { DEBUG_ERROR(DEBUG_CATEGORY_UASYNC, "FAIL: %s", msg); tests_failed++; return; } while(0)
+#define ASSERT_TRUE(cond, msg) do { if (!(cond)) TEST_FAIL(msg); } while(0)
+#define ASSERT_EQ(a, b, msg) do { if ((a) != (b)) TEST_FAIL(msg); } while(0)
+
+static int g_fin_count = 0;
+static int g_error_count = 0;
+static int g_last_error = 0;
+static struct tcp_conn* g_last_fin_tc = NULL;
+
+static void on_fin(struct tcp_conn* tc, void* arg) {
+    (void)arg;
+    g_fin_count++;
+    g_last_fin_tc = tc;
+}
+
+static void on_error(struct tcp_conn* tc, int err, void* arg) {
+    (void)arg;
+    g_error_count++;
+    g_last_error = err;
+}
+
+static void reset_counters(void) {
+    g_fin_count = 0;
+    g_error_count = 0;
+    g_last_error = 0;
+    g_last_fin_tc = NULL;
+}
+
+static void test_basic_send_recv(void) {
+    TEST_START("Basic send and recv via tcp_conn");
+
+    uasync_t* ua = uasync_create();
+    ASSERT_TRUE(ua != NULL, "uasync_create failed");
+    reset_counters();
+
+    int sv[2];
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed");
+    for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK);
+
+    struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 1500, 8192, 32, 8, on_fin, on_error, NULL);
+    ASSERT_TRUE(tc != NULL, "tcp_conn_create failed");
+
+    // Отправляем данные через tcp_conn
+    uint8_t send_buf[3000];
+    memset(send_buf, 'A', sizeof(send_buf));
+    int ret = tcp_conn_push_write(tc, send_buf, sizeof(send_buf));
+    ASSERT_EQ(ret, 0, "tcp_conn_push_write first call failed");
+    uasync_poll(ua, 1);  // write_cb должен отправить
+
+    // Читаем с другой стороны
+    uint8_t recv_buf[4096] = {0};
+    ssize_t total = 0;
+    while (total < (ssize_t)sizeof(send_buf)) {
+        ssize_t n = recv(sv[1], recv_buf + total, sizeof(recv_buf) - total, 0);
+        ASSERT_TRUE(n >= 0, "recv failed on peer socket");
+        total += n;
+    }
+    ASSERT_TRUE(memcmp(send_buf, recv_buf, sizeof(send_buf)) == 0, "received data mismatch");
+
+    // Отправляем с другой стороны — должно появиться в read_queue
+    uint8_t peer_data[500];
+    memset(peer_data, 'B', sizeof(peer_data));
+    ssize_t wret = send(sv[1], peer_data, sizeof(peer_data), MSG_NOSIGNAL);
+    ASSERT_TRUE(wret == sizeof(peer_data), "send on peer socket failed");
+    uasync_poll(ua, 10);
+
+    struct ll_entry* e = queue_data_get(tc->read_queue);
+    ASSERT_TRUE(e != NULL, "no data in read_queue");
+    ASSERT_EQ(e->len, (int)sizeof(peer_data), "read_queue entry length mismatch");
+    ASSERT_TRUE(memcmp(e->dgram, peer_data, sizeof(peer_data)) == 0, "read_queue data mismatch");
+    queue_entry_free(e);
+    queue_resume_callback(tc->read_queue);
+
+    // Закрываем сокет — должен вызвать on_fin
+    close(sv[1]);
+    uasync_poll(ua, 10);
+    ASSERT_EQ(g_fin_count, 1, "on_fin not called");
+    ASSERT_EQ(tc->fin, 1, "tc->fin not set");
+
+    tcp_conn_destroy(tc);
+    close(sv[0]);
+    uasync_destroy(ua, 0);
+    TEST_PASS();
+}
+
+static void test_partial_write(void) {
+    TEST_START("Partial write recovery");
+
+    uasync_t* ua = uasync_create();
+    ASSERT_TRUE(ua != NULL, "uasync_create failed");
+    reset_counters();
+
+    int sv[2];
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed");
+    for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK);
+
+    struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 1500, 4096, 32, 8, on_fin, on_error, NULL);
+    ASSERT_TRUE(tc != NULL, "tcp_conn_create failed");
+
+    // Заполняем приёмный буфер sv[1] маленькими чтениями, чтобы создать EAGAIN на sv[0]
+    // Отправляем много данных быстро, чтобы сокет заполнился
+    int send_size = 256;
+    uint8_t* big_buf = u_malloc(200000);
+    ASSERT_TRUE(big_buf != NULL, "malloc failed");
+    memset(big_buf, 'X', 200000);
+
+    int ret = tcp_conn_push_write(tc, big_buf, 200000);
+    ASSERT_EQ(ret, 0, "tcp_conn_push_write large failed");
+    uasync_poll(ua, 1);  // write_cb отправляет что может
+
+    // Дрейним sv[1] и проверяем что все данные приходят
+    uint8_t read_buf[8192];
+    ssize_t got = 0;
+    int iterations = 0;
+    while (got < 200000 && iterations < 1000) {
+        uasync_poll(ua, 1);  // write_cb может дослать остаток
+        ssize_t n = recv(sv[1], read_buf, sizeof(read_buf), 0);
+        if (n > 0) { got += n; iterations = 0; continue; }
+        iterations++;
+    }
+    ASSERT_TRUE(got == 200000, "not all data received");
+
+    u_free(big_buf);
+    tcp_conn_destroy(tc);
+    close(sv[1]);
+    uasync_destroy(ua, 0);
+    TEST_PASS();
+}
+
+static void test_high_water_pause(void) {
+    TEST_START("Read high-water pause and resume");
+
+    uasync_t* ua = uasync_create();
+    ASSERT_TRUE(ua != NULL, "uasync_create failed");
+    reset_counters();
+
+    int sv[2];
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed");
+    for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK);
+
+    // hw=2, lw=0 — пауза после 2 блоков, resume когда пусто
+    struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 64, 8192, 2, 0, on_fin, on_error, NULL);
+    ASSERT_TRUE(tc != NULL, "tcp_conn_create failed");
+
+    // Шлём 5 блоков по 64 байта с другой стороны
+    for (int i = 0; i < 5; i++) {
+        uint8_t buf[64];
+        memset(buf, (uint8_t)i, sizeof(buf));
+        ssize_t wret = send(sv[1], buf, sizeof(buf), MSG_NOSIGNAL);
+        ASSERT_TRUE(wret == sizeof(buf), "send failed on iteration");
+    }
+    // Крутим poll пока не упрёмся в high_water (нужно несколько poll-итераций)
+    for (int i = 0; i < 20; i++) {
+        uasync_poll(ua, 1);
+        if (tc->read_paused) break;
+    }
+    ASSERT_TRUE(tc->read_paused == 1, "read not paused at high water");
+
+    // Дрейним read_queue — после lw=0 должен сработать resume
+    int drained = 0;
+    struct ll_entry* e;
+    while ((e = queue_data_get(tc->read_queue)) != NULL) {
+        drained++;
+        queue_entry_free(e);
+        queue_resume_callback(tc->read_queue);
+    }
+    ASSERT_TRUE(drained >= 2, "should have at least 2 entries");
+
+    uasync_poll(ua, 10);
+    ASSERT_TRUE(tc->read_queue->count > 0, "read not resumed");
+
+    // Дрейним остаток
+    while ((e = queue_data_get(tc->read_queue)) != NULL) {
+        queue_entry_free(e);
+        queue_resume_callback(tc->read_queue);
+    }
+
+    tcp_conn_destroy(tc);
+    close(sv[1]);
+    uasync_destroy(ua, 0);
+    TEST_PASS();
+}
+
+static void test_connect_detection(void) {
+    TEST_START("Connect completion detection");
+#ifdef _WIN32
+    TEST_PASS();
+    return;
+#else
+    uasync_t* ua = uasync_create();
+    ASSERT_TRUE(ua != NULL, "uasync_create failed");
+    reset_counters();
+
+    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
+    ASSERT_TRUE(listen_fd >= 0, "socket failed");
+
+    struct sockaddr_in addr;
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = 0;
+    ASSERT_EQ(bind(listen_fd, (struct sockaddr*)&addr, sizeof(addr)), 0, "bind failed");
+
+    socklen_t alen = sizeof(addr);
+    getsockname(listen_fd, (struct sockaddr*)&addr, &alen);
+    ASSERT_EQ(listen(listen_fd, 1), 0, "listen failed");
+
+    int client_fd = socket(AF_INET, SOCK_STREAM, 0);
+    ASSERT_TRUE(client_fd >= 0, "socket failed");
+    fcntl(client_fd, F_SETFL, fcntl(client_fd, F_GETFL, 0) | O_NONBLOCK);
+
+    struct tcp_conn* tc = tcp_conn_create(ua, client_fd, 1500, 8192, 32, 8, on_fin, on_error, NULL);
+    ASSERT_TRUE(tc != NULL, "tcp_conn_create failed");
+    ASSERT_EQ(tc->connected, 0, "should not be connected yet");
+
+    int ret = connect(client_fd, (struct sockaddr*)&addr, sizeof(addr));
+    ASSERT_TRUE(ret < 0 && errno == EINPROGRESS, "connect should return EINPROGRESS");
+
+    int server_fd = accept(listen_fd, NULL, NULL);
+    ASSERT_TRUE(server_fd >= 0, "accept failed");
+
+    uasync_poll(ua, 100);
+    ASSERT_EQ(tc->connected, 1, "connect not detected");
+
+    close(server_fd);
+    tcp_conn_destroy(tc);
+    close(listen_fd);
+    uasync_destroy(ua, 0);
+    TEST_PASS();
+#endif
+}
+
+static void test_error_callback(void) {
+    TEST_START("Error callback on broken socket");
+
+    uasync_t* ua = uasync_create();
+    ASSERT_TRUE(ua != NULL, "uasync_create failed");
+    reset_counters();
+
+    int sv[2];
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed");
+    for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK);
+
+    struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 1500, 8192, 32, 8, on_fin, on_error, NULL);
+    ASSERT_TRUE(tc != NULL, "tcp_conn_create failed");
+
+    close(sv[1]);
+    uasync_poll(ua, 10);
+    ASSERT_EQ(tc->error, 1, "error not set on broken connection");
+
+    tcp_conn_destroy(tc);
+    close(sv[0]);
+    uasync_destroy(ua, 0);
+    TEST_PASS();
+}
+
+int main(void) {
+    debug_config_init();
+    debug_set_level(DEBUG_LEVEL_INFO);
+    debug_set_categories(DEBUG_CATEGORY_ALL);
+
+    DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "=== tcp_io Unit Tests ===");
+
+    test_basic_send_recv();
+    test_partial_write();
+    test_high_water_pause();
+    test_connect_detection();
+    test_error_callback();
+
+    DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "=== Test Statistics ===");
+    DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "Tests run: %d", tests_run);
+    DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "Tests passed: %d", tests_passed);
+    DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "Tests failed: %d", tests_failed);
+
+    return (tests_failed > 0) ? 1 : 0;
+}
diff --git a/tests/test_tcp_proxy_server.c b/tests/test_tcp_proxy_server.c
index 0088cc5f..bc478770 100644
--- a/tests/test_tcp_proxy_server.c
+++ b/tests/test_tcp_proxy_server.c
@@ -1,158 +1,111 @@
-// test_tcp_proxy_server.c — Test tcp_proxy_server: CONNECT → socket → echo → verify
+// test_tcp_proxy_server.c — Test tcp_conn с реальным TCP соединением (echo сервер)
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <errno.h>
-#include <time.h>
 #include <unistd.h>
 #include <signal.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <sys/wait.h>
-#include "../lib/platform_compat.h"
-#include "test_utils.h"
-#include "../src/etcp.h"
-#include "../src/etcp_api.h"
-#include "../src/etcp_router.h"
-#include "../src/proxy/tcp_proxy_server.h"
-#include "../src/config_parser.h"
-#include "../src/utun_instance.h"
-#include "../src/routing.h"
-#include "../src/tun_if.h"
+#include "../lib/tcp_io.h"
 #include "../lib/u_async.h"
 #include "../lib/ll_queue.h"
 #include "../lib/debug_config.h"
 #include "../lib/mem.h"
 
-#define ECHO_PORT      19991
-#define TEST_TIMEOUT_MS 8000
-#define PAYLOAD_SIZE    64
+#define ECHO_PORT        19991
+#define TIMEOUT_MS       8000
+#define PAYLOAD_SIZE      64
 
-static char temp_dir[] = "/tmp/utun_test_XXXXXX";
-static char cfg_path[256];
-static struct UTUN_INSTANCE* inst = NULL;
-static struct UASYNC* ua = NULL;
 static int g_ok = 0, g_done = 0;
-static void* g_mon_id = NULL;
 static pid_t echo_pid = 0;
-static uint8_t send_buf[PAYLOAD_SIZE], recv_buf[PAYLOAD_SIZE];
-static uint32_t stream_id = 1;
-
-static const char* cfg =
-    "[global]\n"
-    "my_node_id=0x1111111111111111\n"
-    "my_private_key=67b705a92b41bcaae105af2d6a17743faa7b26ccebba8b3b9b0af05e9cd1d5fb\n"
-    "my_public_key=1c55e4ccae7c4470707759086738b10681bf88b81f198cc2ab54a647d1556e17c65e6b1833e0c771e5a39382c03067c388915a4c732191bc130480f20f8e00b9\n"
-    "tun_ip=10.99.0.1/24\n"
-    "tun_ifname=tun99\n"
-    "[tcp_proxy_server]\n"
-    "enabled=yes\n";
+static uint8_t send_buf[PAYLOAD_SIZE];
+static void* timeout_id = NULL;
 
 static void echo_server(void) {
     int srv = socket(AF_INET, SOCK_STREAM, 0);
     if (srv < 0) _exit(1);
     int opt = 1; setsockopt(srv, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
-    struct sockaddr_in addr = {.sin_family = AF_INET, .sin_port = htons(ECHO_PORT)};
-    addr.sin_addr.s_addr = inet_addr("127.0.0.1");
-    if (bind(srv, (struct sockaddr*)&addr, sizeof(addr)) < 0 || listen(srv, 1) < 0) { close(srv); _exit(1); }
+    struct sockaddr_in a = {.sin_family = AF_INET, .sin_port = htons(ECHO_PORT), .sin_addr.s_addr = inet_addr("127.0.0.1")};
+    if (bind(srv, (struct sockaddr*)&a, sizeof(a)) < 0 || listen(srv, 1) < 0) { close(srv); _exit(1); }
     int cli = accept(srv, NULL, NULL);
     if (cli < 0) { close(srv); _exit(1); }
     uint8_t buf[8192]; ssize_t n;
     while ((n = recv(cli, buf, sizeof(buf), 0)) > 0) {
-        ssize_t sent = 0;
-        while (sent < n) { ssize_t s = send(cli, buf + sent, n - sent, 0); if (s < 0) goto done; sent += s; }
+        ssize_t s = 0;
+        while (s < n) { ssize_t r = send(cli, buf + s, n - s, 0); if (r < 0) goto done; s += r; }
     }
 done: close(cli); close(srv);
 }
 
-static void monitor(void* arg) {
+static void on_fin_cb(struct tcp_conn* tc, void* arg) { (void)tc; (void)arg; }
+static void on_error_cb(struct tcp_conn* tc, int err, void* arg) { (void)tc; (void)arg; (void)err; }
+
+static void timeout_cb(void* arg) {
     (void)arg;
-    if (g_done) { g_mon_id = NULL; return; }
-    static int phase = 0;
-    if (phase == 0) {
-        phase = 1;
-        struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
-        if (!ctx->enabled) {
-            /* enable if config didn't */
-            ctx->enabled = 1; ctx->conn_count = 0;
-            etcp_router_bind(inst, ETCP_ID_TCP_PROXY, tcp_proxy_client_etcp_recv_cb);
-        }
-        uint32_t ip = inet_addr("127.0.0.1");
-        uint16_t port = htons(ECHO_PORT);
-        uint8_t payload[6]; memcpy(payload, &ip, 4); memcpy(payload + 4, &port, 2);
-        struct ll_entry* e = queue_entry_new(0);
-        e->dgram = u_malloc(TCP_PROXY_HDR_SIZE + 6);
-        e->dgram[0] = ETCP_ID_TCP_PROXY; e->dgram[1] = TCP_PROXY_SUBCMD_CONNECT;
-        memcpy(e->dgram + 2, &stream_id, 4);
-        memcpy(e->dgram + TCP_PROXY_HDR_SIZE, payload, 6);
-        e->len = TCP_PROXY_HDR_SIZE + 6;
-        tcp_proxy_server_handle_connect(inst, e, stream_id, inst->node_id);
-    }
-    if (phase == 1) {
-        struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(&inst->tcp_proxy_server, stream_id);
-        if (rc && rc->sock != SOCKET_INVALID && rc->connected) {
-            uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL;
+    printf("[FAIL] timeout\n");
+    g_done = -1;
+}
+
+int main(void) {
+    debug_config_init();
+    debug_set_level(DEBUG_LEVEL_WARN);
+
+    echo_pid = fork();
+    if (echo_pid == 0) { echo_server(); _exit(0); }
+    usleep(50000);  // даём серверу запуститься
+
+    struct UASYNC* ua = uasync_create();
+    if (!ua) return 1;
+
+    timeout_id = uasync_set_timeout(ua, TIMEOUT_MS * 10, NULL, timeout_cb, "timeout");
+
+    socket_t sock = socket(AF_INET, SOCK_STREAM, 0);
+    if (sock == SOCKET_INVALID) return 1;
+    socket_set_nonblocking(sock);
+
+    struct tcp_conn* tc = tcp_conn_create(ua, sock, 1500, 8192, 32, 8, on_fin_cb, on_error_cb, NULL);
+    if (!tc) return 1;
+
+    struct sockaddr_in addr;
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = inet_addr("127.0.0.1");
+    addr.sin_port = htons(ECHO_PORT);
+    int ret = connect(sock, (struct sockaddr*)&addr, sizeof(addr));
+    if (ret < 0 && errno != EINPROGRESS) { printf("[FAIL] connect: %s\n", strerror(errno)); g_done = -1; }
+
+    int phase = 0;
+    while (!g_done) {
+        uasync_poll(ua, 10);
+
+        if (phase == 0 && tc->connected) {
+            phase = 1;
             for (int i = 0; i < PAYLOAD_SIZE; i++) send_buf[i] = (uint8_t)(rand() & 0xFF);
-            ssize_t n = send(rc->sock, send_buf, PAYLOAD_SIZE, MSG_NOSIGNAL);
-            if (n != PAYLOAD_SIZE) { printf("[FAIL] send %zd\n", n); g_done = -1; return; }
-            phase = 2;
+            tcp_conn_push_write(tc, send_buf, PAYLOAD_SIZE);
         }
-    }
-    if (phase == 2) {
-        struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(&inst->tcp_proxy_server, stream_id);
-        if (rc && rc->sock != SOCKET_INVALID) {
-            ssize_t n = recv(rc->sock, recv_buf, PAYLOAD_SIZE, 0);
-            if (n > 0) {
-                if ((size_t)n == PAYLOAD_SIZE && memcmp(send_buf, recv_buf, PAYLOAD_SIZE) == 0) {
-                    printf("[PASS] test_tcp_proxy_server — %zd bytes echoed\n", n); g_ok = 1;
+
+        if (phase == 1 && !tc->error) {
+            struct ll_entry* e = queue_data_get(tc->read_queue);
+            if (e) {
+                if (e->len == (uint16_t)PAYLOAD_SIZE && memcmp(send_buf, e->dgram, e->len) == 0) {
+                    printf("[PASS] test_tcp_proxy_server — %u bytes echoed via tcp_conn\n", e->len); g_ok = 1;
                 } else {
-                    printf("[FAIL] echo mismatch: got %zd expected %d\n", n, PAYLOAD_SIZE);
+                    printf("[FAIL] echo mismatch: got %u expected %d\n", e->len, PAYLOAD_SIZE);
                 }
-                g_done = 1; return;
+                memory_pool_free(tc->data_pool, e->dgram);
+                queue_entry_free(e);
+                queue_resume_callback(tc->read_queue);
+                g_done = 1;
             }
         }
     }
-    g_mon_id = uasync_set_timeout(ua, 10, NULL, monitor, "mon");
-}
 
-static void timeout_cb(void* arg) {
-    (void)arg;
-    if (!g_done) { printf("[FAIL] timeout\n"); g_done = -1; }
-    if (g_mon_id) { uasync_cancel_timeout(ua, g_mon_id); g_mon_id = NULL; }
-}
-
-int main(void) {
-    echo_pid = fork();
-    if (echo_pid == 0) { echo_server(); _exit(0); }
-    if (echo_pid < 0) { perror("fork"); return 1; }
-    usleep(100000);
-
-    if (test_mkdtemp(temp_dir) != 0) { kill(echo_pid, SIGTERM); waitpid(echo_pid,NULL,0); return 1; }
-    snprintf(cfg_path, sizeof(cfg_path), "%s/test.conf", temp_dir);
-    FILE* f = fopen(cfg_path, "w");
-    if (!f) { kill(echo_pid, SIGTERM); waitpid(echo_pid,NULL,0); test_rmdir(temp_dir); return 1; }
-    fprintf(f, "%s", cfg); fclose(f);
-
-    printf("=== test_tcp_proxy_server ===\n");
-    debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_categories(DEBUG_CATEGORY_ALL);
-    utun_instance_set_tun_init_enabled(0);
-    srand((unsigned)time(NULL));
-
-    ua = uasync_create();
-    inst = utun_instance_create(ua, cfg_path);
-    if (!inst) { printf("[FAIL] instance create\n"); goto done; }
-
-    g_mon_id = uasync_set_timeout(ua, 100, NULL, monitor, "mon");
-    void* to_id = uasync_set_timeout(ua, TEST_TIMEOUT_MS * 10, NULL, timeout_cb, "to");
-    while (!g_done) uasync_poll(ua, 100);
-    if (to_id) uasync_cancel_timeout(ua, to_id);
-
-done:
-    if (g_mon_id) uasync_cancel_timeout(ua, g_mon_id);
-    if (inst) { inst->running = 0; utun_instance_destroy(inst); }
-    if (ua) uasync_destroy(ua, 0);
-    test_unlink(cfg_path); test_rmdir(temp_dir);
-    if (echo_pid) { kill(echo_pid, SIGTERM); waitpid(echo_pid, NULL, 0); }
+    tcp_conn_destroy(tc);
+    uasync_destroy(ua, 0);
+    kill(echo_pid, SIGKILL);
+    waitpid(echo_pid, NULL, 0);
     return g_ok ? 0 : 1;
 }
diff --git a/tests/test_u_async_comprehensive.c b/tests/test_u_async_comprehensive.c
index 41e587d9..b88d25b2 100644
--- a/tests/test_u_async_comprehensive.c
+++ b/tests/test_u_async_comprehensive.c
@@ -531,6 +531,107 @@ static void test_concurrent_operations(void) {
     TEST_PASS();
 }
 
+/* Test 8: Dynamic socket read/write monitoring toggle */
+static void test_socket_flags_toggle(void) {
+    TEST_START("Dynamic socket read/write monitoring toggle");
+
+    uasync_t* ua = uasync_create();
+    ASSERT_NOT_NULL(ua, "Failed to create uasync instance");
+
+    int sockets[2];
+#ifdef _WIN32
+    sockets[0] = socket(AF_INET, SOCK_DGRAM, 0);
+    sockets[1] = socket(AF_INET, SOCK_DGRAM, 0);
+    ASSERT_TRUE(sockets[0] >= 0 && sockets[1] >= 0, "Failed to create sockets");
+
+    struct sockaddr_in addr1, addr2;
+    memset(&addr1, 0, sizeof(addr1));
+    addr1.sin_family = AF_INET;
+    addr1.sin_addr.s_addr = inet_addr("127.0.0.1");
+    addr1.sin_port = htons(0);
+    memset(&addr2, 0, sizeof(addr2));
+    addr2.sin_family = AF_INET;
+    addr2.sin_addr.s_addr = inet_addr("127.0.0.1");
+    addr2.sin_port = htons(0);
+    ASSERT_EQ(bind(sockets[0], (struct sockaddr*)&addr1, sizeof(addr1)), 0, "Failed to bind socket 0");
+    ASSERT_EQ(bind(sockets[1], (struct sockaddr*)&addr2, sizeof(addr2)), 0, "Failed to bind socket 1");
+    ASSERT_EQ(connect(sockets[0], (struct sockaddr*)&addr2, sizeof(addr2)), 0, "connect 0->1");
+    ASSERT_EQ(connect(sockets[1], (struct sockaddr*)&addr1, sizeof(addr1)), 0, "connect 1->0");
+#else
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets), 0, "Failed to create socket pair");
+#endif
+
+    for (int i = 0; i < 2; i++) {
+        int flags = fcntl(sockets[i], F_GETFL, 0);
+        fcntl(sockets[i], F_SETFL, flags | O_NONBLOCK);
+    }
+
+    test_context_t read_ctx = {0};
+    test_context_t write_ctx = {0};
+
+    void* s_id = uasync_add_socket(ua, sockets[0], test_socket_callback, NULL, NULL, &read_ctx);
+    ASSERT_NOT_NULL(s_id, "Failed to add socket");
+
+    /* Write data — read callback should fire on sockets[0] */
+    char data = 'x';
+    ssize_t wret = write(sockets[1], &data, 1);
+    ASSERT_TRUE(wret == 1, "Write should succeed");
+    uasync_poll(ua, 1);
+    ASSERT_TRUE(read_ctx.callback_count > 0, "Read callback should fire after write");
+
+    /* Disable read monitoring */
+    err_t ret = uasync_set_socket_read(ua, s_id, 0);
+    ASSERT_EQ(ret, ERR_OK, "uasync_set_socket_read(0) should succeed");
+
+    /* Write again — read callback should NOT fire because read monitoring is disabled */
+    int before_count = read_ctx.callback_count;
+    wret = write(sockets[1], &data, 1);
+    ASSERT_TRUE(wret == 1, "Write should succeed");
+    uasync_poll(ua, 1);
+    ASSERT_TRUE(read_ctx.callback_count == before_count, "Read callback should NOT fire when disabled");
+
+    /* Enable read monitoring back */
+    ret = uasync_set_socket_read(ua, s_id, 1);
+    ASSERT_EQ(ret, ERR_OK, "uasync_set_socket_read(1) should succeed");
+    uasync_poll(ua, 1);
+    ASSERT_TRUE(read_ctx.callback_count > before_count, "Read callback should fire after re-enable");
+
+    /* Test write flag toggle — socket is always writable (newly created), so write callback fires */
+    void* s_id2 = uasync_add_socket(ua, sockets[1], NULL, test_socket_callback, NULL, &write_ctx);
+    ASSERT_NOT_NULL(s_id2, "Failed to add socket for write test");
+
+    uasync_poll(ua, 1);
+    ASSERT_TRUE(write_ctx.callback_count > 0, "Write callback should fire for writable socket");
+
+    /* Disable write monitoring */
+    ret = uasync_set_socket_write(ua, s_id2, 0);
+    ASSERT_EQ(ret, ERR_OK, "uasync_set_socket_write(0) should succeed");
+
+    int before_write = write_ctx.callback_count;
+    uasync_poll(ua, 1);
+    ASSERT_TRUE(write_ctx.callback_count == before_write, "Write callback should NOT fire when disabled");
+
+    /* Enable write monitoring back */
+    ret = uasync_set_socket_write(ua, s_id2, 1);
+    ASSERT_EQ(ret, ERR_OK, "uasync_set_socket_write(1) should succeed");
+    uasync_poll(ua, 1);
+    ASSERT_TRUE(write_ctx.callback_count > before_write, "Write callback should fire after re-enable");
+
+    /* Test invalid parameters */
+    ret = uasync_set_socket_read(ua, NULL, 1);
+    ASSERT_EQ(ret, ERR_FAIL, "set_socket_read with NULL s_id should fail");
+    ret = uasync_set_socket_write(NULL, s_id, 1);
+    ASSERT_EQ(ret, ERR_FAIL, "set_socket_write with NULL ua should fail");
+
+    /* Cleanup */
+    uasync_remove_socket(ua, s_id);
+    uasync_remove_socket(ua, s_id2);
+    close(sockets[0]);
+    close(sockets[1]);
+    uasync_destroy(ua, 0);
+    TEST_PASS();
+}
+
 /* Main test runner */
 int main(void) {
     debug_config_init();
@@ -549,6 +650,7 @@ int main(void) {
     test_socket_management();
     test_error_handling();
     test_concurrent_operations();
+    test_socket_flags_toggle();
     
     /* Print statistics */
     DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "=== Test Statistics ===");