Class: AssistantChatWorker

Inherits:
Object
  • Object
show all
Includes:
AssistantChat::Broadcaster, Sidekiq::Job, Sidekiq::Status::Worker
Defined in:
app/workers/assistant_chat_worker.rb

Overview

Background worker for AI-powered assistant chat

Processes LLM queries asynchronously and broadcasts chunks via Turbo Streams.
This allows real-time streaming updates to the browser without blocking the web process.

Usage:
job_id = AssistantChatWorker.perform_async(
conversation_id: conversation.id,
user_message: "Show me sales by state",
model: "claude-sonnet"
)

Constant Summary collapse

HEARTBEAT_INTERVAL =

Heartbeat interval.

30
MAX_COMPACTION_RETRIES =

Maximum compaction retries.

2
ESCALATION_MODEL =

Model used when escalating from a low-tier model that stalled after tool calls.
A reliable non-Gemini model: escalating into another Gemini snapshot (the old
value, 'gemini-pro') risked the same intermittent body-less 400 the escalation
is trying to route around (#3808).

'claude-sonnet'

Constants included from AssistantChat::Broadcaster

AssistantChat::Broadcaster::STATUS_BROADCAST_INTERVAL, AssistantChat::Broadcaster::TOOLS_WITH_SUMMARIES, AssistantChat::Broadcaster::TOOL_DISPLAY_NAMES, AssistantChat::Broadcaster::TOOL_PREFIXES

Constants included from IconHelper

IconHelper::CUSTOM_ICON_MAP, IconHelper::CUSTOM_SVG_DIR, IconHelper::DEFAULT_FAMILY

Instance Method Summary collapse

Methods included from IconHelper

#account_nav_icon, #fa_icon, #star_rating_html

Instance Method Details

#perform(conversation_id, user_message, model, tool_services = [], user_context = {}, upload_ids = []) ⇒ Object

Parameters:

  • conversation_id (Integer) —

    The conversation to process

  • user_message (String) —

    The user's query

  • model (String) —

    The LLM model key to use

  • tool_services (Array<String>) (defaults to: []) —

    Optional service keys for tool access (e.g. ['content', 'app_db'])

  • user_context (Hash) (defaults to: {}) —

    Optional user identity for personalized queries (name, party_id, department, etc.)

  • upload_ids (Array<Integer>) (defaults to: []) —

    Optional IDs of Upload records to attach to this message



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'app/workers/assistant_chat_worker.rb', line 28

def perform(conversation_id, user_message, model, tool_services = [], user_context = {}, upload_ids = [])
  @conversation = AssistantConversation.find(conversation_id)
  @job_id = jid
  @cancelled = false
  @escalated = false
  @synthesized = false
  @accumulated_content = ''
  @last_stream_broadcast_at = 0.0
  @tool_services = Array(tool_services)
  @column_metadata = {}
  @worker_started_at = Time.current
  @round_count = 0
  initialize_opportunity_briefing_job(user_context)

  # Fork auto-starts render stable `chat-fork-*` targets before Sidekiq runs.
  # Resolve this before the first status broadcast so the queued → working
  # transition reaches those targets too. A regular first turn is also empty,
  # so parent_conversation_id is required to distinguish it from a fork.
  @fork_mode = @conversation.assistant_messages.empty? &&
               @conversation.parent_conversation_id.present?

  # Resolve the sending user from user_context for processing lock
  @sending_user = Party.find_by(id: user_context&.dig('party_id'))

  # Advisory lock prevents concurrent workers from processing the same conversation.
  # Non-blocking (timeout: 0) — if another worker is already processing, skip.
  result = AssistantConversation.with_advisory_lock_result(
    @conversation.processing_lock_key, timeout_seconds: 0
  ) do
    # Acquire the column-based processing lock (visible to UI).
    # If the advisory lock was obtained but the column lock fails, the previous
    # holder must have crashed without cleanup — force-take the lock.
    sender = @sending_user || Party.new(id: 0)
    unless @conversation.acquire_processing_lock!(sender, job_id: @job_id)
      Rails.logger.warn("[AssistantChatWorker] Stale processing lock on conversation #{@conversation.id} " \
                        "(by_id=#{@conversation.processing_by_id}, since=#{@conversation.processing_since}). " \
                        "Force-acquiring — advisory lock proves no other worker is active.")
      @conversation.force_processing_lock!(sender, job_id: @job_id)
    end

    begin
      # Past the advisory lock and the column lock, so this worker owns the
      # turn. Only an owner may record why a turn failed — a worker that lost
      # the lock still broadcasts a "currently being processed" notice, and
      # that is about its own request, not the turn in flight.
      @owns_turn = true
      # A new turn supersedes whatever the last one failed with, so the banner
      # can never show a stale reason next to a fresh question.
      @conversation.clear_turn_error!
      broadcast_lock_state(locked: true)
      broadcast_processing_status('Sunny is working…')
      process_query(user_message, model, tool_services, user_context, Array(upload_ids))
    rescue Assistant::ConversationDestroyed
      # Mark before the ensure runs so the unlock broadcast below is
      # suppressed — the conversation (and its stream) no longer exists.
      @conversation_destroyed = true
      raise
    ensure
      @conversation.release_processing_lock!
      broadcast_lock_state(locked: false) unless @conversation_destroyed
    end
  end

  # If the advisory lock was not acquired, another worker is already processing
  unless result&.lock_was_acquired?
    Rails.logger.info("[AssistantChatWorker] Skipped — conversation #{conversation_id} is locked by another worker")
    broadcast_error('This conversation is currently being processed. Please try again in a moment.')
  end

  # Some handled error paths return normally after broadcasting an error
  # (blank output, cancellation, exhausted context recovery). If no success
  # transition ran, leave the briefing retryable instead of stuck processing.
  transition_opportunity_briefing_failure_if_needed if result&.lock_was_acquired?

  # Schedule continuation AFTER all locks are released so the new worker
  # can immediately acquire the advisory lock without collision.
  if @continuation_pending && !schedule_continuation(
      @continuation_pending[:model],
      @continuation_pending[:tool_services],
      @continuation_pending[:user_context]
    )
    store_opportunity_briefing_continuation_failure
    @continuation_pending = nil
    transition_opportunity_briefing_failure_if_needed
  end
rescue Assistant::ConversationDestroyed
  # The conversation was deleted (owner or Daily Focus cleanup) while this
  # worker was mid-turn. Nothing to persist to and nobody to broadcast to —
  # exit quietly instead of reporting the FK violation. (AppSignal #6043)
  Rails.logger.info("[AssistantChatWorker] Conversation #{conversation_id} destroyed mid-processing — exiting quietly.")
  @cancelled = true
  @conversation_destroyed = true
rescue StandardError => e
  response_body = e.respond_to?(:response) ? e.response&.body.to_s.truncate(500) : nil
  Rails.logger.error("[AssistantChatWorker] Error: #{e.message}\n#{e.backtrace.first(5).join("\n")}#{"\nResponse body: #{response_body}" if response_body}")
  @conversation&.release_processing_lock!
  @conversation&.track_error!
  transition_opportunity_briefing_failure_if_needed
  # `user_id` is a filterable tag; the worker/conversation/model and the
  # provider response body ride along as custom data so they're visible on the
  # AppSignal sample. CAVEAT (#3808): Faraday does not buffer the response body
  # on STREAMING turns (assistant chat), so `response_body` is empty for those
  # — the generic provider 400 carries no body to capture. It is still
  # populated for non-streaming provider calls (e.g. invalid-model 400s).
  ErrorReporting.error(e,
                       source: :background,
                       user_id: @sending_user&.id,
                       custom_data: {
                         worker: 'AssistantChatWorker',
                         conversation_id: @conversation&.id,
                         model: model,
                         response_body: response_body
                       })
  # Pass the raw error message to broadcast_error so its pattern matchers can
  # produce the correct friendly copy (e.g. the /\Amodel:/i pattern for
  # invalid-model 400s). Generic unrecognised messages fall through as-is.
  broadcast_error(e.message)
ensure
  cleanup_attachment_tempfiles
end