Consume Glean's streaming endpoints as they arrive.

Generated operations buffer the whole response before returning, which is the right thing for ordinary calls and the wrong thing for a chat or agent run that is meant to appear a token at a time. These functions send the same requests but hand back a Stream.

Glean uses two wire formats and this module covers both:

• agent runs are text/event-stream, decoded by Gleanex.SSE
• chat is newline-delimited JSON, decoded by Gleanex.NDJSON

Consuming a stream

The response is delivered to the process that made the request, so the stream must be consumed in that same process, and only once.

Examples

{:ok, events} = Gleanex.Streaming.agent_run(config, %{agentId: "abc", input: %{}})

for event <- events do
  case Gleanex.SSE.json_data(event) do
    {:ok, payload} -> IO.inspect(payload)
    {:error, _} -> :ok
  end
end

{:ok, chunks} = Gleanex.Streaming.chat(config, %{messages: messages})

chunks
|> Stream.flat_map(&get_in(&1, ["messages", Access.all(), "fragments"]))
|> Enum.each(&IO.write/1)

Failures before the first byte arrive as {:error, %Gleanex.Error{}}. A failure part way through a stream raises while the stream is being consumed, because by then a response has already been reported as successful.