Merge pull request #532 from itkonen/retry-openai-responses-failures

Retry transient OpenAI Responses failures

Author: ericdallo

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Retry transient OpenAI Responses `response.failed` server errors before output, preserving structured error and request IDs.
 - Fix prompt cache invalidation warning after clearing the chat and changing model. #530
 - Fix missing line break after "Prompt stopped" message when followed by another system message.
 

src/eca/llm_api.clj

@@ -465,8 +465,7 @@
                                            (catch Exception e
                                              (logger/warn logger-tag "on-history-sanitized callback failed" {:exception (ex-message e)})))))
         emit-first-message-fn (fn [& args]
-                                (when-not @first-response-received*
-                                  (reset! first-response-received* true)
+                                (when (compare-and-set! first-response-received* false true)
                                   (apply on-first-response-received args)))
         on-message-received-wrapper (fn [& args]
                                       (apply emit-first-message-fn args)
@@ -516,6 +515,7 @@
                         (if (and (contains? #{:rate-limited :overloaded :retryable-custom :premature-stop} error-type)
                                  (< attempt max-retries)
                                  (not wait-too-long?)
+                                 (not @first-response-received*)
                                  (not (cancelled?)))
                           (let [delay-ms (if rl-wait
                                            (+ (long (:delay-ms rl-wait)) rate-limit-wait-buffer-ms)

src/eca/llm_providers/errors.clj

@@ -36,6 +36,9 @@
    #"(?i)connection refused"
    #"(?i)UnresolvedAddressException"])
 
+(def ^:private openai-transient-message-pattern
+  #"(?i)an error occurred while processing your request\.\s+you can retry your request\b")
+
 (defn ^:private matches-any-pattern? [^String text patterns]
   (when text
     (some #(re-find % text) patterns)))
@@ -65,6 +68,25 @@
 
     :else nil))
 
+(defn ^:private classify-openai-responses-error
+  [{:keys [code type message] source :error/source}]
+  (when (= :openai-responses source)
+    (cond
+      (some #{"server_error"} [code type])
+      {:error/type :overloaded}
+
+      (some #{"rate_limit_exceeded"} [code type])
+      {:error/type :rate-limited}
+
+      (or (some? code) (some? type))
+      {:error/type :unknown}
+
+      (and (string? message)
+           (re-find openai-transient-message-pattern message))
+      {:error/type :overloaded}
+
+      :else nil)))
+
 (defn ^:private classify-by-message
   "Fallback classification from unstructured error message strings
    (e.g. SSE stream errors where HTTP status is not available)."
@@ -112,7 +134,8 @@
 (defn classify-error
   "Classifies an error map into a semantic error type.
 
-   Accepts the standard on-error map shape: {:message :status :body :exception}.
+   Accepts the standard on-error map shape: {:message :status :body :exception},
+   plus optional structured provider fields such as :code, :type, and :error/source.
    Optional `retry-rules` seq of user-configured rules checked before built-in classification.
    Returns a map with :error/type — one of:
      :retryable-custom  — matched a user-configured retry rule (with optional :error/label)
@@ -126,6 +149,7 @@
    (or (when-let [pre-type (:error/type error-data)]
          {:error/type pre-type})
        (classify-by-custom-rules error-data retry-rules)
+       (classify-openai-responses-error error-data)
        (when status
          (classify-by-status-and-body error-data))
        (classify-by-message error-data)

src/eca/llm_providers/openai.clj

@@ -142,8 +142,38 @@
                  ""
                  (:content (last (:output body))))})
 
+(defn ^:private non-blank-str [value]
+  (let [value (if (coll? value) (first value) value)
+        value (some-> value str string/trim)]
+    (when-not (string/blank? value)
+      value)))
+
+(defn ^:private response-header [headers header-name]
+  (some (fn [[k value]]
+          (let [k (if (keyword? k) (name k) (str k))]
+            (when (= header-name (string/lower-case k))
+              (non-blank-str value))))
+        headers))
+
+(defn ^:private response-failed->error-data [data response-headers]
+  (let [response (:response data)
+        error (:error response)
+        request-id (some non-blank-str
+                         [(:request-id error)
+                          (:request_id error)
+                          (:request-id response)
+                          (:request_id response)
+                          (response-header response-headers "x-request-id")])]
+    (assoc-some (or error {})
+                :message (or (:message error) "OpenAI response failed")
+                :error/source :openai-responses
+                :response-id (:id response)
+                :request-id request-id
+                :headers response-headers)))
+
 (defn ^:private base-responses-request! [{:keys [rid body api-url auth-type url-relative-path api-key account-id on-error on-stream http-client extra-headers]}]
   (let [oauth? (= :auth/oauth auth-type)
+        stream? (and on-stream (not= false (:stream body)))
         url (if oauth?
               codex-url
               (join-api-url api-url (or url-relative-path responses-path)))
@@ -174,22 +204,32 @@
                                                           :body (json/generate-string body)
                                                           :throw-exceptions? false
                                                           :http-client (client/merge-with-global-http-client http-client)
-                                                          :as (if on-stream :stream :json)})]
+                                                          :as (if stream? :stream :json)})]
         (if (not= 200 status)
-          (let [body-str (if on-stream (slurp body) body)]
+          (let [body-str (if stream? (slurp body) body)]
             (logger/warn logger-tag "Unexpected response status: %s body: %s" status body-str)
             (on-error {:message (format "OpenAI response status: %s body: %s" status body-str)
                        :status status
                        :body body-str
                        :headers resp-headers}))
-          (if on-stream
-            (with-open [rdr (io/reader body)]
-              (doseq [[event data] (llm-util/event-data-seq rdr)]
-                (llm-util/log-response logger-tag rid event data)
-                (on-stream event data)))
+          (if stream?
+            (let [stream-error
+                  (with-open [rdr (io/reader body)]
+                    (loop [events (seq (llm-util/event-data-seq rdr))]
+                      (when-let [[event data] (first events)]
+                        (llm-util/log-response logger-tag rid event data)
+                        (if (= "response.failed" event)
+                          (response-failed->error-data data resp-headers)
+                          (do
+                            (on-stream event data resp-headers)
+                            (recur (next events)))))))]
+              (when stream-error
+                (on-error stream-error)))
             (do
               (llm-util/log-response logger-tag rid "response" body)
-              (response-body->result body)))))
+              (if (= "failed" (:status body))
+                (on-error (response-failed->error-data {:response body} resp-headers))
+                (response-body->result body))))))
       (catch Exception e
         (on-error {:exception e
                    :message (if (ex-data e)
@@ -327,7 +367,7 @@
         sync-result* (when-not callbacks (atom nil))
         on-stream-fn
         (if callbacks
-          (fn handle-stream [event data]
+          (fn handle-stream [event data & _]
             (case event
               ;; text
               "response.output_text.delta"
@@ -465,23 +505,15 @@
                         :on-stream handle-stream})))
                   (on-message-received {:type :finish
                                         :finish-reason (-> data :response :status)})))
-
-              "response.failed" (do
-                                  (when-let [error (-> data :response :error)]
-                                    (on-error {:message (:message error)}))
-                                  (on-message-received {:type :finish
-                                                        :finish-reason (-> data :response :status)}))
               nil))
           ;; Sync mode: collect text deltas into result atom
           (let [sb (StringBuilder.)]
-            (fn handle-sync-stream [event data]
+            (fn handle-sync-stream [event data & _]
               (case event
                 "response.output_text.delta"
                 (.append sb ^String (:delta data))
                 "response.completed"
                 (reset! sync-result* {:output-text (.toString sb)})
-                "response.failed"
-                (reset! sync-result* {:error {:message (-> data :response :error :message)}})
                 nil))))
         result (base-responses-request!
                 {:rid (llm-util/gen-rid)

test/eca/llm_api_test.clj

@@ -897,6 +897,129 @@
       (is (false? @on-error-called*))
       (is (= "hello" @received-text*)))))
 
+(deftest first-response-callback-is-atomic-test
+  (testing "concurrent output callbacks emit first response once"
+    (let [first-response-count* (atom 0)]
+      (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-message-received]}]
+                                          (let [ready (java.util.concurrent.CountDownLatch. 2)
+                                                go (java.util.concurrent.CountDownLatch. 1)
+                                                workers (mapv (fn [text]
+                                                                (future
+                                                                  (.countDown ready)
+                                                                  (.await go)
+                                                                  (on-message-received {:type :text :text text})))
+                                                              ["one" "two"])]
+                                            (.await ready)
+                                            (.countDown go)
+                                            (doseq [worker workers]
+                                              @worker)))]
+        (llm-api/sync-or-async-prompt!
+         (make-prompt-opts
+          {:on-first-response-received (fn [& _] (swap! first-response-count* inc))
+           :on-message-received identity
+           :on-error identity})))
+      (is (= 1 @first-response-count*)))))
+
+(deftest async-retry-on-structured-server-error-test
+  (testing "retries an OpenAI Responses server_error before output"
+    (let [attempt* (atom 0)
+          retry-events* (atom [])
+          received-text* (atom "")
+          on-error-called* (atom false)]
+      (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-message-received on-error]}]
+                                          (let [attempt (swap! attempt* inc)]
+                                            (if (= 1 attempt)
+                                              (on-error {:code "server_error"
+                                                         :message "Request failed"
+                                                         :error/source :openai-responses
+                                                         :response-id "resp_123"
+                                                         :request-id "req_123"})
+                                              (do
+                                                (on-message-received {:type :text :text "hello"})
+                                                (on-message-received {:type :finish :finish-reason "stop"})))))
+                    eca.llm-api/sleep-with-cancel (fn [_ cancelled?] (not (cancelled?)))]
+        (llm-api/sync-or-async-prompt!
+         (make-prompt-opts
+          {:on-retry (fn [event] (swap! retry-events* conj event))
+           :on-error (fn [_] (reset! on-error-called* true))
+           :on-message-received (fn [{:keys [type text]}]
+                                  (when (= :text type)
+                                    (swap! received-text* str text)))})))
+      (is (= 2 @attempt*))
+      (is (= :overloaded (get-in (first @retry-events*) [:classified :error/type])))
+      (is (= "resp_123" (get-in (first @retry-events*) [:error-data :response-id])))
+      (is (= "req_123" (get-in (first @retry-events*) [:error-data :request-id])))
+      (is (false? @on-error-called*))
+      (is (= "hello" @received-text*)))))
+
+(deftest sync-retry-on-structured-server-error-test
+  (testing "retries an OpenAI Responses server_error in sync mode"
+    (let [attempt* (atom 0)
+          retry-events* (atom [])
+          on-error-called* (atom false)]
+      (with-redefs [eca.llm-api/prompt! (fn [_opts]
+                                          (if (= 1 (swap! attempt* inc))
+                                            {:error {:code "server_error"
+                                                     :message "Request failed"
+                                                     :error/source :openai-responses
+                                                     :response-id "resp_sync"
+                                                     :request-id "req_sync"}}
+                                            {:output-text "success"
+                                             :usage {:input-tokens 1 :output-tokens 1}}))
+                    eca.llm-api/sleep-with-cancel (fn [_ cancelled?] (not (cancelled?)))]
+        (llm-api/sync-or-async-prompt!
+         (make-prompt-opts
+          {:stream false
+           :on-retry (fn [event] (swap! retry-events* conj event))
+           :on-error (fn [_] (reset! on-error-called* true))
+           :on-message-received identity})))
+      (is (= 2 @attempt*))
+      (is (= :overloaded (get-in (first @retry-events*) [:classified :error/type])))
+      (is (= "resp_sync" (get-in (first @retry-events*) [:error-data :response-id])))
+      (is (= "req_sync" (get-in (first @retry-events*) [:error-data :request-id])))
+      (is (false? @on-error-called*)))))
+
+(deftest async-no-retry-after-visible-output-test
+  (doseq [[label emit-output]
+          [["text output"
+            (fn [{:keys [on-message-received]}]
+              (on-message-received {:type :text :text "partial"}))]
+           ["reasoning output"
+            (fn [{:keys [on-reason]}]
+              (on-reason {:status :thinking :id "reason_1" :text "partial"}))]
+           ["tool-call output"
+            (fn [{:keys [on-prepare-tool-call]}]
+              (on-prepare-tool-call {:id "call_1" :full-name "tool" :arguments-text "{"}))]
+           ["web-search output"
+            (fn [{:keys [on-server-web-search]}]
+              (on-server-web-search {:status :started :id "search_1"}))]
+           ["image-generation output"
+            (fn [{:keys [on-server-image-generation]}]
+              (on-server-image-generation {:status :started :id "image_1"}))]]]
+    (testing (str "does not retry after " label)
+      (let [attempt* (atom 0)
+            sleep-calls* (atom 0)
+            errors* (atom [])]
+        (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-error] :as callbacks}]
+                                            (swap! attempt* inc)
+                                            (emit-output callbacks)
+                                            (on-error {:code "server_error"
+                                                       :message "Request failed"
+                                                       :error/source :openai-responses}))
+                      eca.llm-api/sleep-with-cancel (fn [_ _]
+                                                     (swap! sleep-calls* inc)
+                                                     true)]
+          (llm-api/sync-or-async-prompt!
+           (make-prompt-opts
+            {:on-error (fn [error] (swap! errors* conj error))
+             :on-message-received identity})))
+        (is (= 1 @attempt*))
+        (is (zero? @sleep-calls*))
+        (is (= [{:code "server_error"
+                 :message "Request failed"
+                 :error/source :openai-responses}]
+               @errors*))))))
+
 (deftest async-no-retry-on-context-overflow-test
   (testing "does not retry on context overflow"
     (let [attempt* (atom 0)

test/eca/llm_providers/errors_test.clj

@@ -94,6 +94,62 @@
              :body "Bad Gateway"
              :message "LLM response status: 502 body: Bad Gateway"})))))
 
+(deftest classify-openai-response-failed-test
+  (let [generic-message (str "An error occurred while processing your request. "
+                             "You can retry your request, or contact support if the error persists.")
+        responses-error {:error/source :openai-responses}]
+    (testing "structured server_error code is overloaded"
+      (is (= {:error/type :overloaded}
+             (llm-providers.errors/classify-error
+              (assoc responses-error
+                     :code "server_error"
+                     :message "Request failed")))))
+
+    (testing "structured server_error type is overloaded"
+      (is (= {:error/type :overloaded}
+             (llm-providers.errors/classify-error
+              (assoc responses-error
+                     :type "server_error"
+                     :message "Request failed")))))
+
+    (testing "structured rate_limit_exceeded code is rate limited"
+      (is (= {:error/type :rate-limited}
+             (llm-providers.errors/classify-error
+              (assoc responses-error
+                     :code "rate_limit_exceeded"
+                     :message "Request failed")))))
+
+    (testing "generic transient message is a fallback without structured fields"
+      (is (= {:error/type :overloaded}
+             (llm-providers.errors/classify-error
+              (assoc responses-error :message generic-message)))))
+
+    (testing "generic transient message is not applied to unmarked errors"
+      (is (= {:error/type :unknown}
+             (llm-providers.errors/classify-error
+              {:message generic-message}))))
+
+    (testing "unknown structured fields are authoritative over retry-looking messages"
+      (doseq [error-data [{:code "invalid_prompt" :message generic-message}
+                          {:code "invalid_prompt" :message "Internal server error"}
+                          {:type "invalid_request_error" :message "Rate limit exceeded"}
+                          {:code "invalid_prompt"
+                           :message "Request failed"
+                           :exception (Exception. "Connection error")}]]
+        (is (= {:error/type :unknown}
+               (llm-providers.errors/classify-error
+                (merge responses-error error-data))))))
+
+    (testing "custom retry rules still take priority over structured fields"
+      (is (= {:error/type :retryable-custom
+              :error/label "Temporary proxy failure"}
+             (llm-providers.errors/classify-error
+              (assoc responses-error
+                     :code "invalid_prompt"
+                     :message generic-message)
+              [{:errorPattern "processing your request"
+                :label "Temporary proxy failure"}]))))))
+
 (deftest classify-error-unknown-test
   (testing "500 is classified as overloaded"
     (is (= {:error/type :overloaded}