Function openSubscriber

packages/server/test/prompt.e2e.test.ts:212–244

* Open a WS subscriber and wait for server_hello + client_hello ack. * Returns a handle exposing the received frame queue. * * Mirrors the queueing pattern from `ws-broadcast.e2e.test.ts` — message * listener is attached BEFORE the `open` event resolves, so frames that land * in the same tick a

(
  r: RunningServer,
  sid: string,
)

 * in the same tick as the upgrade aren't lost.
 */
async function openSubscriber(
  r: RunningServer,
  sid: string,
): Promise<{
  ws: WebSocket;
  received: Record<string, unknown>[];
}> {
  const wsUrl = r.address.replace('http://', 'ws://') + '/api/v1/ws';
  const received: Record<string, unknown>[] = [];
  const ws = await new Promise<WebSocket>((resolve, reject) => {
    const sock = new WebSocket(wsUrl, ['kimi-code.bearer.test-token']);
    sock.on('message', (data) => {
      try {
        received.push(JSON.parse(wsDataToString(data)) as Record<string, unknown>);
      } catch {
        // ignore
      }
    });
    sock.once('open', () => resolve(sock));
    sock.once('error', reject);
  });
  // Wait for server_hello.
  await waitFor(received, (f) => f['type'] === 'server_hello');
  ws.send(
    JSON.stringify({
      type: 'client_hello',
      id: 'h1',
      payload: { client_id: 'test', subscriptions: [sid] },
    }),
  );
  await waitFor(received, (f) => f['type'] === 'ack' && f['id'] === 'h1');
  return { ws, received };
}

async function waitFor(
  received: Record<string, unknown>[],