Abort in-flight LLM streams promptly on prompt stop for all providers

Extend the stream cancellation watchdog used by Anthropic/Bedrock to the
openai, openai-chat and ollama handlers, so stopping a prompt closes the
connection within ~500ms even while waiting for the first token, instead
of only being noticed on the next received chunk.

Closes #554

Author: ericdallo

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- Abort in-flight LLM streaming requests promptly on prompt stop for OpenAI, GitHub Copilot, Google, Ollama and custom OpenAI-compatible providers (previously only Anthropic/Bedrock), including while waiting for the first token. #554
+
 - Don't reset a chat's model when changing the agent of an already started chat; the agent's `defaultModel` now only applies to chats without an established model. editor-code-assistant/eca-emacs#282
 - Fix plugin-provided agents missing from the editor's agent list when model sync finishes before plugin resolution (common with providers using `fetchModels: false`): re-emit the chat config after plugins resolve.
 - Surface Anthropic `refusal` stop reason (e.g. Fable safety classifiers) as a clear chat message with Anthropic's category/explanation instead of silently ending the prompt.

src/eca/llm_api.clj

@@ -303,7 +303,9 @@
           :api-key api-key
           :auth-type auth-type
           :account-id (:account-id provider-auth)
-          :prompt-cache-key prompt-cache-key}
+          :prompt-cache-key prompt-cache-key
+          :cancelled? cancelled?
+          :stream-idle-timeout-seconds (:streamIdleTimeoutSeconds config)}
          callbacks)
 
         (= "anthropic" provider)
@@ -338,7 +340,9 @@
                          :reasoning-history reasoning-history
                          :api-url api-url
                          :api-key api-key
-                         :prompt-cache-key prompt-cache-key}]
+                         :prompt-cache-key prompt-cache-key
+                         :cancelled? cancelled?
+                         :stream-idle-timeout-seconds (:streamIdleTimeoutSeconds config)}]
           (case (:api api-handler)
             :openai-responses
             (handler
@@ -384,7 +388,9 @@
                                 extra-payload)
           :extra-headers extra-headers
           :api-url api-url
-          :api-key api-key}
+          :api-key api-key
+          :cancelled? cancelled?
+          :stream-idle-timeout-seconds (:streamIdleTimeoutSeconds config)}
          callbacks)
 
         (= "ollama" provider)
@@ -399,7 +405,9 @@
           :tools tools
           :max-output-tokens max-output-tokens
           :extra-payload extra-payload
-          :extra-headers extra-headers}
+          :extra-headers extra-headers
+          :cancelled? cancelled?
+          :stream-idle-timeout-seconds (:streamIdleTimeoutSeconds config)}
          callbacks)
 
         (and (or model-config

src/eca/llm_providers/ollama.clj

@@ -55,7 +55,7 @@
       (logger/warn logger-tag "Error getting model:" (ex-message e))
       [])))
 
-(defn ^:private base-chat-request! [{:keys [rid url body on-error on-stream extra-headers]}]
+(defn ^:private base-chat-request! [{:keys [rid url body on-error on-stream extra-headers cancelled? stream-idle-timeout-seconds]}]
   (let [reason-id (str (random-uuid))
         reasoning?* (atom false)
         response* (atom nil)
@@ -81,10 +81,33 @@
                        :status status
                        :body body-str}))
           (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 rid event data reasoning?* reason-id)))
+            (let [{:keys [touch-fn set-reading-fn stop-fn reason*]}
+                  (llm-util/start-stream-watchdog! body cancelled?
+                                                   (when stream-idle-timeout-seconds
+                                                     {:idle-timeout-ms (* 1000 stream-idle-timeout-seconds)}))]
+              (try
+                (with-open [rdr (io/reader body)]
+                  (doseq [[event data] (llm-util/event-data-seq rdr)]
+                    (set-reading-fn false)
+                    (touch-fn)
+                    (llm-util/log-response logger-tag rid event data)
+                    (on-stream rid event data reasoning?* reason-id)
+                    (set-reading-fn true)))
+                (catch java.io.IOException e
+                  (let [reason @reason*]
+                    (cond
+                      (= :cancelled reason)
+                      (throw (ex-info "Stream cancelled" {:silent? true}))
+
+                      (= :idle-timeout reason)
+                      (on-error {:message (format "Stream idle timeout: no data received for %d seconds"
+                                                  (or stream-idle-timeout-seconds 120))
+                                 :exception e})
+
+                      :else
+                      (throw e))))
+                (finally
+                  (stop-fn))))
             (do
               (llm-util/log-response logger-tag rid "response" body)
               (reset! response*
@@ -139,7 +162,7 @@
         messages))
 
 (defn chat! [{:keys [model user-messages reason? instructions api-url past-messages tools max-output-tokens
-                     extra-headers extra-payload]}
+                     extra-headers extra-payload cancelled? stream-idle-timeout-seconds]}
              {:keys [on-message-received on-error on-prepare-tool-call on-tools-called
                      on-reason] :as callbacks}]
   (let [messages (concat
@@ -182,6 +205,8 @@
                                      :body (assoc body :messages (normalize-messages new-messages)
                                                   :tools (->tools tools))
                                      :extra-headers extra-headers
+                                     :cancelled? cancelled?
+                                     :stream-idle-timeout-seconds stream-idle-timeout-seconds
                                      :on-error on-error
                                      :on-stream handle-stream})))
                                (on-message-received {:type :finish
@@ -209,5 +234,7 @@
       :url url
       :body body
       :extra-headers extra-headers
+      :cancelled? cancelled?
+      :stream-idle-timeout-seconds stream-idle-timeout-seconds
       :on-error on-error
       :on-stream on-stream-fn})))

src/eca/llm_providers/openai.clj

@@ -171,7 +171,7 @@
                 :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]}]
+(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 cancelled? stream-idle-timeout-seconds]}]
   (let [oauth? (= :auth/oauth auth-type)
         stream? (and on-stream (not= false (:stream body)))
         url (if oauth?
@@ -213,18 +213,41 @@
                        :body body-str
                        :headers resp-headers}))
           (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)))
+            (let [{:keys [touch-fn set-reading-fn stop-fn reason*]}
+                  (llm-util/start-stream-watchdog! body cancelled?
+                                                   (when stream-idle-timeout-seconds
+                                                     {:idle-timeout-ms (* 1000 stream-idle-timeout-seconds)}))]
+              (try
+                (let [stream-error
+                      (with-open [rdr (io/reader body)]
+                        (loop [events (seq (llm-util/event-data-seq rdr))]
+                          (when-let [[event data] (first events)]
+                            (set-reading-fn false)
+                            (touch-fn)
+                            (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)
+                                (set-reading-fn true)
+                                (recur (next events)))))))]
+                  (when stream-error
+                    (on-error stream-error)))
+                (catch java.io.IOException e
+                  (let [reason @reason*]
+                    (cond
+                      (= :cancelled reason)
+                      (throw (ex-info "Stream cancelled" {:silent? true}))
+
+                      (= :idle-timeout reason)
+                      (on-error {:message (format "Stream idle timeout: no data received for %d seconds"
+                                                  (or stream-idle-timeout-seconds 120))
+                                 :exception e})
+
+                      :else
+                      (throw e))))
+                (finally
+                  (stop-fn))))
             (do
               (llm-util/log-response logger-tag rid "response" body)
               (if (= "failed" (:status body))
@@ -335,7 +358,7 @@
 
 (defn create-response! [{:keys [model user-messages instructions reason? supports-image? api-key api-url url-relative-path
                                 max-output-tokens past-messages tools web-search image-generation extra-payload extra-headers
-                                auth-type account-id http-client prompt-cache-key]}
+                                auth-type account-id http-client prompt-cache-key cancelled? stream-idle-timeout-seconds]}
                         {:keys [on-message-received on-error on-prepare-tool-call on-tools-called on-reason on-usage-updated
                                 on-server-web-search on-server-image-generation] :as callbacks}]
   (let [oauth? (= :auth/oauth auth-type)
@@ -501,6 +524,8 @@
                         :http-client http-client
                         :extra-headers extra-headers
                         :auth-type auth-type
+                        :cancelled? cancelled?
+                        :stream-idle-timeout-seconds stream-idle-timeout-seconds
                         :on-error on-error
                         :on-stream handle-stream})))
                   (on-message-received {:type :finish
@@ -525,6 +550,8 @@
                  :http-client http-client
                  :extra-headers extra-headers
                  :auth-type auth-type
+                 :cancelled? cancelled?
+                 :stream-idle-timeout-seconds stream-idle-timeout-seconds
                  :on-error on-error
                  :on-stream on-stream-fn})]
     (if callbacks

src/eca/llm_providers/openai_chat.clj

@@ -115,7 +115,7 @@
 
 (defn ^:private base-chat-request!
   [{:keys [rid extra-headers body url-relative-path api-url api-key on-error on-stream
-           on-tools-called-wrapper http-client]}]
+           on-tools-called-wrapper http-client cancelled? stream-idle-timeout-seconds]}]
   (let [url (join-api-url api-url (or url-relative-path chat-completions-path))
         extra-headers (if (fn? extra-headers)
                         (extra-headers {:body body})
@@ -146,11 +146,35 @@
                        :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))
-              (on-stream "stream-end" {}))
+            (let [{:keys [touch-fn set-reading-fn stop-fn reason*]}
+                  (llm-util/start-stream-watchdog! body cancelled?
+                                                   (when stream-idle-timeout-seconds
+                                                     {:idle-timeout-ms (* 1000 stream-idle-timeout-seconds)}))]
+              (try
+                (with-open [rdr (io/reader body)]
+                  (doseq [[event data] (llm-util/event-data-seq rdr)]
+                    (set-reading-fn false)
+                    (touch-fn)
+                    (llm-util/log-response logger-tag rid event data)
+                    (on-stream event data)
+                    (set-reading-fn true))
+                  (set-reading-fn false)
+                  (on-stream "stream-end" {}))
+                (catch java.io.IOException e
+                  (let [reason @reason*]
+                    (cond
+                      (= :cancelled reason)
+                      (throw (ex-info "Stream cancelled" {:silent? true}))
+
+                      (= :idle-timeout reason)
+                      (on-error {:message (format "Stream idle timeout: no data received for %d seconds"
+                                                  (or stream-idle-timeout-seconds 120))
+                                 :exception e})
+
+                      :else
+                      (throw e))))
+                (finally
+                  (stop-fn))))
             (do
               (llm-util/log-response logger-tag rid "full-response" body)
               (response-body->result body on-tools-called-wrapper)))))
@@ -524,7 +548,7 @@
    Compatible with OpenRouter and other OpenAI-compatible providers."
   [{:keys [model user-messages instructions temperature api-key api-url url-relative-path
            max-output-tokens past-messages tools extra-payload extra-headers supports-image?
-           think-tag-start think-tag-end reasoning-history http-client]}
+           think-tag-start think-tag-end reasoning-history http-client cancelled? stream-idle-timeout-seconds]}
    {:keys [on-message-received on-error on-prepare-tool-call on-tools-called on-reason on-usage-updated] :as callbacks}]
   (let [think-tag-start (or think-tag-start "<think>")
         think-tag-end (or think-tag-end "</think>")
@@ -613,6 +637,8 @@
                                         :api-url api-url
                                         :api-key (or fresh-api-key api-key)
                                         :url-relative-path url-relative-path
+                                        :cancelled? cancelled?
+                                        :stream-idle-timeout-seconds stream-idle-timeout-seconds
                                         :on-error wrapped-on-error
                                         :on-stream (when stream? (fn [event data] (handle-response event data tool-calls*)))}))))
 
@@ -734,6 +760,8 @@
       :url-relative-path url-relative-path
       :tool-calls* tool-calls*
       :on-tools-called-wrapper on-tools-called-wrapper
+      :cancelled? cancelled?
+      :stream-idle-timeout-seconds stream-idle-timeout-seconds
       :on-error wrapped-on-error
       :on-stream (when stream?
                    (fn [event data] (handle-response event data tool-calls*)))})))

test/eca/client_test_helpers.clj

@@ -174,6 +174,23 @@
          (finally
            (.abort ^DefaultHttpProxyServer (:px prx#)))))))
 
+(defn blocking-input-stream
+  "An InputStream whose reads block until `close` is called (10s max),
+  then throw IOException. Simulates a hung streaming connection (e.g.
+  waiting for the first LLM token) for stream-watchdog tests."
+  []
+  (let [closed* (promise)
+        block-then-throw! (fn []
+                            (deref closed* 10000 nil)
+                            (throw (java.io.IOException. "Stream closed")))]
+    (proxy [java.io.InputStream] []
+      (read
+        ([] (block-then-throw!))
+        ([_b] (block-then-throw!))
+        ([_b _off _len] (block-then-throw!)))
+      (close []
+        (deliver closed* true)))))
+
 (def ^:dynamic *http-client-captures*
   "A record of all `eca.client-http/merge-with-global-http-client` merge
   requests results done during the call to `with-client-proxied`

test/eca/llm_providers/ollama_test.clj

@@ -2,8 +2,9 @@
   (:require
    [cheshire.core :as json]
    [clojure.test :refer [deftest is testing]]
-   [eca.client-test-helpers :refer [with-client-proxied]]
+   [eca.client-test-helpers :refer [blocking-input-stream with-client-proxied]]
    [eca.llm-providers.ollama :as llm-providers.ollama]
+   [hato.client :as http]
    [matcher-combinators.test :refer [match?]]))
 
 (deftest list-models-test
@@ -134,3 +135,30 @@
                                                                     :error false
                                                                     :text "Allowed directories: /foo/bar"}]}}}
            {:role "assistant" :content "I see /foo/bar"}])))))
+
+(deftest chat-stream-cancelled-test
+  (testing "watchdog aborts a hung stream when cancelled, surfacing a silent error"
+    (let [errors* (atom [])
+          messages* (atom [])
+          stream-body (blocking-input-stream)]
+      (with-redefs [http/post (fn [_url opts]
+                                (is (= :stream (:as opts)))
+                                {:status 200
+                                 :body stream-body})]
+        (llm-providers.ollama/chat!
+         {:model "test-model"
+          :instructions "System prompt"
+          :user-messages [{:role "user" :content "hello"}]
+          :past-messages []
+          :tools []
+          :api-url "http://localhost:1"
+          :cancelled? (constantly true)}
+         {:on-message-received (fn [msg] (swap! messages* conj msg))
+          :on-error (fn [err] (swap! errors* conj err))
+          :on-prepare-tool-call (fn [_])
+          :on-tools-called (fn [_] {:new-messages [] :tools []})
+          :on-reason (fn [_])}))
+      (is (= 1 (count @errors*)))
+      (is (= "Stream cancelled" (ex-message (:exception (first @errors*)))))
+      (is (true? (:silent? (ex-data (:exception (first @errors*))))))
+      (is (empty? @messages*)))))