From bba230a1acddd1fdc94393588cf4e6482f072e2b Mon Sep 17 00:00:00 2001 From: Shivam Mishra Date: Fri, 22 May 2026 21:20:18 +0530 Subject: [PATCH] feat: snapshot and restore metadata from each run --- .../conversation/response_builder_job.rb | 124 ++++++++++++++++-- enterprise/app/models/enterprise/message.rb | 19 +++ .../captain/assistant/agent_runner_service.rb | 53 ++++++-- .../conversation/response_builder_job_spec.rb | 122 +++++++++++++++++ spec/enterprise/models/message_spec.rb | 41 ++++++ .../assistant/agent_runner_service_spec.rb | 83 +++++++++++- 6 files changed, 415 insertions(+), 27 deletions(-) diff --git a/enterprise/app/jobs/captain/conversation/response_builder_job.rb b/enterprise/app/jobs/captain/conversation/response_builder_job.rb index 5050f11b2..6459aea9f 100644 --- a/enterprise/app/jobs/captain/conversation/response_builder_job.rb +++ b/enterprise/app/jobs/captain/conversation/response_builder_job.rb @@ -84,17 +84,26 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob .messages .where(message_type: [:incoming, :outgoing]) .where(private: false) - .map do |message| - message_hash = { - content: prepare_multimodal_message_content(message), - role: determine_role(message) - } + .flat_map { |message| history_entries_for_message(message) } + end - # Include agent_name if present in additional_attributes - message_hash[:agent_name] = message.additional_attributes['agent_name'] if message.additional_attributes&.dig('agent_name').present? + def history_entries_for_message(message) + captain_run_context = message.captain_run_context if captain_v2_enabled? && message.outgoing? + return captain_run_context['messages'] if captain_run_context&.dig('messages').present? - message_hash - end + [legacy_history_entry_for_message(message)] + end + + def legacy_history_entry_for_message(message) + message_hash = { + content: prepare_multimodal_message_content(message), + role: determine_role(message) + } + + agent_name = message.additional_attributes&.dig('captain', 'agent', 'name') || message.additional_attributes&.dig('agent_name') + message_hash[:agent_name] = agent_name if agent_name.present? + + message_hash end def determine_role(message) @@ -153,22 +162,26 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob def create_handoff_message(preserve_waiting_since: false) create_outgoing_message( @assistant.config['handoff_message'].presence || I18n.t('conversations.captain.handoff'), + agent_name: captain_v2_enabled? ? @response&.dig('agent_name') || assistant_agent_name : nil, + captain_metadata: captain_handoff_metadata, preserve_waiting_since: preserve_waiting_since ) end def create_messages validate_message_content!(@response['response']) - create_outgoing_message(@response['response'], agent_name: @response['agent_name']) + create_outgoing_message(@response['response'], agent_name: @response['agent_name'], captain_metadata: captain_response_metadata) end def validate_message_content!(content) raise ArgumentError, 'Message content cannot be blank' if content.blank? end - def create_outgoing_message(message_content, agent_name: nil, preserve_waiting_since: false) + def create_outgoing_message(message_content, agent_name: nil, captain_metadata: nil, preserve_waiting_since: false) additional_attrs = {} - additional_attrs[:agent_name] = agent_name if agent_name.present? + agent_name ||= captain_metadata&.dig('agent', 'name') + additional_attrs['agent_name'] = agent_name if agent_name.present? + additional_attrs['captain'] = captain_metadata if captain_metadata.present? @conversation.messages.create!( message_type: :outgoing, @@ -181,6 +194,93 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob ) end + def captain_response_metadata + return unless captain_v2_enabled? + + { + 'version' => 'v2', + 'agent' => captain_agent_metadata(@response['agent_name']), + 'run' => captain_response_run_context + } + end + + def captain_handoff_metadata + return unless captain_v2_enabled? + + { + 'version' => 'v2', + 'agent' => captain_agent_metadata(@response&.dig('agent_name') || assistant_agent_name), + 'run' => { + 'messages' => [], + 'handoff_tool_called' => true + } + } + end + + def captain_response_run_context + run_context = (@response['run_context'] || {}).deep_stringify_keys + messages = Array(run_context['messages']) + messages = fallback_captain_run_messages if messages.empty? + + { + 'messages' => messages, + 'handoff_tool_called' => @response['handoff_tool_called'] || false + } + end + + def fallback_captain_run_messages + content = {} + content['reasoning'] = @response['reasoning'] if @response['reasoning'].present? + + [ + { + 'role' => 'assistant', + 'content' => content, + 'agent_name' => @response['agent_name'] + }.compact + ] + end + + def captain_agent_metadata(agent_name) + agent_name = agent_name.presence || assistant_agent_name + scenario = @assistant.scenarios.enabled.find { |enabled_scenario| enabled_scenario.handoff_key == agent_name } + + return scenario_agent_metadata(scenario, agent_name) if scenario + return assistant_agent_metadata(agent_name) if agent_name == assistant_agent_name + + unknown_agent_metadata(agent_name) + end + + def scenario_agent_metadata(scenario, agent_name) + { + 'name' => agent_name, + 'type' => 'scenario', + 'assistant_id' => @assistant.id, + 'scenario_id' => scenario.id, + 'handoff_key' => scenario.handoff_key + } + end + + def assistant_agent_metadata(agent_name) + { + 'name' => agent_name, + 'type' => 'assistant', + 'assistant_id' => @assistant.id + } + end + + def unknown_agent_metadata(agent_name) + { + 'name' => agent_name, + 'type' => 'unknown', + 'assistant_id' => @assistant.id + } + end + + def assistant_agent_name + @assistant.name.parameterize(separator: '_') + end + def handle_error(error) log_error(error) @response ||= {} diff --git a/enterprise/app/models/enterprise/message.rb b/enterprise/app/models/enterprise/message.rb index a6336fc55..63e75ee83 100644 --- a/enterprise/app/models/enterprise/message.rb +++ b/enterprise/app/models/enterprise/message.rb @@ -13,8 +13,27 @@ module Enterprise::Message data end + def captain_run_context + captain = additional_attributes&.dig('captain') + return unless captain&.dig('version') == 'v2' + + run_context = (captain['run'] || {}).deep_dup + messages = Array(run_context['messages']).map(&:deep_dup) + restore_captain_final_response(messages) + + run_context.merge('messages' => messages) + end + private + def restore_captain_final_response(messages) + final_assistant_message = messages.reverse.find { |message| message['role'] == 'assistant' } + return unless final_assistant_message + + content_hash = final_assistant_message['content'].is_a?(Hash) ? final_assistant_message['content'] : {} + final_assistant_message['content'] = content_hash.merge('response' => content) + end + def mark_pending_conversation_as_open_for_human_response return unless captain_pending_conversation? return unless human_response? diff --git a/enterprise/app/services/captain/assistant/agent_runner_service.rb b/enterprise/app/services/captain/assistant/agent_runner_service.rb index 21a9c331e..e7d72dc4d 100644 --- a/enterprise/app/services/captain/assistant/agent_runner_service.rb +++ b/enterprise/app/services/captain/assistant/agent_runner_service.rb @@ -45,15 +45,14 @@ class Captain::Assistant::AgentRunnerService def build_context(message_history) conversation_history = message_history.map do |msg| - content = msg[:content] - # Preserve multimodal arrays (with image_url entries) as-is for the runner to restore with attachments. - # Only extract text from non-array formats (hashes from agent structured output, plain strings). - content = extract_text_from_content(content) unless content.is_a?(Array) + content = message_value(msg, :content) + # Preserve multimodal arrays and structured agent output as-is for replay. + content = extract_text_from_content(content) unless content.is_a?(Array) || content.is_a?(Hash) { - role: msg[:role].to_sym, + role: message_value(msg, :role).to_sym, content: content, - agent_name: msg[:agent_name] + agent_name: message_value(msg, :agent_name) } end @@ -65,10 +64,10 @@ class Captain::Assistant::AgentRunnerService end def extract_last_user_message(message_history) - last_user_msg = message_history.reverse.find { |msg| msg[:role] == 'user' } + last_user_msg = message_history.reverse.find { |msg| message_value(msg, :role).to_s == 'user' } return '' if last_user_msg.blank? - content = last_user_msg[:content] + content = message_value(last_user_msg, :content) return extract_text_from_content(content) unless content.is_a?(Array) text, attachments = Captain::OpenAiMessageBuilderService.extract_text_and_attachments(content) @@ -78,12 +77,16 @@ class Captain::Assistant::AgentRunnerService end def message_history_without_last_user_message(message_history) - last_user_index = message_history.rindex { |msg| msg[:role] == 'user' } + last_user_index = message_history.rindex { |msg| message_value(msg, :role).to_s == 'user' } return message_history if last_user_index.nil? message_history.reject.with_index { |_msg, index| index == last_user_index } end + def message_value(message, key) + message[key] || message[key.to_s] + end + def extract_text_from_content(content) # Handle structured output from agents return content[:response] || content['response'] || content.to_s if content.is_a?(Hash) @@ -100,9 +103,41 @@ class Captain::Assistant::AgentRunnerService response = output.is_a?(Hash) ? output.with_indifferent_access : { 'response' => output.to_s, 'reasoning' => 'Processed by agent' } response['agent_name'] = result.context&.dig(:current_agent) response['handoff_tool_called'] = result.context&.dig(:captain_v2_handoff_tool_called) || false + response['run_context'] = build_run_context(result, response) response end + def build_run_context(result, response) + { + 'messages' => normalized_run_messages(result.messages || []), + 'handoff_tool_called' => response['handoff_tool_called'] + } + end + + def normalized_run_messages(messages) + normalized_messages = messages.map { |message| normalize_run_message(message) } + final_assistant_message = normalized_messages.reverse.find { |message| message['role'] == 'assistant' } + strip_final_response(final_assistant_message) if final_assistant_message + normalized_messages + end + + def normalize_run_message(message) + normalized = message.deep_stringify_keys + normalized['role'] = normalized['role'].to_s if normalized['role'].present? + normalized['content'] = normalized['content'].deep_stringify_keys if normalized['content'].is_a?(Hash) + normalized + end + + def strip_final_response(message) + message['content'] = normalized_final_message_content(message['content']) + end + + def normalized_final_message_content(content) + return content.except('response') if content.is_a?(Hash) + + {} + end + def error_response(error_message) { 'response' => 'conversation_handoff', diff --git a/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb b/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb index 8fac81d60..7617689af 100644 --- a/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb +++ b/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb @@ -203,6 +203,124 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do account.reload expect(account.usage_limits[:captain][:responses][:consumed]).to eq(1) end + + it 'stores normalized Captain run metadata on the outgoing message' do + agent_name = assistant.name.parameterize(separator: '_') + allow(mock_agent_runner_service).to receive(:generate_response).and_return({ + 'response' => 'Hey, welcome to Captain V2', + 'agent_name' => agent_name, + 'handoff_tool_called' => false, + 'run_context' => { + 'messages' => [ + { + 'role' => 'assistant', + 'content' => { 'reasoning' => 'Answered from FAQ' }, + 'agent_name' => agent_name + } + ], + 'handoff_tool_called' => false + } + }) + + described_class.perform_now(conversation, assistant) + + attributes = conversation.messages.outgoing.last.additional_attributes + expect(attributes['agent_name']).to eq(agent_name) + expect(attributes.dig('captain', 'version')).to eq('v2') + expect(attributes.dig('captain', 'agent')).to eq({ + 'name' => agent_name, + 'type' => 'assistant', + 'assistant_id' => assistant.id + }) + expect(attributes.dig('captain', 'run', 'messages').last['content']).to eq({ 'reasoning' => 'Answered from FAQ' }) + end + + it 'stores scenario agent metadata when the scenario handles the response' do + scenario = create(:captain_scenario, assistant: assistant, account: account, title: 'Billing support') + allow(mock_agent_runner_service).to receive(:generate_response).and_return({ + 'response' => 'Billing answer', + 'agent_name' => scenario.handoff_key, + 'handoff_tool_called' => false + }) + + described_class.perform_now(conversation, assistant) + + expect(conversation.messages.outgoing.last.additional_attributes.dig('captain', 'agent')).to eq({ + 'name' => scenario.handoff_key, + 'type' => 'scenario', + 'assistant_id' => assistant.id, + 'scenario_id' => scenario.id, + 'handoff_key' => scenario.handoff_key + }) + end + + it 'replays persisted Captain run messages as message history' do + agent_name = assistant.name.parameterize(separator: '_') + create( + :message, + conversation: conversation, + account: account, + inbox: inbox, + sender: assistant, + message_type: :outgoing, + content: 'Prior visible answer', + additional_attributes: { + 'captain' => { + 'version' => 'v2', + 'agent' => { 'name' => agent_name, 'type' => 'assistant', 'assistant_id' => assistant.id }, + 'run' => { + 'messages' => [ + { + 'role' => 'assistant', + 'content' => '', + 'agent_name' => agent_name, + 'tool_calls' => [ + { + 'id' => 'call_1', + 'name' => 'captain--tools--faq_lookup', + 'arguments' => { 'query' => 'billing' } + } + ] + }, + { 'role' => 'tool', 'content' => 'FAQ result', 'tool_call_id' => 'call_1' }, + { + 'role' => 'assistant', + 'content' => { 'reasoning' => 'Prior reasoning' }, + 'agent_name' => agent_name + } + ], + 'handoff_tool_called' => false + } + } + } + ) + + expected_messages = [ + { content: 'Hello', role: 'user' }, + { + 'role' => 'assistant', + 'content' => '', + 'agent_name' => agent_name, + 'tool_calls' => [ + { + 'id' => 'call_1', + 'name' => 'captain--tools--faq_lookup', + 'arguments' => { 'query' => 'billing' } + } + ] + }, + { 'role' => 'tool', 'content' => 'FAQ result', 'tool_call_id' => 'call_1' }, + { + 'role' => 'assistant', + 'content' => { 'reasoning' => 'Prior reasoning', 'response' => 'Prior visible answer' }, + 'agent_name' => agent_name + } + ] + + expect(mock_agent_runner_service).to receive(:generate_response).with(message_history: expected_messages) + + described_class.perform_now(conversation, assistant) + end end context 'when captain_v2 handoff tool fires during agent execution' do @@ -222,6 +340,10 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do public_messages = conversation.messages.outgoing.where(private: false) expect(public_messages.count).to eq(1) expect(public_messages.last.content).to eq(I18n.t('conversations.captain.handoff')) + expect(public_messages.last.additional_attributes.dig('captain', 'run')).to eq({ + 'messages' => [], + 'handoff_tool_called' => true + }) end it 'does not call bot_handoff! again when conversation is already open' do diff --git a/spec/enterprise/models/message_spec.rb b/spec/enterprise/models/message_spec.rb index 36311a567..8dc560619 100644 --- a/spec/enterprise/models/message_spec.rb +++ b/spec/enterprise/models/message_spec.rb @@ -24,6 +24,47 @@ RSpec.describe Message do expect(conversation.waiting_since).to be_nil end + describe '#captain_run_context' do + it 'restores the final response from message content without mutating additional attributes' do + message = create( + :message, + message_type: :outgoing, + conversation: conversation, + content: 'Visible response', + additional_attributes: { + 'captain' => { + 'version' => 'v2', + 'run' => { + 'messages' => [ + { 'role' => 'tool', 'content' => 'FAQ result', 'tool_call_id' => 'call_1' }, + { + 'role' => 'assistant', + 'content' => { 'reasoning' => 'Used FAQ' }, + 'agent_name' => 'assistant' + } + ], + 'handoff_tool_called' => false + } + } + } + ) + + run_context = message.captain_run_context + + expect(run_context['messages'].last['content']).to eq({ + 'reasoning' => 'Used FAQ', + 'response' => 'Visible response' + }) + expect(message.additional_attributes.dig('captain', 'run', 'messages').last['content']).to eq({ 'reasoning' => 'Used FAQ' }) + end + + it 'returns nil when the message has no Captain V2 run context' do + message = create(:message, message_type: :outgoing, conversation: conversation, additional_attributes: {}) + + expect(message.captain_run_context).to be_nil + end + end + describe '#mark_pending_conversation_as_open_for_human_response' do let(:conversation) { create(:conversation, status: :pending) } let(:captain_assistant) { create(:captain_assistant, account: conversation.account) } diff --git a/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb b/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb index d6e57e710..3de687a7c 100644 --- a/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb +++ b/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb @@ -13,7 +13,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do let(:mock_runner) { instance_double(Agents::AgentRunner) } let(:mock_agent) { instance_double(Agents::Agent) } let(:mock_scenario_agent) { instance_double(Agents::Agent) } - let(:mock_result) { instance_double(Agents::RunResult, output: { 'response' => 'Test response' }, context: nil) } + let(:mock_result) { instance_double(Agents::RunResult, output: { 'response' => 'Test response' }, context: nil, messages: []) } let(:message_history) do [ @@ -167,13 +167,67 @@ RSpec.describe Captain::Assistant::AgentRunnerService do it 'processes and formats agent result' do result = service.generate_response(message_history: message_history) - expect(result).to eq({ 'response' => 'Test response', 'agent_name' => nil, 'handoff_tool_called' => false }) + expect(result).to eq({ + 'response' => 'Test response', + 'agent_name' => nil, + 'handoff_tool_called' => false, + 'run_context' => { + 'messages' => [], + 'handoff_tool_called' => false + } + }) + end + + it 'normalizes run messages and removes duplicated final response' do + runner_messages = [ + { + role: :assistant, + content: '', + agent_name: 'assistant', + tool_calls: [{ id: 'call_1', name: 'captain--tools--faq_lookup', arguments: { query: 'billing' } }] + }, + { role: :tool, content: 'FAQ result', tool_call_id: 'call_1' }, + { + role: :assistant, + content: { response: 'Test response', reasoning: 'Answered from FAQ' }, + agent_name: scenario.handoff_key + } + ] + allow(mock_result).to receive(:messages).and_return(runner_messages) + allow(mock_result).to receive(:context).and_return({ current_agent: scenario.handoff_key }) + + result = service.generate_response(message_history: message_history) + + expect(result['agent_name']).to eq(scenario.handoff_key) + expect(result['run_context']).to eq({ + 'messages' => [ + { + 'role' => 'assistant', + 'content' => '', + 'agent_name' => 'assistant', + 'tool_calls' => [ + { + 'id' => 'call_1', + 'name' => 'captain--tools--faq_lookup', + 'arguments' => { 'query' => 'billing' } + } + ] + }, + { 'role' => 'tool', 'content' => 'FAQ result', 'tool_call_id' => 'call_1' }, + { + 'role' => 'assistant', + 'content' => { 'reasoning' => 'Answered from FAQ' }, + 'agent_name' => scenario.handoff_key + } + ], + 'handoff_tool_called' => false + }) end context 'when handoff tool was called during agent execution' do let(:runner_context) { { captain_v2_handoff_tool_called: true } } let(:mock_result) do - instance_double(Agents::RunResult, output: { 'response' => 'Let me connect you' }, context: runner_context) + instance_double(Agents::RunResult, output: { 'response' => 'Let me connect you' }, context: runner_context, messages: []) end it 'includes handoff_tool_called flag in response' do @@ -182,7 +236,11 @@ RSpec.describe Captain::Assistant::AgentRunnerService do expect(result).to eq({ 'response' => 'Let me connect you', 'agent_name' => nil, - 'handoff_tool_called' => true + 'handoff_tool_called' => true, + 'run_context' => { + 'messages' => [], + 'handoff_tool_called' => true + } }) end end @@ -203,7 +261,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do end context 'when agent result is a string' do - let(:mock_result) { instance_double(Agents::RunResult, output: 'Simple string response', context: nil) } + let(:mock_result) { instance_double(Agents::RunResult, output: 'Simple string response', context: nil, messages: []) } it 'formats string response correctly' do result = service.generate_response(message_history: message_history) @@ -212,7 +270,11 @@ RSpec.describe Captain::Assistant::AgentRunnerService do 'response' => 'Simple string response', 'reasoning' => 'Processed by agent', 'agent_name' => nil, - 'handoff_tool_called' => false + 'handoff_tool_called' => false, + 'run_context' => { + 'messages' => [], + 'handoff_tool_called' => false + } }) end end @@ -306,6 +368,15 @@ RSpec.describe Captain::Assistant::AgentRunnerService do ) end + it 'preserves structured assistant content in conversation history' do + structured_content = { 'response' => 'Visible answer', 'reasoning' => 'Used FAQ', 'custom' => { 'score' => 0.9 } } + context = service.send(:build_context, [ + { role: 'assistant', content: structured_content, agent_name: 'assistant' } + ]) + + expect(context[:conversation_history].first[:content]).to eq(structured_content) + end + context 'with multimodal content' do let(:multimodal_content) do [