Add streaming support to LlmComposer.Agent with :tool_call stream chunks

lib/llm_composer/agent/stream_collector.ex

@@ -8,10 +8,16 @@ defmodule LlmComposer.Agent.StreamCollector do
   - **OpenAI / OpenRouter** send `:tool_call_delta` chunks whose `tool_calls` are raw maps keyed by
     `"index"`, with the `function.arguments` JSON split across several chunks. They must be grouped
     by index and concatenated.
+  - **OpenAI Responses** sends a `"function_call_started"` map (carries `"call_id"` and `"name"`)
+    followed by one or more `"function_call_arguments_delta"` maps (carry `"call_id"` and
+    `"arguments_delta"`). Fragments are grouped by `"call_id"` and concatenated in arrival order.
   - **Google** sends already-complete `LlmComposer.FunctionCall` structs (one per chunk).
   - **Bedrock** sends two event types per tool call: a start event with `"toolUseId"` and `"name"`,
     followed by one or more delta events that carry only `"inputJson"` fragments. Fragments are
     grouped by `"toolUseId"` and concatenated in arrival order.
+  - **Ollama** (native provider) does not emit tool-call deltas in its streaming format; text
+    streaming works. For tool-call streaming with Ollama, use the `:open_ai` provider pointed at
+    Ollama's OpenAI-compatible endpoint.
 
   This collector hides those differences. Feed every chunk of a turn through `add/2`, then call
   `tool_turn?/1` to know whether the model requested tools, and `to_llm_response/1` to obtain a
@@ -26,7 +32,7 @@ defmodule LlmComposer.Agent.StreamCollector do
   alias LlmComposer.Message
   alias LlmComposer.StreamChunk
 
-  @supported_providers [:open_ai, :open_router, :google, :bedrock]
+  @supported_providers [:open_ai, :open_router, :open_ai_responses, :google, :bedrock, :ollama]
 
   @type t() :: %__MODULE__{
           provider: atom(),
@@ -133,6 +139,19 @@ defmodule LlmComposer.Agent.StreamCollector do
     end)
   end
 
+  def to_function_calls(%__MODULE__{
+        provider: :open_ai_responses,
+        tool_fragments: fragments,
+        tool_id_sequence: order
+      }) do
+    Enum.map(order, fn call_id ->
+      %{"call_id" => id, "name" => name, "arguments" => args} = Map.fetch!(fragments, call_id)
+      %FunctionCall{id: id, name: name, arguments: args, type: "function"}
+    end)
+  end
+
+  def to_function_calls(%__MODULE__{}), do: []
+
   @doc """
   Builds a synthetic `LlmComposer.LlmResponse` equivalent to a non-streaming response for the turn.
 
@@ -249,6 +268,40 @@ defmodule LlmComposer.Agent.StreamCollector do
     end)
   end
 
+  defp merge_tool_calls(%__MODULE__{provider: :open_ai_responses} = collector, tool_calls) do
+    Enum.reduce(tool_calls, collector, fn
+      %{"type" => "function_call_started", "call_id" => call_id, "name" => name}, acc ->
+        fragment = %{"call_id" => call_id, "name" => name, "arguments" => ""}
+
+        %{
+          acc
+          | tool_fragments: Map.put(acc.tool_fragments, call_id, fragment),
+            tool_id_sequence: acc.tool_id_sequence ++ [call_id],
+            current_tool_id: call_id
+        }
+
+      %{
+        "type" => "function_call_arguments_delta",
+        "call_id" => call_id,
+        "arguments_delta" => delta
+      },
+      acc
+      when is_binary(delta) ->
+        # The API delta event omits call_id; fall back to the most recently started call.
+        target_id = call_id || acc.current_tool_id
+
+        fragments =
+          Map.update!(acc.tool_fragments, target_id, fn f ->
+            Map.update(f, "arguments", delta, &(&1 <> delta))
+          end)
+
+        %{acc | tool_fragments: fragments}
+
+      _other, acc ->
+        acc
+    end)
+  end
+
   @spec merge_fragment(map(), map()) :: map()
   defp merge_fragment(existing, incoming) do
     Map.merge(existing, incoming, fn

lib/llm_composer/function_call_helpers.ex

@@ -9,6 +9,7 @@ defmodule LlmComposer.FunctionCallHelpers do
   behavior.
   """
 
+  alias LlmComposer.Helpers
   alias LlmComposer.LlmResponse
   alias LlmComposer.Message
 
@@ -51,9 +52,18 @@ defmodule LlmComposer.FunctionCallHelpers do
     Enum.map(executed_calls, fn call ->
       %Message{
         type: :tool_result,
-        content: to_string(call.result),
+        content: result_to_content(call.result),
         metadata: %{"tool_call_id" => call.id}
       }
     end)
   end
+
+  @spec result_to_content(term()) :: String.t()
+  defp result_to_content(result) when is_binary(result), do: result
+
+  defp result_to_content(result) do
+    Helpers.json_engine().encode!(result)
+  rescue
+    _ -> inspect(result)
+  end
 end

lib/llm_composer/providers/open_ai_responses.ex

@@ -82,7 +82,7 @@ defmodule LlmComposer.Providers.OpenAIResponses do
 
   defp handle_response({:ok, resp}) do
     Logger.warning(
-      "[open_ai_responses] non-200 response (status=#{Map.get(resp, :status, :unknown)})"
+      "[open_ai_responses] non-200 response (status=#{Map.get(resp, :status, :unknown)}) body=#{inspect(Map.get(resp, :body))}"
     )
 
     {:error, resp}
@@ -273,7 +273,7 @@ defmodule LlmComposer.Providers.OpenAIResponses do
         type: "function",
         name: function.name,
         description: function.description,
-        parameters: function.schema,
+        parameters: Map.put_new(function.schema, "additionalProperties", false),
         strict: true
       }
     end)

lib/llm_composer/providers_runner.ex

@@ -83,6 +83,7 @@ defmodule LlmComposer.ProvidersRunner do
           {{:ok, any()} | {:error, term()}, map()}
   defp run_provider(provider, messages, system_msg, provider_opts) do
     provider_name = provider.name()
+    # Intentional: telemetry must not raise on misconfiguration — nil is acceptable here.
     model = Keyword.get(provider_opts, :model)
 
     :telemetry.span(

lib/llm_composer/stream_chunk.ex

@@ -29,7 +29,14 @@ defmodule LlmComposer.StreamChunk do
   @type t() :: %__MODULE__{
           provider: atom(),
           type:
-            :text_delta | :reasoning_delta | :tool_call_delta | :usage | :done | :error | :unknown,
+            :text_delta
+            | :reasoning_delta
+            | :tool_call_delta
+            | :tool_call
+            | :usage
+            | :done
+            | :error
+            | :unknown,
           text: String.t() | nil,
           reasoning: String.t() | nil,
           reasoning_details: list() | nil,

test/llm_composer/agent/stream_collector_test.exs

@@ -265,7 +265,7 @@ defmodule LlmComposer.Agent.StreamCollectorTest do
   end
 
   test "new/1 raises for unsupported providers" do
-    assert_raise ArgumentError, fn -> StreamCollector.new(:ollama) end
+    assert_raise ArgumentError, fn -> StreamCollector.new(:unknown_provider) end
   end
 
   # --- Helpers ---