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 parse via PartialJson so consumers can read them mid-stream.

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

Instance Method Summary collapse

Constructor Details

#initialize(model:) ⇒ Assembler

Returns a new instance of Assembler.



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

def initialize(model:)
  @model = model
  @blocks = []
  @current = nil
  @usage = Usage.zero
  @stop_reason = nil
  @done = false
end

Instance Method Details

#abort(&emit) ⇒ Object 



48
49
50
51
52
53
54
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 48

def abort(&emit)
  finalize_current
  @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, &emit) ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 64

def fail_stream(reason, &emit)
  finalize_current
  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 



23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 23

def feed(record, &)
  case record["type"]
  when "message_start" then @usage = parse_usage(record.dig("message", "usage"))
  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 @done = true
  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.



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

def finish(&emit)
  return fail_stream(@error, &emit) if @error
  return fail_stream("stream ended without message_stop", &emit) unless @done

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

#message ⇒ Object 



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

def message = @message ||= finish

#wire_error(payload) ⇒ Object

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



58
59
60
61
62
 
# File 'lib/mistri/providers/anthropic/assembler.rb', line 58

def wire_error(payload)
  message = payload&.dig("message") || "provider error"
  klass = payload&.dig("type").to_s.include?("overloaded") ? OverloadedError : ProviderError
  klass.new(message)
end