From 0da149e1052673f67711bde0d5b37072355b0d95 Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Sun, 21 Jun 2026 14:03:41 +0200 Subject: [PATCH] Merge AI runtime request paths --- AUDIT.md | 1 + lib/bds/ai/openai_compatible_runtime.ex | 111 ++++++++++++++---------- test/bds/ai/chat_streaming_test.exs | 17 +++- 3 files changed, 82 insertions(+), 47 deletions(-) diff --git a/AUDIT.md b/AUDIT.md index 2fb99f5..c911afd 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -30,6 +30,7 @@ after each item. - [x] `mcp/server.ex` - [x] `ai/chat_tools.ex` - [x] `ai/chat.ex` +- [x] `ai/openai_compatible_runtime.ex` --- diff --git a/lib/bds/ai/openai_compatible_runtime.ex b/lib/bds/ai/openai_compatible_runtime.ex index cec342a..97296a0 100644 --- a/lib/bds/ai/openai_compatible_runtime.ex +++ b/lib/bds/ai/openai_compatible_runtime.ex @@ -10,9 +10,7 @@ defmodule BDS.AI.OpenAICompatibleRuntime do http_client = Keyword.get(opts, :http_client, HttpClient) url = models_url(endpoint.url) - headers = - %{"accept" => "application/json"} - |> maybe_put_auth(endpoint.api_key) + headers = request_headers(endpoint.api_key, false) with {:ok, response} <- http_client.get(url, headers), 200 <- response.status do @@ -26,21 +24,8 @@ defmodule BDS.AI.OpenAICompatibleRuntime do def generate(endpoint, request, opts) when is_map(endpoint) and is_map(request) do url = completions_url(endpoint.url) - headers = - %{ - "content-type" => "application/json", - "accept" => "application/json" - } - |> maybe_put_auth(endpoint.api_key) - - payload = - %{ - "model" => request.model, - "messages" => request.messages, - "max_tokens" => request.max_output_tokens - } - |> maybe_disable_thinking(request.model) - |> maybe_put_tools(Map.get(request, :tools, [])) + headers = request_headers(endpoint.api_key, true) + payload = base_payload(request) if stream?(request, opts) do generate_streaming(url, headers, payload, request, Keyword.fetch!(opts, :on_stream)) @@ -50,11 +35,9 @@ defmodule BDS.AI.OpenAICompatibleRuntime do end defp generate_blocking(url, headers, payload, request) do - payload_json = Jason.encode!(payload) + payload_json = encode_payload(payload) - Logger.debug( - "AI OpenAI-compatible request operation=#{inspect(Map.get(request, :operation))} model=#{inspect(request.model)} url=#{url} tools=#{payload |> Map.get("tools", []) |> length()} payload_size=#{byte_size(payload_json)}" - ) + log_request(:blocking, url, request, payload, payload_json) case HttpClient.post(url, headers, payload_json) do {:ok, %{status: 200, body: body}} -> @@ -78,15 +61,12 @@ defmodule BDS.AI.OpenAICompatibleRuntime do result {:ok, %{status: status, body: body}} -> - Logger.error( - "AI OpenAI-compatible HTTP error status=#{status} body=#{String.slice(body, 0, 2000)}" - ) - - {:error, %{kind: :http_error, status: status, body: body}} + log_http_status_error("AI OpenAI-compatible", status, body) + http_status_error(status, body) {:error, reason} -> - Logger.error("AI OpenAI-compatible HTTP request failed: #{inspect(reason)}") - {:error, %{kind: :http_error, reason: reason}} + log_http_request_error("AI OpenAI-compatible", reason) + http_request_error(reason) end end @@ -95,15 +75,9 @@ defmodule BDS.AI.OpenAICompatibleRuntime do # snapshots to `on_stream` as they arrive. The assembled message goes # through the same normalization as the blocking path. defp generate_streaming(url, headers, payload, request, on_stream) do - payload_json = - payload - |> Map.put("stream", true) - |> Map.put("stream_options", %{"include_usage" => true}) - |> Jason.encode!() + payload_json = payload |> streaming_payload() |> encode_payload() - Logger.debug( - "AI OpenAI-compatible streaming request operation=#{inspect(Map.get(request, :operation))} model=#{inspect(request.model)} url=#{url} payload_size=#{byte_size(payload_json)}" - ) + log_request(:streaming, url, request, payload, payload_json) sse = SSE.new(on_stream, emit_interval_ms: stream_emit_interval_ms()) @@ -127,15 +101,12 @@ defmodule BDS.AI.OpenAICompatibleRuntime do end {:ok, %{status: status, body: body}, _sse} -> - Logger.error( - "AI OpenAI-compatible streaming HTTP error status=#{status} body=#{String.slice(body, 0, 2000)}" - ) - - {:error, %{kind: :http_error, status: status, body: body}} + log_http_status_error("AI OpenAI-compatible streaming", status, body) + http_status_error(status, body) {:error, reason} -> - Logger.error("AI OpenAI-compatible streaming request failed: #{inspect(reason)}") - {:error, %{kind: :http_error, reason: reason}} + log_http_request_error("AI OpenAI-compatible streaming", reason) + http_request_error(reason) end end @@ -236,6 +207,58 @@ defmodule BDS.AI.OpenAICompatibleRuntime do end end + defp request_headers(api_key, include_content_type?) do + %{"accept" => "application/json"} + |> maybe_put_content_type(include_content_type?) + |> maybe_put_auth(api_key) + end + + defp maybe_put_content_type(headers, true), + do: Map.put(headers, "content-type", "application/json") + + defp maybe_put_content_type(headers, false), do: headers + + defp base_payload(request) do + %{ + "model" => request.model, + "messages" => request.messages, + "max_tokens" => request.max_output_tokens + } + |> maybe_disable_thinking(request.model) + |> maybe_put_tools(Map.get(request, :tools, [])) + end + + defp streaming_payload(payload) do + payload + |> Map.put("stream", true) + |> Map.put("stream_options", %{"include_usage" => true}) + end + + defp encode_payload(payload), do: Jason.encode!(payload) + + defp log_request(mode, url, request, payload, payload_json) do + tools_segment = + case mode do + :blocking -> " tools=#{payload |> Map.get("tools", []) |> length()}" + :streaming -> "" + end + + Logger.debug( + "AI OpenAI-compatible#{if mode == :streaming, do: " streaming", else: ""} request operation=#{inspect(Map.get(request, :operation))} model=#{inspect(request.model)} url=#{url}#{tools_segment} payload_size=#{byte_size(payload_json)}" + ) + end + + defp log_http_status_error(prefix, status, body) do + Logger.error("#{prefix} HTTP error status=#{status} body=#{String.slice(body, 0, 2000)}") + end + + defp log_http_request_error(prefix, reason) do + Logger.error("#{prefix} request failed: #{inspect(reason)}") + end + + defp http_status_error(status, body), do: {:error, %{kind: :http_error, status: status, body: body}} + defp http_request_error(reason), do: {:error, %{kind: :http_error, reason: reason}} + defp maybe_put_auth(headers, nil), do: headers defp maybe_put_auth(headers, ""), do: headers diff --git a/test/bds/ai/chat_streaming_test.exs b/test/bds/ai/chat_streaming_test.exs index 855f762..a23cea5 100644 --- a/test/bds/ai/chat_streaming_test.exs +++ b/test/bds/ai/chat_streaming_test.exs @@ -1,6 +1,13 @@ defmodule BDS.AI.ChatStreamingTest do use ExUnit.Case, async: false + defmodule FakeSecretBackend do + def encrypt(value), do: {:ok, "enc:" <> value} + + def decrypt("enc:" <> value), do: {:ok, value} + def decrypt(_other), do: {:error, :invalid_ciphertext} + end + defmodule StreamingChatPlug do import Plug.Conn @@ -119,7 +126,8 @@ defmodule BDS.AI.ChatStreamingTest do assert {:ok, reply} = BDS.AI.send_chat_message(conversation_id, "tell me a story", - event_target: self() + event_target: self(), + secret_backend: FakeSecretBackend ) assert reply.assistant_message.content == "Once upon a time" @@ -148,7 +156,10 @@ defmodule BDS.AI.ChatStreamingTest do task = Task.async(fn -> - BDS.AI.send_chat_message(conversation_id, "stream forever", event_target: test_pid) + BDS.AI.send_chat_message(conversation_id, "stream forever", + event_target: test_pid, + secret_backend: FakeSecretBackend + ) end) # Wait until tokens are actually flowing before cancelling. @@ -167,7 +178,7 @@ defmodule BDS.AI.ChatStreamingTest do url: "http://127.0.0.1:#{port}/v1", api_key: "sk-stream", model: "stream-model" - }) + }, secret_backend: FakeSecretBackend) assert :ok = BDS.AI.put_model_preference(:chat, "stream-model") assert :ok = BDS.AI.set_airplane_mode(false)