Module: Insika::Evals::Sse

Defined in:
lib/insika/evals/transport.rb

Overview

Pure reduction of the /v1/responses SSE stream. Kept separate from the HTTP so it's testable offline with canned frames (server/responses.rb is the producer).

Class Method Summary collapse

Class Method Details

.payloads(text) ⇒ Object

Raw SSE text -> [parsed JSON payload]. Skips event: lines, blanks, [DONE].



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/insika/evals/transport.rb', line 31

def payloads(text)
  text.to_s.split("\n\n").flat_map do |frame|
    frame.each_line.filter_map do |line|
      next unless line.start_with?("data:")

      p = line.sub(/^data:\s*/, "").strip
      next if p.empty? || p == "[DONE]"

      begin
        JSON.parse(p)
      rescue JSON::ParserError
        nil
      end
    end
  end
end

.reduce(payloads) ⇒ Object

[payload] -> { output_text:, tool_calls:, usage:, error: }. Maps the Responses frames (see server/responses.rb#frame_for): text deltas accumulate; each function_call item contributes a tool NAME (this stream carries no per-tool status — that lives in the ToolTraceStore, an in-process enrichment); a response.failed sets the turn error.



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/insika/evals/transport.rb', line 53

def reduce(payloads)
  text = +""
  tools = []
  usage = nil
  error = nil
  payloads.each do |o|
    case o["type"]
    when "response.output_text.delta"
      text << o["delta"].to_s
    when "response.output_item.added"
      item = o["item"] || {}
      tools << { "name" => item["name"].to_s, "status" => nil } if item["type"] == "function_call"
    when "response.completed"
      usage = o.dig("response", "usage")
    when "response.failed"
      error = o.dig("response", "error", "message") || "response.failed"
    end
  end
  { output_text: text, tool_calls: tools, usage: usage, error: error }
end