path: root/spec/dispatch/adapter/claude/chat_streaming_retry_spec.rb

# frozen_string_literal: true

require "webmock/rspec"

RSpec.describe Dispatch::Adapter::Claude, "#chat (streaming retry)" do
  let(:model_id)  { "claude-sonnet-4-6" }
  let(:api_key)   { "sk-ant-api03-test" }
  let(:base_url)  { "https://api.anthropic.com" }

  subject(:adapter) do
    described_class.new(
      model: model_id,
      api_key: api_key,
      base_url: base_url
    )
  end

  let(:messages) do
    [Dispatch::Adapter::Message.new(
      role: "user",
      content: [Dispatch::Adapter::TextBlock.new(text: "Hello")]
    )]
  end

  before do
    WebMock.disable_net_connect!
    # Suppress actual sleep to keep specs fast
    allow(adapter).to receive(:sleep)
  end

  after { WebMock.reset! }

  # ── Helpers ───────────────────────────────────────────────────────────────

  # A complete, valid SSE stream with text content.
  def complete_sse_stream(text: "Hello!", input_tokens: 10, output_tokens: 5)
    <<~SSE
      event: message_start
      data: {"type":"message_start","message":{"id":"msg_01","model":"#{model_id}","usage":{"input_tokens":#{input_tokens},"output_tokens":0}}}

      event: content_block_start
      data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

      event: content_block_delta
      data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"#{text}"}}

      event: content_block_stop
      data: {"type":"content_block_stop","index":0}

      event: message_delta
      data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":#{output_tokens}}}

      event: message_stop
      data: {"type":"message_stop"}

    SSE
  end

  def stub_stream(body:, status: 200, times: 1)
    stub_request(:post, "#{base_url}/v1/messages")
      .to_return(
        status: status,
        body: body,
        headers: { "Content-Type" => "text/event-stream" }
      ).times(times)
  end

  def stub_streams(*responses)
    # Chain multiple responses for successive retry attempts
    responses.reduce(
      stub_request(:post, "#{base_url}/v1/messages")
    ) { |stub, (status, body)| stub.to_return(status: status, body: body, headers: {}) }
  end

  # ── Happy path ────────────────────────────────────────────────────────────

  describe "happy path (no retry needed)" do
    before { stub_stream(body: complete_sse_stream) }

    it "returns a Response" do
      response = adapter.chat(messages, stream: true)
      expect(response).to be_a(Dispatch::Adapter::Response)
    end

    it "yields StreamDelta events to the block" do
      deltas = []
      adapter.chat(messages, stream: true) { |d| deltas << d }
      expect(deltas.map(&:type)).to include(:text_start, :text_delta, :text_end)
    end

    it "does not call sleep (no retry)" do
      adapter.chat(messages, stream: true)
      expect(adapter).not_to have_received(:sleep)
    end
  end

  # ── First-event timeout retry ─────────────────────────────────────────────

  describe "first-event timeout" do
    # We stub current_time_ms to simulate time advancing:
    # first call → t=0 (request_started_at)
    # subsequent calls → t > deadline, triggering timeout
    before do
      allow(adapter).to receive(:stream_first_event_timeout_ms).and_return(1_000) # 1 second

      # First attempt: "hangs" (empty body, no message_start) — we'll
      # simulate timeout by making time_ms advance past the deadline
      call_count = 0
      allow(adapter).to receive(:current_time_ms) do
        call_count += 1
        # First call: t=0 (baseline)
        # All subsequent calls: t=2000 (past deadline)
        call_count == 1 ? 0 : 2_000
      end

      # First attempt: a stream that sends a chunk but never sees message_start
      # Second attempt (and beyond): complete stream
      stub_request(:post, "#{base_url}/v1/messages")
        .to_return(
          { status: 200,
            body: "event: ping\ndata: {}\n\n",
            headers: { "Content-Type" => "text/event-stream" } },
          { status: 200,
            body: complete_sse_stream,
            headers: { "Content-Type" => "text/event-stream" } }
        )
    end

    it "retries when first event times out and eventually returns a Response" do
      response = adapter.chat(messages, stream: true)
      expect(response).to be_a(Dispatch::Adapter::Response)
    end

    it "calls sleep between retries" do
      adapter.chat(messages, stream: true)
      expect(adapter).to have_received(:sleep).at_least(:once)
    end
  end

  # ── Stream ends before message_start ──────────────────────────────────────

  describe "stream ends before message_start" do
    before do
      stub_request(:post, "#{base_url}/v1/messages")
        .to_return(
          # 3 failures (empty streams), then a success
          { status: 200, body: "", headers: {} },
          { status: 200, body: "", headers: {} },
          { status: 200, body: "", headers: {} },
          { status: 200, body: complete_sse_stream,
            headers: { "Content-Type" => "text/event-stream" } }
        )
    end

    it "retries and eventually succeeds" do
      response = adapter.chat(messages, stream: true)
      expect(response).to be_a(Dispatch::Adapter::Response)
      expect(response.stop_reason).to eq(:end_turn)
    end

    it "sleeps between retries (exponential backoff)" do
      adapter.chat(messages, stream: true)
      expect(adapter).to have_received(:sleep).exactly(3).times
    end
  end

  # ── Stream ends before terminal ───────────────────────────────────────────

  describe "stream ends without message_stop or message_delta" do
    let(:truncated_stream) do
      <<~SSE
        event: message_start
        data: {"type":"message_start","message":{"id":"msg_02","model":"#{model_id}","usage":{"input_tokens":5,"output_tokens":0}}}

      SSE
    end

    before do
      stub_request(:post, "#{base_url}/v1/messages")
        .to_return(
          { status: 200, body: truncated_stream,
            headers: { "Content-Type" => "text/event-stream" } },
          { status: 200, body: complete_sse_stream,
            headers: { "Content-Type" => "text/event-stream" } }
        )
    end

    it "retries and returns a Response" do
      response = adapter.chat(messages, stream: true)
      expect(response).to be_a(Dispatch::Adapter::Response)
      expect(response.stop_reason).to eq(:end_turn)
    end
  end

  # ── Retry exhausted: give-up returns :error Response ─────────────────────

  describe "retry exhausted (all attempts fail before message_start)" do
    before do
      stub_request(:post, "#{base_url}/v1/messages")
        .to_return(status: 200, body: "", headers: {})
        .times(4) # initial + 3 retries = 4 total, all empty
    end

    it "returns a Response with stop_reason: :error" do
      response = adapter.chat(messages, stream: true)
      expect(response.stop_reason).to eq(:error)
    end

    it "does not raise" do
      expect { adapter.chat(messages, stream: true) }.not_to raise_error
    end

    it "sleeps exactly 3 times (one per retry)" do
      adapter.chat(messages, stream: true)
      expect(adapter).to have_received(:sleep).exactly(3).times
    end

    it "uses exponential backoff (2s, 4s, 8s)" do
      delays = []
      allow(adapter).to receive(:sleep) { |s| delays << s }
      adapter.chat(messages, stream: true)
      # Base 2000ms × 2^(attempt-1); converted to seconds
      expect(delays).to eq([2.0, 4.0, 8.0])
    end
  end

  # ── Mid-stream JSON corruption ────────────────────────────────────────────

  describe "mid-stream JSON corruption BEFORE any text deltas" do
    let(:corrupted_stream_no_output) do
      # message_start arrives but no text deltas, then invalid SSE/JSON
      <<~SSE
        event: message_start
        data: {"type":"message_start","message":{"id":"msg_03","model":"#{model_id}","usage":{"input_tokens":5,"output_tokens":0}}}

        event: content_block_start
        data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

        event: content_block_delta
        data: {BROKEN JSON

      SSE
    end

    before do
      stub_request(:post, "#{base_url}/v1/messages")
        .to_return(
          { status: 200,
            body: corrupted_stream_no_output,
            headers: { "Content-Type" => "text/event-stream" } },
          { status: 200,
            body: complete_sse_stream,
            headers: { "Content-Type" => "text/event-stream" } }
        )
    end

    it "retries and eventually succeeds when no output was emitted yet" do
      response = adapter.chat(messages, stream: true)
      # After retry succeeds we get a normal response
      expect(response).to be_a(Dispatch::Adapter::Response)
    end
  end

  describe "mid-stream JSON corruption AFTER text deltas emitted" do
    # A stream where text delta IS emitted to consumer (has_consumer_output? → true),
    # then the stream has a broken JSON frame → RequestError.
    # Since consumer output already happened, we do NOT retry.
    let(:corrupted_stream_with_output) do
      <<~SSE
        event: message_start
        data: {"type":"message_start","message":{"id":"msg_04","model":"#{model_id}","usage":{"input_tokens":5,"output_tokens":0}}}

        event: content_block_start
        data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

        event: content_block_delta
        data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}}

        event: content_block_delta
        data: {BROKEN JSON AFTER TEXT

      SSE
    end

    before do
      stub_stream(body: corrupted_stream_with_output)
    end

    it "does NOT retry when output has been emitted — raises the error" do
      # RequestError from parse_frame; consumer_output? is true → not safe to retry → raises
      expect { adapter.chat(messages, stream: true) }
        .to raise_error(Dispatch::Adapter::RequestError, /invalid JSON|incomplete frame/i)
    end

    it "does not call sleep (no retry)" do
      begin
        adapter.chat(messages, stream: true)
      rescue StandardError
        nil
      end
      expect(adapter).not_to have_received(:sleep)
    end
  end

  # ── Connection error before any output ────────────────────────────────────

  describe "ConnectionError before message_start" do
    # Simulate a connection error that causes ConnectionError (wrapped by HttpClient),
    # then a successful response on the retry.
    before do
      # First call: make the stub raise a connection-level error
      stub_request(:post, "#{base_url}/v1/messages")
        .to_return(
          { status: 200, body: "",
            headers: { "Content-Type" => "text/event-stream" } },
          { status: 200, body: complete_sse_stream,
            headers: { "Content-Type" => "text/event-stream" } }
        )
    end

    it "retries when stream ends before message_start and succeeds" do
      # An empty response also triggers :no_message_start retry
      response = adapter.chat(messages, stream: true)
      expect(response.stop_reason).to eq(:end_turn)
    end
  end

  # ── Non-retriable error (e.g. 401) ────────────────────────────────────────

  describe "AuthenticationError is not retried" do
    before do
      stub_request(:post, "#{base_url}/v1/messages")
        .to_return(
          status: 401,
          body: JSON.generate({ "error" => { "message" => "Unauthorized" } }),
          headers: { "Content-Type" => "application/json" }
        )
    end

    it "raises AuthenticationError without retrying" do
      expect { adapter.chat(messages, stream: true) }
        .to raise_error(Dispatch::Adapter::AuthenticationError)
    end

    it "does not sleep (no retry)" do
      begin
        adapter.chat(messages, stream: true)
      rescue StandardError
        nil
      end
      expect(adapter).not_to have_received(:sleep)
    end
  end

  # ── Constants accessible ──────────────────────────────────────────────────

  describe "constants" do
    it "STREAM_MAX_RETRIES is 3" do
      expect(described_class::STREAM_MAX_RETRIES).to eq(3)
    end

    it "STREAM_BASE_DELAY_MS is 2000" do
      expect(described_class::STREAM_BASE_DELAY_MS).to eq(2_000)
    end

    it "STREAM_FIRST_EVENT_TIMEOUT_MS is a positive integer" do
      expect(described_class::STREAM_FIRST_EVENT_TIMEOUT_MS).to be > 0
    end
  end
end