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

# frozen_string_literal: true

RSpec.describe Dispatch::Adapter::Claude::StreamCollector do
  let(:model_id) { "claude-sonnet-4-6" }
  subject(:collector) { described_class.new(model_id) }

  # Helper: seed message_start so content-block events don't raise
  def seed_message_start(col = collector, id: "msg_01", model: nil, input: 10, output: 0)
    msg = { "id" => id, "usage" => { "input_tokens" => input, "output_tokens" => output } }
    msg["model"] = model if model
    col.handle("message_start", { "message" => msg })
  end

  # Helper: collect all deltas from a sequence of [type, data] pairs
  def collect_deltas(col, events)
    deltas = []
    events.each { |type, data| col.handle(type, data) { |d| deltas << d } }
    deltas
  end

  # ── Initial state ─────────────────────────────────────────────────────────

  describe "#initialize" do
    it "seeds the model from the supplied model_id" do
      expect(collector.model).to eq(model_id)
    end

    it "starts with saw_message_start? == false" do
      expect(collector.saw_message_start?).to be false
    end

    it "starts with a nil response_id" do
      expect(collector.response_id).to be_nil
    end

    it "starts with all usage counters at zero" do
      expect(collector.usage).to eq(
        input: 0, output: 0, cache_read: 0, cache_creation: 0
      )
    end

    it "starts with an empty state[:content_blocks]" do
      expect(collector.state[:content_blocks]).to be_empty
    end

    it "starts with saw_terminal == false" do
      expect(collector.state[:saw_terminal]).to be false
    end

    it "starts with finish_reason == nil" do
      expect(collector.state[:finish_reason]).to be_nil
    end
  end

  # ── message_start handling ────────────────────────────────────────────────

  describe "#handle — message_start" do
    let(:message_start_data) do
      {
        "type" => "message_start",
        "message" => {
          "id" => "msg_01XYZ",
          "model" => "claude-sonnet-4-6-20251101",
          "usage" => {
            "input_tokens" => 42,
            "output_tokens" => 1,
            "cache_read_input_tokens" => 100,
            "cache_creation_input_tokens" => 200
          }
        }
      }
    end

    before { collector.handle("message_start", message_start_data) }

    it "marks saw_message_start? as true" do
      expect(collector.saw_message_start?).to be true
    end

    it "stores the response_id from message.id" do
      expect(collector.response_id).to eq("msg_01XYZ")
    end

    it "updates the model to the value from message.model" do
      expect(collector.model).to eq("claude-sonnet-4-6-20251101")
    end

    it "copies input_tokens into usage[:input]" do
      expect(collector.usage[:input]).to eq(42)
    end

    it "copies output_tokens into usage[:output]" do
      expect(collector.usage[:output]).to eq(1)
    end

    it "copies cache_read_input_tokens into usage[:cache_read]" do
      expect(collector.usage[:cache_read]).to eq(100)
    end

    it "copies cache_creation_input_tokens into usage[:cache_creation]" do
      expect(collector.usage[:cache_creation]).to eq(200)
    end

    it "does NOT yield a StreamDelta" do
      deltas = []
      col2 = described_class.new(model_id)
      col2.handle("message_start", message_start_data) { |d| deltas << d }
      expect(deltas).to be_empty
    end
  end

  # ── ping events ───────────────────────────────────────────────────────────

  describe "#handle — ping events are silently ignored" do
    it "does not raise when ping arrives before message_start" do
      expect { collector.handle("ping", {}) }.not_to raise_error
    end

    it "does not set saw_message_start? after a ping" do
      collector.handle("ping", {})
      expect(collector.saw_message_start?).to be false
    end

    it "does not yield a delta for a ping event" do
      deltas = []
      collector.handle("ping", {}) { |d| deltas << d }
      expect(deltas).to be_empty
    end
  end

  # ── Out-of-order envelope guard ───────────────────────────────────────────

  describe "#handle — events before message_start raise RequestError" do
    %w[content_block_start content_block_delta content_block_stop
       message_delta message_stop].each do |bad_type|
      it "raises RequestError for #{bad_type.inspect} before message_start" do
        expect do
          collector.handle(bad_type, { "type" => bad_type })
        end.to raise_error(
          Dispatch::Adapter::RequestError,
          /stream envelope: received #{Regexp.escape(bad_type.inspect)} before message_start/
        )
      end
    end

    it "does not raise for message_start itself" do
      expect do
        collector.handle("message_start", { "message" => { "id" => "msg_ok", "usage" => {} } })
      end.not_to raise_error
    end
  end

  # ── text block lifecycle ──────────────────────────────────────────────────

  describe "text block lifecycle" do
    before { seed_message_start }

    let(:block_start) do
      {
        "index" => 0,
        "content_block" => { "type" => "text", "text" => "" }
      }
    end

    let(:delta1) do
      { "index" => 0, "delta" => { "type" => "text_delta", "text" => "Hello" } }
    end

    let(:delta2) do
      { "index" => 0, "delta" => { "type" => "text_delta", "text" => " World" } }
    end

    let(:block_stop) { { "index" => 0 } }

    it "yields :text_start on content_block_start" do
      deltas = []
      collector.handle("content_block_start", block_start) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:text_start])
    end

    it "yields :text_delta with correct text" do
      collector.handle("content_block_start", block_start)
      deltas = []
      collector.handle("content_block_delta", delta1) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:text_delta])
      expect(deltas.first.text).to eq("Hello")
    end

    it "yields :text_end on content_block_stop" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", delta1)
      deltas = []
      collector.handle("content_block_stop", block_stop) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:text_end])
    end

    it "concatenates multiple text deltas in the block" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", delta1)
      collector.handle("content_block_delta", delta2)
      blk = collector.content_blocks.first
      expect(blk[:text]).to eq("Hello World")
    end

    it "appends a text block with correct kind" do
      collector.handle("content_block_start", block_start)
      blk = collector.content_blocks.first
      expect(blk[:kind]).to eq("text")
    end

    it "records the correct block index" do
      collector.handle("content_block_start", block_start)
      expect(collector.content_blocks.first[:index]).to eq(0)
    end

    it "yields the full delta sequence in order" do
      events = [
        ["content_block_start", block_start],
        ["content_block_delta", delta1],
        ["content_block_delta", delta2],
        ["content_block_stop",  block_stop]
      ]
      types = collect_deltas(collector, events).map(&:type)
      expect(types).to eq(%i[text_start text_delta text_delta text_end])
    end
  end

  # ── thinking block lifecycle ──────────────────────────────────────────────

  describe "thinking block lifecycle" do
    before { seed_message_start }

    let(:block_start) do
      { "index" => 0, "content_block" => { "type" => "thinking", "thinking" => "" } }
    end

    let(:thinking_delta) do
      { "index" => 0, "delta" => { "type" => "thinking_delta", "thinking" => "step 1" } }
    end

    let(:thinking_delta2) do
      { "index" => 0, "delta" => { "type" => "thinking_delta", "thinking" => " step 2" } }
    end

    let(:signature_delta) do
      { "index" => 0, "delta" => { "type" => "signature_delta", "signature" => "sig_abc" } }
    end

    let(:block_stop) { { "index" => 0 } }

    it "yields :thinking_start on content_block_start" do
      deltas = []
      collector.handle("content_block_start", block_start) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:thinking_start])
    end

    it "yields :thinking_delta with the thinking text" do
      collector.handle("content_block_start", block_start)
      deltas = []
      collector.handle("content_block_delta", thinking_delta) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:thinking_delta])
      expect(deltas.first.text).to eq("step 1")
    end

    it "yields :thinking_end on content_block_stop" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", thinking_delta)
      deltas = []
      collector.handle("content_block_stop", block_stop) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:thinking_end])
    end

    it "accumulates thinking text across multiple deltas" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", thinking_delta)
      collector.handle("content_block_delta", thinking_delta2)
      blk = collector.content_blocks.first
      expect(blk[:thinking]).to eq("step 1 step 2")
    end

    it "does NOT yield a StreamDelta for signature_delta" do
      collector.handle("content_block_start", block_start)
      deltas = []
      collector.handle("content_block_delta", signature_delta) { |d| deltas << d }
      expect(deltas).to be_empty
    end

    it "accumulates the signature in the block" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", signature_delta)
      collector.handle("content_block_delta",
                       { "index" => 0, "delta" => { "type" => "signature_delta", "signature" => "XYZ" } })
      blk = collector.content_blocks.first
      expect(blk[:signature]).to eq("sig_abcXYZ")
    end

    it "uses kind 'thinking' in the block" do
      collector.handle("content_block_start", block_start)
      expect(collector.content_blocks.first[:kind]).to eq("thinking")
    end
  end

  # ── tool_use block lifecycle ──────────────────────────────────────────────

  describe "tool_use block lifecycle" do
    before { seed_message_start }

    let(:block_start) do
      {
        "index" => 0,
        "content_block" => { "type" => "tool_use", "id" => "toolu_01", "name" => "bash" }
      }
    end

    let(:json_delta1) do
      { "index" => 0, "delta" => { "type" => "input_json_delta", "partial_json" => "{\"cmd\":" } }
    end

    let(:json_delta2) do
      { "index" => 0, "delta" => { "type" => "input_json_delta", "partial_json" => "\"ls\"}" } }
    end

    let(:block_stop) { { "index" => 0 } }

    it "yields :tool_use_start with id and name" do
      deltas = []
      collector.handle("content_block_start", block_start) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:tool_use_start])
      expect(deltas.first.tool_call_id).to eq("toolu_01")
      expect(deltas.first.tool_name).to eq("bash")
    end

    it "yields :tool_use_delta with the partial json fragment" do
      collector.handle("content_block_start", block_start)
      deltas = []
      collector.handle("content_block_delta", json_delta1) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:tool_use_delta])
      expect(deltas.first.argument_delta).to eq("{\"cmd\":")
    end

    it "yields :tool_use_end on content_block_stop" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", json_delta1)
      collector.handle("content_block_delta", json_delta2)
      deltas = []
      collector.handle("content_block_stop", block_stop) { |d| deltas << d }
      expect(deltas.map(&:type)).to eq([:tool_use_end])
    end

    it "parses the accumulated partial JSON into arguments at stop" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", json_delta1)
      collector.handle("content_block_delta", json_delta2)
      collector.handle("content_block_stop",  block_stop)
      blk = collector.content_blocks.first
      expect(blk[:arguments]).to eq({ "cmd" => "ls" })
    end

    it "accumulates partial_json across deltas" do
      collector.handle("content_block_start", block_start)
      collector.handle("content_block_delta", json_delta1)
      blk = collector.content_blocks.first
      expect(blk[:partial_json]).to eq("{\"cmd\":")
    end

    it "stores kind 'tool_use' in the block" do
      collector.handle("content_block_start", block_start)
      expect(collector.content_blocks.first[:kind]).to eq("tool_use")
    end

    describe "partial JSON parse failure at tool_use_end" do
      let(:bad_delta) do
        { "index" => 0, "delta" => { "type" => "input_json_delta", "partial_json" => "{broken" } }
      end

      it "falls back to {} on JSON::ParserError without raising" do
        collector.handle("content_block_start", block_start)
        collector.handle("content_block_delta", bad_delta)
        expect { collector.handle("content_block_stop", block_stop) }.not_to raise_error
        blk = collector.content_blocks.first
        expect(blk[:arguments]).to eq({})
      end
    end

    describe "OAuth proxy_ prefix stripping" do
      let(:oauth_collector) { described_class.new(model_id, is_oauth: true) }

      before { seed_message_start(oauth_collector) }

      let(:proxy_block_start) do
        {
          "index" => 0,
          "content_block" => { "type" => "tool_use", "id" => "toolu_02", "name" => "proxy_bash" }
        }
      end

      it "strips proxy_ from the tool name in :tool_use_start delta" do
        deltas = []
        oauth_collector.handle("content_block_start", proxy_block_start) { |d| deltas << d }
        expect(deltas.first.tool_name).to eq("bash")
      end

      it "stores the stripped name in the block" do
        oauth_collector.handle("content_block_start", proxy_block_start)
        blk = oauth_collector.content_blocks.first
        expect(blk[:name]).to eq("bash")
      end

      it "does NOT strip proxy_ when is_oauth: false" do
        non_oauth = described_class.new(model_id, is_oauth: false)
        seed_message_start(non_oauth)
        deltas = []
        non_oauth.handle("content_block_start", proxy_block_start) { |d| deltas << d }
        expect(deltas.first.tool_name).to eq("proxy_bash")
      end
    end
  end

  # ── redacted_thinking block lifecycle ────────────────────────────────────

  describe "redacted_thinking block lifecycle" do
    before { seed_message_start }

    let(:block_start) do
      {
        "index" => 0,
        "content_block" => { "type" => "redacted_thinking", "data" => "REDACTED" }
      }
    end

    let(:block_stop) { { "index" => 0 } }

    it "does NOT yield any StreamDelta for content_block_start" do
      deltas = []
      collector.handle("content_block_start", block_start) { |d| deltas << d }
      expect(deltas).to be_empty
    end

    it "does NOT yield any StreamDelta for content_block_stop" do
      collector.handle("content_block_start", block_start)
      deltas = []
      collector.handle("content_block_stop", block_stop) { |d| deltas << d }
      expect(deltas).to be_empty
    end

    it "appends a redacted_thinking block with data from content_block" do
      collector.handle("content_block_start", block_start)
      blk = collector.content_blocks.first
      expect(blk[:kind]).to eq("redacted_thinking")
      expect(blk[:data]).to eq("REDACTED")
    end
  end

  # ── multiple blocks in one stream ─────────────────────────────────────────

  describe "multiple concurrent/sequential blocks" do
    before { seed_message_start }

    it "tracks two text blocks by separate indices" do
      collector.handle("content_block_start",
                       { "index" => 0, "content_block" => { "type" => "text", "text" => "" } })
      collector.handle("content_block_start",
                       { "index" => 1, "content_block" => { "type" => "text", "text" => "" } })
      collector.handle("content_block_delta",
                       { "index" => 0, "delta" => { "type" => "text_delta", "text" => "A" } })
      collector.handle("content_block_delta",
                       { "index" => 1, "delta" => { "type" => "text_delta", "text" => "B" } })

      expect(collector.content_blocks[0][:text]).to eq("A")
      expect(collector.content_blocks[1][:text]).to eq("B")
    end

    it "handles a text block followed by a tool_use block" do
      collector.handle("content_block_start",
                       { "index" => 0, "content_block" => { "type" => "text", "text" => "" } })
      collector.handle("content_block_delta",
                       { "index" => 0, "delta" => { "type" => "text_delta", "text" => "thinking" } })
      collector.handle("content_block_stop", { "index" => 0 })

      collector.handle("content_block_start",
                       { "index" => 1,
                         "content_block" => { "type" => "tool_use", "id" => "tc1", "name" => "find" } })
      collector.handle("content_block_delta",
                       { "index" => 1,
                         "delta" => { "type" => "input_json_delta", "partial_json" => "{}" } })
      collector.handle("content_block_stop", { "index" => 1 })

      expect(collector.content_blocks.size).to eq(2)
      expect(collector.content_blocks[0][:kind]).to eq("text")
      expect(collector.content_blocks[1][:kind]).to eq("tool_use")
      expect(collector.content_blocks[1][:arguments]).to eq({})
    end
  end

  # ── message_delta handling ───────────────────────────────────────────────

  describe "#handle — message_delta" do
    before { seed_message_start(collector, input: 10, output: 0) }

    let(:message_delta_data) do
      {
        "type" => "message_delta",
        "delta" => { "stop_reason" => "end_turn", "stop_sequence" => nil },
        "usage" => { "output_tokens" => 42 }
      }
    end

    it "stores the stop_reason as finish_reason" do
      collector.handle("message_delta", message_delta_data)
      expect(collector.finish_reason).to eq("end_turn")
    end

    it "marks saw_terminal? as true" do
      collector.handle("message_delta", message_delta_data)
      expect(collector.saw_terminal?).to be true
    end

    it "updates output usage from message_delta usage" do
      collector.handle("message_delta", message_delta_data)
      expect(collector.usage[:output]).to eq(42)
    end

    it "preserves input usage when message_delta has no input_tokens" do
      collector.handle("message_delta", message_delta_data)
      expect(collector.usage[:input]).to eq(10)
    end

    it "does NOT yield a StreamDelta" do
      deltas = []
      collector.handle("message_delta", message_delta_data) { |d| deltas << d }
      expect(deltas).to be_empty
    end

    it "overrides message_start usage when message_delta provides values" do
      # message_start set output=0; message_delta should override to 42
      expect(collector.usage[:output]).to eq(0)
      collector.handle("message_delta", message_delta_data)
      expect(collector.usage[:output]).to eq(42)
    end

    context "when message_delta has all usage fields" do
      let(:full_usage_delta) do
        {
          "type" => "message_delta",
          "delta" => { "stop_reason" => "max_tokens" },
          "usage" => {
            "input_tokens" => 99,
            "output_tokens" => 77,
            "cache_read_input_tokens" => 55,
            "cache_creation_input_tokens" => 33
          }
        }
      end

      it "updates all four usage counters" do
        collector.handle("message_delta", full_usage_delta)
        expect(collector.usage[:input]).to eq(99)
        expect(collector.usage[:output]).to eq(77)
        expect(collector.usage[:cache_read]).to eq(55)
        expect(collector.usage[:cache_creation]).to eq(33)
      end

      it "records the correct finish_reason" do
        collector.handle("message_delta", full_usage_delta)
        expect(collector.finish_reason).to eq("max_tokens")
      end
    end

    context "when message_delta has no stop_reason" do
      let(:no_stop_delta) do
        {
          "type" => "message_delta",
          "delta" => {},
          "usage" => { "output_tokens" => 5 }
        }
      end

      it "leaves finish_reason as nil" do
        collector.handle("message_delta", no_stop_delta)
        expect(collector.finish_reason).to be_nil
      end

      it "still marks saw_terminal? as true" do
        collector.handle("message_delta", no_stop_delta)
        expect(collector.saw_terminal?).to be true
      end
    end

    context "when message_delta has no usage" do
      let(:no_usage_delta) do
        {
          "type" => "message_delta",
          "delta" => { "stop_reason" => "end_turn" },
          "usage" => {}
        }
      end

      it "preserves existing usage values" do
        collector.handle("message_delta", no_usage_delta)
        expect(collector.usage[:input]).to eq(10)
        expect(collector.usage[:output]).to eq(0)
      end
    end
  end

  # ── message_stop handling ─────────────────────────────────────────────────

  describe "#handle — message_stop" do
    before { seed_message_start }

    it "marks saw_terminal? as true" do
      collector.handle("message_stop", { "type" => "message_stop" })
      expect(collector.saw_terminal?).to be true
    end

    it "does NOT yield a StreamDelta" do
      deltas = []
      collector.handle("message_stop", { "type" => "message_stop" }) { |d| deltas << d }
      expect(deltas).to be_empty
    end

    it "does NOT change finish_reason" do
      collector.handle("message_stop", { "type" => "message_stop" })
      expect(collector.finish_reason).to be_nil
    end

    it "does NOT change usage" do
      collector.handle("message_stop", { "type" => "message_stop" })
      expect(collector.usage[:input]).to eq(10)
    end
  end

  # ── message_delta + message_stop together ─────────────────────────────────

  describe "#handle — message_delta followed by message_stop" do
    before do
      seed_message_start(collector, input: 20, output: 0)
      collector.handle("message_delta",
                       { "type" => "message_delta",
                         "delta" => { "stop_reason" => "tool_use" },
                         "usage" => { "output_tokens" => 15 } })
    end

    it "saw_terminal? is true after message_delta" do
      expect(collector.saw_terminal?).to be true
    end

    it "finish_reason is captured from message_delta" do
      expect(collector.finish_reason).to eq("tool_use")
    end

    it "output usage is updated from message_delta" do
      expect(collector.usage[:output]).to eq(15)
    end

    it "message_stop also marks saw_terminal? true (idempotent)" do
      collector.handle("message_stop", {})
      expect(collector.saw_terminal?).to be true
    end
  end

  # ── has_consumer_output? ─────────────────────────────────────────────────

  describe "#has_consumer_output?" do
    before { seed_message_start }

    it "returns false when no content blocks exist" do
      expect(collector.has_consumer_output?).to be false
    end

    it "returns false after content_block_start for text (no delta yet)" do
      collector.handle("content_block_start",
                       { "index" => 0, "content_block" => { "type" => "text", "text" => "" } })
      expect(collector.has_consumer_output?).to be false
    end

    it "returns true after a text_delta is accumulated" do
      collector.handle("content_block_start",
                       { "index" => 0, "content_block" => { "type" => "text", "text" => "" } })
      collector.handle("content_block_delta",
                       { "index" => 0, "delta" => { "type" => "text_delta", "text" => "hi" } })
      expect(collector.has_consumer_output?).to be true
    end

    it "returns false after tool_use_start with no JSON delta yet" do
      collector.handle("content_block_start",
                       { "index" => 0, "content_block" => {
                         "type" => "tool_use", "id" => "tc1", "name" => "bash"
                       } })
      expect(collector.has_consumer_output?).to be false
    end

    it "returns true after an input_json_delta is accumulated" do
      collector.handle("content_block_start",
                       { "index" => 0, "content_block" => {
                         "type" => "tool_use", "id" => "tc1", "name" => "bash"
                       } })
      collector.handle("content_block_delta",
                       { "index" => 0, "delta" => { "type" => "input_json_delta",
                                                    "partial_json" => "{" } })
      expect(collector.has_consumer_output?).to be true
    end

    it "returns false for a redacted_thinking block" do
      collector.handle("content_block_start",
                       { "index" => 0, "content_block" => {
                         "type" => "redacted_thinking", "data" => "DATA"
                       } })
      expect(collector.has_consumer_output?).to be false
    end
  end

  # ── Realistic SSE stream replay ───────────────────────────────────────────

  describe "realistic Anthropic SSE stream replay" do
    let(:sse_parser)  { Dispatch::Adapter::Claude::SseParser.new }
    subject(:col)     { described_class.new(model_id) }

    let(:stream) do
      <<~SSE
        event: message_start
        data: {"type":"message_start","message":{"id":"msg_replay","type":"message","role":"assistant","content":[],"model":"claude-sonnet-4-6","stop_reason":null,"usage":{"input_tokens":10,"output_tokens":0}}}

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

        event: ping
        data: {}

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

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

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

        event: message_delta
        data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":5}}

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

      SSE
    end

    it "produces the expected StreamDelta sequence" do
      deltas = []
      sse_parser.feed(stream) do |event_type, data|
        col.handle(event_type, data) { |d| deltas << d }
      end
      types = deltas.map(&:type)
      expect(types).to eq(%i[text_start text_delta text_delta text_end])
    end

    it "concatenates the text deltas into the final block text" do
      sse_parser.feed(stream) do |event_type, data|
        col.handle(event_type, data)
      end
      blk = col.content_blocks.first
      expect(blk[:text]).to eq("Hello World")
    end

    it "sets saw_message_start? to true" do
      sse_parser.feed(stream) { |t, d| col.handle(t, d) }
      expect(col.saw_message_start?).to be true
    end

    it "stores the correct response_id" do
      sse_parser.feed(stream) { |t, d| col.handle(t, d) }
      expect(col.response_id).to eq("msg_replay")
    end

    it "updates output usage from message_delta (not message_start)" do
      # message_start has output_tokens=0; message_delta has output_tokens=5
      sse_parser.feed(stream) { |t, d| col.handle(t, d) }
      expect(col.usage[:output]).to eq(5)
    end

    it "captures finish_reason from message_delta" do
      sse_parser.feed(stream) { |t, d| col.handle(t, d) }
      expect(col.finish_reason).to eq("end_turn")
    end

    it "marks saw_terminal? as true after message_stop" do
      sse_parser.feed(stream) { |t, d| col.handle(t, d) }
      expect(col.saw_terminal?).to be true
    end
  end

  # ── Realistic tool_use stream replay ─────────────────────────────────────

  describe "realistic tool_use SSE stream replay" do
    let(:sse_parser)  { Dispatch::Adapter::Claude::SseParser.new }
    subject(:col)     { described_class.new(model_id) }

    let(:tool_stream) do
      # NOTE: each `data:` line must be a single, valid JSON object. The two
      # partial_json fragments are JSON STRING VALUES that, when concatenated,
      # yield the final tool input `{"command":"ls -la"}`. Inner double-quotes
      # inside those string values therefore need backslash-escaping in the
      # outer JSON (\" inside the heredoc, which Ruby renders as a literal \"
      # in the emitted SSE bytes).
      <<~'SSE'
        event: message_start
        data: {"type":"message_start","message":{"id":"msg_tool","model":"claude-sonnet-4-6","usage":{"input_tokens":20,"output_tokens":0}}}

        event: content_block_start
        data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_99","name":"bash"}}

        event: content_block_delta
        data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"command\":\""}}

        event: content_block_delta
        data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"ls -la\"}"}}

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

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

      SSE
    end

    it "produces the expected StreamDelta sequence for tool_use" do
      deltas = []
      sse_parser.feed(tool_stream) do |event_type, data|
        col.handle(event_type, data) { |d| deltas << d }
      end
      types = deltas.map(&:type)
      expect(types).to eq(%i[tool_use_start tool_use_delta tool_use_delta tool_use_end])
    end

    it "parses tool arguments correctly" do
      sse_parser.feed(tool_stream) { |t, d| col.handle(t, d) }
      blk = col.content_blocks.first
      expect(blk[:arguments]).to eq({ "command" => "ls -la" })
    end

    it "sets the tool id and name on the block" do
      sse_parser.feed(tool_stream) { |t, d| col.handle(t, d) }
      blk = col.content_blocks.first
      expect(blk[:id]).to eq("toolu_99")
      expect(blk[:name]).to eq("bash")
    end

    it "marks saw_terminal? as true after message_stop" do
      sse_parser.feed(tool_stream) { |t, d| col.handle(t, d) }
      expect(col.saw_terminal?).to be true
    end
  end
end