OpenAgents

Commit

inference: number parallel tool-call SSE indexes instead of always 0

a2d363f87477 · AtlantisPleb · · parent 258a25250208 

inference: number parallel tool-call SSE indexes instead of always 0

Live streaming dropped Enum.with_index and hardcoded index 0 on every
tool_calls delta. The CLI then concatenated names and ids. Number the
calls 0, 1, 2 as they go out, and do not merge two finished calls that
share a wire index in the OpenRouter decoder.

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 475 · 2026-08-28T05:43:45.907072Z

Changed files

  • modified lib/openagents/providers/open_router/stream_decoder.ex
  • modified lib/openagents_web/controllers/inference_proxy_controller.ex
  • modified test/openagents/providers/open_router/stream_decoder_test.exs
  • modified test/openagents_web/controllers/inference_proxy_controller_test.exs
  • modified test/support/providers/test.ex

Diff

5 files changed, +191 -2

lib/openagents/providers/open_router/stream_decoder.ex modified +39 -1

@@ -195,8 +195,9 @@ defmodule OpenAgents.Providers.OpenRouter.StreamDecoder do

195 195 
  defp tool_calls(_state, _invalid), do: {:error, :invalid_provider_event}
196 196
197 197 
  defp tool_call(state, %{} = call) do
198 
    index = if is_integer(call["index"]), do: call["index"], else: 0
199 198 
    function = if is_map(call["function"]), do: call["function"], else: %{}
199 
    incoming_id = binary_or_nil(call["id"])
200 
    index = assign_tool_call_index(state, call["index"], incoming_id)
200 201 
    held = Map.get(state.calls, index, %{id: nil, name: nil, arguments: ""})
201 202
202 203 
    arguments =

@@ -220,6 +221,43 @@ defmodule OpenAgents.Providers.OpenRouter.StreamDecoder do

220 221
221 222 
  defp tool_call(_state, _invalid), do: {:error, :invalid_provider_event}
222 223
224 
  # A second finished call that reuses the wire index is a new call. GLM and
225 
  # the live proxy both emit that shape; merging them produced concatenated
226 
  # names and ids (`openagentsopenagents`, `skillbash`).
227 
  defp assign_tool_call_index(state, raw_index, incoming_id) do
228 
    index = parse_tool_call_index(raw_index)
229
230 
    case Map.get(state.calls, index) do
231 
      %{id: held_id}
232 
      when is_binary(incoming_id) and is_binary(held_id) and incoming_id != held_id ->
233 
        next_tool_call_index(state)
234
235 
      _free_or_same ->
236 
        index
237 
    end
238 
  end
239
240 
  defp parse_tool_call_index(n) when is_integer(n) and n >= 0, do: n
241
242 
  defp parse_tool_call_index(s) when is_binary(s) do
243 
    case Integer.parse(s) do
244 
      {n, ""} when n >= 0 -> n
245 
      _ -> 0
246 
    end
247 
  end
248
249 
  defp parse_tool_call_index(_), do: 0
250
251 
  defp next_tool_call_index(%{calls: calls}) do
252 
    case Map.keys(calls) do
253 
      [] -> 0
254 
      keys -> Enum.max(keys) + 1
255 
    end
256 
  end
257
258 
  defp binary_or_nil(value) when is_binary(value) and value != "", do: value
259 
  defp binary_or_nil(_), do: nil
260
223 261 
  defp text_or(value, _fallback) when is_binary(value) and value != "", do: value
224 262 
  defp text_or(_value, fallback), do: fallback
225 263
lib/openagents_web/controllers/inference_proxy_controller.ex modified +10 -1

@@ -228,6 +228,7 @@ defmodule OpenAgentsWeb.InferenceProxyController do

228 228 
  @state_raw_conn :proxy_stream_raw_conn
229 229 
  @state_opened :proxy_stream_opened
230 230 
  @state_allow_fallback :proxy_stream_allow_fallback
231 
  @state_tool_index :proxy_stream_tool_index
231 232
232 233 
  defp run(conn, grant, model, request) do
233 234 
    allow_fallback? = unnamed_selection?(grant, conn.body_params)

@@ -637,6 +638,8 @@ defmodule OpenAgentsWeb.InferenceProxyController do

637 638 
  end
638 639
639 640 
  defp event_chunks({:tool_call, tool_call}) do
641 
    index = next_tool_call_index()
642
640 643 
    [
641 644 
      %{
642 645 
        "choices" => [

@@ -645,7 +648,7 @@ defmodule OpenAgentsWeb.InferenceProxyController do

645 648 
            "delta" => %{
646 649 
              "tool_calls" => [
647 650 
                %{
648 
                  "index" => 0,
651 
                  "index" => index,
649 652 
                  "id" => tool_call.call_id,
650 653 
                  "type" => "function",
651 654 
                  "function" => %{

@@ -663,6 +666,12 @@ defmodule OpenAgentsWeb.InferenceProxyController do

663 666
664 667 
  defp event_chunks(_event), do: []
665 668
669 
  defp next_tool_call_index do
670 
    index = Process.get(@state_tool_index, 0)
671 
    Process.put(@state_tool_index, index + 1)
672 
    index
673 
  end
674
666 675 
  defp wire_usage(usage) do
667 676 
    input = integer(usage["input_tokens"] || usage[:input_tokens])
668 677 
    output = integer(usage["output_tokens"] || usage[:output_tokens])
test/openagents/providers/open_router/stream_decoder_test.exs modified +73

@@ -125,6 +125,79 @@ defmodule OpenAgents.Providers.OpenRouter.StreamDecoderTest do

125 125 
           ]
126 126 
  end
127 127
128 
  test "parallel tool calls that share an index stay separate" do
129 
    stream =
130 
      frame(%{
131 
        "id" => "gen-p",
132 
        "choices" => [
133 
          %{
134 
            "delta" => %{
135 
              "tool_calls" => [
136 
                %{
137 
                  "index" => 0,
138 
                  "id" => "call_a",
139 
                  "function" => %{"name" => "openagents", "arguments" => "{\"name\":\"x\"}"}
140 
                },
141 
                %{
142 
                  "index" => 0,
143 
                  "id" => "call_b",
144 
                  "function" => %{"name" => "bash", "arguments" => "{\"command\":\"ls\"}"}
145 
                }
146 
              ]
147 
            },
148 
            "finish_reason" => "tool_calls"
149 
          }
150 
        ]
151 
      })
152
153 
    assert {:ok, decoder, events} = StreamDecoder.feed(StreamDecoder.new(), stream)
154 
    assert {:ok, _decoder, final} = StreamDecoder.finish(decoder)
155 
    assert events == [{:response_started, "gen-p"}]
156
157 
    assert [
158 
             {:tool_call,
159 
              %ToolCall{call_id: "call_a", name: "openagents", raw_arguments: "{\"name\":\"x\"}"}},
160 
             {:tool_call,
161 
              %ToolCall{call_id: "call_b", name: "bash", raw_arguments: "{\"command\":\"ls\"}"}},
162 
             {:response_completed, "gen-p"}
163 
           ] = final
164 
  end
165
166 
  test "a string tool-call index is still an index" do
167 
    stream =
168 
      frame(%{
169 
        "id" => "gen-s",
170 
        "choices" => [
171 
          %{
172 
            "delta" => %{
173 
              "tool_calls" => [
174 
                %{
175 
                  "index" => "0",
176 
                  "id" => "call_a",
177 
                  "function" => %{"name" => "skill", "arguments" => "{}"}
178 
                },
179 
                %{
180 
                  "index" => "1",
181 
                  "id" => "call_b",
182 
                  "function" => %{"name" => "bash", "arguments" => "{}"}
183 
                }
184 
              ]
185 
            },
186 
            "finish_reason" => "tool_calls"
187 
          }
188 
        ]
189 
      })
190
191 
    assert {:ok, decoder, _events} = StreamDecoder.feed(StreamDecoder.new(), stream)
192 
    assert {:ok, _decoder, final} = StreamDecoder.finish(decoder)
193
194 
    assert [
195 
             {:tool_call, %ToolCall{call_id: "call_a", name: "skill"}},
196 
             {:tool_call, %ToolCall{call_id: "call_b", name: "bash"}},
197 
             {:response_completed, "gen-s"}
198 
           ] = final
199 
  end
200
128 201 
  test "reports a provider error and closes without a completion" do
129 202 
    stream = frame(%{"error" => %{"code" => "model_not_found", "message" => "no such model"}})
130 203
test/openagents_web/controllers/inference_proxy_controller_test.exs modified +42

@@ -293,6 +293,48 @@ defmodule OpenAgentsWeb.InferenceProxyControllerTest do

293 293 
           )
294 294 
  end
295 295
296 
  test "parallel tool calls are numbered 0, 1 rather than both 0", %{conn: conn} do
297 
    %{token: token} = grant("two-tools")
298
299 
    conn =
300 
      post_chat(conn, token, %{
301 
        "messages" => [%{"role" => "user", "content" => "[two-tools]"}],
302 
        "tools" => [
303 
          %{
304 
            "type" => "function",
305 
            "function" => %{
306 
              "name" => "openagents",
307 
              "description" => "CLI",
308 
              "parameters" => %{"type" => "object"}
309 
            }
310 
          },
311 
          %{
312 
            "type" => "function",
313 
            "function" => %{
314 
              "name" => "bash",
315 
              "description" => "Shell",
316 
              "parameters" => %{"type" => "object"}
317 
            }
318 
          }
319 
        ]
320 
      })
321
322 
    assert conn.status == 200
323
324 
    tool_calls =
325 
      conn.resp_body
326 
      |> sse_events()
327 
      |> Enum.filter(&(&1 != "[DONE]"))
328 
      |> Enum.map(&Jason.decode!/1)
329 
      |> Enum.flat_map(fn chunk ->
330 
        get_in(chunk, ["choices", Access.at(0), "delta", "tool_calls"]) || []
331 
      end)
332
333 
    assert Enum.map(tool_calls, & &1["index"]) == [0, 1]
334 
    assert Enum.map(tool_calls, & &1["id"]) == ["call_a", "call_b"]
335 
    assert Enum.map(tool_calls, &get_in(&1, ["function", "name"])) == ["openagents", "bash"]
336 
  end
337
296 338 
  test "a body naming a model outside the catalog is refused, naming the served set",
297 339 
       %{conn: conn} do
298 340 
    %{grant: grant, token: token} = grant("model-not-served")
test/support/providers/test.ex modified +27

@@ -102,6 +102,33 @@ defmodule OpenAgents.Providers.Test do

102 102 
          ~s({"query":"quartz"})
103 103 
        )
104 104
105 
      "[two-tools]" ->
106 
        on_event.({:response_started, "two-tools-0"})
107
108 
        on_event.(
109 
          {:tool_call,
110 
           %ProviderEvent.ToolCall{
111 
             item_id: "item-call-a",
112 
             call_id: "call_a",
113 
             name: "openagents",
114 
             raw_arguments: ~s({"name":"openagents-cli"})
115 
           }}
116 
        )
117
118 
        on_event.(
119 
          {:tool_call,
120 
           %ProviderEvent.ToolCall{
121 
             item_id: "item-call-b",
122 
             call_id: "call_b",
123 
             name: "bash",
124 
             raw_arguments: ~s({"command":"git status"})
125 
           }}
126 
        )
127
128 
        on_event.({:usage, %{"input_tokens" => 3, "output_tokens" => 1}})
129 
        on_event.({:response_completed, "two-tools-0"})
130 
        :ok
131
105 132 
      "[unknown-tool-loop]" ->
106 133 
        emit_tool_request(on_event, "unknown-loop-0", "call-unknown-1", "missing_tool", "{}")
107 134

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