| import assert from "node:assert/strict"; |
| import { once } from "node:events"; |
| import http from "node:http"; |
| import path from "node:path"; |
| import { Duplex } from "node:stream"; |
| import { setTimeout as delay } from "node:timers/promises"; |
| import { WebSocket } from "../../../packages/gateway-client/src/websocket.js"; |
| import { setLoggerOverride } from "../../logging/logger.js"; |
| import { createNodeDesktopStreamBroker } from "./node-stream-broker.js"; |
| import { handleDesktopObserveUpgrade, mintDesktopObserverToken } from "./observe-bridge.js"; |
|
|
| async function until(predicate: () => boolean) { |
| const deadline = Date.now() + 2_000; |
| while (!predicate()) { |
| assert.ok(Date.now() < deadline, "desktop transport did not settle"); |
| await delay(5); |
| } |
| } |
|
|
| async function listen(server: http.Server) { |
| server.listen(0, "127.0.0.1"); |
| await once(server, "listening"); |
| const address = server.address(); |
| assert.ok(address && typeof address !== "string"); |
| return `ws://127.0.0.1:${address.port}`; |
| } |
|
|
| async function observe(mode: string) { |
| const received: Buffer[] = []; |
| let releaseWrite: (() => void) | undefined; |
| let released = 0; |
| let gatewaySocket: WebSocket | undefined; |
| let closeObserver: ((code: number, reason: string) => void) | undefined; |
| const stream = new Duplex({ |
| writableHighWaterMark: 1, |
| read() {}, |
| write(chunk, _encoding, callback) { |
| received.push(Buffer.from(chunk)); |
| releaseWrite = callback; |
| }, |
| }); |
| const server = http.createServer(); |
| server.on("upgrade", (req, socket, head) => { |
| handleDesktopObserveUpgrade(req, socket, head, { |
| registry: { |
| claimStream: () => stream, |
| attachObserver: (_sourceKey, observer) => { |
| closeObserver = (code, reason) => observer.close(code, reason); |
| return { release: () => released++ }; |
| }, |
| }, |
| getBufferedAmount: (ws) => { |
| gatewaySocket = ws; |
| return ws.bufferedAmount; |
| }, |
| }); |
| }); |
| const url = await listen(server); |
| const { token } = mintDesktopObserverToken({ |
| sourceKey: "worker:runtime-proof", |
| ownerEpoch: 1, |
| control: true, |
| attachment: { kind: "stream", streamId: "runtime-proof" }, |
| }); |
| const client = new WebSocket(`${url}/desktop/observe?token=${token}`); |
| try { |
| await once(client, "open"); |
| const banner = once(client, "message"); |
| stream.push(Buffer.from("RFB 003.008\n")); |
| assert.equal((await banner)[0].toString(), "RFB 003.008\n"); |
|
|
| if (mode === "observer-backpressure") { |
| client.send(Buffer.from("first")); |
| await until(() => Boolean(releaseWrite)); |
| assert.equal(gatewaySocket?.isPaused, true); |
| client.send(Buffer.from("second")); |
| await delay(25); |
| assert.equal(stream.writableLength, 5); |
| assert.equal(Buffer.concat(received).toString(), "first"); |
| const firstWrite = releaseWrite; |
| releaseWrite = undefined; |
| firstWrite?.(); |
| await until(() => Boolean(releaseWrite)); |
| assert.equal(Buffer.concat(received).toString(), "firstsecond"); |
| assert.equal(gatewaySocket?.isPaused, true); |
| const closed = once(client, "close"); |
| closeObserver?.(1012, "desktop stopped"); |
| assert.equal((await closed)[0], 1012); |
| } else if (mode === "observer-payload") { |
| const closed = once(client, "close"); |
| client.send(Buffer.alloc(1024 * 1024 + 1)); |
| assert.equal((await closed)[0], 1009); |
| assert.equal(received.length, 0); |
| } else { |
| const closed = once(client, "close"); |
| client.close(); |
| await closed; |
| } |
| await until(() => stream.destroyed && released === 1); |
| } finally { |
| client.terminate(); |
| stream.destroy(); |
| await new Promise<void>((resolve) => { |
| server.close(() => resolve()); |
| }); |
| } |
| } |
|
|
| async function nodeStream(kind: "desktop" | "portal") { |
| const broker = createNodeDesktopStreamBroker(); |
| const binding = { nodeId: "runtime-node", connId: "runtime-connection", pairingGeneration: "1" }; |
| const ticket = kind === "desktop" ? broker.mint(binding) : broker.mintPortal(binding); |
| const server = http.createServer(); |
| server.on("upgrade", (req, socket, head) => { |
| void broker.handleUpgrade(req, socket, head, { |
| getForPairingGeneration: () => ({ |
| ...binding, |
| client: { |
| connId: binding.connId, |
| socket: client, |
| usesSharedGatewayAuth: false, |
| connect: { |
| minProtocol: 1, |
| maxProtocol: 1, |
| client: { id: "node-host", version: "test", platform: "test", mode: "node" }, |
| }, |
| }, |
| declaredCaps: [], |
| caps: [], |
| declaredCommands: [], |
| commands: [], |
| declaredNodePluginTools: [], |
| nodePluginTools: [], |
| nodeSkills: [], |
| connectedAtMs: 0, |
| }), |
| isConnectionCurrentPairingState: async () => true, |
| }); |
| }); |
| const url = await listen(server); |
| const client = new WebSocket(`${url}${ticket.attachPath}`); |
| let stream: Duplex | undefined; |
| try { |
| await once(client, "open"); |
| client.send( |
| Buffer.from(JSON.stringify(kind === "desktop" ? { auth: "vnc-password" } : { ok: true })), |
| ); |
| ({ stream } = await ticket.attached); |
| const incoming = once(stream, "data"); |
| client.send(Buffer.from("node-to-gateway")); |
| assert.equal((await incoming)[0].toString(), "node-to-gateway"); |
| const outgoing = once(client, "message"); |
| stream.write(Buffer.from("gateway-to-node")); |
| assert.equal((await outgoing)[0].toString(), "gateway-to-node"); |
| const closed = once(client, "close"); |
| stream.on("error", () => undefined); |
| client.send(Buffer.alloc(64 * 1024 + 1)); |
| assert.equal((await closed)[0], 1009); |
| await until(() => Boolean(stream?.destroyed)); |
| } finally { |
| client.terminate(); |
| stream?.destroy(); |
| ticket.cancel(); |
| await new Promise<void>((resolve) => { |
| server.close(() => resolve()); |
| }); |
| } |
| } |
|
|
| export async function runDesktopWebSocketRuntimeProbe(mode: string) { |
| setLoggerOverride({ |
| level: "silent", |
| consoleLevel: "silent", |
| file: path.join(process.cwd(), "desktop-runtime.log"), |
| }); |
| if (mode === "desktop" || mode === "portal") { |
| await nodeStream(mode); |
| } else { |
| await observe(mode); |
| } |
| } |
|
|
