Merge remote-tracking branch 'origin/master'

doc/stream.txt

@@ -755,7 +755,7 @@ Android:
 - Camera/codec callbacks — HandlerThread, Opus encoder — audio capture thread, decoder/AudioTrack — viewer worker.
 - Native bridge держит pthread mutex, ограниченные очереди и один pending post либо timer.
   attach/detach выполняются в uasync вне callbacks. detach отменяет post/timer, закрывает handles,
-  очищает media, snapshots и запрещает producers до уничтожения instance/ua.
+  очищает media, снимает наблюдение каталога и запрещает producers до уничтожения instance/ua.
   UI не сохраняет borrowed inst/ua. Idle без pending data не будит мобильное ядро.
 - TX<=2MiB/128 records, не более 250ms ожидания. Actor публикует <=2 пачек за вызов,
   неполную пачку объединяет с аудио/видео до 20ms; продолжение работы через timer 5ms.
@@ -771,8 +771,16 @@ Android:
   своего микрофона, чтобы исключить акустическую обратную связь. Камера при закрытии preview продолжает работать.
   Viewer закрывается на Activity ON_STOP и создаётся заново на ON_START; publisher от UI не зависит.
   Видимый player удерживает экран включённым. Preview ограничен оставшимся местом, кнопки управления не перекрывает.
-- Source list строится в uasync по stream_member_load и stream_member_online, выдаётся UI копией JSON.
+- Source list общий для Android/desktop: stream_sources_json в uasync использует stream_member_load
-  Свежий snapshot содержит source ID, epoch, self и экранированное имя. Увиденный источник не создаёт подписку.
+  и stream_member_online. stream_watch_sources наблюдает один выбранный канал; group=0 снимает наблюдение.
+  Начальный snapshot и изменения после Merkle/BGP передаются через CHAT_EVT_STREAM_SOURCES=58:
+  group_id:8BE, valid:1, JSON с id(decimal string), epoch(hex string), self, name.
+  События объединяются отложенным callback, одинаковый список повторно не отправляется. Таймера опроса нет.
+  Обнаружение не создаёт подписку, media-буфер и не удерживает ядро в ACTIVE.
+- В шапке канала значок виден при наличии источников; один источник открывается сразу,
+  несколько выбираются из списка по имени/ID. Пункт меню «Трансляции» также открывает управление публикацией.
+  Выбранная эпоха проверяется actor до подписки и по фактически открытому handle. Перезапуск источника
+  требует нового выбора. Ошибка viewer kind=6 передаёт stream_close_reason в flags.
 - Stop сначала завершает capture producers и encoder EOS (максимум 1s), затем FIN.
   Publisher сохраняет конечный буфер ещё 2s для сетевой выдачи, затем закрывается.
   Это ограниченный drain, не подтверждение воспроизведения всеми viewers. Abort/core restart очищает очередь сразу.
@@ -785,3 +793,24 @@ Kotlin progress — DEBUG раз в секунду; C packet-level диагно
 Проверки: test_stream_media (все пары потерь 6+2, parser/границы/GAP), test_android_stream
 (живое ядро, preview, verified snapshots, FIN/cancel, concurrent producer/detach и leak count),
 Android H264AnnexBTest (Annex B/length-prefixed NAL, SPS/PPS/IDR и усечённые данные).
+
+Desktop:
+- Значок трансляций расположен в шапке MessageList. Источники берутся из того же каталога ядра;
+  ответы предыдущего канала не меняют текущую шапку. Личные беседы снимают наблюдение.
+- Один StreamPlayerWindow на процесс. Он содержит имя источника, состояние подключения, громкость,
+  полноэкранный режим и закрытие. Выбор нового источника закрывает предыдущего reader.
+- transport/stream_bridge владеет одним viewer и ограничивает RX 4MiB/256 записей.
+  Attach/detach выполняются в uasync; команды и копирование медиаданных защищены mutex.
+  Post отменяется до destroy ядра; GUI/decoder не хранят заимствованные instance/handle pointers.
+- stream_media_receiver собирает H.264 access units и выдаёт Opus 48kHz mono/20ms.
+  FFmpeg/Opus декодируются отдельным worker; Qt-очередь не накапливает кадры.
+  Очереди ограничены шестью декодированными видеокадрами и 25 аудиопакетами (500ms).
+  Видео декодируется без задержки frame-threading. При GAP сбрасываются decoder refs и timeline до KEY.
+- После prebuffer 150ms показ следует физическим audio callback timestamps; без звука — монотонным часам.
+  Короткие пропуски аудио заполняются Opus PLC; большой разрыв/pressure сбрасывает timeline.
+  Закрытие окна снимает только reader, собственная публикация сохраняется.
+  Удаление канала сообщает STREAM_CLOSE_GROUP; остановка ядра снимает media и catalogue owners.
+- Без FFmpeg окно сообщает, что декодирование недоступно. Нужна desktop-сборка с HAVE_FFMPEG.
+Проверки: test_stream_view — настоящее ядро, старые эпохи/cancel, RX pressure, H.264 с поворотом,
+Opus/громкость на настоящем null audio backend, закрытие канала и полный баланс tracked allocations.
+test_stream_service также проверяет общий каталог двух независимых источников одного канала.

src/chat/chat_event.h

@@ -79,6 +79,7 @@ extern "C" {
 #define CHAT_EVT_RADIO_MUTE_RESULT 55 /* [group_id:8LE][node_id:8LE][success:1] — локальный переключатель рации */
 #define CHAT_EVT_MSG_SUBMITTED 56 /* [request_id:8LE][success:1] — результат локального сохранения сообщения */
 #define CHAT_EVT_CPU_LOAD 57 /* [valid:1][average_percent:1] — CPU потока uasync, среднее за секунду, 0..100 */
+#define CHAT_EVT_STREAM_SOURCES 58 /* [group_id:8BE][valid:1][JSON:utf8] — доступные публикации выбранного канала */
 
 typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);
 

src/stream/stream.c

@@ -1,12 +1,16 @@
 #include <stdio.h>
 #include <string.h>
+#include <limits.h>
 
 #include "../../lib/debug_config.h"
 #include "../../lib/mem.h"
 #include "../../lib/platform_compat.h"
 #include "../../lib/u_async.h"
+#include "../../lib/strbuf.h"
+#include "../../lib/sqlite3.h"
 #include "../utun_instance.h"
 #include "../chat/chat_core.h"
+#include "../chat/chat_event.h"
 #include "stream_member.h"
 #include "stream_source.h"
 #include "stream_priv.h"
@@ -50,12 +54,112 @@ struct stream_service {
     stream_event_fn callback;
     void *arg;
     void *reconcile;
+    void *sources_pending;
+    uint64_t sources_group;
+    char *sources_snapshot;
+    int sources_valid;
     int stopping, dispatching, registered, initializing;
 };
 
 static void session_deliver(void *arg);
 static void service_reconcile(void *arg);
 
+/* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */
+int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) {
+    if (!inst || !out || *out || !group_id) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list invalid arguments"); return STREAM_ERR_ARGUMENT;
+    }
+    struct TOPO_GROUP *group = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
+    if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return STREAM_WAIT;
+    for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source namespace invalid group=%llx", (unsigned long long)group_id);
+        return STREAM_ERR_STATE;
+    }
+    sqlite3 *db = chat_core_get_db(inst);
+    if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list missing DB"); return STREAM_ERR_STATE; }
+    char query[320];
+    snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') "
+             "FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id ORDER BY p.node_id", group->channel_id);
+    struct strbuf json = strbuf_new();
+    sqlite3_stmt *statement = NULL;
+    int result = STREAM_OK, count = 0;
+    if (strbuf_addf(&json, "[") < 0) result = STREAM_ERR_MEMORY;
+    if (result == STREAM_OK && sqlite3_prepare_v2(db, query, -1, &statement, NULL) != SQLITE_OK) result = STREAM_ERR_STATE;
+    int step = SQLITE_DONE;
+    while (result == STREAM_OK && (step = sqlite3_step(statement)) == SQLITE_ROW) {
+        uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0);
+        if (!stream_member_online(inst, group_id, node)) continue;
+        struct stream_member_state state;
+        int loaded = stream_member_load(inst, group_id, node, &state, NULL);
+        if (loaded == STREAM_WAIT || loaded == STREAM_ERR_NOT_MEMBER) continue;
+        if (loaded < 0) {
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source excluded group=%llx node=%llx profile_result=%d",
+                       (unsigned long long)group_id, (unsigned long long)node, loaded); continue;
+        }
+        if (!state.published) continue;
+        if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "",
+            (unsigned long long)node, (unsigned long long)state.source.source_epoch, node == inst->node_id) < 0
+            || strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0
+            || strbuf_addf(&json, "}") < 0) result = STREAM_ERR_MEMORY;
+        count++;
+    }
+    if (result == STREAM_OK && step != SQLITE_DONE) result = STREAM_ERR_STATE;
+    sqlite3_finalize(statement);
+    if (result == STREAM_OK && strbuf_addf(&json, "]") < 0) result = STREAM_ERR_MEMORY;
+    if (result == STREAM_OK) *out = strbuf_detach(&json);
+    strbuf_free(&json);
+    if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list failed group=%llx result=%d DB=%s",
+                                       (unsigned long long)group_id, result, sqlite3_errmsg(db));
+    else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source list group=%llx count=%d", (unsigned long long)group_id, count);
+    return result;
+}
+
+/* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */
+static void sources_deliver(void *arg) {
+    struct stream_service *service = arg;
+    service->sources_pending = NULL;
+    if (service->stopping || !service->sources_group) return;
+    char *json = NULL;
+    int result = stream_sources_json(service->inst, service->sources_group, &json);
+    int valid = result == STREAM_OK || result == STREAM_WAIT;
+    const char *list = json ? json : "[]";
+    if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) {
+        u_free(json); return;
+    }
+    size_t length = strlen(list);
+    uint8_t *event = length <= INT_MAX - 9 ? u_malloc((uint32_t)(9 + length)) : NULL;
+    if (!event) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source event allocation failed bytes=%zu", length); u_free(json); return;
+    }
+    stream_write_be(event, service->sources_group, 8); event[8] = (uint8_t)valid;
+    memcpy(event + 9, list, length);
+    char *snapshot = json ? json : u_strdup(list);
+    if (!snapshot) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source snapshot allocation failed");
+    u_free(service->sources_snapshot); service->sources_snapshot = snapshot; service->sources_valid = valid;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu",
+               (unsigned long long)service->sources_group, valid, length);
+    chat_event_post(service->inst, CHAT_EVT_STREAM_SOURCES, event, (int)(length + 9));
+    u_free(event);
+}
+
+/* Наблюдение принадлежит сервису и снимается до освобождения ядра. */
+int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id) {
+    struct stream_service *service = inst ? inst->stream : NULL;
+    if (!service || service->stopping) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id);
+        return STREAM_ERR_STATE;
+    }
+    service->sources_group = group_id;
+    u_free(service->sources_snapshot); service->sources_snapshot = NULL;
+    if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
+    service->sources_pending = group_id ? uasync_call_soon(inst->ua, service, sources_deliver) : NULL;
+    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx", (unsigned long long)group_id);
+    if (group_id && !service->sources_pending) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY;
+    }
+    return STREAM_OK;
+}
+
 /* Observer выполняет только учёт; пользовательские callbacks и отправка остаются отложенными. */
 static void cache_event(void *arg, const uint8_t *wire, size_t length, int added) {
     struct stream_session *s = arg;
@@ -268,7 +372,13 @@ static void service_reconcile(void *arg) {
 static void member_changed(uint64_t node_id, const char *tags, const char *channel, void *arg) {
     (void)tags;
     struct stream_service *service = arg;
-    if (service->stopping || !service->sessions) return;
+    if (service->stopping) return;
+    struct TOPO_GROUP *watched = service->sources_group ? stream_group(service->inst, service->sources_group) : NULL;
+    if (watched && channel && !strcmp(watched->channel_id, channel) && !service->sources_pending) {
+        service->sources_pending = uasync_call_soon(service->inst->ua, service, sources_deliver);
+        if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed");
+    }
+    if (!service->sessions) return;
     stream_network_membership_changed(service->network, node_id, channel);
     for (struct stream_session *s = service->sessions; s; s = s->next) {
         int own = node_id == service->inst->node_id;
@@ -344,6 +454,8 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
     stream_network_destroy(service->network); service->network = NULL;
     if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service);
     if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile);
+    if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
+    u_free(service->sources_snapshot);
     while (service->sessions) {
         struct stream_session *s = service->sessions;
         if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0);
@@ -360,10 +472,14 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
 void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) {
     struct stream_service *service = inst ? inst->stream : NULL;
     if (!service) return;
+    if (service->sources_group == group_id && !service->sources_pending) {
+        service->sources_pending = uasync_call_soon(inst->ua, service, sources_deliver);
+        if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed");
+    }
     struct stream_session *s = service->sessions;
     while (s) {
         struct stream_session *next = s->next;
-        if (!s->closing && s->descriptor.group_id == group_id) session_close(s, STREAM_CLOSE_GROUP, 1, 0);
+        if (!s->closing && s->descriptor.group_id == group_id) session_close(s, STREAM_CLOSE_GROUP, 1, 1);
         s = next;
     }
 }

src/stream/stream.h

@@ -60,6 +60,12 @@ int stream_service_init(struct UTUN_INSTANCE *inst);
 void stream_service_destroy(struct UTUN_INSTANCE *inst);
 void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id);
 void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg);
+/* Проверенные online-публикации: JSON с id(decimal string), epoch(hex string), name, self.
+ * Только uasync; *out заранее NULL, результат освобождается u_free(). WAIT — группа недоступна. */
+int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out);
+/* Один выбранный канал frontend: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES.
+ * group_id=0 снимает наблюдение. Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */
+int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id);
 /* Локальный owner удерживает эффективный ACTIVE; желаемый frontend-режим ядро восстанавливает после последнего close. */
 int stream_service_needs_activity(const struct UTUN_INSTANCE *inst);
 

tests/test_android_stream.c

@@ -7,6 +7,9 @@
 #include "../lib/mem.h"
 #include "../lib/sqlite3.h"
 #include "../src/chat/chat_core.h"
+#include "../src/chat/chat_event.h"
+#include "../src/stream/stream.h"
+#include "../src/stream/stream_member.h"
 #include "../src/utun_instance.h"
 #include "../tools/chatgui-android/libutun_lite/stream_bridge.h"
 #include "../src/stream/stream_proto.h"
@@ -16,6 +19,18 @@
 static uint8_t video[9000];
 static struct UASYNC *ua;
 static int producer_started, producer_stop;
+static unsigned source_events;
+static uint64_t source_group;
+static char *source_snapshot;
+
+static void chat_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data, int length) {
+    (void)inst;
+    if (type != CHAT_EVT_STREAM_SOURCES) return;
+    CHECK(length >= 11 && data[8] == 1);
+    source_group = stream_read_be(data, 8); source_events++;
+    u_free(source_snapshot); source_snapshot = u_malloc((uint32_t)(length - 8)); CHECK(source_snapshot);
+    memcpy(source_snapshot, data + 9, (size_t)(length - 9)); source_snapshot[length - 9] = '\0';
+}
 
 static uint8_t *wait_event(uint64_t id, int kind, size_t *size) {
     for (unsigned i = 0; i < 1000; i++) {
@@ -48,11 +63,22 @@ int main(void) {
     CHECK(sqlite3_step(statement) == SQLITE_ROW);
     uint64_t group = strtoull((const char *)sqlite3_column_text(statement, 0), NULL, 10); sqlite3_finalize(statement); CHECK(group);
     uasync_mark_running(ua); android_stream_attach(inst);
-    CHECK(android_stream_open(1, group, 0, 1280, 720, 90, 1) == 0);
+    chat_event_set_handler(inst, chat_event);
+    CHECK(android_stream_watch(group) == 0); test_clock_poll(ua, 100);
+    CHECK(source_events == 1 && source_group == group && !strcmp(source_snapshot, "[]"));
+    CHECK(!stream_service_needs_activity(inst));
+    CHECK(android_stream_open(1, group, 0, 1280, 720, 90, 1, 0) == 0);
     size_t size = 0; u_free(wait_event(1, 4, &size));
-    char *sources = android_stream_sources(group); u_free(sources); test_clock_poll(ua, 100);
+    char *sources = NULL; CHECK(stream_sources_json(inst, group, &sources) == STREAM_OK);
-    sources = android_stream_sources(group); CHECK(strstr(sources, "epoch") && strstr(sources, "name")); u_free(sources);
+    CHECK(strstr(sources, "epoch") && strstr(sources, "name")); u_free(sources); test_clock_poll(ua, 100);
-    CHECK(android_stream_open(2, group, inst->node_id, 0, 0, 0, 1) == 0); u_free(wait_event(2, 4, &size));
+    CHECK(source_events == 2 && strstr(source_snapshot, "epoch"));
+    struct stream_member_state member;
+    CHECK(stream_member_load(inst, group, inst->node_id, &member, NULL) == STREAM_OK && member.published);
+    CHECK(android_stream_open(5, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch ^ 1) == 0);
+    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) == 0);
+    u_free(wait_event(2, 4, &size));
+    CHECK(source_events == 2); /* Подписка не меняет каталог публикаций. */
     for (unsigned i = 0; i < sizeof(video); i++) video[i] = (uint8_t)i;
     CHECK(android_stream_feed(1, 1, 100000, 0, video, sizeof(video)) == 0);
     CHECK(android_stream_feed(1, 1, 100000, 1, video, sizeof(video)) == 1);
@@ -69,7 +95,8 @@ int main(void) {
     CHECK(video_seen == 1 && audio_seen == 1);
     android_stream_stop(1); u_free(wait_event(2, 5, &size)); android_stream_stop(2); u_free(wait_event(1, 5, &size));
     while ((output = android_stream_pull(2, &size))) u_free(output);
-    CHECK(android_stream_open(3, group, 0, 1280, 720, 0, 0) == 0); u_free(wait_event(3, 4, &size));
+    CHECK(!strcmp(source_snapshot, "[]"));
+    CHECK(android_stream_open(3, group, 0, 1280, 720, 0, 0, 0) == 0); u_free(wait_event(3, 4, &size));
     pthread_t thread; CHECK(pthread_create(&thread, NULL, producer, NULL) == 0);
     while (!__atomic_load_n(&producer_started, __ATOMIC_ACQUIRE)) {}
     android_stream_detach(inst);
@@ -77,9 +104,10 @@ int main(void) {
     CHECK(android_stream_feed(3, 1, 900000, 1, video, sizeof(video)) == -1);
     while ((output = android_stream_pull(3, &size))) u_free(output);
     android_stream_attach(inst);
-    CHECK(android_stream_open(4, group, 0, 1280, 720, 0, 0) == 0); android_stream_stop(4);
+    CHECK(android_stream_open(4, group, 0, 1280, 720, 0, 0, 0) == 0); android_stream_stop(4);
     u_free(wait_event(4, 5, &size));
     android_stream_detach(inst); uasync_mark_stopped(ua); utun_instance_destroy(inst); uasync_destroy(ua, 0);
+    u_free(source_snapshot);
     CHECK(u_get_allocated_count() == before);
     test_clock_finish();
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS Android stream: preview, source snapshot, FIN, cancellation, concurrent detach and no leaks");

tests/test_stream_service.c

@@ -420,6 +420,14 @@ static void test_network(void) {
     CHECK(stream_publish_open(nodes[3], GROUP1, &parameters, &parallel_publisher) == STREAM_OK);
     struct stream_info parallel_info; CHECK(stream_get_info(parallel_publisher, &parallel_info) == STREAM_OK);
     wait_state(1, 3, GROUP1, parallel_info.descriptor.source_epoch, 0);
+    /* Две независимые публикации одного канала видны через общий каталог; подписчики не становятся источниками. */
+    char *catalog = NULL, origin_id[64], parallel_id[64], subscriber_id[64];
+    CHECK(stream_sources_json(nodes[1], GROUP1, &catalog) == STREAM_OK);
+    snprintf(origin_id, sizeof(origin_id), "\"id\":\"%llu\"", (unsigned long long)nodes[0]->node_id);
+    snprintf(parallel_id, sizeof(parallel_id), "\"id\":\"%llu\"", (unsigned long long)nodes[3]->node_id);
+    snprintf(subscriber_id, sizeof(subscriber_id), "\"id\":\"%llu\"", (unsigned long long)nodes[1]->node_id);
+    CHECK(strstr(catalog, origin_id) && strstr(catalog, parallel_id) && !strstr(catalog, subscriber_id)); u_free(catalog);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] shared catalogue exposes two live publishers in the same channel");
     CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, &parallel_subscriber) == STREAM_OK);
     size_t accepted = 0;
     CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));

tools/chatgui-android/app/src/main/java/com/utun/chat/MainActivity.kt

@@ -394,7 +394,8 @@ class MainActivity : AppCompatActivity() {
                                         },
                                         onInvite = { showInvite(channel) },
                                         onBack = { screen = Screen.ChannelList },
-                                        onStreams = { screen = Screen.Streams },
+                                        onStreams = { vm.selectStream(null); screen = Screen.Streams },
+                                        onWatchStream = { vm.selectStream(it); screen = Screen.Streams },
                                         onPlayVideo = { msg -> videoMessage = msg },
                                         onViewPhoto = { msg -> photoMessage = msg },
                                         onRecordVideo = {
@@ -412,7 +413,12 @@ class MainActivity : AppCompatActivity() {
                             }
                             Screen.Streams -> {
                                 vm.currentChannel.collectAsState().value?.let { channel ->
-                                    StreamScreen(channel.id, channel.name) { screen = Screen.Chat }
+                                    val sources by vm.streamSources.collectAsState()
+                                    val selected by vm.streamSelection.collectAsState()
+                                    val failed by vm.streamSourcesFailed.collectAsState()
+                                    StreamScreen(channel.id, channel.name, sources, selected, failed, vm::selectStream) {
+                                        vm.selectStream(null); screen = Screen.Chat
+                                    }
                                 }
                             }
                             Screen.MemberList -> {

tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt

@@ -199,20 +199,22 @@ object NativeLib {
 
     /* Live media используют отдельную ограниченную очередь, минуя чатовый event bus. */
     fun streamOpen(id: Long, group: Long, source: Long = 0, width: Int = 0, height: Int = 0,
-                   rotation: Int = 0, audio: Boolean = true): Boolean = nativeStreamOpen(id, group, source, width, height, rotation, audio)
+                   rotation: Int = 0, audio: Boolean = true, epoch: Long = 0): Boolean =
+        nativeStreamOpen(id, group, source, width, height, rotation, audio, epoch)
+    fun streamWatch(group: Long): Boolean = nativeStreamWatch(group)
     fun streamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray): Int = nativeStreamFeed(id, video, pts, key, data)
     fun streamStop(id: Long) { nativeStreamStop(id) }
     fun streamPull(id: Long): ByteArray? = nativeStreamPull(id)
-    fun streamSources(group: Long): String = nativeStreamSources(group) ?: "[]"
     fun streamOpusCreate(encode: Boolean): Long = nativeStreamOpusCreate(encode)
     fun streamOpusEncode(handle: Long, pcm: ShortArray): ByteArray? = nativeStreamOpusEncode(handle, pcm)
     fun streamOpusDecode(handle: Long, packet: ByteArray): ShortArray? = nativeStreamOpusDecode(handle, packet)
     fun streamOpusClose(handle: Long) { nativeStreamOpusClose(handle) }
-    private external fun nativeStreamOpen(id: Long, group: Long, source: Long, width: Int, height: Int, rotation: Int, audio: Boolean): Boolean
+    private external fun nativeStreamOpen(id: Long, group: Long, source: Long, width: Int, height: Int, rotation: Int,
+                                         audio: Boolean, epoch: Long): Boolean
+    private external fun nativeStreamWatch(group: Long): Boolean
     private external fun nativeStreamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray): Int
     private external fun nativeStreamStop(id: Long)
     private external fun nativeStreamPull(id: Long): ByteArray?
-    private external fun nativeStreamSources(group: Long): String?
     private external fun nativeStreamOpusCreate(encode: Boolean): Long
     private external fun nativeStreamOpusEncode(handle: Long, pcm: ShortArray): ByteArray?
     private external fun nativeStreamOpusDecode(handle: Long, packet: ByteArray): ShortArray?

tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt

@@ -20,7 +20,8 @@ import kotlin.math.roundToInt
 /** Один worker владеет декодерами и AudioTrack. Hardware audio clock управляет временем показа видео.
  * Просроченные P-кадры декодируются без показа, чтобы сохранить references; при переполнении ждём KEY.
  * viewerLock сериализует stop/open разных экранов до подтверждения освобождения native owner. */
-class StreamPlayerEngine(private val group: Long, private val source: Long, private val muteAudio: Boolean = false) {
+class StreamPlayerEngine(private val group: Long, private val source: Long, private val epoch: Long,
+                         private val muteAudio: Boolean = false) {
     companion object { private val viewerLock = ReentrantLock() }
     private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int, val rotation: Int, val data: ByteArray)
     private val mutableAspect = MutableStateFlow(16f / 9f)
@@ -120,7 +121,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
             }
         }
         try {
-            check(NativeLib.streamOpen(id, group, source)) { "Viewer open refused" }
+            check(NativeLib.streamOpen(id, group, source, epoch = epoch)) { "Viewer open refused" }
             val openDeadline = System.nanoTime() + 10_000_000_000
             var opened = false
             var prefillUntil = 0L
@@ -131,7 +132,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                     if (raw != null) {
                         check(raw.size >= 20)
                         val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN)
-                        val kind = b.get().toInt(); val key = b.get().toInt() and 1 != 0
+                        val kind = b.get().toInt(); val flags = b.get().toInt() and 255; val key = flags and 1 != 0
                         val width = b.short.toInt() and 65535; val height = b.short.toInt() and 65535
                         val rotation = b.short.toInt() and 65535; val pts = b.long; val size = b.int
                         check(size == b.remaining())
@@ -161,7 +162,17 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                             3 -> { LogManager.addLog("INFO", "stream", "viewer GAP id=$id"); resetVideo(); resetAudio(); videoBase = -1; prefillUntil = 0 }
                             4 -> opened = true
                             5 -> ended = true
-                            6 -> error("Native viewer closed")
+                            6 -> {
+                                mutableStatus.value = when (flags) {
+                                    1 -> R.string.stream_group_closed
+                                    2 -> R.string.chat_service_stopped
+                                    3 -> R.string.stream_source_changed
+                                    4 -> R.string.stream_member_removed
+                                    else -> R.string.stream_failed
+                                }
+                                LogManager.addLog("WARN", "stream", "native viewer closed id=$id group=$group source=$source epoch=$epoch reason=$flags")
+                                ended = true; running = false
+                            }
                             else -> error("Unknown media event=$kind")
                         }
                     }

tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamSource.kt

@@ -0,0 +1,21 @@
+package com.utun.chat.data
+
+import org.json.JSONArray
+
+/** Источник идентифицируется участником и эпохой; IDs сохраняют все 64 бита. */
+data class StreamSource(val id: Long, val name: String, val epoch: Long, val self: Boolean) {
+    val displayName: String get() = name.ifEmpty { id.toULong().toString() }
+
+    companion object {
+        fun parse(json: String): List<StreamSource> {
+            val array = JSONArray(json)
+            return (0 until array.length()).map { i ->
+                val item = array.getJSONObject(i)
+                val id = item.getString("id").toULong().toLong()
+                val epoch = item.getString("epoch").toULong(16).toLong()
+                require(id != 0L && epoch != 0L) { "Zero stream identity" }
+                StreamSource(id, item.getString("name"), epoch, item.getInt("self") == 1)
+            }
+        }
+    }
+}

tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/ChatScreen.kt

@@ -15,6 +15,7 @@ import androidx.compose.foundation.combinedClickable
 import androidx.compose.foundation.ExperimentalFoundationApi
 import androidx.compose.foundation.gestures.awaitFirstDown
 import androidx.compose.foundation.horizontalScroll
+import androidx.compose.foundation.verticalScroll
 import androidx.compose.foundation.layout.*
 import androidx.compose.foundation.lazy.rememberLazyListState
 import androidx.compose.foundation.rememberScrollState
@@ -70,6 +71,8 @@ import com.utun.chat.ui.components.GroupAvatar
 import com.utun.chat.ui.components.GroupPropertiesDialog
 import com.utun.chat.data.ConversationTarget
 import com.utun.chat.viewmodel.ChatViewModel
+import com.utun.chat.data.StreamSource
+import androidx.compose.material.icons.filled.LiveTv
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 
@@ -89,7 +92,8 @@ fun ChatScreen(
     onViewPhoto: (Message) -> Unit = {},
     onRecordVideo: () -> Unit = {},
     onCapturePhoto: () -> Unit = {},
-    onStreams: () -> Unit = {}
+    onStreams: () -> Unit = {},
+    onWatchStream: (StreamSource) -> Unit = {}
 ) {
     val listState = rememberLazyListState()
     val context = LocalContext.current
@@ -98,6 +102,23 @@ fun ChatScreen(
     val scope = rememberCoroutineScope()
     var properties by remember(channel.id) { mutableStateOf(false) }
     var menu by remember(channel.id) { mutableStateOf(false) }
+    val streams by viewModel.streamSources.collectAsState()
+    var chooseStream by remember(channel.id) { mutableStateOf(false) }
+    if (chooseStream && streams.size > 1) AlertDialog(
+        onDismissRequest = { chooseStream = false },
+        title = { Text(stringResource(R.string.stream_choose_source)) },
+        text = {
+            Column(Modifier.heightIn(max = 360.dp).verticalScroll(rememberScrollState())) {
+                streams.forEach { source ->
+                    TextButton(onClick = { chooseStream = false; onWatchStream(source) }) {
+                        Text("${source.displayName} · ${source.id.toULong()}")
+                    }
+                }
+            }
+        },
+        confirmButton = { TextButton(onClick = { chooseStream = false }) { Text(stringResource(R.string.cancel)) } }
+    )
+    LaunchedEffect(streams) { if (streams.size <= 1) chooseStream = false }
 
     /* рация */
     val radioActive by viewModel.radioActive.collectAsState()
@@ -172,6 +193,13 @@ fun ChatScreen(
                 IconButton(onClick = onBack) { Icon(Icons.AutoMirrored.Filled.ArrowBack, stringResource(R.string.back)) }
             },
             actions = {
+                if (streams.isNotEmpty()) IconButton(onClick = {
+                    if (streams.size == 1) onWatchStream(streams.single()) else chooseStream = true
+                }) {
+                    BadgedBox(badge = { if (streams.size > 1) Badge { Text(streams.size.toString()) } }) {
+                        Icon(Icons.Filled.LiveTv, stringResource(R.string.stream_available, streams.size), tint = Color(0xFFE53935))
+                    }
+                }
                 IconButton(
                     onClick = { viewModel.toggleRadio() },
                     modifier = Modifier.background(

tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamScreen.kt

@@ -28,27 +28,20 @@ import androidx.lifecycle.Lifecycle
 import androidx.lifecycle.LifecycleEventObserver
 import com.utun.chat.R
 import com.utun.chat.data.LogManager
-import com.utun.chat.data.NativeLib
 import com.utun.chat.data.StreamController
 import com.utun.chat.data.StreamPhase
 import com.utun.chat.data.StreamPlayerEngine
-import kotlinx.coroutines.Dispatchers
+import com.utun.chat.data.StreamSource
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.withContext
-import org.json.JSONArray
-
-private data class LiveSource(val id: Long, val name: String, val epoch: String, val self: Boolean)
 
 /** Выход с экрана закрывает только reader; foreground publisher останавливает явная кнопка/notification. */
 @OptIn(ExperimentalMaterial3Api::class)
 @Composable
-fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) {
+fun StreamScreen(channelId: String, channelName: String, sources: List<StreamSource>, selected: StreamSource?,
+                 sourcesFailed: Boolean, onSelect: (StreamSource?) -> Unit, onBack: () -> Unit) {
     val context = LocalContext.current
     val group = remember(channelId) { channelId.toULongOrNull()?.toLong() ?: 0L }
     val state by StreamController.state.collectAsState()
     val local = state?.takeIf { it.group == group }
-    var sources by remember(group) { mutableStateOf<List<LiveSource>>(emptyList()) }
-    var selected by remember(group) { mutableStateOf<LiveSource?>(null) }
     var previewedSession by remember(group) { mutableStateOf<Long?>(null) }
     var audio by remember { mutableStateOf(true) }
     var camera by remember { mutableStateOf("") }
@@ -64,24 +57,10 @@ fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) {
     val cameraGranted = permissionVersion.let { ContextCompat.checkSelfPermission(context, Manifest.permission.CAMERA) == PackageManager.PERMISSION_GRANTED }
     val micGranted = permissionVersion.let { ContextCompat.checkSelfPermission(context, Manifest.permission.RECORD_AUDIO) == PackageManager.PERMISSION_GRANTED }
 
-    LaunchedEffect(group) {
-        while (true) {
-            if (NativeLib.isReady()) sources = withContext(Dispatchers.IO) {
-                try {
-                    val json = JSONArray(NativeLib.streamSources(group))
-                    (0 until json.length()).map { i ->
-                        val member = json.getJSONObject(i)
-                        LiveSource(member.getString("id").toULong().toLong(), member.getString("name"), member.getString("epoch"), member.optInt("self") == 1)
-                    }
-                } catch (e: Exception) { LogManager.addLog("WARN", "stream", "stream list: ${e.message}"); emptyList() }
-            } else sources = emptyList()
-            delay(1000)
-        }
-    }
     LaunchedEffect(sources, local?.id, local?.phase) {
-        if (local != null && !local.ongoing && selected?.self == true) selected = null
+        if (local != null && !local.ongoing && selected?.self == true) onSelect(null)
         if (local?.phase == StreamPhase.LIVE && previewedSession != local.id) sources.firstOrNull { it.self }?.let {
-            selected = it; previewedSession = local.id
+            onSelect(it); previewedSession = local.id
         }
     }
 
@@ -92,6 +71,16 @@ fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) {
     }) { padding ->
         Column(Modifier.fillMaxSize().padding(padding).padding(12.dp), verticalArrangement = Arrangement.spacedBy(8.dp)) {
             Text(channelName, style = MaterialTheme.typography.titleMedium)
+            if (selected != null) {
+                Text(selected.displayName, style = MaterialTheme.typography.titleSmall)
+                Box(Modifier.weight(1f).fillMaxWidth()) {
+                    key(group, selected.id, selected.epoch) { LiveStreamView(group, selected.id, selected.epoch, selected.self) }
+                }
+                TextButton(onClick = { onSelect(null); onBack() }) { Text(stringResource(R.string.stream_close_view)) }
+                if (selected.self && local?.ongoing == true) Button(onClick = StreamController::stop,
+                    enabled = local.phase != StreamPhase.STOPPING) { Text(stringResource(R.string.stream_stop)) }
+                return@Column
+            }
             if (local?.ongoing == true) {
                 Text(stringResource(when (local.phase) {
                     StreamPhase.LIVE -> R.string.stream_live
@@ -117,24 +106,19 @@ fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) {
                 }, enabled = camera.isNotEmpty() && group != 0L && state?.ongoing != true) { Text(stringResource(R.string.stream_start)) }
             }
             local?.error?.let { Text(stringResource(it), color = MaterialTheme.colorScheme.error) }
-            if (sources.isEmpty() && local?.ongoing != true) Text(stringResource(R.string.stream_none))
+            if (sourcesFailed) Text(stringResource(R.string.stream_list_failed), color = MaterialTheme.colorScheme.error)
+            else if (sources.isEmpty() && local?.ongoing != true) Text(stringResource(R.string.stream_none))
             Column(Modifier.fillMaxWidth().heightIn(max = 160.dp).verticalScroll(rememberScrollState())) {
                 sources.forEach { source ->
-                    TextButton(onClick = { selected = source }) { Text(stringResource(R.string.stream_watch, source.name.ifEmpty { source.id.toULong().toString() })) }
+                    TextButton(onClick = { onSelect(source) }) { Text(stringResource(R.string.stream_watch, source.displayName)) }
                 }
             }
-            selected?.let { source ->
-                Box(Modifier.weight(1f).fillMaxWidth()) {
-                    key(group, source.id, source.epoch) { LiveStreamView(group, source.id, source.self) }
-                }
-                TextButton(onClick = { selected = null }) { Text(stringResource(R.string.stream_close_view)) }
-            }
         }
     }
 }
 
 @Composable
-private fun LiveStreamView(group: Long, source: Long, localPreview: Boolean) {
+private fun LiveStreamView(group: Long, source: Long, epoch: Long, localPreview: Boolean) {
     val owner = LocalLifecycleOwner.current
     var visible by remember(owner) { mutableStateOf(owner.lifecycle.currentState.isAtLeast(Lifecycle.State.STARTED)) }
     DisposableEffect(owner) {
@@ -145,12 +129,12 @@ private fun LiveStreamView(group: Long, source: Long, localPreview: Boolean) {
         owner.lifecycle.addObserver(observer)
         onDispose { owner.lifecycle.removeObserver(observer) }
     }
-    if (visible) LiveStreamSurface(group, source, localPreview)
+    if (visible) LiveStreamSurface(group, source, epoch, localPreview)
 }
 
 @Composable
-private fun LiveStreamSurface(group: Long, source: Long, localPreview: Boolean) {
+private fun LiveStreamSurface(group: Long, source: Long, epoch: Long, localPreview: Boolean) {
-    val engine = remember(group, source, localPreview) { StreamPlayerEngine(group, source, localPreview) }
+    val engine = remember(group, source, epoch, localPreview) { StreamPlayerEngine(group, source, epoch, localPreview) }
     val aspect by engine.aspect.collectAsState()
     val status by engine.status.collectAsState()
     var surface by remember { mutableStateOf<Surface?>(null) }

tools/chatgui-android/app/src/main/java/com/utun/chat/viewmodel/ChatViewModel.kt

@@ -220,6 +220,23 @@ class ChatViewModel : ViewModel() {
 
     private val _currentChannel = MutableStateFlow<Channel?>(null)
     val currentChannel: StateFlow<Channel?> = _currentChannel
+    private val _streamSources = MutableStateFlow<List<StreamSource>>(emptyList())
+    val streamSources: StateFlow<List<StreamSource>> = _streamSources
+    private val _streamSourcesFailed = MutableStateFlow(false)
+    val streamSourcesFailed: StateFlow<Boolean> = _streamSourcesFailed
+    private val _streamSelection = MutableStateFlow<StreamSource?>(null)
+    val streamSelection: StateFlow<StreamSource?> = _streamSelection
+
+    fun selectStream(source: StreamSource?) {
+        _streamSelection.value = source
+        LogManager.addLog("INFO", "stream", "selected channel=${_currentChannel.value?.id} source=${source?.id} epoch=${source?.epoch}")
+    }
+
+    private fun watchStreams() {
+        if (!NativeLib.isReady()) return
+        val group = if (_foreground.value) _currentChannel.value?.id?.toULongOrNull()?.toLong() ?: 0L else 0L
+        if (!NativeLib.streamWatch(group)) LogManager.addLog("WARN", "stream", "source watch rejected group=$group")
+    }
 
     private val _soundOn = MutableStateFlow(true)
     val soundOn: StateFlow<Boolean> = _soundOn
@@ -320,6 +337,7 @@ class ChatViewModel : ViewModel() {
 
     fun setForeground(fg: Boolean) {
         _foreground.value = fg
+        watchStreams()
         if (fg) {
             _isStuck.value = false
             if (dbReady && repo != null) {
@@ -533,8 +551,9 @@ class ChatViewModel : ViewModel() {
                 if (result == 0) refreshChannels()
             }
             4 -> { if (repo != null) { LogManager.addLog("INFO", "VM", "CHANNEL_UPDATED — refreshChannels"); refreshChannels() } else { LogManager.addLog("WARN", "VM", "CHANNEL_UPDATED skipped — repo is NULL") } }
-            12 -> LogManager.addLog("INFO", "VM", "service started")
+            12 -> { LogManager.addLog("INFO", "VM", "service started"); watchStreams() }
             13 -> {
+                _streamSources.value = emptyList(); _streamSourcesFailed.value = false; _streamSelection.value = null
                 for (id in submissions.keys.toList()) submissionResult(id, false)
                 _radioTalkState.value = RadioTalkState()
                 _radioTalking.value = null
@@ -641,6 +660,21 @@ class ChatViewModel : ViewModel() {
                 val cur = _currentChannel.value
                 if (cur != null) refreshMembers(cur.id)
             }
+            58 -> { /* STREAM_SOURCES: group:8BE, valid:1, JSON. Старый канал не меняет новую шапку. */
+                if (data == null || data.size < 11 || data[8].toInt() !in 0..1) {
+                    LogManager.addLog("ERROR", "stream", "invalid source snapshot bytes=${data?.size}"); return
+                }
+                val group = java.nio.ByteBuffer.wrap(data).order(java.nio.ByteOrder.BIG_ENDIAN).long
+                if (_currentChannel.value?.id?.toULongOrNull()?.toLong() != group || !_foreground.value) return
+                try {
+                    _streamSourcesFailed.value = data[8].toInt() == 0
+                    _streamSources.value = StreamSource.parse(String(data, 9, data.size - 9, Charsets.UTF_8))
+                    LogManager.addLog("INFO", "stream", "sources group=$group count=${_streamSources.value.size} valid=${!_streamSourcesFailed.value}")
+                } catch (e: Exception) {
+                    _streamSources.value = emptyList(); _streamSourcesFailed.value = true
+                    LogManager.addLog("ERROR", "stream", "source snapshot parse group=$group: ${e.message}")
+                }
+            }
             19 -> { if (repo != null) refreshLocalSockets() }  /* LOCAL_SOCKETS */
             23 -> { /* MEMBER_LIST: live-снимок дополняет расширенный JSON-список. */
                 if (data == null || data.size < 3) {
@@ -1215,6 +1249,7 @@ class ChatViewModel : ViewModel() {
     fun selectChannel(channel: Channel) {
         val r = repo ?: return
         if (_currentChannel.value?.id != channel.id) {
+            _streamSources.value = emptyList(); _streamSourcesFailed.value = false; _streamSelection.value = null
             _historyAnchor.value = ""
             _messageJump.value = null
             pendingLatestJump = null
@@ -1224,6 +1259,7 @@ class ChatViewModel : ViewModel() {
             memberDetailJob?.cancel(); memberDetailTarget = null; _selectedMember.value = null
         }
         _currentChannel.value = channel
+        watchStreams()
         _radioActive.value = (radioChannelId == channel.id)
         _radioTalking.value = if (_radioActive.value) _radioTalkState.value.speakers.firstOrNull() else null
         _soundOn.value = NativeLib.getSoundOnMessage(channel.id)
@@ -1783,6 +1819,7 @@ class ChatViewModel : ViewModel() {
         _channels.value = emptyList()
         _messages.value = emptyList()
         _currentChannel.value = null
+        _streamSources.value = emptyList(); _streamSelection.value = null; _streamSourcesFailed.value = false
         _generatingKeys.value = false
         _localSockets.value = emptyList()
         _channelOwner.value = false

tools/chatgui-android/app/src/main/res/values-ru/strings.xml

@@ -1,6 +1,12 @@
 <?xml version="1.0" encoding="utf-8"?>
 <resources>
     <string name="stream_title">Трансляции</string>
+    <string name="stream_available">Трансляции: %1$d</string>
+    <string name="stream_choose_source">Выберите трансляцию</string>
+    <string name="stream_list_failed">Не удалось получить список трансляций</string>
+    <string name="stream_source_changed">Трансляция перезапущена или завершена. Выберите источник заново.</string>
+    <string name="stream_member_removed">Источник больше не является участником канала</string>
+    <string name="stream_group_closed">Канал закрыт</string>
     <string name="stream_start">Начать трансляцию</string>
     <string name="stream_stop">Остановить трансляцию</string>
     <string name="stream_live">Прямая трансляция</string>

tools/chatgui-android/app/src/main/res/values/strings.xml

@@ -1,6 +1,12 @@
 <?xml version="1.0" encoding="utf-8"?>
 <resources>
     <string name="stream_title">Live streams</string>
+    <string name="stream_available">Live streams: %1$d</string>
+    <string name="stream_choose_source">Choose a stream</string>
+    <string name="stream_list_failed">Could not load live streams</string>
+    <string name="stream_source_changed">The stream has restarted or ended. Select a source again.</string>
+    <string name="stream_member_removed">The source is no longer a channel member</string>
+    <string name="stream_group_closed">The channel is closed</string>
     <string name="stream_start">Start stream</string>
     <string name="stream_stop">Stop stream</string>
     <string name="stream_live">Streaming live</string>

tools/chatgui-android/jni_bridge/android_stream_jni.c

@@ -11,9 +11,14 @@
 struct android_stream_opus { opus_codec_encoder_t *encoder; opus_codec_decoder_t *decoder; };
 
 JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamOpen(
-    JNIEnv *env, jobject obj, jlong id, jlong group, jlong source, jint width, jint height, jint rotation, jboolean audio) {
+    JNIEnv *env, jobject obj, jlong id, jlong group, jlong source, jint width, jint height, jint rotation, jboolean audio, jlong epoch) {
     (void)env; (void)obj;
-    return android_stream_open((uint64_t)id, (uint64_t)group, (uint64_t)source, width, height, rotation, audio) == 0;
+    return android_stream_open((uint64_t)id, (uint64_t)group, (uint64_t)source, width, height, rotation, audio, (uint64_t)epoch) == 0;
+}
+
+JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamWatch(JNIEnv *env, jobject obj, jlong group) {
+    (void)env; (void)obj;
+    return android_stream_watch((uint64_t)group) == 0;
 }
 
 JNIEXPORT jint JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamFeed(
@@ -32,15 +37,6 @@ JNIEXPORT void JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamStop(JNIEnv
     (void)env; (void)obj; android_stream_stop((uint64_t)id);
 }
 
-JNIEXPORT jstring JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamSources(JNIEnv *env, jobject obj, jlong group) {
-    (void)obj;
-    char *json = android_stream_sources((uint64_t)group);
-    if (!json) return NULL;
-    jstring output = (*env)->NewStringUTF(env, json); u_free(json);
-    if (!output) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI stream source list string allocation failed");
-    return output;
-}
-
 JNIEXPORT jbyteArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPull(JNIEnv *env, jobject obj, jlong id) {
     (void)obj;
     size_t length = 0; uint8_t *data = android_stream_pull((uint64_t)id, &length);

tools/chatgui-android/libutun_lite/stream_bridge.c

@@ -1,15 +1,11 @@
 #include <pthread.h>
-#include <stdio.h>
 #include <string.h>
 
 #include "../../../lib/debug_config.h"
 #include "../../../lib/mem.h"
 #include "../../../lib/u_async.h"
-#include "../../../lib/sqlite3.h"
-#include "../../../lib/strbuf.h"
 #include "../../../src/utun_instance.h"
 #include "../../../src/chat/chat_core.h"
-#include "../../../src/routing_layer/topo_group.h"
 #include "../../../src/stream/stream.h"
 #include "../../../src/stream/stream_member.h"
 #include "../../../src/stream/stream_media.h"
@@ -30,11 +26,11 @@ struct media_record {
 };
 struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned count; };
 struct android_media_session {
-    uint64_t id, group, source, epoch, next_frame, last_pts[2], drain_until;
+    uint64_t id, group, source, epoch, expected_epoch, next_frame, last_pts[2], drain_until;
     struct STREAM_HANDLE *handle;
     struct stream_media_receiver *receiver;
     struct stream_media_format format;
-    int opening, stopping, finished, need_key, seen[2], rx_overflow;
+    int opening, stopping, finished, need_key, seen[2], rx_overflow, reason;
 };
 
 static pthread_mutex_t media_lock = PTHREAD_MUTEX_INITIALIZER;
@@ -43,8 +39,8 @@ static struct posted_task *media_post;
 static void *media_timer;
 static struct android_media_session publisher, viewer;
 static struct media_queue video_queue, audio_queue, output_queue;
-static uint64_t list_group, cached_group, list_updated;
+static uint64_t watch_group;
-static char *source_list;
+static int watch_pending;
 static void media_process(void *arg);
 static void media_close(struct android_media_session *s, int kind);
 
@@ -146,6 +142,7 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e
         } else if (event->type == STREAM_EVENT_GAP) {
             stream_media_receiver_gap(viewer.receiver); media_output(viewer.id, 3, NULL, 0, 0, NULL, 0);
         } else {
+            viewer.reason = event->reason;
             media_close(&viewer, event->type == STREAM_EVENT_END ? 5 : 6);
         }
     }
@@ -154,7 +151,7 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e
 
 /* Publisher держит конечное окно выдачи после FIN; abrupt detach ничего не обещает получателям. */
 static void media_close(struct android_media_session *s, int kind) {
-    if (kind) media_output(s->id, kind, NULL, 0, 0, NULL, 0);
+    if (kind) media_output(s->id, kind, NULL, 0, kind == 6 ? s->reason : 0, NULL, 0);
     if (s->handle) stream_close(s->handle);
     stream_media_receiver_destroy(s->receiver);
     if (s == &publisher) { queue_clear(&video_queue); queue_clear(&audio_queue); }
@@ -196,69 +193,48 @@ static int media_publish(void) {
     return STREAM_OK;
 }
 
-/* Список содержит только проверенные публикации живых мемберов; SQL и BGP читаются в uasync. */
-static void media_list_sources(void) {
-    uint64_t group_id = list_group; list_group = 0;
-    struct TOPO_GROUP *group = media_inst->topo_groups ? topo_groups_find(media_inst->topo_groups, group_id) : NULL;
-    struct strbuf json = strbuf_new();
-    sqlite3_stmt *statement = NULL;
-    int failed = strbuf_addf(&json, "[") < 0;
-    if (group && !failed) {
-        char query[256];
-        for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
-            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source namespace invalid group=%llx", (unsigned long long)group_id);
-            failed = 1; break;
-        }
-        snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') "
-                 "FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id", group->channel_id);
-        sqlite3 *db = chat_core_get_db(media_inst);
-        int result = failed ? SQLITE_ERROR : sqlite3_prepare_v2(db, query, -1, &statement, NULL), count = 0;
-        if (result == SQLITE_OK) {
-            while ((result = sqlite3_step(statement)) == SQLITE_ROW) {
-                uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0);
-                if (!stream_member_online(media_inst, group_id, node)) continue;
-                struct stream_member_state state;
-                if (stream_member_load(media_inst, group_id, node, &state, NULL) != STREAM_OK || !state.published) continue;
-                if (count >= 256) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream source list truncated group=%llx", (unsigned long long)group_id); break; }
-                if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "",
-                    (unsigned long long)node, (unsigned long long)state.source.source_epoch, node == media_inst->node_id) < 0
-                    || strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0
-                    || strbuf_addf(&json, "}") < 0) { failed = 1; break; }
-                count++;
-            }
-        }
-        if (result != SQLITE_DONE && !(result == SQLITE_ROW && (failed || count == 256))) {
-            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream list query failed group=%llx: %s",
-                        (unsigned long long)group_id, sqlite3_errmsg(db)); failed = 1;
-        }
-    }
-    sqlite3_finalize(statement);
-    if (strbuf_addf(&json, "]") < 0) failed = 1;
-    u_free(source_list); source_list = failed ? NULL : strbuf_detach(&json);
-    strbuf_free(&json); cached_group = group_id; list_updated = get_time_us();
-    if (failed) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source snapshot failed group=%llx", (unsigned long long)group_id);
-}
-
 /* Один actor обслуживает команды и не больше двух FEC-пачек за вызов; 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 (list_group) media_list_sources();
+    if (watch_pending) {
+        watch_pending = 0;
+        if (stream_watch_sources(media_inst, watch_group) != STREAM_OK)
+            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source watch failed group=%llx", (unsigned long long)watch_group);
+    }
     struct android_media_session *sessions[] = {&publisher, &viewer};
     for (unsigned i = 0; i < 2; i++) {
         struct android_media_session *s = sessions[i];
         if (s->opening) {
             if (s->stopping) { media_close(s, 5); continue; }
             s->opening = 0;
-            int result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle)
+            int result = STREAM_OK;
-                                   : stream_publish_open(media_inst, s->group, NULL, &s->handle);
+            s->reason = STREAM_CLOSE_FAILED;
+            if (s->source && s->expected_epoch) {
+                struct stream_member_state state;
+                result = stream_member_load(media_inst, s->group, s->source, &state, NULL);
+                if (result == STREAM_OK && (!state.published || state.source.source_epoch != s->expected_epoch)) {
+                    DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android selected source changed group=%llx source=%llx expected=%llx actual=%llx",
+                        (unsigned long long)s->group, (unsigned long long)s->source, (unsigned long long)s->expected_epoch,
+                        (unsigned long long)(state.published ? state.source.source_epoch : 0));
+                    result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
+                }
+            }
+            if (result == STREAM_OK) result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle)
+                                                       : stream_publish_open(media_inst, s->group, NULL, &s->handle);
             if (result == STREAM_OK && s->source) result = stream_media_receiver_create(&s->receiver, media_frame, s);
             struct stream_info info;
             if (result == STREAM_OK) {
                 result = stream_get_info(s->handle, &info);
-                if (result == STREAM_OK) s->epoch = info.descriptor.source_epoch;
+                if (result == STREAM_OK) {
+                    s->epoch = info.descriptor.source_epoch;
+                    if (s->expected_epoch && s->epoch != s->expected_epoch) {
+                        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android viewer opened stale epoch id=%llu", (unsigned long long)s->id);
+                        result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
+                    }
+                }
             }
             if (result != STREAM_OK) {
                 DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open failed id=%llu group=%llx source=%llx result=%d",
@@ -323,12 +299,13 @@ void android_stream_detach(struct UTUN_INSTANCE *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);
         stream_set_event_callback(inst, NULL, NULL);
         queue_clear(&output_queue);
         if (publisher.id) media_close(&publisher, 0);
         if (viewer.id) media_close(&viewer, 0);
         media_inst = NULL;
-        u_free(source_list); source_list = NULL; list_group = cached_group = list_updated = 0;
+        watch_pending = 0; watch_group = 0;
         DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected");
     }
     pthread_mutex_unlock(&media_lock);
@@ -336,9 +313,12 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) {
 
 /* Команда содержит идентификаторы, а не borrowed pointer ядра. */
 int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
-                        unsigned rotation, int audio) {
+                        unsigned rotation, int audio, uint64_t expected_epoch) {
-    if (!id || !group || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) {
+    if (!id || !group || (source && !expected_epoch)
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid arguments"); return -1;
+        || (!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;
@@ -346,13 +326,23 @@ int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned w
     if (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 (media_schedule() == 0) {
-        *s = (struct android_media_session){.id = id, .group = group, .source = source, .opening = 1, .need_key = 1,
+        *s = (struct android_media_session){.id = id, .group = group, .source = source, .expected_epoch = expected_epoch,
+            .opening = 1, .need_key = 1,
             .format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation, STREAM_MEDIA_H264, audio ? STREAM_MEDIA_OPUS : 0}};
         result = 0;
     }
     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) {
     if (!data || !length || length > (video ? STREAM_MEDIA_MAX_FRAME : STREAM_MEDIA_MAX_OPUS) || pts > INT64_MAX) {
@@ -420,16 +410,3 @@ uint8_t *android_stream_pull(uint64_t id, size_t *length) {
     }
     pthread_mutex_unlock(&media_lock); return data;
 }
-
-/* Внешний поток не читает ни borrowed instance, ни SQLite; snapshot строится следующим actor-вызовом. */
-char *android_stream_sources(uint64_t group) {
-    pthread_mutex_lock(&media_lock);
-    if (group && media_inst && uasync_is_running(media_inst->ua)
-        && (group != cached_group || get_time_us() - list_updated >= 500000)) {
-        list_group = group;
-        if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android source snapshot schedule failed");
-    }
-    char *copy = u_strdup(group == cached_group && source_list ? source_list : "[]");
-    if (!copy) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android source snapshot copy allocation failed");
-    pthread_mutex_unlock(&media_lock); return copy;
-}

tools/chatgui-android/libutun_lite/stream_bridge.h

@@ -3,7 +3,7 @@
  * Один publisher и один viewer, TX <=2MB/128 records, RX <=4MB/256 records.
  * Data принимает полные H.264 Annex B кадры (KEY содержит SPS/PPS/IDR) и Opus 48k mono/20ms.
  * pull передаёт владение u_malloc-буфером: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,data.
- * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6. Числа big-endian.
+ * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6; error.flags = stream_close_reason. Числа big-endian.
  * Медиаданные передаются через pull, не через общий чатовый event bus. */
 #ifndef ANDROID_STREAM_BRIDGE_H
 #define ANDROID_STREAM_BRIDGE_H
@@ -15,14 +15,14 @@ struct UTUN_INSTANCE;
 /* 0 attached, -1 invalid/busy: нельзя передать singleton другому ядру до detach старого. */
 int android_stream_attach(struct UTUN_INSTANCE *inst);
 void android_stream_detach(struct UTUN_INSTANCE *inst);
+/* source=0 — publisher; source!=0 — viewer с обязательным expected_epoch. Команда принята=0, результат через pull. */
 int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
-                        unsigned rotation, int audio);
+                        unsigned rotation, int audio, uint64_t expected_epoch);
-/* source=0 — publisher; source!=0 — viewer. 0 означает, что команда принята, результат приходит через pull. */
+/* Наблюдение выбранного канала через CHAT_EVT_STREAM_SOURCES; group=0 снимает его. */
+int android_stream_watch(uint64_t group);
 int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length);
 /* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */
 void android_stream_stop(uint64_t id);
 uint8_t *android_stream_pull(uint64_t id, size_t *length);
-/* Кеш live-снимка источников: запрос в actor, чтение только копии под mutex; caller освобождает u_free(). */
-char *android_stream_sources(uint64_t group);
 
 #endif

tools/chatgui/CMakeLists.txt

@@ -195,6 +195,7 @@ add_executable(vibechat
     src/media_blocks.cpp
     src/videoplayer_engine.cpp
     src/videoplayer_window.cpp
+    src/streamplayer_window.cpp
     src/imageviewer_window.cpp
     src/member_files/browser.cpp
     src/member_files/batch.cpp
@@ -216,6 +217,7 @@ add_executable(vibechat
     transport/node_config.cpp
     transport/config_updater.cpp
     transport/gui_bridge_impl.cpp
+    transport/stream_bridge.cpp
     transport/miniaudio_impl.c
     transport/audio_diagnostics.cpp
     db/db_manager.cpp
@@ -435,7 +437,7 @@ target_link_libraries(test_radio_member_mute PRIVATE ${QT_CORE} utun_voice utun
 add_test(NAME test_radio_member_mute COMMAND test_radio_member_mute)
 set(RADIO_MEMBER_TEST_SOURCES src/memberlistmodel.cpp src/avatars/avatarcache.cpp src/avatars/avatarmanager.cpp
     src/avatars/avatarcrop.cpp src/avatars/avatargallery.cpp src/imageviewer_window.cpp db/db_manager.cpp)
-add_executable(test_radio_mute_bridge tests/test_radio_mute_bridge.cpp transport/utun_node.cpp transport/gui_bridge_impl.cpp
+add_executable(test_radio_mute_bridge tests/test_radio_mute_bridge.cpp transport/utun_node.cpp transport/gui_bridge_impl.cpp transport/stream_bridge.cpp
     src/accountdelegate.cpp src/flagpainter.cpp ${RADIO_MEMBER_TEST_SOURCES})
 target_include_directories(test_radio_mute_bridge PRIVATE ${UTUN_INCLUDE_DIRS})
 target_link_libraries(test_radio_mute_bridge PRIVATE ${QT_LIBS} ${QT_TEST_LIB} utun_voice utun pthread)
@@ -500,6 +502,21 @@ else()
 endif()
 
 add_executable(test_ptt_key_chord tests/test_ptt_key_chord.cpp)
+add_executable(test_stream_view tests/test_stream_view.cpp src/streamplayer_window.cpp transport/stream_bridge.cpp
+    src/videoplayer_window.cpp src/videoplayer_engine.cpp src/sound_manager.cpp src/voice_audio_io.cpp src/audio_device.cpp
+    transport/miniaudio_impl.c transport/audio_diagnostics.cpp)
+target_include_directories(test_stream_view PRIVATE ${UTUN_INCLUDE_DIRS} ${FFMPEG_INCLUDE_DIRS})
+target_link_libraries(test_stream_view PRIVATE ${QT_LIBS} ${QT_TEST_LIB} utun_voice utun pthread ${CMAKE_DL_LIBS} ${FFMPEG_TARGET} ${FFMPEG_LIBRARIES})
+if(FFMPEG_FOUND)
+    target_compile_definitions(test_stream_view PRIVATE HAVE_FFMPEG)
+endif()
+if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
+    target_compile_definitions(test_stream_view PRIVATE TEST_NULL_AUDIO)
+    target_link_options(test_stream_view PRIVATE "-Wl,--wrap=ma_context_init" "-Wl,--wrap=ma_device_init")
+endif()
+add_test(NAME test_stream_view COMMAND test_stream_view)
+set_tests_properties(test_stream_view PROPERTIES TIMEOUT 30 ENVIRONMENT "QT_QPA_PLATFORM=offscreen")
+
 target_link_libraries(test_ptt_key_chord PRIVATE ${QT_CORE})
 add_test(NAME test_ptt_key_chord COMMAND test_ptt_key_chord)
 

tools/chatgui/src/mainwindow.cpp

@@ -1,5 +1,6 @@
 #include <QCoreApplication>
 #include "mainwindow.h"
+#include "streamplayer_window.h"
 #include "channellist.h"
 #include "messagelist.h"
 #include "messagedelegate.h"
@@ -197,6 +198,8 @@ MainWindow::MainWindow(QWidget *parent, DbManager* db, const QString& cfgPath,
 }
 
 MainWindow::~MainWindow() {
+    if (m_streamPlayer) delete m_streamPlayer.data();
+    gui_bridge_set_stream_sources_cb(nullptr);
     s_mainWindow = nullptr;
     m_cpuLoadTimer->stop();
     gui_bridge_set_cpu_load_cb(nullptr);
@@ -585,8 +588,11 @@ void MainWindow::setupBridgeCallbacks() {
             s_mainWindow->m_channelList->setCpuLoad(0);
             s_mainWindow->m_accountList->model()->setRadioState(0, false, false);
             s_mainWindow->m_messageList->clearDmCallRoute();
+            s_mainWindow->m_messageList->clearStreams();
+            if (s_mainWindow->m_streamPlayer) s_mainWindow->m_streamPlayer->close();
             s_mainWindow->m_channelList->clearDmPresence();
         } else {
+            if (type == 12) s_mainWindow->m_messageList->watchStreams();
             s_mainWindow->m_channelList->refreshDmPresence();
         }
     });
@@ -599,12 +605,21 @@ void MainWindow::setupBridgeCallbacks() {
         if (s_mainWindow)
             s_mainWindow->m_messageList->onDmMessages(data, len);
     });
+    gui_bridge_set_stream_sources_cb([](const uint8_t* data, int len) {
+        if (s_mainWindow) s_mainWindow->m_messageList->onStreamSources(data, len);
+    });
     gui_bridge_set_attachment_preparation_cb([](const uint8_t* data, int len) {
         if (s_mainWindow) s_mainWindow->m_messageList->onAttachmentPreparation(data, len);
     });
 }
 
 void MainWindow::setupConnects() {
+    connect(m_messageList, &MessageList::streamRequested, this,
+        [this](quint64 group, quint64 source, quint64 epoch, const QString& name) {
+            if (m_streamPlayer) delete m_streamPlayer.data();
+            m_streamPlayer = new StreamPlayerWindow(group, source, epoch, name);
+            m_streamPlayer->setAttribute(Qt::WA_DeleteOnClose); m_streamPlayer->show();
+        });
     connect(RadioAudioEngine::instance(), &RadioAudioEngine::stateChanged, this, [this] {
         auto* engine = RadioAudioEngine::instance();
         m_accountList->model()->setRadioState(engine->groupId(), engine->isActive(), engine->isTransmitting());

tools/chatgui/src/mainwindow.h

@@ -21,6 +21,7 @@ class UtunNode;
 class AudioRecorder;
 class ConnMonitorWindow;
 class VideoPlayerWindow;
+class StreamPlayerWindow;
 class ImageViewerWindow;
 class CallWindow;
 class CallAudioEngine;
@@ -124,6 +125,7 @@ private:
     AudioRecorder    *m_recorder = nullptr;
     ConnMonitorWindow *m_connMonitor = nullptr;
     VideoPlayerWindow *m_videoPlayer = nullptr;
+    QPointer<StreamPlayerWindow> m_streamPlayer;
     ImageViewerWindow *m_imageViewer = nullptr;
     QPointer<MemberFileBrowser> m_fileBrowser;
 

tools/chatgui/src/messagelist.cpp

@@ -1,5 +1,6 @@
 #include <QCoreApplication>
 #include "messagelist.h"
+#include "../transport/stream_bridge.h"
 #include "debug_ui.h"
 #include "chatview.h"
 #include "messagedelegate.h"
@@ -426,6 +427,30 @@ MessageList::MessageList(DbManager* db, AudioRecorder* recorder, QWidget *parent
         connect(m_radioBtn, &QPushButton::toggled, this, &MessageList::onRadioToggled);
         hl->addWidget(m_radioBtn);
 
+        m_streamBtn = new QPushButton(header); m_streamBtn->setObjectName("streamButton");
+        QPixmap streamIcon(24, 24); streamIcon.fill(Qt::transparent);
+        { QPainter painter(&streamIcon); painter.setRenderHint(QPainter::Antialiasing); painter.setPen(QPen(QColor("#e53935"), 2));
+          painter.drawRoundedRect(QRectF(2, 6, 20, 14), 2, 2); painter.drawLine(8, 1, 12, 5); painter.drawLine(16, 1, 12, 5);
+          painter.setBrush(QColor("#e53935")); painter.drawEllipse(QPointF(12, 13), 3, 3); }
+        m_streamBtn->setIcon(QIcon(streamIcon)); m_streamBtn->setIconSize(QSize(24, 24)); m_streamBtn->hide();
+        m_streamBtn->setAccessibleName(QCoreApplication::translate("ChatGui", "Live streams"));
+        m_streamBtn->setCursor(Qt::PointingHandCursor);
+        connect(m_streamBtn, &QPushButton::clicked, this, [this] {
+            auto watch = [this](const LiveSource& source) {
+                DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop selected stream group=%s source=%llx epoch=%llx",
+                           qPrintable(m_currentChannelId), (unsigned long long)source.id, (unsigned long long)source.epoch);
+                emit streamRequested(m_currentChannelId.toULongLong(), source.id, source.epoch, source.name);
+            };
+            if (m_streamSources.size() == 1) { watch(m_streamSources.front()); return; }
+            QMenu menu(this);
+            for (const auto& source : m_streamSources) {
+                auto* action = menu.addAction(QStringLiteral("%1 · %2").arg(source.name).arg(source.id));
+                connect(action, &QAction::triggered, this, [watch, source] { watch(source); });
+            }
+            menu.exec(m_streamBtn->mapToGlobal(QPoint(0, m_streamBtn->height())));
+        });
+        hl->addWidget(m_streamBtn);
+
         m_groupMenuButton = new QPushButton(QStringLiteral("…"), header);
         m_groupMenuButton->setObjectName("groupMenuButton");
         m_groupMenuButton->setFixedSize(34, 30);
@@ -749,6 +774,7 @@ void MessageList::loadChannel(const QString& channelId) {
     m_avatarCache.clear();
     if (m_currentChannelId != channelId) { m_feedback.clear(); m_pendingPoll.clear(); }
     m_currentChannelId = channelId;
+    clearStreams(); watchStreams();
     m_groupMenuButton->setVisible(channelId.toULongLong() != 0);
     clearDmCallRoute();
     m_callBtn->setVisible(isDmChannel(channelId));
@@ -1451,6 +1477,43 @@ void MessageList::onRadioTalk(quint64 groupId, quint64 srcNodeId, bool on) {
     }
 }
 
+/* Каталог выбранного канала не удерживает media owner; DM/пустой target снимает наблюдение. */
+void MessageList::watchStreams() {
+    gui_stream_watch(isDmChannel(m_currentChannelId) ? 0 : m_currentChannelId.toULongLong());
+}
+
+void MessageList::clearStreams() { m_streamSources.clear(); m_streamBtn->hide(); }
+
+/* Полный снимок применяется атомарно; ответы предыдущего канала не меняют текущую шапку. */
+void MessageList::onStreamSources(const uint8_t* data, int len) {
+    if (!data || len < 11 || data[8] > 1) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source event invalid len=%d", len); return; }
+    quint64 group = stream_read_be(data, 8);
+    if (isDmChannel(m_currentChannelId) || group != m_currentChannelId.toULongLong()) return;
+    QJsonParseError error;
+    auto document = QJsonDocument::fromJson(QByteArray(reinterpret_cast<const char*>(data + 9), len - 9), &error);
+    if (!data[8] || error.error != QJsonParseError::NoError || !document.isArray()) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source list failed group=%llx valid=%u parse=%s",
+                    (unsigned long long)group, data[8], qPrintable(error.errorString()));
+        clearStreams(); return;
+    }
+    QList<LiveSource> sources;
+    for (const auto& value : document.array()) {
+        auto member = value.toObject(); bool idOk = false, epochOk = false;
+        quint64 id = member.value("id").toString().toULongLong(&idOk);
+        quint64 epoch = member.value("epoch").toString().toULongLong(&epochOk, 16);
+        if (!idOk || !epochOk || !id || !epoch || !member.value("name").isString()) {
+            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source identity malformed group=%llx", (unsigned long long)group);
+            clearStreams(); return;
+        }
+        QString name = member.value("name").toString(); if (name.isEmpty()) name = QString::number(id);
+        sources.push_back({id, epoch, name});
+    }
+    m_streamSources = std::move(sources); m_streamBtn->setVisible(!m_streamSources.isEmpty());
+    m_streamBtn->setText(m_streamSources.size() > 1 ? QString::number(m_streamSources.size()) : QString());
+    m_streamBtn->setToolTip(QCoreApplication::translate("ChatGui", "Live streams: %1").arg(m_streamSources.size()));
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream header group=%llx sources=%d", (unsigned long long)group, (int)m_streamSources.size());
+}
+
 void MessageList::onRadioToggled(bool on) {
     if (m_currentChannelId.isEmpty() || isDmChannel(m_currentChannelId)) return;
 

tools/chatgui/src/messagelist.h

@@ -46,11 +46,15 @@ public:
     void onFeedbackResult(const uint8_t* data, int len);
     void onMessageSubmitted(const uint8_t* data, int len);
     void onRadioTalk(quint64 groupId, quint64 srcNodeId, bool on);   /* индикация говорящего */
+    void onStreamSources(const uint8_t* data, int len);
+    void clearStreams();
+    void watchStreams();
 
 signals:
     void readPositionChanged(const QString& channelId, qint64 lastReadMsgId);
     void callRequested(const QString& target, quint64 peer, const QString& name);
     void openVideoRequested(const QString& filePath);
+    void streamRequested(quint64 group, quint64 source, quint64 epoch, const QString& name);
     void openImageRequested(const QString& filePath);
     void inviteRequested(const QString& sourceChannel, quint64 peer, const QString& name);
     void ownFilesRequested(const QString& channelId);
@@ -83,6 +87,9 @@ private:
     quint64 m_callPeer = 0;
     quint64 m_callGroup = 0;
     QPushButton*   m_radioBtn = nullptr;       /* шапка: переключатель рации */
+    QPushButton* m_streamBtn = nullptr;
+    struct LiveSource { quint64 id, epoch; QString name; };
+    QList<LiveSource> m_streamSources;
     RadioPanel*    m_radioPanel = nullptr;    /* верхняя панель устройств, VAD и PTT */
     bool           m_radioOn = false;
     QThreadPool m_thumbnailPool;

tools/chatgui/src/streamplayer_window.cpp

@@ -0,0 +1,305 @@
+#include "streamplayer_window.h"
+#include "videoplayer_window.h"
+#include "sound_manager.h"
+#include "../transport/stream_bridge.h"
+#include <QVBoxLayout>
+#include <QHBoxLayout>
+#include <QLabel>
+#include <QPushButton>
+#include <QSlider>
+#include <QCloseEvent>
+#include <QKeyEvent>
+#include <QTransform>
+#include <atomic>
+#include <thread>
+#include <mutex>
+#include <deque>
+#include <array>
+#include <chrono>
+#include <algorithm>
+#include <cstring>
+
+extern "C" {
+#include "../../../lib/debug_config.h"
+#include "../../../lib/opus_codec.h"
+#include "../../../src/stream/stream.h"
+#ifdef HAVE_FFMPEG
+#include <libavcodec/avcodec.h>
+#include <libswscale/swscale.h>
+#endif
+}
+
+namespace {
+int64_t nowUs() {
+    return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
+}
+std::atomic<uint64_t> nextId{0};
+QString text(const char* value) { return QCoreApplication::translate("ChatGui", value); }
+}
+
+struct StreamPlayerWindow::Playback {
+    struct Video { QImage image; int64_t pts; };
+    struct Audio { std::array<int16_t, 960> pcm; int64_t pts; unsigned offset = 0; };
+    uint64_t id = ++nextId, group = 0, source = 0, epoch = 0;
+    std::atomic<bool> running{true}, audioAvailable{false};
+    std::atomic<float> volume{1.0f};
+    std::thread worker;
+    std::mutex lock;
+    std::deque<Video> videos;
+    std::deque<Audio> audios;
+    int64_t prefillUntil = 0, videoBase = -1, videoAt = 0, audioPts = -1, audioAt = 0;
+    uint64_t shown = 0, underruns = 0;
+    bool ended = false;
+
+    int64_t clock(int64_t now) const {
+        if (audioAvailable && audioPts >= 0 && now - audioAt < 200000) return audioPts + now - audioAt;
+        return videoBase >= 0 ? videoBase + now - videoAt : 0;
+    }
+};
+
+StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent)
+    : QWidget(parent), m_playback(std::make_unique<Playback>()) {
+    m_playback->group = group; m_playback->source = source; m_playback->epoch = epoch;
+    setWindowTitle(text("Live stream: %1").arg(name)); resize(800, 500);
+    auto* layout = new QVBoxLayout(this);
+    layout->addWidget(new QLabel(name, this));
+    m_status = new QLabel(text("Connecting…"), this); layout->addWidget(m_status);
+    m_render = new VideoRenderWidget(this); layout->addWidget(m_render, 1);
+    auto* controls = new QHBoxLayout;
+    auto* volume = new QSlider(Qt::Horizontal, this); volume->setRange(0, 100); volume->setValue(100);
+    volume->setAccessibleName(text("Stream volume")); controls->addWidget(new QLabel(text("Volume"), this)); controls->addWidget(volume);
+    auto* fullscreen = new QPushButton(text("Fullscreen"), this); controls->addWidget(fullscreen);
+    auto* close = new QPushButton(text("Close player"), this); controls->addWidget(close); layout->addLayout(controls);
+    connect(volume, &QSlider::valueChanged, this, [this](int value) { m_playback->volume = value / 100.0f; });
+    connect(fullscreen, &QPushButton::clicked, this, [this] { if (isFullScreen()) showNormal(); else showFullScreen(); });
+    connect(close, &QPushButton::clicked, this, &QWidget::close);
+    connect(this, &StreamPlayerWindow::statusChanged, m_status, &QLabel::setText, Qt::QueuedConnection);
+    connect(SoundManager::instance(), &SoundManager::contextAboutToReset, this, [this] {
+        m_audio.close(); m_playback->audioAvailable = false;
+        std::lock_guard<std::mutex> guard(m_playback->lock);
+        m_playback->audios.clear(); m_playback->audioPts = -1;
+        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream audio context reset id=%llu", (unsigned long long)m_playback->id);
+    });
+    connect(SoundManager::instance(), &SoundManager::contextRestored, this, &StreamPlayerWindow::openAudio);
+    connect(SoundManager::instance(), &SoundManager::devicesChanged, this, [this] { m_audio.close(); openAudio(); });
+    openAudio();
+    connect(&m_timer, &QTimer::timeout, this, &StreamPlayerWindow::tick); m_timer.start(15);
+    m_playback->worker = std::thread(&StreamPlayerWindow::decode, this);
+}
+
+StreamPlayerWindow::~StreamPlayerWindow() { stop(); }
+
+/* Output принадлежит GUI и закрывается до shutdown общего miniaudio-контекста. */
+void StreamPlayerWindow::openAudio() {
+    if (!m_playback->running || m_audio.initialized()) return;
+    ma_context* context = SoundManager::instance()->context();
+    if (!context) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream audio unavailable: no context"); return; }
+    auto config = ma_device_config_init(ma_device_type_playback);
+    config.sampleRate = 48000; config.playback.channels = 1; config.playback.format = ma_format_s16;
+    ma_device_id device;
+    if (SoundManager::deviceId(SoundManager::instance()->playbackDevice(), &device)) config.playback.pDeviceID = &device;
+    bool initialized = m_audio.init(context, config, "stream", m_playback->id, [this](const void*, void* output, unsigned frames) {
+        auto* pcm = static_cast<int16_t*>(output); std::fill(pcm, pcm + frames, 0);
+        auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock);
+        int64_t now = nowUs();
+        if (!p.prefillUntil || now < p.prefillUntil) return;
+        unsigned written = 0; float volume = p.volume.load(); int64_t firstPts = -1;
+        while (written < frames && !p.audios.empty()) {
+            auto& packet = p.audios.front();
+            if (firstPts < 0) firstPts = packet.pts + packet.offset * 1000000LL / 48000 - written * 1000000LL / 48000;
+            unsigned count = std::min(frames - written, 960 - packet.offset);
+            for (unsigned i = 0; i < count; i++) pcm[written + i] = static_cast<int16_t>(packet.pcm[packet.offset + i] * volume);
+            written += count; packet.offset += count;
+            if (packet.offset == 960) p.audios.pop_front();
+        }
+        if (firstPts >= 0) { p.audioPts = firstPts; p.audioAt = m_audio.callbackTimeUs(); }
+        else if (p.audioPts >= 0) {
+            p.audioPts += m_audio.callbackTimeUs() - p.audioAt; p.audioAt = m_audio.callbackTimeUs();
+        }
+        if (written < frames && p.audioPts >= 0 && !p.ended) p.underruns++;
+    });
+    if (!initialized || !m_audio.start()) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio device failed id=%llu", (unsigned long long)m_playback->id);
+        m_audio.close(); m_playback->audioAvailable = false; return;
+    }
+    m_playback->audioAvailable = true;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream audio output ready id=%llu", (unsigned long long)m_playback->id);
+}
+
+/* GUI забирает последний готовый кадр по аппаратным часам, не накапливая Qt-сигналы с QImage. */
+void StreamPlayerWindow::tick() {
+    auto& p = *m_playback;
+    if (m_audio.needsRecovery()) { m_audio.close(); p.audioAvailable = false; openAudio(); }
+    QImage image; bool finished = false;
+    {
+        std::lock_guard<std::mutex> guard(p.lock);
+        int64_t now = nowUs(), clock = p.clock(now);
+        if (now >= p.prefillUntil) while (!p.videos.empty() && p.videos.front().pts <= clock) {
+            image = std::move(p.videos.front().image); p.videos.pop_front();
+        }
+        finished = p.ended && p.videos.empty() && (p.audios.empty() || !p.audioAvailable);
+    }
+    if (!image.isNull()) {
+        m_render->setFrame(image);
+        m_status->setText(text("Live"));
+        if (!p.shown++) {
+            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop first frame id=%llu %dx%d", (unsigned long long)p.id, image.width(), image.height());
+        }
+        emit framePresented();
+    }
+    if (finished) { m_status->setText(text("Stream ended")); m_audio.close(); m_timer.stop(); }
+}
+
+/* Закрытие может происходить до выполнения open actor; session_id отменяет именно эту попытку. */
+void StreamPlayerWindow::stop() {
+    if (!m_playback->running.exchange(false) && !m_playback->worker.joinable()) return;
+    m_timer.stop(); gui_stream_stop(m_playback->id);
+    if (m_playback->worker.joinable()) m_playback->worker.join();
+    m_audio.close(); m_playback->audioAvailable = false;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream player stopped id=%llu shown=%llu underruns=%llu",
+               (unsigned long long)m_playback->id, (unsigned long long)m_playback->shown, (unsigned long long)m_playback->underruns);
+}
+
+void StreamPlayerWindow::closeEvent(QCloseEvent* event) { stop(); QWidget::closeEvent(event); }
+void StreamPlayerWindow::keyPressEvent(QKeyEvent* event) {
+    if (event->key() == Qt::Key_Escape) { if (isFullScreen()) showNormal(); else close(); }
+    else QWidget::keyPressEvent(event);
+}
+
+/* Worker владеет codec contexts. GAP/pressure сбрасывает refs и часы до следующего ключевого кадра. */
+void StreamPlayerWindow::decode() {
+    auto& p = *m_playback;
+#ifndef HAVE_FFMPEG
+    DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop live decoder unavailable: FFmpeg disabled");
+    emit statusChanged(text("Live video playback requires FFmpeg"));
+#else
+    AVCodecContext* decoder = nullptr; AVFrame* frame = av_frame_alloc(); AVPacket* packet = av_packet_alloc();
+    SwsContext* converter = nullptr;
+    opus_codec_decoder_t* opus = nullptr;
+    bool waitKey = true, opened = false;
+    int64_t deadline = nowUs() + 10000000, reportAt = nowUs(), expectedAudio = -1;
+    uint64_t decoded = 0, discarded = 0;
+    auto reset = [&] {
+        avcodec_free_context(&decoder);
+        opus_codec_decoder_destroy(opus); opus = nullptr; expectedAudio = -1; waitKey = true;
+        std::lock_guard<std::mutex> guard(p.lock);
+        p.videos.clear(); p.audios.clear(); p.prefillUntil = 0; p.videoBase = p.audioPts = -1;
+        emit statusChanged(text("Waiting for a key frame…"));
+    };
+    auto fail = [&](const QString& reason) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer failed id=%llu: %s", (unsigned long long)p.id, qPrintable(reason));
+        emit statusChanged(reason);
+    };
+    if (!frame || !packet) fail(text("Stream decoder allocation failed"));
+    else if (!gui_stream_open(p.id, p.group, p.source, p.epoch)) fail(text("Could not connect to the stream"));
+    else while (p.running) {
+        StreamRecord record;
+        if (!gui_stream_pull(p.id, record)) {
+            if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; }
+            if ((!opened || waitKey) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; }
+            std::this_thread::sleep_for(std::chrono::milliseconds(5)); continue;
+        }
+        if (record.kind == StreamRecordKind::Opened) { opened = true; continue; }
+        if (record.kind == StreamRecordKind::Gap) { reset(); deadline = nowUs() + 10000000; continue; }
+        if (record.kind == StreamRecordKind::Closed) {
+            const char* reason = record.reason == STREAM_CLOSE_SOURCE_CHANGED ? "The stream restarted. Select a source again."
+                : record.reason == STREAM_CLOSE_MEMBER ? "The source is no longer a channel member"
+                : record.reason == STREAM_CLOSE_SERVICE ? "Chat service is stopped"
+                : record.reason == STREAM_CLOSE_GROUP ? "The channel is closed" : "Could not continue the stream";
+            fail(text(reason)); break;
+        }
+        if (record.kind == StreamRecordKind::End) { std::lock_guard<std::mutex> guard(p.lock); p.ended = true; break; }
+        if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; }
+        if (record.kind == StreamRecordKind::Audio) {
+            if (!p.audioAvailable) continue;
+            if (!opus) opus = opus_codec_decoder_create(48000, 1);
+            if (!opus) { fail(text("Stream audio decoder failed")); break; }
+            if (expectedAudio >= 0 && (static_cast<int64_t>(record.pts) < expectedAudio - 10000
+                || static_cast<int64_t>(record.pts) > expectedAudio + 100000)) {
+                DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop audio discontinuity id=%llu expected=%lld pts=%llu",
+                           (unsigned long long)p.id, (long long)expectedAudio, (unsigned long long)record.pts);
+                reset(); deadline = nowUs() + 10000000; continue;
+            }
+            int missing = expectedAudio < 0 ? 0 : static_cast<int>((static_cast<int64_t>(record.pts) - expectedAudio + 10000) / 20000);
+            bool failed = false, pressure = false;
+            for (int i = 0; i <= missing; i++) {
+                Playback::Audio audio{}; audio.pts = i == missing ? static_cast<int64_t>(record.pts) : expectedAudio + i * 20000;
+                int count = opus_codec_decode(opus, i == missing ? reinterpret_cast<const uint8_t*>(record.data.constData()) : nullptr,
+                                             i == missing ? record.data.size() : 0, audio.pcm.data(), 960);
+                if (count != 960) { fail(text("Stream audio decoder failed")); failed = true; break; }
+                std::lock_guard<std::mutex> guard(p.lock);
+                if (!p.prefillUntil) p.prefillUntil = nowUs() + 150000;
+                if (p.audios.size() >= 25) { pressure = true; break; }
+                p.audios.push_back(std::move(audio));
+            }
+            if (failed) break;
+            if (pressure) {
+                DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop decoded audio pressure id=%llu; wait KEY", (unsigned long long)p.id);
+                reset(); deadline = nowUs() + 10000000;
+            } else expectedAudio = static_cast<int64_t>(record.pts) + 20000;
+            continue;
+        }
+        if (waitKey && !(record.flags & STREAM_MEDIA_KEY)) continue;
+        if (!decoder) {
+            const AVCodec* codec = avcodec_find_decoder(AV_CODEC_ID_H264);
+            decoder = codec ? avcodec_alloc_context3(codec) : nullptr;
+            if (decoder) { decoder->thread_count = 1; decoder->flags |= AV_CODEC_FLAG_LOW_DELAY; }
+            if (!decoder || avcodec_open2(decoder, codec, nullptr) < 0) { fail(text("Stream video decoder failed")); break; }
+            decoder->pkt_timebase = AVRational{1, 1000000};
+            waitKey = false;
+            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop H264 decoder started id=%llu", (unsigned long long)p.id);
+        }
+        av_packet_unref(packet);
+        if (av_new_packet(packet, record.data.size()) < 0) { fail(text("Stream decoder allocation failed")); break; }
+        std::memcpy(packet->data, record.data.constData(), static_cast<size_t>(record.data.size()));
+        packet->pts = packet->dts = static_cast<int64_t>(record.pts);
+        int sent = avcodec_send_packet(decoder, packet);
+        if (sent < 0) {
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 packet rejected result=%d id=%llu", sent, (unsigned long long)p.id);
+            reset(); deadline = nowUs() + 10000000; continue;
+        }
+        int received;
+        while ((received = avcodec_receive_frame(decoder, frame)) == 0) {
+            decoded++;
+            if (frame->width <= 0 || frame->height <= 0 || frame->width > 4096 || frame->height > 4096) {
+                fail(text("Invalid stream video dimensions")); p.running = false; break;
+            }
+            int64_t pts = frame->best_effort_timestamp == AV_NOPTS_VALUE ? static_cast<int64_t>(record.pts) : frame->best_effort_timestamp;
+            {
+                std::lock_guard<std::mutex> guard(p.lock);
+                if (!p.prefillUntil) p.prefillUntil = nowUs() + 150000;
+                if (p.videoBase < 0) { p.videoBase = pts; p.videoAt = p.prefillUntil; }
+                if (pts < p.clock(nowUs()) - 100000) { discarded++; continue; }
+            }
+            converter = sws_getCachedContext(converter, frame->width, frame->height, static_cast<AVPixelFormat>(frame->format),
+                frame->width, frame->height, AV_PIX_FMT_RGB24, SWS_BILINEAR, nullptr, nullptr, nullptr);
+            QImage image(frame->width, frame->height, QImage::Format_RGB888);
+            if (!converter || image.isNull()) { fail(text("Stream decoder allocation failed")); p.running = false; break; }
+            uint8_t* destination[] = {image.bits(), nullptr, nullptr, nullptr};
+            int strides[] = {static_cast<int>(image.bytesPerLine()), 0, 0, 0};
+            if (sws_scale(converter, frame->data, frame->linesize, 0, frame->height, destination, strides) != frame->height) {
+                fail(text("Stream video conversion failed")); p.running = false; break;
+            }
+            if (record.format.rotation) image = image.transformed(QTransform().rotate(record.format.rotation));
+            std::lock_guard<std::mutex> guard(p.lock);
+            if (p.videos.size() == 6) { p.videos.pop_front(); discarded++; }
+            p.videos.push_back({std::move(image), pts});
+        }
+        if (received != AVERROR(EAGAIN) && received != AVERROR_EOF && p.running) {
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 decode rejected result=%d id=%llu", received, (unsigned long long)p.id);
+            reset(); deadline = nowUs() + 10000000;
+        }
+        if (nowUs() - reportAt >= 1000000) {
+            reportAt = nowUs(); std::lock_guard<std::mutex> guard(p.lock);
+            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "desktop viewer id=%llu decoded=%llu discarded=%llu videoQ=%zu audioQ=%zu underruns=%llu",
+                       (unsigned long long)p.id, (unsigned long long)decoded, (unsigned long long)discarded,
+                       p.videos.size(), p.audios.size(), (unsigned long long)p.underruns);
+        }
+    }
+    avcodec_free_context(&decoder); av_frame_free(&frame); av_packet_free(&packet);
+    if (converter) sws_freeContext(converter);
+    opus_codec_decoder_destroy(opus);
+#endif
+    gui_stream_stop(p.id);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop decoder released id=%llu", (unsigned long long)p.id);
+}

tools/chatgui/src/streamplayer_window.h

@@ -0,0 +1,39 @@
+/* Просмотр одной live-трансляции: FFmpeg/Opus в worker, ограниченные очереди PCM/кадров.
+ * GUI владеет output device и окном; audio callback публикует часы физического воспроизведения.
+ * Закрытие останавливает worker и снимает reader. Публикация и общий транспорт не закрываются. */
+#pragma once
+
+#include <QWidget>
+#include <QTimer>
+#include <memory>
+#include <cstdint>
+#include "audio_device.h"
+
+class VideoRenderWidget;
+class QLabel;
+class QCloseEvent;
+class QKeyEvent;
+
+class StreamPlayerWindow : public QWidget {
+    Q_OBJECT
+public:
+    explicit StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent = nullptr);
+    ~StreamPlayerWindow() override;
+    void stop();
+signals:
+    void statusChanged(const QString& status);
+    void framePresented();
+protected:
+    void closeEvent(QCloseEvent* event) override;
+    void keyPressEvent(QKeyEvent* event) override;
+private:
+    struct Playback;
+    std::unique_ptr<Playback> m_playback;
+    AudioDevice m_audio;
+    QTimer m_timer;
+    VideoRenderWidget* m_render = nullptr;
+    QLabel* m_status = nullptr;
+    void openAudio();
+    void tick();
+    void decode();
+};

tools/chatgui/tests/test_stream_view.cpp

@@ -0,0 +1,215 @@
+/* Настоящее ядро/actor, подписанный медиаблок и H.264 decoder: каталог до подписки, эпоха, показ и teardown. */
+#include "../src/streamplayer_window.h"
+#include "../src/sound_manager.h"
+#include "../transport/stream_bridge.h"
+#include <QtTest>
+#include <QJsonDocument>
+#include <QJsonArray>
+#include <QLabel>
+#include <QSlider>
+#include <atomic>
+#include <cmath>
+
+extern "C" {
+#include "utun_instance.h"
+#include "chat/chat_core.h"
+#include "chat/chat_event.h"
+#include "stream/stream.h"
+#include "stream/stream_member.h"
+#include "debug_config.h"
+#include "mem.h"
+#include "sqlite3.h"
+#include "opus_codec.h"
+#ifdef HAVE_FFMPEG
+#include <libavcodec/avcodec.h>
+#endif
+}
+
+static QList<QByteArray> snapshots;
+#ifdef TEST_NULL_AUDIO
+static std::atomic<ma_device_data_proc> originalAudioCallback{nullptr};
+static std::atomic<uint64_t> nonzeroSamples{0};
+extern "C" ma_result __real_ma_context_init(const ma_backend*, ma_uint32, const ma_context_config*, ma_context*);
+extern "C" ma_result __real_ma_device_init(ma_context*, const ma_device_config*, ma_device*);
+extern "C" ma_result __wrap_ma_context_init(const ma_backend*, ma_uint32, const ma_context_config*, ma_context* context) {
+    ma_backend backend = ma_backend_null; return __real_ma_context_init(&backend, 1, nullptr, context);
+}
+static void observeAudio(ma_device* device, void* out, const void* in, ma_uint32 frames) {
+    originalAudioCallback.load()(device, out, in, frames);
+    const auto* pcm = static_cast<const int16_t*>(out);
+    for (unsigned i = 0; i < frames; i++) if (pcm[i]) nonzeroSamples.fetch_add(1);
+}
+extern "C" ma_result __wrap_ma_device_init(ma_context* context, const ma_device_config* config, ma_device* device) {
+    ma_device_config copy = *config;
+    if (copy.deviceType == ma_device_type_playback && copy.playback.format == ma_format_s16 && copy.playback.channels == 1) {
+        originalAudioCallback = copy.dataCallback; copy.dataCallback = observeAudio;
+    }
+    return __real_ma_device_init(context, &copy, device);
+}
+#endif
+static void sourceEvent(UTUN_INSTANCE*, int type, const uint8_t* data, int size) {
+    if (type == CHAT_EVT_STREAM_SOURCES) snapshots.push_back(QByteArray(reinterpret_cast<const char*>(data), size));
+}
+
+class StreamViewTest : public QObject {
+    Q_OBJECT
+    UASYNC* ua = nullptr;
+    UTUN_INSTANCE* inst = nullptr;
+    STREAM_HANDLE* publisher = nullptr;
+    uint64_t group = 0, epoch = 0, frameId = 0;
+    QTimer poll;
+    size_t allocations = 0;
+
+    bool receive(uint64_t id, StreamRecord& record) {
+        QElapsedTimer elapsed; elapsed.start();
+        while (elapsed.elapsed() < 1000) {
+            if (gui_stream_pull(id, record)) return true;
+            QTest::qWait(1);
+        }
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop test missing record id=%llu", (unsigned long long)id); return false;
+    }
+
+    void publish(const QByteArray& video, const QByteArray& audio = {}) {
+        stream_media_block media{}; media.format = {160, 90, 90, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS};
+        media.first_frame_offset = 0; media.video_count = 1;
+        media.video[0] = {++frameId, frameId * 100000, static_cast<uint32_t>(video.size()), 0,
+            static_cast<uint32_t>(video.size()), STREAM_MEDIA_KEY, reinterpret_cast<const uint8_t*>(video.constData())};
+        if (!audio.isEmpty()) {
+            media.audio_count = 1;
+            media.audio[0] = {frameId * 100000, static_cast<uint32_t>(audio.size()), 960,
+                             reinterpret_cast<const uint8_t*>(audio.constData())};
+        }
+        uint8_t wire[7080]; size_t length = 0;
+        QCOMPARE(stream_media_encode(&media, wire, sizeof(wire), &length), STREAM_OK);
+        QCOMPARE(stream_publish_block(publisher, wire, length), STREAM_OK);
+    }
+
+private slots:
+    void initTestCase() {
+        debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
+        debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR);
+        allocations = u_get_allocated_count(); ua = uasync_create(); QVERIFY(ua);
+        inst = utun_instance_create_from_str(ua,
+            "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n");
+        QVERIFY(inst); QCOMPARE(utun_core_start(inst), 0); QCOMPARE(chat_service_start(inst), 0);
+        chat_core_create_channel_auto(inst, "Stream test");
+        sqlite3_stmt* statement = nullptr;
+        QCOMPARE(sqlite3_prepare_v2(chat_core_get_db(inst), "SELECT channel_id FROM channels LIMIT 1", -1, &statement, nullptr), SQLITE_OK);
+        QCOMPARE(sqlite3_step(statement), SQLITE_ROW);
+        group = QString::fromUtf8(reinterpret_cast<const char*>(sqlite3_column_text(statement, 0))).toULongLong();
+        sqlite3_finalize(statement); QVERIFY(group);
+        chat_event_set_handler(inst, sourceEvent); QVERIFY(gui_stream_watch(group));
+        QVERIFY(gui_stream_attach(inst)); uasync_mark_running(ua);
+        connect(&poll, &QTimer::timeout, this, [this] { uasync_poll(ua, 0); }); poll.start(1);
+    }
+
+    void catalogueWithoutViewer() {
+        QTRY_COMPARE(snapshots.size(), 1);
+        QCOMPARE(snapshots.last().mid(9), QByteArray("[]")); QVERIFY(!stream_service_needs_activity(inst));
+        QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK);
+        stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); epoch = info.descriptor.source_epoch;
+        QTRY_COMPARE(snapshots.size(), 2);
+        auto sources = QJsonDocument::fromJson(snapshots.last().mid(9)).array(); QCOMPARE(sources.size(), 1);
+        QCOMPARE(stream_read_be(reinterpret_cast<const uint8_t*>(snapshots.last().constData()), 8), group);
+        QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);
+    }
+
+    void staleSelectionAndCancellation() {
+        QVERIFY(gui_stream_open(100, group, inst->node_id, epoch ^ 1));
+        StreamRecord event;
+        QVERIFY(receive(100, event)); QCOMPARE(event.kind, StreamRecordKind::Closed);
+        QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED);
+        QVERIFY(gui_stream_open(101, group, inst->node_id, epoch)); gui_stream_stop(101); QTest::qWait(20);
+        stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);
+    }
+
+    void mediaAndPressure() {
+        QVERIFY(gui_stream_open(102, group, inst->node_id, epoch)); StreamRecord event;
+        QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Opened);
+        const QByteArray frame(1500, 'k'); const QByteArray audio("opus"); publish(frame, audio);
+        QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video); QCOMPARE(event.data, frame);
+        QCOMPARE(event.format.rotation, uint16_t(90));
+        QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Audio); QCOMPARE(event.data, audio);
+        for (int i = 0; i < 270; i++) { publish(frame); uasync_poll(ua, 0); }
+        bool gap = false; unsigned records = 0;
+        while (gui_stream_pull(102, event)) { gap |= event.kind == StreamRecordKind::Gap; records++; }
+        QVERIFY(gap); QVERIFY(records <= 256);
+        publish(frame); QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video);
+        gui_stream_stop(102); QTest::qWait(20);
+        stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);
+    }
+
+    void realH264RenderingAndClose() {
+#ifdef HAVE_FFMPEG
+        const AVCodec* encoder = avcodec_find_encoder(AV_CODEC_ID_H264);
+        if (!encoder) QSKIP("H264 encoder is unavailable for the decoder fixture");
+        AVCodecContext* context = avcodec_alloc_context3(encoder); QVERIFY(context);
+        context->width = 160; context->height = 90; context->pix_fmt = AV_PIX_FMT_YUV420P;
+        context->time_base = {1, 10}; context->framerate = {10, 1}; context->max_b_frames = 0;
+        AVDictionary* options = nullptr; av_dict_set(&options, "preset", "ultrafast", 0); av_dict_set(&options, "tune", "zerolatency", 0);
+        QCOMPARE(avcodec_open2(context, encoder, &options), 0); av_dict_free(&options);
+        AVFrame* frame = av_frame_alloc(); QVERIFY(frame); frame->width = 160; frame->height = 90; frame->format = context->pix_fmt;
+        QCOMPARE(av_frame_get_buffer(frame, 32), 0);
+        for (int plane = 0; plane < 3; plane++) for (int y = 0; y < (plane ? 45 : 90); y++)
+            memset(frame->data[plane] + y * frame->linesize[plane], plane ? 128 : 80, plane ? 80 : 160);
+        frame->pts = 0; QCOMPARE(avcodec_send_frame(context, frame), 0);
+        AVPacket* packet = av_packet_alloc(); QVERIFY(packet); QCOMPARE(avcodec_receive_packet(context, packet), 0);
+        QByteArray video(reinterpret_cast<const char*>(packet->data), packet->size);
+        av_packet_free(&packet); av_frame_free(&frame); avcodec_free_context(&context);
+        QByteArray audio;
+#ifdef TEST_NULL_AUDIO
+        QVERIFY(SoundManager::instance()->init());
+        auto* opus = opus_codec_encoder_create(48000, 1); QVERIFY(opus);
+        int16_t pcm[960]; for (unsigned i = 0; i < 960; i++) pcm[i] = static_cast<int16_t>(6000 * std::sin(i * 0.05));
+        uint8_t encoded[1275]; int length = opus_codec_encode(opus, pcm, 960, encoded, sizeof(encoded)); QVERIFY(length > 0);
+        audio = QByteArray(reinterpret_cast<const char*>(encoded), length); opus_codec_encoder_destroy(opus);
+        nonzeroSamples = 0;
+#endif
+        StreamPlayerWindow window(group, inst->node_id, epoch, "Source");
+        QSignalSpy shown(&window, &StreamPlayerWindow::framePresented); window.show();
+        QTimer sender; connect(&sender, &QTimer::timeout, this, [this, video, audio] { publish(video, audio); }); sender.start(100);
+        QTRY_VERIFY_WITH_TIMEOUT(shown.count() > 0, 5000);
+#ifdef TEST_NULL_AUDIO
+        QTRY_VERIFY_WITH_TIMEOUT(nonzeroSamples.load() > 0, 5000);
+        auto* volume = window.findChild<QSlider*>(); QVERIFY(volume); volume->setValue(0);
+        QTest::qWait(50); uint64_t played = nonzeroSamples.load(); QTest::qWait(100); QCOMPARE(nonzeroSamples.load(), played);
+#endif
+        sender.stop(); window.close(); QTest::qWait(20);
+        stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);
+#else
+        QSKIP("FFmpeg is disabled");
+#endif
+    }
+
+    void finishAndDetachPendingOwner() {
+        QCOMPARE(stream_publish_finish(publisher), STREAM_OK); QTRY_COMPARE(snapshots.last().mid(9), QByteArray("[]"));
+        stream_close(publisher); publisher = nullptr;
+        QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK);
+        stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QVERIFY(info.descriptor.source_epoch != epoch);
+        QVERIFY(gui_stream_open(103, group, inst->node_id, epoch)); StreamRecord event;
+        QVERIFY(receive(103, event)); QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED);
+        QVERIFY(gui_stream_open(104, group, inst->node_id, info.descriptor.source_epoch));
+        gui_stream_detach(inst); QVERIFY(!gui_stream_ready()); QTest::qWait(20);
+        QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);
+    }
+
+    void channelClosureNotifiesViewer() {
+        QVERIFY(gui_stream_attach(inst)); QVERIFY(gui_stream_watch(group));
+        stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK);
+        QVERIFY(gui_stream_open(105, group, inst->node_id, info.descriptor.source_epoch));
+        StreamRecord event; QVERIFY(receive(105, event)); QCOMPARE(event.kind, StreamRecordKind::Opened);
+        QByteArray channel = QByteArray::number(group); chat_core_delete_channel(inst, channel.constData());
+        QVERIFY(receive(105, event)); QCOMPARE(event.kind, StreamRecordKind::Closed); QCOMPARE(event.reason, STREAM_CLOSE_GROUP);
+        QTRY_COMPARE(snapshots.last().mid(9), QByteArray("[]"));
+        QVERIFY(!stream_handle_active(publisher)); QCOMPARE(stream_handle_reason(publisher), STREAM_CLOSE_GROUP);
+    }
+
+    void cleanupTestCase() {
+        poll.stop(); gui_stream_detach(inst); stream_close(publisher); uasync_mark_stopped(ua);
+        SoundManager::instance()->shutdown();
+        utun_instance_destroy(inst); uasync_destroy(ua, 0);
+        QCOMPARE(u_get_allocated_count(), allocations);
+    }
+};
+QTEST_MAIN(StreamViewTest)
+#include "test_stream_view.moc"

tools/chatgui/translations/chatgui_en.ts

@@ -3,6 +3,30 @@
 <TS version="2.1" language="en_US" sourcelanguage="en_US">
 <context>
     <name>ChatGui</name>
+    <message><source>Live streams</source><translation>Live streams</translation></message>
+    <message><source>Live streams: %1</source><translation>Live streams: %1</translation></message>
+    <message><source>Live stream: %1</source><translation>Live stream: %1</translation></message>
+    <message><source>Stream volume</source><translation>Stream volume</translation></message>
+    <message><source>Volume</source><translation>Volume</translation></message>
+    <message><source>Fullscreen</source><translation>Fullscreen</translation></message>
+    <message><source>Close player</source><translation>Close player</translation></message>
+    <message><source>Live</source><translation>Live</translation></message>
+    <message><source>Stream ended</source><translation>Stream ended</translation></message>
+    <message><source>Waiting for a key frame…</source><translation>Waiting for a key frame…</translation></message>
+    <message><source>Live video playback requires FFmpeg</source><translation>Live video playback requires FFmpeg</translation></message>
+    <message><source>Stream decoder allocation failed</source><translation>Stream decoder allocation failed</translation></message>
+    <message><source>Could not connect to the stream</source><translation>Could not connect to the stream</translation></message>
+    <message><source>Chat service is stopped</source><translation>Chat service is stopped</translation></message>
+    <message><source>Stream connection timed out</source><translation>Stream connection timed out</translation></message>
+    <message><source>The stream restarted. Select a source again.</source><translation>The stream restarted. Select a source again.</translation></message>
+    <message><source>The source is no longer a channel member</source><translation>The source is no longer a channel member</translation></message>
+    <message><source>The channel is closed</source><translation>The channel is closed</translation></message>
+    <message><source>Could not continue the stream</source><translation>Could not continue the stream</translation></message>
+    <message><source>Invalid stream timestamp</source><translation>Invalid stream timestamp</translation></message>
+    <message><source>Stream audio decoder failed</source><translation>Stream audio decoder failed</translation></message>
+    <message><source>Stream video decoder failed</source><translation>Stream video decoder failed</translation></message>
+    <message><source>Invalid stream video dimensions</source><translation>Invalid stream video dimensions</translation></message>
+    <message><source>Stream video conversion failed</source><translation>Stream video conversion failed</translation></message>
     <message><source>Options</source><translation>Options</translation></message>
     <message><source>Group files</source><translation>Group files</translation></message>
     <message><source>My files</source><translation>My files</translation></message>

tools/chatgui/translations/chatgui_ru.ts

@@ -3,6 +3,30 @@
 <TS version="2.1" language="ru_RU" sourcelanguage="en_US">
 <context>
     <name>ChatGui</name>
+    <message><source>Live streams</source><translation>Трансляции</translation></message>
+    <message><source>Live streams: %1</source><translation>Трансляции: %1</translation></message>
+    <message><source>Live stream: %1</source><translation>Трансляция: %1</translation></message>
+    <message><source>Stream volume</source><translation>Громкость трансляции</translation></message>
+    <message><source>Volume</source><translation>Громкость</translation></message>
+    <message><source>Fullscreen</source><translation>Полный экран</translation></message>
+    <message><source>Close player</source><translation>Закрыть просмотр</translation></message>
+    <message><source>Live</source><translation>Прямой эфир</translation></message>
+    <message><source>Stream ended</source><translation>Трансляция завершена</translation></message>
+    <message><source>Waiting for a key frame…</source><translation>Ожидание ключевого кадра…</translation></message>
+    <message><source>Live video playback requires FFmpeg</source><translation>Для просмотра трансляций необходим FFmpeg</translation></message>
+    <message><source>Stream decoder allocation failed</source><translation>Не удалось выделить память для просмотра трансляции</translation></message>
+    <message><source>Could not connect to the stream</source><translation>Не удалось подключиться к трансляции</translation></message>
+    <message><source>Chat service is stopped</source><translation>Сервис чата остановлен</translation></message>
+    <message><source>Stream connection timed out</source><translation>Истекло время ожидания трансляции</translation></message>
+    <message><source>The stream restarted. Select a source again.</source><translation>Трансляция перезапущена. Выберите источник заново.</translation></message>
+    <message><source>The source is no longer a channel member</source><translation>Источник больше не является участником канала</translation></message>
+    <message><source>The channel is closed</source><translation>Канал закрыт</translation></message>
+    <message><source>Could not continue the stream</source><translation>Не удалось продолжить просмотр трансляции</translation></message>
+    <message><source>Invalid stream timestamp</source><translation>Некорректное время кадра трансляции</translation></message>
+    <message><source>Stream audio decoder failed</source><translation>Ошибка декодирования звука трансляции</translation></message>
+    <message><source>Stream video decoder failed</source><translation>Ошибка декодирования видео трансляции</translation></message>
+    <message><source>Invalid stream video dimensions</source><translation>Некорректный размер кадра трансляции</translation></message>
+    <message><source>Stream video conversion failed</source><translation>Ошибка преобразования кадра трансляции</translation></message>
     <message><source>Options</source><translation>Действия</translation></message>
     <message><source>Group files</source><translation>Файлы группы</translation></message>
     <message><source>My files</source><translation>Мои файлы</translation></message>

tools/chatgui/transport/gui_bridge.h

@@ -253,6 +253,7 @@ void gui_bridge_set_feedback_changed_cb(gui_dm_messages_fn cb);
 void gui_bridge_set_feedback_result_cb(gui_dm_messages_fn cb);
 void gui_bridge_set_msg_submitted_cb(gui_dm_messages_fn cb);
 void gui_bridge_set_radio_mute_result_cb(gui_dm_messages_fn cb);
+void gui_bridge_set_stream_sources_cb(gui_dm_messages_fn cb);
 int gui_bridge_toggle_member_radio_mute(const char* channel_id, uint64_t node_id);
 /* CHAT_EVT_ATTACHMENT_PREPARATION: target + job UUID + state. */
 void gui_bridge_set_attachment_preparation_cb(gui_dm_messages_fn cb);

tools/chatgui/transport/gui_bridge_impl.cpp

@@ -77,6 +77,7 @@ static gui_dm_messages_fn g_feedback_changed_cb = nullptr;
 static gui_dm_messages_fn g_feedback_result_cb = nullptr;
 static gui_dm_messages_fn g_msg_submitted_cb = nullptr;
 static gui_dm_messages_fn g_radio_mute_result_cb = nullptr;
+static gui_dm_messages_fn g_stream_sources_cb = nullptr;
 static gui_network_changed_fn g_network_changed_cb = nullptr;
 static struct UASYNC*        g_ua                 = nullptr;
 static struct UTUN_INSTANCE* g_inst               = nullptr;
@@ -91,6 +92,9 @@ void GuiBridgeReceiver::processPost(int eventType, QByteArray data) {
         eventType == 16 || eventType == 17 || eventType == 24 || eventType == 25 || eventType == 31 || eventType == 33))
         g_network_changed_cb(eventType);
     switch (eventType) {
+    case CHAT_EVT_STREAM_SOURCES:
+        if (g_stream_sources_cb) g_stream_sources_cb(d, dlen);
+        break;
     case CHAT_EVT_CPU_LOAD:
         if (dlen != 2 || d[0] > 1 || d[1] > 100) {
             DEBUG_WARN(DEBUG_CATEGORY_SYS, "invalid CPU load response bytes=%d", dlen);
@@ -615,6 +619,7 @@ void gui_bridge_set_dm_msg_received_cb(gui_dm_msg_received_fn cb) {
 }
 
 void gui_bridge_set_network_changed_cb(gui_network_changed_fn cb) { g_network_changed_cb = cb; }
+void gui_bridge_set_stream_sources_cb(gui_dm_messages_fn cb) { g_stream_sources_cb = cb; }
 void gui_bridge_set_dm_call_route_cb(gui_dm_messages_fn cb) { g_dm_call_route_cb = cb; }
 void gui_bridge_set_dm_invite_result_cb(gui_dm_messages_fn cb) { g_dm_invite_result_cb = cb; }
 void gui_bridge_set_member_files_response_cb(gui_dm_messages_fn cb) { g_member_files_response_cb = cb; }

tools/chatgui/transport/stream_bridge.cpp

@@ -0,0 +1,210 @@
+#include "stream_bridge.h"
+#include <deque>
+#include <mutex>
+#include <algorithm>
+
+extern "C" {
+#include "../../../lib/debug_config.h"
+#include "../../../lib/mem.h"
+#include "../../../lib/u_async.h"
+#include "../../../src/utun_instance.h"
+#include "../../../src/stream/stream.h"
+#include "../../../src/stream/stream_member.h"
+}
+
+namespace {
+struct Viewer { uint64_t id = 0, group = 0, source = 0, epoch = 0; };
+struct Bridge {
+    std::mutex lock;
+    UTUN_INSTANCE* inst = nullptr;
+    posted_task* pending = nullptr;
+    uint64_t watch = 0;
+    bool watchPending = false, opening = false, stopping = false, overflow = false;
+    Viewer viewer, requested;
+    STREAM_HANDLE* handle = nullptr;
+    stream_media_receiver* receiver = nullptr;
+    std::deque<StreamRecord> output;
+    size_t bytes = 0;
+} bridge;
+
+void process(void*);
+
+/* Вызывается под lock; резервирование делает отказ команды наблюдаемым до изменения состояния. */
+bool schedule() {
+    if (!bridge.inst) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream command rejected: core unavailable"); return false;
+    }
+    if (bridge.pending) return true;
+    auto* task = static_cast<posted_task*>(u_calloc(1, sizeof(posted_task)));
+    if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream task allocation failed"); return false; }
+    task->callback = process; bridge.pending = task;
+    uasync_post_reserved(bridge.inst->ua, task);
+    return true;
+}
+
+/* Управляющие записи не теряются при media pressure; старые readers не накапливают очередь. */
+void output(StreamRecord record) {
+    if (bridge.output.size() >= 256 || bridge.bytes + static_cast<size_t>(record.data.size()) > 4u * 1024u * 1024u) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream RX pressure id=%llu records=%zu bytes=%zu; wait KEY",
+                   (unsigned long long)record.id, bridge.output.size(), bridge.bytes);
+        bridge.output.clear(); bridge.bytes = 0; bridge.overflow = true;
+        bridge.output.push_back({record.id, 0, StreamRecordKind::Gap});
+        if (record.kind == StreamRecordKind::Video || record.kind == StreamRecordKind::Audio) return;
+    }
+    bridge.bytes += static_cast<size_t>(record.data.size()); bridge.output.push_back(std::move(record));
+}
+
+/* Только uasync под lock; снимает собственное владение подпиской, не транспорт других владельцев. */
+void closeViewer(StreamRecordKind kind, int reason) {
+    if (bridge.viewer.id) output({bridge.viewer.id, 0, kind, {}, 0, reason});
+    if (bridge.handle) stream_close(bridge.handle);
+    stream_media_receiver_destroy(bridge.receiver);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer released id=%llu reason=%d",
+               (unsigned long long)bridge.viewer.id, reason);
+    bridge.handle = nullptr; bridge.receiver = nullptr; bridge.viewer = {}; bridge.overflow = false;
+}
+
+void mediaFrame(void*, const stream_media_format* format, const stream_media_video* video, const stream_media_audio* audio) {
+    if (bridge.overflow) return;
+    StreamRecord record; record.id = bridge.viewer.id; record.format = *format;
+    if (video) {
+        record.kind = StreamRecordKind::Video; record.pts = video->pts_us; record.flags = video->flags;
+        record.data = QByteArray(reinterpret_cast<const char*>(video->data), static_cast<int>(video->length));
+    } else {
+        record.kind = StreamRecordKind::Audio; record.pts = audio->pts_us;
+        record.data = QByteArray(reinterpret_cast<const char*>(audio->data), static_cast<int>(audio->length));
+    }
+    output(std::move(record));
+}
+
+/* Scratch receiver/stream не покидает callback; decoder получает собственные копии. */
+void mediaEvent(UTUN_INSTANCE* inst, const stream_event* event, void*) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    if (inst != bridge.inst || !bridge.handle || event->descriptor.group_id != bridge.viewer.group
+        || event->descriptor.source_node_id != bridge.viewer.source || event->descriptor.source_epoch != bridge.viewer.epoch) return;
+    if (event->type == STREAM_EVENT_DATA) {
+        int result = stream_media_receiver_feed(bridge.receiver, event->data, event->length);
+        if (result != STREAM_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer media rejected result=%d", result);
+            closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED);
+        } else if (bridge.overflow) {
+            stream_media_receiver_gap(bridge.receiver); bridge.overflow = false;
+        }
+    } else if (event->type == STREAM_EVENT_GAP) {
+        stream_media_receiver_gap(bridge.receiver); output({bridge.viewer.id, 0, StreamRecordKind::Gap});
+    } else if (event->type == STREAM_EVENT_END || !stream_handle_active(bridge.handle)) {
+        closeViewer(event->type == STREAM_EVENT_END ? StreamRecordKind::End : StreamRecordKind::Closed, event->reason);
+    }
+}
+
+/* Сериализованный actor: старый owner закрывается перед открытием нового выбранного источника. */
+void process(void*) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    bridge.pending = nullptr;
+    if (!bridge.inst) return;
+    if (bridge.watchPending) {
+        bridge.watchPending = false;
+        if (stream_watch_sources(bridge.inst, bridge.watch) != STREAM_OK)
+            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream source watch failed group=%llx", (unsigned long long)bridge.watch);
+    }
+    if (bridge.stopping || bridge.opening) {
+        if (bridge.viewer.id) closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_STOPPED);
+        bridge.stopping = false;
+    }
+    if (!bridge.opening) return;
+    bridge.opening = false; bridge.viewer = bridge.requested; bridge.requested = {};
+    stream_member_state member{};
+    int reason = STREAM_CLOSE_FAILED;
+    int result = stream_member_load(bridge.inst, bridge.viewer.group, bridge.viewer.source, &member, nullptr);
+    if (result == STREAM_OK && (!member.published || member.source.source_epoch != bridge.viewer.epoch)) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop selected source changed source=%llx expected=%llx actual=%llx",
+                   (unsigned long long)bridge.viewer.source, (unsigned long long)bridge.viewer.epoch,
+                   (unsigned long long)(member.published ? member.source.source_epoch : 0));
+        result = STREAM_ERR_CONFLICT; reason = STREAM_CLOSE_SOURCE_CHANGED;
+    }
+    if (result == STREAM_OK) result = stream_subscribe_open(bridge.inst, bridge.viewer.group, bridge.viewer.source, &bridge.handle);
+    stream_info info{};
+    if (result == STREAM_OK) result = stream_get_info(bridge.handle, &info);
+    if (result == STREAM_OK && info.descriptor.source_epoch != bridge.viewer.epoch) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop reader encountered stale local epoch");
+        result = STREAM_ERR_CONFLICT; reason = STREAM_CLOSE_SOURCE_CHANGED;
+    }
+    if (result == STREAM_OK) result = stream_media_receiver_create(&bridge.receiver, mediaFrame, nullptr);
+    if (result != STREAM_OK) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer open failed id=%llu group=%llx source=%llx result=%d",
+                   (unsigned long long)bridge.viewer.id, (unsigned long long)bridge.viewer.group,
+                   (unsigned long long)bridge.viewer.source, result);
+        closeViewer(StreamRecordKind::Closed, reason); return;
+    }
+    output({bridge.viewer.id, 0, StreamRecordKind::Opened});
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer opened id=%llu group=%llx source=%llx epoch=%llx",
+               (unsigned long long)bridge.viewer.id, (unsigned long long)bridge.viewer.group,
+               (unsigned long long)bridge.viewer.source, (unsigned long long)bridge.viewer.epoch);
+}
+}
+
+bool gui_stream_attach(UTUN_INSTANCE* inst) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    if (!inst || !inst->stream || bridge.inst) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream attach rejected"); return false;
+    }
+    bridge.inst = inst; bridge.output.clear(); bridge.bytes = 0;
+    stream_set_event_callback(inst, mediaEvent, nullptr);
+    if (bridge.watchPending) {
+        bridge.watchPending = false;
+        if (stream_watch_sources(inst, bridge.watch) != STREAM_OK)
+            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop initial source watch failed group=%llx", (unsigned long long)bridge.watch);
+    }
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream bridge attached"); return true;
+}
+
+void gui_stream_detach(UTUN_INSTANCE* inst) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    if (!inst || bridge.inst != inst) return;
+    if (bridge.pending) uasync_cancel_post(inst->ua, bridge.pending);
+    stream_watch_sources(inst, 0);
+    bridge.pending = nullptr; stream_set_event_callback(inst, nullptr, nullptr);
+    if (bridge.viewer.id) closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_SERVICE);
+    if (bridge.opening) output({bridge.requested.id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_SERVICE});
+    bridge.inst = nullptr; bridge.requested = {}; bridge.opening = bridge.stopping = bridge.watchPending = false;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream bridge detached");
+}
+
+bool gui_stream_watch(uint64_t group) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    if (!bridge.inst) {
+        bridge.watch = group; bridge.watchPending = true;
+        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "desktop source watch queued until attach group=%llx", (unsigned long long)group); return true;
+    }
+    if (!schedule()) return false;
+    bridge.watch = group; bridge.watchPending = true; return true;
+}
+
+bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    if (!id || !group || !source || !epoch) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer invalid identity"); return false; }
+    if (!schedule()) return false;
+    if (bridge.opening) output({bridge.requested.id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_STOPPED});
+    bridge.requested = {id, group, source, epoch}; bridge.opening = true; return true;
+}
+
+void gui_stream_stop(uint64_t id) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    if (bridge.opening && bridge.requested.id == id) {
+        bridge.opening = false; bridge.requested = {}; output({id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_STOPPED});
+    }
+    if (bridge.viewer.id == id && schedule()) bridge.stopping = true;
+    for (auto i = bridge.output.begin(); i != bridge.output.end();) {
+        if (i->id == id) { bridge.bytes -= static_cast<size_t>(i->data.size()); i = bridge.output.erase(i); }
+        else ++i;
+    }
+}
+
+bool gui_stream_pull(uint64_t id, StreamRecord& record) {
+    std::lock_guard<std::mutex> guard(bridge.lock);
+    auto i = std::find_if(bridge.output.begin(), bridge.output.end(), [id](const StreamRecord& r) { return r.id == id; });
+    if (i == bridge.output.end()) return false;
+    bridge.bytes -= static_cast<size_t>(i->data.size()); record = std::move(*i); bridge.output.erase(i); return true;
+}
+
+bool gui_stream_ready() { std::lock_guard<std::mutex> guard(bridge.lock); return bridge.inst != nullptr; }

tools/chatgui/transport/stream_bridge.h

@@ -0,0 +1,27 @@
+/* Desktop live viewer: attach/detach в uasync до destroy ядра; остальные вызовы потокобезопасны.
+ * Один viewer, команды содержат IDs; RX ограничен 4MB/256 записей, borrowed данные копируются.
+ * Watch до attach запоминает выбранную группу. Каталог идёт через CHAT_EVT_STREAM_SOURCES,
+ * медиаданные — через отдельную очередь pull. */
+#pragma once
+
+#include <QByteArray>
+#include <cstdint>
+#include "../../../src/stream/stream_media.h"
+
+struct UTUN_INSTANCE;
+enum class StreamRecordKind { Video, Audio, Gap, Opened, End, Closed };
+struct StreamRecord {
+    uint64_t id = 0, pts = 0;
+    StreamRecordKind kind = StreamRecordKind::Gap;
+    stream_media_format format{};
+    int flags = 0, reason = 0;
+    QByteArray data;
+};
+
+bool gui_stream_attach(UTUN_INSTANCE* inst);
+void gui_stream_detach(UTUN_INSTANCE* inst);
+bool gui_stream_watch(uint64_t group);
+bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch);
+void gui_stream_stop(uint64_t id);
+bool gui_stream_pull(uint64_t id, StreamRecord& record);
+bool gui_stream_ready();

tools/chatgui/transport/utun_node.cpp

@@ -1,5 +1,6 @@
 // utun_node.cpp — embedded uTun node running in a dedicated thread
 #include "utun_node.h"
+#include "stream_bridge.h"
 #include <cinttypes>
 #include <QDir>
 #include "../../lib/socket_compat.h"
@@ -246,6 +247,10 @@ void UtunNode::runLoop() {
     }
     call_audio_init(m_instance);
     radio_audio_init(m_instance);
+    if (!gui_stream_attach(m_instance)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream bridge initialization failed");
+        QMetaObject::invokeMethod(this, [this] { emit error("stream bridge initialization failed"); });
+    }
     chat_core_sync_my_addresses(m_instance);
 
     DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "chat_core + chat_sync initialized");
@@ -262,6 +267,7 @@ void UtunNode::runLoop() {
 
     if (m_instance) {
         DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "runLoop: destroying instance in worker thread");
+        gui_stream_detach(m_instance);
         call_audio_destroy(m_instance);
         radio_audio_destroy(m_instance);
         utun_instance_destroy(m_instance);