Class: Mistri::Providers::OpenAI::Assembler

Inherits:
Object
  • Object
show all
Defined in:
lib/mistri/providers/openai/assembler.rb

Overview

Folds the Responses API stream into the event union. Items arrive sequentially: output_item.added opens a block, typed deltas fill it, output_item.done closes it with the complete item, whose ids and encrypted reasoning land in the signature slots for replay.

Unknown event and item types are skipped by contract.

Defined Under Namespace

Classes: Builder

Constant Summary collapse

KINDS = 
{ "message" => :text, "reasoning" => :thinking,
"function_call" => :toolcall }.freeze

Instance Method Summary collapse

Constructor Details

#initialize(model:) ⇒ Assembler

Returns a new instance of Assembler.



15
16
17
18
19
20
21
22
 
# File 'lib/mistri/providers/openai/assembler.rb', line 15

def initialize(model:)
  @model = model
  @blocks = []
  @current = nil
  @usage = Usage.zero
  @status = nil
  @incomplete_reason = nil
end

Instance Method Details

#abort ⇒ Object 



49
50
51
 
# File 'lib/mistri/providers/openai/assembler.rb', line 49

def abort(&)
  terminal(StopReason::ABORTED, "aborted", &)
end

#fail_stream(reason) ⇒ Object 



65
66
67
68
69
70
71
72
 
# File 'lib/mistri/providers/openai/assembler.rb', line 65

def fail_stream(reason, &)
  text = case reason
         when ProviderError then "#{reason.class}: #{reason.describe}"
         when Exception then "#{reason.class}: #{reason.message}"
         else reason.to_s
         end
  terminal(StopReason::ERROR, text, error: ErrorData.for(reason), &)
end

#feed(record) ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/mistri/providers/openai/assembler.rb', line 24

def feed(record, &)
  case record["type"]
  when "response.output_item.added" then start_item(record["item"], &)
  when "response.output_text.delta" then text_delta(record["delta"], &)
  when "response.reasoning_summary_text.delta"
    thinking_delta(record["delta"], record["summary_index"], &)
  when "response.function_call_arguments.delta" then arguments_delta(record["delta"], &)
  when "response.output_item.done" then finish_item(record["item"], &)
  when "response.completed", "response.incomplete", "response.failed"
    finish_response(record["response"] || {})
  when "error" then @error = wire_error(record)
  end
end

#finish(&emit) ⇒ Object

A stream that ended without a terminal response event was truncated, not cancelled: fail it so the loop can treat it as retryable.



40
41
42
43
44
45
46
47
 
# File 'lib/mistri/providers/openai/assembler.rb', line 40

def finish(&emit)
  return fail_stream(@error, &emit) if @error
  return fail_stream("stream ended without a terminal event", &emit) unless @status

  @message = assemble(stop_reason: stop_reason)
  emit&.call(Event.new(type: :done, reason: @message.stop_reason, message: @message))
  @message
end

#message ⇒ Object 



74
 
# File 'lib/mistri/providers/openai/assembler.rb', line 74

def message = @message ||= finish

#wire_error(record) ⇒ Object

In-stream failures carry a code; rate limits and server errors must classify as retryable, not fold into prose.



55
56
57
58
59
60
61
62
63
 
# File 'lib/mistri/providers/openai/assembler.rb', line 55

def wire_error(record)
  message = record["message"] || "provider error"
  code = record["code"].to_s
  klass = if code.include?("rate_limit") then RateLimitError
          elsif code.include?("server") then ServerError
          else ProviderError
          end
  klass.new(message)
end