fix(codex): retry server overloaded turns

agent/codex/session.go

@@ -408,14 +408,17 @@ func (cs *codexSession) handleEvent(raw map[string]any) {
 
 	case "turn.failed":
 		errMsg := ""
+		errInfo := ""
 		if errObj, ok := raw["error"].(map[string]any); ok {
 			errMsg, _ = errObj["message"].(string)
+			errInfo, _ = errObj["codex_error_info"].(string)
 		}
 		if errMsg == "" {
 			errMsg = "turn failed (no details)"
 		}
-		slog.Warn("codexSession: turn failed", "error", errMsg)
-		evt := core.Event{Type: core.EventError, Error: fmt.Errorf("%s", errMsg)}
+		errKind := codexErrorKind(errInfo, errMsg)
+		slog.Warn("codexSession: turn failed", "error", errMsg, "codex_error_info", errInfo, "error_kind", errKind)
+		evt := core.Event{Type: core.EventError, Error: fmt.Errorf("%s", errMsg), ErrorKind: errKind}
 		select {
 		case cs.events <- evt:
 		case <-cs.ctx.Done():
@@ -435,6 +438,36 @@ func (cs *codexSession) handleEvent(raw map[string]any) {
 	}
 }
 
+func codexErrorKind(errorInfo string, message string) core.ErrorKind {
+	switch strings.TrimSpace(errorInfo) {
+	case "server_overloaded":
+		return core.ErrorKindOverloaded
+	case "rate_limit_exceeded", "rate_limit":
+		return core.ErrorKindRateLimit
+	}
+
+	msg := strings.ToLower(message)
+	if strings.Contains(msg, "at capacity") || strings.Contains(msg, "overloaded") {
+		return core.ErrorKindOverloaded
+	}
+	if strings.Contains(msg, "rate limit") || strings.Contains(msg, "rate_limit") {
+		return core.ErrorKindRateLimit
+	}
+	if strings.Contains(msg, "stream disconnected before completion") ||
+		strings.Contains(msg, "stream closed before response.completed") ||
+		strings.Contains(msg, "you can retry your request") ||
+		strings.Contains(msg, "processing your request") ||
+		strings.Contains(msg, "unexpected status 502") ||
+		strings.Contains(msg, "unexpected status 503") ||
+		strings.Contains(msg, "unexpected status 504") ||
+		strings.Contains(msg, "bad gateway") ||
+		strings.Contains(msg, "service unavailable") ||
+		strings.Contains(msg, "gateway timeout") {
+		return core.ErrorKindOverloaded
+	}
+	return core.ErrorKindUnknown
+}
+
 // flushPendingAsThinking emits all buffered agent_messages as EventThinking.
 func (cs *codexSession) flushPendingAsThinking() {
 	if cs.ctx.Err() != nil {

agent/codex/session_test.go

@@ -609,7 +609,11 @@ func TestSend_HandlesLargeJSONLines(t *testing.T) {
 	if err != nil {
 		t.Fatalf("newCodexSession: %v", err)
 	}
-	defer cs.Close()
+	defer func() {
+		if err := cs.Close(); err != nil {
+			t.Fatalf("Close: %v", err)
+		}
+	}()
 
 	if err := cs.Send("hello", "", nil, nil); err != nil {
 		t.Fatalf("Send: %v", err)
@@ -644,6 +648,93 @@ func TestSend_HandlesLargeJSONLines(t *testing.T) {
 	}
 }
 
+func TestSend_TurnFailedServerOverloadedIsRetriable(t *testing.T) {
+	workDir := t.TempDir()
+	binDir := filepath.Join(workDir, "bin")
+	if err := os.MkdirAll(binDir, 0o755); err != nil {
+		t.Fatalf("mkdir bin: %v", err)
+	}
+
+	shellScript := "#!/bin/sh\n" +
+		"printf '%s\\n' '{\"type\":\"thread.started\",\"thread_id\":\"thread-overloaded\"}'\n" +
+		"printf '%s\\n' '{\"type\":\"turn.failed\",\"error\":{\"message\":\"Selected model is at capacity. Please try a different model.\",\"codex_error_info\":\"server_overloaded\"}}'\n"
+	powershellScript := "[Console]::Out.WriteLine('{\"type\":\"thread.started\",\"thread_id\":\"thread-overloaded\"}')\n" +
+		"[Console]::Out.WriteLine('{\"type\":\"turn.failed\",\"error\":{\"message\":\"Selected model is at capacity. Please try a different model.\",\"codex_error_info\":\"server_overloaded\"}}')\n"
+	writeFakeCodexScript(t, binDir, shellScript, powershellScript)
+
+	t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
+
+	cs, err := newCodexSession(context.Background(), "codex", nil, workDir, "", "", "", "", "", nil, "", "", "")
+	if err != nil {
+		t.Fatalf("newCodexSession: %v", err)
+	}
+	defer func() {
+		if err := cs.Close(); err != nil {
+			t.Fatalf("Close: %v", err)
+		}
+	}()
+
+	if err := cs.Send("hello", "", nil, nil); err != nil {
+		t.Fatalf("Send: %v", err)
+	}
+
+	timeout := time.After(5 * time.Second)
+	for {
+		select {
+		case evt := <-cs.Events():
+			if evt.Type != core.EventError {
+				continue
+			}
+			if evt.ErrorKind != core.ErrorKindOverloaded {
+				t.Fatalf("ErrorKind = %q, want %q", evt.ErrorKind, core.ErrorKindOverloaded)
+			}
+			if evt.Error == nil || !strings.Contains(evt.Error.Error(), "Selected model is at capacity") {
+				t.Fatalf("Error = %v, want capacity message", evt.Error)
+			}
+			return
+		case <-timeout:
+			t.Fatal("timed out waiting for retriable error event")
+		}
+	}
+}
+
+func TestCodexErrorKind_RetriableStreamFailures(t *testing.T) {
+	tests := []struct {
+		name    string
+		message string
+		want    core.ErrorKind
+	}{
+		{
+			name:    "generic processing error with retry hint",
+			message: "stream disconnected before completion: An error occurred while processing your request. You can retry your request, or contact us through our help center at help.openai.com if the error persists. Please include the request ID 0a043d4a-c04a-4ace-becd-deb4cc1f5dd2 in your message.",
+			want:    core.ErrorKindOverloaded,
+		},
+		{
+			name:    "stream closed before response completed",
+			message: "stream disconnected before completion: stream closed before response.completed",
+			want:    core.ErrorKindOverloaded,
+		},
+		{
+			name:    "gateway error",
+			message: "unexpected status 502 Bad Gateway, url: https://aiapi.uu.cc/v1/responses",
+			want:    core.ErrorKindOverloaded,
+		},
+		{
+			name:    "non transient error remains unknown",
+			message: "authentication failed: invalid api key",
+			want:    core.ErrorKindUnknown,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := codexErrorKind("", tc.message); got != tc.want {
+				t.Fatalf("codexErrorKind() = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
 func TestWaitForArgsFile_WaitsForNonEmptyContent(t *testing.T) {
 	workDir := t.TempDir()
 	argsFile := filepath.Join(workDir, "args.txt")