fix: halt stream on terminal AssistantMessage errors

lib/claude_code/stream.ex

@@ -565,16 +565,21 @@ defmodule ClaudeCode.Stream do
     # Use receive_next to get messages from the session (pull-based)
     case GenServer.call(state.session, {:receive_next, state.request_ref}, state.timeout) do
       {:message, message} ->
-        if should_emit?(message, state.filter) do
-          case message do
-            %Message.ResultMessage{} ->
-              {[message], %{state | done: true}}
-
-            _ ->
-              {[message], state}
-          end
-        else
-          next_message(state)
+        cond do
+          terminal_error?(message) ->
+            {[synthesize_error_result(message)], %{state | done: true}}
+
+          should_emit?(message, state.filter) ->
+            case message do
+              %Message.ResultMessage{} ->
+                {[message], %{state | done: true}}
+
+              _ ->
+                {[message], state}
+            end
+
+          true ->
+            next_message(state)
         end
 
       :done ->
@@ -603,6 +608,25 @@ defmodule ClaudeCode.Stream do
   defp process_alive?({name, _node} = ref) when is_atom(name), do: GenServer.whereis(ref) != nil
   defp process_alive?(_), do: false
 
+  defp terminal_error?(%Message.AssistantMessage{error: error}) when not is_nil(error), do: true
+  defp terminal_error?(_), do: false
+
+  defp synthesize_error_result(%Message.AssistantMessage{} = msg) do
+    %Message.ResultMessage{
+      type: :result,
+      subtype: msg.error,
+      is_error: true,
+      duration_ms: 0.0,
+      duration_api_ms: 0.0,
+      num_turns: 0,
+      result: to_string(msg),
+      session_id: msg.session_id,
+      total_cost_usd: 0.0,
+      usage: %{},
+      uuid: msg.uuid
+    }
+  end
+
   defp should_emit?(_message, :all), do: true
 
   defp should_emit?(message, filter) do

test/claude_code/session_adapter_test.exs

@@ -1,6 +1,7 @@
 defmodule ClaudeCode.SessionAdapterTest do
   use ExUnit.Case, async: true
 
+  alias ClaudeCode.Message.ResultMessage
   alias ClaudeCode.Session
   alias ClaudeCode.Test.Factory
 
@@ -382,4 +383,149 @@ defmodule ClaudeCode.SessionAdapterTest do
       GenServer.stop(session)
     end
   end
+
+  # ============================================================================
+  # Stream halts on terminal AssistantMessage errors (issue #49)
+  # ============================================================================
+
+  describe "stream halts on terminal assistant errors" do
+    defmodule RateLimitLoopAdapter do
+      @moduledoc false
+      @behaviour ClaudeCode.Adapter
+
+      use GenServer
+
+      alias ClaudeCode.Adapter
+      alias ClaudeCode.Test.Factory
+
+      @impl ClaudeCode.Adapter
+      def start_link(session, opts), do: GenServer.start_link(__MODULE__, {session, opts})
+
+      @impl ClaudeCode.Adapter
+      def send_query(adapter, request_id, _prompt, _opts) do
+        GenServer.cast(adapter, {:query, request_id})
+        :ok
+      end
+
+      @impl ClaudeCode.Adapter
+      def health(_adapter), do: :healthy
+
+      @impl ClaudeCode.Adapter
+      def stop(adapter), do: GenServer.stop(adapter, :normal)
+
+      @impl GenServer
+      def init({session, _opts}) do
+        Process.link(session)
+        Adapter.notify_status(session, :ready)
+        {:ok, %{session: session}}
+      end
+
+      @impl GenServer
+      def handle_cast({:query, request_id}, state) do
+        user_msg =
+          Factory.user_message(
+            message: %{
+              content: [
+                Factory.text_block(
+                  text: "Stop hook feedback:\nYou MUST call the StructuredOutput tool to complete this request."
+                )
+              ]
+            }
+          )
+
+        error_msg =
+          Factory.assistant_message(
+            error: :rate_limit,
+            message: %{
+              model: "<synthetic>",
+              content: [Factory.text_block(text: "You're out of extra usage")],
+              stop_reason: :stop_sequence
+            }
+          )
+
+        for _ <- 1..10 do
+          Adapter.notify_message(state.session, request_id, user_msg)
+          Adapter.notify_message(state.session, request_id, error_msg)
+        end
+
+        {:noreply, state}
+      end
+    end
+
+    test "stream terminates when assistant message has a terminal error" do
+      {:ok, session} = Session.start_link(adapter: {RateLimitLoopAdapter, []})
+
+      messages =
+        session
+        |> ClaudeCode.stream("test")
+        |> Enum.to_list()
+
+      error_msg = List.last(messages)
+      assert %ResultMessage{is_error: true, subtype: :rate_limit} = error_msg
+      assert error_msg.result == "You're out of extra usage"
+
+      GenServer.stop(session)
+    end
+
+    defmodule BillingErrorAdapter do
+      @moduledoc false
+      @behaviour ClaudeCode.Adapter
+
+      use GenServer
+
+      alias ClaudeCode.Adapter
+      alias ClaudeCode.Test.Factory
+
+      @impl ClaudeCode.Adapter
+      def start_link(session, opts), do: GenServer.start_link(__MODULE__, {session, opts})
+
+      @impl ClaudeCode.Adapter
+      def send_query(adapter, request_id, _prompt, _opts) do
+        GenServer.cast(adapter, {:query, request_id})
+        :ok
+      end
+
+      @impl ClaudeCode.Adapter
+      def health(_adapter), do: :healthy
+
+      @impl ClaudeCode.Adapter
+      def stop(adapter), do: GenServer.stop(adapter, :normal)
+
+      @impl GenServer
+      def init({session, _opts}) do
+        Process.link(session)
+        Adapter.notify_status(session, :ready)
+        {:ok, %{session: session}}
+      end
+
+      @impl GenServer
+      def handle_cast({:query, request_id}, state) do
+        Adapter.notify_message(
+          state.session,
+          request_id,
+          Factory.assistant_message(
+            error: :billing_error,
+            message: %{content: [Factory.text_block(text: "Billing issue")]}
+          )
+        )
+
+        {:noreply, state}
+      end
+    end
+
+    test "stream terminates on other error types" do
+      {:ok, session} = Session.start_link(adapter: {BillingErrorAdapter, []})
+
+      messages =
+        session
+        |> ClaudeCode.stream("test")
+        |> Enum.to_list()
+
+      assert length(messages) == 1
+      assert %ResultMessage{is_error: true, subtype: :billing_error} = hd(messages)
+      assert hd(messages).result == "Billing issue"
+
+      GenServer.stop(session)
+    end
+  end
 end