# 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