Class: Envoy::RunJob

Inherits:
ApplicationJob show all
Extended by:
ActionView::RecordIdentifier
Defined in:
app/jobs/envoy/run_job.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.stream_target(conversation) ⇒ Object

Scoped per conversation: a flat constant meant two panels — or a panel and a console tab — would fight over the same target on one page.



7
8
9
 
# File 'app/jobs/envoy/run_job.rb', line 7

def self.stream_target(conversation)
  dom_id(conversation, :streaming)
end

Instance Method Details

#perform(conversation, content) ⇒ Object

GlobalID serialization of the Conversation (and its polymorphic actor) is handled by ActiveJob automatically.



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'app/jobs/envoy/run_job.rb', line 13

def perform(conversation, content)
  # RubyLLM yields incremental deltas; accumulate them so each broadcast
  # carries the full text so far (broadcast_replace_to replaces, not appends).
  # This also makes streaming race-proof: a missed early broadcast is harmless
  # because every later one contains the whole reply.
  @streamed = +""
  Envoy::Runner.new(conversation: conversation).call(content: content) do |event|
    stream(conversation, event)
  end
  finalize(conversation)
rescue StandardError
  # Without this the panel sits on "working…" forever: nothing else ever
  # replaces the streaming target. Re-raise so the queue still records the
  # failure and any retry policy applies.
  broadcast_error(conversation)
  raise
end