diff --git a/lib/claude_code/stream.ex b/lib/claude_code/stream.ex index 96c58ad..e8f7eff 100644 --- a/lib/claude_code/stream.ex +++ b/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 diff --git a/test/claude_code/session_adapter_test.exs b/test/claude_code/session_adapter_test.exs index 53e622e..2978feb 100644 --- a/test/claude_code/session_adapter_test.exs +++ b/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: "", + 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