diff --git a/backlog/tasks/ml-146 - Honour-Retry-After-and-rate-limit-reset-headers-for-precise-snooze.md b/backlog/tasks/ml-146 - Honour-Retry-After-and-rate-limit-reset-headers-for-precise-snooze.md index b707ec4d..53fecc95 100644 --- a/backlog/tasks/ml-146 - Honour-Retry-After-and-rate-limit-reset-headers-for-precise-snooze.md +++ b/backlog/tasks/ml-146 - Honour-Retry-After-and-rate-limit-reset-headers-for-precise-snooze.md @@ -5,11 +5,12 @@ status: Done assignee: - Codex created_date: '2026-04-24 11:12' -updated_date: '2026-04-25 06:37' +updated_date: '2026-04-27 21:02' labels: [] dependencies: - ML-21 priority: low +ordinal: 1000 --- ## Description diff --git a/backlog/tasks/ml-147 - Upgrade-to-ServerSentEvents-1.0.md b/backlog/tasks/ml-147 - Upgrade-to-ServerSentEvents-1.0.md index 96f8c227..fe76b202 100644 --- a/backlog/tasks/ml-147 - Upgrade-to-ServerSentEvents-1.0.md +++ b/backlog/tasks/ml-147 - Upgrade-to-ServerSentEvents-1.0.md @@ -1,12 +1,14 @@ --- id: ML-147 title: Upgrade to ServerSentEvents 1.0 -status: To Do +status: Done assignee: [] created_date: '2026-04-25 21:26' +updated_date: '2026-04-27 21:02' labels: [] dependencies: [] priority: medium +ordinal: 2000 --- ## Description diff --git a/lib/open_ai/api.ex b/lib/open_ai/api.ex index a1e58df1..e711aca5 100644 --- a/lib/open_ai/api.ex +++ b/lib/open_ai/api.ex @@ -9,6 +9,11 @@ defmodule OpenAI.API do Mid-stream failures in `chat_stream/6` come from parsed SSE event payloads rather than HTTP responses and remain as message strings passed through the stream callback plumbing. + + `chat_stream/6` streams response chunks with `Req`'s `into: :self`, so the + calling process receives and consumes Req's async body messages while the + request is active. Call it from a short-lived worker/task process rather than + directly from a LiveView or GenServer that has unrelated mailbox traffic. """ alias OpenAI.API.ErrorResponse @@ -63,40 +68,18 @@ defmodule OpenAI.API do stream: true, temperature: temperature }, - into: fn {:data, data}, {req, resp} -> - buffer = Request.get_private(req, :sse_buffer, "") - {events, buffer} = ServerSentEvents.parse(buffer <> data) - req = Request.put_private(req, :sse_buffer, buffer) - - decode_events(events, cb, req, resp) - end + into: :self ) - |> do_chat_stream() + |> do_chat_stream(cb) end - defp decode_events([], _cb, req, resp), do: {:cont, {req, resp}} - - defp decode_events([%{data: json} | rest], cb, req, resp) do - case decode_responses_event(json, cb) do - {:error, message} -> - Logger.error(message) - {:halt, {req, Req.Response.put_private(resp, :error, message)}} - - _ -> - decode_events(rest, cb, req, resp) - end - end - - defp do_chat_stream(req) do + defp do_chat_stream(req, cb) do case Req.post(req) do {:ok, %{status: status} = resp} when status in 200..299 -> - if message = Req.Response.get_private(resp, :error) do - {:error, message} - else - :ok - end + decode_response_stream(resp.body, cb) {:ok, response} -> + response = decode_async_error_body(response) {:error, ErrorResponse.from_response(response)} {:error, exception} -> @@ -104,6 +87,39 @@ defmodule OpenAI.API do end end + defp decode_response_stream(stream, cb) do + stream + |> ServerSentEvents.decode_stream() + |> Enum.reduce_while(:ok, fn %{data: json}, :ok -> + case decode_responses_event(json, cb) do + {:error, message} -> + Logger.error(message) + {:halt, {:error, message}} + + _ -> + {:cont, :ok} + end + end) + end + + defp decode_async_error_body(%{body: %Req.Response.Async{} = body} = response) do + body = + body + |> Enum.into("") + |> decode_error_body() + + %{response | body: body} + end + + defp decode_async_error_body(response), do: response + + defp decode_error_body(body) when is_binary(body) do + case JSON.decode(body) do + {:ok, decoded} -> decoded + {:error, _reason} -> body + end + end + @spec get_embeddings(String.t(), OpenAI.Config.t()) :: {:ok, [float()]} | {:error, ErrorResponse.t() | Exception.t()} def get_embeddings(text, config) do diff --git a/mix.exs b/mix.exs index d270fdcf..2998efef 100644 --- a/mix.exs +++ b/mix.exs @@ -129,7 +129,7 @@ defmodule MusicLibrary.MixProject do # HTTP Clients {:finch, "~> 0.21.0"}, {:req, "~> 0.5.8"}, - {:server_sent_events, "~> 0.2"}, + {:server_sent_events, "~> 1.0"}, # Mailers {:idna, "~> 7.1", override: true}, diff --git a/mix.lock b/mix.lock index a74031ec..a7f94ce3 100644 --- a/mix.lock +++ b/mix.lock @@ -67,7 +67,7 @@ "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, "rewrite": {:hex, :rewrite, "1.3.0", "67448ba7975690b35ba7e7f35717efcce317dbd5963cb0577aa7325c1923121a", [:mix], [{:glob_ex, "~> 0.1", [hex: :glob_ex, repo: "hexpm", optional: false]}, {:sourceror, "~> 1.0", [hex: :sourceror, repo: "hexpm", optional: false]}, {:text_diff, "~> 0.1", [hex: :text_diff, repo: "hexpm", optional: false]}], "hexpm", "d111ac7ff3a58a802ef4f193bbd1831e00a9c57b33276e5068e8390a212714a5"}, "rustler_precompiled": {:hex, :rustler_precompiled, "0.9.0", "3a052eda09f3d2436364645cc1f13279cf95db310eb0c17b0d8f25484b233aa0", [:mix], [{:rustler, "~> 0.23", [hex: :rustler, repo: "hexpm", optional: true]}], "hexpm", "471d97315bd3bf7b64623418b3693eedd8e47de3d1cb79a0ac8f9da7d770d94c"}, - "server_sent_events": {:hex, :server_sent_events, "0.2.1", "f83b34f01241302a8bf451efc8dde3a36c533d5715463c31c653f3db8695f636", [:mix], [], "hexpm", "c8099ce4f9acd610eb7c8e0f89dba7d5d1c13300ea9884b0bd8662401d3cf96f"}, + "server_sent_events": {:hex, :server_sent_events, "1.0.0", "e82089ac6b93ebd3c0562fd728492bbe4b5140678ffc891abfa8cce717c2c1ff", [:mix], [], "hexpm", "7899caea3e27850549f671fc9e6c53d55a8e6a78474f6b9623820aae6bb41ec7"}, "sobelow": {:hex, :sobelow, "0.14.1", "2f81e8632f15574cba2402bcddff5497b413c01e6f094bc0ab94e83c2f74db81", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "8fac9a2bd90fdc4b15d6fca6e1608efb7f7c600fa75800813b794ee9364c87f2"}, "sourceror": {:hex, :sourceror, "1.12.0", "da354c5f35aad3cc1132f5d5b0d8437d865e2661c263260480bab51b5eedb437", [:mix], [], "hexpm", "755703683bd014ebcd5de9acc24b68fb874a660a568d1d63f8f98cd8a6ef9cd0"}, "spitfire": {:hex, :spitfire, "0.3.11", "79dfcb033762470de472c1c26ea2b4e3aca74700c685dbffd9a13466272c323d", [:mix], [], "hexpm", "eb6e2dadf63214e8bfe65ca9788cef2b03b01027365d78d3c0e3d9ebd3d5b7b4"},