fix: prevent AI assistant crash on non-2xx streaming responses

Streaming AI requests keep the response body as a lazy stream, but the
error handler assumed a decoded map and keyed into it directly. A
non-success status from a streaming endpoint therefore crashed the
handler and discarded the background job, masking the real error.

Return a clean error message unless the body is a plain map with a
string "message"; any other shape (a stream body, or a nested or
non-string message) falls back to a generic message, and the failing
HTTP status is logged.

Fixes #4941

Author: elias-ba

CHANGELOG.md

@@ -37,6 +37,10 @@ and this project adheres to
 
 ### Fixed
 
+- Return a clean error instead of crashing when the AI assistant's streaming
+  endpoint responds with a non-success status.
+  [#4941](https://github.com/OpenFn/lightning/issues/4941)
+
 ## [2.16.8] - 2026-07-01
 
 ## [2.16.8-pre] - 2026-06-18

lib/lightning/ai_assistant/ai_assistant.ex

@@ -1434,10 +1434,10 @@ defmodule Lightning.AiAssistant do
     case error_response do
       {:ok, %Tesla.Env{status: status, body: body}}
       when status not in @success_status_range ->
-        error_message = body["message"]
+        error_message = error_message_from_body(body)
 
         Logger.error(
-          "AI query failed for session #{session.id}: #{error_message}"
+          "AI query failed for session #{session.id} (status #{status}): #{error_message}"
         )
 
         {:error, error_message}
@@ -1459,6 +1459,16 @@ defmodule Lightning.AiAssistant do
     end
   end
 
+  # A streaming body is a lazy Stream and a JSON "message" may be a nested
+  # object, so only a plain-map body with a string message is safe to return.
+  defp error_message_from_body(body) do
+    message = if is_map(body) and not is_struct(body), do: body["message"]
+
+    if is_binary(message),
+      do: message,
+      else: "Oops! Something went wrong. Please try again."
+  end
+
   defp build_job_message(body) do
     message = body["history"] |> Enum.reverse() |> hd()
     message_attrs = Map.take(message, ["role", "content"])

test/lightning/ai_assistant/ai_assistant_test.exs

@@ -3392,6 +3392,39 @@ defmodule Lightning.AiAssistantTest do
       assert_received {:ai_assistant, :streaming_error,
                        %{error: "Streaming connection lost", session_id: _}}
     end
+
+    test "returns a clean binary error tuple for a non-2xx response with any body shape",
+         %{user: user, workflow: %{jobs: [job_1 | _]}} do
+      session =
+        insert(:chat_session,
+          user: user,
+          job: job_1,
+          messages: [
+            %{
+              role: :user,
+              content: "help me",
+              user: user,
+              status: :pending,
+              inserted_at: DateTime.utc_now() |> DateTime.add(-1)
+            }
+          ]
+        )
+
+      # Each shape (Stream function, %Stream{} struct, nested-message map)
+      # previously crashed the handler; assert it now yields a binary error.
+      for body <- [
+            Stream.unfold(:done, fn _ -> nil end),
+            Stream.map([], & &1),
+            %{"message" => %{"detail" => "invalid input"}}
+          ] do
+        expect(Lightning.Tesla.Mock, :call, fn _env, _opts ->
+          {:ok, %Tesla.Env{status: 500, body: body}}
+        end)
+
+        assert {:error, message} = AiAssistant.query_stream(session, "help me")
+        assert is_binary(message)
+      end
+    end
   end
 
   describe "query_workflow_stream/3" do