Class: OpenAI::Responses::SessionLane

Inherits:
Object
  • Object
show all
Defined in:
lib/openai/helpers/responses_websocket/session.rb

Overview

One routed consumer. Canceling its wait never cancels the socket reader.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(session, stream_id, pending_steers) ⇒ SessionLane

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of SessionLane.



137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 137

def initialize(session, stream_id, pending_steers)
  @session = session
  @stream_id = stream_id
  @queue = []
  @bytes = 0
  @closed = false
  @receiving = false
  @in_flight = false
  @pending_steers = pending_steers
  @response_id = nil
  @changed = Async::Condition.new
  @accumulator = SessionAccumulator.new(session.limits.max_response_bytes)
end

Instance Attribute Details

#stream_id ⇒ Object (readonly)

Returns the value of attribute stream_id.



134
135
136
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 134

def stream_id
  @stream_id
end

Instance Method Details

#close ⇒ Object 



276
277
278
279
280
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 276

def close
  @session.assert_owner!
  @session.detach(self)
  nil
end

#discard(error = nil) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



298
299
300
301
302
303
304
305
306
307
308
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 298

def discard(error = nil)
  return if @closed

  @closed = true
  @detached_error = error
  @queue.each { |_event, bytes| @session.consumed(bytes) }
  @queue.clear
  @bytes = 0
  @accumulator = SessionAccumulator.new(@session.limits.max_response_bytes)
  @changed.signal
end

#enqueue(event, bytes, raw_command_sent = false) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



283
284
285
286
287
288
289
290
291
292
293
294
295
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 283

def enqueue(event, bytes, raw_command_sent = false)
  return if @closed

  limits = @session.limits
  if @queue.length >= limits.max_events_per_lane || @bytes + bytes > limits.max_bytes_per_lane
    raise BufferError, "The Responses lane exceeded its buffered event budget."
  end

  @queue << [event, bytes, raw_command_sent]
  @bytes += bytes
  @changed.signal
  true
end

#get_final_response ⇒ Object 



263
264
265
266
267
268
269
270
271
272
273
274
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 263

def get_final_response
  @session.assert_owner!
  loop do
    raise @accumulator.error if @accumulator.error
    if (response = @accumulator.final_response)
      @accumulator = SessionAccumulator.new(@session.limits.max_response_bytes)
      return response
    end

    receive
  end
end

#receive ⇒ Object

Raises:



194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 194

def receive
  @session.assert_owner!
  raise SessionError, "This Responses lane already has a consumer." if @receiving

  @receiving = true
  begin
    loop do
      raise @detached_error if @detached_error
      raise SessionError, "The Responses lane is detached." if @closed

      unless @queue.empty?
        event, bytes, raw_command_sent = @queue.shift
        @bytes -= bytes
        @session.consumed(bytes)
        if %w[response.steer.pending response.steer.failed].include?(event.type.to_s)
          parent = event.steer.previous_response_id
          pending = @pending_steers[parent]
          id = event.steer.id
          if pending
            previously_pending = pending[:outcomes].delete(id)
            if event.type.to_s == "response.steer.pending" && !id.nil?
              pending[:outcomes][id] = true
            end

            unless previously_pending
              pending[:count] -= 1
            end

            @pending_steers.delete(parent) if pending[:count].zero? && pending[:outcomes].empty?
          end
        end
        # Unregistered named streams remain available for raw inspection,
        # without changing the default lane's response state.
        if event.to_h[:stream_id] == @stream_id
          # Prior raw commands leave generic errors uncorrelated. Use the
          # receipt-time snapshot, so later commands cannot change attribution.
          if event.type.to_s == "error" &&
              @stream_id.nil? &&
              @in_flight &&
              event.to_h[:sequence_number].nil? &&
              raw_command_sent
            discard(RequestError.new(event))
            @session.detach(self)
            return event
          end

          @accumulator.add(event, bytes)
          if event.type.to_s == "response.created"
            @in_flight = true
            @pending_steers.delete(@response_id) if @response_id != event.response.id
            @response_id = event.response.id.dup.freeze
          elsif %w[response.completed response.failed response.incomplete error].include?(event.type.to_s)
            @in_flight = false
            @response_id = event.response.id.dup.freeze unless event.type.to_s == "error"
          end
        end

        return event
      end

      @session.check_reader!
      @changed.wait
    end

  ensure
    @receiving = false
  end
end

#send_event(event) ⇒ Object 



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 151

def send_event(event)
  @session.assert_owner!
  raise @detached_error if @detached_error
  raise SessionError, "The Responses lane is detached." if @closed

  data = @session.encode_client_event(event)
  type = data[:type]
  if type.to_s == "response.create"
    key = data.key?("stream_id") ? "stream_id" : :stream_id
    if data.key?(key) && data[key] != @stream_id
      raise ArgumentError, "The command stream_id does not match its lane."
    end

    data[key] = @stream_id unless @stream_id.nil?
    queued_response = @queue.any? do |queued_event, _bytes|
      queued_event.type.to_s == "response.created" && queued_event.to_h[:stream_id] == @stream_id
    end

    pending = @pending_steers[@response_id]
    if @in_flight || queued_response || (pending && pending[:count].positive?)
      raise SessionError, "Consume the current response before sending another create on this lane."
    end

    @accumulator = SessionAccumulator.new(@session.limits.max_response_bytes)
    @in_flight = true
  elsif type.to_s == "response.steer" &&
      (steer_parent = data[:previous_response_id]).is_a?(String) &&
      !steer_parent.empty?
    pending = (@pending_steers[steer_parent] ||= {count: 0, outcomes: {}})
    pending[:count] += 1
  end

  @session.send_event(data)
rescue OpenAI::Errors::ResponsesClientEventError
  @in_flight = false if type.to_s == "response.create"
  if type.to_s == "response.steer" && pending
    pending[:count] -= 1
    @pending_steers.delete(steer_parent) if pending[:count].zero? && pending[:outcomes].empty?
  end

  raise
end

#wake ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



311
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 311

def wake = @changed.signal