Skip to content

Retry transient OpenAI Responses failures #532

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ericdallo merged 1 commit into from
Jul 15, 2026
Merged

Retry transient OpenAI Responses failures #532

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
4 changes: 2 additions & 2 deletions src/eca/llm_api.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
26 changes: 25 additions & 1 deletion src/eca/llm_providers/errors.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)))
Expand Down Expand Up @@ -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)."
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down
68 changes: 50 additions & 18 deletions src/eca/llm_providers/openai.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)))
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down
123 changes: 123 additions & 0 deletions test/eca/llm_api_test.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
56 changes: 56 additions & 0 deletions test/eca/llm_providers/errors_test.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down
Loading
Loading