kunthawat/moreminimore-chat
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix: Captain handles message bursts with a single reply (#15133)

Browse Source 
Captain now handles a burst of customer messages with one reply. If more
messages arrive before Captain replies, the latest job uses the full
conversation history. The change applies only to Captain V2 and works
across every channel that Captain supports.

Blocked on: https://github.com/chatwoot/chatwoot/pull/15212

## Closes

Closes https://github.com/chatwoot/chatwoot/issues/14545

## How to reproduce

1. Start a pending conversation with Captain V2.
2. Send several messages while Captain is preparing a reply.
3. Captain can generate and send a separate reply for each message.

## What changed

* Each Captain V2 job records the incoming message that started it.
* A job stops before generation if a newer message already exists.
* Captain discards a generated reply if a newer message arrived during
generation.
* The latest job replies using the full conversation history.
* Langfuse records whether a generation was discarded and whether a
customer credit was used.
* Captain V1 keeps its existing behavior.

## Tradeoffs

Discarded generations still cost money. Message bursts can also increase
background job work and model provider load. A continuous stream of
incoming messages can delay the reply until one generation finishes
without a newer message.

A small timing window remains if a message arrives after the final check
and before Captain saves the reply. A handoff also cannot be undone if a
newer message arrives after Captain has already changed the conversation
status.

## How to test

1. Enable Captain V2 and start a pending Captain conversation.
2. Send several messages while Captain is preparing a reply.
3. Confirm that Captain sends one reply based on the full message
history.
4. Confirm that Langfuse marks discarded runs with `discarded=true` and
`credit_used=false`.
5. Disable Captain V2 and confirm that Captain V1 behavior is unchanged.

---------

Co-authored-by: Sony Mathew <sony@chatwoot.com>
This commit is contained in:
Aakash Bakhle
2026-07-30 15:58:54 +05:30
committed by GitHub
parent 34d63454db
commit 38a317962b
17 changed files with 635 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
app/models/conversation.rb
View File

@@ -172,10 +172,14 @@ class Conversation < ApplicationRecord
save
end
def bot_handoff!
def bot_handoff!(dispatch_event: true)
update(waiting_since: Time.current) if waiting_since.blank?
self.assignee_agent_bot = nil
open!
dispatch_bot_handoff_event if dispatch_event
end
def dispatch_bot_handoff_event
dispatcher_dispatch(CONVERSATION_BOT_HANDOFF)
end

89
enterprise/app/jobs/captain/conversation/response_builder_job.rb
View File

@@ -7,20 +7,20 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
retry_on ActiveStorage::FileNotFoundError, attempts: 3, wait: 2.seconds
retry_on Faraday::BadRequestError, attempts: 3, wait: 2.seconds
def perform(conversation, assistant, responding_to_message_id = nil) # rubocop:disable Lint/UnusedMethodArgument
def perform(conversation, assistant, responding_to_message_id = nil)
@conversation = conversation
@inbox = conversation.inbox
@assistant = assistant
@responding_to_message_id = responding_to_message_id if captain_v2_enabled?
return unless conversation_pending?
Current.executed_by = @assistant
if captain_v2_enabled?
generate_response_with_v2
else
generate_and_process_response
end
return generate_and_process_response unless captain_v2_enabled?
return if newer_customer_message_arrived?
generate_response_with_v2
rescue ActiveStorage::FileNotFoundError, Faraday::BadRequestError => e
handle_error(e)
raise e
@@ -45,30 +45,27 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
end
def generate_response_with_v2
runner_service = Captain::Assistant::AgentRunnerService.new(assistant: @assistant, conversation: @conversation)
runner_service = v2_runner_service
message_history = Captain::Conversation::MessageHistoryBuilderService.new(conversation: @conversation).perform
@response = runner_service.generate_response(message_history: message_history)
@run_result = runner_service.last_run_result
@v2_handoff_tool_completed = runner_service.handoff_completed?
return process_response if v2_handoff_tool_completed?
return if runner_service.response_discarded? || newer_customer_message_arrived?
process_response
end
def v2_runner_service
runner_args = { assistant: @assistant, conversation: @conversation }
runner_args[:responding_to_message_id] = @responding_to_message_id if @responding_to_message_id.present?
Captain::Assistant::AgentRunnerService.new(**runner_args)
end
def process_response
# Check V2 before V1: error_response can set both signals at once when HandoffTool
# fired before the runner errored. V2 must win — running V1 on top would duplicate
# OOO and re-dispatch the bot_handoff event.
if v2_handoff_tool_fired?
if conversation_pending?
# HandoffTool flipped the flag without committing — its perform returned a
# failure string (e.g. "Conversation not found") before bot_handoff! ran. Fall
# back to a full V1 handoff so the customer still ends up with a human.
process_v1_handoff
else
# HandoffTool already opened the conversation inside the agent loop. All that's
# left is the customer-facing follow-up message.
process_v2_handoff
end
capture_assistant_session(result_message: @handoff_message, credits_consumed: 0.0)
process_v2_handoff_response
elsif v1_handoff_requested?
# V1 only signals via the response string — no state has been touched yet. If
# the conversation isn't pending anymore, a human took over mid-run; bail out
@@ -77,16 +74,38 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
process_v1_handoff
elsif conversation_pending?
message = nil
ActiveRecord::Base.transaction do
message = create_messages
Rails.logger.info("[CAPTAIN][ResponseBuilderJob] Incrementing response usage for #{account.id}")
account.increment_response_usage
end
capture_assistant_session(result_message: message, credits_consumed: 1.0)
process_standard_response
end
end
def process_standard_response
message = nil
ActiveRecord::Base.transaction do
next if captain_v2_enabled? && newer_customer_message_arrived?
message = create_messages
Rails.logger.info("[CAPTAIN][ResponseBuilderJob] Incrementing response usage for #{account.id}")
account.increment_response_usage
end
return unless message
capture_assistant_session(result_message: message, credits_consumed: 1.0)
end
def process_v2_handoff_response
# Captain V1 infers completion from status. Captain V2 uses the completion
# marker set inside the locked handoff.
if captain_v2_enabled?
return unless v2_handoff_tool_completed? || conversation_pending?
v2_handoff_tool_completed? ? process_v2_handoff : process_v1_handoff
else
conversation_pending? ? process_v1_handoff : process_v2_handoff
end
capture_assistant_session(result_message: @handoff_message, credits_consumed: 0.0)
end
def v1_handoff_requested?
legacy_v1_handoff_token? || classifier_v1_handoff_requested?
end
@@ -103,6 +122,8 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
@response['handoff_tool_called']
end
def v2_handoff_tool_completed? = @v2_handoff_tool_completed == true
def process_v1_handoff
I18n.with_locale(@assistant.account.locale) do
Rails.logger.info(
@@ -152,7 +173,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
@response ||= {}
@response['action_source'] ||= 'error'
@response['action_reason'] ||= error_action_reason(error)
process_v1_handoff if conversation_pending?
process_v1_handoff if conversation_pending? && (!captain_v2_enabled? || !newer_customer_message_arrived?)
true
end
@@ -181,4 +202,14 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
status = Conversation.uncached { Conversation.where(id: @conversation.id).pick(:status) }
status == 'pending' || status == Conversation.statuses[:pending]
end
def newer_customer_message_arrived?
return false if @responding_to_message_id.blank?
Conversation.uncached do
@conversation.messages
.captain_response_triggering
.exists?(['messages.id > ?', @responding_to_message_id])
end
end
end

14
enterprise/app/models/enterprise/message.rb
View File

@@ -4,6 +4,14 @@ module Enterprise::Message
has_one :call, class_name: 'Call', foreign_key: :message_id, dependent: :nullify, inverse_of: :message
scope :with_call, -> { includes(call: [:contact, { inbox: :channel }]) }
# Scheduling and freshness checks must share this scope so an email auto reply cannot cancel a pending response.
scope :captain_response_triggering, lambda {
incoming.joins(:inbox).where(
"((messages.content_attributes #>> '{}')::jsonb -> 'email' ->> 'auto_reply') IS DISTINCT FROM 'true' OR " \
"(messages.content_type != :incoming_email AND inboxes.channel_type != 'Channel::Email')",
incoming_email: content_types[:incoming_email]
)
}
end
end
@@ -13,6 +21,12 @@ module Enterprise::Message
data
end
def captain_response_triggering?
return incoming? && !auto_reply_email? unless persisted?
self.class.captain_response_triggering.exists?(id: id)
end
private
def mark_pending_conversation_as_open_for_human_response

75
enterprise/app/services/captain/assistant/agent_runner_service.rb
View File

@@ -2,19 +2,21 @@ require 'agents'
require 'agents/instrumentation'
class Captain::Assistant::AgentRunnerService
include Integrations::LlmInstrumentationConstants
include Captain::Assistant::RunnerCallbacksHelper
include Captain::Assistant::RunnerInstrumentationHelper
include Captain::Assistant::TracePayloadHelper
include Captain::Assistant::RunnerStateHelper
attr_reader :last_run_result
def initialize(assistant:, conversation: nil, callbacks: {}, source: nil)
def initialize(assistant:, conversation: nil, callbacks: {}, source: nil, responding_to_message_id: nil)
@assistant = assistant
@conversation = conversation
@callbacks = callbacks
@source = source
@responding_to_message_id = responding_to_message_id
@handoff_tool_called = false
@handoff_tool_completed = false
end
def generate_response(message_history: [])
@@ -35,6 +37,10 @@ class Captain::Assistant::AgentRunnerService
error_response(e.message)
end
def response_discarded? = @response_discarded == true
def handoff_completed? = @handoff_tool_completed == true
private
def build_context(message_history)
@@ -144,71 +150,6 @@ class Captain::Assistant::AgentRunnerService
[assistant_agent] + scenario_agents
end
def install_instrumentation(runner)
return unless ChatwootApp.otel_enabled?
Agents::Instrumentation.install(
runner,
tracer: OpentelemetryConfig.tracer,
trace_name: 'llm.captain_v2',
span_attributes: {
ATTR_LANGFUSE_TAGS => ['captain_v2'].to_json
},
attribute_provider: Captain::Assistant::InstrumentationAttributeProvider.new(self)
)
register_trace_input_callback(runner)
end
def dynamic_trace_attributes(context_wrapper)
state = context_wrapper&.context&.dig(:state) || {}
conversation = state[:conversation] || {}
trace_input = context_wrapper&.context&.dig(:captain_v2_trace_input)
{
ATTR_LANGFUSE_USER_ID => state[:account_id],
format(ATTR_LANGFUSE_METADATA, 'assistant_id') => state[:assistant_id],
format(ATTR_LANGFUSE_METADATA, 'conversation_display_id') => conversation[:display_id],
format(ATTR_LANGFUSE_METADATA, 'channel_type') => state[:channel_type],
format(ATTR_LANGFUSE_METADATA, 'source') => state[:source],
ATTR_LANGFUSE_TRACE_INPUT => trace_input,
ATTR_LANGFUSE_OBSERVATION_INPUT => trace_input
}.compact.transform_values(&:to_s)
end
def add_usage_metadata_callback(runner)
handoff_tool_name = Captain::Tools::HandoffTool.new(@assistant).name
# Tool tracking always runs — process_response in the job consumes the resulting
# handoff_tool_called flag regardless of whether OTEL is enabled.
runner.on_tool_complete do |tool_name, _tool_result, context_wrapper|
track_handoff_usage(tool_name, handoff_tool_name, context_wrapper)
end
if ChatwootApp.otel_enabled?
runner.on_run_complete do |_agent_name, _result, context_wrapper|
write_credits_used_metadata(context_wrapper)
end
end
runner
end
def track_handoff_usage(tool_name, handoff_tool_name, context_wrapper)
return unless context_wrapper&.context
return unless tool_name.to_s == handoff_tool_name
# Mirror the flag onto the instance so error_response can surface it even when
# the runner raises before returning a result (the context is unreachable then).
context_wrapper.context[:captain_v2_handoff_tool_called] = true
@handoff_tool_called = true
end
def write_credits_used_metadata(context_wrapper)
root_span = context_wrapper&.context&.dig(:__otel_tracing, :root_span)
return unless root_span
root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), @handoff_tool_called ? 'false' : 'true')
end
def runner
@runner ||= begin
configured_runner = Agents::Runner.with_agents(*build_and_wire_agents)

6
enterprise/app/services/captain/assistant/instrumentation_attribute_provider.rb
View File

@@ -12,9 +12,13 @@ class Captain::Assistant::InstrumentationAttributeProvider
end
def generation_attributes(_context_wrapper, _chat, message)
{
attributes = {
format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'generation_stage') => generation_stage(message)
}
if @service.send(:message_burst_protection_active?)
attributes[format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'discarded')] = @service.send(:newer_customer_message_arrived?).to_s
end
attributes
end
private

99
enterprise/app/services/captain/assistant/runner_instrumentation_helper.rb Normal file
View File

@@ -0,0 +1,99 @@
module Captain::Assistant::RunnerInstrumentationHelper
include Integrations::LlmInstrumentationConstants
private
def install_instrumentation(runner)
return unless ChatwootApp.otel_enabled?
Agents::Instrumentation.install(
runner,
tracer: OpentelemetryConfig.tracer,
trace_name: 'llm.captain_v2',
span_attributes: {
ATTR_LANGFUSE_TAGS => ['captain_v2'].to_json
},
attribute_provider: Captain::Assistant::InstrumentationAttributeProvider.new(self)
)
register_trace_input_callback(runner)
end
def dynamic_trace_attributes(context_wrapper)
state = context_wrapper&.context&.dig(:state) || {}
conversation = state[:conversation] || {}
trace_input = context_wrapper&.context&.dig(:captain_v2_trace_input)
{
ATTR_LANGFUSE_USER_ID => state[:account_id],
format(ATTR_LANGFUSE_METADATA, 'assistant_id') => state[:assistant_id],
format(ATTR_LANGFUSE_METADATA, 'conversation_display_id') => conversation[:display_id],
format(ATTR_LANGFUSE_METADATA, 'channel_type') => state[:channel_type],
format(ATTR_LANGFUSE_METADATA, 'source') => state[:source],
ATTR_LANGFUSE_TRACE_INPUT => trace_input,
ATTR_LANGFUSE_OBSERVATION_INPUT => trace_input
}.compact.transform_values(&:to_s)
end
def add_usage_metadata_callback(runner)
handoff_tool_name = Captain::Tools::HandoffTool.new(@assistant).name
# Tool tracking always runs — process_response in the job consumes the resulting
# handoff_tool_called flag regardless of whether OTEL is enabled.
runner.on_tool_complete do |tool_name, _tool_result, context_wrapper|
track_handoff_usage(tool_name, handoff_tool_name, context_wrapper)
end
if message_burst_protection_active?
runner.on_run_complete do |_agent_name, _result, context_wrapper|
@response_discarded = newer_customer_message_arrived?
write_run_metadata(context_wrapper) if ChatwootApp.otel_enabled?
end
elsif ChatwootApp.otel_enabled?
runner.on_run_complete do |_agent_name, _result, context_wrapper|
write_credits_used_metadata(context_wrapper)
end
end
runner
end
def track_handoff_usage(tool_name, handoff_tool_name, context_wrapper)
return unless context_wrapper&.context
return unless tool_name.to_s == handoff_tool_name
# Mirror the flag onto the instance so error_response can surface it even when
# the runner raises before returning a result (the context is unreachable then).
context_wrapper.context[:captain_v2_handoff_tool_called] = true
@handoff_tool_called = true
return unless context_wrapper.context.dig(:state, :captain_v2_handoff_tool_completed)
@handoff_tool_completed = true
end
def write_run_metadata(context_wrapper)
root_span = context_wrapper&.context&.dig(:__otel_tracing, :root_span)
return unless root_span
root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'discarded'), response_discarded?.to_s)
root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), (!@handoff_tool_called && !response_discarded?).to_s)
end
def write_credits_used_metadata(context_wrapper)
root_span = context_wrapper&.context&.dig(:__otel_tracing, :root_span)
return unless root_span
root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), @handoff_tool_called ? 'false' : 'true')
end
def message_burst_protection_active? = @responding_to_message_id.present?
def newer_customer_message_arrived?
return false if @responding_to_message_id.blank? || @conversation.blank?
Conversation.uncached do
@conversation.messages
.captain_response_triggering
.exists?(['messages.id > ?', @responding_to_message_id])
end
end
end

1
enterprise/app/services/captain/assistant/runner_state_helper.rb
View File

@@ -23,6 +23,7 @@ module Captain::Assistant::RunnerStateHelper
timezone: @conversation&.inbox&.timezone.presence || 'UTC'
}
state[:source] = @source if @source.present?
state[:responding_to_message_id] = @responding_to_message_id if @responding_to_message_id.present?
build_conversation_state(state) if @conversation
state

27
enterprise/app/services/enterprise/message_templates/hook_execution_service.rb
View File

@@ -31,17 +31,34 @@ module Enterprise::MessageTemplates::HookExecutionService
def schedule_captain_response
job_args = [conversation, conversation.inbox.captain_assistant]
captain_v2_enabled = conversation.account.feature_enabled?('captain_integration_v2')
job_args << message.id if captain_v2_enabled
wait_time = attachment_wait_time(captain_v2_enabled)
if message.attachments.blank?
if wait_time.zero?
Captain::Conversation::ResponseBuilderJob.perform_later(*job_args)
else
wait_time = calculate_attachment_wait_time
Captain::Conversation::ResponseBuilderJob.set(wait: wait_time).perform_later(*job_args)
end
end
def calculate_attachment_wait_time
attachment_count = message.attachments.size
def attachment_wait_time(captain_v2_enabled)
attachment_count = captain_v2_enabled ? recent_attachment_count : message.attachments.size
return 0.seconds if attachment_count.zero?
calculate_attachment_wait_time(attachment_count)
end
def recent_attachment_count
maximum_wait = (MAX_ATTACHMENT_WAIT_SECONDS + 1).seconds
conversation.messages.incoming
.joins(:attachments)
.where(attachments: { created_at: maximum_wait.ago.. })
.count
end
def calculate_attachment_wait_time(attachment_count)
base_wait = 1.second
# Wait longer for more attachments or larger files
@@ -50,7 +67,7 @@ module Enterprise::MessageTemplates::HookExecutionService
end
def should_process_captain_response?
conversation.pending? && message.incoming? && inbox.captain_assistant.present?
conversation.pending? && message.captain_response_triggering? && inbox.captain_assistant.present?
end
def perform_handoff

30
enterprise/lib/captain/tools/base_public_tool.rb
View File

@@ -6,6 +6,14 @@ class Captain::Tools::BasePublicTool < Agents::Tool
super()
end
def execute(tool_context, **params)
return super unless captain_v2_enabled?
return super if safe_to_run_after_new_customer_message?
return 'Tool skipped because a newer customer message arrived' if newer_customer_message_arrived?(tool_context.state)
super
end
def active?
# Public tools are always active
true
@@ -37,6 +45,28 @@ class Captain::Tools::BasePublicTool < Agents::Tool
account_scoped(::Contact).find_by(id: contact_id)
end
def safe_to_run_after_new_customer_message?
false
end
def captain_v2_enabled?
@assistant.account.feature_enabled?('captain_integration_v2')
end
def newer_customer_message_arrived?(state)
responding_to_message_id = state&.dig(:responding_to_message_id)
return false if responding_to_message_id.blank?
conversation_id = state&.dig(:conversation, :id)
::Message.uncached do
account_scoped(::Message)
.where(conversation_id: conversation_id)
.captain_response_triggering
.exists?(['messages.id > ?', responding_to_message_id])
end
end
def log_tool_usage(action, details = {})
Rails.logger.info do
"#{self.class.name}: #{action} for assistant #{@assistant&.id} - #{details.inspect}"

4
enterprise/lib/captain/tools/faq_lookup_tool.rb
View File

@@ -20,6 +20,10 @@ class Captain::Tools::FaqLookupTool < Captain::Tools::BasePublicTool
private
def safe_to_run_after_new_customer_message?
true
end
def record_retrieved_sources(tool_context, responses)
return if responses.empty?

58
enterprise/lib/captain/tools/handoff_tool.rb
View File

@@ -13,7 +13,9 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
})
# Use existing handoff mechanism from ResponseBuilderJob
trigger_handoff(tool_context, conversation, reason)
handoff_result = trigger_handoff(tool_context, conversation, reason)
return 'Handoff skipped because a newer customer message arrived' if handoff_result == :stale
return 'Handoff skipped because the conversation changed' unless handoff_result == :completed
"Conversation handed off to human support team#{" (Reason: #{reason})" if reason}"
rescue StandardError => e
@@ -24,30 +26,54 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
private
def trigger_handoff(tool_context, conversation, reason)
# post the reason as a private note
note = conversation.messages.create!(
message_type: :outgoing,
private: true,
sender: @assistant,
account: conversation.account,
inbox: conversation.inbox,
content: reason
)
return trigger_legacy_handoff(tool_context, conversation, reason) unless captain_v2_enabled?
note = nil
handoff_result = conversation.with_lock do
next :changed unless conversation.pending?
next :stale if newer_customer_message_arrived?(tool_context.state)
# post the reason as a private note
note = conversation.messages.create!(
message_type: :outgoing, private: true, sender: @assistant,
account: conversation.account, inbox: conversation.inbox, content: reason
)
conversation.bot_handoff!(dispatch_event: false)
:completed
end
return handoff_result unless handoff_result == :completed
# Session capture attributes the run to this note so agents can inspect the
# generation path on the handoff reason instead of the canned follow-up message.
# A reason-less note has no content and never renders in the dashboard, so
# leave it unrecorded and let capture fall back to the follow-up message.
if reason.present?
metadata = tool_context.state[:cw_metadata] ||= {}
metadata[:handoff_note_id] = note.id
end
record_handoff_note(tool_context, note) if reason.present?
# Trigger the bot handoff (sets status to open + dispatches events)
conversation.bot_handoff!
tool_context.state[:captain_v2_handoff_tool_completed] = true
# Queue the event after the state change commits so notification jobs always see the open conversation.
conversation.dispatch_bot_handoff_event
# Send out of office message if applicable (since template messages were suppressed while Captain was handling)
send_out_of_office_message_if_applicable(conversation)
:completed
end
def trigger_legacy_handoff(tool_context, conversation, reason)
note = conversation.messages.create!(
message_type: :outgoing, private: true, sender: @assistant,
account: conversation.account, inbox: conversation.inbox, content: reason
)
record_handoff_note(tool_context, note) if reason.present?
conversation.bot_handoff!
send_out_of_office_message_if_applicable(conversation)
:completed
end
def record_handoff_note(tool_context, note)
metadata = tool_context.state[:cw_metadata] ||= {}
metadata[:handoff_note_id] = note.id
end
def send_out_of_office_message_if_applicable(conversation)

9
enterprise/lib/captain/tools/http_tool.rb
View File

@@ -1,10 +1,9 @@
require 'agents'
class Captain::Tools::HttpTool < Agents::Tool
class Captain::Tools::HttpTool < Captain::Tools::BasePublicTool
def initialize(assistant, custom_tool)
@assistant = assistant
@custom_tool = custom_tool
super()
super(assistant)
end
def active?
@@ -24,6 +23,10 @@ class Captain::Tools::HttpTool < Agents::Tool
private
def safe_to_run_after_new_customer_message?
@custom_tool.http_method == 'GET'
end
# Limit response size to prevent memory exhaustion and match LLM token limits
# 1MB of text ≈ 250K tokens, which exceeds most LLM context windows
MAX_RESPONSE_SIZE = 1.megabyte

69
spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb
View File

@@ -13,6 +13,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
let(:mock_action_classifier_service) { instance_double(Captain::Llm::AssistantActionClassifierService) }
let(:mock_false_promise_service) { instance_double(Captain::Llm::AssistantFalsePromiseService) }
let(:assistant_model) { Llm::Models.default_model_for('assistant') }
let(:responding_to_message) { conversation.messages.find_by!(content: 'Hello') }
before do
create(:message, conversation: conversation, content: 'Hello', message_type: :incoming)
@@ -23,6 +24,8 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
allow(Captain::Assistant::AgentRunnerService).to receive(:new).and_return(mock_agent_runner_service)
allow(mock_agent_runner_service).to receive(:generate_response).and_return({ 'response' => 'Hey, welcome to Captain V2' })
allow(mock_agent_runner_service).to receive(:last_run_result).and_return(nil)
allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(false)
allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(false)
allow(Captain::Llm::AssistantActionClassifierService).to receive(:new).and_return(mock_action_classifier_service)
allow(mock_action_classifier_service).to receive(:classify).and_return({ 'action' => 'continue' })
allow(Captain::Llm::AssistantFalsePromiseService).to receive(:new).and_return(mock_false_promise_service)
@@ -355,15 +358,57 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
allow(account).to receive(:feature_enabled?).with('captain_integration_v2').and_return(true)
end
it 'uses Captain::Assistant::AgentRunnerService' do
expect(Captain::Assistant::AgentRunnerService).to receive(:new).with(
assistant: assistant,
conversation: conversation
)
expect(Captain::Llm::AssistantChatService).not_to receive(:new)
context 'with message burst protection' do
it 'passes the responding message id to the runner' do
expect(Captain::Assistant::AgentRunnerService).to receive(:new).with(
assistant: assistant,
conversation: conversation,
responding_to_message_id: responding_to_message.id
)
described_class.perform_now(conversation, assistant)
expect(conversation.messages.last.content).to eq('Hey, welcome to Captain V2')
described_class.perform_now(conversation, assistant, responding_to_message.id)
end
it 'discards a response when the runner sees a newer customer message' do
allow(mock_agent_runner_service).to receive(:generate_response) do
create(:message, conversation: conversation, content: 'New context', message_type: :incoming)
{ 'response' => 'Stale response', 'handoff_tool_called' => false }
end
allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(true)
described_class.perform_now(conversation, assistant, responding_to_message.id)
expect(conversation.messages.outgoing.count).to eq(0)
expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(0)
end
it 'checks freshness itself after generation' do
allow(mock_agent_runner_service).to receive(:generate_response) do
create(:message, conversation: conversation, content: 'New context', message_type: :incoming)
{ 'response' => 'Stale response', 'handoff_tool_called' => false }
end
allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(false)
described_class.perform_now(conversation, assistant, responding_to_message.id)
expect(conversation.messages.outgoing.count).to eq(0)
expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(0)
end
it 'keeps the pending response fresh when an email auto reply arrives' do
create(
:message,
conversation: conversation,
message_type: :incoming,
content_type: :incoming_email,
content_attributes: { email: { auto_reply: true } }
)
described_class.perform_now(conversation, assistant, responding_to_message.id)
expect(conversation.messages.outgoing.last.content).to eq('Hey, welcome to Captain V2')
expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(1)
end
end
it 'passes message history with resolution markers to agent runner service' do
@@ -417,9 +462,11 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
before do
allow(account).to receive(:feature_enabled?).and_return(false)
allow(account).to receive(:feature_enabled?).with('captain_integration_v2').and_return(true)
allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(true)
end
it 'creates a public handoff message visible to the customer' do
it 'creates a public handoff message after the generated response is discarded' do
allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(true)
allow(mock_agent_runner_service).to receive(:generate_response) do
conversation.update!(status: :open)
{ 'response' => 'Let me connect you', 'handoff_tool_called' => true }
@@ -469,6 +516,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
end
it 'does not hand off when handoff_tool_called is false' do
allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(false)
allow(mock_agent_runner_service).to receive(:generate_response).and_return({
'response' => 'Hi! How can I help you?',
'handoff_tool_called' => false
@@ -482,6 +530,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
end
it 'falls back to a full V1 handoff when HandoffTool fired but failed to commit' do
allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(false)
allow(mock_agent_runner_service).to receive(:generate_response).and_return({
'response' => 'I tried to hand off',
'handoff_tool_called' => true
@@ -548,6 +597,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
end
it 'creates a zero-credit session when the handoff tool fired' do
allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(true)
allow(mock_agent_runner_service).to receive(:generate_response) do
conversation.update!(status: :open)
{ 'response' => 'Let me connect you', 'handoff_tool_called' => true }
@@ -562,6 +612,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
end
it 'attributes the handoff session to the private reason note when the tool recorded one' do
allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(true)
handoff_note = create(:message, conversation: conversation, account: account, message_type: :outgoing,
private: true, sender: assistant, content: 'Needs a human')
run_context[:state][:cw_metadata][:handoff_note_id] = handoff_note.id

30
spec/enterprise/lib/captain/tools/add_private_note_tool_spec.rb
View File

@@ -116,6 +116,36 @@ RSpec.describe Captain::Tools::AddPrivateNoteTool, type: :model do
end
end
describe '#execute' do
let(:responding_to_message) do
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
end
let(:tool_context) do
Struct.new(:state).new({ conversation: { id: conversation.id }, responding_to_message_id: responding_to_message.id })
end
before do
responding_to_message
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
end
it 'keeps legacy public-tool side effects for Captain V1' do
expect do
result = tool.execute(tool_context, note: 'Keep the legacy side effect')
expect(result).to eq('Private note added successfully')
end.to change(Message, :count).by(1)
end
it 'skips stale public-tool side effects for Captain V2' do
account.enable_features!(:captain_integration_v2)
expect do
result = tool.execute(tool_context, note: 'Do not create this note')
expect(result).to eq('Tool skipped because a newer customer message arrived')
end.not_to change(Message, :count)
end
end
describe '#active?' do
it 'returns true for public tools' do
expect(tool.active?).to be true

94
spec/enterprise/lib/captain/tools/handoff_tool_spec.rb
View File

@@ -7,7 +7,7 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
let(:user) { create(:user, account: account) }
let(:inbox) { create(:inbox, account: account) }
let(:contact) { create(:contact, account: account) }
let(:conversation) { create(:conversation, account: account, inbox: inbox, contact: contact) }
let(:conversation) { create(:conversation, account: account, inbox: inbox, contact: contact, status: :pending) }
let(:tool_context) { Struct.new(:state).new({ conversation: { id: conversation.id } }) }
describe '#description' do
@@ -28,6 +28,98 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
describe '#perform' do
context 'when conversation exists' do
context 'when Captain is responding to a customer message' do
let(:responding_to_message) do
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
end
let(:tool_context) do
Struct.new(:state).new({ conversation: { id: conversation.id }, responding_to_message_id: responding_to_message.id })
end
before do
account.enable_features!(:captain_integration_v2)
responding_to_message
end
it 'hands off when no newer customer message has arrived' do
found_conversation = Conversation.find(conversation.id)
scoped_conversations = Conversation.where(account_id: assistant.account_id)
allow(Conversation).to receive(:where).with(account_id: assistant.account_id).and_return(scoped_conversations)
allow(scoped_conversations).to receive(:find_by).with(id: conversation.id).and_return(found_conversation)
expect(found_conversation).to receive(:with_lock).and_call_original
expect do
result = tool.perform(tool_context, reason: 'Customer needs specialized support')
expect(result).to include('Conversation handed off')
end.to change(Message, :count).by(1)
expect(tool_context.state[:captain_v2_handoff_tool_completed]).to be true
end
it 'dispatches the handoff event after leaving the lock transaction' do
found_conversation = Conversation.find(conversation.id)
scoped_conversations = Conversation.where(account_id: assistant.account_id)
allow(Conversation).to receive(:where).with(account_id: assistant.account_id).and_return(scoped_conversations)
allow(scoped_conversations).to receive(:find_by).with(id: conversation.id).and_return(found_conversation)
open_transactions_before_handoff = ActiveRecord::Base.connection.open_transactions
expect(found_conversation).to receive(:dispatch_bot_handoff_event) do
expect(ActiveRecord::Base.connection.open_transactions).to eq(open_transactions_before_handoff)
end
tool.perform(tool_context, reason: 'Customer needs specialized support')
end
it 'notifies inbox members after the committed handoff' do
create(:inbox_member, user: user, inbox: inbox)
notification_setting = user.notification_settings.find_by!(account: account)
notification_setting.selected_email_flags = [:email_conversation_creation]
notification_setting.selected_push_flags = []
notification_setting.save!
perform_enqueued_jobs do
tool.perform(tool_context, reason: 'Customer needs specialized support')
end
expect(user.notifications.find_by(primary_actor: conversation, notification_type: :conversation_creation)).to be_present
end
it 'skips the handoff when a newer message has arrived' do
conversation.update!(status: :pending)
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
expect do
result = tool.perform(tool_context, reason: 'Customer needs specialized support')
expect(result).to eq('Handoff skipped because a newer customer message arrived')
end.not_to change(Message, :count)
expect(conversation.reload.status).to eq('pending')
end
end
context 'with Captain V1' do
let(:tool_context) do
Struct.new(:state).new({ conversation: { id: conversation.id }, responding_to_message_id: responding_to_message.id })
end
let(:responding_to_message) do
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
end
it 'uses the legacy handoff without a lock or stale-message guard' do
responding_to_message
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
found_conversation = Conversation.find(conversation.id)
scoped_conversations = Conversation.where(account_id: assistant.account_id)
allow(Conversation).to receive(:where).with(account_id: assistant.account_id).and_return(scoped_conversations)
allow(scoped_conversations).to receive(:find_by).with(id: conversation.id).and_return(found_conversation)
expect(found_conversation).not_to receive(:with_lock)
result = tool.perform(tool_context, reason: 'Customer needs specialized support')
expect(result).to include('Conversation handed off')
expect(conversation.reload.status).to eq('open')
expect(tool_context.state).not_to have_key(:captain_v2_handoff_tool_completed)
end
end
context 'with reason provided' do
it 'creates a private note with reason and hands off conversation' do
reason = 'Customer needs specialized support'

98
spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb
View File

@@ -52,6 +52,12 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
expect(service.instance_variable_get(:@callbacks)).to eq(callbacks)
end
it 'accepts the message id it is responding to' do
service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: 123)
expect(service.instance_variable_get(:@responding_to_message_id)).to eq(123)
end
end
describe '#generate_response' do
@@ -99,6 +105,18 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
service.generate_response(message_history: message_history)
end
it 'adds the responding message id to the runner state' do
service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: 123)
expect(mock_runner).to receive(:run).with(
'I need help with my account',
context: hash_including(state: hash_including(responding_to_message_id: 123)),
max_turns: 10
)
service.generate_response(message_history: message_history)
end
context 'when the latest user message is multimodal' do
let(:multimodal_message_history) do
[
@@ -446,6 +464,19 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
attributes = provider.generation_attributes(nil, nil, message)
expect(attributes['langfuse.observation.metadata.generation_stage']).to eq('final_response')
expect(attributes).not_to have_key('langfuse.observation.metadata.discarded')
end
it 'marks a protected generation as not discarded when no newer message has arrived' do
responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
runner_service = described_class.new(assistant: assistant, conversation: conversation,
responding_to_message_id: responding_to_message.id)
attribute_provider = Captain::Assistant::InstrumentationAttributeProvider.new(runner_service)
message = instance_double(RubyLLM::Message, tool_calls: {})
attributes = attribute_provider.generation_attributes(nil, nil, message)
expect(attributes['langfuse.observation.metadata.discarded']).to eq('false')
end
it 'marks tool call generations separately from final responses' do
@@ -456,6 +487,19 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
expect(attributes['langfuse.observation.metadata.generation_stage']).to eq('tool_call')
end
it 'marks a generation as discarded when a newer message has arrived' do
responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
runner_service = described_class.new(assistant: assistant, conversation: conversation,
responding_to_message_id: responding_to_message.id)
attribute_provider = Captain::Assistant::InstrumentationAttributeProvider.new(runner_service)
message = instance_double(RubyLLM::Message, tool_calls: {})
create(:message, conversation: conversation, message_type: :incoming)
attributes = attribute_provider.generation_attributes(nil, nil, message)
expect(attributes['langfuse.observation.metadata.discarded']).to eq('true')
end
end
describe '#build_state' do
@@ -588,18 +632,40 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
tool_complete_callback = block
runner
end
allow(runner).to receive(:on_run_complete).and_return(runner)
service.send(:add_usage_metadata_callback, runner)
context_wrapper = Struct.new(:context).new({})
context_wrapper = Struct.new(:context).new({ state: { captain_v2_handoff_tool_completed: true } })
expect(tool_complete_callback).not_to be_nil
tool_complete_callback.call(Captain::Tools::HandoffTool.new(assistant).name, 'ok', context_wrapper)
expect(context_wrapper.context[:captain_v2_handoff_tool_called]).to be true
expect(service.handoff_completed?).to be true
end
it 'does not register OTEL run callback when OTEL is disabled' do
it 'tracks discarded responses when OTEL is disabled' do
responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: responding_to_message.id)
runner = instance_double(Agents::AgentRunner)
run_complete_callback = nil
allow(ChatwootApp).to receive(:otel_enabled?).and_return(false)
allow(runner).to receive(:on_tool_complete).and_return(runner)
allow(runner).to receive(:on_run_complete) do |&block|
run_complete_callback = block
runner
end
service.send(:add_usage_metadata_callback, runner)
create(:message, conversation: conversation, message_type: :incoming)
run_complete_callback.call('assistant', nil, Struct.new(:context).new({}))
expect(service.response_discarded?).to be true
end
it 'does not register a run callback when OTEL and burst protection are disabled' do
service = described_class.new(assistant: assistant, conversation: conversation)
runner = instance_double(Agents::AgentRunner)
@@ -632,6 +698,34 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'true')
run_complete_callback.call('assistant', nil, context_wrapper)
end
it 'marks the trace discarded and does not use credit when a newer message arrived' do
responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: responding_to_message.id)
runner = instance_double(Agents::AgentRunner)
run_complete_callback = nil
span_class = Class.new do
def set_attribute(*); end
end
root_span = instance_double(span_class)
context_wrapper = Struct.new(:context).new({ __otel_tracing: { root_span: root_span } })
allow(ChatwootApp).to receive(:otel_enabled?).and_return(true)
allow(runner).to receive(:on_tool_complete).and_return(runner)
allow(runner).to receive(:on_run_complete) do |&block|
run_complete_callback = block
runner
end
service.send(:add_usage_metadata_callback, runner)
create(:message, conversation: conversation, message_type: :incoming)
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.discarded', 'true')
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'false')
run_complete_callback.call('assistant', nil, context_wrapper)
expect(service.response_discarded?).to be true
end
end
describe 'constants' do

62
spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb
View File

@@ -21,10 +21,68 @@ RSpec.describe MessageTemplates::HookExecutionService do
)
end
it 'schedules captain response job for incoming messages on pending conversations' do
expect(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later).with(conversation, assistant)
it 'keeps the legacy job arguments for Captain V1' do
allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
create(:message, conversation: conversation, message_type: :incoming, account: account)
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).with(conversation, assistant)
end
it 'passes the responding message id for Captain V2' do
account.enable_features!(:captain_integration_v2)
allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
message = create(:message, conversation: conversation, message_type: :incoming, account: account)
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).with(conversation, assistant, message.id)
end
it 'does not lock or schedule a job for an email auto reply' do
account.enable_features!(:captain_integration_v2)
allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
customer_message = create(:message, conversation: conversation, message_type: :incoming, account: account)
auto_reply = build(
:message,
conversation: conversation,
message_type: :incoming,
content_type: :incoming_email,
content_attributes: { email: { auto_reply: true } },
account: account
)
auto_reply.save!
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).once
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).with(conversation, assistant, customer_message.id)
expect(conversation.messages.captain_response_triggering).to contain_exactly(customer_message)
expect(conversation.messages.captain_response_triggering).not_to include(auto_reply)
end
end
context 'when calculating attachment wait time' do
let(:configured_job) { instance_double(ActiveJob::ConfiguredJob, perform_later: true) }
before do
allow(Captain::Conversation::ResponseBuilderJob).to receive(:set).and_return(configured_job)
end
it 'uses only the current message attachments for Captain V1' do
create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:set).with(wait: 2.seconds).twice
expect(Captain::Conversation::ResponseBuilderJob).not_to have_received(:set).with(wait: 3.seconds)
end
it 'recalculates the wait from recent burst attachments for Captain V2' do
account.enable_features!(:captain_integration_v2)
create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:set).with(wait: 2.seconds).once
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:set).with(wait: 3.seconds).once
end
end