diff --git a/CHANGELOG.md b/CHANGELOG.md index 17f8b96cc..7069f2bb2 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/eca/llm_api.clj b/src/eca/llm_api.clj index caa87c660..f7181f7ab 100644 --- a/src/eca/llm_api.clj +++ b/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) diff --git a/src/eca/llm_providers/errors.clj b/src/eca/llm_providers/errors.clj index cf0956fd1..ac40498f1 100644 --- a/src/eca/llm_providers/errors.clj +++ b/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) diff --git a/src/eca/llm_providers/openai.clj b/src/eca/llm_providers/openai.clj index e05776632..72b99e2b2 100644 --- a/src/eca/llm_providers/openai.clj +++ b/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) diff --git a/test/eca/llm_api_test.clj b/test/eca/llm_api_test.clj index ca08df4fb..97d68475b 100644 --- a/test/eca/llm_api_test.clj +++ b/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) diff --git a/test/eca/llm_providers/errors_test.clj b/test/eca/llm_providers/errors_test.clj index 88765ac9b..cdbe88138 100644 --- a/test/eca/llm_providers/errors_test.clj +++ b/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} diff --git a/test/eca/llm_providers/openai_test.clj b/test/eca/llm_providers/openai_test.clj index fded20800..a8540f302 100644 --- a/test/eca/llm_providers/openai_test.clj +++ b/test/eca/llm_providers/openai_test.clj @@ -366,6 +366,83 @@ :message "OpenAI response status: 401 body: unauthorized"}} result)))))) +(deftest create-response-failed-event-test + (let [failed-response {:id "resp_123" + :status "failed" + :error {:code "server_error" + :type "server_error" + :message "An error occurred while processing your request. You can retry your request." + :provider-detail "preserved"}} + response-headers {"X-Request-ID" ["req_123"]} + expected-error {:code "server_error" + :type "server_error" + :message "An error occurred while processing your request. You can retry your request." + :provider-detail "preserved" + :error/source :openai-responses + :response-id "resp_123" + :request-id "req_123" + :headers response-headers}] + (testing "streaming failure closes the stream before error handling and ignores trailing events" + (let [closed?* (atom false) + closed-at-error?* (atom false) + errors* (atom []) + messages* (atom []) + tool-prepares* (atom []) + stream-text (str "event: response.failed\n" + "data: {\"type\":\"response.failed\",\"response\":{\"id\":\"resp_123\",\"status\":\"failed\",\"error\":{\"code\":\"server_error\",\"type\":\"server_error\",\"message\":\"An error occurred while processing your request. You can retry your request.\",\"provider-detail\":\"preserved\"}}}\n\n" + "event: response.output_text.delta\n" + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"stale\"}\n\n" + "event: response.output_item.added\n" + "data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"function_call\",\"id\":\"item_1\",\"call_id\":\"call_1\",\"name\":\"tool\",\"arguments\":\"{}\"}}\n\n" + "event: response.completed\n" + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n") + stream-body (proxy [java.io.ByteArrayInputStream] + [(.getBytes ^String stream-text java.nio.charset.StandardCharsets/UTF_8)] + (close [] + (reset! closed?* true) + (proxy-super close)))] + (with-redefs [http/post (fn [_url opts] + (is (= :stream (:as opts))) + {:status 200 + :headers response-headers + :body stream-body})] + (llm-providers.openai/create-response! + (base-provider-params) + (base-callbacks + {:on-error (fn [error] + (reset! closed-at-error?* @closed?*) + (swap! errors* conj error)) + :on-message-received (fn [message] (swap! messages* conj message)) + :on-prepare-tool-call (fn [tool] (swap! tool-prepares* conj tool))}))) + (is (= [expected-error] @errors*)) + (is (true? @closed-at-error?*) + "retry/error handling must start only after the failed stream is closed") + (is (empty? @messages*) + "events after response.failed must not emit text or finish") + (is (empty? @tool-prepares*) + "events after response.failed must not prepare tool calls"))) + + (testing "stream false JSON failure returns the same structured error" + (with-redefs [http/post (fn [_url opts] + (is (= :json (:as opts))) + {:status 200 + :headers response-headers + :body failed-response})] + (is (= {:error expected-error} + (llm-providers.openai/create-response! + (assoc (base-provider-params) :extra-payload {:stream false}) + nil))))) + + (testing "structured request ID takes precedence over response and header IDs" + (let [error (#'llm-providers.openai/response-failed->error-data + {:response {:id "resp_123" + :request_id "response_req" + :error {:message "failed" + :request_id "error_req"}}} + {"x-request-id" "header_req"})] + (is (= "error_req" (:request-id error))) + (is (= "error_req" (:request_id error))))))) + (deftest normalize-messages-tool-call-output-image-test (let [tool-output-with-image {:role "tool_call_output"