From 57d28376b6b0c6a280a2081deb31750cd91fcd65 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Tue, 20 Oct 2026 16:58:25 +0200
Subject: [PATCH] Use independent shared viewers in the Android native bridge

---
 tests/test_android_stream.c                   |  10 +-
 .../libutun_lite/stream_bridge.c              | 443 +++++-------------
 .../libutun_lite/stream_bridge.h              |   2 +-
 3 files changed, 132 insertions(+), 323 deletions(-)

diff --git a/tests/test_android_stream.c b/tests/test_android_stream.c
index c390fbda..e77679e9 100644
--- a/tests/test_android_stream.c
+++ b/tests/test_android_stream.c
@@ -41,7 +41,7 @@ static void chat_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data
 
 static uint8_t *wait_event(uint64_t id, int kind, size_t *size) {
     for (unsigned i = 0; i < 1000; i++) {
-        if (id == 2) CHECK(android_stream_read_request(id, kind == 5 ? 1000000 : -1, 1, 80000) == 0);
+        if (id == 2 || id == 7) CHECK(android_stream_read_request(id, kind == 5 ? 1000000 : -1, 1, 80000) == 0);
         uint8_t *data = android_stream_pull(id, size);
         if (data && data[0] == kind) return data;
         u_free(data); test_clock_poll(ua, 100);
@@ -98,6 +98,10 @@ int main(void) {
     uint8_t *error = wait_event(5, 6, &size); CHECK(error[1] == STREAM_CLOSE_SOURCE_CHANGED); u_free(error);
     CHECK(android_stream_open(2, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0);
     u_free(wait_event(2, 4, &size));
+    CHECK(android_stream_open(7, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0);
+    u_free(wait_event(7, 4, &size));
+    CHECK(android_stream_open(8, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0);
+    android_stream_stop(8);
     CHECK(source_events == 2); /* Подписка не меняет каталог публикаций. */
     for (unsigned i = 0; i < sizeof(video); i++) video[i] = (uint8_t)i;
     CHECK(feed(1, 1, 100000, 0, video, sizeof(video)) == 0);
@@ -114,6 +118,10 @@ int main(void) {
         u_free(output); test_clock_poll(ua, 100);
     }
     CHECK(video_seen == 1 && audio_seen == 1);
+    output = wait_event(7, 1, &size);
+    CHECK(stream_read_be(output + 8, 8) == 100000 && !memcmp(output + ANDROID_STREAM_HEADER_SIZE, video, sizeof(video)));
+    u_free(output); android_stream_stop(7); test_clock_poll(ua, 100);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android independent viewers: delayed reader consumed its own frame; primary remains active");
     /* Commit новой revision удаляет старое видео, сохраняет аудио/epoch и не требует нового owner. */
     CHECK(feed(1, 1, 200000, 0, video, sizeof(video)) == 1);
     CHECK(feed(1, 0, 120000, 0, (const uint8_t *)"ops", 3) == 1);
diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c
index 204f94f2..d5606921 100644
--- a/tools/chatgui-android/libutun_lite/stream_bridge.c
+++ b/tools/chatgui-android/libutun_lite/stream_bridge.c
@@ -1,50 +1,100 @@
 #include <pthread.h>
 #include <string.h>
-
 #include "../../../lib/debug_config.h"
 #include "../../../lib/mem.h"
 #include "../../../lib/u_async.h"
 #include "../../../src/utun_instance.h"
-#include "../../../src/chat/chat_core.h"
 #include "../../../src/stream/stream.h"
-#include "../../../src/stream/stream_media.h"
-#include "../../../src/stream/stream_network.h"
 #include "../../../src/stream/stream_publisher.h"
+#include "../../../src/stream/stream_viewer.h"
 #include "stream_bridge.h"
 
-#define RX_LIMIT (4u * 1024u * 1024u)
-#define RX_RECORDS 256
-
-struct media_record {
-    struct media_record *next;
-    uint64_t id, pts, frame_id, created;
-    size_t length, offset;
-    int key;
-    uint8_t data[];
-};
-struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned count; };
-struct android_media_session {
-    uint64_t id, group, source, epoch, expected_epoch, next_frame, last_pts[2], drain_until;
-    struct STREAM_HANDLE *handle;
-    struct stream_media_format format;
-    int opening, stopping, finished, need_key, seen[2], rx_overflow, reason;
-    int want_read, want_video, waiting, resync, listen_audio;
-    int64_t position, audio_ahead;
-};
-
 static pthread_mutex_t media_lock = PTHREAD_MUTEX_INITIALIZER;
 static struct UTUN_INSTANCE *media_inst;
-static struct posted_task *media_post;
-static void *media_timer;
 static struct stream_publisher *transmitter;
-static struct android_media_session publisher, viewer;
-static struct media_queue output_queue;
-static struct stream_media_status viewer_status;
-static uint64_t viewer_status_id;
-static uint64_t watch_group;
-static int watch_pending;
-static void media_process(void *arg);
-static void media_close(struct android_media_session *s, int kind);
+static struct stream_viewers *viewers;
+static uint64_t publisher_id, watch_group;
+static int publisher_opening;
+static struct posted_task *watch_post;
+
+/* Отложенная команда каталога; прочие команды обслуживают общие publisher/viewer managers. */
+static void watch_process(void *arg) {
+    (void)arg;
+    pthread_mutex_lock(&media_lock); watch_post = NULL;
+    if (stream_watch_sources(media_inst, watch_group, 0) != STREAM_OK)
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android source watch failed group=%llx", (unsigned long long)watch_group);
+    pthread_mutex_unlock(&media_lock);
+}
+
+int android_stream_attach(struct UTUN_INSTANCE *inst) {
+    pthread_mutex_lock(&media_lock);
+    if (!inst || !inst->stream || (media_inst && media_inst != inst)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream attach rejected inst=%p owner=%p", (void *)inst, (void *)media_inst);
+        pthread_mutex_unlock(&media_lock); return -1;
+    }
+    if (media_inst == inst) { pthread_mutex_unlock(&media_lock); return 0; }
+    if (stream_publisher_create(&transmitter, inst) < 0) { pthread_mutex_unlock(&media_lock); return -1; }
+    if (stream_viewers_create(&viewers, inst) < 0) {
+        stream_publisher_destroy(transmitter); transmitter = NULL; pthread_mutex_unlock(&media_lock); return -1;
+    }
+    media_inst = inst;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge attached");
+    pthread_mutex_unlock(&media_lock); return 0;
+}
+
+void android_stream_detach(struct UTUN_INSTANCE *inst) {
+    pthread_mutex_lock(&media_lock);
+    if (media_inst == inst && inst) {
+        if (watch_post) uasync_cancel_post(inst->ua, watch_post);
+        watch_post = NULL; watch_group = 0; stream_watch_sources(inst, 0, 0);
+        stream_viewers_destroy(viewers); viewers = NULL;
+        stream_publisher_destroy(transmitter); transmitter = NULL;
+        publisher_id = 0; publisher_opening = 0; media_inst = NULL;
+        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected");
+    }
+    pthread_mutex_unlock(&media_lock);
+}
+
+int android_stream_watch(uint64_t group) {
+    pthread_mutex_lock(&media_lock);
+    int result = -1;
+    if (!media_inst || !uasync_is_running(media_inst->ua))
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android source watch rejected group=%llx", (unsigned long long)group);
+    else {
+        if (!watch_post) {
+            watch_post = u_calloc(1, sizeof(*watch_post));
+            if (watch_post) { watch_post->callback = watch_process; uasync_post_reserved(media_inst->ua, watch_post); }
+            else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android source watch allocation failed");
+        }
+        if (watch_post) { watch_group = group; result = 0; }
+    }
+    pthread_mutex_unlock(&media_lock); return result;
+}
+
+/* Каждому viewer — собственный cursor. Publisher использует отдельное состояние. */
+int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
+                        unsigned rotation, int audio, uint64_t epoch, unsigned codec) {
+    if (!id || !group || (source && !epoch) || codec < STREAM_MEDIA_H264 || codec > STREAM_MEDIA_H266
+        || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid id=%llu group=%llx source=%llx codec=%u",
+                    (unsigned long long)id, (unsigned long long)group, (unsigned long long)source, codec); return -1;
+    }
+    pthread_mutex_lock(&media_lock);
+    int result = -1;
+    struct stream_media_status unused;
+    if (!media_inst || !uasync_is_running(media_inst->ua) || id == publisher_id
+        || (!source && (publisher_id || stream_viewer_status(viewers, id, &unused) == 0)))
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open busy/detached id=%llu", (unsigned long long)id);
+    else if (source)
+        result = stream_viewer_open(viewers, id, group, source, epoch, STREAM_MEDIA_TRACK_VIDEO | (audio ? STREAM_MEDIA_TRACK_AUDIO : 0));
+    else {
+        struct stream_media_format format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation,
+            (uint8_t)codec, audio ? STREAM_MEDIA_OPUS_STEREO : 0, 1};
+        result = stream_publisher_open(transmitter, id, group, &format);
+        if (!result) { publisher_id = id; publisher_opening = 1; }
+    }
+    pthread_mutex_unlock(&media_lock); return result;
+}
 
 int android_stream_tx_snapshot(uint64_t id, struct android_stream_tx_stats *out) {
     if (!out) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "TX stats output missing"); return -1; }
@@ -56,245 +106,6 @@ int android_stream_tx_snapshot(uint64_t id, struct android_stream_tx_stats *out)
     pthread_mutex_unlock(&media_lock); return result;
 }
 
-/* Очереди обмена pthread: LL_QUEUE принадлежит uasync и сюда не подходит. */
-static void queue_append(struct media_queue *q, struct media_record *record) {
-    record->next = NULL;
-    if (q->tail) q->tail->next = record; else q->head = record;
-    q->tail = record; q->bytes += record->length; q->count++;
-}
-static struct media_record *queue_take(struct media_queue *q) {
-    struct media_record *record = q->head;
-    if (!record) return NULL;
-    q->head = record->next; if (!q->head) q->tail = NULL;
-    q->bytes -= record->length; q->count--; record->next = NULL;
-    return record;
-}
-static void queue_clear(struct media_queue *q) {
-    struct media_record *record;
-    while ((record = queue_take(q))) u_free(record);
-}
-
-/* RX pressure удаляет media/GAP, сохраняя подтверждения открытия/закрытия owners. */
-static void media_drop_output(void) {
-    struct media_record **link = &output_queue.head, *previous = NULL;
-    while (*link) {
-        struct media_record *r = *link;
-        if (r->data[0] <= 3 || (r->id != publisher.id && r->id != viewer.id)) {
-            *link = r->next; if (output_queue.tail == r) output_queue.tail = previous;
-            output_queue.bytes -= r->length; output_queue.count--; u_free(r);
-        } else { previous = r; link = &r->next; }
-    }
-}
-
-/* Все обращения к ua, включая post, защищены тем же mutex, что и detach перед destroy. */
-static int media_schedule(void) {
-    if (media_post || media_timer) return 0;
-    if (!media_inst || !uasync_is_running(media_inst->ua)) {
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream schedule: core detached/stopped"); return -1;
-    }
-    struct posted_task *task = u_calloc(1, sizeof(*task));
-    if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream post allocation failed"); return -1; }
-    task->callback = media_process;
-    media_post = task; uasync_post_reserved(media_inst->ua, task);
-    return 0;
-}
-
-/* Выдача владеет копией: lifetime scratch stream callback не выходит в Kotlin. */
-static void media_output(uint64_t id, int kind, const struct stream_media_format *format, uint64_t pts,
-                         int flags, const uint8_t *data, size_t size) {
-    if (!id || size > STREAM_MEDIA_MAX_FRAME) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android output invalid id/size=%zu", size); return; }
-    if (output_queue.count >= RX_RECORDS || size + ANDROID_STREAM_HEADER_SIZE > RX_LIMIT - output_queue.bytes) {
-        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream RX full id=%llu bytes=%zu count=%u",
-                   (unsigned long long)id, output_queue.bytes, output_queue.count);
-        viewer.rx_overflow = 1;
-        media_drop_output();
-        if (kind == 1 || kind == 2) { media_output(id, 3, NULL, 0, 0, NULL, 0); return; }
-    }
-    struct media_record *record = u_calloc(1, sizeof(*record) + size + ANDROID_STREAM_HEADER_SIZE);
-    if (!record) {
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream RX allocation failed bytes=%zu", size); viewer.rx_overflow = 1; return;
-    }
-    record->id = id; record->length = size + ANDROID_STREAM_HEADER_SIZE; record->created = get_time_us();
-    uint8_t *p = record->data; p[0] = (uint8_t)kind; p[1] = (uint8_t)flags;
-    if (format) {
-        stream_write_be(p + 2, format->width, 2); stream_write_be(p + 4, format->height, 2);
-        stream_write_be(p + 6, format->rotation, 2); stream_write_be(p + 20, format->revision, 4);
-        p[24] = format->video_codec;
-    }
-    stream_write_be(p + 8, pts, 8); stream_write_be(p + 16, size, 4);
-    if (size) memcpy(p + ANDROID_STREAM_HEADER_SIZE, data, size);
-    queue_append(&output_queue, record);
-}
-
-/* Callback receiver работает под media_lock, декодирование видео/аудио выполняет внешний worker. */
-static void media_frame(void *arg, const struct stream_media_format *format,
-                        const struct stream_media_video *video, const struct stream_media_audio *audio) {
-    struct android_media_session *s = arg;
-    if (s->rx_overflow) return;
-    if (video) media_output(s->id, 1, format, video->pts_us, video->flags, video->data, video->length);
-    if (audio) media_output(s->id, 2, format, audio->pts_us, 0, audio->data, audio->length);
-}
-
-/* Состояния потока не передаются через chat SharedFlow; здесь только владельцы выбранного viewer. */
-static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
-    (void)arg;
-    pthread_mutex_lock(&media_lock);
-    if (inst == media_inst && viewer.handle && event->descriptor.group_id == viewer.group
-        && event->descriptor.source_node_id == viewer.source && event->descriptor.source_epoch == viewer.epoch
-        && !viewer.stopping && (event->type != STREAM_EVENT_CLOSED || !stream_handle_active(viewer.handle))) {
-        if (event->type == STREAM_EVENT_CLOSED) {
-            viewer.reason = event->reason;
-            media_close(&viewer, 6);
-        } else if (event->type == STREAM_EVENT_AVAILABLE) {
-            viewer.waiting = 0;
-            if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media ready scheduling failed");
-        }
-    }
-    pthread_mutex_unlock(&media_lock);
-}
-
-/* Publisher держит конечное окно выдачи после FIN; abrupt detach ничего не обещает получателям. */
-static void media_close(struct android_media_session *s, int kind) {
-    if (kind) media_output(s->id, kind, NULL, 0, kind == 6 ? s->reason : 0, NULL, 0);
-    if (s->handle) stream_close(s->handle);
-    memset(s, 0, sizeof(*s));
-}
-
-/* Один actor обслуживает команды и один ответ на запрос; idle не копирует поток из кольца. */
-static void media_process(void *arg) {
-    (void)arg;
-    pthread_mutex_lock(&media_lock);
-    media_post = NULL; media_timer = NULL;
-    if (!media_inst) { pthread_mutex_unlock(&media_lock); return; }
-    if (watch_pending) {
-        watch_pending = 0;
-        if (stream_watch_sources(media_inst, watch_group, 0) != STREAM_OK)
-            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source watch failed group=%llx", (unsigned long long)watch_group);
-    }
-    if (viewer.opening) {
-        if (viewer.stopping) media_close(&viewer, 5);
-        else {
-            viewer.opening = 0;
-            int result = stream_media_open(media_inst, viewer.group, viewer.source, viewer.expected_epoch,
-                STREAM_MEDIA_TRACK_VIDEO | (viewer.listen_audio ? STREAM_MEDIA_TRACK_AUDIO : 0), &viewer.handle);
-            viewer.reason = result == STREAM_ERR_CONFLICT ? STREAM_CLOSE_SOURCE_CHANGED : STREAM_CLOSE_FAILED;
-            if (result != STREAM_OK) {
-                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open failed id=%llu group=%llx source=%llx result=%d",
-                    (unsigned long long)viewer.id, (unsigned long long)viewer.group, (unsigned long long)viewer.source, result);
-                media_close(&viewer, 6);
-            } else {
-                viewer.epoch = viewer.expected_epoch;
-                viewer_status_id = viewer.id; memset(&viewer_status, 0, sizeof(viewer_status));
-                media_output(viewer.id, 4, NULL, 0, 0, NULL, 0);
-                DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android viewer opened id=%llu group=%llx source=%llx epoch=%llx",
-                    (unsigned long long)viewer.id, (unsigned long long)viewer.group,
-                    (unsigned long long)viewer.source, (unsigned long long)viewer.epoch);
-            }
-        }
-    }
-    if (viewer.stopping && viewer.id) media_close(&viewer, 5);
-    if (viewer.handle && (viewer.resync || viewer.rx_overflow)) {
-        viewer.resync = viewer.rx_overflow = 0; viewer.position = -1; viewer.waiting = 0;
-        if (stream_media_resync(viewer.handle) != STREAM_OK) media_close(&viewer, 6);
-    }
-    if (viewer.handle) {
-        int result = stream_media_get_status(viewer.handle, &viewer_status);
-        if (result < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media index failed result=%d", result); media_close(&viewer, 6); }
-    }
-    unsigned queued_viewer = 0;
-    for (struct media_record *r = output_queue.head; r; r = r->next) if (r->id == viewer.id) queued_viewer++;
-    if (viewer.handle && viewer.want_read && !queued_viewer
-        && output_queue.bytes + STREAM_MEDIA_MAX_FRAME + ANDROID_STREAM_HEADER_SIZE <= RX_LIMIT) {
-        viewer.want_read = 0;
-        struct stream_media_packet packet;
-        int result = stream_media_read(viewer.handle, viewer.position, viewer.want_video, viewer.audio_ahead, &packet);
-        if (stream_media_get_status(viewer.handle, &viewer_status) < 0) result = STREAM_ERR_FORMAT;
-        if (result == STREAM_OK) {
-            if (packet.kind == STREAM_MEDIA_READ_END) media_close(&viewer, 5);
-            else media_frame(&viewer, &packet.format, packet.kind == STREAM_MEDIA_READ_VIDEO ? &packet.video : NULL,
-                             packet.kind == STREAM_MEDIA_READ_AUDIO ? &packet.audio : NULL);
-        } else if (result == STREAM_MORE) {
-            viewer.want_read = 1;
-            if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media assembly scheduling failed");
-        } else if (result == STREAM_WAIT) viewer.want_read = viewer.waiting = 1;
-        else if (result == STREAM_GAP || result == STREAM_EXPIRED) {
-            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android playback resync id=%llu result=%d", (unsigned long long)viewer.id, result);
-            stream_media_resync(viewer.handle); viewer.position = -1; media_output(viewer.id, 3, NULL, 0, 0, NULL, 0);
-            memset(&viewer_status, 0, sizeof(viewer_status));
-        } else { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android playback read failed result=%d", result); media_close(&viewer, 6); }
-    }
-    pthread_mutex_unlock(&media_lock);
-}
-
-/* Attach/detach — обязательная граница lifetime ядра. */
-int android_stream_attach(struct UTUN_INSTANCE *inst) {
-    pthread_mutex_lock(&media_lock);
-    if (!inst || !inst->stream || (media_inst && media_inst != inst)) {
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream attach rejected inst=%p owner=%p", (void *)inst, (void *)media_inst);
-        pthread_mutex_unlock(&media_lock); return -1;
-    }
-    if (!transmitter && stream_publisher_create(&transmitter, inst) < 0) {
-        pthread_mutex_unlock(&media_lock); return -1;
-    }
-    media_inst = inst; stream_set_event_callback(inst, media_event, NULL);
-    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge attached");
-    pthread_mutex_unlock(&media_lock);
-    return 0;
-}
-void android_stream_detach(struct UTUN_INSTANCE *inst) {
-    pthread_mutex_lock(&media_lock);
-    if (media_inst == inst) {
-        if (media_post) uasync_cancel_post(inst->ua, media_post);
-        if (media_timer) uasync_cancel_timeout(inst->ua, media_timer);
-        media_post = NULL; media_timer = NULL;
-        stream_watch_sources(inst, 0, 0);
-        stream_set_event_callback(inst, NULL, NULL);
-        queue_clear(&output_queue);
-        stream_publisher_destroy(transmitter); transmitter = NULL; memset(&publisher, 0, sizeof(publisher));
-        if (viewer.id) media_close(&viewer, 0);
-        media_inst = NULL;
-        watch_pending = 0; watch_group = 0;
-        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected");
-    }
-    pthread_mutex_unlock(&media_lock);
-}
-
-/* Команда содержит идентификаторы, а не borrowed pointer ядра. */
-int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
-                        unsigned rotation, int audio, uint64_t expected_epoch, unsigned codec) {
-    if (!id || !group || (source && !expected_epoch)
-        || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) {
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid id=%llu group=%llx source=%llx epoch=%llx dimensions=%ux%u",
-                    (unsigned long long)id, (unsigned long long)group, (unsigned long long)source,
-                    (unsigned long long)expected_epoch, width, height); return -1;
-    }
-    pthread_mutex_lock(&media_lock);
-    struct android_media_session *s = source ? &viewer : &publisher;
-    int result = -1;
-    if (codec < STREAM_MEDIA_H264 || codec > STREAM_MEDIA_H266 || s->id || id == publisher.id || id == viewer.id
-        || !media_inst || !uasync_is_running(media_inst->ua)) {
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open busy/detached id=%llu", (unsigned long long)id);
-    } else if ((!source || media_schedule() == 0)) {
-        *s = (struct android_media_session){.id = id, .group = group, .source = source, .expected_epoch = expected_epoch,
-            .opening = 1, .need_key = 1, .listen_audio = !!audio, .position = -1,
-            .audio_ahead = STREAM_MEDIA_READ_AHEAD_US,
-            .format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation, (uint8_t)codec, audio ? STREAM_MEDIA_OPUS_STEREO : 0, 1}};
-        result = source ? 0 : stream_publisher_open(transmitter, id, group, &s->format);
-        if (result < 0) memset(s, 0, sizeof(*s));
-    }
-    pthread_mutex_unlock(&media_lock); return result;
-}
-
-/* Последняя выбранная группа заменяет ещё не обработанную команду, указатели ядра не покидают mutex. */
-int android_stream_watch(uint64_t group) {
-    pthread_mutex_lock(&media_lock);
-    int result = media_inst && uasync_is_running(media_inst->ua) ? media_schedule() : -1;
-    if (result == 0) { watch_group = group; watch_pending = 1; }
-    else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream watch rejected group=%llx", (unsigned long long)group);
-    pthread_mutex_unlock(&media_lock); return result;
-}
-
-/* Перегрузка удаляет целые видео-кадры и блокирует P-кадры до нового KEY. */
 int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length,
                         const struct stream_media_format *format) {
     pthread_mutex_lock(&media_lock);
@@ -308,46 +119,51 @@ int android_stream_set_audio_enabled(uint64_t id, int enabled) {
     pthread_mutex_unlock(&media_lock); return result;
 }
 
-/* Stop идемпотентен; publisher дренирует принятую очередь, viewer закрывается сразу в actor. */
 void android_stream_stop(uint64_t id) {
     pthread_mutex_lock(&media_lock);
-    struct android_media_session *s = publisher.id == id ? &publisher : viewer.id == id ? &viewer : NULL;
-    if (s == &publisher) { s->stopping = 1; stream_publisher_stop(transmitter, id); }
-    else if (s) { s->stopping = 1; if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream stop schedule failed"); }
+    if (id == publisher_id && id) stream_publisher_stop(transmitter, id);
+    else stream_viewer_stop(viewers, id);
     pthread_mutex_unlock(&media_lock);
 }
 
-/* Выбрать только события текущего owner; старые owner-события имеют ограниченный TTL. */
+/* Kotlin получает owned wire-копию только ближайшей записи своего reader. */
+static uint8_t *encode_record(int kind, int flags, const struct stream_media_format *format, uint64_t pts,
+                              const uint8_t *data, size_t bytes, size_t *length) {
+    uint8_t *p = u_calloc(1, ANDROID_STREAM_HEADER_SIZE + bytes);
+    if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android output allocation failed bytes=%zu", bytes); return NULL; }
+    p[0] = (uint8_t)kind; p[1] = (uint8_t)flags;
+    if (format) {
+        stream_write_be(p + 2, format->width, 2); stream_write_be(p + 4, format->height, 2);
+        stream_write_be(p + 6, format->rotation, 2); stream_write_be(p + 20, format->revision, 4); p[24] = format->video_codec;
+    }
+    stream_write_be(p + 8, pts, 8); stream_write_be(p + 16, bytes, 4);
+    if (bytes) memcpy(p + ANDROID_STREAM_HEADER_SIZE, data, bytes);
+    *length = ANDROID_STREAM_HEADER_SIZE + bytes; return p;
+}
+
 uint8_t *android_stream_pull(uint64_t id, size_t *length) {
     if (!length) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream pull missing length"); return NULL; }
     *length = 0; pthread_mutex_lock(&media_lock);
-    if (publisher.id) {
+    uint8_t *data = NULL;
+    if (id == publisher_id && id) {
         struct stream_publisher_status status;
-        if (stream_publisher_snapshot(transmitter, publisher.id, &status) == 0) {
-            if (publisher.opening && status.state >= STREAM_PUBLISHER_LIVE && status.state <= STREAM_PUBLISHER_DRAINING) {
-                publisher.opening = 0; publisher.epoch = status.epoch; media_output(publisher.id, 4, NULL, 0, 0, NULL, 0);
-            }
+        if (stream_publisher_snapshot(transmitter, id, &status) == 0) {
             if (status.state == STREAM_PUBLISHER_ENDED || status.state == STREAM_PUBLISHER_FAILED) {
-                media_output(publisher.id, status.state == STREAM_PUBLISHER_ENDED ? 5 : 6, NULL, 0, status.reason, NULL, 0);
-                memset(&publisher, 0, sizeof(publisher));
+                data = encode_record(status.state == STREAM_PUBLISHER_ENDED ? 5 : 6, status.reason, NULL, 0, NULL, 0, length);
+                if (data) { publisher_id = 0; publisher_opening = 0; }
+            } else if (publisher_opening && status.state >= STREAM_PUBLISHER_LIVE && status.state <= STREAM_PUBLISHER_DRAINING) {
+                data = encode_record(4, 0, NULL, 0, NULL, 0, length);
+                if (data) publisher_opening = 0;
             }
         }
-    }
-    struct media_record **link = &output_queue.head, *previous = NULL;
-    uint8_t *data = NULL; uint64_t now = get_time_us();
-    while (*link) {
-        struct media_record *r = *link;
-        if (r->id == id || now - r->created > 5000000) {
-            *link = r->next; if (output_queue.tail == r) output_queue.tail = previous;
-            output_queue.bytes -= r->length; output_queue.count--;
-            if (r->id == id) {
-                data = u_malloc((uint32_t)r->length);
-                if (data) { memcpy(data, r->data, r->length); *length = r->length; }
-                else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream pull allocation failed");
-                u_free(r); break;
-            }
+    } else {
+        struct stream_viewer_record *r = stream_viewer_pull(viewers, id);
+        if (r) {
+            data = encode_record(r->kind, r->kind == STREAM_VIEWER_CLOSED ? r->reason : r->flags,
+                                 &r->format, r->pts, r->data, r->length, length);
+            if (!data) stream_viewer_resync(viewers, id);
             u_free(r);
-        } else { previous = r; link = &r->next; }
+        }
     }
     pthread_mutex_unlock(&media_lock); return data;
 }
@@ -358,35 +174,20 @@ int android_stream_read_request(uint64_t id, int64_t position, int want_video, i
                     want_video, (long long)audio_ahead); return -1;
     }
     pthread_mutex_lock(&media_lock);
-    int result = 0;
-    if (viewer.id == id) {
-        int queued = 0;
-        for (struct media_record *r = output_queue.head; r; r = r->next) if (r->id == id) queued = 1;
-        if (!queued) {
-            int changed = viewer.position != position || viewer.want_video != want_video || viewer.audio_ahead != audio_ahead;
-            viewer.position = position; viewer.want_video = want_video; viewer.audio_ahead = audio_ahead;
-            viewer.want_read = want_video || audio_ahead >= 0;
-            if (viewer.want_read && (!viewer.waiting || changed)) { viewer.waiting = 0; result = media_schedule(); }
-        }
-    }
-    pthread_mutex_unlock(&media_lock); return result;
+    if (viewers && id != publisher_id) stream_viewer_request(viewers, id, position, want_video, audio_ahead);
+    /* CLOSED/END остаются доступны pull даже после закрытия cursor; неизвестный ID даёт пустой pull. */
+    pthread_mutex_unlock(&media_lock); return 0;
 }
 
 int android_stream_get_status(uint64_t id, struct stream_media_status *status) {
-    if (!status) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media status: missing output"); return -1; }
+    if (!status) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media status missing output"); return -1; }
     pthread_mutex_lock(&media_lock);
-    int result = viewer_status_id == id ? 0 : -1;
-    if (!result) *status = viewer_status;
+    int result = stream_viewer_status(viewers, id, status);
     pthread_mutex_unlock(&media_lock); return result;
 }
 
 int android_stream_resync(uint64_t id) {
     pthread_mutex_lock(&media_lock);
-    int result = -1;
-    if (viewer.id == id && viewer.handle) {
-        media_drop_output(); viewer.resync = viewer.want_read = 1; viewer.waiting = 0; viewer.position = -1;
-        memset(&viewer_status, 0, sizeof(viewer_status));
-        result = media_schedule();
-    }
+    int result = stream_viewer_resync(viewers, id);
     pthread_mutex_unlock(&media_lock); return result;
 }
diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.h b/tools/chatgui-android/libutun_lite/stream_bridge.h
index 3b4e1e2d..06b4f8c5 100644
--- a/tools/chatgui-android/libutun_lite/stream_bridge.h
+++ b/tools/chatgui-android/libutun_lite/stream_bridge.h
@@ -1,6 +1,6 @@
 /* Android-очереди live media: attach/detach только в uasync, вне callbacks.
  * UI/кодировщики используют только session_id, никогда не сохраняют inst/ua.
- * Один publisher и один viewer, TX <=2MB/128 records, RX <=4MB/256 records.
+ * Один publisher и независимые viewers, TX <=2MB/128 records, RX — одна ближайшая запись на viewer.
  * Data принимает полные access units пяти кодеков uAV2 (KEY: config_length:4/config/frame) и Opus 48k mono/stereo/20ms.
  * pull передаёт u_malloc-буфер: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,revision:4,codec:1,reserved:3,data.
  * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6; error.flags = stream_close_reason. Числа big-endian.