Add device regression for independent stream audio and video workers

tools/chatgui-android/tests/stream_worker_smoke/README.md

@@ -0,0 +1,28 @@
+# Проверка независимости аудио и видео на устройстве
+
+Из корня репозитория после сборки `assembleDebug`:
+
+```bash
+python3 tools/chatgui-android/tests/stream_worker_smoke/run.py SERIAL > /tmp/stream-worker-smoke.log 2>&1
+```
+
+Нужны подключённое ADB-устройство, JDK, FFmpeg с libx264, Android SDK API 36 и build-tools.
+Путь SDK берётся из `--sdk`, `ANDROID_HOME`, `ANDROID_SDK_ROOT` или `local.properties` проекта.
+Другой собранный APK можно передать через `--apk`.
+
+Тест запускает классы собранного APK в отдельном `app_process`, без переустановки приложения.
+Он использует настоящий MediaCodec с ImageReader Surface, native Opus/SoundTouch, StreamMixedPlayback
+и общий AudioTrack. Аудиоматериал — цифровая тишина.
+
+Проверки:
+
+- Callback показа блокирует video worker на 500 мс. За это время аппаратные аудиочасы должны
+  пройти минимум 20 000 кадров, а `underruns` не должен увеличиться.
+- Все 120 видеокадров декодируются до EOF; поздние после остановки кадры разрешено отбросить.
+- Тот же video worker после reset декодирует короткую трансляцию из трёх кадров и снова завершает EOF.
+- После каждого сценария дожидаемся физического завершения аудиохвоста и освобождаем input.
+
+Успех отмечается строками `PASS`; нарушение инвариантов завершает процесс с ненулевым кодом.
+Подробные сообщения production workers доступны в `adb logcat -s utun-gui:D`.
+Тестовые APK, библиотеки и fixture остаются в `/data/local/tmp/utun-stream-worker-smoke` для диагностики.
+Сетевой путь и UI этот сценарий не проверяет.

tools/chatgui-android/tests/stream_worker_smoke/StreamWorkerSmoke.java

@@ -0,0 +1,132 @@
+import android.media.ImageReader;
+import android.graphics.ImageFormat;
+import android.os.Handler;
+import android.os.HandlerThread;
+import android.os.Looper;
+import android.view.Surface;
+import java.lang.reflect.*;
+import java.nio.file.*;
+import java.util.concurrent.atomic.*;
+
+/** Device smoke test: uses production classes from APK, without installing/stopping the user's app. */
+public class StreamWorkerSmoke {
+    static Object invoke(Object owner, String name, Object... args) throws Exception {
+        for (Method m : owner.getClass().getDeclaredMethods()) {
+            if (m.getName().equals(name) && m.getParameterCount() == args.length) {
+                m.setAccessible(true);
+                try { return m.invoke(owner, args); }
+                catch (InvocationTargetException e) { throw new RuntimeException(name, e.getCause()); }
+            }
+        }
+        throw new NoSuchMethodException(name);
+    }
+    static Class<?> type(String name) throws Exception { return Class.forName("com.utun.chat.data." + name); }
+    static Object callback(Runnable body) throws Exception {
+        Class<?> function = Class.forName("kotlin.jvm.functions.Function0");
+        Object unit = Class.forName("kotlin.Unit").getField("INSTANCE").get(null);
+        return Proxy.newProxyInstance(function.getClassLoader(), new Class<?>[]{function}, (p, m, a) -> {
+            if (m.getName().equals("invoke")) { body.run(); return unit; }
+            if (m.getName().equals("toString")) return "smoke callback";
+            return null;
+        });
+    }
+    static long underruns(String text) {
+        return Long.parseLong(text.split("underruns=")[1].split(" ")[0]);
+    }
+    static long played(String text) { return Long.parseLong(text.split("played=")[1].split(" ")[0]); }
+    public static void main(String[] args) throws Exception {
+        Thread.setDefaultUncaughtExceptionHandler((t, e) -> { e.printStackTrace(); System.exit(1); });
+        if (Looper.getMainLooper() == null) Looper.prepareMainLooper();
+        HandlerThread images = new HandlerThread("smoke-images"); images.start();
+        ImageReader reader = ImageReader.newInstance(128, 72, ImageFormat.PRIVATE, 4);
+        AtomicInteger received = new AtomicInteger();
+        reader.setOnImageAvailableListener(r -> {
+            try (android.media.Image image = r.acquireLatestImage()) { if (image != null) received.incrementAndGet(); }
+        }, new Handler(images.getLooper()));
+        Object noop = callback(() -> {});
+        Object nativeLib = type("NativeLib").getField("INSTANCE").get(null);
+        long opus = (Long)invoke(nativeLib, "streamOpusCreate", true);
+        byte[] packet = (byte[])invoke(nativeLib, "nativeStreamOpusEncode", opus, new short[1920], false, 25, 1);
+        invoke(nativeLib, "streamOpusClose", opus);
+        if (packet == null) throw new AssertionError("Opus encode failed");
+        byte[] frame = Files.readAllBytes(Paths.get(args[0]));
+        Object codec = type("StreamVideoCodec").getField("H264").get(null);
+        Class<?> function = Class.forName("kotlin.jvm.functions.Function0");
+        Constructor<?> audioCtor = type("StreamMixedPlayback").getConstructor(long.class, long.class, boolean.class, function);
+        Constructor<?> frameCtor = type("StreamVideoPlayback$Frame").getConstructor(long.class, boolean.class,
+            int.class, int.class, int.class, long.class, type("StreamVideoCodec"), byte[].class, byte[].class);
+        AtomicReference<Object> currentAudio = new AtomicReference<>();
+        AtomicInteger shown = new AtomicInteger();
+        AtomicBoolean stalled = new AtomicBoolean();
+        AtomicReference<Throwable> problem = new AtomicReference<>();
+        Object render = callback(() -> {
+            if (shown.incrementAndGet() == 15) {
+                try {
+                    Object audio = currentAudio.get();
+                    String before = (String)invoke(audio, "getDescription");
+                    System.out.println("STALL start " + before);
+                    Thread.sleep(500);
+                    String after = (String)invoke(audio, "getDescription");
+                    System.out.println("STALL end " + after);
+                    if (underruns(after) != underruns(before)) throw new AssertionError("Audio gap during video stall");
+                    if (played(after) - played(before) < 20000) throw new AssertionError("Audio clock stalled with video");
+                    stalled.set(true);
+                } catch (Throwable e) { problem.set(e); }
+            }
+        });
+        Object video = type("StreamVideoPlayback").getConstructor(long.class, Surface.class, long.class, function, function)
+            .newInstance(4242L, reader.getSurface(), System.nanoTime() / 1000, noop, render);
+        try {
+            for (int phase = 0; phase < 2; phase++) {
+                if (phase != 0) invoke(video, "reset");
+                int frames = phase == 0 ? 120 : 3;
+                long duration = frames * 33333L;
+                Object audio = audioCtor.newInstance(4242L + phase, 0L, true, noop);
+                currentAudio.set(audio);
+                long audioPts = 0, position = 0, deadline = System.nanoTime() + 15_000_000_000L;
+                int supplied = 0;
+                boolean videoFinished = false, audioFinished = false, done = false;
+                try {
+                    while (System.nanoTime() < deadline) {
+                        Object failure = invoke(video, "getFailure");
+                        if (failure != null) throw new AssertionError("Video failed", (Throwable)failure);
+                        if (problem.get() != null) throw new AssertionError("Stall test failed", problem.get());
+                        while (audioPts < duration && (Boolean)invoke(audio, "getNeedsInput")) {
+                            int result = (Integer)invoke(audio, "feed", audioPts, packet);
+                            if (result < 0) throw new AssertionError("Opus feed=" + result);
+                            audioPts += 20000;
+                        }
+                        if (audioPts >= duration && !audioFinished) { invoke(audio, "finish"); audioFinished = true; }
+                        invoke(audio, "prepare");
+                        while (supplied < frames && (Boolean)invoke(video, "getWantsInput")) {
+                            invoke(video, "offer", frameCtor.newInstance(supplied++ * 33333L, true, 128, 72, 0, 1L,
+                                codec, new byte[0], frame));
+                        }
+                        if (supplied == frames && !videoFinished) { invoke(video, "finish"); videoFinished = true; }
+                        Object state = invoke(video, "getState");
+                        long lead = (Long)invoke(audio, "getLeadUs");
+                        if (invoke(state, "getFirstOutputPts") != null) {
+                            long[] clock = (long[])invoke(audio, "update", System.nanoTime() / 1000, audioPts, Math.max(40000L, lead + 20000));
+                            if (clock == null) throw new AssertionError("Clock failed");
+                            position = clock[0];
+                        }
+                        invoke(video, "update", position, audioPts, lead);
+                        if ((Boolean)invoke(state, "getEnded") && (Boolean)invoke(audio, "getDrained")) {
+                            long decoded = (Long)invoke(state, "getDecoded");
+                            if (decoded != frames) throw new AssertionError("Lost frames " + state);
+                            System.out.println("PASS phase=" + phase + " " + state + " " + invoke(audio, "getDescription"));
+                            done = true; break;
+                        }
+                        Thread.sleep(5);
+                    }
+                    if (!done) throw new AssertionError("Drain timeout " + invoke(video, "getState") + " " + invoke(audio, "getDescription"));
+                } finally { invoke(audio, "close"); }
+            }
+            if (!stalled.get()) throw new AssertionError("Stall was not exercised");
+            System.out.println("PASS 500ms video stall: no new audio gaps; EOF/reset: 120 then 3 decoded; surfaces=" + received.get());
+        } finally {
+            invoke(video, "close"); reader.close(); images.quitSafely(); images.join();
+        }
+        System.exit(0);
+    }
+}

tools/chatgui-android/tests/stream_worker_smoke/run.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""Run production APK media workers on a device without replacing the installed app."""
+import argparse
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import tempfile
+import zipfile
+
+
+def main():
+    project = Path(__file__).resolve().parents[2]
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("serial", help="adb serial of the test device")
+    parser.add_argument("--sdk", type=Path, default=os.getenv("ANDROID_HOME") or os.getenv("ANDROID_SDK_ROOT"))
+    parser.add_argument("--apk", type=Path, default=project / "app/build/outputs/apk/debug/app-debug.apk")
+    args = parser.parse_args()
+    sdk = args.sdk
+    if sdk is None:
+        properties = project / "local.properties"
+        if properties.exists():
+            for line in properties.read_text().splitlines():
+                if line.startswith("sdk.dir="):
+                    sdk = Path(line.split("=", 1)[1])
+    if sdk is None or not args.apk.is_file():
+        parser.error("Build assembleDebug first and specify --sdk or sdk.dir in local.properties")
+    android = sdk / "platforms/android-36/android.jar"
+    versions = sorted((sdk / "build-tools").glob("*/d8"), key=lambda p: tuple(int(n) for n in p.parent.name.split(".")))
+    if not android.is_file() or not versions:
+        parser.error("Android API 36 and build-tools with d8 are required")
+    adb = ["adb", "-s", args.serial]
+    remote = "/data/local/tmp/utun-stream-worker-smoke"
+
+    def run(*command):
+        subprocess.run([str(part) for part in command], check=True)
+
+    with tempfile.TemporaryDirectory(prefix="utun-stream-worker-smoke-") as directory:
+        work = Path(directory)
+        classes = work / "classes"
+        classes.mkdir()
+        run("javac", "-cp", android, "-d", classes, Path(__file__).with_name("StreamWorkerSmoke.java"))
+        jar = work / "smoke.jar"
+        run(versions[-1], "--lib", android, "--min-api", "26", "--output", jar, classes / "StreamWorkerSmoke.class")
+        frame = work / "key.h264"
+        run("ffmpeg", "-nostdin", "-loglevel", "error", "-f", "lavfi", "-i", "color=size=128x72:rate=30:color=blue",
+            "-frames:v", "1", "-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency",
+            "-x264-params", "repeat-headers=1", "-f", "h264", frame)
+        abi = subprocess.check_output(adb + ["shell", "getprop", "ro.product.cpu.abi"], text=True).strip()
+        libraries = []
+        with zipfile.ZipFile(args.apk) as apk:
+            for name in apk.namelist():
+                if name.startswith(f"lib/{abi}/") and name.endswith(".so"):
+                    target = work / Path(name).name
+                    target.write_bytes(apk.read(name))
+                    libraries.append(target)
+        if not libraries:
+            parser.error(f"APK has no native libraries for {abi}")
+        run(*adb, "shell", "mkdir", "-p", remote)
+        run(*adb, "push", jar, frame, *libraries, remote + "/")
+        run(*adb, "push", args.apk, remote + "/app.apk")
+        command = [f"CLASSPATH={remote}/smoke.jar:{remote}/app.apk", "app_process",
+                   f"-Djava.library.path={remote}", remote, "StreamWorkerSmoke", remote + "/key.h264"]
+        print("Testing real MediaCodec/Opus/AudioTrack: 500ms video stall, EOF and reset; audio content is silence", flush=True)
+        run(*adb, "shell", " ".join(shlex.quote(part) for part in command))
+
+
+if __name__ == "__main__":
+    main()