Class: Mistri::Providers::Anthropic::Assembler

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

Overview

Folds the Messages API stream into the event union, building the assistant message block by block. Every emitted event carries an immutable snapshot of the message so far; in-flight tool arguments expose a bounded, cached PartialJson preview so hostile fragmentation cannot make snapshot construction quadratic.

Unknown event and block types are skipped by contract: the API adds types over time and a live stream must survive them.

Defined Under Namespace

Classes: Builder

Constant Summary collapse

STREAM_EVENTS = 
%w[
  message_start content_block_start content_block_delta content_block_stop
  message_delta message_stop error
].freeze
DELTA_KINDS = 
{
  "text_delta" => :text,
  "thinking_delta" => :thinking,
  "signature_delta" => :thinking,
  "input_json_delta" => :toolcall
}.freeze
PREVIEW_EAGER_BYTES = 
4 * 1024
PREVIEW_STEP_BYTES = 
4 * 1024
PREVIEW_MAX_BYTES = 
64 * 1024
MAX_SIGNATURE_BYTES = 
ToolArguments::MAX_BYTES
MAX_REFUSAL_BYTES = 
2048

Instance Method Summary collapse

Constructor Details

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

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



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

def initialize(model:, catalog_pricing: true)
  @model = model
  @catalog_pricing = catalog_pricing
  @pricing_at = Time.now
  @blocks = []
  @current = nil
  @usage = Usage.new
  @stop_reason = nil
  @done = false
  @next_wire_index = 0
  @open_wire_index = nil
  @message_phase = false
end

Instance Method Details

#abort(&emit) ⇒ Object 



72
73
74
75
76
77
78
79
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 72

def abort(&emit)
  close_current(interrupted: true, &emit)
  invalidate_cost
  @message = assemble(stop_reason: StopReason::ABORTED, error_message: "aborted")
  emit&.call(Event.new(type: :error, reason: StopReason::ABORTED, message: @message,
                       error_message: "aborted"))
  @message
end

#fail_stream(reason, usage_known: false, &emit) ⇒ Object 



98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 98

def fail_stream(reason, usage_known: false, &emit)
  close_current(interrupted: true, &emit)
  invalidate_cost unless usage_known
  text = case reason
         when ProviderError then "#{reason.class}: #{reason.describe}"
         when Exception then "#{reason.class}: #{reason.message}"
         else reason.to_s
         end
  @message = assemble(stop_reason: StopReason::ERROR, error_message: text,
                      error: ErrorData.for(reason))
  emit&.call(Event.new(type: :error, reason: StopReason::ERROR, message: @message,
                       error_message: text))
  @message
end

#feed(record) ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 34

def feed(record, &)
  type = record["type"]
  if @done && STREAM_EVENTS.include?(type)
    return protocol_error("stream continued after message_stop", &)
  end

  case type
  when "message_start" then message_start(record)
  when "content_block_start" then start_block(record, &)
  when "content_block_delta" then delta_block(record, &)
  when "content_block_stop" then stop_block(record, &)
  when "message_delta" then message_delta(record, &)
  when "message_stop" then stop_message(&)
  when "error" then @error ||= wire_error(record["error"])
  end
end

#finish(&emit) ⇒ Object

Close the stream: the terminal event reflects how it ended. A stream that ended without message_stop was truncated (a dropped proxy, say), not user-aborted, so it fails for the loop to retry rather than reading as a cancellation.



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

def finish(&emit)
  return fail_stream(@error, &emit) if @error
  if @refused
    return fail_stream(refusal_error,
                       usage_known: @done && @usage_authoritative, &emit)
  end
  return fail_stream("stream ended without message_stop", &emit) unless @done
  if @open_wire_index
    return fail_stream("content block ended without content_block_stop", &emit)
  end

  invalidate_cost unless @usage_authoritative
  @message = assemble(stop_reason: @stop_reason || StopReason::STOP)
  emit&.call(Event.new(type: :done, reason: @message.stop_reason, message: @message))
  @message
end

#message ⇒ Object 



113
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 113

def message = @message ||= finish

#start(&emit) ⇒ Object 



130
131
132
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 130

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

#wire_error(payload) ⇒ Object

In-stream failures carry a wire type; overloaded ones must classify as retryable, not fold into prose.



83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 83

def wire_error(payload)
  message = payload&.dig("message") || "provider error"
  type = payload&.dig("type").to_s
  klass = case type
          when "authentication_error" then AuthenticationError
          when "rate_limit_error" then RateLimitError
          when "overloaded_error" then OverloadedError
          when "api_error" then ServerError
          when "timeout_error" then ProviderError
          else InvalidRequestError
          end
  detail = type.empty? ? message : "#{type}: #{message}"
  klass.new(detail)
end