diff --git a/app/dispatchers/sync_dispatcher.rb b/app/dispatchers/sync_dispatcher.rb
index 509a42727..837e8af7c 100644
--- a/app/dispatchers/sync_dispatcher.rb
+++ b/app/dispatchers/sync_dispatcher.rb
@@ -8,3 +8,5 @@ class SyncDispatcher < BaseDispatcher
     [ActionCableListener.instance, AgentBotListener.instance]
   end
 end
+
+SyncDispatcher.prepend_mod_with('SyncDispatcher')
diff --git a/enterprise/app/dispatchers/enterprise/sync_dispatcher.rb b/enterprise/app/dispatchers/enterprise/sync_dispatcher.rb
new file mode 100644
index 000000000..3e32eb6fc
--- /dev/null
+++ b/enterprise/app/dispatchers/enterprise/sync_dispatcher.rb
@@ -0,0 +1,7 @@
+module Enterprise::SyncDispatcher
+  def listeners
+    super + [
+      Captain::ConversationOutcomeEventListener.instance
+    ]
+  end
+end
diff --git a/enterprise/app/jobs/captain/conversation/response_builder_job.rb b/enterprise/app/jobs/captain/conversation/response_builder_job.rb
index 45b6d9ad2..66f3259b0 100644
--- a/enterprise/app/jobs/captain/conversation/response_builder_job.rb
+++ b/enterprise/app/jobs/captain/conversation/response_builder_job.rb
@@ -88,7 +88,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     return unless conversation_pending?
 
     process_v1_handoff
-    record_v2_failure_handoff if v2_generation_errored?
+    record_v2_failure_handoff(source: Captain::ConversationEvents::Sources::GENERATION_FAILURE) if v2_generation_errored?
   end
 
   def process_standard_response
@@ -112,11 +112,8 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     if captain_v2_enabled?
       return unless v2_handoff_tool_completed? || conversation_pending?
 
-      # Known gap, accepted for now: the fallback V1 handoff (tool fired but never
-      # completed) emits no captain.conversation.handed_off event — the tool emits
-      # only after a successful bot_handoff!. If outcome data ever needs it, emit
-      # here with a distinct source such as 'tool_fallback'.
       v2_handoff_tool_completed? ? process_v2_handoff : process_v1_handoff
+      record_v2_failure_handoff(source: Captain::ConversationEvents::Sources::TOOL) unless v2_handoff_tool_completed?
     else
       conversation_pending? ? process_v1_handoff : process_v2_handoff
     end
@@ -136,10 +133,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     @response['response'] == 'conversation_handoff'
   end
 
-  def v2_handoff_tool_fired?
-    @response['handoff_tool_called']
-  end
-
+  def v2_handoff_tool_fired? = @response['handoff_tool_called']
   def v2_handoff_tool_completed? = @v2_handoff_tool_completed == true
 
   def process_v1_handoff
@@ -201,7 +195,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     return if captain_v2_enabled? && newer_customer_message_arrived?
 
     process_v1_handoff
-    record_v2_failure_handoff if captain_v2_enabled?
+    record_v2_failure_handoff(source: Captain::ConversationEvents::Sources::GENERATION_FAILURE) if captain_v2_enabled?
   end
 
   def log_error(error)
diff --git a/enterprise/app/jobs/captain/conversation/v2_lifecycle_events.rb b/enterprise/app/jobs/captain/conversation/v2_lifecycle_events.rb
index 37501e9ff..dc1738946 100644
--- a/enterprise/app/jobs/captain/conversation/v2_lifecycle_events.rb
+++ b/enterprise/app/jobs/captain/conversation/v2_lifecycle_events.rb
@@ -18,11 +18,11 @@ module Captain::Conversation::V2LifecycleEvents
     Captain::ConversationEvents.response_failed(conversation: @conversation, assistant: @assistant, reason: reason, at: Time.current)
   end
 
-  def record_v2_failure_handoff
+  def record_v2_failure_handoff(source:)
     Captain::ConversationEvents.handed_off(
       conversation: @conversation,
       assistant: @assistant,
-      source: Captain::ConversationEvents::Sources::GENERATION_FAILURE,
+      source: source,
       reason_category: :tool_failure,
       at: Time.current
     )
diff --git a/enterprise/app/listeners/captain/conversation_outcome_event_listener.rb b/enterprise/app/listeners/captain/conversation_outcome_event_listener.rb
new file mode 100644
index 000000000..fa5848920
--- /dev/null
+++ b/enterprise/app/listeners/captain/conversation_outcome_event_listener.rb
@@ -0,0 +1,25 @@
+class Captain::ConversationOutcomeEventListener < BaseListener
+  def message_created(event)
+    message = event.data[:message]
+    Captain::ConversationOutcomeTracker.new(conversation: message.conversation).record_human_reply(message: message)
+  end
+
+  def captain_conversation_handed_off(event)
+    tracker(event).record_handoff(at: event.timestamp, reason_category: event.data[:reason_category])
+  end
+
+  def conversation_updated(event)
+    return unless event.data[:changed_attributes]&.dig('status', 0) == 'resolved'
+
+    tracker(event).record_reopen(at: event.timestamp)
+  end
+
+  private
+
+  def tracker(event)
+    Captain::ConversationOutcomeTracker.new(
+      conversation: event.data[:conversation],
+      assistant: event.data[:assistant]
+    )
+  end
+end
diff --git a/enterprise/app/listeners/captain_listener.rb b/enterprise/app/listeners/captain_listener.rb
index cd6d8fd1f..01b2605d3 100644
--- a/enterprise/app/listeners/captain_listener.rb
+++ b/enterprise/app/listeners/captain_listener.rb
@@ -1,8 +1,20 @@
 class CaptainListener < BaseListener
   include ::Events::Types
 
+  def message_updated(event)
+    message = event.data[:message]
+    return unless message.input_csat?
+
+    response = CsatSurveyResponse.find_by(message: message)
+    return unless response
+
+    tracker(message.conversation).record_csat(response: response)
+  end
+
   def conversation_resolved(event)
     conversation = extract_conversation_and_account(event)[0]
+    tracker(conversation).record_resolution(at: event.timestamp)
+
     assistant = conversation.inbox.captain_assistant
 
     return unless conversation.inbox.captain_active?
@@ -10,4 +22,10 @@ class CaptainListener < BaseListener
     Captain::Llm::ContactNotesService.new(assistant, conversation).generate_and_update_notes if assistant.config['feature_memory'].present?
     Captain::Llm::ConversationFaqJob.perform_later(conversation, assistant) if assistant.config['feature_faq'].present?
   end
+
+  private
+
+  def tracker(conversation)
+    Captain::ConversationOutcomeTracker.new(conversation: conversation)
+  end
 end
diff --git a/enterprise/app/services/captain/conversation/response_scheduler_service.rb b/enterprise/app/services/captain/conversation/response_scheduler_service.rb
index 0e8f6c508..692e10501 100644
--- a/enterprise/app/services/captain/conversation/response_scheduler_service.rb
+++ b/enterprise/app/services/captain/conversation/response_scheduler_service.rb
@@ -8,8 +8,6 @@ class Captain::Conversation::ResponseSchedulerService
   end
 
   def perform
-    track_captain_engagement
-
     wait_time = attachment_wait_time
     return Captain::Conversation::ResponseBuilderJob.perform_later(*job_args) if wait_time.zero?
 
@@ -24,16 +22,6 @@ class Captain::Conversation::ResponseSchedulerService
     args
   end
 
-  def track_captain_engagement
-    return unless captain_v2_enabled?
-
-    Captain::ConversationEvents.engaged(
-      conversation: @conversation,
-      assistant: @assistant,
-      at: @message.created_at
-    )
-  end
-
   def captain_v2_enabled?
     @conversation.account.feature_enabled?('captain_integration_v2')
   end
diff --git a/enterprise/app/services/captain/conversation_events.rb b/enterprise/app/services/captain/conversation_events.rb
index 964b2d365..9ad8c3793 100644
--- a/enterprise/app/services/captain/conversation_events.rb
+++ b/enterprise/app/services/captain/conversation_events.rb
@@ -8,16 +8,7 @@ class Captain::ConversationEvents
   end
 
   class << self
-    def engaged(conversation:, assistant:, at:)
-      dispatch(
-        Events::Types::CAPTAIN_CONVERSATION_ENGAGED,
-        at: at,
-        conversation: conversation,
-        assistant: assistant
-      )
-    end
-
-    def handed_off(conversation:, assistant:, source:, at:, reason_category: nil)
+    def handed_off(conversation:, assistant:, source:, reason_category:, at:)
       dispatch(
         Events::Types::CAPTAIN_CONVERSATION_HANDED_OFF,
         at: at,
diff --git a/enterprise/app/services/captain/conversation_outcome_tracker.rb b/enterprise/app/services/captain/conversation_outcome_tracker.rb
new file mode 100644
index 000000000..3cae8201d
--- /dev/null
+++ b/enterprise/app/services/captain/conversation_outcome_tracker.rb
@@ -0,0 +1,129 @@
+class Captain::ConversationOutcomeTracker
+  pattr_initialize [:conversation!, :assistant]
+
+  def record_eligibility(at:)
+    safely_track(:eligibility, at: at) do
+      episodes.trigger_initial.take || create_initial_episode(at)
+    end
+  end
+
+  def record_reopen(at:)
+    safely_track(:reopen, at: at) do
+      next if episodes.none?
+
+      ApplicationRecord.transaction do
+        predecessor = episodes.find_by!(ended_at: nil)
+        predecessor.update!(ended_at: at)
+        ConversationOutcome.create!(
+          account: account,
+          assistant: conversation.inbox.captain_assistant || predecessor.assistant,
+          conversation: conversation,
+          inbox: conversation.inbox,
+          episode_trigger: 'reopen',
+          started_at: at
+        )
+      end
+    end
+  end
+
+  def record_handoff(at:, reason_category:)
+    snapshot_episode(:handoff, at: at) do |episode|
+      episode.handoff_at = at
+      episode.handoff_reason_category = reason_category
+    end
+  end
+
+  def record_resolution(at:)
+    snapshot_episode(:resolution, at: at) do |episode|
+      episode.resolved_at = at
+    end
+  end
+
+  def record_human_reply(message:)
+    return unless public_human_reply?(message)
+
+    safely_track(:human_reply, at: message.created_at) do
+      episode = attributed_episode(message.created_at)
+      next unless episode
+      next episode if episode.first_human_reply_at.present?
+
+      episode.update!(first_human_reply_at: message.created_at)
+      episode
+    end
+  end
+
+  def record_csat(response:)
+    safely_track(:csat, at: response.message.created_at) do
+      episode = attributed_episode(response.message.created_at)
+      next unless episode
+
+      episode.update!(csat_rating: response.rating, csat_received_at: response.created_at)
+      episode
+    end
+  end
+
+  private
+
+  def account
+    conversation.account
+  end
+
+  def episodes
+    ConversationOutcome.where(account_id: conversation.account_id, conversation_id: conversation.id)
+  end
+
+  def create_initial_episode(at)
+    ConversationOutcome.create!(
+      account: account,
+      assistant: assistant,
+      conversation: conversation,
+      inbox: conversation.inbox,
+      episode_trigger: 'initial',
+      started_at: at
+    )
+  end
+
+  def snapshot_episode(action, at:)
+    safely_track(action, at: at) do
+      episode = attributed_episode(at)
+      next unless episode
+
+      yield episode
+      snapshot_message_facts(episode, through: at)
+      episode.save! if episode.changed?
+      episode
+    end
+  end
+
+  def attributed_episode(at)
+    episodes.covering(at).chronological.last
+  end
+
+  def snapshot_message_facts(episode, through:)
+    messages = conversation.messages.where(created_at: episode.started_at..through)
+    replies = messages.where(sender_type: 'Captain::Assistant', message_type: :outgoing, private: false)
+
+    episode.captain_reply_count = replies.count
+    episode.first_captain_reply_at = replies.minimum(:created_at)
+    episode.last_captain_reply_at = replies.maximum(:created_at)
+  end
+
+  def public_human_reply?(message)
+    return false unless message.outgoing? && !message.private?
+    return false if message.content_attributes['automation_rule_id'].present?
+    return false if message.additional_attributes['campaign_id'].present?
+
+    message.sender.is_a?(User) || message.content_attributes['external_echo'].present?
+  end
+
+  def safely_track(action, at:)
+    yield
+  rescue StandardError => e
+    ChatwootExceptionTracker.new(e, account: account).capture_exception
+    Rails.logger.error(
+      "[CAPTAIN][ConversationOutcomeTracker] Failed to record #{action} for conversation=#{conversation.display_id} " \
+      "at=#{at.iso8601(6)}: #{e.message}"
+    )
+    nil
+  end
+end
diff --git a/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb b/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb
index 22997b0ca..6f289339c 100644
--- a/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb
+++ b/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb
@@ -1,7 +1,15 @@
 module Enterprise::MessageTemplates::HookExecutionService
   def trigger_templates
     super
-    return unless should_process_captain_response?
+    return unless captain_conversation_message?
+
+    # Eligibility is demand-level: every inbound customer message in a
+    # Captain-connected inbox counts, including conversations a human grabbed
+    # first or that arrive while the account is over its usage limit —
+    # otherwise the coverage denominator only ever contains conversations
+    # Captain was already about to answer.
+    track_captain_eligibility
+    return unless conversation.pending?
     return perform_handoff unless inbox.captain_active?
 
     Captain::Conversation::ResponseSchedulerService.new(message: message).perform
@@ -27,19 +35,20 @@ module Enterprise::MessageTemplates::HookExecutionService
 
   private
 
-  def captain_v2_enabled?
-    conversation.account.feature_enabled?('captain_integration_v2')
+  def track_captain_eligibility
+    return unless conversation.account.feature_enabled?('captain_integration_v2')
+
+    Captain::ConversationOutcomeTracker.new(
+      conversation: conversation,
+      assistant: inbox.captain_assistant
+    ).record_eligibility(at: message.created_at)
   end
 
-  def should_process_captain_response?
-    # Audience and schedule are decided when Captain first takes or reopens a conversation.
-    # Do not re-evaluate an existing pending conversation for each new message.
-    conversation.pending? && message.captain_response_triggering? && captain_assistant_configured? && !inbox.external_bot_active?
+  def captain_conversation_message?
+    message.captain_response_triggering? && captain_assistant_configured? && !inbox.external_bot_active?
   end
 
   def perform_handoff
-    return unless conversation.pending?
-
     Rails.logger.info("Captain limit exceeded, performing handoff mid-conversation for conversation: #{conversation.id}")
     conversation.messages.create!(
       message_type: :outgoing,
@@ -48,15 +57,13 @@ module Enterprise::MessageTemplates::HookExecutionService
       content: 'Transferring to another agent for further assistance.'
     )
     conversation.bot_handoff!
-    if captain_v2_enabled?
-      Captain::ConversationEvents.handed_off(
-        conversation: conversation,
-        assistant: inbox.captain_assistant,
-        source: Captain::ConversationEvents::Sources::USAGE_LIMIT,
-        reason_category: :usage_limit,
-        at: Time.current
-      )
-    end
+    Captain::ConversationEvents.handed_off(
+      conversation: conversation,
+      assistant: inbox.captain_assistant,
+      source: Captain::ConversationEvents::Sources::USAGE_LIMIT,
+      reason_category: :usage_limit,
+      at: Time.current
+    )
     send_out_of_office_message_after_handoff
   end
 
diff --git a/enterprise/lib/captain/tools/handoff_tool.rb b/enterprise/lib/captain/tools/handoff_tool.rb
index e1d8a11c8..deb8b0ccf 100644
--- a/enterprise/lib/captain/tools/handoff_tool.rb
+++ b/enterprise/lib/captain/tools/handoff_tool.rb
@@ -1,8 +1,24 @@
 class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
-  description 'Hand off the conversation to a human agent when unable to assist further'
-  param :reason, type: 'string', desc: 'The reason why handoff is needed (optional)', required: false
+  # LLM-selectable reasons are a subset of the outcome enum. System lifecycle
+  # paths emit the remaining categories, such as usage limits and pending clarification.
+  REASON_CATEGORIES = %w[customer_request missing_knowledge unsupported_request policy_restriction tool_failure].freeze
 
-  def perform(tool_context, reason: nil)
+  description 'Hand off the conversation to a human agent when unable to assist further'
+  params do
+    string :reason, description: 'The reason why handoff is needed (optional)', required: false
+    string :reason_category, enum: REASON_CATEGORIES, description: 'Reporting category for why the handoff is needed'
+  end
+
+  # Agents::ToolWrapper reads `tool.class.params`, while ruby_llm treats a
+  # no-argument call as a schema reset. Keep the compatibility fix local to the
+  # only tool that uses ruby_llm's block schema DSL.
+  def self.params(schema = nil, &)
+    return params_schema_definition if schema.nil? && !block_given?
+
+    super
+  end
+
+  def perform(tool_context, reason: nil, reason_category: nil)
     conversation = find_conversation(tool_context.state)
     return 'Conversation not found' unless conversation
 
@@ -13,7 +29,7 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
                    })
 
     # Use existing handoff mechanism from ResponseBuilderJob
-    handoff_result = trigger_handoff(tool_context, conversation, reason)
+    handoff_result = trigger_handoff(tool_context, conversation, reason, reason_category)
     return 'Handoff skipped because a newer customer message arrived' if handoff_result == :stale
     return 'Handoff skipped because the conversation changed' unless handoff_result == :completed
 
@@ -25,8 +41,8 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
 
   private
 
-  def trigger_handoff(tool_context, conversation, reason)
-    return trigger_legacy_handoff(tool_context, conversation, reason) unless captain_v2_enabled?
+  def trigger_handoff(tool_context, conversation, reason, reason_category)
+    return trigger_legacy_handoff(tool_context, conversation, reason, reason_category) unless captain_v2_enabled?
 
     note = nil
     handoff_result = conversation.with_lock do
@@ -54,28 +70,38 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
     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
-    emit_tool_handoff_event(conversation)
+    emit_tool_handoff_event(conversation, reason_category)
 
     # 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)
+  def trigger_legacy_handoff(tool_context, conversation, reason, reason_category)
     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!
-    emit_tool_handoff_event(conversation)
+    emit_tool_handoff_event(conversation, reason_category)
     send_out_of_office_message_if_applicable(conversation)
     :completed
   end
 
-  def emit_tool_handoff_event(conversation)
+  def emit_tool_handoff_event(conversation, reason_category)
     Captain::ConversationEvents.handed_off(conversation: conversation, assistant: @assistant,
-                                           source: Captain::ConversationEvents::Sources::TOOL, at: Time.current)
+                                           source: Captain::ConversationEvents::Sources::TOOL,
+                                           reason_category: normalize_reason_category(reason_category),
+                                           at: Time.current)
+  end
+
+  # Tool execution does not enforce the schema enum, and an unknown category
+  # would fail the outcome's validated enum after the handoff already happened.
+  # Record those handoffs as unclassified instead.
+  def normalize_reason_category(reason_category)
+    category = reason_category.to_s
+    category if REASON_CATEGORIES.include?(category)
   end
 
   def record_handoff_note(tool_context, note)
diff --git a/lib/events/types.rb b/lib/events/types.rb
index 653739826..83f3d0d3c 100644
--- a/lib/events/types.rb
+++ b/lib/events/types.rb
@@ -22,7 +22,6 @@ module Events::Types
   # FIXME: deprecate the opened and resolved events in future in favor of status changed event.
   CONVERSATION_OPENED = 'conversation.opened'
   CONVERSATION_RESOLVED = 'conversation.resolved'
-  CAPTAIN_CONVERSATION_ENGAGED = 'captain.conversation.engaged'
   CAPTAIN_CONVERSATION_HANDED_OFF = 'captain.conversation.handed_off'
   CAPTAIN_CONVERSATION_RESOLVED = 'captain.conversation.resolved'
   CAPTAIN_RESPONSE_COMPLETED = 'captain.response.completed'
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 1d7fe151f..0033e3619 100644
--- a/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb
+++ b/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb
@@ -817,6 +817,8 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
                                                                                      'response' => 'I tried to hand off',
                                                                                      'handoff_tool_called' => true
                                                                                    })
+        expect(Captain::ConversationEvents).to receive(:handed_off)
+          .with(conversation: conversation, assistant: assistant, source: 'tool', reason_category: :tool_failure, at: kind_of(Time))
 
         described_class.perform_now(conversation, assistant)
 
diff --git a/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb b/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb
index d928e7d8f..51f7ffdda 100644
--- a/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb
+++ b/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb
@@ -16,13 +16,23 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
     end
   end
 
-  describe '#parameters' do
-    it 'returns the correct parameters' do
-      expect(tool.parameters).to have_key(:reason)
-      expect(tool.parameters[:reason].name).to eq(:reason)
-      expect(tool.parameters[:reason].type).to eq('string')
-      expect(tool.parameters[:reason].description).to eq('The reason why handoff is needed (optional)')
-      expect(tool.parameters[:reason].required).to be false
+  describe '#params_schema' do
+    it 'constrains the reason category to the supported values and keeps the reason optional' do
+      schema = tool.params_schema
+
+      expect(schema['properties']['reason']['type']).to eq('string')
+      expect(schema['properties']['reason_category']['enum']).to eq(described_class::REASON_CATEGORIES)
+      expect(schema['required']).to eq(['reason_category'])
+    end
+
+    it 'uses only reason categories supported by conversation outcomes' do
+      expect(ConversationOutcome::HANDOFF_REASON_CATEGORIES).to include(*described_class::REASON_CATEGORIES)
+    end
+
+    it 'survives Agents::ToolWrapper reading the class params at wrap time' do
+      described_class.params
+
+      expect(described_class.new(assistant).params_schema['properties'].keys).to contain_exactly('reason', 'reason_category')
     end
   end
 
@@ -96,9 +106,16 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
 
         it 'emits a captain handoff event with the tool source after the locked handoff completes' do
           expect(Captain::ConversationEvents).to receive(:handed_off)
-            .with(conversation: conversation, assistant: assistant, source: 'tool', at: kind_of(Time))
+            .with(conversation: conversation, assistant: assistant, source: 'tool', reason_category: 'customer_request', at: kind_of(Time))
 
-          tool.perform(tool_context, reason: 'Customer needs specialized support')
+          tool.perform(tool_context, reason: 'Customer needs specialized support', reason_category: 'customer_request')
+        end
+
+        it 'emits an unclassified handoff when the model supplies an unknown reason category' do
+          expect(Captain::ConversationEvents).to receive(:handed_off)
+            .with(conversation: conversation, assistant: assistant, source: 'tool', reason_category: nil, at: kind_of(Time))
+
+          tool.perform(tool_context, reason: 'Customer needs specialized support', reason_category: 'hallucinated_category')
         end
 
         it 'does not emit a captain handoff event when the handoff is skipped as stale' do
@@ -170,11 +187,33 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
           tool.perform(tool_context, reason: 'Test reason')
         end
 
-        it 'emits a captain handoff event with the tool source' do
+        it 'emits a captain handoff event with the tool source and reason category' do
           expect(Captain::ConversationEvents).to receive(:handed_off)
-            .with(conversation: conversation, assistant: assistant, source: 'tool', at: kind_of(Time))
+            .with(conversation: conversation, assistant: assistant, source: 'tool', reason_category: 'unsupported_request', at: kind_of(Time))
 
-          tool.perform(tool_context, reason: 'Test reason')
+          tool.perform(tool_context, reason: 'Test reason', reason_category: 'unsupported_request')
+        end
+
+        it 'records the handoff on an existing V2 outcome' do
+          account.enable_features!('captain_integration_v2')
+          create(
+            :conversation_outcome,
+            account: account,
+            assistant: assistant,
+            conversation: conversation,
+            inbox: inbox
+          )
+
+          tool.perform(
+            tool_context,
+            reason: 'Customer needs specialized support',
+            reason_category: 'unsupported_request'
+          )
+
+          expect(ConversationOutcome.last).to have_attributes(
+            handoff_reason_category: 'unsupported_request',
+            handoff_at: be_present
+          )
         end
 
         it 'creates a conversation_bot_handoff reporting event' do
diff --git a/spec/enterprise/listeners/captain_listener_spec.rb b/spec/enterprise/listeners/captain_listener_spec.rb
index 4e25c40c4..060fce2ec 100644
--- a/spec/enterprise/listeners/captain_listener_spec.rb
+++ b/spec/enterprise/listeners/captain_listener_spec.rb
@@ -11,7 +11,7 @@ describe CaptainListener do
     let(:conversation) { create(:conversation, account: account, inbox: inbox, assignee: user) }
 
     let(:event_name) { :conversation_resolved }
-    let(:event) { Events::Base.new(event_name, Time.zone.now, conversation: conversation) }
+    let(:event) { Events::Base.new(event_name, Time.zone.now.change(usec: 0), conversation: conversation) }
 
     before do
       create(:captain_inbox, captain_assistant: assistant, inbox: inbox)
@@ -49,5 +49,74 @@ describe CaptainListener do
         listener.conversation_resolved(event)
       end
     end
+
+    it 'records the resolution on an existing V2 outcome' do
+      assistant.update!(config: {})
+      outcome = create(
+        :conversation_outcome,
+        account: account,
+        assistant: assistant,
+        conversation: conversation,
+        inbox: inbox,
+        started_at: 10.minutes.ago
+      )
+      captain_reply = create(
+        :message,
+        account: account,
+        inbox: inbox,
+        conversation: conversation,
+        sender: assistant,
+        message_type: :outgoing,
+        created_at: 5.minutes.ago.change(usec: 0)
+      )
+
+      listener.conversation_resolved(event)
+
+      expect(outcome.reload).to have_attributes(
+        captain_reply_count: 1,
+        first_captain_reply_at: captain_reply.created_at,
+        resolved_at: event.timestamp
+      )
+    end
+  end
+
+  describe '#message_updated' do
+    let(:conversation) { create(:conversation, account: account, inbox: inbox) }
+    let!(:outcome) do
+      create(
+        :conversation_outcome,
+        account: account,
+        assistant: assistant,
+        conversation: conversation,
+        inbox: inbox
+      )
+    end
+
+    it 'records a submitted CSAT response' do
+      message = create(
+        :message,
+        account: account,
+        inbox: inbox,
+        conversation: conversation,
+        content_type: :input_csat,
+        message_type: :outgoing
+      )
+      response = create(
+        :csat_survey_response,
+        account: account,
+        conversation: conversation,
+        contact: conversation.contact,
+        message: message,
+        rating: 5
+      )
+      event = Events::Base.new(:message_updated, Time.current, message: message)
+
+      listener.message_updated(event)
+
+      expect(outcome.reload).to have_attributes(
+        csat_rating: 5,
+        csat_received_at: response.created_at
+      )
+    end
   end
 end
diff --git a/spec/enterprise/services/captain/conversation_events_spec.rb b/spec/enterprise/services/captain/conversation_events_spec.rb
index b6f22d6cb..5ab854578 100644
--- a/spec/enterprise/services/captain/conversation_events_spec.rb
+++ b/spec/enterprise/services/captain/conversation_events_spec.rb
@@ -6,18 +6,6 @@ RSpec.describe Captain::ConversationEvents do
   let(:assistant) { create(:captain_assistant, account: account) }
   let(:timestamp) { Time.zone.now }
 
-  describe '.engaged' do
-    it 'dispatches the engagement event with normalized context' do
-      expect(Rails.configuration.dispatcher).to receive(:dispatch).with(
-        Events::Types::CAPTAIN_CONVERSATION_ENGAGED,
-        timestamp,
-        { conversation: conversation, assistant: assistant }
-      )
-
-      described_class.engaged(conversation: conversation, assistant: assistant, at: timestamp)
-    end
-  end
-
   describe '.handed_off' do
     it 'dispatches the handoff event with source and reason category' do
       expect(Rails.configuration.dispatcher).to receive(:dispatch).with(
@@ -69,7 +57,9 @@ RSpec.describe Captain::ConversationEvents do
         .and_return(instance_double(ChatwootExceptionTracker, capture_exception: true))
 
       expect do
-        described_class.engaged(conversation: conversation, assistant: assistant, at: timestamp)
+        described_class.handed_off(
+          conversation: conversation, assistant: assistant, source: 'tool', reason_category: nil, at: timestamp
+        )
       end.not_to raise_error
     end
   end
diff --git a/spec/enterprise/services/captain/conversation_outcome_tracker_spec.rb b/spec/enterprise/services/captain/conversation_outcome_tracker_spec.rb
new file mode 100644
index 000000000..9112283ce
--- /dev/null
+++ b/spec/enterprise/services/captain/conversation_outcome_tracker_spec.rb
@@ -0,0 +1,195 @@
+require 'rails_helper'
+
+RSpec.describe Captain::ConversationOutcomeTracker do
+  let(:account) { create(:account) }
+  let(:assistant) { create(:captain_assistant, account: account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:conversation) { create(:conversation, account: account, inbox: inbox, status: :pending) }
+  let(:tracker) { described_class.new(conversation: conversation, assistant: assistant) }
+
+  def episodes
+    ConversationOutcome.where(conversation_id: conversation.id).order(:started_at)
+  end
+
+  before do
+    account.enable_features!('captain_integration_v2')
+    create(:captain_inbox, captain_assistant: assistant, inbox: inbox)
+  end
+
+  describe '#record_eligibility' do
+    it 'creates the initial episode anchored to the demand time' do
+      eligible_at = 2.minutes.ago.change(usec: 0)
+
+      episode = tracker.record_eligibility(at: eligible_at)
+
+      expect(episode.reload).to have_attributes(
+        account: account,
+        assistant: assistant,
+        conversation: conversation,
+        inbox: inbox,
+        episode_trigger: 'initial',
+        started_at: eligible_at,
+        ended_at: nil
+      )
+    end
+
+    it 'is idempotent' do
+      tracker.record_eligibility(at: Time.current)
+
+      expect do
+        tracker.record_eligibility(at: Time.current)
+      end.not_to change(ConversationOutcome, :count)
+    end
+  end
+
+  describe '#record_reopen' do
+    let(:initial_at) { 30.minutes.ago.change(usec: 0) }
+    let!(:initial) { tracker.record_eligibility(at: initial_at) }
+
+    it 'closes the open episode and opens a reopen episode' do
+      boundary_at = 5.minutes.ago.change(usec: 0)
+
+      tracker.record_reopen(at: boundary_at)
+
+      expect(initial.reload.ended_at).to eq(boundary_at)
+      expect(episodes.last).to have_attributes(
+        episode_trigger: 'reopen',
+        assistant: assistant,
+        started_at: boundary_at,
+        ended_at: nil
+      )
+    end
+
+    it 'creates exactly one episode per boundary regardless of history' do
+      tracker.record_reopen(at: 20.minutes.ago)
+      tracker.record_reopen(at: 10.minutes.ago)
+
+      expect do
+        tracker.record_reopen(at: 1.minute.ago)
+      end.to change(ConversationOutcome, :count).by(1)
+    end
+
+    it 'ignores boundaries for conversations without Captain history' do
+      ConversationOutcome.delete_all
+
+      expect do
+        tracker.record_reopen(at: Time.current)
+      end.not_to change(ConversationOutcome, :count)
+    end
+
+    it 'captures boundary failures without raising' do
+      error = ActiveRecord::StatementInvalid.new('database unavailable')
+      exception_tracker = instance_double(ChatwootExceptionTracker)
+      allow(ConversationOutcome).to receive(:create!).and_raise(error)
+      expect(ChatwootExceptionTracker).to receive(:new).with(error, account: account).and_return(exception_tracker)
+      expect(exception_tracker).to receive(:capture_exception)
+
+      expect do
+        tracker.record_reopen(at: Time.current)
+      end.not_to raise_error
+    end
+  end
+
+  describe '#record_handoff' do
+    let!(:initial) { tracker.record_eligibility(at: 30.minutes.ago) }
+
+    it 'snapshots message facts and records the handoff reason' do
+      first_reply = create(
+        :message,
+        account: account, inbox: inbox, conversation: conversation,
+        sender: assistant, message_type: :outgoing, created_at: 20.minutes.ago.change(usec: 0)
+      )
+      last_reply = create(
+        :message,
+        account: account, inbox: inbox, conversation: conversation,
+        sender: assistant, message_type: :outgoing, created_at: 10.minutes.ago.change(usec: 0)
+      )
+      handoff_at = 5.minutes.ago.change(usec: 0)
+      tracker.record_handoff(at: handoff_at, reason_category: 'missing_knowledge')
+
+      expect(initial.reload).to have_attributes(
+        captain_reply_count: 2,
+        first_captain_reply_at: first_reply.created_at,
+        last_captain_reply_at: last_reply.created_at,
+        handoff_at: handoff_at,
+        handoff_reason_category: 'missing_knowledge'
+      )
+    end
+  end
+
+  describe '#record_resolution' do
+    let!(:initial) { tracker.record_eligibility(at: 30.minutes.ago) }
+
+    it 'snapshots the episode active at the resolution event time' do
+      captain_reply = create(
+        :message,
+        account: account, inbox: inbox, conversation: conversation,
+        sender: assistant, message_type: :outgoing, created_at: 20.minutes.ago.change(usec: 0)
+      )
+      resolved_at = 15.minutes.ago.change(usec: 0)
+      tracker.record_reopen(at: 10.minutes.ago.change(usec: 0))
+
+      tracker.record_resolution(at: resolved_at)
+
+      expect(initial.reload).to have_attributes(
+        captain_reply_count: 1,
+        first_captain_reply_at: captain_reply.created_at,
+        resolved_at: resolved_at
+      )
+      expect(episodes.last.resolved_at).to be_nil
+    end
+
+    it 'snapshots Captain and human replies through resolution' do
+      captain_reply = create(
+        :message,
+        account: account, inbox: inbox, conversation: conversation,
+        sender: assistant, message_type: :outgoing, created_at: 20.minutes.ago.change(usec: 0)
+      )
+      agent = create(:user, account: account)
+      create(
+        :message,
+        account: account, inbox: inbox, conversation: conversation,
+        sender: agent, message_type: :outgoing, created_at: 15.minutes.ago,
+        content_attributes: { automation_rule_id: 1 }
+      )
+      human_reply = create(
+        :message,
+        account: account, inbox: inbox, conversation: conversation,
+        sender: agent, message_type: :outgoing, created_at: 10.minutes.ago.change(usec: 0)
+      )
+      resolved_at = 5.minutes.ago.change(usec: 0)
+
+      tracker.record_resolution(at: resolved_at)
+
+      expect(initial.reload).to have_attributes(
+        captain_reply_count: 1,
+        first_captain_reply_at: captain_reply.created_at,
+        last_captain_reply_at: captain_reply.created_at,
+        first_human_reply_at: human_reply.created_at,
+        resolved_at: resolved_at
+      )
+    end
+  end
+
+  describe '#record_csat' do
+    let!(:initial) { tracker.record_eligibility(at: 30.minutes.ago) }
+
+    it 'attributes the response to the episode that issued its survey' do
+      survey_message = create(
+        :message,
+        account: account, inbox: inbox, conversation: conversation,
+        content_type: :input_csat, created_at: 20.minutes.ago
+      )
+      tracker.record_reopen(at: 10.minutes.ago)
+      response = create(
+        :csat_survey_response,
+        account: account, conversation: conversation, message: survey_message, rating: 4
+      )
+
+      tracker.record_csat(response: response)
+
+      expect(initial.reload).to have_attributes(csat_rating: 4, csat_received_at: response.created_at)
+      expect(episodes.last.csat_rating).to be_nil
+    end
+  end
+end
diff --git a/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb b/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb
index f4974c1b5..118cbdeda 100644
--- a/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb
+++ b/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb
@@ -137,19 +137,23 @@ RSpec.describe MessageTemplates::HookExecutionService do
         create(:message, conversation: conversation, message_type: :incoming, account: account)
       end
 
-      it 'emits the engagement event when captain V2 is enabled' do
+      it 'records a conversation outcome when captain V2 is enabled' do
         account.enable_features!('captain_integration_v2')
 
-        expect(Captain::ConversationEvents).to receive(:engaged)
-          .with(conversation: conversation, assistant: assistant, at: kind_of(Time))
+        expect do
+          create(:message, conversation: conversation, message_type: :incoming, account: account)
+        end.to change(ConversationOutcome, :count).by(1)
 
-        create(:message, conversation: conversation, message_type: :incoming, account: account)
+        expect(ConversationOutcome.last).to have_attributes(
+          assistant: assistant,
+          conversation: conversation
+        )
       end
 
-      it 'does not emit the engagement event when captain V2 is disabled' do
-        expect(Captain::ConversationEvents).not_to receive(:engaged)
-
-        create(:message, conversation: conversation, message_type: :incoming, account: account)
+      it 'does not record a conversation outcome when captain V2 is disabled' do
+        expect do
+          create(:message, conversation: conversation, message_type: :incoming, account: account)
+        end.not_to change(ConversationOutcome, :count)
       end
     end
 
@@ -173,19 +177,30 @@ RSpec.describe MessageTemplates::HookExecutionService do
         expect(conversation.reload.status).to eq('open')
       end
 
-      it 'emits a usage limit handoff event when captain V2 is enabled' do
-        account.enable_features!('captain_integration_v2')
-
+      it 'emits a usage limit handoff event' do
         expect(Captain::ConversationEvents).to receive(:handed_off)
           .with(conversation: conversation, assistant: assistant, source: 'usage_limit', reason_category: :usage_limit, at: kind_of(Time))
 
         create(:message, conversation: conversation, message_type: :incoming, account: account)
       end
 
-      it 'does not emit a handoff event when captain V2 is disabled' do
-        expect(Captain::ConversationEvents).not_to receive(:handed_off)
+      it 'records the handoff on the outcome when captain V2 is enabled' do
+        account.enable_features!('captain_integration_v2')
 
-        create(:message, conversation: conversation, message_type: :incoming, account: account)
+        expect do
+          create(:message, conversation: conversation, message_type: :incoming, account: account)
+        end.to change(ConversationOutcome, :count).by(1)
+
+        expect(ConversationOutcome.last).to have_attributes(
+          handoff_reason_category: 'usage_limit',
+          handoff_at: be_present
+        )
+      end
+
+      it 'does not record an outcome when captain V2 is disabled' do
+        expect do
+          create(:message, conversation: conversation, message_type: :incoming, account: account)
+        end.not_to change(ConversationOutcome, :count)
       end
     end
   end
@@ -222,6 +237,14 @@ RSpec.describe MessageTemplates::HookExecutionService do
 
       create(:message, conversation: conversation, message_type: :incoming, account: account)
     end
+
+    it 'still records the conversation as eligible demand when captain V2 is enabled' do
+      account.enable_features!('captain_integration_v2')
+
+      expect do
+        create(:message, conversation: conversation, message_type: :incoming, account: account)
+      end.to change(ConversationOutcome, :count).by(1)
+    end
   end
 
   context 'when the contact is inside the assistant audience' do