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.

Argument deltas expose a bounded, cached preview; the completed item stays authoritative. 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
STREAM_EVENTS = 
%w[
  response.output_item.added response.output_text.delta
  response.reasoning_summary_text.delta response.refusal.delta
  response.function_call_arguments.delta response.output_item.done
  response.completed response.incomplete response.failed error
].freeze
TERMINAL_STATUSES = 
{
  "response.completed" => "completed",
  "response.incomplete" => "incomplete",
  "response.failed" => "failed"
}.freeze
PREVIEW_EAGER_BYTES = 
4 * 1024
PREVIEW_STEP_BYTES = 
4 * 1024
PREVIEW_MAX_BYTES = 
64 * 1024
MAX_REFUSAL_BYTES = 
2048

Instance Method Summary collapse

Constructor Details

#initialize(model:, catalog_pricing: true) ⇒ Assembler

rubocop:disable Metrics/ClassLength -- one stream owns item order



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

def initialize(model:, catalog_pricing: true)
  @model = model
  @catalog_pricing = catalog_pricing
  @pricing_at = Time.now
  @blocks = []
  @current = nil
  @usage = Usage.new
  @status = nil
  @incomplete_reason = nil
end

Instance Method Details

#abort ⇒ Object 



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

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

#fail_stream(reason) ⇒ Object 



84
85
86
87
88
89
90
91
 
# File 'lib/mistri/providers/openai/assembler.rb', line 84

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 



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

def feed(record, &)
  type = record["type"]
  if @status && STREAM_EVENTS.include?(type)
    return protocol_error("response continued after its terminal event", &)
  end

  dispatch_record(record, &)
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.



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/mistri/providers/openai/assembler.rb', line 55

def finish(&emit)
  return fail_stream(@error, &emit) if @error
  return fail_stream(@refusal_error, &emit) if @refusal_error
  return fail_stream(filter_error, &emit) if @incomplete_reason == "content_filter"

  if @status == "incomplete" && @incomplete_reason != "max_output_tokens"
    detail = @incomplete_reason || "unspecified reason"
    return fail_stream("response was incomplete: #{detail}", &emit)
  end
  return fail_stream("stream ended without a terminal event", &emit) unless @status
  return fail_stream("output item ended without output_item.done", &emit) if @current

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

#message ⇒ Object 



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

def message = @message ||= finish

#start(&emit) ⇒ Object 



116
117
118
 
# File 'lib/mistri/providers/openai/assembler.rb', line 116

def start(&emit)
  emit&.call(Event.new(type: :start, partial: assemble))
end

#wire_error(record) ⇒ Object

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



78
79
80
81
82
 
# File 'lib/mistri/providers/openai/assembler.rb', line 78

def wire_error(record)
  message = record["message"] || "provider error"
  code = record["code"].to_s
  classify_error(code, message, unknown: ProviderError)
end