diff --git a/package-lock.json b/package-lock.json index 3aca3c96..0c7a53a9 100644 --- a/package-lock.json +++ b/package-lock.json @@ -401,7 +401,6 @@ "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz", "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -423,7 +422,6 @@ "resolved": "https://registry.npmjs.org/vite/-/vite-6.4.1.tgz", "integrity": "sha512-+Oxm7q9hDoLMyJOYfUYBuHQo+dkAloi33apOPP56pzj+vsdJDzr+j1NISE5pyaAuKL4A3UD34qd0lx5+kfKp2g==", "license": "MIT", - "peer": true, "dependencies": { "esbuild": "^0.25.0", "fdir": "^6.4.4", @@ -563,7 +561,6 @@ "resolved": "https://registry.npmjs.org/@babel/core/-/core-7.28.5.tgz", "integrity": "sha512-e7jT4DxYvIDLk1ZHmU/m/mB19rex9sv0c2ftBtjSBv+kVM/902eh0fINUzD7UwLLNR+jU585GxUJ8/EBfAM5fw==", "license": "MIT", - "peer": true, "dependencies": { "@babel/code-frame": "^7.27.1", "@babel/generator": "^7.28.5", @@ -1509,8 +1506,7 @@ "resolved": "https://registry.npmjs.org/@cloudflare/workers-types/-/workers-types-4.20251121.0.tgz", "integrity": "sha512-jzFg7hEGKzpEalxTCanN6lM8IdkvO/brsERp/+OyMms4Zi0nhDPUAg9dUcKU8wDuDUnzbjkplY6YRwle7Cq6gA==", "devOptional": true, - "license": "MIT OR Apache-2.0", - "peer": true + "license": "MIT OR Apache-2.0" }, "node_modules/@cspotcode/source-map-support": { "version": "0.8.1", @@ -2824,7 +2820,6 @@ "integrity": "sha512-/g2d4sW9nUDJOMz3mabVQvOGhVa4e/BN/Um7yca9Bb2XTzPPnfTWHWQg+IsEYO7M3Vx+EXvaM/I2pJWIMun1bg==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@octokit/auth-token": "^4.0.0", "@octokit/graphql": "^7.1.0", @@ -4294,7 +4289,6 @@ "resolved": "https://registry.npmjs.org/@types/node/-/node-24.10.1.tgz", "integrity": "sha512-GNWcUTRBgIRJD5zj+Tq0fKOJ5XZajIiBroOF0yvj2bSU1WvNdYS/dn9UxwsujGW4JX06dnHyjV2y9rRaybH0iQ==", "license": "MIT", - "peer": true, "dependencies": { "undici-types": "~7.16.0" } @@ -4310,7 +4304,6 @@ "resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.6.tgz", "integrity": "sha512-p/jUvulfgU7oKtj6Xpk8cA2Y1xKTtICGpJYeJXz2YVO2UcvjQgeRMLDGfDeqeRW2Ta+0QNFwcc8X3GH8SxZz6w==", "license": "MIT", - "peer": true, "dependencies": { "csstype": "^3.2.2" } @@ -4320,7 +4313,6 @@ "resolved": "https://registry.npmjs.org/@types/react-dom/-/react-dom-19.2.3.tgz", "integrity": "sha512-jp2L/eY6fn+KgVVQAOqYItbF0VY/YApe5Mz2F0aykSO8gx31bYCZyvSeYxCHKvzHG5eZjc+zyaS5BrBWya2+kQ==", "license": "MIT", - "peer": true, "peerDependencies": { "@types/react": "^19.2.0" } @@ -4447,7 +4439,6 @@ "integrity": "sha512-oukfKT9Mk41LreEW09vt45f8wx7DordoWUZMYdY/cyAk7w5TWkTRCNZYF7sX7n2wB7jyGAl74OxgwhPgKaqDMQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@vitest/utils": "3.2.4", "pathe": "^2.0.3", @@ -4463,7 +4454,6 @@ "integrity": "sha512-dEYtS7qQP2CjU27QBC5oUOxLE/v5eLkGqPE0ZKEIDGMs4vKWe7IjgLOeauHsR0D5YuuycGRO5oSRXnwnmA78fQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@vitest/pretty-format": "3.2.4", "magic-string": "^0.30.17", @@ -4492,7 +4482,6 @@ "integrity": "sha512-hGISOaP18plkzbWEcP/QvtRW1xDXF2+96HbEX6byqQhAUbiS5oH6/9JwW+QsQCIYON2bI6QZBF+2PvOmrRZ9wA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@vitest/utils": "3.2.4", "fflate": "^0.8.2", @@ -4640,7 +4629,6 @@ "resolved": "https://registry.npmjs.org/ajv/-/ajv-8.17.1.tgz", "integrity": "sha512-B/gBuNg5SiMTrPkC+A2+cW0RszwxYmn6VYxB/inlBStS5nx6xHIt/ehKRhIMhqusl7a8LjQoZnjCs5vhwxOQ1g==", "license": "MIT", - "peer": true, "dependencies": { "fast-deep-equal": "^3.1.3", "fast-uri": "^3.0.1", @@ -5386,7 +5374,6 @@ "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz", "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -5671,7 +5658,6 @@ } ], "license": "MIT", - "peer": true, "dependencies": { "baseline-browser-mapping": "^2.8.25", "caniuse-lite": "^1.0.30001754", @@ -7609,7 +7595,6 @@ "integrity": "sha512-ekilCSN1jwRvIbgeg/57YFh8qQDNbwDb9xT/qu2DAHbFFZUicIl4ygVaAvzveMhMVr3LnpSKTNnwt8PoOfmKhQ==", "devOptional": true, "license": "MIT", - "peer": true, "bin": { "jiti": "lib/jiti-cli.mjs" } @@ -7874,7 +7859,6 @@ "integrity": "sha512-utfs7Pr5uJyyvDETitgsaqSyjCb2qNRAtuqUeWIAKztsOYdcACf2KtARYXg2pSvhkt+9NfoaNY7fxjl6nuMjIQ==", "devOptional": true, "license": "MPL-2.0", - "peer": true, "dependencies": { "detect-libc": "^2.0.3" }, @@ -7911,6 +7895,7 @@ "os": [ "android" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -7931,6 +7916,7 @@ "os": [ "darwin" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -7951,6 +7937,7 @@ "os": [ "darwin" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -7971,6 +7958,7 @@ "os": [ "freebsd" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -7991,6 +7979,7 @@ "os": [ "linux" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -8011,6 +8000,7 @@ "os": [ "linux" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -8031,6 +8021,7 @@ "os": [ "linux" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -8051,6 +8042,7 @@ "os": [ "linux" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -8071,6 +8063,7 @@ "os": [ "linux" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -8091,6 +8084,7 @@ "os": [ "win32" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -8111,6 +8105,7 @@ "os": [ "win32" ], + "peer": true, "engines": { "node": ">= 12.0.0" }, @@ -8160,6 +8155,7 @@ "resolved": "https://registry.npmjs.org/loose-envify/-/loose-envify-1.4.0.tgz", "integrity": "sha512-lyuxPGr/Wfhrlem2CL/UcnUc1zcqKAImBDzukY7Y5F/yQiNdko6+fRLevlw1HgMySw7f611UIY408EtxRSoK3Q==", "license": "MIT", + "peer": true, "dependencies": { "js-tokens": "^3.0.0 || ^4.0.0" }, @@ -9728,7 +9724,6 @@ "resolved": "https://registry.npmjs.org/prettier/-/prettier-3.6.2.tgz", "integrity": "sha512-I7AIg5boAr5R0FFtJ6rCfD+LFsWHp81dolrFD8S79U9tb8Az2nGrJncnMSnys+bpQJfRUzqs9hnA81OAA3hCuQ==", "license": "MIT", - "peer": true, "bin": { "prettier": "bin/prettier.cjs" }, @@ -9945,7 +9940,6 @@ "resolved": "https://registry.npmjs.org/react/-/react-19.2.0.tgz", "integrity": "sha512-tmbWg6W31tQLeB5cdIBOicJDJRR2KzXsV7uSK9iNfLWQ5bIZfxuPEHp7M8wiHyHnn0DD1i7w3Zmin0FtkrwoCQ==", "license": "MIT", - "peer": true, "engines": { "node": ">=0.10.0" } @@ -9955,7 +9949,6 @@ "resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.2.0.tgz", "integrity": "sha512-UlbRu4cAiGaIewkPyiRGJk0imDN2T3JjieT6spoL2UeSf5od4n5LB/mQ4ejmxhCFT1tYe8IvaFulzynWovsEFQ==", "license": "MIT", - "peer": true, "dependencies": { "scheduler": "^0.27.0" }, @@ -9967,7 +9960,8 @@ "version": "16.13.1", "resolved": "https://registry.npmjs.org/react-is/-/react-is-16.13.1.tgz", "integrity": "sha512-24e6ynE2H+OKt4kqsOvNd8kBpV65zoxbA4BVsEOB3ARVWQki/DHzaUoC5KuON/BiccDaCCTZBuOcfZs70kR8bQ==", - "license": "MIT" + "license": "MIT", + "peer": true }, "node_modules/react-katex": { "version": "3.1.0", @@ -10341,7 +10335,6 @@ "integrity": "sha512-ZRLgPlS91l4JztLYEZnmMcd3Umcla1hkXJgiEiR4HloRJBBoeaX8qogTu5Jfu36rRMVLndzqYv0h+M5gJAkUfg==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@oxc-project/types": "=0.98.0", "@rolldown/pluginutils": "1.0.0-beta.51" @@ -11157,7 +11150,6 @@ "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz", "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -11357,7 +11349,6 @@ "integrity": "sha512-ytQKuwgmrrkDTFP4LjR0ToE2nqgy886GpvRSpU0JAnrdBYppuY5rLkRUYPU1yCryb24SsKBTL/hlDQAEFVwtZg==", "devOptional": true, "license": "MIT", - "peer": true, "dependencies": { "esbuild": "~0.25.0", "get-tsconfig": "^4.7.5" @@ -11517,7 +11508,6 @@ "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", "license": "Apache-2.0", - "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -11587,7 +11577,6 @@ "resolved": "https://registry.npmjs.org/unenv/-/unenv-2.0.0-rc.24.tgz", "integrity": "sha512-i7qRCmY42zmCwnYlh9H2SvLEypEFGye5iRmEMKjcGi7zk9UquigRjFtTLz0TYqr0ZGLZhaMHl/foy1bZR+Cwlw==", "license": "MIT", - "peer": true, "dependencies": { "pathe": "^2.0.3" } @@ -12029,7 +12018,6 @@ "resolved": "https://registry.npmjs.org/vite/-/vite-7.2.4.tgz", "integrity": "sha512-NL8jTlbo0Tn4dUEXEsUg8KeyG/Lkmc4Fnzb8JXN/Ykm9G4HNImjtABMJgkQoVjOBN/j2WAwDTRytdqJbZsah7w==", "license": "MIT", - "peer": true, "dependencies": { "esbuild": "^0.25.0", "fdir": "^6.5.0", @@ -12144,7 +12132,6 @@ "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz", "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -12177,7 +12164,6 @@ "integrity": "sha512-LUCP5ev3GURDysTWiP47wRRUpLKMOfPh+yKTx3kVIEiu5KOMeqzpnYNsKyOoVrULivR8tLcks4+lga33Whn90A==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@types/chai": "^5.2.2", "@vitest/expect": "3.2.4", @@ -12603,7 +12589,6 @@ "integrity": "sha512-Om5ns0Lyx/LKtYI04IV0bjIrkBgoFNg0p6urzr2asekJlfP18RqFzyqMFZKf0i9Gnjtz/JfAS/Ol6tjCe5JJsQ==", "hasInstallScript": true, "license": "Apache-2.0", - "peer": true, "bin": { "workerd": "bin/workerd" }, @@ -13367,7 +13352,6 @@ "resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz", "integrity": "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==", "license": "MIT", - "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" } diff --git a/packages/sandbox-container/src/handlers/ws-handler.ts b/packages/sandbox-container/src/handlers/ws-handler.ts new file mode 100644 index 00000000..7d6c9aa0 --- /dev/null +++ b/packages/sandbox-container/src/handlers/ws-handler.ts @@ -0,0 +1,347 @@ +/** + * WebSocket Handler for Container + * + * Handles WebSocket connections and routes messages to HTTP handlers. + * This enables multiplexing multiple requests over a single WebSocket connection, + * reducing sub-request count when the SDK runs inside Workers/Durable Objects. + */ + +import type { Logger } from '@repo/shared'; +import { + isWSRequest, + type WSError, + type WSRequest, + type WSResponse, + type WSServerMessage, + type WSStreamChunk +} from '@repo/shared'; +import type { ServerWebSocket } from 'bun'; +import type { Router } from '../core/router'; + +/** + * WebSocket data attached to each connection + */ +export interface WSData { + /** Connection ID for logging */ + connectionId: string; +} + +/** + * WebSocket handler that bridges WebSocket messages to HTTP handlers + */ +export class WebSocketHandler { + private router: Router; + private logger: Logger; + + constructor(router: Router, logger: Logger) { + this.router = router; + this.logger = logger.child({ component: 'ws-handler' }); + } + + /** + * Handle WebSocket connection open + */ + onOpen(ws: ServerWebSocket): void { + this.logger.debug('WebSocket connection opened', { + connectionId: ws.data.connectionId + }); + } + + /** + * Handle WebSocket connection close + */ + onClose(ws: ServerWebSocket, code: number, reason: string): void { + this.logger.debug('WebSocket connection closed', { + connectionId: ws.data.connectionId, + code, + reason + }); + } + + /** + * Handle incoming WebSocket message + */ + async onMessage( + ws: ServerWebSocket, + message: string | Buffer + ): Promise { + const messageStr = + typeof message === 'string' ? message : message.toString('utf-8'); + + let parsed: unknown; + try { + parsed = JSON.parse(messageStr); + } catch (error) { + this.sendError(ws, undefined, 'PARSE_ERROR', 'Invalid JSON message', 400); + return; + } + + if (!isWSRequest(parsed)) { + this.sendError( + ws, + undefined, + 'INVALID_REQUEST', + 'Message must be a valid WSRequest', + 400 + ); + return; + } + + const request = parsed as WSRequest; + + this.logger.debug('WebSocket request received', { + connectionId: ws.data.connectionId, + id: request.id, + method: request.method, + path: request.path + }); + + try { + await this.handleRequest(ws, request); + } catch (error) { + this.logger.error( + 'Error handling WebSocket request', + error instanceof Error ? error : new Error(String(error)), + { requestId: request.id } + ); + this.sendError( + ws, + request.id, + 'INTERNAL_ERROR', + error instanceof Error ? error.message : 'Unknown error', + 500 + ); + } + } + + /** + * Handle a WebSocket request by routing it to HTTP handlers + */ + private async handleRequest( + ws: ServerWebSocket, + request: WSRequest + ): Promise { + // Build URL for the request + const url = `http://localhost:3000${request.path}`; + + // Build headers + const headers: Record = { + 'Content-Type': 'application/json', + ...request.headers + }; + + // Build request options + const requestInit: RequestInit = { + method: request.method, + headers + }; + + // Add body for POST/PUT + if ( + request.body !== undefined && + (request.method === 'POST' || request.method === 'PUT') + ) { + requestInit.body = JSON.stringify(request.body); + } + + // Create a fetch Request object + const httpRequest = new Request(url, requestInit); + + // Route through the existing router + const httpResponse = await this.router.route(httpRequest); + + // Check if this is a streaming response + const contentType = httpResponse.headers.get('Content-Type') || ''; + const isStreaming = contentType.includes('text/event-stream'); + + if (isStreaming && httpResponse.body) { + // Handle SSE streaming response + await this.handleStreamingResponse(ws, request.id, httpResponse); + } else { + // Handle regular response + await this.handleRegularResponse(ws, request.id, httpResponse); + } + } + + /** + * Handle a regular (non-streaming) HTTP response + */ + private async handleRegularResponse( + ws: ServerWebSocket, + requestId: string, + response: Response + ): Promise { + let body: unknown; + + try { + const text = await response.text(); + body = text ? JSON.parse(text) : undefined; + } catch { + body = undefined; + } + + const wsResponse: WSResponse = { + type: 'response', + id: requestId, + status: response.status, + body, + done: true + }; + + this.send(ws, wsResponse); + } + + /** + * Handle a streaming (SSE) HTTP response + */ + private async handleStreamingResponse( + ws: ServerWebSocket, + requestId: string, + response: Response + ): Promise { + if (!response.body) { + this.sendError(ws, requestId, 'STREAM_ERROR', 'No response body', 500); + return; + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + try { + while (true) { + const { done, value } = await reader.read(); + + if (done) { + break; + } + + // Decode chunk and add to buffer + buffer += decoder.decode(value, { stream: true }); + + // Parse SSE events from buffer + const events = this.parseSSEEvents(buffer); + buffer = events.remaining; + + // Send each parsed event as a stream chunk + for (const event of events.events) { + const chunk: WSStreamChunk = { + type: 'stream', + id: requestId, + event: event.event, + data: event.data + }; + this.send(ws, chunk); + } + } + + // Send final response to close the stream + const wsResponse: WSResponse = { + type: 'response', + id: requestId, + status: response.status, + done: true + }; + this.send(ws, wsResponse); + } catch (error) { + this.logger.error( + 'Error reading stream', + error instanceof Error ? error : new Error(String(error)), + { requestId } + ); + this.sendError( + ws, + requestId, + 'STREAM_ERROR', + error instanceof Error ? error.message : 'Stream read failed', + 500 + ); + } finally { + reader.releaseLock(); + } + } + + /** + * Parse SSE events from a buffer + */ + private parseSSEEvents(buffer: string): { + events: Array<{ event?: string; data: string }>; + remaining: string; + } { + const events: Array<{ event?: string; data: string }> = []; + const lines = buffer.split('\n'); + let currentEvent: { event?: string; data: string[] } = { data: [] }; + let processedIndex = 0; + + for (let i = 0; i < lines.length; i++) { + const line = lines[i]; + + // Check if we have a complete event (empty line after data) + if (line === '' && currentEvent.data.length > 0) { + events.push({ + event: currentEvent.event, + data: currentEvent.data.join('\n') + }); + currentEvent = { data: [] }; + processedIndex = buffer.indexOf(line, processedIndex) + line.length + 1; + continue; + } + + if (line.startsWith('event:')) { + currentEvent.event = line.substring(6).trim(); + processedIndex = buffer.indexOf(line, processedIndex) + line.length + 1; + } else if (line.startsWith('data:')) { + currentEvent.data.push(line.substring(5).trim()); + processedIndex = buffer.indexOf(line, processedIndex) + line.length + 1; + } else if (line === '') { + processedIndex = buffer.indexOf(line, processedIndex) + line.length + 1; + } + } + + return { + events, + remaining: buffer.substring(processedIndex) + }; + } + + /** + * Send a message over WebSocket + */ + private send(ws: ServerWebSocket, message: WSServerMessage): void { + try { + ws.send(JSON.stringify(message)); + } catch (error) { + this.logger.error( + 'Failed to send WebSocket message', + error instanceof Error ? error : new Error(String(error)) + ); + } + } + + /** + * Send an error message over WebSocket + */ + private sendError( + ws: ServerWebSocket, + requestId: string | undefined, + code: string, + message: string, + status: number + ): void { + const error: WSError = { + type: 'error', + id: requestId, + code, + message, + status + }; + this.send(ws, error); + } +} + +/** + * Generate a unique connection ID + */ +export function generateConnectionId(): string { + return `conn_${Date.now()}_${Math.random().toString(36).substring(2, 8)}`; +} diff --git a/packages/sandbox-container/src/index.ts b/packages/sandbox-container/src/index.ts index f5d8dd49..aa266c1a 100644 --- a/packages/sandbox-container/src/index.ts +++ b/packages/sandbox-container/src/index.ts @@ -1,14 +1,24 @@ import { createLogger } from '@repo/shared'; +import type { ServerWebSocket } from 'bun'; import { serve } from 'bun'; import { Container } from './core/container'; import { Router } from './core/router'; +import { + generateConnectionId, + WebSocketHandler, + type WSData +} from './handlers/ws-handler'; import { setupRoutes } from './routes/setup'; // Create module-level logger for server lifecycle events const logger = createLogger({ component: 'container' }); +// WebSocket handler (initialized after router is ready) +let wsHandler: WebSocketHandler | null = null; + async function createApplication(): Promise<{ - fetch: (req: Request) => Promise; + fetch: (req: Request, server: ReturnType) => Promise; + router: Router; }> { // Initialize dependency injection container const container = new Container(); @@ -24,23 +34,61 @@ async function createApplication(): Promise<{ setupRoutes(router, container); return { - fetch: (req: Request) => router.route(req) + fetch: async (req: Request, server: ReturnType) => { + // Check for WebSocket upgrade request + const upgradeHeader = req.headers.get('Upgrade'); + if (upgradeHeader?.toLowerCase() === 'websocket') { + // Handle WebSocket upgrade for control plane + const url = new URL(req.url); + if (url.pathname === '/ws' || url.pathname === '/api/ws') { + const upgraded = server.upgrade(req, { + data: { + connectionId: generateConnectionId() + } as WSData + }); + if (upgraded) { + return undefined as unknown as Response; // Bun handles the upgrade + } + return new Response('WebSocket upgrade failed', { status: 500 }); + } + } + + // Regular HTTP request + return router.route(req); + }, + router }; } // Initialize the application const app = await createApplication(); +// Initialize WebSocket handler with the router +wsHandler = new WebSocketHandler(app.router, logger); + // Start the Bun server const server = serve({ idleTimeout: 255, - fetch: app.fetch, + fetch: (req, server) => app.fetch(req, server), hostname: '0.0.0.0', port: 3000, - // Enhanced WebSocket placeholder for future streaming features + // WebSocket handlers for control plane multiplexing websocket: { - async message() { - // WebSocket functionality can be added here in the future + open(ws) { + wsHandler?.onOpen(ws as unknown as ServerWebSocket); + }, + close(ws, code: number, reason: string) { + wsHandler?.onClose( + ws as unknown as ServerWebSocket, + code, + reason + ); + }, + async message(ws, message: string | Buffer) { + await wsHandler?.onMessage( + ws as unknown as ServerWebSocket, + message + ); } } }); diff --git a/packages/sandbox-container/tests/handlers/ws-handler.test.ts b/packages/sandbox-container/tests/handlers/ws-handler.test.ts new file mode 100644 index 00000000..8028f55a --- /dev/null +++ b/packages/sandbox-container/tests/handlers/ws-handler.test.ts @@ -0,0 +1,382 @@ +import type { Logger, WSError, WSRequest, WSResponse } from '@repo/shared'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { Router } from '../../src/core/router'; +import { + generateConnectionId, + WebSocketHandler, + type WSData +} from '../../src/handlers/ws-handler'; + +// Mock ServerWebSocket +class MockServerWebSocket { + data: WSData; + sentMessages: string[] = []; + + constructor(data: WSData) { + this.data = data; + } + + send(message: string) { + this.sentMessages.push(message); + } + + getSentMessages(): T[] { + return this.sentMessages.map((m) => JSON.parse(m)); + } + + getLastMessage(): T { + return JSON.parse(this.sentMessages[this.sentMessages.length - 1]); + } +} + +// Mock Router +function createMockRouter(): Router { + return { + route: vi.fn() + } as unknown as Router; +} + +// Mock Logger +function createMockLogger(): Logger { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + child: vi.fn(() => createMockLogger()) + } as unknown as Logger; +} + +describe('WebSocketHandler', () => { + let handler: WebSocketHandler; + let mockRouter: Router; + let mockLogger: Logger; + let mockWs: MockServerWebSocket; + + beforeEach(() => { + vi.clearAllMocks(); + mockRouter = createMockRouter(); + mockLogger = createMockLogger(); + handler = new WebSocketHandler(mockRouter, mockLogger); + mockWs = new MockServerWebSocket({ connectionId: 'test-conn-123' }); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + describe('onOpen', () => { + it('should log connection open', () => { + handler.onOpen(mockWs as any); + + expect(mockLogger.child).toHaveBeenCalled(); + }); + }); + + describe('onClose', () => { + it('should log connection close with code and reason', () => { + handler.onClose(mockWs as any, 1000, 'Normal closure'); + + expect(mockLogger.child).toHaveBeenCalled(); + }); + }); + + describe('onMessage', () => { + it('should handle valid request and return response', async () => { + const request: WSRequest = { + type: 'request', + id: 'req-123', + method: 'GET', + path: '/api/health' + }; + + // Mock router to return a successful response + (mockRouter.route as any).mockResolvedValue( + new Response(JSON.stringify({ status: 'ok' }), { + status: 200, + headers: { 'Content-Type': 'application/json' } + }) + ); + + await handler.onMessage(mockWs as any, JSON.stringify(request)); + + expect(mockRouter.route).toHaveBeenCalled(); + + const response = mockWs.getLastMessage(); + expect(response.type).toBe('response'); + expect(response.id).toBe('req-123'); + expect(response.status).toBe(200); + expect(response.body).toEqual({ status: 'ok' }); + expect(response.done).toBe(true); + }); + + it('should handle POST request with body', async () => { + const request: WSRequest = { + type: 'request', + id: 'req-456', + method: 'POST', + path: '/api/execute', + body: { command: 'echo hello', sessionId: 'sess-1' } + }; + + (mockRouter.route as any).mockResolvedValue( + new Response( + JSON.stringify({ + success: true, + stdout: 'hello\n', + exitCode: 0 + }), + { status: 200 } + ) + ); + + await handler.onMessage(mockWs as any, JSON.stringify(request)); + + // Verify router was called with correct Request + const routerCall = (mockRouter.route as any).mock.calls[0][0] as Request; + expect(routerCall.method).toBe('POST'); + expect(routerCall.url).toContain('/api/execute'); + + const body = (await routerCall.clone().json()) as { command: string }; + expect(body.command).toBe('echo hello'); + }); + + it('should return error for invalid JSON', async () => { + await handler.onMessage(mockWs as any, 'not valid json'); + + const response = mockWs.getLastMessage(); + expect(response.type).toBe('error'); + expect(response.code).toBe('PARSE_ERROR'); + expect(response.status).toBe(400); + }); + + it('should return error for invalid request format', async () => { + await handler.onMessage( + mockWs as any, + JSON.stringify({ notARequest: true }) + ); + + const response = mockWs.getLastMessage(); + expect(response.type).toBe('error'); + expect(response.code).toBe('INVALID_REQUEST'); + expect(response.status).toBe(400); + }); + + it('should handle router errors gracefully', async () => { + const request: WSRequest = { + type: 'request', + id: 'req-err', + method: 'GET', + path: '/api/fail' + }; + + (mockRouter.route as any).mockRejectedValue(new Error('Router failed')); + + await handler.onMessage(mockWs as any, JSON.stringify(request)); + + const response = mockWs.getLastMessage(); + expect(response.type).toBe('error'); + expect(response.id).toBe('req-err'); + expect(response.code).toBe('INTERNAL_ERROR'); + expect(response.message).toContain('Router failed'); + expect(response.status).toBe(500); + }); + + it('should handle 404 responses', async () => { + const request: WSRequest = { + type: 'request', + id: 'req-404', + method: 'GET', + path: '/api/notfound' + }; + + (mockRouter.route as any).mockResolvedValue( + new Response( + JSON.stringify({ + code: 'NOT_FOUND', + message: 'Resource not found' + }), + { status: 404 } + ) + ); + + await handler.onMessage(mockWs as any, JSON.stringify(request)); + + const response = mockWs.getLastMessage(); + expect(response.type).toBe('response'); + expect(response.id).toBe('req-404'); + expect(response.status).toBe(404); + }); + + it('should handle streaming responses', async () => { + const request: WSRequest = { + type: 'request', + id: 'req-stream', + method: 'POST', + path: '/api/execute/stream', + body: { command: 'echo test' } + }; + + // Create a mock SSE stream + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode('event: start\ndata: {"type":"start"}\n\n') + ); + controller.enqueue( + encoder.encode('data: {"type":"stdout","text":"test\\n"}\n\n') + ); + controller.enqueue( + encoder.encode( + 'event: complete\ndata: {"type":"complete","exitCode":0}\n\n' + ) + ); + controller.close(); + } + }); + + (mockRouter.route as any).mockResolvedValue( + new Response(stream, { + status: 200, + headers: { 'Content-Type': 'text/event-stream' } + }) + ); + + await handler.onMessage(mockWs as any, JSON.stringify(request)); + + // Should have received stream chunks and final response + const messages = mockWs.getSentMessages(); + + // Find stream chunks + const streamChunks = messages.filter((m) => m.type === 'stream'); + expect(streamChunks.length).toBeGreaterThan(0); + + // Find final response + const finalResponse = messages.find((m) => m.type === 'response'); + expect(finalResponse).toBeDefined(); + expect(finalResponse.done).toBe(true); + }); + + it('should handle Buffer messages', async () => { + const request: WSRequest = { + type: 'request', + id: 'req-buffer', + method: 'GET', + path: '/api/test' + }; + + (mockRouter.route as any).mockResolvedValue( + new Response(JSON.stringify({ ok: true }), { status: 200 }) + ); + + // Send as Buffer + const buffer = Buffer.from(JSON.stringify(request)); + await handler.onMessage(mockWs as any, buffer); + + expect(mockRouter.route).toHaveBeenCalled(); + }); + }); + + describe('generateConnectionId', () => { + it('should generate unique connection IDs', () => { + const id1 = generateConnectionId(); + const id2 = generateConnectionId(); + + expect(id1).toMatch(/^conn_\d+_[a-z0-9]+$/); + expect(id2).toMatch(/^conn_\d+_[a-z0-9]+$/); + expect(id1).not.toBe(id2); + }); + }); +}); + +describe('WebSocket Integration', () => { + let handler: WebSocketHandler; + let mockRouter: Router; + let mockLogger: Logger; + + beforeEach(() => { + mockRouter = createMockRouter(); + mockLogger = createMockLogger(); + handler = new WebSocketHandler(mockRouter, mockLogger); + }); + + it('should handle multiple concurrent requests', async () => { + const mockWs = new MockServerWebSocket({ connectionId: 'concurrent-test' }); + + const requests: WSRequest[] = [ + { type: 'request', id: 'req-1', method: 'GET', path: '/api/one' }, + { type: 'request', id: 'req-2', method: 'GET', path: '/api/two' }, + { type: 'request', id: 'req-3', method: 'GET', path: '/api/three' } + ]; + + // Router returns different responses based on path + (mockRouter.route as any).mockImplementation((req: Request) => { + const path = new URL(req.url).pathname; + return new Response(JSON.stringify({ path }), { status: 200 }); + }); + + // Process all requests concurrently + await Promise.all( + requests.map((req) => + handler.onMessage(mockWs as any, JSON.stringify(req)) + ) + ); + + const responses = mockWs.getSentMessages(); + expect(responses).toHaveLength(3); + + // Verify each request got its correct response + const responseIds = responses.map((r) => r.id).sort(); + expect(responseIds).toEqual(['req-1', 'req-2', 'req-3']); + + // Verify response bodies match request paths + responses.forEach((r) => { + expect(r.body).toBeDefined(); + }); + }); + + it('should maintain request isolation', async () => { + const mockWs = new MockServerWebSocket({ connectionId: 'isolation-test' }); + + // First request fails + const failRequest: WSRequest = { + type: 'request', + id: 'fail-req', + method: 'GET', + path: '/api/fail' + }; + + // Second request succeeds + const successRequest: WSRequest = { + type: 'request', + id: 'success-req', + method: 'GET', + path: '/api/success' + }; + + (mockRouter.route as any).mockImplementation((req: Request) => { + const path = new URL(req.url).pathname; + if (path === '/api/fail') { + throw new Error('Intentional failure'); + } + return new Response(JSON.stringify({ ok: true }), { status: 200 }); + }); + + // Process both requests + await handler.onMessage(mockWs as any, JSON.stringify(failRequest)); + await handler.onMessage(mockWs as any, JSON.stringify(successRequest)); + + const messages = mockWs.getSentMessages(); + expect(messages).toHaveLength(2); + + // First should be error + const errorMsg = messages.find((m) => m.id === 'fail-req'); + expect(errorMsg.type).toBe('error'); + + // Second should succeed + const successMsg = messages.find((m) => m.id === 'success-req'); + expect(successMsg.type).toBe('response'); + expect(successMsg.status).toBe(200); + }); +}); diff --git a/packages/sandbox/src/clients/base-client.ts b/packages/sandbox/src/clients/base-client.ts index fcd64634..15f7175c 100644 --- a/packages/sandbox/src/clients/base-client.ts +++ b/packages/sandbox/src/clients/base-client.ts @@ -4,6 +4,7 @@ import { getHttpStatus } from '@repo/shared/errors'; import type { ErrorResponse as NewErrorResponse } from '../errors'; import { createErrorFromResponse, ErrorCode } from '../errors'; import type { SandboxError } from '../errors/classes'; +import { createTransport, type Transport } from './transport'; import type { HttpClientOptions, ResponseHandler } from './types'; // Container startup retry configuration @@ -11,26 +12,107 @@ const TIMEOUT_MS = 120_000; // 2 minutes total retry budget const MIN_TIME_FOR_RETRY_MS = 15_000; // Need at least 15s remaining to retry (allows for longer container startups) /** - * Abstract base class providing common HTTP functionality for all domain clients + * Abstract base class providing common HTTP/WebSocket functionality for all domain clients + * + * Supports two transport modes: + * - HTTP (default): Each request is a separate HTTP call + * - WebSocket: All requests multiplexed over a single connection + * + * WebSocket mode is useful when running inside Workers/Durable Objects + * where sub-request limits apply. */ export abstract class BaseHttpClient { protected baseUrl: string; protected options: HttpClientOptions; protected logger: Logger; + protected transport: Transport | null = null; constructor(options: HttpClientOptions = {}) { this.options = options; this.logger = options.logger ?? createNoOpLogger(); this.baseUrl = this.options.baseUrl!; + + // Use provided transport or create one if WebSocket mode is enabled + if (options.transport) { + this.transport = options.transport; + } else if (options.transportMode === 'websocket' && options.wsUrl) { + this.transport = createTransport({ + mode: 'websocket', + wsUrl: options.wsUrl, + logger: this.logger + }); + } + } + + /** + * Check if using WebSocket transport + */ + protected isWebSocketMode(): boolean { + return this.transport?.getMode() === 'websocket'; } /** * Core HTTP request method with automatic retry for container startup delays * Retries both 503 (provisioning) and 500 (startup failure) errors when they're container-related + * + * When WebSocket transport is enabled, this creates a Response-like object + * from the WebSocket response for compatibility with existing code. */ protected async doFetch( path: string, options?: RequestInit + ): Promise { + // Use WebSocket transport if available + if (this.transport?.getMode() === 'websocket') { + return this.doWebSocketFetch(path, options); + } + + // Fall back to HTTP transport + return this.doHttpFetch(path, options); + } + + /** + * WebSocket-based fetch implementation + * Converts WebSocket request/response to Response object for compatibility + */ + private async doWebSocketFetch( + path: string, + options?: RequestInit + ): Promise { + if (!this.transport) { + throw new Error('WebSocket transport not initialized'); + } + + const method = (options?.method || 'GET') as + | 'GET' + | 'POST' + | 'PUT' + | 'DELETE'; + let body: unknown; + + if (options?.body && typeof options.body === 'string') { + try { + body = JSON.parse(options.body); + } catch { + body = options.body; + } + } + + const result = await this.transport.request(method, path, body); + + // Create a Response-like object for compatibility + return new Response(JSON.stringify(result.body), { + status: result.status, + headers: { 'Content-Type': 'application/json' } + }); + } + + /** + * HTTP-based fetch implementation with retry logic + */ + private async doHttpFetch( + path: string, + options?: RequestInit ): Promise { const startTime = Date.now(); let attempt = 0; @@ -201,6 +283,29 @@ export abstract class BaseHttpClient { return response.body; } + /** + * Stream request handler for WebSocket transport + * Returns a ReadableStream that receives data over WebSocket + */ + protected async doStreamFetch( + path: string, + body?: unknown + ): Promise> { + // Use WebSocket transport if available + if (this.transport?.getMode() === 'websocket') { + return this.transport.requestStream('POST', path, body); + } + + // Fall back to HTTP streaming + const response = await this.doFetch(path, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: body ? JSON.stringify(body) : undefined + }); + + return this.handleStreamResponse(response); + } + /** * Utility method to log successful operations */ diff --git a/packages/sandbox/src/clients/index.ts b/packages/sandbox/src/clients/index.ts index 840ef1ce..1201689a 100644 --- a/packages/sandbox/src/clients/index.ts +++ b/packages/sandbox/src/clients/index.ts @@ -41,6 +41,13 @@ export type { } from './process-client'; export { ProcessClient } from './process-client'; export { SandboxClient } from './sandbox-client'; +// Transport layer +export type { + TransportMode, + TransportOptions, + TransportResponse +} from './transport'; +export { createTransport, Transport } from './transport'; // Types and interfaces export type { BaseApiResponse, @@ -62,3 +69,4 @@ export type { VersionResponse } from './utility-client'; export { UtilityClient } from './utility-client'; +export { WSTransport } from './ws-transport'; diff --git a/packages/sandbox/src/clients/sandbox-client.ts b/packages/sandbox/src/clients/sandbox-client.ts index fa1599eb..2023f98c 100644 --- a/packages/sandbox/src/clients/sandbox-client.ts +++ b/packages/sandbox/src/clients/sandbox-client.ts @@ -4,12 +4,23 @@ import { GitClient } from './git-client'; import { InterpreterClient } from './interpreter-client'; import { PortClient } from './port-client'; import { ProcessClient } from './process-client'; +import { + createTransport, + type Transport, + type TransportMode +} from './transport'; import type { HttpClientOptions } from './types'; import { UtilityClient } from './utility-client'; /** * Main sandbox client that composes all domain-specific clients * Provides organized access to all sandbox functionality + * + * Supports two transport modes: + * - HTTP (default): Each request is a separate HTTP call + * - WebSocket: All requests multiplexed over a single connection + * + * WebSocket mode reduces sub-request count when running inside Workers/Durable Objects. */ export class SandboxClient { public readonly commands: CommandClient; @@ -20,11 +31,27 @@ export class SandboxClient { public readonly interpreter: InterpreterClient; public readonly utils: UtilityClient; + private transport: Transport | null = null; + constructor(options: HttpClientOptions) { + // Create shared transport if WebSocket mode is enabled + if (options.transportMode === 'websocket' && options.wsUrl) { + this.transport = createTransport({ + mode: 'websocket', + wsUrl: options.wsUrl, + baseUrl: options.baseUrl, + logger: options.logger, + stub: options.stub, + port: options.port + }); + } + // Ensure baseUrl is provided for all clients const clientOptions: HttpClientOptions = { baseUrl: 'http://localhost:3000', - ...options + ...options, + // Share transport across all clients + transport: this.transport ?? options.transport }; // Initialize all domain clients with shared options @@ -36,4 +63,39 @@ export class SandboxClient { this.interpreter = new InterpreterClient(clientOptions); this.utils = new UtilityClient(clientOptions); } + + /** + * Get the current transport mode + */ + getTransportMode(): TransportMode { + return this.transport?.getMode() ?? 'http'; + } + + /** + * Check if WebSocket is connected (only relevant in WebSocket mode) + */ + isWebSocketConnected(): boolean { + return this.transport?.isWebSocketConnected() ?? false; + } + + /** + * Connect WebSocket transport (no-op in HTTP mode) + * Called automatically on first request, but can be called explicitly + * to establish connection upfront. + */ + async connect(): Promise { + if (this.transport) { + await this.transport.connect(); + } + } + + /** + * Disconnect WebSocket transport (no-op in HTTP mode) + * Should be called when the sandbox is destroyed. + */ + disconnect(): void { + if (this.transport) { + this.transport.disconnect(); + } + } } diff --git a/packages/sandbox/src/clients/transport.ts b/packages/sandbox/src/clients/transport.ts new file mode 100644 index 00000000..92ecaf86 --- /dev/null +++ b/packages/sandbox/src/clients/transport.ts @@ -0,0 +1,240 @@ +import type { Logger } from '@repo/shared'; +import { createNoOpLogger } from '@repo/shared'; +import type { ContainerStub } from './types'; +import { WSTransport } from './ws-transport'; + +/** + * Transport mode for SDK communication + */ +export type TransportMode = 'http' | 'websocket'; + +/** + * Transport configuration options + */ +export interface TransportOptions { + /** Transport mode */ + mode: TransportMode; + + /** Base URL for HTTP mode */ + baseUrl?: string; + + /** WebSocket URL for WebSocket mode */ + wsUrl?: string; + + /** Logger instance */ + logger?: Logger; + + /** Container stub for DO-internal requests */ + stub?: ContainerStub; + + /** Port number */ + port?: number; + + /** Request timeout in milliseconds */ + requestTimeoutMs?: number; +} + +/** + * HTTP response-like structure + */ +export interface TransportResponse { + status: number; + ok: boolean; + body: unknown; + stream?: ReadableStream; +} + +/** + * Transport abstraction layer + * + * Provides a unified interface for HTTP and WebSocket transports. + * The SandboxClient uses this to communicate with the container. + */ +export class Transport { + private mode: TransportMode; + private baseUrl: string; + private wsTransport: WSTransport | null = null; + private logger: Logger; + private stub?: ContainerStub; + private port?: number; + + constructor(options: TransportOptions) { + this.mode = options.mode; + this.baseUrl = options.baseUrl ?? 'http://localhost:3000'; + this.logger = options.logger ?? createNoOpLogger(); + this.stub = options.stub; + this.port = options.port; + + if (this.mode === 'websocket' && options.wsUrl) { + this.wsTransport = new WSTransport(options.wsUrl, { + logger: this.logger, + requestTimeoutMs: options.requestTimeoutMs + }); + } + } + + /** + * Get the current transport mode + */ + getMode(): TransportMode { + return this.mode; + } + + /** + * Check if WebSocket is connected + */ + isWebSocketConnected(): boolean { + return this.wsTransport?.isConnected() ?? false; + } + + /** + * Connect WebSocket (no-op for HTTP mode) + */ + async connect(): Promise { + if (this.mode === 'websocket' && this.wsTransport) { + await this.wsTransport.connect(); + } + } + + /** + * Disconnect WebSocket (no-op for HTTP mode) + */ + disconnect(): void { + if (this.wsTransport) { + this.wsTransport.disconnect(); + } + } + + /** + * Make a request using the configured transport + */ + async request( + method: 'GET' | 'POST' | 'PUT' | 'DELETE', + path: string, + body?: unknown + ): Promise { + if (this.mode === 'websocket' && this.wsTransport) { + return this.wsRequest(method, path, body); + } + return this.httpRequest(method, path, body); + } + + /** + * Make a streaming request using the configured transport + */ + async requestStream( + method: 'POST', + path: string, + body?: unknown + ): Promise> { + if (this.mode === 'websocket' && this.wsTransport) { + return this.wsTransport.requestStream(method, path, body); + } + return this.httpRequestStream(method, path, body); + } + + /** + * Make an HTTP request + */ + private async httpRequest( + method: 'GET' | 'POST' | 'PUT' | 'DELETE', + path: string, + body?: unknown + ): Promise { + const url = this.stub + ? `http://localhost:${this.port}${path}` + : `${this.baseUrl}${path}`; + + const options: RequestInit = { + method, + headers: body ? { 'Content-Type': 'application/json' } : undefined, + body: body ? JSON.stringify(body) : undefined + }; + + let response: Response; + if (this.stub) { + response = await this.stub.containerFetch(url, options, this.port); + } else { + response = await fetch(url, options); + } + + // Parse JSON body if possible + let responseBody: unknown; + try { + responseBody = await response.json(); + } catch { + responseBody = undefined; + } + + return { + status: response.status, + ok: response.ok, + body: responseBody + }; + } + + /** + * Make an HTTP streaming request + */ + private async httpRequestStream( + method: 'POST', + path: string, + body?: unknown + ): Promise> { + const url = this.stub + ? `http://localhost:${this.port}${path}` + : `${this.baseUrl}${path}`; + + const options: RequestInit = { + method, + headers: { 'Content-Type': 'application/json' }, + body: body ? JSON.stringify(body) : undefined + }; + + let response: Response; + if (this.stub) { + response = await this.stub.containerFetch(url, options, this.port); + } else { + response = await fetch(url, options); + } + + if (!response.ok) { + const errorBody = await response.text(); + throw new Error(`HTTP error! status: ${response.status} - ${errorBody}`); + } + + if (!response.body) { + throw new Error('No response body for streaming'); + } + + return response.body; + } + + /** + * Make a WebSocket request + */ + private async wsRequest( + method: 'GET' | 'POST' | 'PUT' | 'DELETE', + path: string, + body?: unknown + ): Promise { + if (!this.wsTransport) { + throw new Error('WebSocket transport not initialized'); + } + + const result = await this.wsTransport.request(method, path, body); + + return { + status: result.status, + ok: result.status >= 200 && result.status < 300, + body: result.body + }; + } +} + +/** + * Create a transport instance based on options + */ +export function createTransport(options: TransportOptions): Transport { + return new Transport(options); +} diff --git a/packages/sandbox/src/clients/types.ts b/packages/sandbox/src/clients/types.ts index b59d8af3..7c0545a7 100644 --- a/packages/sandbox/src/clients/types.ts +++ b/packages/sandbox/src/clients/types.ts @@ -1,4 +1,5 @@ import type { Logger } from '@repo/shared'; +import type { Transport, TransportMode } from './transport'; /** * Minimal interface for container fetch functionality @@ -27,6 +28,25 @@ export interface HttpClientOptions { command: string ) => void; onError?: (error: string, command?: string) => void; + + /** + * Transport mode: 'http' (default) or 'websocket' + * WebSocket mode multiplexes all requests over a single connection, + * reducing sub-request count in Workers/Durable Objects. + */ + transportMode?: TransportMode; + + /** + * WebSocket URL for WebSocket transport mode. + * Required when transportMode is 'websocket'. + */ + wsUrl?: string; + + /** + * Shared transport instance (for internal use). + * When provided, clients will use this transport instead of creating their own. + */ + transport?: Transport; } /** diff --git a/packages/sandbox/src/clients/ws-transport.ts b/packages/sandbox/src/clients/ws-transport.ts new file mode 100644 index 00000000..2d1a5e91 --- /dev/null +++ b/packages/sandbox/src/clients/ws-transport.ts @@ -0,0 +1,419 @@ +import type { Logger } from '@repo/shared'; +import { + createNoOpLogger, + generateRequestId, + isWSError, + isWSResponse, + isWSStreamChunk, + type WSMethod, + type WSRequest, + type WSResponse, + type WSServerMessage, + type WSStreamChunk +} from '@repo/shared'; + +/** + * Pending request tracker for response matching + */ +interface PendingRequest { + resolve: (response: WSResponse) => void; + reject: (error: Error) => void; + streamController?: ReadableStreamDefaultController; + isStreaming: boolean; +} + +/** + * WebSocket transport configuration + */ +export interface WSTransportOptions { + /** Logger instance */ + logger?: Logger; + + /** Connection timeout in milliseconds */ + connectTimeoutMs?: number; + + /** Request timeout in milliseconds */ + requestTimeoutMs?: number; +} + +/** + * WebSocket transport state + */ +type WSTransportState = 'disconnected' | 'connecting' | 'connected' | 'error'; + +/** + * WebSocket transport layer for multiplexing HTTP-like requests + * + * Maintains a single WebSocket connection and multiplexes requests using + * unique IDs. Supports both request/response and streaming patterns. + */ +export class WSTransport { + private ws: WebSocket | null = null; + private state: WSTransportState = 'disconnected'; + private pendingRequests: Map = new Map(); + private connectPromise: Promise | null = null; + private logger: Logger; + private options: WSTransportOptions; + private url: string; + + // Bound event handlers for proper add/remove + private boundHandleMessage: (event: MessageEvent) => void; + private boundHandleClose: (event: CloseEvent) => void; + + constructor(url: string, options: WSTransportOptions = {}) { + this.url = url; + this.options = options; + this.logger = options.logger ?? createNoOpLogger(); + + // Bind handlers once in constructor + this.boundHandleMessage = this.handleMessage.bind(this); + this.boundHandleClose = this.handleClose.bind(this); + } + + /** + * Check if WebSocket is connected + */ + isConnected(): boolean { + return this.state === 'connected' && this.ws?.readyState === WebSocket.OPEN; + } + + /** + * Connect to the WebSocket server + */ + async connect(): Promise { + // Already connected + if (this.isConnected()) { + return; + } + + // Connection in progress + if (this.connectPromise) { + return this.connectPromise; + } + + this.state = 'connecting'; + + this.connectPromise = new Promise((resolve, reject) => { + const timeoutMs = this.options.connectTimeoutMs ?? 30000; + const timeout = setTimeout(() => { + this.cleanup(); + reject(new Error(`WebSocket connection timeout after ${timeoutMs}ms`)); + }, timeoutMs); + + try { + this.ws = new WebSocket(this.url); + + // One-time open handler for connection + const onOpen = () => { + clearTimeout(timeout); + this.ws?.removeEventListener('open', onOpen); + this.ws?.removeEventListener('error', onConnectError); + this.state = 'connected'; + this.logger.debug('WebSocket connected', { url: this.url }); + resolve(); + }; + + // One-time error handler for connection + const onConnectError = () => { + clearTimeout(timeout); + this.ws?.removeEventListener('open', onOpen); + this.ws?.removeEventListener('error', onConnectError); + this.state = 'error'; + this.logger.error( + 'WebSocket error', + new Error('WebSocket connection failed') + ); + reject(new Error('WebSocket connection failed')); + }; + + this.ws.addEventListener('open', onOpen); + this.ws.addEventListener('error', onConnectError); + this.ws.addEventListener('close', this.boundHandleClose); + this.ws.addEventListener('message', this.boundHandleMessage); + } catch (error) { + clearTimeout(timeout); + this.state = 'error'; + reject(error); + } + }); + + try { + await this.connectPromise; + } finally { + this.connectPromise = null; + } + } + + /** + * Disconnect from the WebSocket server + */ + disconnect(): void { + this.cleanup(); + } + + /** + * Send a request and wait for response + */ + async request( + method: WSMethod, + path: string, + body?: unknown + ): Promise<{ status: number; body: T }> { + await this.connect(); + + const id = generateRequestId(); + const request: WSRequest = { + type: 'request', + id, + method, + path, + body + }; + + return new Promise((resolve, reject) => { + const timeoutMs = this.options.requestTimeoutMs ?? 120000; + const timeout = setTimeout(() => { + this.pendingRequests.delete(id); + reject( + new Error(`Request timeout after ${timeoutMs}ms: ${method} ${path}`) + ); + }, timeoutMs); + + this.pendingRequests.set(id, { + resolve: (response: WSResponse) => { + clearTimeout(timeout); + this.pendingRequests.delete(id); + resolve({ status: response.status, body: response.body as T }); + }, + reject: (error: Error) => { + clearTimeout(timeout); + this.pendingRequests.delete(id); + reject(error); + }, + isStreaming: false + }); + + this.send(request); + }); + } + + /** + * Send a streaming request and return a ReadableStream + * + * The stream will receive data chunks as they arrive over the WebSocket. + * Format matches SSE for compatibility with existing streaming code. + */ + async requestStream( + method: WSMethod, + path: string, + body?: unknown + ): Promise> { + await this.connect(); + + const id = generateRequestId(); + const request: WSRequest = { + type: 'request', + id, + method, + path, + body + }; + + const encoder = new TextEncoder(); + + return new ReadableStream({ + start: (controller) => { + const timeoutMs = this.options.requestTimeoutMs ?? 120000; + const timeout = setTimeout(() => { + this.pendingRequests.delete(id); + controller.error( + new Error(`Stream timeout after ${timeoutMs}ms: ${method} ${path}`) + ); + }, timeoutMs); + + this.pendingRequests.set(id, { + resolve: (response: WSResponse) => { + clearTimeout(timeout); + this.pendingRequests.delete(id); + // Final response - close the stream + if (response.status >= 400) { + controller.error( + new Error( + `Stream error: ${response.status} - ${JSON.stringify(response.body)}` + ) + ); + } else { + controller.close(); + } + }, + reject: (error: Error) => { + clearTimeout(timeout); + this.pendingRequests.delete(id); + controller.error(error); + }, + streamController: controller, + isStreaming: true + }); + + this.send(request); + }, + cancel: () => { + this.pendingRequests.delete(id); + // Could send a cancel message to server if needed + } + }); + } + + /** + * Send a message over the WebSocket + */ + private send(message: WSRequest): void { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { + throw new Error('WebSocket not connected'); + } + + this.ws.send(JSON.stringify(message)); + this.logger.debug('WebSocket sent', { + id: message.id, + method: message.method, + path: message.path + }); + } + + /** + * Handle incoming WebSocket messages + */ + private handleMessage(event: MessageEvent): void { + try { + const message = JSON.parse(event.data) as WSServerMessage; + + if (isWSResponse(message)) { + this.handleResponse(message); + } else if (isWSStreamChunk(message)) { + this.handleStreamChunk(message); + } else if (isWSError(message)) { + this.handleError(message); + } else { + this.logger.warn('Unknown WebSocket message type', { message }); + } + } catch (error) { + this.logger.error( + 'Failed to parse WebSocket message', + error instanceof Error ? error : new Error(String(error)) + ); + } + } + + /** + * Handle a response message + */ + private handleResponse(response: WSResponse): void { + const pending = this.pendingRequests.get(response.id); + if (!pending) { + this.logger.warn('Received response for unknown request', { + id: response.id + }); + return; + } + + this.logger.debug('WebSocket response', { + id: response.id, + status: response.status, + done: response.done + }); + + // Only resolve when done is true + if (response.done) { + pending.resolve(response); + } + } + + /** + * Handle a stream chunk message + */ + private handleStreamChunk(chunk: WSStreamChunk): void { + const pending = this.pendingRequests.get(chunk.id); + if (!pending || !pending.streamController) { + this.logger.warn('Received stream chunk for unknown request', { + id: chunk.id + }); + return; + } + + // Convert to SSE format for compatibility with existing parsers + const encoder = new TextEncoder(); + let sseData: string; + if (chunk.event) { + sseData = `event: ${chunk.event}\ndata: ${chunk.data}\n\n`; + } else { + sseData = `data: ${chunk.data}\n\n`; + } + + try { + pending.streamController.enqueue(encoder.encode(sseData)); + } catch (error) { + // Stream may have been cancelled + this.logger.debug('Failed to enqueue stream chunk', { + id: chunk.id, + error: error instanceof Error ? error.message : String(error) + }); + } + } + + /** + * Handle an error message + */ + private handleError(error: { + id?: string; + code: string; + message: string; + status: number; + }): void { + if (error.id) { + const pending = this.pendingRequests.get(error.id); + if (pending) { + pending.reject(new Error(`${error.code}: ${error.message}`)); + return; + } + } + + // Global error - log it + this.logger.error('WebSocket error message', new Error(error.message), { + code: error.code, + status: error.status + }); + } + + /** + * Handle WebSocket close + */ + private handleClose(event: CloseEvent): void { + this.state = 'disconnected'; + this.ws = null; + + // Reject all pending requests + for (const [id, pending] of this.pendingRequests) { + pending.reject( + new Error( + `WebSocket closed: ${event.code} ${event.reason || 'No reason'}` + ) + ); + } + this.pendingRequests.clear(); + } + + /** + * Cleanup resources + */ + private cleanup(): void { + if (this.ws) { + this.ws.removeEventListener('close', this.boundHandleClose); + this.ws.removeEventListener('message', this.boundHandleMessage); + this.ws.close(); + this.ws = null; + } + this.state = 'disconnected'; + this.connectPromise = null; + this.pendingRequests.clear(); + } +} diff --git a/packages/sandbox/src/sandbox.ts b/packages/sandbox/src/sandbox.ts index b5014691..f59c5987 100644 --- a/packages/sandbox/src/sandbox.ts +++ b/packages/sandbox/src/sandbox.ts @@ -85,6 +85,10 @@ export function getSandbox( stub.setContainerTimeouts(options.containerTimeouts); } + if (options?.useWebSocket !== undefined) { + stub.setUseWebSocket(options.useWebSocket); + } + return Object.assign(stub, { wsConnect: connect(stub) }); @@ -119,6 +123,7 @@ export class Sandbox extends Container implements ISandbox { private logger: ReturnType; private keepAliveEnabled: boolean = false; private activeMounts: Map = new Map(); + private useWebSocketTransport: boolean = false; /** * Default container startup timeouts (conservative for production) @@ -202,6 +207,23 @@ export class Sandbox extends Container implements ISandbox { ...storedTimeouts }; } + + // Load WebSocket transport setting + const storedUseWebSocket = + (await this.ctx.storage.get('useWebSocket')) || false; + if (storedUseWebSocket) { + this.useWebSocketTransport = true; + // Recreate client with WebSocket transport + this.client = new SandboxClient({ + logger: this.logger, + port: 3000, + stub: this, + transportMode: 'websocket', + wsUrl: 'ws://localhost:3000/ws' + }); + // Re-initialize code interpreter with new client + this.codeInterpreter = new CodeInterpreter(this); + } }); } @@ -247,6 +269,48 @@ export class Sandbox extends Container implements ISandbox { } } + /** + * RPC method to enable WebSocket transport for control plane communication + * + * When enabled, all sandbox operations are multiplexed over a single WebSocket + * connection instead of individual HTTP requests, reducing sub-request count. + * + * Note: The WebSocket connection is established on the first request. + */ + async setUseWebSocket(useWebSocket: boolean): Promise { + if (this.useWebSocketTransport === useWebSocket) { + return; // No change needed + } + + this.useWebSocketTransport = useWebSocket; + await this.ctx.storage.put('useWebSocket', useWebSocket); + + if (useWebSocket) { + this.logger.info( + 'WebSocket transport enabled - requests will be multiplexed over single connection' + ); + // Recreate client with WebSocket transport + this.client = new SandboxClient({ + logger: this.logger, + port: 3000, + stub: this, + transportMode: 'websocket', + wsUrl: 'ws://localhost:3000/ws' + }); + } else { + this.logger.info('WebSocket transport disabled - using HTTP requests'); + // Recreate client with HTTP transport + this.client = new SandboxClient({ + logger: this.logger, + port: 3000, + stub: this + }); + } + + // Re-initialize code interpreter with new client + this.codeInterpreter = new CodeInterpreter(this); + } + // RPC method to set environment variables async setEnvVars(envVars: Record): Promise { // Update local state for new sessions diff --git a/packages/sandbox/tests/transport.test.ts b/packages/sandbox/tests/transport.test.ts new file mode 100644 index 00000000..34c22e35 --- /dev/null +++ b/packages/sandbox/tests/transport.test.ts @@ -0,0 +1,264 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { createTransport, Transport } from '../src/clients/transport'; + +describe('Transport', () => { + let mockFetch: ReturnType; + + beforeEach(() => { + vi.clearAllMocks(); + mockFetch = vi.fn(); + global.fetch = mockFetch as unknown as typeof fetch; + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + describe('HTTP mode', () => { + it('should create transport in HTTP mode by default', () => { + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + expect(transport.getMode()).toBe('http'); + }); + + it('should make HTTP GET request', async () => { + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + mockFetch.mockResolvedValue( + new Response(JSON.stringify({ data: 'test' }), { status: 200 }) + ); + + const result = await transport.request('GET', '/api/test'); + + expect(result.status).toBe(200); + expect(result.body).toEqual({ data: 'test' }); + expect(mockFetch).toHaveBeenCalledWith( + 'http://localhost:3000/api/test', + expect.objectContaining({ method: 'GET' }) + ); + }); + + it('should make HTTP POST request with body', async () => { + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + mockFetch.mockResolvedValue( + new Response(JSON.stringify({ success: true }), { status: 200 }) + ); + + const result = await transport.request('POST', '/api/execute', { + command: 'echo hello' + }); + + expect(result.status).toBe(200); + expect(mockFetch).toHaveBeenCalledWith( + 'http://localhost:3000/api/execute', + expect.objectContaining({ + method: 'POST', + body: JSON.stringify({ command: 'echo hello' }) + }) + ); + }); + + it('should handle HTTP errors', async () => { + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + mockFetch.mockResolvedValue( + new Response(JSON.stringify({ error: 'Not found' }), { status: 404 }) + ); + + const result = await transport.request('GET', '/api/missing'); + + expect(result.status).toBe(404); + }); + + it('should stream HTTP responses', async () => { + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + const mockStream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: test\n\n')); + controller.close(); + } + }); + + mockFetch.mockResolvedValue( + new Response(mockStream, { + status: 200, + headers: { 'Content-Type': 'text/event-stream' } + }) + ); + + const stream = await transport.requestStream('POST', '/api/stream', {}); + + expect(stream).toBeInstanceOf(ReadableStream); + }); + + it('should use stub.containerFetch when stub is provided', async () => { + const mockContainerFetch = vi + .fn() + .mockResolvedValue( + new Response(JSON.stringify({ ok: true }), { status: 200 }) + ); + + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000', + stub: { containerFetch: mockContainerFetch }, + port: 3000 + }); + + await transport.request('GET', '/api/test'); + + expect(mockContainerFetch).toHaveBeenCalledWith( + 'http://localhost:3000/api/test', + expect.any(Object), + 3000 + ); + expect(mockFetch).not.toHaveBeenCalled(); + }); + }); + + describe('WebSocket mode', () => { + // Note: Full WebSocket tests are in ws-transport.test.ts + // These tests verify the Transport wrapper behavior + + it('should create transport in WebSocket mode', () => { + const transport = createTransport({ + mode: 'websocket', + wsUrl: 'ws://localhost:3000/ws' + }); + + expect(transport.getMode()).toBe('websocket'); + }); + + it('should report WebSocket connection state', () => { + const transport = createTransport({ + mode: 'websocket', + wsUrl: 'ws://localhost:3000/ws' + }); + + // Initially not connected + expect(transport.isWebSocketConnected()).toBe(false); + }); + + it('should handle missing WebSocket URL gracefully', () => { + // When wsUrl is missing, transport is created but won't connect + const transport = createTransport({ + mode: 'websocket' + // wsUrl missing - will fail on connect attempt + }); + + // Transport is created but in an invalid state for WebSocket + expect(transport.getMode()).toBe('websocket'); + expect(transport.isWebSocketConnected()).toBe(false); + }); + }); + + describe('createTransport factory', () => { + it('should create HTTP transport with minimal options', () => { + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + expect(transport).toBeInstanceOf(Transport); + expect(transport.getMode()).toBe('http'); + }); + + it('should create WebSocket transport with URL', () => { + const transport = createTransport({ + mode: 'websocket', + wsUrl: 'ws://localhost:3000/ws' + }); + + expect(transport).toBeInstanceOf(Transport); + expect(transport.getMode()).toBe('websocket'); + }); + + it('should pass logger to transport', () => { + const mockLogger = { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + child: vi.fn() + }; + + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000', + logger: mockLogger as any + }); + + expect(transport).toBeDefined(); + }); + }); + + describe('mode switching', () => { + it('should maintain mode throughout lifecycle', async () => { + const httpTransport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + expect(httpTransport.getMode()).toBe('http'); + + mockFetch.mockResolvedValue( + new Response(JSON.stringify({}), { status: 200 }) + ); + + await httpTransport.request('GET', '/test'); + + // Mode should still be http + expect(httpTransport.getMode()).toBe('http'); + }); + }); +}); + +describe('Transport with SandboxClient', () => { + let mockFetch: ReturnType; + + beforeEach(() => { + vi.clearAllMocks(); + mockFetch = vi.fn(); + global.fetch = mockFetch as unknown as typeof fetch; + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('should work with shared transport across clients', async () => { + // Create a shared transport + const transport = createTransport({ + mode: 'http', + baseUrl: 'http://localhost:3000' + }); + + mockFetch.mockResolvedValue( + new Response(JSON.stringify({ success: true }), { status: 200 }) + ); + + // Multiple requests through same transport + await transport.request('POST', '/api/mkdir', { path: '/test' }); + await transport.request('POST', '/api/write', { path: '/test/file.txt' }); + await transport.request('GET', '/api/read'); + + expect(mockFetch).toHaveBeenCalledTimes(3); + }); +}); diff --git a/packages/sandbox/tests/ws-transport.test.ts b/packages/sandbox/tests/ws-transport.test.ts new file mode 100644 index 00000000..af9966c1 --- /dev/null +++ b/packages/sandbox/tests/ws-transport.test.ts @@ -0,0 +1,315 @@ +import type { + WSError, + WSRequest, + WSResponse, + WSStreamChunk +} from '@repo/shared'; +import { + generateRequestId, + isWSError, + isWSRequest, + isWSResponse, + isWSStreamChunk +} from '@repo/shared'; +import { describe, expect, it } from 'vitest'; + +/** + * Tests for WebSocket protocol types and utilities. + * + * Note: Full WSTransport integration tests require a real WebSocket environment + * and are covered in E2E tests. These unit tests focus on the protocol layer: + * message types, type guards, and request ID generation. + */ +describe('WebSocket Protocol Types', () => { + describe('generateRequestId', () => { + it('should generate unique request IDs', () => { + const id1 = generateRequestId(); + const id2 = generateRequestId(); + const id3 = generateRequestId(); + + expect(id1).toMatch(/^ws_\d+_[a-z0-9]+$/); + expect(id2).toMatch(/^ws_\d+_[a-z0-9]+$/); + expect(id3).toMatch(/^ws_\d+_[a-z0-9]+$/); + + // All should be unique + expect(new Set([id1, id2, id3]).size).toBe(3); + }); + + it('should include timestamp in ID', () => { + const before = Date.now(); + const id = generateRequestId(); + const after = Date.now(); + + // Extract timestamp from ID (format: ws__) + const parts = id.split('_'); + const timestamp = parseInt(parts[1], 10); + + expect(timestamp).toBeGreaterThanOrEqual(before); + expect(timestamp).toBeLessThanOrEqual(after); + }); + }); + + describe('isWSRequest', () => { + it('should return true for valid WSRequest', () => { + const request: WSRequest = { + type: 'request', + id: 'req-123', + method: 'POST', + path: '/api/execute', + body: { command: 'echo hello' } + }; + + expect(isWSRequest(request)).toBe(true); + }); + + it('should return true for minimal WSRequest', () => { + const request = { + type: 'request', + id: 'req-456', + method: 'GET', + path: '/api/health' + }; + + expect(isWSRequest(request)).toBe(true); + }); + + it('should return false for non-request types', () => { + expect(isWSRequest(null)).toBe(false); + expect(isWSRequest(undefined)).toBe(false); + expect(isWSRequest('string')).toBe(false); + expect(isWSRequest({ type: 'response' })).toBe(false); + expect(isWSRequest({ type: 'error' })).toBe(false); + }); + }); + + describe('isWSResponse', () => { + it('should return true for valid WSResponse', () => { + const response: WSResponse = { + type: 'response', + id: 'req-123', + status: 200, + body: { data: 'test' }, + done: true + }; + + expect(isWSResponse(response)).toBe(true); + }); + + it('should return true for minimal WSResponse', () => { + const response = { + type: 'response', + id: 'req-456', + status: 404, + done: false + }; + + expect(isWSResponse(response)).toBe(true); + }); + + it('should return false for non-response types', () => { + expect(isWSResponse(null)).toBe(false); + expect(isWSResponse(undefined)).toBe(false); + expect(isWSResponse('string')).toBe(false); + expect(isWSResponse({ type: 'error' })).toBe(false); + expect(isWSResponse({ type: 'stream' })).toBe(false); + expect(isWSResponse({ type: 'request' })).toBe(false); + }); + }); + + describe('isWSError', () => { + it('should return true for valid WSError', () => { + const error: WSError = { + type: 'error', + id: 'req-123', + code: 'NOT_FOUND', + message: 'Resource not found', + status: 404 + }; + + expect(isWSError(error)).toBe(true); + }); + + it('should return true for WSError without id', () => { + const error = { + type: 'error', + code: 'PARSE_ERROR', + message: 'Invalid JSON', + status: 400 + }; + + expect(isWSError(error)).toBe(true); + }); + + it('should return false for non-error types', () => { + expect(isWSError(null)).toBe(false); + expect(isWSError(undefined)).toBe(false); + expect(isWSError({ type: 'response' })).toBe(false); + expect(isWSError({ type: 'stream' })).toBe(false); + }); + }); + + describe('isWSStreamChunk', () => { + it('should return true for valid WSStreamChunk', () => { + const chunk: WSStreamChunk = { + type: 'stream', + id: 'req-123', + data: 'chunk data' + }; + + expect(isWSStreamChunk(chunk)).toBe(true); + }); + + it('should return true for WSStreamChunk with event', () => { + const chunk = { + type: 'stream', + id: 'req-456', + event: 'output', + data: 'line of output' + }; + + expect(isWSStreamChunk(chunk)).toBe(true); + }); + + it('should return false for non-stream types', () => { + expect(isWSStreamChunk(null)).toBe(false); + expect(isWSStreamChunk({ type: 'response' })).toBe(false); + expect(isWSStreamChunk({ type: 'error' })).toBe(false); + }); + }); +}); + +describe('WebSocket Message Serialization', () => { + it('should serialize WSRequest correctly', () => { + const request: WSRequest = { + type: 'request', + id: generateRequestId(), + method: 'POST', + path: '/api/execute', + body: { command: 'echo hello', sessionId: 'sess-1' } + }; + + const serialized = JSON.stringify(request); + const parsed = JSON.parse(serialized); + + expect(parsed.type).toBe('request'); + expect(parsed.method).toBe('POST'); + expect(parsed.path).toBe('/api/execute'); + expect(parsed.body.command).toBe('echo hello'); + expect(isWSRequest(parsed)).toBe(true); + }); + + it('should serialize WSResponse correctly', () => { + const response: WSResponse = { + type: 'response', + id: 'req-123', + status: 200, + body: { + success: true, + stdout: 'hello\n', + stderr: '', + exitCode: 0 + }, + done: true + }; + + const serialized = JSON.stringify(response); + const parsed = JSON.parse(serialized); + + expect(parsed.type).toBe('response'); + expect(parsed.status).toBe(200); + expect(parsed.body.stdout).toBe('hello\n'); + expect(parsed.done).toBe(true); + expect(isWSResponse(parsed)).toBe(true); + }); + + it('should serialize WSError correctly', () => { + const error: WSError = { + type: 'error', + id: 'req-123', + code: 'FILE_NOT_FOUND', + message: 'File not found: /test.txt', + status: 404, + context: { path: '/test.txt' } + }; + + const serialized = JSON.stringify(error); + const parsed = JSON.parse(serialized); + + expect(parsed.type).toBe('error'); + expect(parsed.code).toBe('FILE_NOT_FOUND'); + expect(parsed.status).toBe(404); + expect(isWSError(parsed)).toBe(true); + }); + + it('should serialize WSStreamChunk correctly', () => { + const chunk: WSStreamChunk = { + type: 'stream', + id: 'req-123', + event: 'stdout', + data: 'output line\n' + }; + + const serialized = JSON.stringify(chunk); + const parsed = JSON.parse(serialized); + + expect(parsed.type).toBe('stream'); + expect(parsed.event).toBe('stdout'); + expect(parsed.data).toBe('output line\n'); + expect(isWSStreamChunk(parsed)).toBe(true); + }); + + it('should handle special characters in body', () => { + const response: WSResponse = { + type: 'response', + id: 'req-123', + status: 200, + body: { + content: 'Line 1\nLine 2\tTabbed\r\nWindows line' + }, + done: true + }; + + const serialized = JSON.stringify(response); + const parsed = JSON.parse(serialized); + + expect(parsed.body.content).toBe('Line 1\nLine 2\tTabbed\r\nWindows line'); + }); + + it('should handle binary data as base64', () => { + const binaryData = 'SGVsbG8gV29ybGQ='; // "Hello World" in base64 + + const response: WSResponse = { + type: 'response', + id: 'req-123', + status: 200, + body: { + content: binaryData, + encoding: 'base64' + }, + done: true + }; + + const serialized = JSON.stringify(response); + const parsed = JSON.parse(serialized); + + expect(parsed.body.content).toBe(binaryData); + expect(parsed.body.encoding).toBe('base64'); + }); + + it('should handle large payloads', () => { + const largeContent = 'x'.repeat(100000); + + const response: WSResponse = { + type: 'response', + id: 'req-123', + status: 200, + body: { content: largeContent }, + done: true + }; + + const serialized = JSON.stringify(response); + const parsed = JSON.parse(serialized); + + expect(parsed.body.content.length).toBe(100000); + }); +}); diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index d4367084..45548fe8 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -106,3 +106,20 @@ export type { WriteFileResult } from './types.js'; export { isExecResult, isProcess, isProcessStatus } from './types.js'; +// Export WebSocket protocol types +export type { + WSClientMessage, + WSError, + WSMethod, + WSRequest, + WSResponse, + WSServerMessage, + WSStreamChunk +} from './ws-types.js'; +export { + generateRequestId, + isWSError, + isWSRequest, + isWSResponse, + isWSStreamChunk +} from './ws-types.js'; diff --git a/packages/shared/src/logger/types.ts b/packages/shared/src/logger/types.ts index 1b6edb59..6de0e4c9 100644 --- a/packages/shared/src/logger/types.ts +++ b/packages/shared/src/logger/types.ts @@ -14,7 +14,11 @@ export enum LogLevel { ERROR = 3 } -export type LogComponent = 'container' | 'sandbox-do' | 'executor'; +export type LogComponent = + | 'container' + | 'sandbox-do' + | 'executor' + | 'ws-handler'; /** * Context metadata included in every log entry diff --git a/packages/shared/src/types.ts b/packages/shared/src/types.ts index 50238211..659a187a 100644 --- a/packages/shared/src/types.ts +++ b/packages/shared/src/types.ts @@ -358,6 +358,22 @@ export interface SandboxOptions { */ waitIntervalMS?: number; }; + + /** + * Use WebSocket transport for control plane communication + * + * When enabled, all sandbox operations (file I/O, command execution, etc.) + * are multiplexed over a single WebSocket connection instead of individual + * HTTP requests. This significantly reduces sub-request count when running + * inside Workers or Durable Objects. + * + * **Use cases:** + * - Agent loops with many file operations inside a Worker/DO + * - Any scenario where sub-request limits are a concern + * + * @default false + */ + useWebSocket?: boolean; } /** diff --git a/packages/shared/src/ws-types.ts b/packages/shared/src/ws-types.ts new file mode 100644 index 00000000..bd29c847 --- /dev/null +++ b/packages/shared/src/ws-types.ts @@ -0,0 +1,166 @@ +/** + * WebSocket transport protocol types + * + * Enables multiplexing HTTP-like requests over a single WebSocket connection. + * This reduces sub-request count when running inside Workers/Durable Objects. + * + * Protocol: + * - Client sends WSRequest messages + * - Server responds with WSResponse messages (matched by id) + * - For streaming endpoints, server sends multiple WSStreamChunk messages + * followed by a final WSResponse + */ + +/** + * HTTP methods supported over WebSocket + */ +export type WSMethod = 'GET' | 'POST' | 'PUT' | 'DELETE'; + +/** + * WebSocket request message sent from client to server + */ +export interface WSRequest { + /** Message type discriminator */ + type: 'request'; + + /** Unique request ID for response matching */ + id: string; + + /** HTTP method */ + method: WSMethod; + + /** Request path (e.g., '/api/execute', '/api/read') */ + path: string; + + /** Request body (for POST/PUT requests) */ + body?: unknown; + + /** Request headers (optional, for special cases) */ + headers?: Record; +} + +/** + * WebSocket response message sent from server to client + */ +export interface WSResponse { + /** Message type discriminator */ + type: 'response'; + + /** Request ID this response corresponds to */ + id: string; + + /** HTTP status code */ + status: number; + + /** Response body (JSON parsed) */ + body?: unknown; + + /** Whether this is the final response (for streaming, false until complete) */ + done: boolean; +} + +/** + * WebSocket stream chunk for streaming responses (SSE replacement) + * Sent for streaming endpoints like /api/execute/stream, /api/read/stream + */ +export interface WSStreamChunk { + /** Message type discriminator */ + type: 'stream'; + + /** Request ID this chunk belongs to */ + id: string; + + /** Stream event type (matches SSE event types) */ + event?: string; + + /** Chunk data */ + data: string; +} + +/** + * WebSocket error response + */ +export interface WSError { + /** Message type discriminator */ + type: 'error'; + + /** Request ID this error corresponds to (if available) */ + id?: string; + + /** Error code */ + code: string; + + /** Error message */ + message: string; + + /** HTTP status code equivalent */ + status: number; + + /** Additional error context */ + context?: Record; +} + +/** + * Union type for all WebSocket messages from server to client + */ +export type WSServerMessage = WSResponse | WSStreamChunk | WSError; + +/** + * Union type for all WebSocket messages from client to server + */ +export type WSClientMessage = WSRequest; + +/** + * Type guard for WSRequest + */ +export function isWSRequest(msg: unknown): msg is WSRequest { + return ( + typeof msg === 'object' && + msg !== null && + 'type' in msg && + (msg as WSRequest).type === 'request' + ); +} + +/** + * Type guard for WSResponse + */ +export function isWSResponse(msg: unknown): msg is WSResponse { + return ( + typeof msg === 'object' && + msg !== null && + 'type' in msg && + (msg as WSResponse).type === 'response' + ); +} + +/** + * Type guard for WSStreamChunk + */ +export function isWSStreamChunk(msg: unknown): msg is WSStreamChunk { + return ( + typeof msg === 'object' && + msg !== null && + 'type' in msg && + (msg as WSStreamChunk).type === 'stream' + ); +} + +/** + * Type guard for WSError + */ +export function isWSError(msg: unknown): msg is WSError { + return ( + typeof msg === 'object' && + msg !== null && + 'type' in msg && + (msg as WSError).type === 'error' + ); +} + +/** + * Generate a unique request ID + */ +export function generateRequestId(): string { + return `ws_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`; +} diff --git a/tests/e2e/helpers/test-fixtures.ts b/tests/e2e/helpers/test-fixtures.ts index 862338e5..bbec88c2 100644 --- a/tests/e2e/helpers/test-fixtures.ts +++ b/tests/e2e/helpers/test-fixtures.ts @@ -27,11 +27,23 @@ export function createSessionId(): string { return `session-${randomUUID()}`; } +/** + * Options for creating test headers + */ +export interface TestHeaderOptions { + /** Session ID for session isolation tests */ + sessionId?: string; + /** Enable WebSocket transport instead of HTTP */ + useWebSocket?: boolean; + /** Enable keepAlive mode */ + keepAlive?: boolean; +} + /** * Create headers for sandbox/session identification * * @param sandboxId - Which container instance to use - * @param sessionId - (Optional) Which session within that container (SDK defaults to auto-managed session) + * @param options - Optional configuration (sessionId, useWebSocket, keepAlive) * * @example * // Most tests: unique sandbox, default session @@ -40,20 +52,36 @@ export function createSessionId(): string { * @example * // Session isolation tests: one sandbox, multiple sessions * const sandboxId = createSandboxId(); - * const headers1 = createTestHeaders(sandboxId, createSessionId()); - * const headers2 = createTestHeaders(sandboxId, createSessionId()); + * const headers1 = createTestHeaders(sandboxId, { sessionId: createSessionId() }); + * const headers2 = createTestHeaders(sandboxId, { sessionId: createSessionId() }); + * + * @example + * // WebSocket transport tests + * const headers = createTestHeaders(createSandboxId(), { useWebSocket: true }); */ export function createTestHeaders( sandboxId: string, - sessionId?: string + options?: TestHeaderOptions | string ): Record { + // Support legacy signature: createTestHeaders(sandboxId, sessionId) + const opts: TestHeaderOptions = + typeof options === 'string' ? { sessionId: options } : options || {}; + const headers: Record = { 'Content-Type': 'application/json', 'X-Sandbox-Id': sandboxId }; - if (sessionId) { - headers['X-Session-Id'] = sessionId; + if (opts.sessionId) { + headers['X-Session-Id'] = opts.sessionId; + } + + if (opts.useWebSocket) { + headers['X-Use-WebSocket'] = 'true'; + } + + if (opts.keepAlive) { + headers['X-Sandbox-KeepAlive'] = 'true'; } return headers; diff --git a/tests/e2e/test-worker/index.ts b/tests/e2e/test-worker/index.ts index 96501b48..72d62475 100644 --- a/tests/e2e/test-worker/index.ts +++ b/tests/e2e/test-worker/index.ts @@ -55,8 +55,13 @@ export default { const keepAliveHeader = request.headers.get('X-Sandbox-KeepAlive'); const keepAlive = keepAliveHeader === 'true'; + // Check if WebSocket transport is requested + const useWebSocketHeader = request.headers.get('X-Use-WebSocket'); + const useWebSocket = useWebSocketHeader === 'true'; + const sandbox = getSandbox(env.Sandbox, sandboxId, { - keepAlive + keepAlive, + useWebSocket }); // Get session ID from header (optional) diff --git a/tests/e2e/websocket-transport.test.ts b/tests/e2e/websocket-transport.test.ts new file mode 100644 index 00000000..115c484e --- /dev/null +++ b/tests/e2e/websocket-transport.test.ts @@ -0,0 +1,469 @@ +/** + * E2E Test: WebSocket Transport + * + * Tests that SDK operations work correctly when using WebSocket transport + * instead of HTTP transport. This verifies the WebSocket control plane + * multiplexing implementation. + * + * These tests mirror the HTTP tests but use the X-Use-WebSocket header + * to enable WebSocket transport mode. + */ + +import { afterAll, afterEach, beforeAll, describe, expect, test } from 'vitest'; +import type { + WriteFileResult, + ReadFileResult, + MkdirResult, + ListFilesResult, + ExecResult +} from '@repo/shared'; +import { getTestWorkerUrl, WranglerDevRunner } from './helpers/wrangler-runner'; +import { + createSandboxId, + createTestHeaders, + cleanupSandbox +} from './helpers/test-fixtures'; + +describe('WebSocket Transport (E2E)', () => { + let runner: WranglerDevRunner | null; + let workerUrl: string; + let currentSandboxId: string | null = null; + + // Helper to create headers with WebSocket transport enabled + const createWSHeaders = (sandboxId: string) => + createTestHeaders(sandboxId, { useWebSocket: true }); + + beforeAll(async () => { + // Get test worker URL (CI: uses deployed URL, Local: spawns wrangler dev) + const result = await getTestWorkerUrl(); + workerUrl = result.url; + runner = result.runner; + }, 120000); // 2 minute timeout for wrangler startup + + afterEach(async () => { + // Cleanup sandbox container after each test + if (currentSandboxId) { + await cleanupSandbox(workerUrl, currentSandboxId); + currentSandboxId = null; + } + }); + + afterAll(async () => { + if (runner) { + await runner.stop(); + } + }); + + describe('File Operations via WebSocket', () => { + test('should create directories via WebSocket transport', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + // Create nested directory structure + const mkdirResponse = await fetch(`${workerUrl}/api/file/mkdir`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-test/nested/directory', + recursive: true + }) + }); + + const mkdirData = (await mkdirResponse.json()) as MkdirResult; + expect(mkdirData.success).toBe(true); + + // Verify directory exists + const lsResponse = await fetch(`${workerUrl}/api/execute`, { + method: 'POST', + headers, + body: JSON.stringify({ + command: 'ls -la /workspace/ws-test/nested/directory' + }) + }); + + const lsData = (await lsResponse.json()) as ExecResult; + expect(lsResponse.status).toBe(200); + expect(lsData.success).toBe(true); + }, 90000); + + test('should write and read files via WebSocket transport', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + // Create directory + await fetch(`${workerUrl}/api/file/mkdir`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-files', + recursive: true + }) + }); + + // Write file + const content = JSON.stringify({ + transport: 'websocket', + timestamp: Date.now() + }); + + const writeResponse = await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-files/config.json', + content + }) + }); + + expect(writeResponse.status).toBe(200); + + // Read file back + const readResponse = await fetch(`${workerUrl}/api/file/read`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-files/config.json' + }) + }); + + const readData = (await readResponse.json()) as ReadFileResult; + expect(readResponse.status).toBe(200); + expect(readData.content).toContain('websocket'); + }, 90000); + + test('should delete files via WebSocket transport', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + // Create and write file + await fetch(`${workerUrl}/api/file/mkdir`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-delete', + recursive: true + }) + }); + + await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-delete/temp.txt', + content: 'Delete me via WebSocket' + }) + }); + + // Delete file + const deleteResponse = await fetch(`${workerUrl}/api/file/delete`, { + method: 'DELETE', + headers, + body: JSON.stringify({ + path: '/workspace/ws-delete/temp.txt' + }) + }); + + expect(deleteResponse.status).toBe(200); + + // Verify file is deleted + const readResponse = await fetch(`${workerUrl}/api/file/read`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-delete/temp.txt' + }) + }); + + expect(readResponse.status).toBe(500); // File not found + }, 90000); + + test('should list files via WebSocket transport', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + // Create directory with files + await fetch(`${workerUrl}/api/file/mkdir`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-list', + recursive: true + }) + }); + + await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-list/file1.txt', + content: 'File 1' + }) + }); + + await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-list/file2.txt', + content: 'File 2' + }) + }); + + // List files + const listResponse = await fetch(`${workerUrl}/api/list-files`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-list' + }) + }); + + expect(listResponse.status).toBe(200); + const listData = (await listResponse.json()) as ListFilesResult; + + expect(listData.success).toBe(true); + expect(listData.count).toBe(2); + expect(listData.files.map((f) => f.name).sort()).toEqual([ + 'file1.txt', + 'file2.txt' + ]); + }, 90000); + }); + + describe('Command Execution via WebSocket', () => { + test('should execute commands via WebSocket transport', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + const execResponse = await fetch(`${workerUrl}/api/execute`, { + method: 'POST', + headers, + body: JSON.stringify({ + command: 'echo "WebSocket transport works!"' + }) + }); + + const execData = (await execResponse.json()) as ExecResult; + expect(execResponse.status).toBe(200); + expect(execData.success).toBe(true); + expect(execData.stdout).toContain('WebSocket transport works!'); + }, 90000); + + test('should handle command with environment variables via WebSocket', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + const execResponse = await fetch(`${workerUrl}/api/execute`, { + method: 'POST', + headers, + body: JSON.stringify({ + command: 'echo "Transport: $TRANSPORT_MODE"', + env: { TRANSPORT_MODE: 'websocket' } + }) + }); + + const execData = (await execResponse.json()) as ExecResult; + expect(execResponse.status).toBe(200); + expect(execData.stdout).toContain('Transport: websocket'); + }, 90000); + + test('should handle command failures via WebSocket', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + const execResponse = await fetch(`${workerUrl}/api/execute`, { + method: 'POST', + headers, + body: JSON.stringify({ + command: 'exit 42' + }) + }); + + const execData = (await execResponse.json()) as ExecResult; + expect(execResponse.status).toBe(200); + expect(execData.exitCode).toBe(42); + expect(execData.success).toBe(false); + }, 90000); + }); + + describe('Multiple Operations via WebSocket', () => { + test('should handle many sequential operations efficiently', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + // Create base directory + await fetch(`${workerUrl}/api/file/mkdir`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-batch', + recursive: true + }) + }); + + // Perform many file operations (this is where WebSocket shines) + const operationCount = 10; + const startTime = Date.now(); + + for (let i = 0; i < operationCount; i++) { + // Write file + await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: `/workspace/ws-batch/file${i}.txt`, + content: `Content ${i}` + }) + }); + + // Read file back + const readResponse = await fetch(`${workerUrl}/api/file/read`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: `/workspace/ws-batch/file${i}.txt` + }) + }); + + const readData = (await readResponse.json()) as ReadFileResult; + expect(readData.content).toBe(`Content ${i}`); + } + + const duration = Date.now() - startTime; + console.log( + `WebSocket: ${operationCount * 2} operations completed in ${duration}ms` + ); + + // Verify all files exist + const listResponse = await fetch(`${workerUrl}/api/list-files`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-batch' + }) + }); + + const listData = (await listResponse.json()) as ListFilesResult; + expect(listData.count).toBe(operationCount); + }, 120000); + + test('should handle mixed operations via WebSocket', async () => { + currentSandboxId = createSandboxId(); + const headers = createWSHeaders(currentSandboxId); + + // Create directory structure + await fetch(`${workerUrl}/api/file/mkdir`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-mixed/src', + recursive: true + }) + }); + + // Write a script + await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers, + body: JSON.stringify({ + path: '/workspace/ws-mixed/src/script.sh', + content: '#!/bin/bash\necho "Hello from WebSocket!"' + }) + }); + + // Make it executable + await fetch(`${workerUrl}/api/execute`, { + method: 'POST', + headers, + body: JSON.stringify({ + command: 'chmod +x /workspace/ws-mixed/src/script.sh' + }) + }); + + // Execute the script + const execResponse = await fetch(`${workerUrl}/api/execute`, { + method: 'POST', + headers, + body: JSON.stringify({ + command: '/workspace/ws-mixed/src/script.sh' + }) + }); + + const execData = (await execResponse.json()) as ExecResult; + expect(execData.success).toBe(true); + expect(execData.stdout).toContain('Hello from WebSocket!'); + + // Clean up + await fetch(`${workerUrl}/api/execute`, { + method: 'POST', + headers, + body: JSON.stringify({ + command: 'rm -rf /workspace/ws-mixed' + }) + }); + }, 90000); + }); + + describe('WebSocket vs HTTP Comparison', () => { + test('should work identically to HTTP transport', async () => { + const sandboxId = createSandboxId(); + currentSandboxId = sandboxId; + + // Test with WebSocket + const wsHeaders = createTestHeaders(sandboxId, { useWebSocket: true }); + + await fetch(`${workerUrl}/api/file/mkdir`, { + method: 'POST', + headers: wsHeaders, + body: JSON.stringify({ + path: '/workspace/compare-test', + recursive: true + }) + }); + + const wsWriteResponse = await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers: wsHeaders, + body: JSON.stringify({ + path: '/workspace/compare-test/ws-file.txt', + content: 'Written via WebSocket' + }) + }); + expect(wsWriteResponse.status).toBe(200); + + // Now test with HTTP (same sandbox, different transport) + const httpHeaders = createTestHeaders(sandboxId); // No useWebSocket + + const httpWriteResponse = await fetch(`${workerUrl}/api/file/write`, { + method: 'POST', + headers: httpHeaders, + body: JSON.stringify({ + path: '/workspace/compare-test/http-file.txt', + content: 'Written via HTTP' + }) + }); + expect(httpWriteResponse.status).toBe(200); + + // Read both files to verify they work + const wsReadResponse = await fetch(`${workerUrl}/api/file/read`, { + method: 'POST', + headers: wsHeaders, + body: JSON.stringify({ + path: '/workspace/compare-test/ws-file.txt' + }) + }); + const wsReadData = (await wsReadResponse.json()) as ReadFileResult; + expect(wsReadData.content).toBe('Written via WebSocket'); + + const httpReadResponse = await fetch(`${workerUrl}/api/file/read`, { + method: 'POST', + headers: httpHeaders, + body: JSON.stringify({ + path: '/workspace/compare-test/http-file.txt' + }) + }); + const httpReadData = (await httpReadResponse.json()) as ReadFileResult; + expect(httpReadData.content).toBe('Written via HTTP'); + }, 90000); + }); +});