fix: Add pluggable stream parser for Gemini JSON array streaming (#26)

Gemini's streamGenerateContent endpoint returns a JSON array by default,
not SSE. Instead of forcing SSE with alt=sse, this adds a proper JSON
array stream parser and makes HTTP.stream's buffer parser pluggable.

- Add Nous.Providers.HTTP.JSONArrayParser for JSON array responses
- Add :stream_parser option to HTTP.stream/4 (defaults to SSE)
- Gemini provider uses JSONArrayParser for native format support
- 18 new tests for the JSON array parser

Bumps version to 0.12.2.

Author: nyo16

CHANGELOG.md

@@ -2,6 +2,17 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.12.2] - 2026-03-04
+
+### Fixed
+
+- **Gemini streaming**: Fixed streaming responses returning 0 events. The Gemini `streamGenerateContent` endpoint returns a JSON array (`application/json`) by default, not Server-Sent Events. Instead of forcing SSE via `alt=sse` query parameter, added a pluggable stream parser to `Nous.Providers.HTTP`.
+
+### Added
+
+- `Nous.Providers.HTTP.JSONArrayParser` — stream buffer parser for JSON array responses. Extracts complete JSON objects from a streaming `[{...},{...},...]` response by tracking `{}` nesting depth while respecting string literals and escape sequences.
+- `:stream_parser` option on `HTTP.stream/4` — accepts any module implementing `parse_buffer/1` with the same `{events, remaining_buffer}` contract as SSE parsing. Defaults to the existing SSE parser. Enables any provider with a non-SSE streaming format to plug in a custom parser.
+
 ## [0.12.0] - 2026-02-28
 
 ### Added

lib/nous/providers/gemini.ex

@@ -110,7 +110,11 @@ defmodule Nous.Providers.Gemini do
     # Remove model from params (it's in the URL)
     body = params |> Map.delete("model") |> Map.delete(:model)
 
-    HTTP.stream(url, body, headers, timeout: timeout, finch_name: finch_name)
+    HTTP.stream(url, body, headers,
+      timeout: timeout,
+      finch_name: finch_name,
+      stream_parser: Nous.Providers.HTTP.JSONArrayParser
+    )
   end
 
   # Build URL with model and API key in query params

lib/nous/providers/http.ex

@@ -95,6 +95,9 @@ defmodule Nous.Providers.HTTP do
   ## Options
     * `:timeout` - Request timeout in ms (default: 60_000)
     * `:finch_name` - Finch pool name (default: Nous.Finch)
+    * `:stream_parser` - Module for parsing the stream buffer (default: SSE parsing).
+      Must implement `parse_buffer/1` returning `{events, remaining_buffer}`.
+      See `Nous.Providers.HTTP.JSONArrayParser` for an example.
 
   ## Error Handling
   The stream will emit `{:stream_error, reason}` on errors and then halt.
@@ -106,6 +109,7 @@ defmodule Nous.Providers.HTTP do
       when is_binary(url) and is_map(body) and is_list(headers) do
     timeout = Keyword.get(opts, :timeout, @default_timeout)
     finch_name = Keyword.get(opts, :finch_name, Nous.Finch)
+    stream_parser = Keyword.get(opts, :stream_parser)
 
     case Jason.encode(body) do
       {:ok, json_body} ->
@@ -114,7 +118,9 @@ defmodule Nous.Providers.HTTP do
 
         stream =
           Stream.resource(
-            fn -> start_streaming(url, headers, json_body, finch_name, timeout) end,
+            fn ->
+              start_streaming(url, headers, json_body, finch_name, timeout, stream_parser)
+            end,
             &next_chunk/1,
             &cleanup/1
           )
@@ -356,7 +362,7 @@ defmodule Nous.Providers.HTTP do
   end
 
   # Start streaming - spawn a process to handle Finch.stream
-  defp start_streaming(url, headers, body, finch_name, timeout) do
+  defp start_streaming(url, headers, body, finch_name, timeout, stream_parser) do
     parent = self()
 
     pid =
@@ -408,7 +414,8 @@ defmodule Nous.Providers.HTTP do
       done: false,
       status: nil,
       timeout: timeout,
-      error: nil
+      error: nil,
+      stream_parser: stream_parser
     }
   end
 
@@ -482,7 +489,7 @@ defmodule Nous.Providers.HTTP do
           Logger.error("SSE buffer overflow, terminating stream")
           {[{:stream_error, %{reason: :buffer_overflow}}], %{state | done: true}}
         else
-          {events, remaining_buffer} = parse_sse_buffer(new_buffer)
+          {events, remaining_buffer} = parse_stream_buffer(new_buffer, state.stream_parser)
 
           # Filter out parse errors if we want to be lenient
           {valid_events, errors} =
@@ -505,7 +512,7 @@ defmodule Nous.Providers.HTTP do
 
       {:sse, :done, :ok} ->
         # Flush any remaining buffer
-        {events, _} = parse_sse_buffer(state.buffer <> "\n\n")
+        {events, _} = flush_stream_buffer(state.buffer, state.stream_parser)
 
         final_events =
           Enum.reject(events, fn
@@ -530,6 +537,16 @@ defmodule Nous.Providers.HTTP do
     end
   end
 
+  # Parse stream buffer using the configured parser (default: SSE)
+  defp parse_stream_buffer(buffer, nil), do: parse_sse_buffer(buffer)
+  defp parse_stream_buffer(buffer, parser_mod), do: parser_mod.parse_buffer(buffer)
+
+  # Flush remaining buffer at end of stream
+  # SSE needs a trailing \n\n to force the last event through;
+  # custom parsers just re-parse the remaining buffer as-is.
+  defp flush_stream_buffer(buffer, nil), do: parse_sse_buffer(buffer <> "\n\n")
+  defp flush_stream_buffer(buffer, parser_mod), do: parser_mod.parse_buffer(buffer)
+
   # Cleanup when stream is done
   defp cleanup(state) do
     if state[:pid] && Process.alive?(state.pid) do

lib/nous/providers/http/json_array_parser.ex

@@ -0,0 +1,126 @@
+defmodule Nous.Providers.HTTP.JSONArrayParser do
+  @moduledoc """
+  Stream parser for JSON array responses.
+
+  Parses streaming HTTP responses where the body is a JSON array of objects:
+
+      [{"candidates":[...]},{"candidates":[...]},...]
+
+  Used by providers (like Gemini) that stream responses as a JSON array
+  rather than Server-Sent Events. Has the same interface as
+  `Nous.Providers.HTTP.parse_sse_buffer/1` so it can be used as a
+  drop-in `:stream_parser` for `HTTP.stream/4`.
+
+  ## How it works
+
+  Chunks arrive at arbitrary byte boundaries. The parser accumulates them
+  in a buffer, skips array-level syntax (`[`, `]`, `,`, whitespace), and
+  extracts complete top-level JSON objects by tracking `{}` nesting depth
+  while respecting string literals and escape sequences.
+  """
+
+  require Logger
+
+  @doc """
+  Parse a buffer containing chunks of a JSON array into individual events.
+
+  Returns `{events, remaining_buffer}` where events is a list of parsed
+  JSON maps (same contract as `HTTP.parse_sse_buffer/1`).
+
+  ## Examples
+
+      iex> parse_buffer(~s|[{"text":"hi"},{"text":"there"}]|)
+      {[%{"text" => "hi"}, %{"text" => "there"}], ""}
+
+      iex> parse_buffer(~s|[{"text":"hi"},{"tex|)
+      {[%{"text" => "hi"}], ~s|{"tex|}
+
+      iex> parse_buffer("")
+      {[], ""}
+  """
+  @spec parse_buffer(String.t()) :: {list(), String.t()}
+  def parse_buffer(buffer) when is_binary(buffer) do
+    extract_objects(buffer, [])
+  end
+
+  def parse_buffer(_), do: {[], ""}
+
+  # Recursively extract complete JSON objects from the buffer
+  defp extract_objects(buffer, acc) do
+    trimmed = skip_array_syntax(buffer)
+
+    case extract_next_object(trimmed) do
+      {:ok, json_str, rest} ->
+        case Jason.decode(json_str) do
+          {:ok, parsed} ->
+            extract_objects(rest, [parsed | acc])
+
+          {:error, error} ->
+            Logger.debug("JSON array parser: failed to decode object: #{inspect(error)}")
+            # Don't consume more — the buffer might just need more data
+            {Enum.reverse(acc), trimmed}
+        end
+
+      :incomplete ->
+        {Enum.reverse(acc), trimmed}
+    end
+  end
+
+  # Skip array-level syntax: [ ] , and whitespace between objects
+  defp skip_array_syntax(<<c, rest::binary>>) when c in ~c|[,] \t\n\r|,
+    do: skip_array_syntax(rest)
+
+  defp skip_array_syntax(buffer), do: buffer
+
+  # Extract the next complete top-level JSON object from the buffer.
+  # Only starts extraction when buffer begins with `{`.
+  defp extract_next_object(<<"{", _::binary>> = buffer) do
+    case find_object_end(buffer, 0, 0, false) do
+      {:ok, end_pos} ->
+        <<json::binary-size(^end_pos), rest::binary>> = buffer
+        {:ok, json, rest}
+
+      :incomplete ->
+        :incomplete
+    end
+  end
+
+  defp extract_next_object(_), do: :incomplete
+
+  # Walk the buffer character by character tracking {} depth.
+  # Respects JSON string boundaries and escape sequences.
+  #
+  # Returns {:ok, end_position} when a complete object is found,
+  # or :incomplete if the buffer ends mid-object.
+
+  # Buffer exhausted before object closed
+  defp find_object_end(<<>>, _pos, _depth, _in_string), do: :incomplete
+
+  # Escaped character inside a string — skip both bytes
+  defp find_object_end(<<"\\", _, rest::binary>>, pos, depth, true) do
+    find_object_end(rest, pos + 2, depth, true)
+  end
+
+  # Quote toggles string state
+  defp find_object_end(<<"\"", rest::binary>>, pos, depth, in_string) do
+    find_object_end(rest, pos + 1, depth, not in_string)
+  end
+
+  # Open brace outside string — increase depth
+  defp find_object_end(<<"{", rest::binary>>, pos, depth, false) do
+    find_object_end(rest, pos + 1, depth + 1, false)
+  end
+
+  # Close brace outside string — decrease depth, check if object complete
+  defp find_object_end(<<"}", rest::binary>>, pos, depth, false) do
+    case depth - 1 do
+      0 -> {:ok, pos + 1}
+      new_depth -> find_object_end(rest, pos + 1, new_depth, false)
+    end
+  end
+
+  # Any other character — advance position
+  defp find_object_end(<<_, rest::binary>>, pos, depth, in_string) do
+    find_object_end(rest, pos + 1, depth, in_string)
+  end
+end

mix.exs

@@ -1,7 +1,7 @@
 defmodule Nous.MixProject do
   use Mix.Project
 
-  @version "0.12.1"
+  @version "0.12.2"
   @source_url "https://github.com/nyo16/nous"
 
   def project do