Keep Codex stream-disconnect classification on the terminal error

plugins/provider-codex/src/delta-translation.ts

@@ -78,6 +78,18 @@ export interface CodexInjectedTool {
   presentation?: DeltaPresentation;
 }
 
+/**
+ * The structured classification Codex attached to a retried failure, keyed by
+ * `threadId\0turnId`. Codex labels every reconnect attempt with the specific
+ * error info (for example `responseStreamDisconnected`) but downgrades the
+ * terminal error for the same failure to `other` once its retry budget is
+ * exhausted, so the bridge carries the retry-time classification forward.
+ */
+interface CodexRetryErrorContext {
+  errorInfo: CodexErrorInfo;
+  failureText: string;
+}
+
 interface CodexEventTranslationState {
   rateLimits: CodexRateLimitSnapshot | null;
   /**
@@ -86,10 +98,15 @@ interface CodexEventTranslationState {
    * says; every other dynamic tool call is codex's own.
    */
   injectedToolsByName: Map<string, CodexInjectedTool>;
+  retryErrorsByTurnKey: Map<string, CodexRetryErrorContext>;
 }
 
 export function createCodexEventTranslationState(): CodexEventTranslationState {
-  return { rateLimits: null, injectedToolsByName: new Map() };
+  return {
+    rateLimits: null,
+    injectedToolsByName: new Map(),
+    retryErrorsByTurnKey: new Map(),
+  };
 }
 
 export function setCodexInjectedTools(
@@ -245,7 +262,7 @@ function normalizeCodexRateLimits(
 }
 
 type CodexErrorEvent = Extract<CodexHandledEvent, { method: "error" }>;
-type CodexErrorPayload = CodexErrorEvent["params"]["error"];
+type CodexErrorParams = CodexErrorEvent["params"];
 
 type CodexItemTranslationResult =
   | {
@@ -349,9 +366,8 @@ function getProviderErrorCategory(
 }
 
 function toProviderErrorInfo(
-  error: CodexErrorPayload,
+  errorInfo: CodexErrorInfo | null | undefined,
 ): ProviderErrorInfo | null {
-  const errorInfo = error.codexErrorInfo;
   if (!errorInfo) {
     return null;
   }
@@ -362,6 +378,63 @@ function toProviderErrorInfo(
   };
 }
 
+function codexTurnKey(scope: { threadId: string; turnId?: string }): string {
+  return `${scope.threadId}\0${scope.turnId ?? ""}`;
+}
+
+function takeCodexRetryError(
+  state: CodexEventTranslationState,
+  scope: { threadId: string; turnId?: string },
+): CodexRetryErrorContext | undefined {
+  const key = codexTurnKey(scope);
+  const retryError = state.retryErrorsByTurnKey.get(key);
+  state.retryErrorsByTurnKey.delete(key);
+  return retryError;
+}
+
+export function clearCodexEventTranslationThreadState(
+  state: CodexEventTranslationState,
+  threadId: string,
+): void {
+  const prefix = codexTurnKey({ threadId });
+  for (const key of state.retryErrorsByTurnKey.keys()) {
+    if (key.startsWith(prefix)) {
+      state.retryErrorsByTurnKey.delete(key);
+    }
+  }
+}
+
+/**
+ * Codex reports the underlying failure in `additionalDetails` while it is
+ * retrying ("Reconnecting... n/m"), then moves the same text to `message` and
+ * downgrades `codexErrorInfo` to `other` on the terminal event. Correlate the
+ * two by failure text, scoped to the turn, so the terminal error keeps the
+ * structured classification without interpreting provider prose.
+ */
+function resolveCodexErrorInfo(
+  state: CodexEventTranslationState,
+  params: CodexErrorParams,
+): CodexErrorInfo | null | undefined {
+  const errorInfo = params.error.codexErrorInfo;
+  const failureText = params.error.additionalDetails ?? params.error.message;
+  if (params.willRetry === true) {
+    if (errorInfo && errorInfo !== "other") {
+      state.retryErrorsByTurnKey.set(codexTurnKey(params), {
+        errorInfo,
+        failureText,
+      });
+    }
+    return errorInfo;
+  }
+  if (params.willRetry !== false) {
+    return errorInfo;
+  }
+  const retryError = takeCodexRetryError(state, params);
+  return errorInfo === "other" && retryError?.failureText === failureText
+    ? retryError.errorInfo
+    : errorInfo;
+}
+
 function toRawEvent(rawEvent: JsonRpcMessage): ProviderRawEvent {
   const parsed = providerRawEventSchema.safeParse(rawEvent);
   if (parsed.success) {
@@ -903,6 +976,10 @@ export function translateCodexEventToDeltas(
         { kind: "turn.open", providerTurnId: handledEvent.params.turn.id },
       ];
     case "turn/completed": {
+      takeCodexRetryError(state, {
+        threadId: handledEvent.params.threadId,
+        turnId: handledEvent.params.turn.id,
+      });
       const status = toTurnStatus(handledEvent.params.turn.status);
       return [
         {
@@ -1164,7 +1241,9 @@ export function translateCodexEventToDeltas(
         },
       ];
     case "error": {
-      const errorInfo = toProviderErrorInfo(handledEvent.params.error);
+      const errorInfo = toProviderErrorInfo(
+        resolveCodexErrorInfo(state, handledEvent.params),
+      );
       return [
         {
           kind: "provider.error",

plugins/provider-codex/src/translator.test.ts

@@ -1214,6 +1214,149 @@ describe("codex subagent activity correlation", () => {
   });
 });
 
+// ---------------------------------------------------------------------------
+// Terminal retry-error classification (#1840)
+// ---------------------------------------------------------------------------
+
+const STREAM_DISCONNECT_MESSAGE =
+  "stream disconnected before completion: error sending request for url (https://chatgpt.com/backend-api/codex/responses)";
+
+const STREAM_DISCONNECTED_ERROR_INFO = {
+  category: "stream-disconnected",
+  providerCode: "responseStreamDisconnected",
+  httpStatusCode: 502,
+};
+
+const UNKNOWN_ERROR_INFO = {
+  category: "unknown",
+  providerCode: "other",
+  httpStatusCode: null,
+};
+
+function codexReconnectError(turnId: string) {
+  return codexEvent("error", {
+    threadId: "t1",
+    turnId,
+    error: {
+      message: "Reconnecting... 5/5",
+      codexErrorInfo: { responseStreamDisconnected: { httpStatusCode: 502 } },
+      additionalDetails: STREAM_DISCONNECT_MESSAGE,
+    },
+    willRetry: true,
+  });
+}
+
+function codexTerminalOtherError(turnId: string, message: string) {
+  return codexEvent("error", {
+    threadId: "t1",
+    turnId,
+    error: { message, codexErrorInfo: "other", additionalDetails: null },
+    willRetry: false,
+  });
+}
+
+describe("codex terminal retry-error classification", () => {
+  // Codex labels each reconnect attempt `responseStreamDisconnected`, then
+  // reports the terminal failure for the same stream error as `other` once
+  // its retry budget is exhausted (codex-rs maps `CodexErrorDetails::Stream`
+  // to `CodexErrorInfo::Other`). The translator keeps the structured
+  // classification for the terminal row without parsing provider prose.
+  it("carries the retry classification into the degraded terminal error", () => {
+    const harness = createHarness();
+    harness.translate(codexReconnectError("turn-1"));
+
+    expect(
+      harness.translate(
+        codexTerminalOtherError("turn-1", STREAM_DISCONNECT_MESSAGE),
+      ),
+    ).toContainEqual(
+      expect.objectContaining({
+        type: "provider/error",
+        scope: turnScope(harness.turnId("turn-1")),
+        willRetry: false,
+        detail: STREAM_DISCONNECT_MESSAGE,
+        errorInfo: STREAM_DISCONNECTED_ERROR_INFO,
+      }),
+    );
+
+    // The context is consumed by the terminal event: a repeat stays `other`.
+    expect(
+      harness.translate(
+        codexTerminalOtherError("turn-1", STREAM_DISCONNECT_MESSAGE),
+      ),
+    ).toContainEqual(
+      expect.objectContaining({
+        type: "provider/error",
+        errorInfo: UNKNOWN_ERROR_INFO,
+      }),
+    );
+  });
+
+  it("does not relabel an unrelated terminal error after a reconnect", () => {
+    const harness = createHarness();
+    harness.translate(codexReconnectError("turn-1"));
+
+    expect(
+      harness.translate(codexTerminalOtherError("turn-1", "request failed")),
+    ).toContainEqual(
+      expect.objectContaining({
+        type: "provider/error",
+        errorInfo: UNKNOWN_ERROR_INFO,
+      }),
+    );
+  });
+
+  it("scopes the retry context to the turn and drops it on turn/completed", () => {
+    const harness = createHarness();
+    harness.translate(codexReconnectError("turn-1"));
+
+    expect(
+      harness.translate(
+        codexTerminalOtherError("turn-2", STREAM_DISCONNECT_MESSAGE),
+      ),
+    ).toContainEqual(
+      expect.objectContaining({
+        type: "provider/error",
+        errorInfo: UNKNOWN_ERROR_INFO,
+      }),
+    );
+
+    harness.translate(
+      codexEvent("turn/completed", {
+        threadId: "t1",
+        turn: codexTurn({ id: "turn-1", status: "completed", error: null }),
+      }),
+    );
+    expect(
+      harness.translate(
+        codexTerminalOtherError("turn-1", STREAM_DISCONNECT_MESSAGE),
+      ),
+    ).toContainEqual(
+      expect.objectContaining({
+        type: "provider/error",
+        errorInfo: UNKNOWN_ERROR_INFO,
+      }),
+    );
+  });
+
+  it("drops the retry context when the codex thread closes", () => {
+    const harness = createHarness();
+    harness.translate(codexReconnectError("turn-1"));
+    harness.translate(codexEvent("thread/closed", { threadId: "t1" }));
+
+    expect(
+      harness.translate(
+        codexTerminalOtherError("turn-1", STREAM_DISCONNECT_MESSAGE),
+      ),
+    ).toContainEqual(
+      expect.objectContaining({
+        type: "provider/error",
+        errorInfo: UNKNOWN_ERROR_INFO,
+      }),
+    );
+  });
+});
+
 // ---------------------------------------------------------------------------
 // Accepted-turn correlation via turn/started (68d80092f, current semantics)
 // ---------------------------------------------------------------------------

plugins/provider-codex/src/translator.ts

@@ -22,6 +22,7 @@ import {
 import { z } from "zod";
 import {
   applyCodexRateLimitUpdate,
+  clearCodexEventTranslationThreadState,
   createCodexEventTranslationState,
   setCodexInjectedTools,
   translateCodexEventToDeltas,
@@ -568,6 +569,10 @@ export function createCodexEventTranslator(
     const closed = clearExitedChildThreadState({
       providerThreadId: paramsResult.data.threadId,
     });
+    clearCodexEventTranslationThreadState(
+      eventTranslationState,
+      paramsResult.data.threadId,
+    );
     clearGitWritableRootsByProviderThreadId({
       providerThreadId: paramsResult.data.threadId,
     });