feat(bridge): carry bridge traffic over simlock's multiplexed link

.nx/version-plans/version-plan-1792067880723.md

@@ -0,0 +1,5 @@
+---
+__default__: patch
+---
+
+The connection between the Harness CLI and the app now sends screenshot data in flow-controlled chunks on its own channel, with test events and heartbeats sent ahead of it. The CLI and the app runtime must use the same Harness version.

packages/bridge/package.json

@@ -33,6 +33,7 @@
     "pngjs": "^7.0.0",
     "ssim.js": "^3.5.0",
     "tslib": "^2.3.0",
+    "web-streams-polyfill": "^4.3.0",
     "ws": "^8.18.2"
   },
   "devDependencies": {

packages/bridge/src/__tests__/client.test.ts

@@ -1,99 +1,132 @@
 import { describe, expect, it, vi } from 'vitest';
+import { receiveBinary } from '../binary.js';
 import { connectToHarness } from '../client.js';
-import type {
-  BridgeTransport,
-  BridgeTransportCloseEvent,
-  BridgeTransportMessage,
-  BridgeTransportState,
-} from '../transport.js';
-
-const createMockTransport = (initialState: BridgeTransportState = 'connecting') => {
-  let state = initialState;
-  const openListeners = new Set<() => void>();
-  const messageListeners = new Set<(message: BridgeTransportMessage) => void>();
-  const closeListeners = new Set<(event: BridgeTransportCloseEvent) => void>();
-  const errorListeners = new Set<(error: Error) => void>();
-  const sent: BridgeTransportMessage[] = [];
-
-  const transport: BridgeTransport = {
-    get state() {
-      return state;
-    },
-    send: (message) => {
-      sent.push(message);
-    },
-    close: (code = 1000, reason = '') => {
-      state = 'closed';
-      for (const listener of closeListeners) {
-        listener({ code, reason });
-      }
-    },
-    onOpen: (listener) => {
-      openListeners.add(listener);
-      return () => openListeners.delete(listener);
-    },
-    onMessage: (listener) => {
-      messageListeners.add(listener);
-      return () => messageListeners.delete(listener);
-    },
-    onClose: (listener) => {
-      closeListeners.add(listener);
-      return () => closeListeners.delete(listener);
-    },
-    onError: (listener) => {
-      errorListeners.add(listener);
-      return () => errorListeners.delete(listener);
-    },
-  };
+import { createLink, globalClock, type Logger } from '../link/index.js';
+import { createConnectionPair } from '../link/__tests__/ports.js';
+import type { BridgeMessage } from '../protocol.js';
 
-  return {
-    transport,
-    sent,
-    open: () => {
-      state = 'open';
-      for (const listener of openListeners) {
-        listener();
-      }
-    },
-    receive: (message: BridgeTransportMessage) => {
-      for (const listener of messageListeners) {
-        listener(message);
-      }
-    },
-    fail: (error: Error) => {
-      for (const listener of errorListeners) {
-        listener(error);
+const silentLogger: Logger = {
+  debug: () => undefined,
+  info: () => undefined,
+  warn: () => undefined,
+  error: () => undefined,
+  child: () => silentLogger,
+};
+
+const setup = async () => {
+  const [appEnd, cliEnd] = createConnectionPair();
+  const cli = createLink(cliEnd, {
+    end: 'gateway',
+    clock: globalClock,
+    logger: silentLogger,
+  });
+  const received: BridgeMessage[] = [];
+  const waiters: Array<() => void> = [];
+  cli.onData((chunk) => {
+    received.push(JSON.parse(chunk) as BridgeMessage);
+    for (const waiter of waiters.splice(0)) {
+      waiter();
+    }
+  });
+  const nextMessage = async (type: BridgeMessage['type']) => {
+    for (;;) {
+      const message = received.find((candidate) => candidate.type === type);
+      if (message) {
+        received.splice(received.indexOf(message), 1);
+        return message;
       }
-    },
+      await new Promise<void>((resolve) => waiters.push(resolve));
+    }
   };
+
+  const handle = await connectToHarness(
+    'ws://unused',
+    { runTests: vi.fn(), resetEnvironment: vi.fn() },
+    { transport: appEnd },
+  );
+
+  return { cli, handle, nextMessage };
 };
 
-describe('connectToHarness transport injection', () => {
+describe('connectToHarness', () => {
   it('uses the injected transport instead of constructing a WebSocket', async () => {
-    const mockTransport = createMockTransport();
-    const handlePromise = connectToHarness(
-      'ws://unused',
-      { runTests: vi.fn(), resetEnvironment: vi.fn() },
-      { transport: mockTransport.transport },
-    );
-
-    mockTransport.open();
+    const { handle, nextMessage } = await setup();
 
-    const handle = await handlePromise;
     handle.reportReady({
       platform: 'ios',
       manufacturer: 'Apple',
       model: 'Injected',
       osVersion: '18.0',
     });
 
-    expect(mockTransport.sent).toHaveLength(1);
-    expect(mockTransport.sent[0]).toBeTypeOf('string');
-    expect(JSON.parse(mockTransport.sent[0] as string)).toMatchObject({
+    expect(await nextMessage('ready')).toMatchObject({
       type: 'ready',
       device: { model: 'Injected' },
     });
 
     handle.disconnect();
   });
+
+  it('answers pings with pongs', async () => {
+    const { cli, handle, nextMessage } = await setup();
+
+    await cli.write(JSON.stringify({ type: 'ping', id: 7 }));
+
+    expect(await nextMessage('pong')).toEqual({ type: 'pong', id: 7 });
+
+    handle.disconnect();
+  });
+
+  it('sends screenshot bytes over a link channel, not inside the RPC message', async () => {
+    const { cli, handle, nextMessage } = await setup();
+    const data = new Uint8Array(3 * 1024 * 1024).map((_, index) => index % 251);
+
+    const transfer = handle.transferScreenshot(data, { width: 10, height: 20 });
+    const invoke = await nextMessage('invoke');
+
+    if (invoke.type !== 'invoke') {
+      throw new Error('expected an invoke');
+    }
+
+    const [reference, metadata] = invoke.args as [
+      { channelId: string; size: number },
+      unknown,
+    ];
+    expect(invoke.method).toBe('device.screenshot.receive');
+    expect(metadata).toEqual({ width: 10, height: 20 });
+    expect(reference.size).toBe(data.length);
+
+    const bytes = await receiveBinary(cli, reference.channelId);
+    expect(Buffer.from(bytes).equals(Buffer.from(data))).toBe(true);
+
+    await cli.write(
+      JSON.stringify({
+        type: 'return',
+        id: invoke.id,
+        ok: true,
+        value: { path: '/tmp/shot.png' },
+      }),
+    );
+
+    await expect(transfer).resolves.toEqual({ path: '/tmp/shot.png' });
+
+    handle.disconnect();
+  });
+
+  it('rejects pending calls once the link closes', async () => {
+    const { cli, handle, nextMessage } = await setup();
+
+    const call = handle.matchImageSnapshot(
+      { path: '/tmp/shot.png' },
+      '/tests/a.test.ts',
+      { name: 'a' },
+      'ios',
+    );
+    await nextMessage('invoke');
+    await cli.close();
+
+    await expect(call).rejects.toThrow('Harness connection closed');
+
+    handle.disconnect();
+  });
 });

packages/bridge/src/__tests__/rpc-peer.test.ts

@@ -1,7 +1,7 @@
 import { describe, expect, it, vi } from 'vitest';
 import { createRpcPeer } from '../rpc-peer.js';
 import type { BridgeEvents } from '../shared.js';
-import type { RpcTransport } from '../transport.js';
+import type { RpcTransport } from '../rpc-peer.js';
 
 type LocalMethods = {
   add: (left: number, right: number) => Promise<number>;
@@ -12,7 +12,6 @@ const createMockTransport = (): RpcTransport & { messages: string[] } => {
   const messages: string[] = [];
 
   return {
-    state: 'open',
     messages,
     send: (message) => {
       messages.push(message);
@@ -27,7 +26,6 @@ describe('rpc-peer', () => {
     const client = createRpcPeer<Record<string, never>, { add: LocalMethods['add'] }, BridgeEvents>({
       localMethods: {},
       transport: {
-        state: 'open',
         send: (message) => {
           if (!serverHandleMessage) {
             throw new Error('Server peer not initialized');
@@ -46,7 +44,6 @@ describe('rpc-peer', () => {
         },
       },
       transport: {
-        state: 'open',
         send: (message) => {
           void client.handleMessage(message);
         },
@@ -77,7 +74,6 @@ describe('rpc-peer', () => {
         },
       },
       transport: {
-        state: 'open',
         send: (message) => {
           void client.handleMessage(message);
         },
@@ -96,7 +92,6 @@ describe('rpc-peer', () => {
     const peer = createRpcPeer<Record<string, never>, Record<string, never>, BridgeEvents>({
       localMethods: {},
       transport: {
-        state: 'open',
         send: (message) => {
           response = message;
         },