From 5cbe82a87006a7ee472f0f9c9fce5d5f24caa795 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 11 Oct 2026 13:35:57 -0700 Subject: [PATCH] fix(devices): refresh backlogged iOS video and require input admission (#18326) --- .../src/device/DeviceHubProxy.stream.test.ts | 89 +++++++++++++++++ .../server/src/device/deviceToolVersions.json | 2 +- .../device/DeviceStreamView.test.tsx | 1 + .../src/device/duoStream.test.ts | 1 + .../client-runtime/src/device/stream.test.ts | 96 +++++++++++++++++-- packages/client-runtime/src/device/stream.ts | 57 +++++------ 6 files changed, 209 insertions(+), 37 deletions(-) create mode 100644 apps/server/src/device/DeviceHubProxy.stream.test.ts diff --git a/apps/server/src/device/DeviceHubProxy.stream.test.ts b/apps/server/src/device/DeviceHubProxy.stream.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..0251dabf7ace7ecc29b83e5c32428af9fd3797dd --- /dev/null +++ b/apps/server/src/device/DeviceHubProxy.stream.test.ts @@ -0,0 +1,89 @@ +import { expect, it } from "@effect/vitest"; +import { NodeWS } from "@effect/platform-node/NodeSocket"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import { HttpRouter, HttpServer } from "effect/http"; +import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; +import { + AuthOrchestrationOperateScope, + AuthSessionId, + LOCAL_DEVICE_HOST_ID, +} from "@t3tools/contracts"; +import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; +import * as DeviceService from "./DeviceService.ts"; +import * as DeviceHubProxy from "./DeviceHubProxy.ts"; + +it.effect("preserves binary input and admission frames through the real socket proxy", () => + Effect.gen(function* () { + const hub = yield* Effect.acquireRelease( + Effect.sync(() => new NodeWS.WebSocketServer({ port: 0 })), + (server) => + Effect.promise( + () => + new Promise((resolve) => { + for (const client of server.clients) client.terminate(); + server.close(() => resolve()); + }), + ), + ); + if (!hub.address()) + yield* Effect.promise(() => new Promise((resolve) => hub.once("listening", resolve))); + const address = hub.address(); + if (typeof address !== "object" || !address) throw new Error("Missing hub port"); + hub.on("connection", (socket) => { + socket.send(Uint8Array.of(0x83)); + socket.on("message", (data, binary) => socket.send(data, { binary })); + }); + const services = yield* Layer.build( + HttpRouter.serve(DeviceHubProxy.layer, { disableListenLog: true }).pipe( + Layer.provideMerge(NodeHttpServer.layerTest), + Layer.provide( + Layer.succeed(EnvironmentAuth.EnvironmentAuth, { + authenticateWebSocketUpgrade: () => + Effect.succeed({ + sessionId: AuthSessionId.make("test"), + subject: "test", + method: "bearer-access-token", + scopes: [AuthOrchestrationOperateScope], + }), + } as unknown as EnvironmentAuth.EnvironmentAuth["Service"]), + ), + Layer.provide( + Layer.succeed(DeviceService.DeviceService, { + currentReadiness: () => + Effect.succeed({ + hostId: LOCAL_DEVICE_HOST_ID, + hub: { origin: `http://127.0.0.1:${address.port}` }, + }), + } as DeviceService.DeviceService["Service"]), + ), + ), + ); + const origin = HttpServer.formatAddress( + Context.get(services, HttpServer.HttpServer).address, + ).replace(/^http/, "ws"); + const client = yield* Effect.acquireRelease( + Effect.sync( + () => + new WebSocket( + `${origin}/api/device-hub/vendor/serve-sim/helper/ws?device=test&wsTicket=test`, + ), + ), + (socket) => Effect.sync(() => socket.close()), + ); + client.binaryType = "arraybuffer"; + const frames: number[][] = []; + const received = Promise.withResolvers(); + client.addEventListener("error", () => received.reject(new Error("Proxy failed"))); + client.addEventListener("message", ({ data }) => { + frames.push(Array.from(new Uint8Array(data as ArrayBuffer))); + if (frames.length === 1) { + client.send(Uint8Array.of(0x03, 1)); + client.send(Uint8Array.of(0x03, 2)); + } else if (frames.length === 3) received.resolve(); + }); + yield* Effect.promise(() => received.promise); + expect(frames).toEqual([[0x83], [0x03, 1], [0x03, 2]]); + }), +); diff --git a/apps/server/src/device/deviceToolVersions.json b/apps/server/src/device/deviceToolVersions.json index 8d2e5a3efebc709f34e1e7551fea7fd5ce8d61f1..e0c2880e3f657a4c08adf2a52e71914d24b21546 100644 --- a/apps/server/src/device/deviceToolVersions.json +++ b/apps/server/src/device/deviceToolVersions.json @@ -1,4 +1,4 @@ { - "agent-device": "0.21.24", + "agent-device": "0.21.25", "expo-device-hub": "0.15.3" } diff --git a/apps/web/src/components/device/DeviceStreamView.test.tsx b/apps/web/src/components/device/DeviceStreamView.test.tsx index a7f4079b5683610475ba86073a9287415232f1cd..559050303cf4ff58d0aa98429a1a41612e36d1d6 100644 --- a/apps/web/src/components/device/DeviceStreamView.test.tsx +++ b/apps/web/src/components/device/DeviceStreamView.test.tsx @@ -157,6 +157,7 @@ async function setup(h264 = false) { const packet = new Uint8Array(1 + json.length); packet[0] = 0x82; packet.set(json, 1); + sockets[0]?.onmessage?.({ data: Uint8Array.of(0x83).buffer }); sockets[0]?.onmessage?.({ data: packet.buffer }); videoBody?.enqueue(new Uint8Array([0, 0, 0, 5, 1, 1, 0x64, 0, 0x1f])); }, diff --git a/packages/client-runtime/src/device/duoStream.test.ts b/packages/client-runtime/src/device/duoStream.test.ts index f290bef1712d4dad3de873048d7feb8c4107fcdf..9ac072957e0a78181e9754db5245c5515632ca8c 100644 --- a/packages/client-runtime/src/device/duoStream.test.ts +++ b/packages/client-runtime/src/device/duoStream.test.ts @@ -111,6 +111,7 @@ it("switches to fixed authenticated feeds while retaining one HID socket, routes client.start(); await socketConstructed; const ws = Socket.instances[0]!; + ws.onmessage?.({ data: Uint8Array.of(0x83).buffer }); const config = ( id: number, orientation: DeviceScreenSize["orientation"] = "portrait", diff --git a/packages/client-runtime/src/device/stream.test.ts b/packages/client-runtime/src/device/stream.test.ts index f29484eb00b43f69caa9c4ec2a57e39a124a4c87..e92f1a8fa19847b5ae154c59d4328afc4ceb068f 100644 --- a/packages/client-runtime/src/device/stream.test.ts +++ b/packages/client-runtime/src/device/stream.test.ts @@ -64,6 +64,7 @@ describe("native device stream transport", () => { readyState = 1; binaryType = ""; onopen: (() => void) | null = null; + onmessage: ((event: { data: ArrayBuffer }) => void) | null = null; onclose: ((event: { code: number; reason: string }) => void) | null = null; send = vi.fn(); close = vi.fn(); @@ -111,10 +112,12 @@ describe("native device stream transport", () => { expect(imageUrl.searchParams.get("hostId")).toBe("ssh-host"); expect(fetch).toHaveBeenCalledTimes(1); socket.onopen?.(); + expect(socket.send).not.toHaveBeenCalled(); + socket.onmessage?.({ data: Uint8Array.of(0x83).buffer }); client.setOrientation("landscape_right"); client.pressButton("home"); client.sendTouch("begin", 0.25, 0.75); - const messages = socket.send.mock.calls.slice(1).map(([bytes]) => { + const messages = socket.send.mock.calls.map(([bytes]) => { const packet = bytes as Uint8Array; return { tag: packet[0], @@ -179,6 +182,7 @@ describe("native device stream transport", () => { client.start(); const replacement = await reopened; replacement.onopen?.(); + if (platform === "ios") replacement.onmessage?.({ data: Uint8Array.of(0x83).buffer }); replacement.send.mockClear(); events.onInputConnected.mockClear(); @@ -336,6 +340,7 @@ describe("iOS input startup", () => { packet[0] = 0x82; packet.set(json, 1); socket.onmessage?.({ data: packet.buffer }); + socket.onmessage?.({ data: Uint8Array.of(0x83).buffer }); client.sendTouch("begin", 0.2, 0.7); const sent = socket.send.mock.calls[0]![0] as Uint8Array; expect(JSON.parse(new TextDecoder().decode(sent.subarray(1)))).toEqual({ @@ -519,27 +524,100 @@ describe("shared device stream readiness and recovery", () => { expect(vi.getTimerCount()).toBe(0); }); - it("drops an iOS decode backlog and resumes at a keyframe without switching to MJPEG", async () => { - const { client, videoBodies, decoders, events, decodedFrame } = recoveryFixture("ios", false); + it("requires admission on each iOS socket and does not mistake config for admission", async () => { + const { client, sockets, events, imageReady } = recoveryFixture(); client.start(); await vi.advanceTimersByTimeAsync(0); + imageReady(); + const config = (socket: (typeof sockets)[number], unavailable: boolean) => { + const json = new TextEncoder().encode( + JSON.stringify({ + width: 400, + height: 800, + orientation: "portrait", + inputUnavailable: unavailable, + supportsHingeAngle: true, + }), + ); + socket.onmessage?.({ data: Uint8Array.of(0x82, ...json).buffer }); + }; + const controls = () => { + client.sendTouch("begin", 0.2, 0.3); + client.pressButton("home"); + client.sendKey({ code: "KeyA" } as KeyboardEvent, "down"); + client.setOrientation("landscape_left"); + client.controlDuo({ control: "angle", value: 90 }); + }; + const first = sockets[0]!; + config(first, false); + controls(); + expect(first.send).not.toHaveBeenCalled(); + expect(events.onInputConnected).not.toHaveBeenCalledWith(true, undefined); + config(first, true); + first.onmessage?.({ data: Uint8Array.of(0x83).buffer }); + controls(); + expect(first.send).not.toHaveBeenCalled(); + expect(events.onInputConnected).toHaveBeenLastCalledWith(false); + config(first, false); + client.pressButton("home"); + expect(first.send).toHaveBeenCalled(); + first.onclose?.({ code: 1013, reason: "retry" }); + await vi.advanceTimersByTimeAsync(1_000); + const replacement = sockets[1]!; + config(replacement, false); + controls(); + expect(replacement.send).not.toHaveBeenCalled(); + first.onmessage?.({ data: Uint8Array.of(0x83).buffer }); + expect(events.onInputConnected).toHaveBeenLastCalledWith(false, undefined); + replacement.onmessage?.({ data: Uint8Array.of(0x83).buffer }); + client.pressButton("home"); + expect(replacement.send).toHaveBeenCalledOnce(); + client.stop(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("reopens a backlogged iOS feed for a fresh keyframe while keeping input connected", async () => { + const { client, videoBodies, decoders, events, decodedFrame, signals, sockets, drawImage } = + recoveryFixture("ios", false); + client.start(); + await vi.advanceTimersByTimeAsync(0); + const socket = sockets[0]!; + socket.onmessage?.({ data: Uint8Array.of(0x83).buffer }); const body = videoBodies[0]!; body.enqueue(new Uint8Array([...envelope(1, [1, 0x64, 0, 0x1f]), ...envelope(2, [1])])); await vi.advanceTimersByTimeAsync(0); decodedFrame(); const decoder = decoders[0]!; - expect(decoder.decode).toHaveBeenCalledOnce(); decoder.decodeQueueSize = 9; body.enqueue(new Uint8Array([...envelope(3, [2]), ...envelope(3, [3])])); await vi.advanceTimersByTimeAsync(0); - expect(decoder.reset).toHaveBeenCalledOnce(); - expect(decoder.decode).toHaveBeenCalledOnce(); - body.enqueue(new Uint8Array([...envelope(2, [4]), ...envelope(3, [5])])); + expect(signals[1]!.aborted).toBe(true); + expect(decoder.state).toBe("closed"); + expect(events.onStatus).toHaveBeenLastCalledWith( + "connecting", + expect.stringContaining("fell behind"), + ); + expect(socket.close).not.toHaveBeenCalled(); + client.sendTouch("begin", 0.2, 0.3); + expect(socket.send).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1_000); + expect(videoBodies).toHaveLength(2); + // The old feed never produces another IDR. Only a fresh subscription can recover. + videoBodies[1]!.enqueue( + new Uint8Array([...envelope(1, [1, 0x64, 0, 0x1f]), ...envelope(2, [4])]), + ); await vi.advanceTimersByTimeAsync(0); - expect(decoder.decode).toHaveBeenCalledTimes(3); + expect(decoders[1]!.decode).toHaveBeenCalledOnce(); + drawImage.mockClear(); + decodedFrame(0); + expect(drawImage).not.toHaveBeenCalled(); + decodedFrame(); + expect(drawImage).toHaveBeenCalledOnce(); expect(events.onMjpegFallback).not.toHaveBeenCalled(); expect(events.onStatus).toHaveBeenLastCalledWith("streaming", undefined); + expect(events.onInputConnected).toHaveBeenLastCalledWith(true); client.stop(); + expect(vi.getTimerCount()).toBe(0); }); it("does not repaint a delayed JPEG seed over newer decoded video", async () => { @@ -614,7 +692,7 @@ describe("shared device stream readiness and recovery", () => { expect(sockets[0]!.close).toHaveBeenCalledOnce(); expect(image.src).toBe(""); client.pressButton("home"); - expect(sockets[0]!.send).toHaveBeenCalledTimes(1); + expect(sockets[0]!.send).not.toHaveBeenCalled(); expect(vi.getTimerCount()).toBe(0); }); diff --git a/packages/client-runtime/src/device/stream.ts b/packages/client-runtime/src/device/stream.ts index 27ab7c93b416673fa6ef81e22e58472a5815ca16..ae83bf93555c52b556ee3b07641f3d27f8a3d3c5 100644 --- a/packages/client-runtime/src/device/stream.ts +++ b/packages/client-runtime/src/device/stream.ts @@ -128,7 +128,6 @@ const IOS_MSG_TOUCH = 0x03; const IOS_MSG_BUTTON = 0x04; const IOS_MSG_KEY = 0x06; const IOS_MSG_ORIENTATION = 0x07; -const IOS_MSG_HARDWARE_KEYBOARD = 0x0d; // helper -> browser. const IOS_TAG_SCREEN_CONFIG = 0x82; const IOS_TAG_INPUT_ADMITTED = 0x83; @@ -366,8 +365,8 @@ export function createDeviceStreamClient( const retryTimers = new Map<"video" | "input", ReturnType>(); let primeController: AbortController | null = null; let videoDecoder: VideoDecoder | null = null; - let decoderConfig: VideoDecoderConfig | null = null; let decodedVideoGeneration: number | null = null; + let iosInputAdmitted = false; let iosInputUnavailable = false; let timestamp = 0; let awaitingKeyframe = true; @@ -388,7 +387,13 @@ export function createDeviceStreamClient( let pendingOrientation: { requestId: number } | null = null; const duoControl = createDuoControl({ send(request) { - if (socket?.readyState !== WebSocket.OPEN || !screen?.supportsHingeAngle) return false; + if ( + !iosInputAdmitted || + iosInputUnavailable || + socket?.readyState !== WebSocket.OPEN || + !screen?.supportsHingeAngle + ) + return false; if (request.command.control === "physical" && !screen.supportsPhysicalOrientation) return false; try { @@ -522,7 +527,6 @@ export function createDeviceStreamClient( // Already closed. } videoDecoder = null; - decoderConfig = null; awaitingKeyframe = true; }; @@ -580,7 +584,6 @@ export function createDeviceStreamClient( try { if (!videoDecoder || videoDecoder.state === "closed") videoDecoder = makeDecoder(); videoDecoder.configure(full); - decoderConfig = full; return true; } catch (cause) { if (platform === "android") fail(`Video decoder: ${(cause as Error).message}`); @@ -595,22 +598,15 @@ export function createDeviceStreamClient( awaitingKeyframe = false; } if (videoDecoder.decodeQueueSize > SOFT_DECODE_QUEUE) { - if (platform !== "ios" || !decoderConfig) { - recoverDecoder(); - return; - } - // A slow viewer can accumulate healthy H.264 frames. Drop that backlog - // and resume at an IDR instead of permanently giving up its 3D view. - try { - videoDecoder.reset(); - videoDecoder.configure(decoderConfig); - awaitingKeyframe = true; - if (!isKey) return; - awaitingKeyframe = false; - } catch { + if (platform !== "ios") { recoverDecoder(); return; } + // Native IDRs are spaced by frame count, so an idle simulator may take + // minutes to produce one. A new AVCC subscription forces a fresh IDR. + controller?.abort(); + closeDecoder(); + return; } try { videoDecoder.decode( @@ -698,6 +694,7 @@ export function createDeviceStreamClient( if (!isCurrent()) return; if (done) break; for (const chunk of demuxer.push(value)) { + if (videoController.signal.aborted) break; switch (chunk.type) { case "seed": void createImageBitmap(new Blob([chunk.payload as BlobPart], { type: "image/jpeg" })) @@ -740,7 +737,9 @@ export function createDeviceStreamClient( } } catch (cause) { if (!isCurrent()) return; - retryDetail ??= (cause as Error).message; + retryDetail ??= videoController.signal.aborted + ? "Video decoder fell behind. Reconnecting…" + : (cause as Error).message; } if (isCurrent()) { controller = null; @@ -822,18 +821,16 @@ export function createDeviceStreamClient( const ws = new WebSocket(wsUrl(`/helper/ws?device=${device}`)); ws.binaryType = "arraybuffer"; socket = ws; - ws.onopen = () => { - if (stopped || socket !== ws) return; - ws.send(taggedJson(IOS_MSG_HARDWARE_KEYBOARD, { enabled: false })); - }; + iosInputAdmitted = false; + iosInputUnavailable = false; ws.onmessage = (event) => { if (stopped || socket !== ws) return; if (!(event.data instanceof ArrayBuffer)) return; const bytes = new Uint8Array(event.data); if (socket !== ws || stopped || bytes.length < 1) return; if (bytes[0] === IOS_TAG_INPUT_ADMITTED) { - iosInputUnavailable = false; - events.onInputConnected(true); + iosInputAdmitted = true; + events.onInputConnected(!iosInputUnavailable); return; } try { @@ -846,7 +843,7 @@ export function createDeviceStreamClient( if (Option.isSome(config)) { iosInputUnavailable = config.value.inputUnavailable === true; events.onInputConnected( - !iosInputUnavailable, + iosInputAdmitted && !iosInputUnavailable, iosInputUnavailable ? "Simulator input is unavailable" : undefined, ); const previous = screen; @@ -884,6 +881,7 @@ export function createDeviceStreamClient( ws.onclose = (event) => { if (socket !== ws) return; socket = null; + iosInputAdmitted = false; duoControl.clear(); rotationCursor = null; if (stopped) return; @@ -979,6 +977,7 @@ export function createDeviceStreamClient( stopped = false; generation++; configuring = false; + iosInputAdmitted = false; iosInputUnavailable = false; connecting(); if (platform === "ios") { @@ -1022,7 +1021,11 @@ export function createDeviceStreamClient( }; const send = (payload: Uint8Array | string) => { - if (!stopped && !iosInputUnavailable && socket?.readyState === WebSocket.OPEN) + if ( + !stopped && + (platform !== "ios" || (iosInputAdmitted && !iosInputUnavailable)) && + socket?.readyState === WebSocket.OPEN + ) socket.send(payload); }; -- 2.54.0