mirror of
https://github.com/wu736139669/hapi.git
synced 2026-11-07 01:21:37 +00:00
fix(codex): allow larger histories over shared WebSockets (#2171)
This commit is contained in:
@@ -4,6 +4,7 @@ import WebSocket from 'ws';
|
||||
import { logger } from '@/ui/logger';
|
||||
import { JsonLineParser } from '@/utils/jsonLineParser';
|
||||
import { killProcessByChildProcess } from '@/utils/process';
|
||||
import { CODEX_WEBSOCKET_MAX_PAYLOAD } from './websocketLimits';
|
||||
import type {
|
||||
CollaborationModeListResponse,
|
||||
InitializeParams,
|
||||
@@ -229,7 +230,7 @@ export class CodexAppServerClient extends JsonLineParser {
|
||||
const url = endpoint.startsWith('unix://') ? `ws+unix://${endpoint.slice(7)}:/` : endpoint;
|
||||
const socket = new WebSocket(url, {
|
||||
headers: { Host: 'localhost', ...(this.options.token ? { Authorization: `Bearer ${this.options.token}` } : {}) },
|
||||
maxPayload: 64 * 1024 * 1024,
|
||||
maxPayload: CODEX_WEBSOCKET_MAX_PAYLOAD,
|
||||
perMessageDeflate: false,
|
||||
handshakeTimeout: 10_000
|
||||
});
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import { WebSocket, WebSocketServer } from 'ws';
|
||||
import { CodexAppServerClient } from '../codexAppServerClient';
|
||||
import { startCodexGateway, type Envelope } from './gateway';
|
||||
|
||||
async function fixture(after: (request: Envelope, response: Envelope, connection: string) => Promise<void>) {
|
||||
@@ -31,6 +32,26 @@ async function fixture(after: (request: Envelope, response: Envelope, connection
|
||||
} };
|
||||
}
|
||||
describe('native gateway barriers', () => {
|
||||
it('round-trips payloads larger than 64 MiB through the gateway and app-server client', async () => {
|
||||
const f = await fixture(async () => {});
|
||||
const client = new CodexAppServerClient({ endpoint: f.gateway.endpoint, token: 'secret' });
|
||||
try {
|
||||
await client.connect();
|
||||
await client.initialize({ clientInfo: { name: 'large-history-test', version: '1.0.0' }, capabilities: null });
|
||||
// Exercises all three receivers: gateway downstream, gateway upstream,
|
||||
// and CodexAppServerClient. A small request afterwards checks usability.
|
||||
const payload = 'x'.repeat(65 * 1024 * 1024);
|
||||
const response = await client.request<{ thread: { id: string } }>('thread/read', payload);
|
||||
expect(response.thread.id.length).toBe(payload.length);
|
||||
expect(response.thread.id === payload).toBe(true);
|
||||
const next = await client.request<{ thread: { id: string } }>('thread/read', 'small');
|
||||
expect(next.thread.id).toBe('small');
|
||||
} finally {
|
||||
await client.disconnect();
|
||||
await f.close();
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it('namespaces colliding native request IDs by connection and withholds replies until binding', async () => {
|
||||
let release!: () => void; const barrier = new Promise<void>(resolve => { release = resolve; });
|
||||
const after = vi.fn(async () => { await barrier; }); const f = await fixture(after);
|
||||
|
||||
@@ -3,6 +3,7 @@ import { chmod, unlink } from 'node:fs/promises';
|
||||
import { timingSafeEqual, randomUUID } from 'node:crypto';
|
||||
import WebSocket, { WebSocketServer } from 'ws';
|
||||
import { z } from 'zod';
|
||||
import { CODEX_WEBSOCKET_MAX_PAYLOAD } from '../websocketLimits';
|
||||
|
||||
export const EnvelopeSchema = z.object({
|
||||
id: z.union([z.string(), z.number()]).optional(),
|
||||
@@ -30,7 +31,7 @@ export async function startCodexGateway(options: {
|
||||
upstream: string; upstreamToken?: string; path?: string; token: string; hooks: GatewayHooks;
|
||||
}): Promise<{ endpoint: string; close(): Promise<void> }> {
|
||||
const server = createServer((_request, response) => { response.writeHead(404).end(); });
|
||||
const websocket = new WebSocketServer({ noServer: true, maxPayload: 64 * 1024 * 1024 });
|
||||
const websocket = new WebSocketServer({ noServer: true, maxPayload: CODEX_WEBSOCKET_MAX_PAYLOAD });
|
||||
const connections = new Map<WebSocket, WebSocket>();
|
||||
server.on('upgrade', (request, socket, head) => {
|
||||
if (!options.path) {
|
||||
@@ -47,7 +48,7 @@ export async function startCodexGateway(options: {
|
||||
const connectionId = randomUUID();
|
||||
const upstream = new WebSocket(socketUrl(options.upstream), {
|
||||
headers: { Host: 'localhost', ...(options.upstreamToken ? { Authorization: `Bearer ${options.upstreamToken}` } : {}) },
|
||||
handshakeTimeout: 10_000, maxPayload: 64 * 1024 * 1024, perMessageDeflate: false
|
||||
handshakeTimeout: 10_000, maxPayload: CODEX_WEBSOCKET_MAX_PAYLOAD, perMessageDeflate: false
|
||||
});
|
||||
connections.set(downstream, upstream);
|
||||
const pending = new Map<string | number, Envelope>();
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
// A full thread/read or thread/resume reply can exceed 64 MiB for long sessions.
|
||||
// Keep a finite bound on the authenticated loopback transport while allowing
|
||||
// large histories to pass through both the gateway and app-server client.
|
||||
export const CODEX_WEBSOCKET_MAX_PAYLOAD = 256 * 1024 * 1024;
|
||||
Reference in New Issue
Block a user