OpenAgents

Commit

Carry the cached-token split through usage and grants

cb314d14067d · AtlantisPleb · · parent 56f2cadb0c5d 

Carry the cached-token split through usage and grants

Cost figures overstated agentic work badly. A measured trial spent
124,941 input tokens that were mostly the same prefix re-sent every
round, and both providers bill cached prefix tokens at a fraction of
fresh — but the usage record carried only totals, so the number was
wrong on exactly the workloads that matter.

The proxy now reads the provider's cache fields where they exist —
OpenAI's prompt_tokens_details.cached_tokens and Gemini's
cachedContentTokenCount — and records them on grant usage beside the
totals.

The distinction that makes this honest: a provider reporting no split
records none. Absent, not zero. The usage merge no longer coerces a
missing field to 0 and adds it, because a zero read as fact says the
prefix was fresh when nobody measured it.

The ATIF and CLI halves of #220 stay open; this is the server side.

Built by a Devin child through the openagents coder's delegate tool;
185 inference, provider, chat, and proxy tests re-run before landing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GoYpb8FEmdxVErsv7ABCYi
Co-Authored-By
Claude Fable 5 <noreply@anthropic.com>

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

pushed
by user · WAL seq 344 · 2026-08-25T05:50:01.943443Z

Changed files

  • modified lib/openagents/chat/gemini/stream_decoder.ex
  • modified lib/openagents/inference.ex
  • modified lib/openagents/providers/open_ai/stream_decoder.ex
  • modified lib/openagents_web/controllers/inference_proxy_controller.ex
  • modified test/openagents/chat/gemini/stream_decoder_test.exs
  • modified test/openagents/inference_test.exs
  • modified test/openagents/providers/open_ai/stream_decoder_test.exs

Diff

7 files changed, +212 -13

lib/openagents/chat/gemini/stream_decoder.ex modified +2 -2

@@ -188,7 +188,7 @@ defmodule OpenAgents.Chat.Gemini.StreamDecoder do

188 188 
        "output_tokens" => count(usage["candidatesTokenCount"]),
189 189 
        "total_tokens" => count(usage["totalTokenCount"]),
190 190 
        "reasoning_tokens" => count(usage["thoughtsTokenCount"]),
191 
        "cached_tokens" => count(usage["cachedContentTokenCount"])
191 
        "cache_read_input_tokens" => count(usage["cachedContentTokenCount"])
192 192 
      }
193 193 
      |> Enum.reject(fn {_key, value} -> is_nil(value) end)
194 194 
      |> Map.new()

@@ -246,7 +246,7 @@ defmodule OpenAgents.Chat.Gemini.StreamDecoder do

246 246 
        "output_tokens" => usage["output_tokens"],
247 247 
        "total_tokens" => usage["total_tokens"],
248 248 
        "output_tokens_details" => details("reasoning_tokens", usage["reasoning_tokens"]),
249 
        "input_tokens_details" => details("cached_tokens", usage["cached_tokens"])
249 
        "input_tokens_details" => details("cached_tokens", usage["cache_read_input_tokens"])
250 250 
      }
251 251 
      |> Enum.reject(fn {_key, value} -> is_nil(value) end)
252 252 
      |> Map.new()
lib/openagents/inference.ex modified +24 -2

@@ -383,7 +383,19 @@ defmodule OpenAgents.Inference do

383 383
384 384 
    merged =
385 385 
      Enum.reduce(@cost_fields, %{}, fn field, acc ->
386 
        Map.put(acc, field, integer(existing[field]) + integer(normalized[field]))
386 
        case {existing[field], normalized[field]} do
387 
          {nil, nil} ->
388 
            acc
389
390 
          {existing_value, nil} ->
391 
            Map.put(acc, field, integer(existing_value))
392
393 
          {nil, normalized_value} ->
394 
            Map.put(acc, field, integer(normalized_value))
395
396 
          {existing_value, normalized_value} ->
397 
            Map.put(acc, field, integer(existing_value) + integer(normalized_value))
398 
        end
387 399 
      end)
388 400
389 401 
    merged

@@ -412,10 +424,20 @@ defmodule OpenAgents.Inference do

412 424
413 425 
  defp normalize_usage(usage) do
414 426 
    Enum.reduce(@cost_fields, %{}, fn field, acc ->
415 
      Map.put(acc, field, integer(usage[field] || usage[String.to_atom(field)]))
427 
      case raw_value(usage, field) do
428 
        nil -> acc
429 
        value -> Map.put(acc, field, integer(value))
430 
      end
416 431 
    end)
417 432 
  end
418 433
434 
  defp raw_value(usage, field) do
435 
    case usage[field] do
436 
      nil -> usage[String.to_atom(field)]
437 
      value -> value
438 
    end
439 
  end
440
419 441 
  # ── helpers ─────────────────────────────────────────────────────────────
420 442
421 443 
  defp expire(%Grant{} = grant) do
lib/openagents/providers/open_ai/stream_decoder.ex modified +17 -7

@@ -201,13 +201,21 @@ defmodule OpenAgents.Providers.OpenAI.StreamDecoder do

201 201
202 202 
  defp normalize_usage(usage) when is_map(usage) and map_size(usage) <= 32 do
203 203 
    normalized =
204 
      usage
205 
      |> Map.take(["input_tokens", "output_tokens", "total_tokens"])
206 
      |> Map.put("cached_input_tokens", get_in(usage, ["input_tokens_details", "cached_tokens"]))
207 
      |> Map.put(
208 
        "reasoning_output_tokens",
209 
        get_in(usage, ["output_tokens_details", "reasoning_tokens"])
210 
      )
204 
      %{
205 
        "input_tokens" => first_value([usage["input_tokens"], usage["prompt_tokens"]]),
206 
        "output_tokens" => first_value([usage["output_tokens"], usage["completion_tokens"]]),
207 
        "total_tokens" => first_value([usage["total_tokens"]]),
208 
        "cache_read_input_tokens" =>
209 
          first_value([
210 
            get_in(usage, ["input_tokens_details", "cached_tokens"]),
211 
            get_in(usage, ["prompt_tokens_details", "cached_tokens"])
212 
          ]),
213 
        "reasoning_output_tokens" =>
214 
          first_value([
215 
            get_in(usage, ["output_tokens_details", "reasoning_tokens"]),
216 
            get_in(usage, ["completion_tokens_details", "reasoning_tokens"])
217 
          ])
218 
      }
211 219 
      |> Enum.reject(fn {_key, value} -> is_nil(value) end)
212 220 
      |> Map.new()
213 221

@@ -218,6 +226,8 @@ defmodule OpenAgents.Providers.OpenAI.StreamDecoder do

218 226
219 227 
  defp normalize_usage(_usage), do: {:error, :invalid_provider_event}
220 228
229 
  defp first_value(values), do: Enum.find(values, fn value -> not is_nil(value) end)
230
221 231 
  defp usage_events(nil), do: []
222 232 
  defp usage_events(usage), do: [{:usage, usage}]
223 233
lib/openagents_web/controllers/inference_proxy_controller.ex modified +15 -1

@@ -297,12 +297,26 @@ defmodule OpenAgentsWeb.InferenceProxyController do

297 297 
    input = integer(usage["input_tokens"] || usage[:input_tokens])
298 298 
    output = integer(usage["output_tokens"] || usage[:output_tokens])
299 299 
    total = integer(usage["total_tokens"] || usage[:total_tokens])
300 
    cache = cache_read_tokens(usage)
300 301
301 
    %{
302 
    base = %{
302 303 
      "prompt_tokens" => input,
303 304 
      "completion_tokens" => output,
304 305 
      "total_tokens" => if(total > 0, do: total, else: input + output)
305 306 
    }
307
308 
    if is_nil(cache) do
309 
      base
310 
    else
311 
      Map.put(base, "prompt_tokens_details", %{"cached_tokens" => cache})
312 
    end
313 
  end
314
315 
  defp cache_read_tokens(usage) do
316 
    case usage["cache_read_input_tokens"] || usage[:cache_read_input_tokens] do
317 
      value when is_integer(value) and value >= 0 -> value
318 
      _ -> nil
319 
    end
306 320 
  end
307 321
308 322 
  defp data(payload), do: ["data: ", Jason.encode!(payload), "\n\n"]
test/openagents/chat/gemini/stream_decoder_test.exs modified +37

@@ -239,6 +239,43 @@ defmodule OpenAgents.Chat.Gemini.StreamDecoderTest do

239 239 
    end
240 240 
  end
241 241
242 
  test "reports cache read input tokens on the internal usage map when present" do
243 
    stream =
244 
      frame(chunk([part("hi")])) <>
245 
        frame(%{
246 
          "candidates" => [%{"content" => %{"parts" => []}, "finishReason" => "STOP"}],
247 
          "usageMetadata" => %{
248 
            "promptTokenCount" => 17,
249 
            "candidatesTokenCount" => 14,
250 
            "totalTokenCount" => 304,
251 
            "cachedContentTokenCount" => 5
252 
          }
253 
        })
254
255 
    {state, _events} = feed_all(StreamDecoder.new(model: "m"), [stream])
256 
    assert state.usage["cache_read_input_tokens"] == 5
257 
    assert state.usage["input_tokens"] == 17
258 
    assert {:ok, _completion} = StreamDecoder.finish(state)
259 
  end
260
261 
  test "omits cache read input tokens from the internal usage map when absent" do
262 
    stream =
263 
      frame(chunk([part("hi")])) <>
264 
        frame(%{
265 
          "candidates" => [%{"content" => %{"parts" => []}, "finishReason" => "STOP"}],
266 
          "usageMetadata" => %{
267 
            "promptTokenCount" => 8,
268 
            "candidatesTokenCount" => 3,
269 
            "totalTokenCount" => 11
270 
          }
271 
        })
272
273 
    {state, _events} = feed_all(StreamDecoder.new(model: "m"), [stream])
274 
    refute Map.has_key?(state.usage, "cache_read_input_tokens")
275 
    refute Map.has_key?(state.usage, "cache_write_input_tokens")
276 
    assert {:ok, _completion} = StreamDecoder.finish(state)
277 
  end
278
242 279 
  describe "failure" do
243 280 
    test "a stream with no finishReason is truncated, not a short answer" do
244 281 
      {state, _events} =
test/openagents/inference_test.exs modified +54

@@ -114,6 +114,60 @@ defmodule OpenAgents.InferenceTest do

114 114 
      assert twice.usage["total_tokens"] == 155
115 115 
    end
116 116
117 
    test "records cache read input tokens when the provider reports them" do
118 
      {:ok, grant, _token} = Inference.mint(scope("usage-cache"))
119
120 
      {:ok, once} =
121 
        Inference.record_usage(grant, %{
122 
          "input_tokens" => 100,
123 
          "output_tokens" => 40,
124 
          "cache_read_input_tokens" => 20
125 
        })
126
127 
      assert once.usage["input_tokens"] == 100
128 
      assert once.usage["output_tokens"] == 40
129 
      assert once.usage["cache_read_input_tokens"] == 20
130 
      assert once.usage["total_tokens"] == 140
131
132 
      {:ok, twice} =
133 
        Inference.record_usage(once, %{
134 
          "input_tokens" => 10,
135 
          "output_tokens" => 5,
136 
          "cache_read_input_tokens" => 3
137 
        })
138
139 
      assert twice.usage["cache_read_input_tokens"] == 23
140 
    end
141
142 
    test "omits cache splits when the provider does not report them" do
143 
      {:ok, grant, _token} = Inference.mint(scope("usage-no-cache"))
144
145 
      {:ok, once} =
146 
        Inference.record_usage(grant, %{"input_tokens" => 100, "output_tokens" => 40})
147
148 
      refute Map.has_key?(once.usage, "cache_read_input_tokens")
149 
      refute Map.has_key?(once.usage, "cache_write_input_tokens")
150 
      assert once.usage["input_tokens"] == 100
151 
      assert once.usage["total_tokens"] == 140
152 
    end
153
154 
    test "preserves existing cache read tokens when a later call reports none" do
155 
      {:ok, grant, _token} = Inference.mint(scope("usage-cache-persist"))
156
157 
      {:ok, once} =
158 
        Inference.record_usage(grant, %{
159 
          "input_tokens" => 100,
160 
          "output_tokens" => 40,
161 
          "cache_read_input_tokens" => 20
162 
        })
163
164 
      {:ok, twice} =
165 
        Inference.record_usage(once, %{"input_tokens" => 10, "output_tokens" => 5})
166
167 
      assert twice.usage["cache_read_input_tokens"] == 20
168 
      refute Map.has_key?(twice.usage, "cache_write_input_tokens")
169 
    end
170
117 171 
    test "flips the grant to exhausted when a ceiling is reached" do
118 172 
      {:ok, grant, _token} = Inference.mint(scope("usage-exhaust"))
119 173
test/openagents/providers/open_ai/stream_decoder_test.exs modified +63 -1

@@ -57,7 +57,7 @@ defmodule OpenAgents.Providers.OpenAI.StreamDecoderTest do

57 57 
              }},
58 58 
             {:usage,
59 59 
              %{
60 
                "cached_input_tokens" => 2,
60 
                "cache_read_input_tokens" => 2,
61 61 
                "input_tokens" => 10,
62 62 
                "output_tokens" => 4,
63 63 
                "reasoning_output_tokens" => 1,

@@ -67,6 +67,68 @@ defmodule OpenAgents.Providers.OpenAI.StreamDecoderTest do

67 67 
           ]
68 68 
  end
69 69
70 
  test "decodes chat-completions usage and omits cache read tokens when absent" do
71 
    stream =
72 
      frame(%{"type" => "response.created", "response" => %{"id" => "resp_cc"}}) <>
73 
        frame(%{
74 
          "type" => "response.output_text.delta",
75 
          "delta" => "Ok"
76 
        }) <>
77 
        frame(%{
78 
          "type" => "response.completed",
79 
          "response" => %{
80 
            "id" => "resp_cc",
81 
            "usage" => %{
82 
              "prompt_tokens" => 7,
83 
              "completion_tokens" => 2,
84 
              "total_tokens" => 9
85 
            }
86 
          }
87 
        })
88
89 
    assert {:ok, decoder, events} = feed_all([stream])
90 
    assert {:ok, _decoder, final_events} = StreamDecoder.finish(decoder)
91
92 
    assert events ++ final_events == [
93 
             {:response_started, "resp_cc"},
94 
             {:text_delta, "Ok"},
95 
             {:usage, %{"input_tokens" => 7, "output_tokens" => 2, "total_tokens" => 9}},
96 
             {:response_completed, "resp_cc"}
97 
           ]
98 
  end
99
100 
  test "decodes prompt token cache details from chat-completions usage" do
101 
    stream =
102 
      frame(%{"type" => "response.created", "response" => %{"id" => "resp_cache"}}) <>
103 
        frame(%{
104 
          "type" => "response.completed",
105 
          "response" => %{
106 
            "id" => "resp_cache",
107 
            "usage" => %{
108 
              "prompt_tokens" => 12,
109 
              "completion_tokens" => 3,
110 
              "total_tokens" => 15,
111 
              "prompt_tokens_details" => %{"cached_tokens" => 6}
112 
            }
113 
          }
114 
        })
115
116 
    assert {:ok, decoder, events} = feed_all([stream])
117 
    assert {:ok, _decoder, final_events} = StreamDecoder.finish(decoder)
118
119 
    assert events ++ final_events == [
120 
             {:response_started, "resp_cache"},
121 
             {:usage,
122 
              %{
123 
                "input_tokens" => 12,
124 
                "output_tokens" => 3,
125 
                "total_tokens" => 15,
126 
                "cache_read_input_tokens" => 6
127 
              }},
128 
             {:response_completed, "resp_cache"}
129 
           ]
130 
  end
131
70 132 
  test "decodes reasoning summary and reasoning text deltas as reasoning events" do
71 133 
    stream =
72 134 
      frame(%{"type" => "response.created", "response" => %{"id" => "resp_r"}}) <>

This page updates live while a promote is in flight · changelog