[codex] fix responses lifecycle parity and codex session serialization (#278)

* fix responses lifecycle parity * remove websocket test debug logging * fix responses proxy stream typecheck * update compact upstream response expectations * add community and deepwiki badges * update linuxdo banner link * fix responses review follow-ups * preserve whitespace in responses terminal deltas * fix responses stream reconciliation edges * fix responses lifecycle parity and codex session serialization * fix review follow-up edge cases --------- Co-authored-by: apple <apple@appledeMacBook-Pro.local>
2026-03-25 18:56:36 +08:00
parent 9ba4369a83
commit 639c387532
13 changed files with 777 additions and 67 deletions
@@ -1,3 +1,4 @@
+import { zstdCompressSync } from 'node:zlib';
 import Fastify, { type FastifyInstance } from 'fastify';
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';

@@ -191,4 +192,166 @@ describe('testRoutes proxy tester transport', () => {
    expect(url).toMatch(/\/v1\/files$/);
    expect(requestInit.body?.constructor?.name).toBe('FormData');
  });
+
+  it('decodes zstd-compressed buffered proxy test responses', async () => {
+    const payload = JSON.stringify({
+      choices: [
+        {
+          message: {
+            content: '你好，来自压缩响应',
+          },
+        },
+      ],
+    });
+    fetchMock.mockResolvedValue(new Response(zstdCompressSync(Buffer.from(payload)), {
+      status: 200,
+      headers: {
+        'content-encoding': 'zstd',
+        'content-type': 'application/json; charset=utf-8',
+      },
+    }));
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/api/test/proxy',
+      payload: {
+        method: 'POST',
+        path: '/v1/chat/completions',
+        requestKind: 'json',
+        stream: false,
+        jobMode: false,
+        rawMode: false,
+        jsonBody: {
+          model: 'gpt-4o-mini',
+          messages: [{ role: 'user', content: 'ping' }],
+        },
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.json()).toEqual({
+      choices: [
+        {
+          message: {
+            content: '你好，来自压缩响应',
+          },
+        },
+      ],
+    });
+  });
+
+  it('decodes zstd-compressed non-SSE stream fallback responses', async () => {
+    const payload = JSON.stringify({
+      choices: [
+        {
+          message: {
+            content: '你好，来自流式回退',
+          },
+        },
+      ],
+    });
+    fetchMock.mockResolvedValue(new Response(zstdCompressSync(Buffer.from(payload)), {
+      status: 200,
+      headers: {
+        'content-encoding': 'zstd',
+        'content-type': 'application/json; charset=utf-8',
+      },
+    }));
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/api/test/proxy/stream',
+      payload: {
+        method: 'POST',
+        path: '/v1/chat/completions',
+        requestKind: 'json',
+        stream: true,
+        jobMode: false,
+        rawMode: false,
+        jsonBody: {
+          model: 'gpt-4o-mini',
+          messages: [{ role: 'user', content: 'ping' }],
+        },
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.headers['content-type']).toContain('text/event-stream');
+    expect(response.payload).toBe(`data: ${payload}\n\ndata: [DONE]\n\n`);
+  });
+
+  it('encodes multiline non-SSE stream fallback responses as valid SSE', async () => {
+    const payload = 'line one\nline two\n';
+    fetchMock.mockResolvedValue(new Response(payload, {
+      status: 200,
+      headers: {
+        'content-type': 'text/plain; charset=utf-8',
+      },
+    }));
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/api/test/proxy/stream',
+      payload: {
+        method: 'POST',
+        path: '/v1/chat/completions',
+        requestKind: 'json',
+        stream: true,
+        jobMode: false,
+        rawMode: false,
+        jsonBody: {
+          model: 'gpt-4o-mini',
+          messages: [{ role: 'user', content: 'ping' }],
+        },
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.headers['content-type']).toContain('text/event-stream');
+    expect(response.payload).toBe('data: line one\ndata: line two\ndata: \n\ndata: [DONE]\n\n');
+  });
+
+  it('cancels and releases SSE readers when stream forwarding fails', async () => {
+    const cancel = vi.fn().mockResolvedValue(undefined);
+    const releaseLock = vi.fn();
+    const read = vi.fn().mockRejectedValue(new Error('boom'));
+    fetchMock.mockResolvedValue({
+      ok: true,
+      status: 200,
+      headers: new Headers({
+        'content-type': 'text/event-stream; charset=utf-8',
+      }),
+      body: {
+        getReader: () => ({
+          read,
+          cancel,
+          releaseLock,
+        }),
+      },
+    });
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/api/test/proxy/stream',
+      payload: {
+        method: 'POST',
+        path: '/v1/chat/completions',
+        requestKind: 'json',
+        stream: true,
+        jobMode: false,
+        rawMode: false,
+        jsonBody: {
+          model: 'gpt-4o-mini',
+          messages: [{ role: 'user', content: 'ping' }],
+        },
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.headers['content-type']).toContain('text/event-stream');
+    expect(response.payload).toContain('event: error');
+    expect(response.payload).toContain('"message":"boom"');
+    expect(cancel).toHaveBeenCalledTimes(1);
+    expect(releaseLock).toHaveBeenCalledTimes(1);
+  });
 });
@@ -2,6 +2,7 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
 import { randomUUID } from 'node:crypto';
 import { fetch, File as UndiciFile, FormData as UndiciFormData } from 'undici';
 import { config } from '../../config.js';
+import { readRuntimeResponseText } from '../../proxy-core/executors/types.js';

 type UndiciRequestInit = Parameters<typeof fetch>[1];

@@ -521,7 +522,7 @@ async function fetchProxyBuffered(
  });

  const contentType = upstream.headers.get('content-type') || '';
-  const text = await upstream.text();
+  const text = await readRuntimeResponseText(upstream);

  if (!upstream.ok) {
    throw new UpstreamProxyError(upstream.status, normalizeErrorPayload(text));
@@ -656,14 +657,14 @@ async function sendStreamingEnvelope(
  }

  if (!upstream.ok) {
-    const text = await upstream.text();
+    const text = await readRuntimeResponseText(upstream);
    cleanupClientListeners();
    return reply.code(upstream.status).send(normalizeErrorPayload(text));
  }

  const contentType = upstream.headers.get('content-type') || '';
-  const reader = upstream.body?.getReader();
-  if (!reader) {
+  const reader = contentType.includes('text/event-stream') ? upstream.body?.getReader() : null;
+  if (contentType.includes('text/event-stream') && !reader) {
    cleanupClientListeners();
    return reply.code(502).send({
      error: {
@@ -682,25 +683,30 @@ async function sendStreamingEnvelope(

  try {
    if (contentType.includes('text/event-stream')) {
-      while (true) {
-        const { done, value } = await reader.read();
-        if (done) break;
-        if (value) {
-          reply.raw.write(Buffer.from(value));
+      const streamReader = reader!;
+      try {
+        while (true) {
+          const { done, value } = await streamReader.read();
+          if (done) break;
+          if (value) {
+            reply.raw.write(Buffer.from(value));
+          }
+        }
+      } finally {
+        try {
+          await streamReader.cancel();
+        } catch {
+          // no-op
+        } finally {
+          streamReader.releaseLock();
        }
      }
    } else {
-      let text = '';
-      const decoder = new TextDecoder('utf-8');
-      while (true) {
-        const { done, value } = await reader.read();
-        if (done) break;
-        if (value) {
-          text += decoder.decode(value, { stream: true });
-        }
+      const text = await readRuntimeResponseText(upstream);
+      for (const line of text.split(/\r?\n/)) {
+        reply.raw.write(`data: ${line}\n`);
      }
-      text += decoder.decode();
-      reply.raw.write(`data: ${text}\n\n`);
+      reply.raw.write('\n');
      reply.raw.write('data: [DONE]\n\n');
    }
  } catch (error) {
@@ -711,11 +717,6 @@ async function sendStreamingEnvelope(
      reply.raw.write(`event: error\ndata: ${message}\n\n`);
    }
  } finally {
-    try {
-      await reader.cancel();
-    } catch {
-      // no-op
-    }
    cleanupClientListeners();
    if (!reply.raw.writableEnded) {
      reply.raw.end();
@@ -177,7 +177,7 @@ describe('chat proxy codex oauth compatibility', () => {
    const forwardedBody = JSON.parse(options.body);
    expect(forwardedBody.stream).toBe(true);
    expect(forwardedBody.instructions).toBe('');
-    expect(forwardedBody.store).toBeUndefined();
+    expect(forwardedBody.store).toBe(false);
    expect(forwardedBody.parallel_tool_calls).toBeUndefined();
    expect(forwardedBody.include).toBeUndefined();
    expect(forwardedBody.max_output_tokens).toBe(256);
@@ -1,3 +1,4 @@
+import { zstdCompressSync } from 'node:zlib';
 import Fastify, { type FastifyInstance } from 'fastify';
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
 import { config } from '../../config.js';
@@ -169,6 +170,79 @@ describe('chat proxy stream behavior', () => {
    expect(response.body).toContain('data: [DONE]');
  });

+  it('decodes zstd-compressed non-stream chat responses before serializing downstream JSON', async () => {
+    const payload = JSON.stringify({
+      id: 'chatcmpl-zstd',
+      object: 'chat.completion',
+      created: 1_706_000_000,
+      model: 'upstream-gpt',
+      choices: [{
+        index: 0,
+        message: { role: 'assistant', content: '你好，来自 zstd 非流式响应' },
+        finish_reason: 'stop',
+      }],
+      usage: { prompt_tokens: 11, completion_tokens: 7, total_tokens: 18 },
+    });
+    fetchMock.mockResolvedValue(new Response(zstdCompressSync(Buffer.from(payload)), {
+      status: 200,
+      headers: {
+        'content-encoding': 'zstd',
+        'content-type': 'application/json; charset=utf-8',
+      },
+    }));
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/v1/chat/completions',
+      payload: {
+        model: 'gpt-4o-mini',
+        messages: [{ role: 'user', content: 'hi' }],
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.json()?.choices?.[0]?.message?.content).toBe('你好，来自 zstd 非流式响应');
+  });
+
+  it('decodes zstd-compressed non-SSE streaming chat responses before SSE conversion', async () => {
+    const payload = JSON.stringify({
+      id: 'chatcmpl-zstd-stream',
+      object: 'chat.completion',
+      created: 1_706_000_000,
+      model: 'upstream-gpt',
+      choices: [{
+        index: 0,
+        message: { role: 'assistant', content: '你好，来自 zstd 流式回退' },
+        finish_reason: 'stop',
+      }],
+      usage: { prompt_tokens: 11, completion_tokens: 7, total_tokens: 18 },
+    });
+    fetchMock.mockResolvedValue(new Response(zstdCompressSync(Buffer.from(payload)), {
+      status: 200,
+      headers: {
+        'content-encoding': 'zstd',
+        'content-type': 'application/json; charset=utf-8',
+      },
+    }));
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/v1/chat/completions',
+      payload: {
+        model: 'gpt-4o-mini',
+        stream: true,
+        messages: [{ role: 'user', content: 'hi' }],
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.headers['content-type']).toContain('text/event-stream');
+    expect(response.body).toContain('"chat.completion.chunk"');
+    expect(response.body).toContain('你好，来自 zstd 流式回退');
+    expect(response.body).not.toContain('(�/�');
+    expect(response.body).toContain('data: [DONE]');
+  });
+
  it('returns upstream_error for empty non-stream chat responses when empty-content failure is enabled', async () => {
    config.proxyEmptyContentFailEnabled = true;

@@ -1,6 +1,7 @@
 import Fastify, { type FastifyInstance } from 'fastify';
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
 import { config } from '../../config.js';
+import { resetCodexHttpSessionQueue } from '../../proxy-core/runtime/codexHttpSessionQueue.js';

 const fetchMock = vi.fn();
 const selectChannelMock = vi.fn();
@@ -91,6 +92,26 @@ vi.mock('../../db/index.js', () => ({
 describe('responses proxy codex oauth refresh', () => {
  let app: FastifyInstance;

+  const createDeferred = <T>() => {
+    let resolve!: (value: T | PromiseLike<T>) => void;
+    let reject!: (reason?: unknown) => void;
+    const promise = new Promise<T>((res, rej) => {
+      resolve = res;
+      reject = rej;
+    });
+    return { promise, resolve, reject };
+  };
+
+  const waitFor = async (predicate: () => boolean, timeoutMs = 1000) => {
+    const startedAt = Date.now();
+    while (!predicate()) {
+      if ((Date.now() - startedAt) >= timeoutMs) {
+        throw new Error('Timed out waiting for test condition');
+      }
+      await new Promise((resolve) => setTimeout(resolve, 0));
+    }
+  };
+
  const createSseResponse = (chunks: string[], status = 200) => {
    const encoder = new TextEncoder();
    return new Response(new ReadableStream<Uint8Array>({
@@ -113,6 +134,7 @@ describe('responses proxy codex oauth refresh', () => {
  });

  beforeEach(() => {
+    resetCodexHttpSessionQueue();
    config.proxyEmptyContentFailEnabled = false;
    fetchMock.mockReset();
    selectChannelMock.mockReset();
@@ -461,6 +483,84 @@ describe('responses proxy codex oauth refresh', () => {
    expect(options.headers.referer).toBeUndefined();
  });

+  it('serializes concurrent codex HTTP responses requests that share the same session id', async () => {
+    const firstUpstream = createDeferred<Response>();
+    fetchMock
+      .mockImplementationOnce(() => firstUpstream.promise)
+      .mockResolvedValueOnce(new Response(JSON.stringify({
+        id: 'resp_codex_serial_2',
+        object: 'response',
+        model: 'gpt-5.2-codex',
+        status: 'completed',
+        output_text: 'second request finished',
+        usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 },
+      }), {
+        status: 200,
+        headers: { 'content-type': 'application/json' },
+      }));
+
+    const firstResponsePromise = app.inject({
+      method: 'POST',
+      url: '/v1/responses',
+      headers: {
+        session_id: 'session-http-serial-1',
+      },
+      payload: {
+        model: 'gpt-5.2-codex',
+        input: 'first request',
+      },
+    });
+
+    await waitFor(() => fetchMock.mock.calls.length === 1);
+
+    const secondResponsePromise = app.inject({
+      method: 'POST',
+      url: '/v1/responses',
+      headers: {
+        session_id: 'session-http-serial-1',
+      },
+      payload: {
+        model: 'gpt-5.2-codex',
+        input: 'second request',
+      },
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+
+    firstUpstream.resolve(new Response(JSON.stringify({
+      id: 'resp_codex_serial_1',
+      object: 'response',
+      model: 'gpt-5.2-codex',
+      status: 'completed',
+      output_text: 'first request finished',
+      usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 },
+    }), {
+      status: 200,
+      headers: { 'content-type': 'application/json' },
+    }));
+
+    const firstResponse = await firstResponsePromise;
+    expect(firstResponse.statusCode).toBe(200);
+    expect(firstResponse.json()).toMatchObject({
+      output_text: 'first request finished',
+    });
+
+    await waitFor(() => fetchMock.mock.calls.length === 2);
+
+    const secondResponse = await secondResponsePromise;
+    expect(secondResponse.statusCode).toBe(200);
+    expect(secondResponse.json()).toMatchObject({
+      output_text: 'second request finished',
+    });
+
+    const [, firstOptions] = fetchMock.mock.calls[0] as [string, any];
+    const [, secondOptions] = fetchMock.mock.calls[1] as [string, any];
+    expect(firstOptions.headers.Session_id || firstOptions.headers.session_id).toBe('session-http-serial-1');
+    expect(secondOptions.headers.Session_id || secondOptions.headers.session_id).toBe('session-http-serial-1');
+  });
+
  it('records codex usage_limit_reached reset hints on upstream 429 failures', async () => {
    fetchMock.mockResolvedValue(new Response(JSON.stringify({
      error: {
@@ -525,7 +625,7 @@ describe('responses proxy codex oauth refresh', () => {
    const forwardedBody = JSON.parse(options.body);
    expect(forwardedBody.stream).toBe(true);
    expect(forwardedBody.instructions).toBe('');
-    expect(forwardedBody.store).toBeUndefined();
+    expect(forwardedBody.store).toBe(false);
    expect(forwardedBody.max_output_tokens).toBeUndefined();

    expect(response.json()).toMatchObject({
@@ -568,11 +668,11 @@ describe('responses proxy codex oauth refresh', () => {
    const secondBody = JSON.parse(secondOptions.body);

    expect(firstBody.instructions).toBe('');
-    expect(firstBody.store).toBeUndefined();
+    expect(firstBody.store).toBe(false);
    expect(firstBody.stream).toBe(true);
    expect(firstBody.max_output_tokens).toBeUndefined();
    expect(secondBody.instructions).toBe('');
-    expect(secondBody.store).toBeUndefined();
+    expect(secondBody.store).toBe(false);
    expect(secondBody.stream).toBe(true);
    expect(secondBody.max_output_tokens).toBeUndefined();
  });
@@ -628,7 +628,7 @@ describe('buildUpstreamEndpointRequest', () => {
    expect(request.body.instructions).toBe('');
    expect(request.body.prompt_cache_key).toBeUndefined();
    expect(request.body.stream).toBe(false);
-    expect(request.body.store).toBeUndefined();
+    expect(request.body.store).toBe(false);
    expect(request.body.parallel_tool_calls).toBeUndefined();
    expect(request.body.include).toBeUndefined();
    expect(request.body.max_output_tokens).toBe(4096);
@@ -762,7 +762,7 @@ describe('buildUpstreamEndpointRequest', () => {
    expect(request.body.prompt_cache_key).toBeUndefined();
    expect(request.body.instructions).toBe('');
    expect(request.body.stream).toBe(false);
-    expect(request.body.store).toBe(true);
+    expect(request.body.store).toBe(false);
    expect(request.body.parallel_tool_calls).toBe(false);
    expect(request.body.include).toEqual(['reasoning.encrypted_content', 'mcp_approval_request.details']);
    expect(request.body.previous_response_id).toBe('resp_prev_123');
@@ -848,7 +848,7 @@ describe('buildUpstreamEndpointRequest', () => {
    expect(httpRequest.headers['x-codex-beta-features']).toBeUndefined();
  });

-  it('applies configured payload rules before preparing codex responses requests', () => {
+  it('applies configured payload rules before preparing codex responses requests while forcing store false', () => {
    (config as any).payloadRules = {
      default: [
        {
@@ -864,6 +864,7 @@ describe('buildUpstreamEndpointRequest', () => {
          models: [{ name: 'gpt-5.4', protocol: 'codex' }],
          params: {
            'text.verbosity': 'low',
+            store: true,
          },
        },
      ],
@@ -899,6 +900,7 @@ describe('buildUpstreamEndpointRequest', () => {
    expect(request.body.reasoning).toEqual({ effort: 'high' });
    expect(request.body.text).toEqual({ verbosity: 'low' });
    expect(request.body.safety_identifier).toBeUndefined();
+    expect(request.body.store).toBe(false);
  });

  it('builds gemini-cli native requests with project envelope and bearer headers', () => {
@@ -413,7 +413,11 @@ function ensureCodexResponsesStoreFalse(
  body: Record<string, unknown>,
  sitePlatform: string,
 ): Record<string, unknown> {
-  return body;
+  if (sitePlatform !== 'codex') return body;
+  return {
+    ...body,
+    store: false,
+  };
 }

 function convertCodexSystemRoleToDeveloper(input: unknown): unknown {
@@ -1256,7 +1260,10 @@ export function buildUpstreamEndpointRequest(input: {
      ),
      sitePlatform,
    );
-    const configuredResponsesBody = applyConfiguredPayloadRules(body);
+    const configuredResponsesBody = ensureCodexResponsesStoreFalse(
+      applyConfiguredPayloadRules(body),
+      sitePlatform,
+    );

    if (sitePlatform === 'codex') {
      if (providerProfile?.id !== 'codex') {