[codex] fix codex responses continuation across channel drift (#404)

* fix codex responses continuation across channel drift * fix: clear codex responses continuation fallback on scoped reset * fix: harden codex continuation store keying
2026-04-03 11:00:53 +08:00
parent b9b553d772
commit af5b62fd8b
3 changed files with 288 additions and 6 deletions
@@ -1,6 +1,7 @@
 import { describe, expect, it } from 'vitest';
 import {
  buildCodexSessionResponseStoreKey,
+  clearCodexSessionResponseId,
  getCodexSessionResponseId,
  resetCodexSessionResponseStore,
  setCodexSessionResponseId,
@@ -45,4 +46,84 @@ describe('codexSessionResponseStore', () => {

    resetCodexSessionResponseStore();
  });
+
+  it('falls back to the bare downstream session id across channel scope drift', () => {
+    resetCodexSessionResponseStore();
+
+    const originalScopedKey = buildCodexSessionResponseStoreKey({
+      sessionId: 'session-drift',
+      siteId: 10,
+      accountId: 20,
+      channelId: 30,
+    });
+    const driftedScopedKey = buildCodexSessionResponseStoreKey({
+      sessionId: 'session-drift',
+      siteId: 10,
+      accountId: 21,
+      channelId: 31,
+    });
+
+    setCodexSessionResponseId(originalScopedKey, 'resp-drift');
+
+    expect(getCodexSessionResponseId(originalScopedKey)).toBe('resp-drift');
+    expect(getCodexSessionResponseId(driftedScopedKey)).toBe('resp-drift');
+
+    resetCodexSessionResponseStore();
+  });
+
+  it('does not collapse distinct downstream sessions when scoped session ids contain delimiters', () => {
+    resetCodexSessionResponseStore();
+
+    const delimitedScopedKey = buildCodexSessionResponseStoreKey({
+      sessionId: 'session-delimited|tail',
+      siteId: 10,
+      accountId: 20,
+      channelId: 30,
+    });
+    const delimitedDriftedScopedKey = buildCodexSessionResponseStoreKey({
+      sessionId: 'session-delimited|tail',
+      siteId: 10,
+      accountId: 21,
+      channelId: 31,
+    });
+    const plainScopedKey = buildCodexSessionResponseStoreKey({
+      sessionId: 'session-delimited',
+      siteId: 10,
+      accountId: 22,
+      channelId: 32,
+    });
+
+    setCodexSessionResponseId(delimitedScopedKey, 'resp-delimited');
+    setCodexSessionResponseId(plainScopedKey, 'resp-plain');
+
+    expect(getCodexSessionResponseId(delimitedDriftedScopedKey)).toBe('resp-delimited');
+    expect(getCodexSessionResponseId(plainScopedKey)).toBe('resp-plain');
+
+    resetCodexSessionResponseStore();
+  });
+
+  it('clears the bare downstream session fallback when removing a scoped continuation id', () => {
+    resetCodexSessionResponseStore();
+
+    const originalScopedKey = buildCodexSessionResponseStoreKey({
+      sessionId: 'session-clear',
+      siteId: 10,
+      accountId: 20,
+      channelId: 30,
+    });
+    const driftedScopedKey = buildCodexSessionResponseStoreKey({
+      sessionId: 'session-clear',
+      siteId: 10,
+      accountId: 21,
+      channelId: 31,
+    });
+
+    setCodexSessionResponseId(originalScopedKey, 'resp-clear');
+    clearCodexSessionResponseId(originalScopedKey);
+
+    expect(getCodexSessionResponseId(originalScopedKey)).toBeNull();
+    expect(getCodexSessionResponseId(driftedScopedKey)).toBeNull();
+
+    resetCodexSessionResponseStore();
+  });
 });
@@ -2,6 +2,61 @@ const MAX_CODEX_SESSION_RESPONSE_IDS = 10_000;

 const codexSessionResponseIds = new Map<string, string>();

+const SCOPED_SESSION_SEGMENT_PREFIX = 'session:';
+const SCOPED_STORE_KEY_SEGMENT_PATTERN = /^(site|account|channel):\d+$/;
+
+function buildScopedSessionSegment(sessionId: string): string {
+  return `${SCOPED_SESSION_SEGMENT_PREFIX}${encodeURIComponent(sessionId)}`;
+}
+
+function extractScopedSessionSegment(sessionId: string): string {
+  const normalizedSessionId = normalizeSessionId(sessionId);
+  if (!normalizedSessionId) return '';
+
+  if (normalizedSessionId.startsWith(SCOPED_SESSION_SEGMENT_PREFIX)) {
+    return normalizedSessionId;
+  }
+
+  const scopedSegments = normalizedSessionId
+    .split('|')
+    .map((segment) => segment.trim())
+    .filter(Boolean);
+  if (scopedSegments.length <= 1) return '';
+
+  const sessionSegment = scopedSegments[scopedSegments.length - 1];
+  if (!sessionSegment.startsWith(SCOPED_SESSION_SEGMENT_PREFIX)) {
+    return '';
+  }
+  const scopeSegments = scopedSegments.slice(0, -1);
+  if (!scopeSegments.every((segment) => SCOPED_STORE_KEY_SEGMENT_PATTERN.test(segment))) {
+    return '';
+  }
+  return sessionSegment;
+}
+
+function getBareSessionStoreKey(sessionId: string): string {
+  return extractScopedSessionSegment(sessionId);
+}
+
+function getFallbackSessionStoreKeys(sessionId: string): string[] {
+  const normalizedSessionId = normalizeSessionId(sessionId);
+  if (!normalizedSessionId) return [];
+
+  const bareSessionKey = getBareSessionStoreKey(normalizedSessionId);
+  if (!bareSessionKey) return [];
+  if (normalizedSessionId === bareSessionKey) return [];
+  return [bareSessionKey];
+}
+
+function getSessionStoreKeys(sessionId: string): string[] {
+  const normalizedSessionId = normalizeSessionId(sessionId);
+  if (!normalizedSessionId) return [];
+  return [
+    normalizedSessionId,
+    ...getFallbackSessionStoreKeys(normalizedSessionId),
+  ];
+}
+
 function normalizeSessionId(sessionId: string): string {
  return sessionId.trim();
 }
@@ -18,7 +73,7 @@ export function buildCodexSessionResponseStoreKey(input: {
    Number.isFinite(input.siteId as number) && Number(input.siteId) > 0 ? `site:${Math.trunc(Number(input.siteId))}` : '',
    Number.isFinite(input.accountId as number) && Number(input.accountId) > 0 ? `account:${Math.trunc(Number(input.accountId))}` : '',
    Number.isFinite(input.channelId as number) && Number(input.channelId) > 0 ? `channel:${Math.trunc(Number(input.channelId))}` : '',
-    `session:${normalizedSessionId}`,
+    buildScopedSessionSegment(normalizedSessionId),
  ].filter(Boolean);
  return parts.join('|');
 }
@@ -26,17 +81,32 @@ export function buildCodexSessionResponseStoreKey(input: {
 export function getCodexSessionResponseId(sessionId: string): string | null {
  const normalized = normalizeSessionId(sessionId);
  if (!normalized) return null;
-  return codexSessionResponseIds.get(normalized) ?? null;
+
+  const direct = codexSessionResponseIds.get(normalized);
+  if (direct) return direct;
+
+  for (const fallbackKey of getFallbackSessionStoreKeys(normalized)) {
+    const fallback = codexSessionResponseIds.get(fallbackKey);
+    if (fallback) return fallback;
+  }
+
+  return null;
 }

 export function setCodexSessionResponseId(sessionId: string, responseId: string): void {
  const normalizedSessionId = normalizeSessionId(sessionId);
  const normalizedResponseId = responseId.trim();
  if (!normalizedSessionId || !normalizedResponseId) return;
-  if (codexSessionResponseIds.has(normalizedSessionId)) {
-    codexSessionResponseIds.delete(normalizedSessionId);
+
+  const keysToWrite = new Set<string>(getSessionStoreKeys(normalizedSessionId));
+
+  for (const key of keysToWrite) {
+    if (codexSessionResponseIds.has(key)) {
+      codexSessionResponseIds.delete(key);
+    }
+    codexSessionResponseIds.set(key, normalizedResponseId);
  }
-  codexSessionResponseIds.set(normalizedSessionId, normalizedResponseId);
+
  while (codexSessionResponseIds.size > MAX_CODEX_SESSION_RESPONSE_IDS) {
    const oldestKey = codexSessionResponseIds.keys().next().value;
    if (!oldestKey) break;
@@ -47,7 +117,9 @@ export function setCodexSessionResponseId(sessionId: string, responseId: string)
 export function clearCodexSessionResponseId(sessionId: string): void {
  const normalized = normalizeSessionId(sessionId);
  if (!normalized) return;
-  codexSessionResponseIds.delete(normalized);
+  for (const key of getSessionStoreKeys(normalized)) {
+    codexSessionResponseIds.delete(key);
+  }
 }

 export function resetCodexSessionResponseStore(): void {
@@ -571,6 +571,135 @@ describe('responses proxy codex oauth refresh', () => {
    ]);
  });

+  it('reuses previous_response_id for codex tool-output follow-up turns after channel/account drift on the same downstream session', async () => {
+    const firstSelected = {
+      channel: { id: 11, routeId: 22 },
+      site: { name: 'codex-site', url: 'https://chatgpt.com/backend-api/codex', platform: 'codex' },
+      account: {
+        id: 33,
+        username: 'codex-user-a@example.com',
+        extraConfig: JSON.stringify({
+          credentialMode: 'session',
+          oauth: {
+            provider: 'codex',
+            accountId: 'chatgpt-account-123',
+            email: 'codex-user-a@example.com',
+            planType: 'plus',
+          },
+        }),
+      },
+      tokenName: 'default-a',
+      tokenValue: 'expired-access-token-a',
+      actualModel: 'gpt-5.2-codex',
+    };
+    const secondSelected = {
+      channel: { id: 12, routeId: 23 },
+      site: { name: 'codex-site', url: 'https://chatgpt.com/backend-api/codex', platform: 'codex' },
+      account: {
+        id: 34,
+        username: 'codex-user-b@example.com',
+        extraConfig: JSON.stringify({
+          credentialMode: 'session',
+          oauth: {
+            provider: 'codex',
+            accountId: 'chatgpt-account-456',
+            email: 'codex-user-b@example.com',
+            planType: 'plus',
+          },
+        }),
+      },
+      tokenName: 'default-b',
+      tokenValue: 'expired-access-token-b',
+      actualModel: 'gpt-5.2-codex',
+    };
+
+    selectChannelMock
+      .mockReset()
+      .mockReturnValueOnce(firstSelected)
+      .mockReturnValueOnce(secondSelected);
+    selectPreferredChannelMock
+      .mockReset()
+      .mockResolvedValueOnce(null)
+      .mockResolvedValueOnce(null);
+
+    fetchMock
+      .mockResolvedValueOnce(new Response(JSON.stringify({
+        id: 'resp_codex_prev_drift_1',
+        object: 'response',
+        model: 'gpt-5.2-codex',
+        status: 'completed',
+        output_text: 'tool call issued',
+        usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 },
+      }), {
+        status: 200,
+        headers: { 'content-type': 'application/json' },
+      }))
+      .mockResolvedValueOnce(new Response(JSON.stringify({
+        id: 'resp_codex_prev_drift_2',
+        object: 'response',
+        model: 'gpt-5.2-codex',
+        status: 'completed',
+        output_text: 'tool result accepted after drift',
+        usage: { input_tokens: 5, output_tokens: 2, total_tokens: 7 },
+      }), {
+        status: 200,
+        headers: { 'content-type': 'application/json' },
+      }));
+
+    const headers = {
+      session_id: 'session-http-prev-drift',
+      'user-agent': 'CodexClient/1.0',
+    };
+
+    const firstResponse = await app.inject({
+      method: 'POST',
+      url: '/v1/responses',
+      headers,
+      payload: {
+        model: 'gpt-5.2-codex',
+        input: 'start codex tool flow',
+      },
+    });
+    expect(firstResponse.statusCode).toBe(200);
+
+    const secondResponse = await app.inject({
+      method: 'POST',
+      url: '/v1/responses',
+      headers,
+      payload: {
+        model: 'gpt-5.2-codex',
+        input: [
+          {
+            id: 'tool_out_drift_1',
+            type: 'function_call_output',
+            call_id: 'call_1',
+            output: '{"ok":true}',
+          },
+        ],
+      },
+    });
+
+    expect(secondResponse.statusCode).toBe(200);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(selectChannelMock).toHaveBeenCalledTimes(2);
+    expect(selectPreferredChannelMock).toHaveBeenCalledTimes(2);
+
+    const [, firstOptions] = fetchMock.mock.calls[0] as [string, any];
+    const [, secondOptions] = fetchMock.mock.calls[1] as [string, any];
+    const secondBody = JSON.parse(secondOptions.body);
+    expect(firstOptions.headers['Chatgpt-Account-Id'] || firstOptions.headers['chatgpt-account-id']).toBe('chatgpt-account-123');
+    expect(secondOptions.headers['Chatgpt-Account-Id'] || secondOptions.headers['chatgpt-account-id']).toBe('chatgpt-account-456');
+    expect(secondBody.previous_response_id).toBe('resp_codex_prev_drift_1');
+    expect(secondBody.input).toEqual([
+      {
+        id: 'tool_out_drift_1',
+        type: 'function_call_output',
+        call_id: 'call_1',
+        output: '{"ok":true}',
+      },
+    ]);
+  });
+
  it('drops stale previous_response_id and retries codex responses requests once', async () => {
    fetchMock
      .mockResolvedValueOnce(new Response(JSON.stringify({