Merge AI runtime request paths

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`
 
 ---
 

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 =
+    headers = request_headers(endpoint.api_key, false)
-      %{"accept" => "application/json"}
-      |> maybe_put_auth(endpoint.api_key)
 
     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 =
+    headers = request_headers(endpoint.api_key, true)
-      %{
+    payload = base_payload(request)
-        "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, []))
 
     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(
+    log_request(:blocking, url, request, payload, payload_json)
-      "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)}"
-    )
 
     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(
+        log_http_status_error("AI OpenAI-compatible", status, body)
-          "AI OpenAI-compatible HTTP error status=#{status} body=#{String.slice(body, 0, 2000)}"
+        http_status_error(status, body)
-        )
-
-        {:error, %{kind: :http_error, status: status, body: body}}
 
       {:error, reason} ->
-        Logger.error("AI OpenAI-compatible HTTP request failed: #{inspect(reason)}")
+        log_http_request_error("AI OpenAI-compatible", reason)
-        {:error, %{kind: :http_error, reason: 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_json = payload |> streaming_payload() |> encode_payload()
-      payload
-      |> Map.put("stream", true)
-      |> Map.put("stream_options", %{"include_usage" => true})
-      |> Jason.encode!()
 
-    Logger.debug(
+    log_request(:streaming, url, request, payload, payload_json)
-      "AI OpenAI-compatible streaming request operation=#{inspect(Map.get(request, :operation))} model=#{inspect(request.model)} url=#{url} payload_size=#{byte_size(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(
+        log_http_status_error("AI OpenAI-compatible streaming", status, body)
-          "AI OpenAI-compatible streaming HTTP error status=#{status} body=#{String.slice(body, 0, 2000)}"
+        http_status_error(status, body)
-        )
-
-        {:error, %{kind: :http_error, status: status, body: body}}
 
       {:error, reason} ->
-        Logger.error("AI OpenAI-compatible streaming request failed: #{inspect(reason)}")
+        log_http_request_error("AI OpenAI-compatible streaming", reason)
-        {:error, %{kind: :http_error, reason: 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
 

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)