From 658f75f3aa7db896403bb5c509b8aa2472c9c6f5 Mon Sep 17 00:00:00 2001 From: Jakub Zika Date: Tue, 20 Oct 2026 10:52:54 +0200 Subject: [PATCH] Show output TPS in chat usage Measure streaming responses from the first generation event to the usage update and expose the result as messageOutputTps. The window ends before tool execution so approval and tool time cannot lower the value. Non-streaming requests and providers without usage keep the field absent. --- CHANGELOG.md | 2 ++ docs/protocol.md | 10 +++++++++ src/eca/llm_api.clj | 35 +++++++++++++++++++++++++----- src/eca/shared.clj | 14 +++++++++++- test/eca/llm_api_test.clj | 45 +++++++++++++++++++++++++++++++++++++++ test/eca/shared_test.clj | 27 ++++++++++++++++++++++- 6 files changed, 126 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d191c1b46..c5d6f2d0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Show output tokens per second (TPS) in the chat usage payload for streamed model responses. + ## 0.165.1 - Add `toolCall.askUser.trustModeTimeoutSeconds`: in trust mode, an unanswered `ask_user` question times out and the LLM continues with its best judgment. diff --git a/docs/protocol.md b/docs/protocol.md index 222e33143..d6ffbf3b3 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -963,6 +963,16 @@ interface ChatUsageContent { * The total input + output tokens of the whole chat session so far. */ sessionTokens: number; + + /** + * Output tokens per second of the last model response, rounded to an + * integer. Measured from the first generated token to the end of the + * response. Includes reasoning tokens when the provider counts them as + * output and server-side tool waits such as web search. Excludes client + * tool execution and approval waits. Absent for non-streaming requests or + * when usage or timing is unknown. + */ + messageOutputTps?: number; /** * The cost of the last sent message summing input + output tokens. diff --git a/src/eca/llm_api.clj b/src/eca/llm_api.clj index cdda2a750..055f3fac4 100644 --- a/src/eca/llm_api.clj +++ b/src/eca/llm_api.clj @@ -539,6 +539,10 @@ on-history-sanitized identity cancelled? (constantly false)}}] (let [first-response-received* (atom false) + ;; First generation event of the current model response. Consumed by + ;; the next usage update, so tool-call loops start a new measurement. + generation-start-ms* (atom nil) + mark-generation-start! #(compare-and-set! generation-start-ms* nil (System/currentTimeMillis)) ;; Fire :on-history-sanitized at most once per sync-or-async-prompt! call: ;; prompt! is invoked multiple times on retries and we don't want to ;; spam the chat with repeated "history sanitized" notices. @@ -558,24 +562,45 @@ ;; tool call. Wrapped here rather than in `prompt!` because the sync path ;; invokes the callback itself, without going through it. on-tools-called-wrapper (fn [tool-calls] + ;; Tool execution can block on user approval. + ;; End the generation window first so a usage + ;; event that arrives after tools (Bedrock) + ;; cannot include that time. + (reset! generation-start-ms* nil) (let [result (on-tools-called tool-calls)] (cond-> result (map? result) (update :tools tools-for-request)))) - on-message-received-wrapper (fn [& args] + on-message-received-wrapper (fn [& [message :as args]] + (when (#{:text :url :refusal} (:type message)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-message-received args)) - on-reason-wrapper (fn [& args] + on-reason-wrapper (fn [& [reason :as args]] + (when (#{:started :thinking} (:status reason)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-reason args)) on-prepare-tool-call-wrapper (fn [& args] + (mark-generation-start!) (apply emit-first-message-fn args) (apply on-prepare-tool-call args)) - on-server-web-search-wrapper (fn [& args] + on-server-web-search-wrapper (fn [& [event :as args]] + (when (= :started (:status event)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-server-web-search args)) - on-server-image-generation-wrapper (fn [& args] + on-server-image-generation-wrapper (fn [& [event :as args]] + (when (= :started (:status event)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-server-image-generation args)) + on-usage-updated-wrapper (fn [usage] + (let [start-ms @generation-start-ms* + elapsed-ms (when start-ms (- (System/currentTimeMillis) start-ms)) + tps (shared/tokens-per-second (:output-tokens usage) elapsed-ms)] + (reset! generation-start-ms* nil) + (on-usage-updated (cond-> usage + tps (assoc :message-output-tps tps))))) ;; Deliver at most one terminal error per prompt: when a shared ;; connection dies, every stacked tool-continuation request unwinds ;; and fires its own on-error for the same failure. #547 @@ -718,7 +743,7 @@ :on-message-received on-message-received-wrapper :on-prepare-tool-call on-prepare-tool-call-wrapper :on-tools-called on-tools-called-wrapper - :on-usage-updated on-usage-updated + :on-usage-updated on-usage-updated-wrapper :on-server-web-search on-server-web-search-wrapper :on-server-image-generation on-server-image-generation-wrapper :on-reason on-reason-wrapper diff --git a/src/eca/shared.clj b/src/eca/shared.clj index f8eccd66c..7e8b303d2 100644 --- a/src/eca/shared.clj +++ b/src/eca/shared.clj @@ -330,11 +330,20 @@ (format "%.2f" (+ input-cost (* output-tokens output-token-cost))))))) +(defn tokens-per-second + "Output tokens per second as an integer. Returns nil when the token count or + elapsed time is not positive." + [tokens elapsed-ms] + (when (and (number? tokens) (number? elapsed-ms) + (pos? (long tokens)) (pos? (long elapsed-ms))) + (Math/round (/ (* 1000.0 tokens) elapsed-ms)))) + (defn ^:private usage-sumary* [chat-id full-model db] (let [last-input-tokens (or (get-in db [:chats chat-id :usage :last-input-tokens]) 0) last-output-tokens (or (get-in db [:chats chat-id :usage :last-output-tokens]) 0) last-input-cache-creation-tokens (or (get-in db [:chats chat-id :usage :last-input-cache-creation-tokens]) 0) last-input-cache-read-tokens (or (get-in db [:chats chat-id :usage :last-input-cache-read-tokens]) 0) + message-output-tps (get-in db [:chats chat-id :usage :last-output-tps]) total-input-tokens (or (get-in db [:chats chat-id :usage :total-input-tokens]) 0) total-input-cache-creation-tokens (or (get-in db [:chats chat-id :usage :total-input-cache-creation-tokens]) 0) total-input-cache-read-tokens (or (get-in db [:chats chat-id :usage :total-input-cache-read-tokens]) 0) @@ -344,6 +353,7 @@ last-input-cache-read-tokens last-input-cache-creation-tokens last-output-tokens)} + :message-output-tps message-output-tps :limit (:limit model-capabilities) :last-message-cost (tokens->cost last-input-tokens last-input-cache-creation-tokens last-input-cache-read-tokens last-output-tokens model-capabilities) :session-cost (tokens->cost total-input-tokens total-input-cache-creation-tokens total-input-cache-read-tokens total-output-tokens model-capabilities)))) @@ -368,12 +378,14 @@ tokens considered, but we save them for cost calculation - cost: we count the tokens in past requests done + current one" [{:keys [input-tokens output-tokens - input-cache-creation-tokens input-cache-read-tokens]} + input-cache-creation-tokens input-cache-read-tokens + message-output-tps]} full-model {:keys [chat-id db*] :as chat-ctx}] (when (and output-tokens input-tokens) (swap! db* assoc-in [:chats chat-id :usage :last-input-tokens] input-tokens) (swap! db* assoc-in [:chats chat-id :usage :last-output-tokens] output-tokens) + (swap! db* assoc-in [:chats chat-id :usage :last-output-tps] message-output-tps) (swap! db* update-in [:chats chat-id :usage :total-input-tokens] (fnil + 0) input-tokens) (swap! db* update-in [:chats chat-id :usage :total-output-tokens] (fnil + 0) output-tokens) (when input-cache-creation-tokens diff --git a/test/eca/llm_api_test.clj b/test/eca/llm_api_test.clj index 5b49d1ff1..8947ca01a 100644 --- a/test/eca/llm_api_test.clj +++ b/test/eca/llm_api_test.clj @@ -11,6 +11,7 @@ [eca.llm-providers.openai :as llm-providers.openai] [eca.llm-providers.openai-chat :as llm-providers.openai-chat] [eca.secrets :as secrets] + [eca.shared :as shared] [eca.test-helper :as h] [matcher-combinators.test :refer [match?]])) @@ -1214,6 +1215,50 @@ (is (= 1 @attempt*)) (is (true? @on-error-called*))))) +(deftest message-output-tps-test + (testing "usage gets TPS from output tokens and measured time" + (let [usage* (atom nil) + tps-args* (atom nil)] + (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-message-received on-usage-updated]}] + (on-message-received {:type :text :text "hi"}) + (Thread/sleep 20) + (on-usage-updated {:input-tokens 1000 :output-tokens 2})) + shared/tokens-per-second (fn [tokens elapsed-ms] + (reset! tps-args* [tokens elapsed-ms]) + 42)] + (llm-api/sync-or-async-prompt! + (make-prompt-opts + {:on-message-received identity + :on-usage-updated #(reset! usage* %)}))) + (let [[tokens elapsed-ms] @tps-args*] + (is (= 2 tokens) "uses output tokens") + (is (and (number? elapsed-ms) (pos? elapsed-ms)) "measures elapsed time")) + (is (= 42 (:message-output-tps @usage*)) "attaches the computed TPS"))) + + (testing "usage has no TPS without a generation event" + (let [usage* (atom nil)] + (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-usage-updated]}] + (on-usage-updated {:input-tokens 1 :output-tokens 2}))] + (llm-api/sync-or-async-prompt! + (make-prompt-opts + {:on-message-received identity + :on-usage-updated #(reset! usage* %)}))) + (is (nil? (:message-output-tps @usage*))))) + + (testing "usage after tools has no TPS because tool time is not generation time" + (let [usage* (atom nil)] + (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-message-received on-tools-called on-usage-updated]}] + (on-message-received {:type :text :text "hi"}) + (Thread/sleep 20) + (on-tools-called []) + (on-usage-updated {:input-tokens 1 :output-tokens 2}))] + (llm-api/sync-or-async-prompt! + (make-prompt-opts + {:on-message-received identity + :on-tools-called (constantly nil) + :on-usage-updated #(reset! usage* %)}))) + (is (nil? (:message-output-tps @usage*)))))) + (deftest async-retry-on-custom-retry-rule-error-pattern-test (testing "retries async when custom retryRules errorPattern matches error message" (let [attempt* (atom 0) diff --git a/test/eca/shared_test.clj b/test/eca/shared_test.clj index d06a2c871..857645756 100644 --- a/test/eca/shared_test.clj +++ b/test/eca/shared_test.clj @@ -14,8 +14,13 @@ :limit {:context 1000 :output 100}}} :provider-quotas {"anthropic" {:updated-at 1 :windows [{:id "5h" :label "5h" :used-percent 42.0}]}}}] (testing "API billing shows costs" - (is (match? {:session-tokens 15 :last-message-cost "0.20" :session-cost "2.00" :quota m/absent} + (is (match? {:session-tokens 15 :last-message-cost "0.20" :session-cost "2.00" + :quota m/absent :message-output-tps m/absent} (shared/usage-sumary "c1" "anthropic/claude" db)))) + (testing "includes the last response TPS when known" + (is (match? {:message-output-tps 42} + (shared/usage-sumary "c1" "anthropic/claude" + (assoc-in db [:chats "c1" :usage :last-output-tps] 42))))) (testing "subscription logins show the provider quota instead of costs" (is (match? {:session-tokens 15 :limit {:context 1000} @@ -174,6 +179,26 @@ (testing "returns nil when mandatory costs are missing" (is (nil? (shared/tokens->cost 30 nil nil 20 {:input-token-cost 0.01})))))) +(deftest tokens-per-second-test + (is (= 50 (shared/tokens-per-second 100 2000))) + (is (= 3 (shared/tokens-per-second 10 3000))) + (is (= 3 (shared/tokens-per-second 10 4000)) "rounds 2.5 up") + (is (nil? (shared/tokens-per-second 0 1000))) + (is (nil? (shared/tokens-per-second 10 0))) + (is (nil? (shared/tokens-per-second nil 1000)))) + +(deftest usage-msg->usage-test + (let [db* (atom {:chats {"c1" {:usage {}}}}) + chat-ctx {:chat-id "c1" :db* db*}] + (shared/usage-msg->usage {:input-tokens 10 :output-tokens 20 :message-output-tps 42} + "anthropic/claude" + chat-ctx) + (is (= 42 (get-in @db* [:chats "c1" :usage :last-output-tps]))) + (shared/usage-msg->usage {:input-tokens 10 :output-tokens 20} + "anthropic/claude" + chat-ctx) + (is (nil? (get-in @db* [:chats "c1" :usage :last-output-tps]))))) + (deftest deep-merge-test (testing "second map as nil returns first map" (is (match?