fix(media-server): only record the overlapping call window

For a reported 6-second call the agent OGG was 11.82s and the combined
OGG was 11.77s. Root cause: the browser's mic starts sending RTP to the
media server as soon as Peer B completes WebRTC — ~5s before the
contact actually picks up on an outbound call. Those pre-answer seconds
were captured into agent.ogg, and ffmpeg's longest-duration mix
inherited the inflated length.

Recorder now drops RTP on each side until the other side has produced
at least one packet. Writes start only when both tracks are live, which
aligns the recording with the real conversation window and matches
Meta's authoritative duration.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.env.example

@@ -277,3 +277,10 @@ AZURE_APP_SECRET=
 
 # REDIS_ALFRED_SIZE=10
 # REDIS_VELMA_SIZE=10
+
+# Media Server (WhatsApp Calling - Server-Side WebRTC)
+# Enable server-side WebRTC relay for call persistence across page reloads
+# and server-side recording. Requires the chatwoot-media-server sidecar.
+# MEDIA_SERVER_URL=http://localhost:4000
+# MEDIA_SERVER_AUTH_TOKEN=
+# MEDIA_SERVER_PUBLIC_IP=

Procfile.dev

@@ -2,3 +2,4 @@ backend: bin/rails s -p 3000
 # https://github.com/mperham/sidekiq/issues/3090#issuecomment-389748695
 worker: dotenv bundle exec sidekiq -C config/sidekiq.yml
 vite: bin/vite dev
+media_server: cd enterprise/media-server && AUTH_TOKEN=${MEDIA_SERVER_AUTH_TOKEN:-devtoken} RAILS_CALLBACK_URL=http://localhost:3000 PUBLIC_IP=${MEDIA_SERVER_PUBLIC_IP:-127.0.0.1} RECORDINGS_DIR=/tmp/chatwoot-recordings HTTP_PORT=4000 go run ./cmd/server

app/javascript/dashboard/api/whatsappCalls.js

@@ -0,0 +1,78 @@
+/* global axios */
+import ApiClient from './ApiClient';
+
+class WhatsappCallsAPI extends ApiClient {
+  constructor() {
+    super('whatsapp_calls', { accountScoped: true });
+  }
+
+  show(callId) {
+    return axios.get(`${this.url}/${callId}`);
+  }
+
+  // Accept a ringing call. sdpAnswer is optional — omitted in server-relay mode
+  // where the media server handles WebRTC negotiation.
+  accept(callId, sdpAnswer) {
+    const body = sdpAnswer ? { sdp_answer: sdpAnswer } : {};
+    return axios.post(`${this.url}/${callId}/accept`, body);
+  }
+
+  reject(callId) {
+    return axios.post(`${this.url}/${callId}/reject`);
+  }
+
+  terminate(callId) {
+    return axios.post(`${this.url}/${callId}/terminate`);
+  }
+
+  // Initiate an outbound call. sdpOffer is optional — omitted in server-relay
+  // mode where the media server generates the SDP offer for Meta.
+  initiate(conversationId, sdpOffer) {
+    const body = { conversation_id: conversationId };
+    if (sdpOffer) body.sdp_offer = sdpOffer;
+    return axios.post(`${this.url}/initiate`, body);
+  }
+
+  // Send the agent's SDP answer for the Peer B connection (server-relay mode).
+  agentAnswer(callId, sdpAnswer) {
+    return axios.post(`${this.url}/${callId}/agent_answer`, {
+      sdp_answer: sdpAnswer,
+    });
+  }
+
+  // Get the current agent's active call (if any). Used for reconnection on page load.
+  active() {
+    return axios.get(`${this.url}/active`);
+  }
+
+  // Reconnect to an active call after page reload. Server creates a new Peer B
+  // and returns a fresh SDP offer via ActionCable.
+  reconnect(callId) {
+    return axios.post(`${this.url}/${callId}/reconnect`);
+  }
+
+  // Join an existing call as a supervisor (listen-only by default).
+  join(callId, role = 'listen_only') {
+    return axios.post(`${this.url}/${callId}/join`, { role });
+  }
+
+  // Play an audio file to the caller via the media server.
+  playAudio(callId, { filePath, mode = 'replace', loop = false }) {
+    return axios.post(`${this.url}/${callId}/play_audio`, {
+      file_path: filePath,
+      mode,
+      loop,
+    });
+  }
+
+  // Legacy: upload browser-side recording. Deprecated when media server is enabled.
+  uploadRecording(callId, blob) {
+    const formData = new FormData();
+    formData.append('recording', blob, `call-${callId}.webm`);
+    return axios.post(`${this.url}/${callId}/upload_recording`, formData, {
+      headers: { 'Content-Type': 'multipart/form-data' },
+    });
+  }
+}
+
+export default new WhatsappCallsAPI();

app/javascript/dashboard/components-next/message/bubbles/VoiceCall.vue

@@ -1,7 +1,9 @@
 <script setup>
-import { computed } from 'vue';
+import { computed, ref } from 'vue';
+import { useRouter } from 'vue-router';
 import { useMessageContext } from '../provider.js';
 import { MESSAGE_TYPES, VOICE_CALL_STATUS } from '../constants';
+import { acceptWhatsappCallById } from 'dashboard/composables/useWhatsappCallSession';
 
 import Icon from 'dashboard/components-next/icon/Icon.vue';
 import BaseBubble from 'next/message/bubbles/Base.vue';
@@ -30,8 +32,11 @@ const BG_COLOR_MAP = {
   [VOICE_CALL_STATUS.FAILED]: 'bg-n-ruby-9',
 };
 
+const router = useRouter();
 const { contentAttributes, messageType } = useMessageContext();
 
+// NOTE: contentAttributes.data keys are camelCase because MessageList.vue
+// applies useCamelCase(messages, { deep: true }) before rendering.
 const data = computed(() => contentAttributes.value?.data);
 const status = computed(() => data.value?.status?.toString());
 
@@ -40,6 +45,52 @@ const isFailed = computed(() =>
   [VOICE_CALL_STATUS.NO_ANSWER, VOICE_CALL_STATUS.FAILED].includes(status.value)
 );
 
+// Call source and metadata — all camelCase due to deep transform
+const isWhatsappCall = computed(() => data.value?.callSource === 'whatsapp');
+const callId = computed(() => data.value?.callId);
+const acceptedBy = computed(() => data.value?.acceptedBy);
+const durationSeconds = computed(() => data.value?.durationSeconds);
+const recordingUrl = computed(() => data.value?.recordingUrl);
+const transcript = computed(() => data.value?.transcript);
+const isJoining = ref(false);
+const showTranscript = ref(false);
+
+const formattedDuration = computed(() => {
+  const seconds = durationSeconds.value;
+  if (!seconds || seconds <= 0) return '';
+  const mins = Math.floor(seconds / 60);
+  const secs = seconds % 60;
+  return mins > 0 ? `${mins}m ${secs}s` : `${secs}s`;
+});
+
+// Show join/accept button logic
+// WhatsApp with media server: ringing + in_progress (server-relay supports rejoin)
+// WhatsApp without media server: only ringing (peer-to-peer WebRTC — cannot rejoin after accept)
+// Twilio: ringing + in-progress (conference model supports rejoin)
+const showJoinButton = computed(() => {
+  if (isWhatsappCall.value) {
+    // Server-relay mode enables rejoining in-progress calls
+    const isMediaServerMode = data.value?.mediaServerEnabled;
+    if (isMediaServerMode) {
+      return [
+        VOICE_CALL_STATUS.RINGING,
+        VOICE_CALL_STATUS.IN_PROGRESS,
+      ].includes(status.value);
+    }
+    return status.value === VOICE_CALL_STATUS.RINGING;
+  }
+  return [VOICE_CALL_STATUS.RINGING, VOICE_CALL_STATUS.IN_PROGRESS].includes(
+    status.value
+  );
+});
+
+const joinButtonLabel = computed(() => {
+  if (isWhatsappCall.value && status.value === VOICE_CALL_STATUS.RINGING) {
+    return 'CONVERSATION.VOICE_CALL.ACCEPT_CALL';
+  }
+  return 'CONVERSATION.VOICE_CALL.JOIN_CALL';
+});
+
 const labelKey = computed(() => {
   if (LABEL_MAP[status.value]) return LABEL_MAP[status.value];
   if (status.value === VOICE_CALL_STATUS.RINGING) {
@@ -53,6 +104,15 @@ const labelKey = computed(() => {
 });
 
 const subtextKey = computed(() => {
+  if (
+    acceptedBy.value?.name &&
+    [VOICE_CALL_STATUS.IN_PROGRESS, VOICE_CALL_STATUS.COMPLETED].includes(
+      status.value
+    )
+  ) {
+    return null;
+  }
+
   if (SUBTEXT_MAP[status.value]) return SUBTEXT_MAP[status.value];
   if (status.value === VOICE_CALL_STATUS.IN_PROGRESS) {
     return isOutbound.value
@@ -64,12 +124,39 @@ const subtextKey = computed(() => {
     : 'CONVERSATION.VOICE_CALL.NOT_ANSWERED_YET';
 });
 
+const answeredByText = computed(() => {
+  if (!acceptedBy.value?.name) return '';
+  return acceptedBy.value.name;
+});
+
 const iconName = computed(() => {
   if (ICON_MAP[status.value]) return ICON_MAP[status.value];
   return isOutbound.value ? 'i-ph-phone-outgoing' : 'i-ph-phone-incoming';
 });
 
 const bgColor = computed(() => BG_COLOR_MAP[status.value] || 'bg-n-teal-9');
+
+const handleJoinCall = async () => {
+  if (isJoining.value) return;
+  isJoining.value = true;
+
+  try {
+    if (isWhatsappCall.value) {
+      const result = await acceptWhatsappCallById(callId.value);
+      if (result?.success && result.call) {
+        router.push({
+          name: 'inbox_conversation',
+          params: { conversation_id: result.call.conversationId },
+        });
+      }
+    }
+  } catch (err) {
+    // eslint-disable-next-line no-console
+    console.error('[WhatsApp Call] Accept from bubble failed:', err);
+  } finally {
+    isJoining.value = false;
+  }
+};
 </script>
 
 <template>
@@ -90,14 +177,75 @@ const bgColor = computed(() => BG_COLOR_MAP[status.value] || 'bg-n-teal-9');
           />
         </div>
 
-        <div class="flex overflow-hidden flex-col flex-grow">
+        <div class="flex overflow-hidden flex-col flex-grow gap-0.5">
           <span class="text-sm font-medium truncate text-n-slate-12">
             {{ $t(labelKey) }}
           </span>
-          <span class="text-xs text-n-slate-11">
+          <span v-if="answeredByText" class="text-xs text-n-slate-11">
+            {{
+              $t('CONVERSATION.VOICE_CALL.ANSWERED_BY', {
+                name: answeredByText,
+              })
+            }}
+          </span>
+          <span v-else-if="subtextKey" class="text-xs text-n-slate-11">
             {{ $t(subtextKey) }}
           </span>
+          <span
+            v-if="formattedDuration && status === VOICE_CALL_STATUS.COMPLETED"
+            class="text-xs text-n-slate-10"
+          >
+            {{ formattedDuration }}
+          </span>
         </div>
+
+        <button
+          v-if="showJoinButton"
+          :disabled="isJoining"
+          class="flex items-center gap-1 px-3 py-1.5 text-xs font-medium text-white bg-n-teal-9 hover:bg-n-teal-10 rounded-lg transition-colors shrink-0"
+          :class="{ 'opacity-75 cursor-wait': isJoining }"
+          @click="handleJoinCall"
+        >
+          <i
+            v-if="isJoining"
+            class="i-ph-circle-notch-bold text-sm animate-spin"
+          />
+          <i v-else class="i-ph-phone-bold text-sm" />
+          {{ $t(joinButtonLabel) }}
+        </button>
+      </div>
+
+      <div
+        v-if="recordingUrl && status === VOICE_CALL_STATUS.COMPLETED"
+        class="px-3 pb-2"
+      >
+        <audio controls class="w-full h-8" :src="recordingUrl">
+          {{ $t('CONVERSATION.VOICE_CALL.AUDIO_NOT_SUPPORTED') }}
+        </audio>
+      </div>
+
+      <div
+        v-if="transcript && status === VOICE_CALL_STATUS.COMPLETED"
+        class="px-3 pb-3"
+      >
+        <button
+          class="flex items-center gap-1 text-xs text-n-slate-11 hover:text-n-slate-12 transition-colors"
+          @click="showTranscript = !showTranscript"
+        >
+          <i
+            class="text-sm"
+            :class="
+              showTranscript ? 'i-ph-caret-up-bold' : 'i-ph-caret-down-bold'
+            "
+          />
+          {{ $t('CONVERSATION.VOICE_CALL.TRANSCRIPT') }}
+        </button>
+        <p
+          v-if="showTranscript"
+          class="mt-1 text-xs leading-relaxed text-n-slate-11 whitespace-pre-wrap"
+        >
+          {{ transcript }}
+        </p>
       </div>
     </div>
   </BaseBubble>

app/javascript/dashboard/components/widgets/WhatsappCallWidget.vue

@@ -0,0 +1,246 @@
+<script setup>
+import { watch, onUnmounted, onMounted } from 'vue';
+import { useRouter } from 'vue-router';
+import { useWhatsappCallSession } from 'dashboard/composables/useWhatsappCallSession';
+import Avatar from 'dashboard/components-next/avatar/Avatar.vue';
+import { useI18n } from 'vue-i18n';
+import { emitter } from 'shared/helpers/mitt';
+import { BUS_EVENTS } from 'shared/constants/busEvents';
+
+const { t } = useI18n();
+const router = useRouter();
+
+const {
+  activeCall,
+  incomingCalls,
+  hasActiveCall,
+  hasIncomingCall,
+  isAccepting,
+  isMuted,
+  isOutboundRinging,
+  isReconnecting,
+  callError,
+  formattedCallDuration,
+  acceptCall,
+  rejectCall,
+  endActiveCall,
+  toggleMute,
+  dismissIncomingCall,
+  startDurationTimer,
+} = useWhatsappCallSession();
+
+// In server-relay mode, the timer starts when the Peer B WebRTC handshake
+// completes (not when the agent clicks accept). Listen for this event.
+const onAgentWebRTCConnected = () => {
+  startDurationTimer();
+};
+
+const onPermissionGranted = ({ contactName }) => {
+  emitter.emit(BUS_EVENTS.SHOW_ALERT, {
+    message: t('WHATSAPP_CALL.PERMISSION_GRANTED', { contactName }),
+    type: 'success',
+  });
+};
+
+onMounted(() => {
+  emitter.on('whatsapp_call:agent_webrtc_connected', onAgentWebRTCConnected);
+  emitter.on('whatsapp_call:permission_granted', onPermissionGranted);
+});
+
+// Auto-dismiss ringing calls after 30 seconds
+const autoRejectTimers = new Map();
+
+const startAutoRejectTimer = call => {
+  if (autoRejectTimers.has(call.callId)) return;
+  const timer = setTimeout(() => {
+    dismissIncomingCall(call);
+    autoRejectTimers.delete(call.callId);
+  }, 30000);
+  autoRejectTimers.set(call.callId, timer);
+};
+
+const clearAutoRejectTimer = callId => {
+  const timer = autoRejectTimers.get(callId);
+  if (timer) {
+    clearTimeout(timer);
+    autoRejectTimers.delete(callId);
+  }
+};
+
+const handleAccept = async call => {
+  clearAutoRejectTimer(call.callId);
+  await acceptCall(call);
+  if (activeCall.value) {
+    router.push({
+      name: 'inbox_conversation',
+      params: { conversation_id: call.conversationId },
+    });
+  }
+};
+
+const handleReject = async call => {
+  clearAutoRejectTimer(call.callId);
+  await rejectCall(call);
+};
+
+const handleEndCall = async () => {
+  await endActiveCall();
+};
+
+// Start auto-reject timers for each newly added incoming call
+watch(
+  incomingCalls,
+  calls => {
+    calls.forEach(call => startAutoRejectTimer(call));
+  },
+  { immediate: true, deep: true }
+);
+
+onUnmounted(() => {
+  autoRejectTimers.forEach(timer => clearTimeout(timer));
+  autoRejectTimers.clear();
+  emitter.off('whatsapp_call:agent_webrtc_connected', onAgentWebRTCConnected);
+  emitter.off('whatsapp_call:permission_granted', onPermissionGranted);
+});
+</script>
+
+<template>
+  <div
+    v-if="hasIncomingCall || hasActiveCall"
+    class="fixed ltr:right-4 rtl:left-4 bottom-20 z-50 flex flex-col gap-2 w-72"
+  >
+    <!-- Error banner -->
+    <div
+      v-if="callError"
+      class="px-3 py-2 bg-n-ruby-3 border border-n-ruby-6 rounded-lg text-xs text-n-ruby-11"
+    >
+      {{ callError }}
+    </div>
+
+    <!-- Incoming calls (shown when there's no active call yet) -->
+    <template v-if="!hasActiveCall">
+      <div
+        v-for="call in incomingCalls"
+        :key="call.callId"
+        class="flex items-center gap-3 p-4 bg-n-solid-2 rounded-xl shadow-xl outline outline-1 outline-n-strong"
+      >
+        <div
+          class="animate-pulse ring-2 ring-n-teal-9 rounded-full inline-flex"
+        >
+          <Avatar
+            :src="call.caller?.avatar"
+            :name="call.caller?.name || call.caller?.phone"
+            :size="40"
+            rounded-full
+          />
+        </div>
+        <div class="flex-1 min-w-0">
+          <p class="text-sm font-medium text-n-slate-12 truncate mb-0">
+            {{
+              call.caller?.name ||
+              call.caller?.phone ||
+              t('WHATSAPP_CALL.UNKNOWN_CALLER')
+            }}
+          </p>
+          <p class="text-xs text-n-slate-11 truncate">
+            {{ t('WHATSAPP_CALL.INCOMING_WHATSAPP_CALL') }}
+          </p>
+        </div>
+        <div class="flex shrink-0 gap-2">
+          <button
+            class="flex justify-center items-center w-10 h-10 bg-n-ruby-9 hover:bg-n-ruby-10 rounded-full transition-colors"
+            :title="t('WHATSAPP_CALL.REJECT')"
+            @click="handleReject(call)"
+          >
+            <i class="text-lg text-white i-ph-phone-x-bold" />
+          </button>
+          <button
+            class="flex justify-center items-center w-10 h-10 bg-n-teal-9 hover:bg-n-teal-10 rounded-full transition-colors"
+            :disabled="isAccepting"
+            :title="t('WHATSAPP_CALL.ACCEPT')"
+            @click="handleAccept(call)"
+          >
+            <i
+              v-if="isAccepting"
+              class="text-lg text-white i-ph-circle-notch animate-spin"
+            />
+            <i v-else class="text-lg text-white i-ph-phone-bold" />
+          </button>
+        </div>
+      </div>
+    </template>
+
+    <!-- Active call widget -->
+    <div
+      v-if="hasActiveCall"
+      class="flex items-center gap-3 p-4 bg-n-solid-2 rounded-xl shadow-xl outline outline-1 outline-n-strong"
+    >
+      <div
+        class="ring-2 ring-n-teal-9 rounded-full inline-flex"
+        :class="{ 'animate-pulse': isOutboundRinging }"
+      >
+        <Avatar
+          :src="activeCall.caller?.avatar"
+          :name="activeCall.caller?.name || activeCall.caller?.phone"
+          :size="40"
+          rounded-full
+        />
+      </div>
+      <div class="flex-1 min-w-0">
+        <p class="text-sm font-medium text-n-slate-12 truncate mb-0">
+          {{
+            activeCall.caller?.name ||
+            activeCall.caller?.phone ||
+            t('WHATSAPP_CALL.UNKNOWN_CALLER')
+          }}
+        </p>
+        <p
+          class="text-sm"
+          :class="
+            isOutboundRinging || isReconnecting
+              ? 'text-n-slate-11'
+              : 'font-mono text-n-teal-9'
+          "
+        >
+          <template v-if="isReconnecting">
+            {{ t('WHATSAPP_CALL.RECONNECTING') }}
+          </template>
+          <template v-else-if="isOutboundRinging">
+            {{ t('WHATSAPP_CALL.RINGING') }}
+          </template>
+          <template v-else>
+            {{ formattedCallDuration }}
+          </template>
+        </p>
+      </div>
+      <div class="flex shrink-0 gap-2">
+        <!-- Mute toggle -->
+        <button
+          class="flex justify-center items-center w-9 h-9 rounded-full transition-colors"
+          :class="
+            isMuted
+              ? 'bg-n-amber-9 hover:bg-n-amber-10'
+              : 'bg-n-slate-4 hover:bg-n-slate-5'
+          "
+          :title="isMuted ? t('WHATSAPP_CALL.UNMUTE') : t('WHATSAPP_CALL.MUTE')"
+          @click="toggleMute"
+        >
+          <i
+            class="text-base text-white"
+            :class="
+              isMuted ? 'i-ph-microphone-slash-bold' : 'i-ph-microphone-bold'
+            "
+          />
+        </button>
+        <!-- Hang up -->
+        <button
+          class="flex justify-center items-center w-9 h-9 bg-n-ruby-9 hover:bg-n-ruby-10 rounded-full transition-colors"
+          :title="t('WHATSAPP_CALL.HANG_UP')"
+          @click="handleEndCall"
+        >
+          <i class="text-base text-white i-ph-phone-x-bold" />
+        </button>
+      </div>
+    </div>
+  </div>
+</template>

app/javascript/dashboard/components/widgets/conversation/ConversationHeader.vue

@@ -13,6 +13,14 @@ import { conversationListPageURL } from 'dashboard/helper/URLHelper';
 import { snoozedReopenTime } from 'dashboard/helper/snoozeHelpers';
 import { useInbox } from 'dashboard/composables/useInbox';
 import { useI18n } from 'vue-i18n';
+import WhatsappCallsAPI from 'dashboard/api/whatsappCalls';
+import { emitter } from 'shared/helpers/mitt';
+import { BUS_EVENTS } from 'shared/constants/busEvents';
+import {
+  useWhatsappCallsStore,
+  setOutboundCallProperty,
+} from 'dashboard/stores/whatsappCalls';
+import { startCallRecording } from 'dashboard/composables/useWhatsappCallSession';
 
 const props = defineProps({
   chat: {
@@ -30,7 +38,9 @@ const store = useStore();
 const route = useRoute();
 const conversationHeader = ref(null);
 const { width } = useElementSize(conversationHeader);
-const { isAWebWidgetInbox } = useInbox();
+const { isAWebWidgetInbox, isAWhatsAppCloudChannel } = useInbox();
+const whatsappCallsStore = useWhatsappCallsStore();
+const isInitiatingCall = ref(false);
 
 const currentChat = computed(() => store.getters.getSelectedChat);
 const accountId = computed(() => store.getters.getCurrentAccountId);
@@ -91,6 +101,211 @@ const hasMultipleInboxes = computed(
 );
 
 const hasSlaPolicyId = computed(() => props.chat?.sla_policy_id);
+
+const canInitiateWhatsappCall = computed(() => {
+  if (!isAWhatsAppCloudChannel.value) return false;
+  if (!inbox.value?.calling_enabled) return false;
+  if (whatsappCallsStore.hasWhatsappCall) return false;
+  return true;
+});
+
+// Detect if the media server is enabled for this inbox.
+// When enabled, the browser should NOT create its own WebRTC offer.
+const isMediaServerEnabled = computed(
+  () => !!inbox.value?.media_server_enabled
+);
+
+const waitForOutboundIceGathering = pc =>
+  new Promise((resolve, reject) => {
+    if (pc.iceGatheringState === 'complete') {
+      resolve();
+      return;
+    }
+
+    let timeout = null;
+
+    const cleanup = () => {
+      clearTimeout(timeout);
+      pc.onicegatheringstatechange = null;
+      pc.oniceconnectionstatechange = null;
+    };
+
+    timeout = setTimeout(() => {
+      cleanup();
+      resolve();
+    }, 10000);
+
+    pc.onicegatheringstatechange = () => {
+      if (pc.iceGatheringState === 'complete') {
+        cleanup();
+        resolve();
+      }
+    };
+    pc.oniceconnectionstatechange = () => {
+      if (pc.iceConnectionState === 'failed') {
+        cleanup();
+        reject(new Error('ICE connection failed'));
+      }
+    };
+  });
+
+/**
+ * Server-relay mode: POST /initiate without SDP. The media server creates
+ * Peer A (Meta-side) and later sends the agent Peer B offer via ActionCable
+ * (whatsapp_call.outbound_connected with sdp_offer).
+ */
+const initiateServerRelayCall = async () => {
+  if (isInitiatingCall.value || !currentChat.value?.id) return;
+  isInitiatingCall.value = true;
+
+  try {
+    const response = await WhatsappCallsAPI.initiate(currentChat.value.id);
+
+    const callStatus = response.data?.status;
+    if (
+      callStatus === 'permission_requested' ||
+      callStatus === 'permission_pending'
+    ) {
+      const message =
+        callStatus === 'permission_requested'
+          ? t('WHATSAPP_CALL.PERMISSION_REQUESTED')
+          : t('WHATSAPP_CALL.PERMISSION_PENDING');
+      emitter.emit(BUS_EVENTS.SHOW_ALERT, { message, type: 'info' });
+      return;
+    }
+
+    emitter.emit(BUS_EVENTS.SHOW_ALERT, {
+      message: t('WHATSAPP_CALL.CALLING'),
+      type: 'success',
+    });
+
+    const outboundCallId = response.data?.call_id;
+
+    // Set active call — WebRTC setup happens when ActionCable delivers agent_offer
+    whatsappCallsStore.setActiveCall({
+      id: response.data?.id,
+      callId: outboundCallId,
+      direction: 'outbound',
+      status: 'ringing',
+      serverRelay: true,
+      conversationId: currentChat.value.id,
+      caller: {
+        name: currentContact.value?.name,
+        phone: currentContact.value?.phone_number,
+        avatar: currentContact.value?.thumbnail,
+      },
+    });
+  } catch (err) {
+    const errorMessage =
+      err.response?.data?.error || t('WHATSAPP_CALL.CALL_FAILED');
+    emitter.emit(BUS_EVENTS.SHOW_ALERT, {
+      message: errorMessage,
+      type: 'error',
+    });
+  } finally {
+    isInitiatingCall.value = false;
+  }
+};
+
+/**
+ * Legacy mode: Browser creates RTCPeerConnection, generates SDP offer,
+ * sends it to backend which forwards to Meta.
+ */
+const initiateLegacyCall = async () => {
+  if (isInitiatingCall.value || !currentChat.value?.id) return;
+  isInitiatingCall.value = true;
+  let pc = null;
+  let localStream = null;
+  let recordCallId = null;
+  try {
+    localStream = await navigator.mediaDevices.getUserMedia({ audio: true });
+    pc = new RTCPeerConnection({
+      iceServers: [{ urls: 'stun:stun.l.google.com:19302' }],
+    });
+    localStream.getTracks().forEach(track => pc.addTrack(track, localStream));
+
+    pc.ontrack = event => {
+      const [stream] = event.streams;
+      if (!stream) return;
+      const audio = document.createElement('audio');
+      audio.srcObject = stream;
+      audio.autoplay = true;
+      document.body.appendChild(audio);
+      setOutboundCallProperty('audio', audio);
+      whatsappCallsStore.markActiveCallConnected();
+      if (recordCallId) startCallRecording(pc, localStream, recordCallId);
+    };
+
+    const offer = await pc.createOffer();
+    await pc.setLocalDescription(offer);
+
+    await waitForOutboundIceGathering(pc);
+    const completeSdp = pc.localDescription.sdp;
+
+    const response = await WhatsappCallsAPI.initiate(
+      currentChat.value.id,
+      completeSdp
+    );
+
+    const callStatus = response.data?.status;
+    if (
+      callStatus === 'permission_requested' ||
+      callStatus === 'permission_pending'
+    ) {
+      pc.close();
+      localStream.getTracks().forEach(track => track.stop());
+      const message =
+        callStatus === 'permission_requested'
+          ? t('WHATSAPP_CALL.PERMISSION_REQUESTED')
+          : t('WHATSAPP_CALL.PERMISSION_PENDING');
+      emitter.emit(BUS_EVENTS.SHOW_ALERT, { message, type: 'info' });
+      return;
+    }
+
+    emitter.emit(BUS_EVENTS.SHOW_ALERT, {
+      message: t('WHATSAPP_CALL.CALLING'),
+      type: 'success',
+    });
+
+    const outboundCallId = response.data?.call_id;
+    recordCallId = response.data?.id;
+    setOutboundCallProperty('pc', pc);
+    setOutboundCallProperty('stream', localStream);
+    setOutboundCallProperty('callId', outboundCallId);
+
+    whatsappCallsStore.setActiveCall({
+      id: response.data?.id,
+      callId: outboundCallId,
+      direction: 'outbound',
+      status: 'ringing',
+      serverRelay: false,
+      conversationId: currentChat.value.id,
+      caller: {
+        name: currentContact.value?.name,
+        phone: currentContact.value?.phone_number,
+        avatar: currentContact.value?.thumbnail,
+      },
+    });
+  } catch (err) {
+    if (pc) pc.close();
+    if (localStream) localStream.getTracks().forEach(track => track.stop());
+    const errorMessage =
+      err.response?.data?.error || t('WHATSAPP_CALL.CALL_FAILED');
+    emitter.emit(BUS_EVENTS.SHOW_ALERT, {
+      message: errorMessage,
+      type: 'error',
+    });
+  } finally {
+    isInitiatingCall.value = false;
+  }
+};
+
+const initiateWhatsappCall = () => {
+  if (isMediaServerEnabled.value) {
+    return initiateServerRelayCall();
+  }
+  return initiateLegacyCall();
+};
 </script>
 
 <template>
@@ -152,6 +367,19 @@ const hasSlaPolicyId = computed(() => props.chat?.sla_policy_id);
         :parent-width="width"
         class="hidden md:flex"
       />
+      <button
+        v-if="canInitiateWhatsappCall"
+        v-tooltip="$t('WHATSAPP_CALL.INITIATE_CALL')"
+        class="flex items-center justify-center w-8 h-8 rounded-lg text-n-slate-11 hover:text-n-slate-12 hover:bg-n-slate-3 transition-colors"
+        :disabled="isInitiatingCall"
+        @click="initiateWhatsappCall"
+      >
+        <i
+          v-if="isInitiatingCall"
+          class="text-base i-ph-circle-notch animate-spin"
+        />
+        <i v-else class="text-base i-ph-phone-bold" />
+      </button>
       <MoreActions :conversation-id="currentChat.id" />
     </div>
   </div>

app/javascript/dashboard/composables/useCallReconnection.js

@@ -0,0 +1,63 @@
+import { onMounted, computed } from 'vue';
+import { useWhatsappCallsStore } from 'dashboard/stores/whatsappCalls';
+import WhatsappCallsAPI from 'dashboard/api/whatsappCalls';
+
+/**
+ * Checks for an active WhatsApp call on page load and reconnects if found.
+ * This handles the server-relay scenario where the call persists on the media
+ * server even after the agent's browser reloads.
+ *
+ * NOTE: This composable intentionally does NOT call useWhatsappCallSession()
+ * to avoid creating duplicate side effects (beforeunload handlers, cleanup
+ * callbacks, timers). The WhatsappCallWidget owns the useWhatsappCallSession
+ * instance. This composable only sets store state and triggers the reconnect
+ * API call — the actual WebRTC setup happens when the ActionCable agent_offer
+ * event arrives and is handled by handleAgentOffer.
+ *
+ * Usage: call `useCallReconnection()` in the app-level layout component that
+ * mounts once on page load.
+ */
+export function useCallReconnection() {
+  const callsStore = useWhatsappCallsStore();
+
+  const isReconnecting = computed(() => callsStore.isReconnecting);
+
+  onMounted(async () => {
+    // Skip if there's already an active or incoming call in the store
+    if (callsStore.hasActiveCall || callsStore.hasIncomingCall) return;
+
+    try {
+      const { data } = await WhatsappCallsAPI.active();
+      if (!data?.call) return;
+
+      const activeCallData = data.call;
+
+      callsStore.setReconnecting(true);
+      callsStore.setActiveCall({
+        id: activeCallData.id,
+        callId: activeCallData.call_id,
+        direction: activeCallData.direction,
+        conversationId: activeCallData.conversation_id,
+        status: 'reconnecting',
+        serverRelay: true,
+        caller: activeCallData.caller,
+      });
+
+      // Set timer offset so the timer resumes from the correct elapsed time
+      if (activeCallData.elapsed_seconds) {
+        callsStore.setTimerOffset(activeCallData.elapsed_seconds);
+      }
+
+      // Tell the server to create a new Peer B and send us a fresh SDP offer.
+      // The server will broadcast whatsapp_call.agent_offer via ActionCable,
+      // which is handled by handleAgentOffer in actionCable.js.
+      await WhatsappCallsAPI.reconnect(activeCallData.id);
+    } catch {
+      // No active call or API/reconnect error — clear state and silent fail.
+      // clearActiveCall() also resets isReconnecting and callTimerOffset.
+      callsStore.clearActiveCall();
+    }
+  });
+
+  return { isReconnecting };
+}

app/javascript/dashboard/composables/useWhatsappCallSession.js

@@ -0,0 +1,542 @@
+import { ref, computed, watch, onUnmounted } from 'vue';
+import { useI18n } from 'vue-i18n';
+import {
+  useWhatsappCallsStore,
+  getOutboundCallState,
+  cleanupOutboundCall,
+} from 'dashboard/stores/whatsappCalls';
+import WhatsappCallsAPI from 'dashboard/api/whatsappCalls';
+import Auth from 'dashboard/api/auth';
+import Timer from 'dashboard/helper/Timer';
+
+// ── Module-level WebRTC state (shared across legacy inbound + server-relay) ──
+let inboundPc = null;
+let inboundStream = null;
+let inboundAudio = null;
+
+// ── Module-level recording state (legacy mode only) ──
+let mediaRecorder = null;
+let recordedChunks = [];
+let recordingCallId = null;
+
+function cleanupInboundWebRTC() {
+  if (inboundStream) {
+    inboundStream.getTracks().forEach(track => track.stop());
+    inboundStream = null;
+  }
+  if (inboundPc) {
+    inboundPc.close();
+    inboundPc = null;
+  }
+  if (inboundAudio) {
+    inboundAudio.srcObject = null;
+    if (inboundAudio.parentNode) {
+      inboundAudio.parentNode.removeChild(inboundAudio);
+    }
+    inboundAudio = null;
+  }
+}
+
+/**
+ * Start recording both local and remote audio tracks via MediaRecorder.
+ * Mixes them into a single stream using AudioContext.
+ * Used ONLY in legacy (browser-direct) mode.
+ */
+export function startCallRecording(pc, localStream, callId) {
+  try {
+    const ctx = new AudioContext();
+    const dest = ctx.createMediaStreamDestination();
+
+    if (localStream) {
+      const localSource = ctx.createMediaStreamSource(localStream);
+      localSource.connect(dest);
+    }
+
+    pc.getReceivers().forEach(receiver => {
+      if (receiver.track && receiver.track.kind === 'audio') {
+        const remoteStream = new MediaStream([receiver.track]);
+        const remoteSource = ctx.createMediaStreamSource(remoteStream);
+        remoteSource.connect(dest);
+      }
+    });
+
+    recordedChunks = [];
+    recordingCallId = callId;
+    const recorder = new MediaRecorder(dest.stream, {
+      mimeType: 'audio/webm;codecs=opus',
+    });
+
+    recorder.ondataavailable = e => {
+      if (e.data.size > 0) recordedChunks.push(e.data);
+    };
+
+    mediaRecorder = recorder;
+    recorder.start(1000);
+  } catch (err) {
+    // eslint-disable-next-line no-console
+    console.error('[WhatsApp Call] Failed to start recording:', err);
+  }
+}
+
+/**
+ * Stop recording and upload the audio blob to the backend.
+ * Used ONLY in legacy (browser-direct) mode.
+ */
+function stopAndUploadRecording(callId) {
+  if (!mediaRecorder || mediaRecorder.state === 'inactive') return;
+
+  const id = callId || recordingCallId;
+
+  mediaRecorder.onstop = () => {
+    if (recordedChunks.length === 0 || !id) return;
+
+    const blob = new Blob(recordedChunks, { type: 'audio/webm' });
+    recordedChunks = [];
+    recordingCallId = null;
+
+    WhatsappCallsAPI.uploadRecording(id, blob).catch(err => {
+      // eslint-disable-next-line no-console
+      console.error('[WhatsApp Call] Failed to upload recording:', err);
+    });
+  };
+
+  mediaRecorder.stop();
+  mediaRecorder = null;
+}
+
+function waitForIceGatheringComplete(pc) {
+  return new Promise((resolve, reject) => {
+    if (pc.iceGatheringState === 'complete') {
+      resolve();
+      return;
+    }
+
+    let timeout = null;
+
+    const cleanup = () => {
+      clearTimeout(timeout);
+      pc.onicegatheringstatechange = null;
+      pc.oniceconnectionstatechange = null;
+    };
+
+    timeout = setTimeout(() => {
+      cleanup();
+      // eslint-disable-next-line no-console
+      console.warn(
+        '[WhatsApp Call] ICE gathering timed out, sending partial SDP'
+      );
+      resolve();
+    }, 10000);
+
+    pc.onicegatheringstatechange = () => {
+      if (pc.iceGatheringState === 'complete') {
+        cleanup();
+        resolve();
+      }
+    };
+    pc.oniceconnectionstatechange = () => {
+      if (pc.iceConnectionState === 'failed') {
+        cleanup();
+        reject(new Error('ICE connection failed'));
+      }
+    };
+  });
+}
+
+// ── Server-relay mode detection ──
+// Prefer the explicit flag set by Rails whenever it's available (call object
+// from GET /whatsapp_calls/:id, inbox serializer, or the incoming ActionCable
+// broadcast). Fall back to the presence of sdpOffer so legacy deployments keep
+// working when the field isn't set.
+function isServerRelayCall(call) {
+  if (call?.mediaServerEnabled === true) return true;
+  if (call?.mediaServerEnabled === false) return false;
+  return !call?.sdpOffer;
+}
+
+/**
+ * Handle an SDP offer from the media server (Peer B). Used in server-relay mode
+ * for both inbound accept and outbound connect flows.
+ *
+ * Flow: getUserMedia -> RTCPeerConnection(iceServers) -> setRemoteDescription(offer)
+ *       -> createAnswer -> waitForICE -> POST /agent_answer
+ */
+async function handleAgentOffer(callId, sdpOffer, iceServers) {
+  cleanupInboundWebRTC();
+
+  try {
+    const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
+    inboundStream = stream;
+
+    const servers = iceServers?.length
+      ? iceServers
+      : [{ urls: 'stun:stun.l.google.com:19302' }];
+
+    const pc = new RTCPeerConnection({ iceServers: servers });
+    inboundPc = pc;
+
+    stream.getTracks().forEach(track => pc.addTrack(track, stream));
+
+    pc.ontrack = event => {
+      const [remoteStream] = event.streams;
+      if (!remoteStream) return;
+      if (!inboundAudio) {
+        const audio = document.createElement('audio');
+        audio.autoplay = true;
+        document.body.appendChild(audio);
+        inboundAudio = audio;
+      }
+      inboundAudio.srcObject = remoteStream;
+      inboundAudio.play().catch(() => {});
+
+      // No client-side recording in server-relay mode — the media server records
+    };
+
+    await pc.setRemoteDescription({ type: 'offer', sdp: sdpOffer });
+    const answer = await pc.createAnswer();
+    await pc.setLocalDescription(answer);
+    await waitForIceGatheringComplete(pc);
+
+    const completeSdp = pc.localDescription.sdp;
+    await WhatsappCallsAPI.agentAnswer(callId, completeSdp);
+
+    return { success: true };
+  } catch (err) {
+    cleanupInboundWebRTC();
+    throw err;
+  }
+}
+
+// Expose handleAgentOffer so ActionCable handler can invoke it
+export { handleAgentOffer };
+
+/**
+ * Legacy mode: creates WebRTC session and posts SDP to backend (browser ↔ Meta).
+ * Can be called from anywhere — composable, widget, or bubble.
+ */
+async function doAcceptCall(call) {
+  // Server-relay mode: just POST /accept without SDP. Wait for agent_offer event.
+  if (isServerRelayCall(call)) {
+    await WhatsappCallsAPI.accept(call.id);
+    return { success: true, awaitingAgentOffer: true };
+  }
+
+  // Legacy mode: full browser-side WebRTC handshake
+  cleanupInboundWebRTC();
+
+  try {
+    const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
+    inboundStream = stream;
+
+    const iceServers = call.iceServers?.length
+      ? call.iceServers
+      : [{ urls: 'stun:stun.l.google.com:19302' }];
+
+    const pc = new RTCPeerConnection({ iceServers });
+    inboundPc = pc;
+
+    stream.getTracks().forEach(track => pc.addTrack(track, stream));
+
+    pc.ontrack = event => {
+      const [remoteStream] = event.streams;
+      if (!remoteStream) return;
+      if (!inboundAudio) {
+        const audio = document.createElement('audio');
+        audio.autoplay = true;
+        document.body.appendChild(audio);
+        inboundAudio = audio;
+      }
+      inboundAudio.srcObject = remoteStream;
+      inboundAudio.play().catch(() => {});
+
+      // Start recording once remote audio is available (legacy mode only)
+      startCallRecording(pc, stream, call.id);
+    };
+
+    await pc.setRemoteDescription({ type: 'offer', sdp: call.sdpOffer });
+    const answer = await pc.createAnswer();
+    await pc.setLocalDescription(answer);
+    await waitForIceGatheringComplete(pc);
+
+    const completeSdp = pc.localDescription.sdp;
+    await WhatsappCallsAPI.accept(call.id, completeSdp);
+
+    return { success: true };
+  } catch (err) {
+    cleanupInboundWebRTC();
+    throw err;
+  }
+}
+
+/**
+ * Standalone function callable from VoiceCall bubble.
+ * Fetches call data if needed, runs WebRTC accept, updates store.
+ */
+export async function acceptWhatsappCallById(callId) {
+  const callsStore = useWhatsappCallsStore();
+
+  if (callsStore.hasActiveCall) {
+    return { success: false, error: 'active_call_exists' };
+  }
+
+  let call = callsStore.incomingCalls.find(
+    c => c.id === callId || c.callId === String(callId)
+  );
+
+  if (!call) {
+    const { data } = await WhatsappCallsAPI.show(callId);
+    if (data.status !== 'ringing') {
+      return { success: false, error: 'not_ringing' };
+    }
+    call = {
+      id: data.id,
+      callId: data.call_id,
+      direction: data.direction,
+      inboxId: data.inbox_id,
+      conversationId: data.conversation_id,
+      sdpOffer: data.sdp_offer,
+      iceServers: data.ice_servers,
+      mediaServerEnabled: data.media_server_enabled,
+      caller: data.caller,
+    };
+    callsStore.addIncomingCall(call);
+  }
+
+  try {
+    const result = await doAcceptCall(call);
+
+    callsStore.removeIncomingCall(call.callId);
+
+    // In server-relay mode the call becomes active but awaits the agent_offer
+    // ActionCable event to complete WebRTC setup. Mark it with serverRelay flag.
+    const activeCallData = {
+      ...call,
+      serverRelay: isServerRelayCall(call),
+    };
+    callsStore.setActiveCall(activeCallData);
+
+    return { success: true, call: activeCallData, ...result };
+  } catch (err) {
+    callsStore.removeIncomingCall(call.callId);
+    throw err;
+  }
+}
+
+/**
+ * Fire-and-forget terminate request using fetch + keepalive.
+ * Works reliably inside beforeunload / pagehide where axios won't complete.
+ * Used ONLY in legacy mode. Server-relay mode does NOT terminate on unload.
+ */
+function terminateCallOnUnload(callId) {
+  const authData = Auth.hasAuthCookie() ? Auth.getAuthData() : {};
+  const accountId =
+    window.location.pathname.includes('/app/accounts') &&
+    window.location.pathname.split('/')[3];
+  if (!accountId) return;
+
+  const url = `/api/v1/accounts/${accountId}/whatsapp_calls/${callId}/terminate`;
+  fetch(url, {
+    method: 'POST',
+    keepalive: true,
+    headers: {
+      'Content-Type': 'application/json',
+      'access-token': authData['access-token'] || '',
+      'token-type': authData['token-type'] || '',
+      client: authData.client || '',
+      expiry: authData.expiry || '',
+      uid: authData.uid || '',
+    },
+  }).catch(() => {});
+}
+
+// ── Composable (used by WhatsappCallWidget for floating UI + timer) ──
+export function useWhatsappCallSession() {
+  const { t } = useI18n();
+  const callsStore = useWhatsappCallsStore();
+
+  const isAccepting = ref(false);
+  const isMuted = ref(false);
+  const callError = ref(null);
+  const callDuration = ref(0);
+  const isReconnecting = computed(() => callsStore.isReconnecting);
+
+  const durationTimer = new Timer(elapsed => {
+    callDuration.value = callsStore.callTimerOffset + elapsed;
+  });
+
+  const activeCall = computed(() => callsStore.activeCall);
+  const incomingCalls = computed(() => callsStore.incomingCalls);
+  const hasActiveCall = computed(() => callsStore.hasActiveCall);
+  const hasIncomingCall = computed(() => callsStore.hasIncomingCall);
+  const firstIncomingCall = computed(() => callsStore.firstIncomingCall);
+
+  const isOutboundRinging = computed(
+    () =>
+      activeCall.value?.direction === 'outbound' &&
+      activeCall.value?.status === 'ringing'
+  );
+
+  const formattedCallDuration = computed(() => {
+    const minutes = Math.floor(callDuration.value / 60);
+    const seconds = callDuration.value % 60;
+    return `${minutes.toString().padStart(2, '0')}:${seconds.toString().padStart(2, '0')}`;
+  });
+
+  // Register cleanup so external call-end events can teardown WebRTC
+  callsStore.registerCleanupCallback(() => {
+    // Only do recording cleanup in legacy mode
+    if (!callsStore.isMediaServerEnabled) {
+      stopAndUploadRecording();
+    }
+    cleanupInboundWebRTC();
+    durationTimer.stop();
+    callDuration.value = 0;
+  });
+
+  // On page close / reload:
+  // - Legacy mode: terminate call (current behavior)
+  // - Server-relay mode: just clean up local WebRTC resources, call persists
+  const handleBeforeUnload = () => {
+    const call = callsStore.activeCall;
+    if (!call?.id) return;
+
+    if (call.serverRelay) {
+      // Server-relay: only clean up local resources, do NOT terminate
+      cleanupInboundWebRTC();
+    } else {
+      // Legacy: terminate and clean up
+      terminateCallOnUnload(call.id);
+      cleanupInboundWebRTC();
+    }
+  };
+  window.addEventListener('beforeunload', handleBeforeUnload);
+
+  // Start timer when outbound call becomes connected
+  watch(activeCall, call => {
+    if (
+      call?.direction === 'outbound' &&
+      call?.status === 'connected' &&
+      !durationTimer.intervalId
+    ) {
+      durationTimer.start();
+    }
+  });
+
+  /**
+   * Accept an incoming call — used by the floating widget buttons.
+   */
+  const acceptCall = async call => {
+    if (isAccepting.value) return;
+    isAccepting.value = true;
+    callError.value = null;
+
+    try {
+      const result = await doAcceptCall(call);
+      callsStore.removeIncomingCall(call.callId);
+
+      const activeCallData = {
+        ...call,
+        serverRelay: isServerRelayCall(call),
+      };
+      callsStore.setActiveCall(activeCallData);
+
+      // In legacy mode, WebRTC is already established so start timer now.
+      // In server-relay mode, timer starts when handleAgentOffer completes
+      // (triggered by the whatsapp_call.agent_offer ActionCable event).
+      if (!result.awaitingAgentOffer) {
+        durationTimer.start();
+      }
+    } catch (err) {
+      callError.value =
+        err.name === 'NotAllowedError'
+          ? t('WHATSAPP_CALL.MIC_DENIED')
+          : t('WHATSAPP_CALL.CALL_FAILED');
+      // eslint-disable-next-line no-console
+      console.error('[WhatsApp Call] acceptCall error:', err);
+      // Note: doAcceptCall already cleans up WebRTC resources on error
+    } finally {
+      isAccepting.value = false;
+    }
+  };
+
+  const rejectCall = async call => {
+    try {
+      await WhatsappCallsAPI.reject(call.id);
+    } catch {
+      // Best effort
+    } finally {
+      callsStore.removeIncomingCall(call.callId);
+    }
+  };
+
+  const endActiveCall = async () => {
+    const call = activeCall.value;
+    if (!call) return;
+
+    // Only upload recording in legacy mode
+    if (!call.serverRelay) {
+      stopAndUploadRecording(call.id);
+    }
+
+    try {
+      await WhatsappCallsAPI.terminate(call.id);
+    } catch {
+      // Best effort
+    } finally {
+      cleanupInboundWebRTC();
+      cleanupOutboundCall();
+      // Clear state directly — do NOT use handleCallEnded here since that is
+      // meant for external events (ActionCable) and would invoke cleanupCallback
+      // which would duplicate the cleanup we just performed.
+      callsStore.clearActiveCall();
+      durationTimer.stop();
+      callDuration.value = 0;
+    }
+  };
+
+  const toggleMute = () => {
+    const stream = inboundStream || getOutboundCallState().stream;
+    if (!stream) return;
+    const audioTrack = stream.getAudioTracks()[0];
+    if (!audioTrack) return;
+    audioTrack.enabled = !audioTrack.enabled;
+    isMuted.value = !audioTrack.enabled;
+  };
+
+  const dismissIncomingCall = call => {
+    callsStore.removeIncomingCall(call.callId);
+  };
+
+  /**
+   * Start the duration timer. Called externally after server-relay WebRTC
+   * setup completes (handleAgentOffer).
+   */
+  const startDurationTimer = () => {
+    durationTimer.start();
+  };
+
+  onUnmounted(() => {
+    window.removeEventListener('beforeunload', handleBeforeUnload);
+    durationTimer.stop();
+  });
+
+  return {
+    activeCall,
+    incomingCalls,
+    hasActiveCall,
+    hasIncomingCall,
+    firstIncomingCall,
+    isAccepting,
+    isMuted,
+    isOutboundRinging,
+    isReconnecting,
+    callError,
+    formattedCallDuration,
+    acceptCall,
+    rejectCall,
+    endActiveCall,
+    toggleMute,
+    dismissIncomingCall,
+    startDurationTimer,
+  };
+}

app/javascript/dashboard/featureFlags.js

@@ -45,6 +45,7 @@ export const FEATURE_FLAGS = {
   COMPANIES: 'companies',
   ADVANCED_SEARCH: 'advanced_search',
   CONVERSATION_REQUIRED_ATTRIBUTES: 'conversation_required_attributes',
+  WHATSAPP_CALL: 'whatsapp_call',
 };
 
 export const PREMIUM_FEATURES = [
@@ -57,4 +58,5 @@ export const PREMIUM_FEATURES = [
   FEATURE_FLAGS.SAML,
   FEATURE_FLAGS.CONVERSATION_REQUIRED_ATTRIBUTES,
   FEATURE_FLAGS.ADVANCED_ASSIGNMENT,
+  FEATURE_FLAGS.WHATSAPP_CALL,
 ];

app/javascript/dashboard/helper/actionCable.js

@@ -4,6 +4,11 @@ import DashboardAudioNotificationHelper from './AudioAlerts/DashboardAudioNotifi
 import { BUS_EVENTS } from 'shared/constants/busEvents';
 import { emitter } from 'shared/helpers/mitt';
 import { useImpersonation } from 'dashboard/composables/useImpersonation';
+import {
+  useWhatsappCallsStore,
+  getOutboundCallState,
+} from 'dashboard/stores/whatsappCalls';
+import { handleAgentOffer } from 'dashboard/composables/useWhatsappCallSession';
 
 const { isImpersonating } = useImpersonation();
 
@@ -34,6 +39,12 @@ class ActionCableConnector extends BaseActionCableConnector {
       'conversation.updated': this.onConversationUpdated,
       'account.cache_invalidated': this.onCacheInvalidate,
       'copilot.message.created': this.onCopilotMessageCreated,
+      'whatsapp_call.incoming': this.onWhatsappCallIncoming,
+      'whatsapp_call.accepted': this.onWhatsappCallAccepted,
+      'whatsapp_call.ended': this.onWhatsappCallEnded,
+      'whatsapp_call.outbound_connected': this.onWhatsappCallOutboundConnected,
+      'whatsapp_call.permission_granted': this.onWhatsappCallPermissionGranted,
+      'whatsapp_call.agent_offer': this.onWhatsappCallAgentOffer,
     };
   }
 
@@ -200,6 +211,111 @@ class ActionCableConnector extends BaseActionCableConnector {
     this.app.$store.dispatch('inboxes/revalidate', { newKey: keys.inbox });
     this.app.$store.dispatch('teams/revalidate', { newKey: keys.team });
   };
+
+  // eslint-disable-next-line class-methods-use-this
+  onWhatsappCallIncoming = data => {
+    const whatsappCallsStore = useWhatsappCallsStore();
+    // In server-relay mode, sdp_offer and ice_servers are absent — the media
+    // server handles WebRTC with Meta, and the browser only needs call metadata.
+    whatsappCallsStore.addIncomingCall({
+      id: data.id,
+      callId: data.call_id,
+      direction: data.direction,
+      inboxId: data.inbox_id,
+      conversationId: data.conversation_id,
+      caller: data.caller,
+      sdpOffer: data.sdp_offer || null,
+      iceServers: data.ice_servers || null,
+      mediaServerEnabled: data.media_server_enabled,
+    });
+  };
+
+  onWhatsappCallAccepted = data => {
+    const whatsappCallsStore = useWhatsappCallsStore();
+    const currentUserId = this.app.$store.getters.getCurrentUserID;
+    // If accepted by a different agent, remove from incoming list for this agent
+    if (data.accepted_by_agent_id !== currentUserId) {
+      whatsappCallsStore.handleCallAcceptedByOther(data.call_id);
+    }
+  };
+
+  // eslint-disable-next-line class-methods-use-this
+  onWhatsappCallEnded = data => {
+    const whatsappCallsStore = useWhatsappCallsStore();
+    whatsappCallsStore.handleCallEnded(data.call_id);
+  };
+
+  // eslint-disable-next-line class-methods-use-this
+  onWhatsappCallOutboundConnected = data => {
+    const whatsappCallsStore = useWhatsappCallsStore();
+
+    // Server-relay mode: data contains sdp_offer (media server generated offer
+    // for Peer B) instead of sdp_answer.
+    if (data.sdp_offer) {
+      const activeCall = whatsappCallsStore.activeCall;
+      if (activeCall && activeCall.callId === data.call_id) {
+        handleAgentOffer(activeCall.id, data.sdp_offer, data.ice_servers)
+          .then(() => {
+            whatsappCallsStore.markActiveCallConnected();
+            // Emit event so the composable can start the timer
+            emitter.emit('whatsapp_call:agent_webrtc_connected');
+          })
+          .catch(err => {
+            // eslint-disable-next-line no-console
+            console.error(
+              '[WhatsApp Call] Failed to handle outbound agent offer:',
+              err
+            );
+          });
+      }
+      return;
+    }
+
+    // Legacy mode: data contains sdp_answer (Meta's answer to browser's offer)
+    const { pc, callId } = getOutboundCallState();
+    if (pc && callId === data.call_id && data.sdp_answer) {
+      pc.setRemoteDescription({ type: 'answer', sdp: data.sdp_answer }).catch(
+        err => {
+          // eslint-disable-next-line no-console
+          console.error(
+            '[WhatsApp Call] Failed to set remote SDP answer:',
+            err
+          );
+        }
+      );
+    }
+  };
+
+  // eslint-disable-next-line class-methods-use-this
+  onWhatsappCallPermissionGranted = data => {
+    emitter.emit('whatsapp_call:permission_granted', {
+      contactName: data.contact_name,
+    });
+  };
+
+  // Server-relay mode: the media server created Peer B and sent an SDP offer
+  // for the agent's browser. This fires after POST /accept or POST /reconnect.
+  // eslint-disable-next-line class-methods-use-this
+  onWhatsappCallAgentOffer = data => {
+    const whatsappCallsStore = useWhatsappCallsStore();
+    const activeCall = whatsappCallsStore.activeCall;
+
+    if (!activeCall) return;
+    // Verify this offer is for the current active call
+    if (activeCall.callId !== data.call_id && activeCall.id !== data.id) return;
+
+    handleAgentOffer(activeCall.id, data.sdp_offer, data.ice_servers)
+      .then(() => {
+        whatsappCallsStore.markActiveCallConnected();
+        whatsappCallsStore.setReconnecting(false);
+        // Emit event so the composable can start the timer
+        emitter.emit('whatsapp_call:agent_webrtc_connected');
+      })
+      .catch(err => {
+        // eslint-disable-next-line no-console
+        console.error('[WhatsApp Call] Failed to handle agent offer:', err);
+      });
+  };
 }
 
 export default {

app/javascript/dashboard/helper/voice.js

@@ -18,6 +18,10 @@ const isVoiceCallMessage = message => {
   return CONTENT_TYPES.VOICE_CALL === message?.content_type;
 };
 
+const isWhatsappCall = message => {
+  return message?.content_attributes?.data?.call_source === 'whatsapp';
+};
+
 const shouldSkipCall = (callDirection, senderId, currentUserId) => {
   return callDirection === 'outbound' && senderId !== currentUserId;
 };
@@ -36,6 +40,10 @@ function extractCallData(message) {
 export function handleVoiceCallCreated(message, currentUserId) {
   if (!isVoiceCallMessage(message)) return;
 
+  // WhatsApp calls are managed by their own store (whatsappCalls),
+  // don't add them to the Twilio calls store.
+  if (isWhatsappCall(message)) return;
+
   const { callSid, callDirection, conversationId, senderId } =
     extractCallData(message);
 
@@ -56,14 +64,18 @@ export function handleVoiceCallUpdated(commit, message, currentUserId) {
   const { callSid, status, callDirection, conversationId, senderId } =
     extractCallData(message);
 
-  const callsStore = useCallsStore();
+  // Vuex message/conversation status updates apply to all call sources
-
-  callsStore.handleCallStatusChanged({ callSid, status, conversationId });
-
   const callInfo = { conversationId, callStatus: status };
   commit(types.UPDATE_CONVERSATION_CALL_STATUS, callInfo);
   commit(types.UPDATE_MESSAGE_CALL_STATUS, callInfo);
 
+  // Twilio-specific store interactions — skip for WhatsApp calls
+  if (isWhatsappCall(message)) return;
+
+  const callsStore = useCallsStore();
+
+  callsStore.handleCallStatusChanged({ callSid, status, conversationId });
+
   const isNewCall =
     status === 'ringing' &&
     !shouldSkipCall(callDirection, senderId, currentUserId);

app/javascript/dashboard/i18n/locale/en/conversation.json

@@ -83,7 +83,13 @@
       "CALL_ENDED": "Call ended",
       "NOT_ANSWERED_YET": "Not answered yet",
       "THEY_ANSWERED": "They answered",
-      "YOU_ANSWERED": "You answered"
+      "YOU_ANSWERED": "You answered",
+      "ANSWERED_BY": "Answered by {name}",
+      "DURATION": "{duration}",
+      "ACCEPT_CALL": "Accept",
+      "JOIN_CALL": "Join",
+      "TRANSCRIPT": "Transcript",
+      "AUDIO_NOT_SUPPORTED": "Your browser does not support audio playback."
     },
     "HEADER": {
       "RESOLVE_ACTION": "Resolve",

app/javascript/dashboard/i18n/locale/en/inboxMgmt.json

@@ -790,6 +790,9 @@
       "WHATSAPP_TEMPLATES_SYNC_SUBHEADER": "Manually sync message templates from WhatsApp to update your available templates.",
       "WHATSAPP_TEMPLATES_SYNC_BUTTON": "Sync Templates",
       "WHATSAPP_TEMPLATES_SYNC_SUCCESS": "Templates sync initiated successfully. It may take a couple of minutes to update.",
+      "WHATSAPP_CALLING_TITLE": "WhatsApp Calling",
+      "WHATSAPP_CALLING_SUBHEADER": "Enable agents to make and receive WhatsApp voice calls directly from Chatwoot. Requires your WhatsApp number to be approved for the Calling API by Meta.",
+      "WHATSAPP_CALLING_LABEL": "Enable WhatsApp Calling",
       "UPDATE_PRE_CHAT_FORM_SETTINGS": "Update Pre Chat Form Settings"
     },
     "HELP_CENTER": {

app/javascript/dashboard/i18n/locale/en/index.js

@@ -37,6 +37,7 @@ import sla from './sla.json';
 import snooze from './snooze.json';
 import teamsSettings from './teamsSettings.json';
 import whatsappTemplates from './whatsappTemplates.json';
+import whatsappCall from './whatsappCall.json';
 import contentTemplates from './contentTemplates.json';
 import mfa from './mfa.json';
 import yearInReview from './yearInReview.json';
@@ -81,6 +82,7 @@ export default {
   ...snooze,
   ...teamsSettings,
   ...whatsappTemplates,
+  ...whatsappCall,
   ...contentTemplates,
   ...mfa,
   ...yearInReview,

app/javascript/dashboard/i18n/locale/en/whatsappCall.json

@@ -0,0 +1,22 @@
+{
+  "WHATSAPP_CALL": {
+    "INCOMING_WHATSAPP_CALL": "Incoming WhatsApp Call",
+    "OUTGOING_WHATSAPP_CALL": "Outgoing WhatsApp Call",
+    "ACCEPT": "Accept",
+    "REJECT": "Reject",
+    "HANG_UP": "Hang Up",
+    "MUTE": "Mute",
+    "UNMUTE": "Unmute",
+    "INITIATE_CALL": "Call via WhatsApp",
+    "CALLING": "Calling…",
+    "RINGING": "Ringing…",
+    "CALL_FAILED": "Call failed. Please try again.",
+    "PERMISSION_REQUESTED": "Call permission request sent to the contact. You can call once they approve.",
+    "PERMISSION_PENDING": "Waiting for the contact to approve the call permission request. Please try again shortly.",
+    "UNKNOWN_CALLER": "Unknown caller",
+    "MIC_DENIED": "Microphone access denied. Please allow mic access and try again.",
+    "CALL_TAKEN": "Call accepted by another agent",
+    "RECONNECTING": "Reconnecting…",
+    "PERMISSION_GRANTED": "{contactName} approved the call permission request. You can now call them."
+  }
+}

app/javascript/dashboard/routes/dashboard/Dashboard.vue

@@ -20,11 +20,16 @@ const FloatingCallWidget = defineAsyncComponent(
   () => import('dashboard/components/widgets/FloatingCallWidget.vue')
 );
 
+const WhatsappCallWidget = defineAsyncComponent(
+  () => import('dashboard/components/widgets/WhatsappCallWidget.vue')
+);
+
 import CopilotLauncher from 'dashboard/components-next/copilot/CopilotLauncher.vue';
 import CopilotContainer from 'dashboard/components/copilot/CopilotContainer.vue';
 
 import MobileSidebarLauncher from 'dashboard/components-next/sidebar/MobileSidebarLauncher.vue';
 import { useCallsStore } from 'dashboard/stores/calls';
+import { useWhatsappCallsStore } from 'dashboard/stores/whatsappCalls';
 
 export default {
   components: {
@@ -36,6 +41,7 @@ export default {
     CopilotLauncher,
     CopilotContainer,
     FloatingCallWidget,
+    WhatsappCallWidget,
     MobileSidebarLauncher,
   },
   setup() {
@@ -44,6 +50,7 @@ export default {
     const { accountId } = useAccount();
     const { width: windowWidth } = useWindowSize();
     const callsStore = useCallsStore();
+    const whatsappCallsStore = useWhatsappCallsStore();
 
     return {
       uiSettings,
@@ -53,6 +60,10 @@ export default {
       windowWidth,
       hasActiveCall: computed(() => callsStore.hasActiveCall),
       hasIncomingCall: computed(() => callsStore.hasIncomingCall),
+      hasWhatsappCall: computed(
+        () =>
+          whatsappCallsStore.hasActiveCall || whatsappCallsStore.hasIncomingCall
+      ),
     };
   },
   data() {
@@ -163,6 +174,7 @@ export default {
         />
         <CopilotContainer />
         <FloatingCallWidget v-if="hasActiveCall || hasIncomingCall" />
+        <WhatsappCallWidget v-if="hasWhatsappCall" />
       </template>
       <AddAccountModal
         :show="showCreateAccountModal"

app/javascript/dashboard/routes/dashboard/settings/inbox/settingsPage/ConfigurationPage.vue

@@ -43,6 +43,7 @@ export default {
       isSyncingTemplates: false,
       allowedDomains: '',
       isUpdatingAllowedDomains: false,
+      callingEnabled: false,
       isSettingDefaults: false,
     };
   },
@@ -83,6 +84,8 @@ export default {
         this.inbox.selected_feature_flags || []
       ).includes('allow_mobile_webview');
       this.allowedDomains = this.inbox.allowed_domains || '';
+      this.callingEnabled =
+        this.inbox.provider_config?.calling_enabled || false;
       this.$nextTick(() => {
         this.isSettingDefaults = false;
       });
@@ -171,6 +174,23 @@ export default {
         await this.$refs.whatsappReauth.requestAuthorization();
       }
     },
+    async updateCallingEnabled() {
+      try {
+        await this.$store.dispatch('inboxes/updateInbox', {
+          id: this.inbox.id,
+          formData: false,
+          channel: {
+            provider_config: {
+              ...this.inbox.provider_config,
+              calling_enabled: this.callingEnabled,
+            },
+          },
+        });
+        useAlert(this.$t('INBOX_MGMT.EDIT.API.SUCCESS_MESSAGE'));
+      } catch (error) {
+        useAlert(this.$t('INBOX_MGMT.EDIT.API.ERROR_MESSAGE'));
+      }
+    },
     async syncTemplates() {
       this.isSyncingTemplates = true;
       try {
@@ -452,6 +472,22 @@ export default {
           {{ $t('INBOX_MGMT.SETTINGS_POPUP.WHATSAPP_TEMPLATES_SYNC_BUTTON') }}
         </NextButton>
       </SettingsFieldSection>
+      <SettingsFieldSection
+        :label="$t('INBOX_MGMT.SETTINGS_POPUP.WHATSAPP_CALLING_TITLE')"
+        :help-text="$t('INBOX_MGMT.SETTINGS_POPUP.WHATSAPP_CALLING_SUBHEADER')"
+      >
+        <div class="flex gap-2 items-center">
+          <input
+            id="callingEnabled"
+            v-model="callingEnabled"
+            type="checkbox"
+            @change="updateCallingEnabled"
+          />
+          <label for="callingEnabled" class="text-body-main text-n-slate-12">
+            {{ $t('INBOX_MGMT.SETTINGS_POPUP.WHATSAPP_CALLING_LABEL') }}
+          </label>
+        </div>
+      </SettingsFieldSection>
     </div>
     <WhatsappReauthorize
       v-if="isEmbeddedSignupWhatsApp"

app/javascript/dashboard/stores/whatsappCalls.js

@@ -0,0 +1,121 @@
+import { defineStore } from 'pinia';
+
+// Module-scoped (non-reactive) state for outbound call WebRTC objects.
+// These cannot be in Pinia state because RTCPeerConnection/MediaStream are not serializable.
+// Used ONLY in legacy (browser-direct) mode. In server-relay mode outbound calls
+// go through the same inbound WebRTC path via handleAgentOffer.
+const outboundCall = { pc: null, stream: null, audio: null, callId: null };
+
+export function getOutboundCallState() {
+  return outboundCall;
+}
+
+export function setOutboundCallProperty(key, value) {
+  outboundCall[key] = value;
+}
+
+export function cleanupOutboundCall() {
+  if (outboundCall.pc) outboundCall.pc.close();
+  if (outboundCall.stream) {
+    outboundCall.stream.getTracks().forEach(t => t.stop());
+  }
+  if (outboundCall.audio) {
+    outboundCall.audio.srcObject = null;
+    outboundCall.audio.remove();
+  }
+  outboundCall.pc = null;
+  outboundCall.stream = null;
+  outboundCall.audio = null;
+  outboundCall.callId = null;
+}
+
+export const useWhatsappCallsStore = defineStore('whatsappCalls', {
+  state: () => ({
+    // Incoming ringing calls waiting for agent action
+    incomingCalls: [],
+    // The single active call (accepted + audio connected)
+    activeCall: null,
+    // Cleanup callback registered by the composable — called when a call ends externally
+    cleanupCallback: null,
+    // True while the agent is reconnecting to an active call after page reload
+    isReconnecting: false,
+    // Seconds already elapsed when reconnecting — timer resumes from this offset
+    callTimerOffset: 0,
+  }),
+
+  getters: {
+    hasIncomingCall: state => state.incomingCalls.length > 0,
+    hasActiveCall: state => state.activeCall !== null,
+    hasWhatsappCall: state =>
+      state.incomingCalls.length > 0 || state.activeCall !== null,
+    firstIncomingCall: state => state.incomingCalls[0] || null,
+
+    // Returns true when the active call is operating through the media server
+    // (server-relay mode). Detected by the absence of sdpOffer in the call data
+    // — in legacy mode the incoming call ActionCable event includes sdpOffer.
+    isMediaServerEnabled() {
+      return this.activeCall?.serverRelay === true;
+    },
+  },
+
+  actions: {
+    addIncomingCall(callData) {
+      const exists = this.incomingCalls.some(c => c.callId === callData.callId);
+      if (exists) return;
+      this.incomingCalls.push(callData);
+    },
+
+    removeIncomingCall(callId) {
+      this.incomingCalls = this.incomingCalls.filter(c => c.callId !== callId);
+    },
+
+    setActiveCall(callData) {
+      this.activeCall = callData;
+    },
+
+    clearActiveCall() {
+      this.activeCall = null;
+      this.callTimerOffset = 0;
+      this.isReconnecting = false;
+    },
+
+    markActiveCallConnected() {
+      if (this.activeCall) {
+        this.activeCall = { ...this.activeCall, status: 'connected' };
+      }
+    },
+
+    registerCleanupCallback(callback) {
+      this.cleanupCallback = callback;
+    },
+
+    setReconnecting(value) {
+      this.isReconnecting = value;
+    },
+
+    setTimerOffset(seconds) {
+      this.callTimerOffset = seconds;
+    },
+
+    handleCallAcceptedByOther(callId) {
+      this.removeIncomingCall(callId);
+    },
+
+    handleCallEnded(callId) {
+      this.removeIncomingCall(callId);
+      if (this.activeCall?.callId === callId) {
+        // Invoke cleanup BEFORE clearing activeCall so the callback can
+        // check isMediaServerEnabled (which depends on activeCall.serverRelay)
+        if (this.cleanupCallback) {
+          this.cleanupCallback();
+        }
+        this.activeCall = null;
+        this.callTimerOffset = 0;
+        this.isReconnecting = false;
+      }
+      if (outboundCall.callId === callId) {
+        cleanupOutboundCall();
+      }
+    },
+  },
+});

app/jobs/webhooks/whatsapp_events_job.rb

@@ -100,3 +100,5 @@ class Webhooks::WhatsappEventsJob < ApplicationJob
     return channel if channel && channel.provider_config['phone_number_id'] == phone_number_id
   end
 end
+
+Webhooks::WhatsappEventsJob.prepend_mod_with('Webhooks::WhatsappEventsJob')

app/services/whatsapp/facebook_api_client.rb

@@ -86,7 +86,7 @@ class Whatsapp::FacebookApiClient
       body: {
         override_callback_uri: callback_url,
         verify_token: verify_token,
-        subscribed_fields: %w[messages smb_message_echoes]
+        subscribed_fields: webhook_subscribed_fields
       }.to_json
     )
 
@@ -102,6 +102,10 @@ class Whatsapp::FacebookApiClient
     handle_response(response, 'Webhook unsubscription failed')
   end
 
+  def webhook_subscribed_fields
+    %w[messages smb_message_echoes]
+  end
+
   private
 
   def request_headers
@@ -123,3 +127,5 @@ class Whatsapp::FacebookApiClient
     response.parsed_response
   end
 end
+
+Whatsapp::FacebookApiClient.prepend_mod_with('Whatsapp::FacebookApiClient')

app/services/whatsapp/providers/whatsapp_cloud_service.rb

@@ -89,9 +89,9 @@ class Whatsapp::Providers::WhatsappCloudService < Whatsapp::Providers::BaseServi
     ENV.fetch('WHATSAPP_CLOUD_BASE_URL', 'https://graph.facebook.com')
   end
 
-  # TODO: See if we can unify the API versions and for both paths and make it consistent with out facebook app API versions
   def phone_id_path
-    "#{api_base_path}/v13.0/#{whatsapp_channel.provider_config['phone_number_id']}"
+    api_version = GlobalConfigService.load('WHATSAPP_API_VERSION', 'v22.0')
+    "#{api_base_path}/#{api_version}/#{whatsapp_channel.provider_config['phone_number_id']}"
   end
 
   def business_account_path
@@ -205,3 +205,5 @@ class Whatsapp::Providers::WhatsappCloudService < Whatsapp::Providers::BaseServi
     process_response(response, message)
   end
 end
+
+Whatsapp::Providers::WhatsappCloudService.include_mod_with('Whatsapp::Providers::WhatsappCloudService')

app/views/api/v1/models/_inbox.json.jbuilder

@@ -129,6 +129,8 @@ if resource.whatsapp?
   json.message_templates resource.channel.try(:message_templates)
   json.provider_config resource.channel.try(:provider_config) if Current.account_user&.administrator?
   json.reauthorization_required resource.channel.try(:reauthorization_required?)
+  json.calling_enabled resource.channel.try(:provider_config)&.dig('calling_enabled') || false
+  json.media_server_enabled Call.media_server_enabled?
 end
 
 ## Voice Channel Attributes

config/features.yml

@@ -104,6 +104,10 @@
   display_name: Audit Logs
   enabled: false
   premium: true
+- name: whatsapp_call
+  display_name: WhatsApp Calling
+  enabled: false
+  premium: true
 - name: custom_tools
   display_name: Custom Tools
   enabled: false

config/routes.rb

@@ -303,6 +303,23 @@ Rails.application.routes.draw do
             resource :authorization, only: [:create]
           end
 
+          resources :whatsapp_calls, only: [:show] do
+            collection do
+              get :active
+              post :initiate
+            end
+            member do
+              post :accept
+              post :reject
+              post :terminate
+              post :agent_answer
+              post :reconnect
+              post :join
+              post :play_audio
+              post :upload_recording
+            end
+          end
+
           resources :webhooks, only: [:index, :create, :update, :destroy]
           namespace :integrations do
             resources :apps, only: [:index, :show]
@@ -606,6 +623,16 @@ Rails.application.routes.draw do
   get 'instagram/callback', to: 'instagram/callbacks#show'
   get 'tiktok/callback', to: 'tiktok/callbacks#show'
   get 'notion/callback', to: 'notion/callbacks#show'
+
+  # Media server callbacks — authenticated by shared MEDIA_SERVER_AUTH_TOKEN,
+  # not a user session. Intentionally top-level and not account-scoped.
+  namespace :callbacks do
+    namespace :media_server do
+      post :agent_disconnected, to: '/media_server/callbacks#agent_disconnected'
+      post :recording_ready, to: '/media_server/callbacks#recording_ready'
+      post :session_terminated, to: '/media_server/callbacks#session_terminated'
+    end
+  end
   # ----------------------------------------------------------------------
   # Routes for external service verifications
   get '.well-known/assetlinks.json' => 'android_app#assetlinks'

db/migrate/20260421042235_add_media_server_fields_to_calls.rb

@@ -0,0 +1,7 @@
+class AddMediaServerFieldsToCalls < ActiveRecord::Migration[7.0]
+  def change
+    add_column :calls, :media_session_id, :string
+    add_index :calls, :media_session_id, unique: true
+    add_index :calls, [:accepted_by_agent_id, :status]
+  end
+end

db/schema.rb

@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[7.1].define(version: 2026_04_10_092753) do
+ActiveRecord::Schema[7.1].define(version: 2026_04_21_042235) do
   # These extensions should be enabled to support this database
   enable_extension "pg_stat_statements"
   enable_extension "pg_trgm"
@@ -279,8 +279,11 @@ ActiveRecord::Schema[7.1].define(version: 2026_04_10_092753) do
     t.text "transcript"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.string "media_session_id"
+    t.index ["accepted_by_agent_id", "status"], name: "index_calls_on_accepted_by_agent_id_and_status"
     t.index ["account_id", "contact_id"], name: "index_calls_on_account_id_and_contact_id"
     t.index ["account_id", "conversation_id"], name: "index_calls_on_account_id_and_conversation_id"
+    t.index ["media_session_id"], name: "index_calls_on_media_session_id", unique: true
     t.index ["message_id"], name: "index_calls_on_message_id"
     t.index ["provider", "provider_call_id"], name: "index_calls_on_provider_and_provider_call_id", unique: true
   end

docker-compose.production.yaml

@@ -57,7 +57,34 @@ services:
     ports:
       - '127.0.0.1:6379:6379'
 
+  # Optional: WhatsApp Calling media server (server-side WebRTC relay)
+  # Uncomment to enable call persistence across page reloads and server-side recording.
+  # media-server:
+  #   image: chatwoot/media-server:latest
+  #   env_file: .env
+  #   ports:
+  #     - '4000:4000'
+  #     - '10000-10100:10000-10100/udp'
+  #   environment:
+  #     - AUTH_TOKEN=${MEDIA_SERVER_AUTH_TOKEN}
+  #     - RAILS_CALLBACK_URL=http://rails:3000
+  #     - RECORDINGS_DIR=/recordings
+  #     - LOG_LEVEL=info
+  #     - UDP_PORT_MIN=10000
+  #     - UDP_PORT_MAX=10100
+  #     - PUBLIC_IP=${MEDIA_SERVER_PUBLIC_IP}
+  #     - STUN_SERVERS=stun:stun.l.google.com:19302
+  #   volumes:
+  #     - media_recordings:/recordings
+  #   restart: always
+  #   healthcheck:
+  #     test: ['CMD', 'wget', '--spider', '-q', 'http://localhost:4000/health']
+  #     interval: 10s
+  #     timeout: 5s
+  #     retries: 3
+
 volumes:
   storage_data:
   postgres_data:
   redis_data:
+  # media_recordings:

docker-compose.yaml

@@ -110,6 +110,32 @@ services:
       - 1025:1025
       - 8025:8025
 
+  media-server:
+    image: chatwoot/media-server:latest
+    build:
+      context: ./enterprise/media-server
+      dockerfile: Dockerfile
+    ports:
+      - '4000:4000'
+      - '10000-10100:10000-10100/udp'
+    environment:
+      - AUTH_TOKEN=${MEDIA_SERVER_AUTH_TOKEN:-}
+      - RAILS_CALLBACK_URL=http://rails:3000
+      - RECORDINGS_DIR=/recordings
+      - LOG_LEVEL=info
+      - UDP_PORT_MIN=10000
+      - UDP_PORT_MAX=10100
+      - PUBLIC_IP=${MEDIA_SERVER_PUBLIC_IP:-}
+      - STUN_SERVERS=stun:stun.l.google.com:19302
+    volumes:
+      - media_recordings:/recordings
+    restart: unless-stopped
+    healthcheck:
+      test: ['CMD', 'wget', '--spider', '-q', 'http://localhost:4000/health']
+      interval: 10s
+      timeout: 5s
+      retries: 3
+
 volumes:
   postgres:
   redis:
@@ -117,3 +143,4 @@ volumes:
   node_modules:
   cache:
   bundle:
+  media_recordings:

enterprise/app/builders/enterprise/messages/message_builder.rb

@@ -1,8 +1,10 @@
 module Enterprise::Messages::MessageBuilder
   private
 
+  INCOMING_ALLOWED_CHANNEL_TYPES = %w[Channel::Voice Channel::Whatsapp].freeze
+
   def message_type
-    return @message_type if @message_type == 'incoming' && @conversation.inbox.channel_type == 'Channel::Voice'
+    return @message_type if @message_type == 'incoming' && INCOMING_ALLOWED_CHANNEL_TYPES.include?(@conversation.inbox.channel_type)
 
     super
   end

enterprise/app/controllers/api/v1/accounts/whatsapp_calls_controller.rb

@@ -0,0 +1,273 @@
+class Api::V1::Accounts::WhatsappCallsController < Api::V1::Accounts::BaseController
+  ALLOWED_PEER_ROLES = %w[listen_only participant].freeze
+  ALLOWED_AUDIO_MODES = %w[replace mix].freeze
+
+  before_action :ensure_whatsapp_call_enabled
+  before_action :set_call, only: [:show, :accept, :reject, :terminate, :upload_recording, :agent_answer, :reconnect, :join, :play_audio]
+
+  def show
+    render json: {
+      id: @call.id,
+      call_id: @call.provider_call_id,
+      status: @call.status,
+      direction: @call.direction_label,
+      conversation_id: @call.conversation_id,
+      inbox_id: @call.inbox_id,
+      message_id: @call.message_id,
+      # In server-relay mode the browser must not talk WebRTC to Meta directly
+      # — the media server owns that peer connection. Omitting sdp_offer forces
+      # the FE's isServerRelayCall() check to return true so the accept flow
+      # waits for the whatsapp_call.agent_offer broadcast from Rails instead of
+      # negotiating straight with Meta.
+      sdp_offer: @call.ringing? && !Call.media_server_enabled? ? @call.sdp_offer : nil,
+      ice_servers: Call.media_server_enabled? ? [] : @call.ice_servers,
+      media_server_enabled: Call.media_server_enabled?,
+      caller: caller_info
+    }
+  end
+
+  def accept
+    if Call.media_server_enabled?
+      call = Whatsapp::CallService.new(call: @call, agent: current_user).accept
+      render json: { id: call.id, status: call.status, message_id: call.message_id, media_session_id: call.media_session_id }
+    else
+      sdp_answer = params[:sdp_answer]
+      return render json: { error: 'sdp_answer is required' }, status: :unprocessable_entity if sdp_answer.blank?
+
+      call = Whatsapp::CallService.new(call: @call, agent: current_user).pre_accept_and_accept(sdp_answer)
+      render json: { id: call.id, status: call.status, message_id: call.message_id }
+    end
+  rescue Whatsapp::CallErrors::NotRinging, Whatsapp::CallErrors::AlreadyAccepted => e
+    render json: { error: e.message }, status: :unprocessable_entity
+  rescue StandardError => e
+    Rails.logger.error "[WHATSAPP CALL] accept failed: #{e.message}"
+    render json: { error: 'Failed to accept call' }, status: :internal_server_error
+  end
+
+  def reject
+    call = Whatsapp::CallService.new(call: @call, agent: current_user).reject
+    render json: { id: call.id, status: call.status }
+  rescue StandardError => e
+    Rails.logger.error "[WHATSAPP CALL] reject failed: #{e.message}"
+    render json: { error: 'Failed to reject call' }, status: :internal_server_error
+  end
+
+  def terminate
+    call = Whatsapp::CallService.new(call: @call, agent: current_user).terminate
+    render json: { id: call.id, status: call.status }
+  rescue StandardError => e
+    Rails.logger.error "[WHATSAPP CALL] terminate failed: #{e.message}"
+    render json: { error: 'Failed to terminate call' }, status: :internal_server_error
+  end
+
+  def upload_recording
+    return render json: { error: 'No recording file provided' }, status: :unprocessable_entity if params[:recording].blank?
+    return render json: { error: 'Call is not ended' }, status: :unprocessable_entity unless @call.terminal?
+
+    attach_recording_and_enqueue_transcription
+    render json: { id: @call.id, status: 'uploaded' }
+  rescue StandardError => e
+    Rails.logger.error "[WHATSAPP CALL] upload_recording failed: #{e.message}"
+    render json: { error: 'Failed to upload recording' }, status: :internal_server_error
+  end
+
+  def active
+    call = current_account.calls.whatsapp.active_for_agent(current_user.id).last
+    if call
+      elapsed = call.started_at ? (Time.current - call.started_at).to_i : 0
+      render json: {
+        id: call.id,
+        call_id: call.provider_call_id,
+        conversation_id: call.conversation_id,
+        status: call.status,
+        elapsed_seconds: elapsed,
+        media_session_id: call.media_session_id
+      }
+    else
+      render json: { call: nil }
+    end
+  end
+
+  def agent_answer
+    return render json: { error: 'sdp_answer is required' }, status: :unprocessable_entity if params[:sdp_answer].blank?
+    return render json: { error: 'No media session' }, status: :unprocessable_entity if @call.media_session_id.blank?
+
+    client = Whatsapp::MediaServerClient.new
+    client.set_agent_answer(@call.media_session_id, sdp_answer: params[:sdp_answer])
+    render json: { success: true }
+  rescue Whatsapp::MediaServerClient::SessionError, Whatsapp::MediaServerClient::ConnectionError => e
+    Rails.logger.error "[WHATSAPP CALL] agent_answer failed: #{e.message}"
+    render json: { error: 'Failed to set agent answer' }, status: :internal_server_error
+  end
+
+  def reconnect
+    return render json: { error: 'No media session' }, status: :unprocessable_entity if @call.media_session_id.blank?
+    return render json: { error: 'Call is not in progress' }, status: :unprocessable_entity unless @call.in_progress?
+
+    client = Whatsapp::MediaServerClient.new
+    response = client.reconnect_agent(@call.media_session_id)
+    render json: {
+      sdp_offer: response['sdp_offer'],
+      ice_servers: response['ice_servers']
+    }
+  rescue Whatsapp::MediaServerClient::SessionError, Whatsapp::MediaServerClient::ConnectionError => e
+    Rails.logger.error "[WHATSAPP CALL] reconnect failed: #{e.message}"
+    render json: { error: 'Failed to reconnect' }, status: :internal_server_error
+  end
+
+  def join
+    return render json: { error: 'No media session' }, status: :unprocessable_entity if @call.media_session_id.blank?
+
+    role = params[:role] || 'listen_only'
+    return render json: { error: 'Invalid role' }, status: :unprocessable_entity unless ALLOWED_PEER_ROLES.include?(role)
+
+    client = Whatsapp::MediaServerClient.new
+    response = client.add_peer(@call.media_session_id, role: role, label: current_user.name)
+    render json: {
+      peer_id: response['peer_id'],
+      sdp_offer: response['sdp_offer'],
+      ice_servers: response['ice_servers']
+    }
+  rescue Whatsapp::MediaServerClient::SessionError, Whatsapp::MediaServerClient::ConnectionError => e
+    Rails.logger.error "[WHATSAPP CALL] join failed: #{e.message}"
+    render json: { error: 'Failed to join call' }, status: :internal_server_error
+  end
+
+  def play_audio
+    return render json: { error: 'No media session' }, status: :unprocessable_entity if @call.media_session_id.blank?
+    return render json: { error: 'file_path is required' }, status: :unprocessable_entity if params[:file_path].blank?
+    return render json: { error: 'Invalid file_path' }, status: :unprocessable_entity if params[:file_path].include?('..')
+
+    mode = params[:mode] || 'replace'
+    return render json: { error: 'Invalid mode' }, status: :unprocessable_entity unless ALLOWED_AUDIO_MODES.include?(mode)
+
+    client = Whatsapp::MediaServerClient.new
+    response = client.inject_audio(
+      @call.media_session_id,
+      file_path: params[:file_path],
+      mode: mode,
+      loop: ActiveModel::Type::Boolean.new.cast(params[:loop])
+    )
+    render json: { injection_id: response['injection_id'] }
+  rescue Whatsapp::MediaServerClient::SessionError, Whatsapp::MediaServerClient::ConnectionError => e
+    Rails.logger.error "[WHATSAPP CALL] play_audio failed: #{e.message}"
+    render json: { error: 'Failed to play audio' }, status: :internal_server_error
+  end
+
+  def initiate
+    conversation = current_account.conversations.find_by!(display_id: params[:conversation_id])
+    authorize conversation, :show?
+    error = validate_whatsapp_calling(conversation)
+    return render json: { error: error }, status: :unprocessable_entity if error
+
+    call = create_outbound_call(conversation)
+    message = Whatsapp::CallMessageBuilder.create!(conversation: conversation, call: call, user: current_user)
+    call.update!(message_id: message.id)
+    render json: { status: 'calling', call_id: call.provider_call_id, id: call.id, message_id: message.id }
+  rescue Whatsapp::CallErrors::NoCallPermission
+    handle_no_call_permission(conversation)
+  rescue ActiveRecord::RecordNotFound
+    render json: { error: 'Conversation not found' }, status: :not_found
+  rescue StandardError => e
+    Rails.logger.error "[WHATSAPP CALL] initiate failed: #{e.message}"
+    render json: { error: e.message }, status: :unprocessable_entity
+  end
+
+  private
+
+  def create_outbound_call(conversation)
+    contact_phone = conversation.contact&.phone_number
+    raise ArgumentError, 'Contact phone number not available' if contact_phone.blank?
+    raise ArgumentError, 'sdp_offer is required' if params[:sdp_offer].blank? && !Call.media_server_enabled?
+
+    if Call.media_server_enabled?
+      create_outbound_call_via_media_server(conversation, contact_phone)
+    else
+      create_outbound_call_direct(conversation, contact_phone)
+    end
+  end
+
+  def create_outbound_call_direct(conversation, contact_phone)
+    result = conversation.inbox.channel.provider_service.initiate_call(contact_phone.delete('+'), params[:sdp_offer])
+    provider_call_id = result.dig('calls', 0, 'id') || result['call_id']
+
+    current_account.calls.create!(
+      provider: :whatsapp,
+      inbox: conversation.inbox, conversation: conversation, contact: conversation.contact,
+      provider_call_id: provider_call_id, direction: :outgoing, status: 'ringing',
+      meta: { sdp_offer: params[:sdp_offer] }
+    )
+  end
+
+  def create_outbound_call_via_media_server(conversation, contact_phone)
+    client = Whatsapp::MediaServerClient.new
+
+    # Step 1: Create session on media server (generates SDP offer for Meta)
+    session_response = client.create_session(
+      call_id: "pending_#{SecureRandom.hex(8)}",
+      direction: 'outgoing',
+      sdp_offer: nil,
+      ice_servers: [{ urls: ['stun:stun.l.google.com:19302'] }],
+      account_id: current_account.id
+    )
+
+    # Step 2: Send the media server's SDP offer to Meta to initiate the call
+    sdp_offer = session_response['meta_sdp_offer']
+    result = conversation.inbox.channel.provider_service.initiate_call(contact_phone.delete('+'), sdp_offer)
+    provider_call_id = result.dig('calls', 0, 'id') || result['call_id']
+
+    current_account.calls.create!(
+      provider: :whatsapp,
+      inbox: conversation.inbox, conversation: conversation, contact: conversation.contact,
+      provider_call_id: provider_call_id, direction: :outgoing, status: 'ringing',
+      media_session_id: session_response['session_id'],
+      meta: { sdp_offer: sdp_offer }
+    )
+  end
+
+  def handle_no_call_permission(conversation)
+    last_requested = conversation.additional_attributes&.dig('call_permission_requested_at')
+
+    return render json: { status: 'permission_pending' } if last_requested.present? && Time.zone.parse(last_requested) > 5.minutes.ago
+
+    contact_phone = conversation.contact.phone_number.delete('+')
+    result = conversation.inbox.channel.provider_service.send_call_permission_request(contact_phone)
+    return render json: { error: 'Failed to send call permission request' }, status: :unprocessable_entity unless result
+
+    attrs = (conversation.additional_attributes || {}).merge('call_permission_requested_at' => Time.current.iso8601)
+    conversation.update!(additional_attributes: attrs)
+    render json: { status: 'permission_requested' }
+  end
+
+  def validate_whatsapp_calling(conversation)
+    channel = conversation.inbox.channel
+    return 'Calling is only supported on WhatsApp Cloud inboxes' unless channel.is_a?(Channel::Whatsapp) && channel.provider == 'whatsapp_cloud'
+    return 'Calling is not enabled for this inbox' unless channel.provider_config['calling_enabled']
+
+    nil
+  end
+
+  def ensure_whatsapp_call_enabled
+    render_payment_required('WhatsApp calling is not enabled for this account') unless current_account.feature_enabled?('whatsapp_call')
+  end
+
+  def set_call
+    @call = current_account.calls.whatsapp.find(params[:id])
+    authorize @call.conversation, :show?
+  rescue ActiveRecord::RecordNotFound
+    render json: { error: 'Call not found' }, status: :not_found
+  end
+
+  def attach_recording_and_enqueue_transcription
+    @call.recording.attach(params[:recording])
+    Whatsapp::CallMessageBuilder.update_recording_url!(call: @call)
+    Whatsapp::CallTranscriptionJob.perform_later(@call.id)
+  end
+
+  def caller_info
+    contact = @call.conversation&.contact
+    return {} unless contact
+
+    { name: contact.name, phone: contact.phone_number, avatar: contact.avatar_url }
+  end
+end

enterprise/app/controllers/media_server/callbacks_controller.rb

@@ -0,0 +1,63 @@
+class MediaServer::CallbacksController < ApplicationController
+  before_action :validate_media_server_token
+
+  def agent_disconnected
+    call = find_call_by_session
+    return head :not_found unless call
+
+    ActionCable.server.broadcast(
+      "account_#{call.account_id}",
+      {
+        event: 'whatsapp_call.agent_disconnected',
+        data: { id: call.id, call_id: call.provider_call_id, conversation_id: call.conversation_id }
+      }
+    )
+    head :ok
+  end
+
+  def recording_ready
+    call = find_call_by_session
+    return head :not_found unless call
+
+    Whatsapp::CallRecordingFetchJob.perform_later(call.id)
+    head :ok
+  end
+
+  def session_terminated
+    call = find_call_by_session
+    return head :not_found unless call
+    return head :ok if call.terminal?
+
+    reason = params[:reason] || 'media_server'
+    was_answered = call.in_progress? || call.accepted_by_agent_id.present?
+    final_status = was_answered ? 'completed' : 'failed'
+
+    call.update!(status: final_status, end_reason: reason)
+
+    ActionCable.server.broadcast(
+      "account_#{call.account_id}",
+      {
+        event: 'whatsapp_call.ended',
+        data: { id: call.id, call_id: call.provider_call_id, status: final_status, conversation_id: call.conversation_id }
+      }
+    )
+
+    call.inbox.channel.provider_service.terminate_call(call.provider_call_id)
+  rescue StandardError => e
+    Rails.logger.error "[MEDIA SERVER] Failed to terminate on provider: #{e.message}"
+  ensure
+    head :ok unless performed?
+  end
+
+  private
+
+  def validate_media_server_token
+    token = request.headers['Authorization']&.sub('Bearer ', '')
+    expected = ENV.fetch('MEDIA_SERVER_AUTH_TOKEN', '')
+    head :unauthorized unless expected.present? && token.present? && ActiveSupport::SecurityUtils.secure_compare(token, expected)
+  end
+
+  def find_call_by_session
+    Call.find_by(media_session_id: params[:session_id])
+  end
+end

enterprise/app/jobs/enterprise/webhooks/whatsapp_events_job.rb

@@ -0,0 +1,41 @@
+module Enterprise::Webhooks::WhatsappEventsJob
+  def handle_message_events(channel, params)
+    if call_event?(params)
+      handle_call_events(channel, params)
+      return
+    end
+
+    if call_permission_reply?(params)
+      handle_call_permission_reply(channel, params)
+      return
+    end
+
+    super
+  end
+
+  private
+
+  def call_event?(params)
+    params.dig(:entry, 0, :changes, 0, :field) == 'calls'
+  end
+
+  def call_permission_reply?(params)
+    message = params.dig(:entry, 0, :changes, 0, :value, :messages, 0)
+    message&.dig(:type) == 'interactive' && message&.dig(:interactive, :type) == 'call_permission_reply'
+  end
+
+  def handle_call_events(channel, params)
+    Whatsapp::IncomingCallService.new(
+      inbox: channel.inbox,
+      params: extract_call_params(params)
+    ).perform
+  end
+
+  def handle_call_permission_reply(channel, params)
+    Whatsapp::CallPermissionReplyService.new(inbox: channel.inbox, params: params).perform
+  end
+
+  def extract_call_params(params)
+    params.dig(:entry, 0, :changes, 0, :value) || {}
+  end
+end

enterprise/app/jobs/whatsapp/call_cleanup_job.rb

@@ -0,0 +1,33 @@
+class Whatsapp::CallCleanupJob < ApplicationJob
+  queue_as :low
+
+  def perform
+    expire_stale_ringing_calls
+    expire_stale_in_progress_calls
+  end
+
+  private
+
+  def expire_stale_ringing_calls
+    Call.whatsapp.ringing.where('created_at < ?', 2.minutes.ago).find_each do |call|
+      call.update!(status: 'no_answer', end_reason: 'timeout')
+      Whatsapp::CallMessageBuilder.update_status!(call: call, status: 'no_answer')
+    end
+  end
+
+  def expire_stale_in_progress_calls
+    Call.whatsapp.where(status: 'in_progress').where('started_at < ?', 3.hours.ago).find_each do |call|
+      terminate_media_session(call)
+      call.update!(status: 'failed', end_reason: 'timeout')
+      Whatsapp::CallMessageBuilder.update_status!(call: call, status: 'failed')
+    end
+  end
+
+  def terminate_media_session(call)
+    return unless call.media_session_id.present?
+
+    Whatsapp::MediaServerClient.new.terminate_session(call.media_session_id)
+  rescue Whatsapp::MediaServerClient::ConnectionError, Whatsapp::MediaServerClient::SessionError => e
+    Rails.logger.error "[WHATSAPP CALL CLEANUP] Failed to terminate media session #{call.media_session_id}: #{e.message}"
+  end
+end

enterprise/app/jobs/whatsapp/call_recording_fetch_job.rb

@@ -0,0 +1,42 @@
+class Whatsapp::CallRecordingFetchJob < ApplicationJob
+  queue_as :default
+
+  retry_on Whatsapp::MediaServerClient::ConnectionError, wait: 5.seconds, attempts: 5
+  discard_on ActiveRecord::RecordNotFound
+
+  def perform(call_id)
+    call = Call.find(call_id)
+    return unless call.media_session_id.present?
+
+    client = Whatsapp::MediaServerClient.new
+
+    # combined.ogg is only produced when the media-server session terminates
+    # (ffmpeg mix runs in Recorder.Finalize). For calls that ended via Meta's
+    # terminate webhook — not agent hang-up — nothing has triggered Finalize
+    # yet. Call terminate first; it's idempotent on the server and returns
+    # only after Finalize has written combined.ogg.
+    safe_terminate(client, call.media_session_id)
+
+    recording_data = client.download_recording(call.media_session_id)
+    return if recording_data.blank?
+
+    call.recording.attach(
+      io: StringIO.new(recording_data.force_encoding('BINARY')),
+      filename: "call_#{call.id}_#{call.provider_call_id}.ogg",
+      content_type: 'audio/ogg'
+    )
+
+    Whatsapp::CallMessageBuilder.update_recording_url!(call: call)
+    Whatsapp::CallTranscriptionJob.perform_later(call.id) if call.recording.attached?
+  rescue Whatsapp::MediaServerClient::SessionError => e
+    Rails.logger.warn "[WHATSAPP CALL] Recording not available for session #{call.media_session_id}: #{e.message}"
+  end
+
+  private
+
+  def safe_terminate(client, session_id)
+    client.terminate_session(session_id)
+  rescue Whatsapp::MediaServerClient::SessionError, Whatsapp::MediaServerClient::ConnectionError => e
+    Rails.logger.info "[WHATSAPP CALL] terminate_session during fetch (#{session_id}): #{e.message}"
+  end
+end

enterprise/app/jobs/whatsapp/call_transcription_job.rb

@@ -0,0 +1,15 @@
+class Whatsapp::CallTranscriptionJob < ApplicationJob
+  queue_as :low
+
+  retry_on ActiveStorage::FileNotFoundError, wait: 2.seconds, attempts: 3
+  discard_on Faraday::BadRequestError do |job, error|
+    Rails.logger.warn("[WHATSAPP CALL] Discarding transcription job: call_id=#{job.arguments.first}, status=#{error.response&.dig(:status)}")
+  end
+
+  def perform(call_id)
+    call = Call.whatsapp.find_by(id: call_id)
+    return if call.blank? || !call.recording.attached?
+
+    Whatsapp::CallTranscriptionService.new(call).perform
+  end
+end

enterprise/app/models/call.rb

@@ -6,6 +6,7 @@
 #  direction            :integer          not null
 #  duration_seconds     :integer
 #  end_reason           :string
+#  media_session_id     :string
 #  meta                 :jsonb
 #  provider             :integer          default("twilio"), not null
 #  started_at           :datetime
@@ -23,10 +24,12 @@
 #
 # Indexes
 #
-#  index_calls_on_account_id_and_contact_id       (account_id,contact_id)
+#  index_calls_on_accepted_by_agent_id_and_status  (accepted_by_agent_id,status)
-#  index_calls_on_account_id_and_conversation_id  (account_id,conversation_id)
+#  index_calls_on_account_id_and_contact_id        (account_id,contact_id)
-#  index_calls_on_message_id                      (message_id)
+#  index_calls_on_account_id_and_conversation_id   (account_id,conversation_id)
-#  index_calls_on_provider_and_provider_call_id   (provider,provider_call_id) UNIQUE
+#  index_calls_on_media_session_id                 (media_session_id) UNIQUE
+#  index_calls_on_message_id                       (message_id)
+#  index_calls_on_provider_and_provider_call_id    (provider,provider_call_id) UNIQUE
 #
 class Call < ApplicationRecord
   # All valid call statuses
@@ -52,4 +55,41 @@ class Call < ApplicationRecord
   validates :status, presence: true, inclusion: { in: STATUSES }
 
   scope :active, -> { where.not(status: TERMINAL_STATUSES) }
+  scope :ringing, -> { where(status: 'ringing') }
+  scope :active_for_agent, ->(agent_id) { active.where(accepted_by_agent_id: agent_id) }
+
+  def self.media_server_enabled?
+    ENV['MEDIA_SERVER_URL'].present?
+  end
+
+  def ringing?
+    status == 'ringing'
+  end
+
+  def in_progress?
+    status == 'in_progress'
+  end
+
+  def terminal?
+    TERMINAL_STATUSES.include?(status)
+  end
+
+  # Frontend-facing direction label: incoming→inbound, outgoing→outbound
+  def direction_label
+    incoming? ? 'inbound' : 'outbound'
+  end
+
+  def sdp_offer
+    meta&.dig('sdp_offer')
+  end
+
+  def ice_servers
+    meta&.dig('ice_servers') || []
+  end
+
+  def recording_url
+    return unless recording.attached?
+
+    Rails.application.routes.url_helpers.rails_blob_path(recording, only_path: true)
+  end
 end

enterprise/app/services/enterprise/whatsapp/facebook_api_client.rb

@@ -0,0 +1,5 @@
+module Enterprise::Whatsapp::FacebookApiClient
+  def webhook_subscribed_fields
+    super + %w[calls]
+  end
+end

enterprise/app/services/enterprise/whatsapp/providers/whatsapp_cloud_service.rb

@@ -0,0 +1,7 @@
+module Enterprise::Whatsapp::Providers::WhatsappCloudService
+  extend ActiveSupport::Concern
+
+  included do
+    include ::Whatsapp::Providers::WhatsappCloudCallMethods
+  end
+end

enterprise/app/services/whatsapp/call_message_builder.rb

@@ -0,0 +1,108 @@
+class Whatsapp::CallMessageBuilder
+  # Maps Call model statuses to voice_call display statuses (hyphenated for frontend)
+  CALL_TO_VOICE_STATUS = {
+    'ringing' => 'ringing',
+    'in_progress' => 'in-progress',
+    'failed' => 'failed',
+    'no_answer' => 'no-answer',
+    'completed' => 'completed'
+  }.freeze
+
+  def self.create!(conversation:, call:, user: nil)
+    new(conversation: conversation, call: call, user: user).create!
+  end
+
+  def self.update_status!(call:, status: nil, agent: nil, duration_seconds: nil)
+    new(conversation: call.conversation, call: call).update_status!(
+      status: status, agent: agent, duration_seconds: duration_seconds
+    )
+  end
+
+  def self.update_recording_url!(call:)
+    message = call.message
+    return unless message
+
+    data = (message.content_attributes || {}).dup
+    data['data'] ||= {}
+    data['data']['recording_url'] = call.recording_url
+    message.update!(content_attributes: data)
+  end
+
+  def initialize(conversation:, call:, user: nil)
+    @conversation = conversation
+    @call = call
+    @user = user
+  end
+
+  def create!
+    params = {
+      content: 'WhatsApp Call',
+      message_type: message_type,
+      content_type: 'voice_call',
+      content_attributes: { 'data' => build_data_payload }
+    }
+
+    Messages::MessageBuilder.new(sender, conversation, params).perform
+  end
+
+  def update_status!(status:, agent: nil, duration_seconds: nil)
+    message = call.message
+    return unless message
+
+    data = (message.content_attributes || {}).dup
+    data['data'] ||= {}
+    data['data']['status'] = map_status(status) if status
+    data['data']['accepted_by'] = { 'id' => agent.id, 'name' => agent.name } if agent
+    data['data']['duration_seconds'] = duration_seconds if duration_seconds
+
+    message.update!(content_attributes: data)
+    message
+  end
+
+  private
+
+  attr_reader :conversation, :call, :user
+
+  def build_data_payload
+    {
+      'call_sid' => call.provider_call_id,
+      'status' => map_status(call.status),
+      'call_direction' => call.direction_label,
+      'call_source' => 'whatsapp',
+      'call_id' => call.id,
+      'from_number' => from_number,
+      'to_number' => to_number,
+      'meta' => { 'created_at' => Time.zone.now.to_i }
+    }
+  end
+
+  def message_type
+    call.outgoing? ? 'outgoing' : 'incoming'
+  end
+
+  def sender
+    return user if call.outgoing? && user
+
+    conversation.contact
+  end
+
+  def from_number
+    if call.incoming?
+      conversation.contact&.phone_number
+    else
+      conversation.inbox.channel&.phone_number
+    end
+  end
+
+  def to_number
+    if call.incoming?
+      conversation.inbox.channel&.phone_number
+    else
+      conversation.contact&.phone_number
+    end
+  end
+
+  def map_status(status)
+    CALL_TO_VOICE_STATUS[status] || status
+  end
+end

enterprise/app/services/whatsapp/call_permission_reply_service.rb

@@ -0,0 +1,61 @@
+class Whatsapp::CallPermissionReplyService
+  pattr_initialize [:inbox!, :params!]
+
+  def perform
+    return unless inbox.account.feature_enabled?('whatsapp_call')
+
+    reply_data = extract_reply_data
+    return unless reply_data&.dig(:accepted)
+
+    contact = find_contact(reply_data[:from_number])
+    return unless contact
+
+    conversation = find_active_conversation(contact)
+    return unless conversation
+
+    clear_permission_flag(conversation)
+    broadcast_permission_granted(contact, conversation)
+  end
+
+  private
+
+  def extract_reply_data
+    value = params.dig(:entry, 0, :changes, 0, :value)
+    message = value&.dig(:messages, 0)
+    reply = message&.dig(:interactive, :call_permission_reply)
+    return unless reply
+
+    accepted = reply[:response] == 'accept'
+    Rails.logger.info "[WHATSAPP CALL] call_permission_reply from=#{message[:from]} accepted=#{accepted} permanent=#{reply[:is_permanent]}"
+
+    { from_number: message[:from], accepted: accepted }
+  end
+
+  def find_contact(from_number)
+    inbox.contact_inboxes.joins(:contact)
+         .where(contacts: { phone_number: "+#{from_number}" })
+         .first&.contact
+  end
+
+  def find_active_conversation(contact)
+    inbox.conversations.where(contact: contact).where.not(status: :resolved).last
+  end
+
+  def clear_permission_flag(conversation)
+    attrs = conversation.additional_attributes || {}
+    attrs.delete('call_permission_requested_at')
+    conversation.update!(additional_attributes: attrs)
+  end
+
+  def broadcast_permission_granted(contact, conversation)
+    ActionCable.server.broadcast("account_#{inbox.account_id}", {
+                                   event: 'whatsapp_call.permission_granted',
+                                   data: {
+                                     account_id: inbox.account_id,
+                                     conversation_id: conversation.id,
+                                     contact_name: contact.name,
+                                     contact_phone: contact.phone_number
+                                   }
+                                 })
+  end
+end

enterprise/app/services/whatsapp/call_service.rb

@@ -0,0 +1,194 @@
+class Whatsapp::CallService
+  pattr_initialize [:call!, :agent!]
+
+  def accept(params = {})
+    if media_server_enabled?
+      accept_via_media_server
+    else
+      pre_accept_and_accept(params[:sdp_answer])
+    end
+  end
+
+  def pre_accept_and_accept(sdp_answer)
+    call.with_lock do
+      ensure_ringing!
+      ensure_not_already_taken!
+
+      provider = call.inbox.channel.provider_service
+      fixed_sdp = fix_sdp_setup(sdp_answer)
+
+      # Step 1: pre_accept (with SDP answer - required by Meta)
+      pre_response = provider.pre_accept_call(call.provider_call_id, fixed_sdp)
+      raise Whatsapp::CallErrors::NotRinging, 'Meta pre_accept failed' unless pre_response
+
+      # Step 2: accept with same SDP answer
+      accept_response = provider.accept_call(call.provider_call_id, fixed_sdp)
+      raise Whatsapp::CallErrors::NotRinging, 'Meta accept failed' unless accept_response
+
+      call.update!(
+        status: 'in_progress',
+        accepted_by_agent_id: agent.id,
+        started_at: Time.current
+      )
+    end
+
+    Whatsapp::CallMessageBuilder.update_status!(call: call, status: 'in_progress', agent: agent)
+    update_conversation_call_status('in-progress')
+    broadcast_accepted
+    call
+  end
+
+  def reject
+    call.reload
+    return call if call.terminal? || call.in_progress?
+
+    provider = call.inbox.channel.provider_service
+    success = provider.reject_call(call.provider_call_id)
+    Rails.logger.error "[WHATSAPP CALL] reject_call API returned false for call #{call.provider_call_id}" unless success
+
+    call.update!(status: 'failed')
+    Whatsapp::CallMessageBuilder.update_status!(call: call, status: 'failed')
+    update_conversation_call_status('failed')
+    broadcast_call_ended
+    call
+  end
+
+  def terminate
+    return call if call.terminal?
+
+    terminate_media_session if call.media_session_id.present?
+    terminate_on_provider
+
+    call.update!(status: 'completed')
+    Whatsapp::CallMessageBuilder.update_status!(call: call, status: 'completed')
+    update_conversation_call_status('completed')
+    broadcast_call_ended
+    call
+  end
+
+  def terminate_on_provider
+    provider = call.inbox.channel.provider_service
+    success = provider.terminate_call(call.provider_call_id)
+    Rails.logger.error "[WHATSAPP CALL] terminate_call API returned false for call #{call.provider_call_id}" unless success
+  end
+
+  private
+
+  def accept_via_media_server
+    agent_offer = nil
+
+    call.with_lock do
+      ensure_ringing!
+      ensure_not_already_taken!
+
+      client = Whatsapp::MediaServerClient.new
+
+      # Step 1: Create session on Go server with Meta's SDP
+      session_response = client.create_session(
+        call_id: call.provider_call_id,
+        direction: 'incoming',
+        sdp_offer: call.sdp_offer,
+        ice_servers: call.ice_servers,
+        account_id: call.account_id
+      )
+
+      # Step 2: Send Go-generated SDP answer to Meta
+      provider = call.inbox.channel.provider_service
+      pre_response = provider.pre_accept_call(call.provider_call_id, session_response['meta_sdp_answer'])
+      raise Whatsapp::CallErrors::NotRinging, 'Meta pre_accept failed' unless pre_response
+
+      accept_response = provider.accept_call(call.provider_call_id, session_response['meta_sdp_answer'])
+      raise Whatsapp::CallErrors::NotRinging, 'Meta accept failed' unless accept_response
+
+      # Step 3: Generate agent offer (Peer B)
+      agent_offer = client.generate_agent_offer(session_response['session_id'])
+
+      # Step 4: Update call record
+      call.update!(
+        status: 'in_progress',
+        accepted_by_agent_id: agent.id,
+        started_at: Time.current,
+        media_session_id: session_response['session_id']
+      )
+    end
+
+    # Step 5: Broadcast events (outside lock)
+    Whatsapp::CallMessageBuilder.update_status!(call: call, status: 'in_progress', agent: agent)
+    update_conversation_call_status('in-progress')
+    broadcast_agent_offer(agent_offer)
+    broadcast_accepted
+    call
+  end
+
+  def media_server_enabled?
+    Call.media_server_enabled?
+  end
+
+  def terminate_media_session
+    Whatsapp::MediaServerClient.new.terminate_session(call.media_session_id)
+  rescue Whatsapp::MediaServerClient::ConnectionError, Whatsapp::MediaServerClient::SessionError => e
+    Rails.logger.error "[WHATSAPP CALL] Failed to terminate media session: #{e.message}"
+  end
+
+  def ensure_ringing!
+    raise Whatsapp::CallErrors::NotRinging, 'Call is not in ringing state' unless call.ringing?
+  end
+
+  def ensure_not_already_taken!
+    raise Whatsapp::CallErrors::AlreadyAccepted, 'Call already accepted by another agent' if call.in_progress?
+  end
+
+  def fix_sdp_setup(sdp)
+    sdp.gsub('a=setup:actpass', 'a=setup:active')
+  end
+
+  def update_conversation_call_status(mapped_status)
+    conversation = call.conversation
+    attrs = (conversation.additional_attributes || {}).merge('call_status' => mapped_status)
+    conversation.update!(additional_attributes: attrs)
+  end
+
+  def broadcast_accepted
+    payload = {
+      event: 'whatsapp_call.accepted',
+      data: {
+        account_id: call.account_id,
+        id: call.id,
+        call_id: call.provider_call_id,
+        accepted_by_agent_id: agent.id,
+        conversation_id: call.conversation_id
+      }
+    }
+    ActionCable.server.broadcast("account_#{call.account_id}", payload)
+  end
+
+  def broadcast_agent_offer(agent_offer)
+    payload = {
+      event: 'whatsapp_call.agent_offer',
+      data: {
+        account_id: call.account_id,
+        id: call.id,
+        call_id: call.provider_call_id,
+        conversation_id: call.conversation_id,
+        accepted_by_agent_id: agent.id,
+        sdp_offer: agent_offer['sdp_offer'],
+        ice_servers: agent_offer['ice_servers']
+      }
+    }
+    ActionCable.server.broadcast("account_#{call.account_id}", payload)
+  end
+
+  def broadcast_call_ended
+    payload = {
+      event: 'whatsapp_call.ended',
+      data: {
+        account_id: call.account_id,
+        id: call.id,
+        call_id: call.provider_call_id,
+        status: call.status,
+        conversation_id: call.conversation_id
+      }
+    }
+    ActionCable.server.broadcast("account_#{call.account_id}", payload)
+  end
+end

enterprise/app/services/whatsapp/call_transcription_service.rb

@@ -0,0 +1,142 @@
+class Whatsapp::CallTranscriptionService < Llm::LegacyBaseOpenAiService
+  WHISPER_MODEL = 'whisper-1'.freeze
+
+  attr_reader :call, :account
+
+  def initialize(call)
+    super()
+    @call = call
+    @account = call.account
+  end
+
+  def perform
+    return { error: 'Transcription not available' } unless can_transcribe?
+    return { error: 'No recording attached' } unless call.recording.attached?
+
+    transcribed_text = transcribe_audio
+    update_call_and_message(transcribed_text)
+    { success: true, transcript: transcribed_text }
+  rescue Faraday::UnauthorizedError
+    Rails.logger.warn('[WHATSAPP CALL] Skipping transcription: OpenAI configuration is invalid (401)')
+    { error: 'OpenAI configuration is invalid' }
+  end
+
+  private
+
+  def can_transcribe?
+    account.feature_enabled?('captain_integration') &&
+      account.usage_limits[:captain][:responses][:current_available].positive?
+  end
+
+  # Transcribe per-direction recordings separately when possible so lines can
+  # be attributed to Customer vs Agent. Falls back to the combined recording
+  # if the media server isn't available or per-side files are missing.
+  def transcribe_audio
+    if call.media_session_id.present?
+      diarized = diarized_transcript
+      return diarized if diarized.present?
+    end
+
+    transcribe_combined
+  end
+
+  def diarized_transcript
+    temp_dir = Rails.root.join('tmp/uploads/call-transcriptions')
+    FileUtils.mkdir_p(temp_dir)
+    customer_path = File.join(temp_dir, "#{call.media_session_id}_customer.ogg")
+    agent_path    = File.join(temp_dir, "#{call.media_session_id}_agent.ogg")
+
+    client = Whatsapp::MediaServerClient.new
+    File.binwrite(customer_path, client.download_recording(call.media_session_id, side: 'customer'))
+    File.binwrite(agent_path,    client.download_recording(call.media_session_id, side: 'agent'))
+
+    segments = transcribe_segments(customer_path, 'Customer') +
+               transcribe_segments(agent_path, 'Agent')
+    return nil if segments.empty?
+
+    segments.sort_by { |s| s[:start] }
+            .map { |s| "[#{format_ts(s[:start])}] #{s[:speaker]}: #{s[:text].strip}" }
+            .reject { |line| line.end_with?(': ') }
+            .join("\n")
+  rescue Whatsapp::MediaServerClient::ConnectionError, Whatsapp::MediaServerClient::SessionError => e
+    Rails.logger.warn "[WHATSAPP CALL] Per-side recording unavailable, falling back to combined transcription: #{e.message}"
+    nil
+  ensure
+    FileUtils.rm_f(customer_path) if defined?(customer_path) && customer_path
+    FileUtils.rm_f(agent_path)    if defined?(agent_path) && agent_path
+  end
+
+  def transcribe_segments(file_path, speaker)
+    return [] unless File.exist?(file_path) && File.size(file_path).positive?
+
+    File.open(file_path, 'rb') do |file|
+      response = @client.audio.transcribe(
+        parameters: {
+          model: WHISPER_MODEL,
+          file: file,
+          temperature: 0.2,
+          response_format: 'verbose_json',
+          timestamp_granularities: ['segment']
+        }
+      )
+      (response['segments'] || []).map do |seg|
+        { speaker: speaker, start: seg['start'].to_f, text: seg['text'].to_s }
+      end
+    end
+  end
+
+  def transcribe_combined
+    temp_file_path = fetch_combined_recording
+    File.open(temp_file_path, 'rb') do |file|
+      response = @client.audio.transcribe(
+        parameters: { model: WHISPER_MODEL, file: file, temperature: 0.2 }
+      )
+      return response['text']
+    end
+  ensure
+    FileUtils.rm_f(temp_file_path) if defined?(temp_file_path) && temp_file_path
+  end
+
+  def fetch_combined_recording
+    blob = call.recording.blob
+    temp_dir = Rails.root.join('tmp/uploads/call-transcriptions')
+    FileUtils.mkdir_p(temp_dir)
+
+    extension = extension_from_content_type(blob.content_type)
+    temp_file_path = File.join(temp_dir, "#{blob.key}.#{extension}")
+
+    File.open(temp_file_path, 'wb') do |file|
+      blob.open { |blob_file| IO.copy_stream(blob_file, file) }
+    end
+
+    temp_file_path
+  end
+
+  def format_ts(seconds)
+    total = seconds.to_i
+    mins = total / 60
+    secs = total % 60
+    format('%<mins>02d:%<secs>02d', mins: mins, secs: secs)
+  end
+
+  def update_call_and_message(transcribed_text)
+    return if transcribed_text.blank?
+
+    call.update!(transcript: transcribed_text)
+    account.increment_response_usage
+
+    message = call.message
+    return unless message
+
+    data = (message.content_attributes || {}).dup
+    data['data'] ||= {}
+    data['data']['transcript'] = transcribed_text
+    data['data']['recording_url'] = call.recording_url
+    message.update!(content_attributes: data)
+  end
+
+  def extension_from_content_type(content_type)
+    subtype = content_type.to_s.downcase.split(';').first.to_s.split('/').last.to_s
+    { 'webm' => 'webm', 'ogg' => 'ogg', 'x-m4a' => 'm4a', 'x-wav' => 'wav', 'mpeg' => 'mp3' }.fetch(subtype, 'webm')
+  end
+end

enterprise/app/services/whatsapp/incoming_call_service.rb

@@ -0,0 +1,248 @@
+class Whatsapp::IncomingCallService
+  pattr_initialize [:inbox!, :params!]
+
+  def perform
+    return unless inbox.account.feature_enabled?('whatsapp_call')
+
+    calls = params[:calls]
+    return if calls.blank?
+
+    calls.each do |call_payload|
+      process_call_event(call_payload.with_indifferent_access)
+    end
+  end
+
+  private
+
+  def process_call_event(call_payload)
+    case call_payload[:event]
+    when 'connect' then handle_call_connect(call_payload)
+    when 'terminate' then handle_call_terminate(call_payload)
+    else Rails.logger.warn "[WHATSAPP CALL] Unknown call event: #{call_payload[:event]}"
+    end
+  end
+
+  def handle_call_connect(call_payload)
+    provider_call_id = call_payload[:id]
+    direction = map_direction(call_payload[:direction])
+
+    # For outbound calls, a Call record already exists from initiate.
+    # Update it instead of creating a duplicate.
+    existing_call = Call.whatsapp.find_by(provider_call_id: provider_call_id)
+    if existing_call
+      Rails.logger.info "[WHATSAPP CALL] call_connect for existing call #{provider_call_id} (direction=#{direction})"
+      # Guard against race condition: skip if already in_progress (agent accepted via CallService)
+      return if existing_call.in_progress?
+
+      sdp_answer = fix_sdp_setup(call_payload.dig(:session, :sdp))
+      existing_call.update!(status: 'in_progress', started_at: Time.current, meta: (existing_call.meta || {}).merge('sdp_answer' => sdp_answer))
+      Whatsapp::CallMessageBuilder.update_status!(call: existing_call, status: 'in_progress')
+      update_conversation_call_status(existing_call.conversation, 'in-progress', existing_call.direction_label)
+
+      if existing_call.media_session_id.present?
+        finalize_outbound_server_relay(existing_call, sdp_answer)
+      else
+        broadcast_outbound_call_connected(existing_call, sdp_answer)
+      end
+      return
+    end
+
+    contact = find_or_create_contact("+#{call_payload[:from]}")
+    return unless contact
+
+    conversation = find_or_create_conversation(contact)
+    return unless conversation
+
+    call = create_call_record(call_payload, conversation, contact, direction)
+    create_voice_call_message(conversation, call)
+    update_conversation_call_status(conversation, 'ringing', call.direction_label)
+    broadcast_incoming_call(call, contact, call_payload.dig(:session, :sdp))
+  rescue ActiveRecord::RecordNotUnique
+    Rails.logger.warn "[WHATSAPP CALL] Duplicate provider_call_id received: #{provider_call_id}"
+  end
+
+  def create_voice_call_message(conversation, call, user: nil)
+    message = Whatsapp::CallMessageBuilder.create!(conversation: conversation, call: call, user: user)
+    call.update!(message_id: message.id)
+  rescue StandardError => e
+    Rails.logger.error "[WHATSAPP CALL] Failed to create voice_call message: #{e.message}"
+  end
+
+  def create_call_record(call_payload, conversation, contact, direction)
+    Call.create!(
+      provider: :whatsapp,
+      account: inbox.account,
+      inbox: inbox,
+      conversation: conversation,
+      contact: contact,
+      provider_call_id: call_payload[:id],
+      direction: direction,
+      status: 'ringing',
+      meta: { sdp_offer: call_payload.dig(:session, :sdp), ice_servers: default_ice_servers }
+    )
+  end
+
+  def handle_call_terminate(call_payload)
+    provider_call_id = call_payload[:id]
+    duration = call_payload[:duration]&.to_i
+    end_reason = call_payload[:terminate_reason]
+
+    call = Call.whatsapp.find_by(provider_call_id: provider_call_id)
+    return unless call
+
+    # Determine if the call was answered: check in_progress status, duration > 0,
+    # or accepted_by_agent_id presence (handles webhook race conditions)
+    was_answered = call.in_progress? || duration.to_i.positive? || call.accepted_by_agent_id.present?
+    final_status = was_answered ? 'completed' : 'no_answer'
+    call.update!(
+      status: final_status,
+      duration_seconds: duration,
+      end_reason: end_reason
+    )
+
+    agent = call.accepted_by_agent if call.accepted_by_agent_id.present?
+    Whatsapp::CallMessageBuilder.update_status!(call: call, status: final_status, agent: agent, duration_seconds: duration)
+    mapped = Whatsapp::CallMessageBuilder::CALL_TO_VOICE_STATUS[final_status] || final_status
+    update_conversation_call_status(call.conversation, mapped, call.direction_label)
+    broadcast_call_ended(call)
+
+    # Fetch recording from media server if a session was active
+    Whatsapp::CallRecordingFetchJob.perform_later(call.id) if call.media_session_id.present?
+  end
+
+  def find_or_create_contact(phone_number)
+    waid = phone_number.delete('+')
+
+    contact_inbox = ::ContactInboxWithContactBuilder.new(
+      source_id: waid,
+      inbox: inbox,
+      contact_attributes: {
+        name: phone_number,
+        phone_number: phone_number
+      }
+    ).perform
+
+    contact_inbox&.contact
+  end
+
+  def find_or_create_conversation(contact)
+    contact_inbox = contact.contact_inboxes.find_by(inbox: inbox)
+    return unless contact_inbox
+
+    conversation = contact_inbox.conversations.where.not(status: :resolved).last
+    return conversation if conversation
+
+    ::Conversation.create!(
+      account_id: inbox.account_id,
+      inbox: inbox,
+      contact: contact,
+      contact_inbox: contact_inbox,
+      additional_attributes: { channel: 'whatsapp' }
+    )
+  end
+
+  def update_conversation_call_status(conversation, call_status, direction)
+    attrs = (conversation.additional_attributes || {}).merge(
+      'call_status' => call_status,
+      'call_direction' => direction
+    )
+    conversation.update!(additional_attributes: attrs)
+  end
+
+  def broadcast_incoming_call(call, contact, sdp_offer)
+    data = {
+      account_id: inbox.account_id,
+      id: call.id,
+      call_id: call.provider_call_id,
+      direction: call.direction_label,
+      inbox_id: call.inbox_id,
+      conversation_id: call.conversation_id,
+      media_server_enabled: Call.media_server_enabled?,
+      caller: {
+        name: contact.name,
+        phone: contact.phone_number,
+        avatar: contact.avatar_url
+      }
+    }
+
+    # When media server is enabled, the browser does not need Meta's SDP since
+    # the Go sidecar handles the Meta-side peer connection directly.
+    unless Call.media_server_enabled?
+      data[:sdp_offer] = sdp_offer
+      data[:ice_servers] = default_ice_servers
+    end
+
+    ActionCable.server.broadcast("account_#{inbox.account_id}", { event: 'whatsapp_call.incoming', data: data })
+  end
+
+  def broadcast_call_ended(call)
+    payload = {
+      event: 'whatsapp_call.ended',
+      data: {
+        account_id: inbox.account_id,
+        id: call.id,
+        call_id: call.provider_call_id,
+        status: call.status,
+        duration_seconds: call.duration_seconds,
+        conversation_id: call.conversation_id
+      }
+    }
+
+    ActionCable.server.broadcast("account_#{inbox.account_id}", payload)
+  end
+
+  def broadcast_outbound_call_connected(call, sdp_answer)
+    payload = {
+      event: 'whatsapp_call.outbound_connected',
+      data: {
+        account_id: inbox.account_id,
+        id: call.id,
+        call_id: call.provider_call_id,
+        conversation_id: call.conversation_id,
+        sdp_answer: sdp_answer
+      }
+    }
+
+    ActionCable.server.broadcast("account_#{inbox.account_id}", payload)
+  end
+
+  # Server-relay outbound: deliver Meta's SDP answer to the media server so it
+  # completes Peer A, then ask it for an SDP offer to send to the agent (Peer B).
+  # Broadcast that offer so the agent browser can negotiate directly with the
+  # media server.
+  def finalize_outbound_server_relay(call, sdp_answer)
+    client = Whatsapp::MediaServerClient.new
+    client.set_meta_answer(call.media_session_id, sdp_answer: sdp_answer)
+    agent_offer = client.generate_agent_offer(call.media_session_id)
+
+    payload = {
+      event: 'whatsapp_call.outbound_connected',
+      data: {
+        account_id: inbox.account_id,
+        id: call.id,
+        call_id: call.provider_call_id,
+        conversation_id: call.conversation_id,
+        sdp_offer: agent_offer['sdp_offer'],
+        ice_servers: agent_offer['ice_servers']
+      }
+    }
+    ActionCable.server.broadcast("account_#{inbox.account_id}", payload)
+  rescue Whatsapp::MediaServerClient::ConnectionError, Whatsapp::MediaServerClient::SessionError => e
+    Rails.logger.error "[WHATSAPP CALL] Failed to finalize outbound server-relay: #{e.message}"
+  end
+
+  # Meta sends "USER_INITIATED" / "BUSINESS_INITIATED", map to Call enum values
+  def map_direction(raw_direction)
+    return :outgoing if raw_direction&.upcase == 'BUSINESS_INITIATED'
+
+    :incoming
+  end
+
+  def default_ice_servers
+    [{ urls: ['stun:stun.l.google.com:19302'] }]
+  end
+
+  def fix_sdp_setup(sdp)
+    sdp.present? ? sdp.gsub('a=setup:actpass', 'a=setup:active') : sdp
+  end
+end

enterprise/app/services/whatsapp/media_server_client.rb

@@ -0,0 +1,129 @@
+class Whatsapp::MediaServerClient
+  class ConnectionError < StandardError; end
+  class SessionError < StandardError; end
+
+  TIMEOUT = 10
+
+  def create_session(call_id:, direction:, sdp_offer:, ice_servers:, account_id: nil)
+    body = { call_id: call_id, direction: direction, meta_sdp_offer: sdp_offer,
+             ice_servers: normalize_ice_servers(ice_servers), account_id: account_id&.to_s }.compact
+    post('/sessions', body)
+  end
+
+  def generate_agent_offer(session_id)
+    post("/sessions/#{session_id}/agent-offer")
+  end
+
+  def set_agent_answer(session_id, sdp_answer:)
+    post("/sessions/#{session_id}/agent-answer", { sdp_answer: sdp_answer })
+  end
+
+  def set_meta_answer(session_id, sdp_answer:)
+    post("/sessions/#{session_id}/meta-answer", { sdp_answer: sdp_answer })
+  end
+
+  def reconnect_agent(session_id)
+    post("/sessions/#{session_id}/agent-reconnect")
+  end
+
+  def terminate_session(session_id)
+    post("/sessions/#{session_id}/terminate")
+  end
+
+  def download_recording(session_id, side: nil)
+    path = "/sessions/#{session_id}/recording"
+    path = "#{path}?side=#{side}" if side
+    response = execute_request(:get, path)
+    unless response.success?
+      Rails.logger.error "[MEDIA SERVER] Recording download failed: side=#{side.inspect} status=#{response.code}"
+      raise SessionError, "Recording download failed (#{response.code})"
+    end
+    response.body
+  end
+
+  def add_peer(session_id, role:, label:)
+    post("/sessions/#{session_id}/peers", { role: role, label: label })
+  end
+
+  def remove_peer(session_id, peer_id:)
+    delete("/sessions/#{session_id}/peers/#{peer_id}")
+  end
+
+  def inject_audio(session_id, file_path:, mode: 'replace', loop: false, target: 'peer_a')
+    post("/sessions/#{session_id}/inject-audio", { file_path: file_path, mode: mode, loop: loop, target: target })
+  end
+
+  def stop_audio_injection(session_id, injection_id:)
+    delete("/sessions/#{session_id}/inject-audio/#{injection_id}")
+  end
+
+  def health_check
+    get('/health')
+  end
+
+  private
+
+  def post(path, body = {})
+    response = execute_request(:post, path, body)
+    parse_response(response)
+  end
+
+  def get(path)
+    response = execute_request(:get, path)
+    parse_response(response)
+  end
+
+  def delete(path)
+    response = execute_request(:delete, path)
+    parse_response(response)
+  end
+
+  def execute_request(method, path, body = nil)
+    url = "#{base_url}#{path}"
+    options = { headers: auth_headers, timeout: TIMEOUT }
+    options[:body] = body.to_json if body.present?
+
+    Rails.logger.info "[MEDIA SERVER] #{method.upcase} #{path}"
+    HTTParty.send(method, url, options)
+  rescue Errno::ECONNREFUSED, Net::OpenTimeout, Net::ReadTimeout, SocketError => e
+    Rails.logger.error "[MEDIA SERVER] Connection failed: #{e.class} #{e.message}"
+    raise ConnectionError, "Media server unavailable: #{e.message}"
+  end
+
+  def parse_response(response)
+    unless response.success?
+      Rails.logger.error "[MEDIA SERVER] Request failed: status=#{response.code} body=#{response.body}"
+      raise SessionError, "Media server error (#{response.code}): #{response.body}"
+    end
+
+    response.parsed_response
+  end
+
+  def base_url
+    ENV.fetch('MEDIA_SERVER_URL', 'http://localhost:4000')
+  end
+
+  def auth_token
+    ENV.fetch('MEDIA_SERVER_AUTH_TOKEN', '')
+  end
+
+  def auth_headers
+    {
+      'Content-Type' => 'application/json',
+      'Authorization' => "Bearer #{auth_token}"
+    }
+  end
+
+  # Go media server expects `urls` to always be an array of strings.
+  # Accept legacy data that may have `urls` as a single string.
+  def normalize_ice_servers(servers)
+    return [] if servers.blank?
+
+    Array(servers).map do |srv|
+      s = srv.respond_to?(:to_h) ? srv.to_h.transform_keys(&:to_s) : srv.stringify_keys
+      urls = s['urls']
+      s['urls'] = urls.is_a?(Array) ? urls : Array(urls).compact
+      s
+    end
+  end
+end

enterprise/app/services/whatsapp/providers/whatsapp_cloud_call_methods.rb

@@ -0,0 +1,85 @@
+module Whatsapp::Providers::WhatsappCloudCallMethods
+  def pre_accept_call(call_id, sdp_answer)
+    call_api('pre_accept_call', call_action_body(call_id, 'pre_accept', sdp_answer))
+  end
+
+  def accept_call(call_id, sdp_answer)
+    call_api('accept_call', call_action_body(call_id, 'accept', sdp_answer))
+  end
+
+  def reject_call(call_id)
+    call_api('reject_call', { messaging_product: 'whatsapp', call_id: call_id, action: 'reject' })
+  end
+
+  def terminate_call(call_id)
+    call_api('terminate_call', { messaging_product: 'whatsapp', call_id: call_id, action: 'terminate' })
+  end
+
+  def send_call_permission_request(to_phone_number, body_text = 'We would like to call you regarding your conversation.')
+    response = HTTParty.post(
+      "#{phone_id_path}/messages", headers: api_headers, body: permission_request_body(to_phone_number, body_text).to_json
+    )
+
+    unless response.success?
+      Rails.logger.error "[WHATSAPP CALL] send_call_permission_request failed: status=#{response.code} body=#{response.body}"
+      return nil
+    end
+
+    response.parsed_response
+  end
+
+  def initiate_call(to_phone_number, sdp_offer)
+    response = HTTParty.post(
+      "#{phone_id_path}/calls", headers: api_headers, body: initiate_call_body(to_phone_number, sdp_offer).to_json
+    )
+    process_initiate_call_response(response)
+  end
+
+  private
+
+  def call_action_body(call_id, action, sdp_answer = nil)
+    body = { messaging_product: 'whatsapp', call_id: call_id, action: action }
+    body[:session] = { sdp: sdp_answer, sdp_type: 'answer' } if sdp_answer
+    body
+  end
+
+  def call_api(action_name, body)
+    url = "#{phone_id_path}/calls"
+    Rails.logger.info "[WHATSAPP CALL] #{action_name} POST #{url} body=#{body.except(:session).to_json}"
+    response = HTTParty.post(url, headers: api_headers, body: body.to_json)
+    Rails.logger.error "[WHATSAPP CALL] #{action_name} failed: status=#{response.code} body=#{response.body}" unless response.success?
+    response.success?
+  end
+
+  def permission_request_body(to_phone_number, body_text)
+    {
+      messaging_product: 'whatsapp', recipient_type: 'individual', to: to_phone_number,
+      type: 'interactive',
+      interactive: {
+        type: 'call_permission_request',
+        action: { name: 'call_permission_request' },
+        body: { text: body_text }
+      }
+    }
+  end
+
+  def initiate_call_body(to_phone_number, sdp_offer)
+    {
+      messaging_product: 'whatsapp', to: to_phone_number, type: 'audio',
+      session: { sdp: sdp_offer, sdp_type: 'offer' }
+    }
+  end
+
+  def process_initiate_call_response(response)
+    return response.parsed_response if response.success?
+
+    parsed = response.parsed_response
+    error_code = parsed&.dig('error', 'code')
+    error_msg = parsed&.dig('error', 'error_user_msg') || 'Failed to initiate call'
+    Rails.logger.error "[WHATSAPP CALL] initiate_call failed: status=#{response.code} body=#{response.body}"
+
+    raise Whatsapp::CallErrors::NoCallPermission, error_msg if error_code == 138_006
+
+    raise StandardError, error_msg
+  end
+end

enterprise/lib/whatsapp/call_errors.rb

@@ -0,0 +1,6 @@
+module Whatsapp::CallErrors
+  class NotRinging < StandardError; end
+  class AlreadyAccepted < StandardError; end
+  class CallFailed < StandardError; end
+  class NoCallPermission < StandardError; end
+end

enterprise/media-server/Dockerfile

@@ -0,0 +1,31 @@
+FROM golang:1.22-alpine AS builder
+
+RUN apk add --no-cache git gcc musl-dev
+
+WORKDIR /build
+
+# Copy module files and download dependencies (layer caching).
+COPY go.mod go.sum* ./
+RUN if [ -f go.sum ]; then go mod download; fi
+
+# Build the binary (go mod tidy ensures go.sum is present).
+COPY . .
+RUN go mod tidy && CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o chatwoot-media-server ./cmd/server/
+
+# ---
+
+FROM alpine:3.19
+
+RUN apk add --no-cache ca-certificates wget
+
+COPY --from=builder /build/chatwoot-media-server /usr/local/bin/
+
+RUN mkdir -p /recordings
+
+EXPOSE 4000
+EXPOSE 10000-12000/udp
+
+HEALTHCHECK --interval=10s --timeout=5s --retries=3 \
+  CMD wget --spider -q http://localhost:4000/health || exit 1
+
+ENTRYPOINT ["chatwoot-media-server"]

enterprise/media-server/README.md

@@ -0,0 +1,155 @@
+# chatwoot-media-server
+
+A Pion WebRTC media server sidecar for Chatwoot's WhatsApp Calling feature. It acts as a back-to-back user agent (B2BUA) between Meta's media servers and the agent's browser, providing call persistence across page reloads, server-side recording, and centralized call lifecycle management.
+
+## Architecture
+
+The media server maintains two independent WebRTC peer connections per call:
+
+- **Peer A (Meta-side):** Receives the customer's audio from Meta and sends the agent's audio back.
+- **Peer B (Agent-side):** Receives the agent's microphone audio and sends the customer's audio to the browser.
+
+An audio bridge forwards RTP packets between the two peers while simultaneously recording both streams to OGG/Opus files.
+
+## Build
+
+```bash
+# Local build
+go build -o chatwoot-media-server ./cmd/server/
+
+# Docker build
+docker build -t chatwoot-media-server .
+```
+
+## Run
+
+```bash
+# Locally
+AUTH_TOKEN=secret RAILS_CALLBACK_URL=http://localhost:3000 ./chatwoot-media-server
+
+# Docker
+docker run -p 4000:4000 -p 10000-10100:10000-10100/udp \
+  -e AUTH_TOKEN=secret \
+  -e RAILS_CALLBACK_URL=http://rails:3000 \
+  -v media-recordings:/recordings \
+  chatwoot-media-server
+```
+
+## Environment Variables
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `AUTH_TOKEN` | (empty) | Shared secret for Bearer token auth. Empty disables auth (dev only). |
+| `RAILS_CALLBACK_URL` | `http://localhost:3000` | Base URL for Rails callbacks. |
+| `STUN_SERVERS` | `stun:stun.l.google.com:19302` | Comma-separated STUN server URLs. |
+| `TURN_SERVERS` | (empty) | Comma-separated TURN server URLs. |
+| `TURN_USERNAME` | (empty) | TURN credential username. |
+| `TURN_PASSWORD` | (empty) | TURN credential password. |
+| `PUBLIC_IP` | (empty) | Server's public IP for ICE candidates. |
+| `UDP_PORT_MIN` | `10000` | Lower bound of UDP port range. |
+| `UDP_PORT_MAX` | `12000` | Upper bound of UDP port range. |
+| `RECORDINGS_DIR` | `/recordings` | Directory for recording files. |
+| `HTTP_PORT` | `4000` | HTTP API listen port. |
+| `LOG_LEVEL` | `info` | Log level (debug, info, warn, error). |
+| `MAX_SESSION_DURATION` | `7200` | Max call duration in seconds (2 hours). |
+| `RECONNECT_TIMEOUT` | `30` | Seconds to wait for agent reconnect. |
+| `MAX_CONCURRENT_SESSIONS` | `0` | Max active sessions (0 = unlimited). |
+
+## API
+
+All endpoints except `/health` require a `Authorization: Bearer <token>` header.
+
+### Sessions
+
+| Method | Path | Description |
+|--------|------|-------------|
+| `POST` | `/sessions` | Create session (Meta SDP offer -> Peer A) |
+| `GET` | `/sessions/:id` | Get session status |
+| `POST` | `/sessions/:id/agent-offer` | Generate agent-side SDP offer (Peer B) |
+| `POST` | `/sessions/:id/agent-answer` | Set agent's SDP answer, complete Peer B |
+| `POST` | `/sessions/:id/agent-reconnect` | Tear down old Peer B, create new one |
+| `POST` | `/sessions/:id/terminate` | End call, finalize recording |
+| `GET` | `/sessions/:id/recording` | Download recording (binary OGG) |
+| `DELETE` | `/sessions/:id` | Cleanup session and files |
+
+### Multi-participant
+
+| Method | Path | Description |
+|--------|------|-------------|
+| `POST` | `/sessions/:id/peers` | Add a peer |
+| `DELETE` | `/sessions/:id/peers/:peer_id` | Remove a peer |
+| `PATCH` | `/sessions/:id/peers/:peer_id/role` | Change peer role |
+
+### Audio Injection
+
+| Method | Path | Description |
+|--------|------|-------------|
+| `POST` | `/sessions/:id/inject-audio` | Start audio injection |
+| `DELETE` | `/sessions/:id/inject-audio/:inj_id` | Stop injection |
+
+### System
+
+| Method | Path | Description |
+|--------|------|-------------|
+| `GET` | `/health` | Health check (no auth) |
+| `GET` | `/metrics` | Session metrics |
+
+### Example: Create Session (Incoming Call)
+
+```bash
+curl -X POST http://localhost:4000/sessions \
+  -H "Authorization: Bearer secret" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "call_id": "call_123",
+    "account_id": "1",
+    "direction": "incoming",
+    "meta_sdp_offer": "v=0\r\no=- ...",
+    "ice_servers": [{"urls": ["stun:stun.l.google.com:19302"]}]
+  }'
+```
+
+Response:
+
+```json
+{
+  "session_id": "sess_20240101120000_1",
+  "meta_sdp_answer": "v=0\r\no=- ...",
+  "status": "created"
+}
+```
+
+## Recording
+
+Recordings are written in real time as OGG/Opus files to the configured recordings directory. Three files are produced per call:
+
+- `{session_id}.ogg` -- Combined audio for playback
+- `{session_id}_customer.ogg` -- Customer channel only (for transcription)
+- `{session_id}_agent.ogg` -- Agent channel only (for transcription)
+
+On call termination, a callback is sent to Rails which fetches the recording via `GET /sessions/:id/recording` and stores it in ActiveStorage.
+
+## Deployment
+
+Add to `docker-compose.yml`:
+
+```yaml
+media-server:
+  build:
+    context: ./enterprise/media-server
+  ports:
+    - "4000:4000"
+    - "10000-10100:10000-10100/udp"
+  environment:
+    - AUTH_TOKEN=${MEDIA_SERVER_AUTH_TOKEN}
+    - RAILS_CALLBACK_URL=http://web:3000
+    - STUN_SERVERS=stun:stun.l.google.com:19302
+    - RECORDINGS_DIR=/recordings
+    - UDP_PORT_MIN=10000
+    - UDP_PORT_MAX=10100
+  volumes:
+    - media-recordings:/recordings
+  restart: unless-stopped
+```
+
+UDP ports must be exposed to the internet for WebRTC connectivity. If direct UDP exposure is not possible, configure a TURN server as a relay.

enterprise/media-server/cmd/server/main.go

@@ -0,0 +1,133 @@
+// Package main is the entry point for the chatwoot-media-server binary. It
+// loads configuration, initializes the session manager, sets up HTTP routes,
+// and starts the server with graceful shutdown support.
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/callback"
+	"github.com/chatwoot/chatwoot-media-server/internal/config"
+	"github.com/chatwoot/chatwoot-media-server/internal/server"
+	"github.com/chatwoot/chatwoot-media-server/internal/session"
+)
+
+func main() {
+	if err := run(); err != nil {
+		slog.Error("fatal error", "error", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	// Load configuration from environment variables.
+	cfg, err := config.Load()
+	if err != nil {
+		return fmt.Errorf("load config: %w", err)
+	}
+
+	// Configure structured logging.
+	setupLogging(cfg.LogLevel)
+
+	// Log configuration warnings.
+	for _, w := range cfg.Validate() {
+		slog.Warn("config warning", "message", w)
+	}
+
+	slog.Info("starting chatwoot-media-server",
+		"http_port", cfg.HTTPPort,
+		"udp_port_range", fmt.Sprintf("%d-%d", cfg.UDPPortMin, cfg.UDPPortMax),
+		"recordings_dir", cfg.RecordingsDir,
+	)
+
+	// Ensure recordings directory exists.
+	if err := os.MkdirAll(cfg.RecordingsDir, 0o755); err != nil {
+		return fmt.Errorf("create recordings directory: %w", err)
+	}
+
+	// Initialize the Rails callback client.
+	railsClient := callback.NewRailsClient(cfg.RailsCallbackURL, cfg.AuthToken)
+
+	// Initialize the session manager.
+	mgr := session.NewManager(cfg, railsClient)
+
+	// Recover any orphaned recordings from a previous crash.
+	mgr.RecoverOrphanedRecordings()
+
+	// Build the HTTP router.
+	router := server.NewRouter(cfg, mgr)
+	handler := router.Build()
+
+	// Create the HTTP server.
+	srv := &http.Server{
+		Addr:         fmt.Sprintf(":%d", cfg.HTTPPort),
+		Handler:      handler,
+		ReadTimeout:  15 * time.Second,
+		WriteTimeout: 30 * time.Second,
+		IdleTimeout:  60 * time.Second,
+	}
+
+	// Start the server in a goroutine.
+	errCh := make(chan error, 1)
+	go func() {
+		slog.Info("HTTP server listening", "addr", srv.Addr)
+		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			errCh <- fmt.Errorf("HTTP server error: %w", err)
+		}
+	}()
+
+	// Wait for shutdown signal or server error.
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+
+	select {
+	case sig := <-sigCh:
+		slog.Info("received shutdown signal", "signal", sig.String())
+	case err := <-errCh:
+		return err
+	}
+
+	// Graceful shutdown: stop accepting new connections, drain existing ones.
+	slog.Info("initiating graceful shutdown")
+
+	// First, terminate all active sessions so recordings are finalized.
+	mgr.Shutdown()
+
+	// Then shut down the HTTP server with a timeout.
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer shutdownCancel()
+
+	if err := srv.Shutdown(shutdownCtx); err != nil {
+		return fmt.Errorf("HTTP server shutdown error: %w", err)
+	}
+
+	slog.Info("server stopped gracefully")
+	return nil
+}
+
+// setupLogging configures the global slog logger with the given level.
+func setupLogging(level string) {
+	var logLevel slog.Level
+	switch level {
+	case "debug":
+		logLevel = slog.LevelDebug
+	case "warn":
+		logLevel = slog.LevelWarn
+	case "error":
+		logLevel = slog.LevelError
+	default:
+		logLevel = slog.LevelInfo
+	}
+
+	handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
+		Level: logLevel,
+	})
+	slog.SetDefault(slog.New(handler))
+}

enterprise/media-server/go.mod

@@ -0,0 +1,30 @@
+module github.com/chatwoot/chatwoot-media-server
+
+go 1.22
+
+require (
+	github.com/pion/interceptor v0.1.37
+	github.com/pion/rtp v1.8.9
+	github.com/pion/webrtc/v4 v4.0.5
+)
+
+require (
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/pion/datachannel v1.5.10 // indirect
+	github.com/pion/dtls/v3 v3.0.4 // indirect
+	github.com/pion/ice/v4 v4.0.3 // indirect
+	github.com/pion/logging v0.2.2 // indirect
+	github.com/pion/mdns/v2 v2.0.7 // indirect
+	github.com/pion/randutil v0.1.0 // indirect
+	github.com/pion/rtcp v1.2.14 // indirect
+	github.com/pion/sctp v1.8.35 // indirect
+	github.com/pion/sdp/v3 v3.0.9 // indirect
+	github.com/pion/srtp/v3 v3.0.4 // indirect
+	github.com/pion/stun/v3 v3.0.0 // indirect
+	github.com/pion/transport/v3 v3.0.7 // indirect
+	github.com/pion/turn/v4 v4.0.0 // indirect
+	github.com/wlynxg/anet v0.0.5 // indirect
+	golang.org/x/crypto v0.29.0 // indirect
+	golang.org/x/net v0.31.0 // indirect
+	golang.org/x/sys v0.27.0 // indirect
+)

enterprise/media-server/go.sum

@@ -0,0 +1,61 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/pion/datachannel v1.5.10 h1:ly0Q26K1i6ZkGf42W7D4hQYR90pZwzFOjTq5AuCKk4o=
+github.com/pion/datachannel v1.5.10/go.mod h1:p/jJfC9arb29W7WrxyKbepTU20CFgyx5oLo8Rs4Py/M=
+github.com/pion/dtls/v3 v3.0.4 h1:44CZekewMzfrn9pmGrj5BNnTMDCFwr+6sLH+cCuLM7U=
+github.com/pion/dtls/v3 v3.0.4/go.mod h1:R373CsjxWqNPf6MEkfdy3aSe9niZvL/JaKlGeFphtMg=
+github.com/pion/ice/v4 v4.0.3 h1:9s5rI1WKzF5DRqhJ+Id8bls/8PzM7mau0mj1WZb4IXE=
+github.com/pion/ice/v4 v4.0.3/go.mod h1:VfHy0beAZ5loDT7BmJ2LtMtC4dbawIkkkejHPRZNB3Y=
+github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI=
+github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y=
+github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
+github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
+github.com/pion/mdns/v2 v2.0.7 h1:c9kM8ewCgjslaAmicYMFQIde2H9/lrZpjBkN8VwoVtM=
+github.com/pion/mdns/v2 v2.0.7/go.mod h1:vAdSYNAT0Jy3Ru0zl2YiW3Rm/fJCwIeM0nToenfOJKA=
+github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
+github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
+github.com/pion/rtcp v1.2.14 h1:KCkGV3vJ+4DAJmvP0vaQShsb0xkRfWkO540Gy102KyE=
+github.com/pion/rtcp v1.2.14/go.mod h1:sn6qjxvnwyAkkPzPULIbVqSKI5Dv54Rv7VG0kNxh9L4=
+github.com/pion/rtp v1.8.9 h1:E2HX740TZKaqdcPmf4pw6ZZuG8u5RlMMt+l3dxeu6Wk=
+github.com/pion/rtp v1.8.9/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU=
+github.com/pion/sctp v1.8.35 h1:qwtKvNK1Wc5tHMIYgTDJhfZk7vATGVHhXbUDfHbYwzA=
+github.com/pion/sctp v1.8.35/go.mod h1:EcXP8zCYVTRy3W9xtOF7wJm1L1aXfKRQzaM33SjQlzg=
+github.com/pion/sdp/v3 v3.0.9 h1:pX++dCHoHUwq43kuwf3PyJfHlwIj4hXA7Vrifiq0IJY=
+github.com/pion/sdp/v3 v3.0.9/go.mod h1:B5xmvENq5IXJimIO4zfp6LAe1fD9N+kFv+V/1lOdz8M=
+github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M=
+github.com/pion/srtp/v3 v3.0.4/go.mod h1:1Jx3FwDoxpRaTh1oRV8A/6G1BnFL+QI82eK4ms8EEJQ=
+github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw=
+github.com/pion/stun/v3 v3.0.0/go.mod h1:HvCN8txt8mwi4FBvS3EmDghW6aQJ24T+y+1TKjB5jyU=
+github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1o0=
+github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo=
+github.com/pion/turn/v4 v4.0.0 h1:qxplo3Rxa9Yg1xXDxxH8xaqcyGUtbHYw4QSCvmFWvhM=
+github.com/pion/turn/v4 v4.0.0/go.mod h1:MuPDkm15nYSklKpN8vWJ9W2M0PlyQZqYt1McGuxG7mA=
+github.com/pion/webrtc/v4 v4.0.5 h1:8cVPojcv3cQTwVga2vF1rzCNvkiEimnYdCCG7yF317I=
+github.com/pion/webrtc/v4 v4.0.5/go.mod h1:LvP8Np5b/sM0uyJIcUPvJcCvhtjHxJwzh2H2PYzE6cQ=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
+github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
+golang.org/x/crypto v0.29.0 h1:L5SG1JTTXupVV3n6sUqMTeWbjAyfPwoda2DLX8J8FrQ=
+golang.org/x/crypto v0.29.0/go.mod h1:+F4F4N5hv6v38hfeYwTdx20oUvLLc+QfrE9Ax9HtgRg=
+golang.org/x/net v0.31.0 h1:68CPQngjLL0r2AlUKiSxtQFKvzRVbnzLwMUn5SzcLHo=
+golang.org/x/net v0.31.0/go.mod h1:P4fl1q7dY2hnZFxEk4pPSkDHF+QqjitcnDjUQyMM+pM=
+golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s=
+golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

enterprise/media-server/internal/auth/middleware.go

@@ -0,0 +1,65 @@
+// Package auth provides HTTP authentication middleware for the media server API.
+package auth
+
+import (
+	"crypto/subtle"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"strings"
+)
+
+// Middleware returns an HTTP middleware that validates Bearer token authentication.
+// Requests without a valid token receive a 401 Unauthorized response. If the
+// configured token is empty, all requests are allowed (development mode).
+func Middleware(token string) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			// Skip auth if no token is configured (development mode).
+			if token == "" {
+				next.ServeHTTP(w, r)
+				return
+			}
+
+			authHeader := r.Header.Get("Authorization")
+			if authHeader == "" {
+				slog.Warn("missing Authorization header",
+					"path", r.URL.Path,
+					"remote_addr", r.RemoteAddr,
+				)
+				writeAuthError(w, "missing Authorization header")
+				return
+			}
+
+			// Extract Bearer token.
+			parts := strings.SplitN(authHeader, " ", 2)
+			if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") {
+				slog.Warn("invalid Authorization header format",
+					"path", r.URL.Path,
+					"remote_addr", r.RemoteAddr,
+				)
+				writeAuthError(w, "invalid Authorization header format")
+				return
+			}
+
+			// Constant-time comparison to prevent timing attacks.
+			if subtle.ConstantTimeCompare([]byte(parts[1]), []byte(token)) != 1 {
+				slog.Warn("invalid auth token",
+					"path", r.URL.Path,
+					"remote_addr", r.RemoteAddr,
+				)
+				writeAuthError(w, "invalid auth token")
+				return
+			}
+
+			next.ServeHTTP(w, r)
+		})
+	}
+}
+
+// writeAuthError writes a JSON-formatted 401 error response.
+func writeAuthError(w http.ResponseWriter, msg string) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusUnauthorized)
+	fmt.Fprintf(w, `{"error":%q}`, msg)
+}

enterprise/media-server/internal/callback/rails_client.go

@@ -0,0 +1,138 @@
+// Package callback provides an HTTP client for sending event notifications
+// back to the Chatwoot Rails application.
+package callback
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"time"
+)
+
+// RailsClient sends HTTP callbacks to the Rails application to notify it of
+// media server events such as agent disconnection, recording availability,
+// session termination, and errors.
+type RailsClient struct {
+	baseURL   string
+	authToken string
+	client    *http.Client
+}
+
+// NewRailsClient creates a new callback client configured with the Rails base
+// URL and shared authentication token. The HTTP client uses a 10-second
+// timeout to avoid blocking the media server on slow Rails responses.
+func NewRailsClient(baseURL, authToken string) *RailsClient {
+	return &RailsClient{
+		baseURL:   baseURL,
+		authToken: authToken,
+		client: &http.Client{
+			Timeout: 10 * time.Second,
+		},
+	}
+}
+
+// AgentDisconnectedPayload is the request body sent when an agent's peer
+// connection drops unexpectedly.
+type AgentDisconnectedPayload struct {
+	SessionID string `json:"session_id"`
+	CallID    string `json:"call_id"`
+	Reason    string `json:"reason"`
+}
+
+// RecordingReadyPayload is the request body sent when a call recording has
+// been finalized and is available for download.
+type RecordingReadyPayload struct {
+	SessionID      string `json:"session_id"`
+	CallID         string `json:"call_id"`
+	FilePath       string `json:"file_path"`
+	DurationSec    int    `json:"duration_seconds"`
+	FileSizeBytes  int64  `json:"file_size_bytes"`
+}
+
+// SessionTerminatedPayload is the request body sent when a call session has
+// been fully terminated and cleaned up.
+type SessionTerminatedPayload struct {
+	SessionID   string `json:"session_id"`
+	CallID      string `json:"call_id"`
+	Reason      string `json:"reason"`
+	DurationSec int    `json:"duration_seconds"`
+}
+
+// ErrorPayload is the request body sent when the media server encounters
+// an error that the Rails application should be aware of.
+type ErrorPayload struct {
+	SessionID string `json:"session_id"`
+	CallID    string `json:"call_id"`
+	Error     string `json:"error"`
+	Code      string `json:"code"`
+}
+
+// NotifyAgentDisconnected informs Rails that an agent's WebRTC peer connection
+// has dropped. Rails can then start the reconnection timer and update the call
+// status accordingly.
+func (c *RailsClient) NotifyAgentDisconnected(ctx context.Context, payload AgentDisconnectedPayload) error {
+	return c.post(ctx, "/callbacks/media_server/agent_disconnected", payload)
+}
+
+// NotifyRecordingReady informs Rails that a call recording has been finalized
+// and is available for download via the GET /sessions/:id/recording endpoint.
+// Rails should enqueue a job to fetch and attach the recording to ActiveStorage.
+func (c *RailsClient) NotifyRecordingReady(ctx context.Context, payload RecordingReadyPayload) error {
+	return c.post(ctx, "/callbacks/media_server/recording_ready", payload)
+}
+
+// NotifySessionTerminated informs Rails that a call session has ended. This
+// is sent after both peer connections are closed and the recording is finalized.
+func (c *RailsClient) NotifySessionTerminated(ctx context.Context, payload SessionTerminatedPayload) error {
+	return c.post(ctx, "/callbacks/media_server/session_terminated", payload)
+}
+
+// NotifyError informs Rails of a media server error that may require attention,
+// such as a failed ICE negotiation or recording write failure.
+func (c *RailsClient) NotifyError(ctx context.Context, payload ErrorPayload) error {
+	return c.post(ctx, "/callbacks/media_server/error", payload)
+}
+
+func (c *RailsClient) post(ctx context.Context, path string, payload any) error {
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return fmt.Errorf("marshal callback payload: %w", err)
+	}
+
+	url := c.baseURL + path
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("create callback request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if c.authToken != "" {
+		req.Header.Set("Authorization", "Bearer "+c.authToken)
+	}
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		slog.Error("callback request failed",
+			"url", url,
+			"error", err,
+		)
+		return fmt.Errorf("callback request to %s: %w", path, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 400 {
+		slog.Error("callback returned error status",
+			"url", url,
+			"status", resp.StatusCode,
+		)
+		return fmt.Errorf("callback to %s returned status %d", path, resp.StatusCode)
+	}
+
+	slog.Debug("callback sent successfully",
+		"url", url,
+		"status", resp.StatusCode,
+	)
+	return nil
+}

enterprise/media-server/internal/config/config.go

@@ -0,0 +1,197 @@
+// Package config provides environment-based configuration for the media server.
+package config
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Config holds all configuration values for the media server, loaded from
+// environment variables at startup. Sensible defaults are provided for
+// development; production deployments should set AUTH_TOKEN and RAILS_CALLBACK_URL
+// at a minimum.
+type Config struct {
+	// AuthToken is the shared secret used for Bearer token authentication
+	// between Rails and the media server. Required in production.
+	AuthToken string
+
+	// RailsCallbackURL is the base URL for HTTP callbacks to the Rails app
+	// (e.g., "http://rails:3000").
+	RailsCallbackURL string
+
+	// STUNServers is a list of STUN server URLs for ICE candidate gathering.
+	STUNServers []string
+
+	// TURNServers is a list of TURN server URLs for relay candidates.
+	TURNServers []string
+
+	// TURNUsername is the credential username for TURN servers.
+	TURNUsername string
+
+	// TURNPassword is the credential password for TURN servers.
+	TURNPassword string
+
+	// PublicIP is the server's public IP address used for ICE candidate
+	// generation via NAT1To1IPs. Leave empty to rely on STUN discovery.
+	PublicIP string
+
+	// UDPPortMin is the lower bound of the ephemeral UDP port range used
+	// for WebRTC media transport.
+	UDPPortMin uint16
+
+	// UDPPortMax is the upper bound of the ephemeral UDP port range.
+	UDPPortMax uint16
+
+	// RecordingsDir is the filesystem path where call recordings are stored.
+	RecordingsDir string
+
+	// HTTPPort is the port on which the HTTP API listens.
+	HTTPPort int
+
+	// LogLevel controls the verbosity of structured logging.
+	// Valid values: "debug", "info", "warn", "error".
+	LogLevel string
+
+	// MaxSessionDuration is the maximum allowed duration for a single call
+	// session before automatic termination.
+	MaxSessionDuration time.Duration
+
+	// ReconnectTimeout is the duration the server waits for an agent to
+	// reconnect after their peer connection drops before terminating the call.
+	ReconnectTimeout time.Duration
+
+	// MaxConcurrentSessions limits the total number of active sessions across
+	// all accounts. Zero means unlimited.
+	MaxConcurrentSessions int
+}
+
+// Load reads configuration from environment variables and returns a validated
+// Config. It returns an error if any required value is missing or invalid.
+func Load() (*Config, error) {
+	cfg := &Config{
+		AuthToken:        getEnv("AUTH_TOKEN", ""),
+		RailsCallbackURL: getEnv("RAILS_CALLBACK_URL", "http://localhost:3000"),
+		TURNUsername:     getEnv("TURN_USERNAME", ""),
+		TURNPassword:     getEnv("TURN_PASSWORD", ""),
+		PublicIP:         getEnv("PUBLIC_IP", ""),
+		RecordingsDir:    getEnv("RECORDINGS_DIR", "/recordings"),
+		LogLevel:         getEnv("LOG_LEVEL", "info"),
+	}
+
+	// Parse STUN servers (comma-separated).
+	stunStr := getEnv("STUN_SERVERS", "stun:stun.l.google.com:19302")
+	if stunStr != "" {
+		cfg.STUNServers = splitAndTrim(stunStr)
+	}
+
+	// Parse TURN servers (comma-separated).
+	turnStr := getEnv("TURN_SERVERS", "")
+	if turnStr != "" {
+		cfg.TURNServers = splitAndTrim(turnStr)
+	}
+
+	// Parse UDP port range.
+	portMin, err := getEnvUint16("UDP_PORT_MIN", 10000)
+	if err != nil {
+		return nil, fmt.Errorf("invalid UDP_PORT_MIN: %w", err)
+	}
+	cfg.UDPPortMin = portMin
+
+	portMax, err := getEnvUint16("UDP_PORT_MAX", 12000)
+	if err != nil {
+		return nil, fmt.Errorf("invalid UDP_PORT_MAX: %w", err)
+	}
+	cfg.UDPPortMax = portMax
+
+	if cfg.UDPPortMin >= cfg.UDPPortMax {
+		return nil, fmt.Errorf("UDP_PORT_MIN (%d) must be less than UDP_PORT_MAX (%d)", cfg.UDPPortMin, cfg.UDPPortMax)
+	}
+
+	// Parse HTTP port.
+	httpPort, err := getEnvInt("HTTP_PORT", 4000)
+	if err != nil {
+		return nil, fmt.Errorf("invalid HTTP_PORT: %w", err)
+	}
+	cfg.HTTPPort = httpPort
+
+	// Parse max session duration.
+	maxDur, err := getEnvInt("MAX_SESSION_DURATION", 7200)
+	if err != nil {
+		return nil, fmt.Errorf("invalid MAX_SESSION_DURATION: %w", err)
+	}
+	cfg.MaxSessionDuration = time.Duration(maxDur) * time.Second
+
+	// Parse reconnect timeout.
+	reconTimeout, err := getEnvInt("RECONNECT_TIMEOUT", 30)
+	if err != nil {
+		return nil, fmt.Errorf("invalid RECONNECT_TIMEOUT: %w", err)
+	}
+	cfg.ReconnectTimeout = time.Duration(reconTimeout) * time.Second
+
+	// Parse max concurrent sessions.
+	maxSessions, err := getEnvInt("MAX_CONCURRENT_SESSIONS", 0)
+	if err != nil {
+		return nil, fmt.Errorf("invalid MAX_CONCURRENT_SESSIONS: %w", err)
+	}
+	cfg.MaxConcurrentSessions = maxSessions
+
+	return cfg, nil
+}
+
+// Validate checks that required configuration values are set for production
+// use. It returns a list of warnings for missing optional values.
+func (c *Config) Validate() []string {
+	var warnings []string
+	if c.AuthToken == "" {
+		warnings = append(warnings, "AUTH_TOKEN is not set; all API requests will be unauthenticated")
+	}
+	if c.RailsCallbackURL == "" {
+		warnings = append(warnings, "RAILS_CALLBACK_URL is not set; callbacks to Rails will fail")
+	}
+	if len(c.STUNServers) == 0 {
+		warnings = append(warnings, "No STUN servers configured; ICE candidate gathering may fail")
+	}
+	return warnings
+}
+
+func getEnv(key, defaultVal string) string {
+	if val := os.Getenv(key); val != "" {
+		return val
+	}
+	return defaultVal
+}
+
+func getEnvInt(key string, defaultVal int) (int, error) {
+	str := os.Getenv(key)
+	if str == "" {
+		return defaultVal, nil
+	}
+	return strconv.Atoi(str)
+}
+
+func getEnvUint16(key string, defaultVal uint16) (uint16, error) {
+	str := os.Getenv(key)
+	if str == "" {
+		return defaultVal, nil
+	}
+	v, err := strconv.ParseUint(str, 10, 16)
+	if err != nil {
+		return 0, err
+	}
+	return uint16(v), nil
+}
+
+func splitAndTrim(s string) []string {
+	parts := strings.Split(s, ",")
+	result := make([]string, 0, len(parts))
+	for _, p := range parts {
+		trimmed := strings.TrimSpace(p)
+		if trimmed != "" {
+			result = append(result, trimmed)
+		}
+	}
+	return result
+}

enterprise/media-server/internal/media/bridge.go

@@ -0,0 +1,323 @@
+package media
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/peer"
+)
+
+// AudioConsumer is an interface for components that want to receive audio
+// frames from the bridge, such as real-time transcription or AI services.
+type AudioConsumer interface {
+	// OnAudioFrame is called for each RTP packet passing through the bridge.
+	// source is either "customer" or "agent".
+	OnAudioFrame(sessionID, source string, packet *rtp.Packet)
+}
+
+// Bridge connects two WebRTC peers (Meta-side and Agent-side) by forwarding
+// RTP audio packets between them. It also taps into both audio streams for
+// recording and external consumers.
+type Bridge struct {
+	sessionID  string
+	metaPeer   *peer.MetaPeer
+	agentPeers map[string]*peer.AgentPeer
+	recorder   *Recorder
+	consumers  []AudioConsumer
+
+	cancel context.CancelFunc
+	mu     sync.RWMutex
+	active bool
+}
+
+// NewBridge creates a new audio bridge for the given session. The bridge does
+// not start forwarding automatically; call Start after both peers are connected.
+func NewBridge(sessionID string, metaPeer *peer.MetaPeer, recorder *Recorder) *Bridge {
+	return &Bridge{
+		sessionID:  sessionID,
+		metaPeer:   metaPeer,
+		agentPeers: make(map[string]*peer.AgentPeer),
+		recorder:   recorder,
+	}
+}
+
+// AddAgentPeer registers an agent peer with the bridge. If the bridge is
+// already running, forwarding to the new peer begins immediately.
+func (b *Bridge) AddAgentPeer(ap *peer.AgentPeer) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.agentPeers[ap.ID] = ap
+	slog.Info("bridge: agent peer added",
+		"session_id", b.sessionID,
+		"peer_id", ap.ID,
+		"role", string(ap.Role),
+	)
+}
+
+// RemoveAgentPeer removes an agent peer from the bridge. Its forwarding
+// goroutines will terminate when the peer's tracks are closed.
+func (b *Bridge) RemoveAgentPeer(peerID string) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	delete(b.agentPeers, peerID)
+	slog.Info("bridge: agent peer removed",
+		"session_id", b.sessionID,
+		"peer_id", peerID,
+	)
+}
+
+// AddConsumer registers an AudioConsumer that receives copies of all audio
+// packets passing through the bridge.
+func (b *Bridge) AddConsumer(c AudioConsumer) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.consumers = append(b.consumers, c)
+}
+
+// Start begins forwarding audio between the Meta peer and all agent peers.
+// It spawns goroutines for each direction of audio flow. The bridge runs
+// until Stop is called or the provided context is cancelled.
+func (b *Bridge) Start(ctx context.Context) {
+	b.mu.Lock()
+	if b.active {
+		b.mu.Unlock()
+		return
+	}
+	b.active = true
+	ctx, b.cancel = context.WithCancel(ctx)
+	b.mu.Unlock()
+
+	slog.Info("bridge: started", "session_id", b.sessionID)
+
+	// Forward Meta audio (customer) to all agent peers.
+	go b.forwardMetaToAgents(ctx)
+
+	// For each agent peer, forward their audio to Meta.
+	b.mu.RLock()
+	for _, ap := range b.agentPeers {
+		go b.forwardAgentToMeta(ctx, ap)
+	}
+	b.mu.RUnlock()
+}
+
+// StartAgentForwarding begins forwarding a specific agent peer's audio to
+// Meta. This is used when a new agent peer is added after the bridge has
+// already started.
+func (b *Bridge) StartAgentForwarding(ctx context.Context, ap *peer.AgentPeer) {
+	b.mu.RLock()
+	active := b.active
+	b.mu.RUnlock()
+
+	if !active {
+		return
+	}
+
+	go b.forwardAgentToMeta(ctx, ap)
+}
+
+// Stop halts all audio forwarding and finalizes the recording. This method
+// is idempotent.
+func (b *Bridge) Stop() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if !b.active {
+		return
+	}
+	b.active = false
+
+	if b.cancel != nil {
+		b.cancel()
+	}
+
+	slog.Info("bridge: stopped", "session_id", b.sessionID)
+}
+
+// IsActive returns whether the bridge is currently forwarding audio.
+func (b *Bridge) IsActive() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.active
+}
+
+// forwardMetaToAgents reads RTP packets from the Meta peer's remote audio
+// track (customer audio) and writes them to every connected agent peer's
+// local track. Each packet is also sent to the recorder and any consumers.
+func (b *Bridge) forwardMetaToAgents(ctx context.Context) {
+	metaTrack := b.metaPeer.AudioTrack()
+	if metaTrack == nil {
+		slog.Warn("bridge: Meta audio track not yet available, waiting via OnTrack",
+			"session_id", b.sessionID,
+		)
+		// The track will be set via OnTrack callback. We wait for the track
+		// by polling with a channel. In production, the OnTrack callback
+		// mechanism in the session handles this coordination.
+		return
+	}
+
+	b.readAndForwardMetaTrack(ctx, metaTrack)
+}
+
+// ReadAndForwardMetaTrack is the core loop that reads from a Meta remote
+// track and fans out to agent peers. It is exported so the session layer can
+// call it directly from the OnTrack callback.
+func (b *Bridge) ReadAndForwardMetaTrack(ctx context.Context, track *webrtc.TrackRemote) {
+	b.readAndForwardMetaTrack(ctx, track)
+}
+
+func (b *Bridge) readAndForwardMetaTrack(ctx context.Context, track *webrtc.TrackRemote) {
+	slog.Info("bridge: forwarding Meta audio to agents",
+		"session_id", b.sessionID,
+		"codec", track.Codec().MimeType,
+	)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		pkt, _, readErr := track.ReadRTP()
+		if readErr != nil {
+			slog.Debug("bridge: Meta track read ended",
+				"session_id", b.sessionID,
+				"error", readErr,
+			)
+			return
+		}
+
+		// Record customer audio.
+		if b.recorder != nil {
+			if err := b.recorder.WriteCustomerRTP(pkt); err != nil {
+				slog.Warn("bridge: failed to record customer audio",
+					"session_id", b.sessionID,
+					"error", err,
+				)
+			}
+		}
+
+		// Notify consumers.
+		b.mu.RLock()
+		for _, c := range b.consumers {
+			c.OnAudioFrame(b.sessionID, "customer", pkt)
+		}
+		b.mu.RUnlock()
+
+		// Fan-out to all agent peers.
+		b.mu.RLock()
+		for _, ap := range b.agentPeers {
+			if ap.Role == peer.RoleInjectOnly {
+				continue // inject-only peers do not receive audio
+			}
+			raw, marshalErr := pkt.Marshal()
+			if marshalErr != nil {
+				slog.Warn("bridge: failed to marshal RTP packet",
+					"session_id", b.sessionID,
+					"error", marshalErr,
+				)
+				continue
+			}
+			if _, writeErr := ap.LocalTrack().Write(raw); writeErr != nil {
+				slog.Debug("bridge: failed to write to agent peer",
+					"session_id", b.sessionID,
+					"peer_id", ap.ID,
+					"error", writeErr,
+				)
+			}
+		}
+		b.mu.RUnlock()
+	}
+}
+
+// forwardAgentToMeta reads RTP packets from an agent peer's remote audio
+// track (agent microphone) and writes them to the Meta peer's local track.
+func (b *Bridge) forwardAgentToMeta(ctx context.Context, ap *peer.AgentPeer) {
+	agentTrack := ap.AudioTrack()
+	if agentTrack == nil {
+		slog.Debug("bridge: agent audio track not yet available, waiting via OnTrack",
+			"session_id", b.sessionID,
+			"peer_id", ap.ID,
+		)
+		return
+	}
+
+	b.readAndForwardAgentTrack(ctx, ap, agentTrack)
+}
+
+// ReadAndForwardAgentTrack is the core loop that reads from an agent's remote
+// track and forwards to Meta. Exported so the session layer can call it from
+// the OnTrack callback.
+func (b *Bridge) ReadAndForwardAgentTrack(ctx context.Context, ap *peer.AgentPeer, track *webrtc.TrackRemote) {
+	b.readAndForwardAgentTrack(ctx, ap, track)
+}
+
+func (b *Bridge) readAndForwardAgentTrack(ctx context.Context, ap *peer.AgentPeer, track *webrtc.TrackRemote) {
+	slog.Info("bridge: forwarding agent audio to Meta",
+		"session_id", b.sessionID,
+		"peer_id", ap.ID,
+		"codec", track.Codec().MimeType,
+	)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		pkt, _, readErr := track.ReadRTP()
+		if readErr != nil {
+			slog.Debug("bridge: agent track read ended",
+				"session_id", b.sessionID,
+				"peer_id", ap.ID,
+				"error", readErr,
+			)
+			return
+		}
+
+		// Only active peers send audio to Meta.
+		if ap.Role != peer.RoleActive {
+			continue
+		}
+
+		// Record agent audio.
+		if b.recorder != nil {
+			if err := b.recorder.WriteAgentRTP(pkt); err != nil {
+				slog.Warn("bridge: failed to record agent audio",
+					"session_id", b.sessionID,
+					"error", err,
+				)
+			}
+		}
+
+		// Notify consumers.
+		b.mu.RLock()
+		for _, c := range b.consumers {
+			c.OnAudioFrame(b.sessionID, "agent", pkt)
+		}
+		b.mu.RUnlock()
+
+		// Forward to Meta.
+		raw, marshalErr := pkt.Marshal()
+		if marshalErr != nil {
+			slog.Warn("bridge: failed to marshal agent RTP packet",
+				"session_id", b.sessionID,
+				"error", marshalErr,
+			)
+			continue
+		}
+		if _, writeErr := b.metaPeer.LocalTrack().Write(raw); writeErr != nil {
+			slog.Debug("bridge: failed to write to Meta peer",
+				"session_id", b.sessionID,
+				"error", writeErr,
+			)
+		}
+	}
+}

enterprise/media-server/internal/media/injector.go

@@ -0,0 +1,189 @@
+package media
+
+import (
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4/pkg/media/oggreader"
+)
+
+// Injector reads Opus frames from an OGG file and injects them into an RTP
+// stream at the correct 20ms pacing interval. This is used for hold music,
+// announcements, and other pre-recorded audio injection.
+type Injector struct {
+	ID     string
+	Source string // file path of the OGG/Opus file
+	Mode   string // "replace" or "mix"
+	Loop   bool
+	Target string // "meta", "agents", or "all"
+
+	stopCh chan struct{}
+	mu     sync.Mutex
+	active bool
+}
+
+// InjectorTarget is a writable RTP destination.
+type InjectorTarget interface {
+	Write(b []byte) (n int, err error)
+}
+
+// NewInjector creates a new audio injector configured with the given source
+// file and injection parameters.
+func NewInjector(id, source, mode, target string, loop bool) *Injector {
+	return &Injector{
+		ID:     id,
+		Source: source,
+		Mode:   mode,
+		Loop:   loop,
+		Target: target,
+		stopCh: make(chan struct{}),
+	}
+}
+
+// Start begins reading the OGG file and injecting Opus frames into the target
+// track at 20ms intervals. It runs in its own goroutine and returns immediately.
+// The injection stops when Stop is called, the file ends (if Loop is false),
+// or an error occurs.
+func (inj *Injector) Start(target InjectorTarget) error {
+	inj.mu.Lock()
+	if inj.active {
+		inj.mu.Unlock()
+		return fmt.Errorf("injector %s is already active", inj.ID)
+	}
+	inj.active = true
+	inj.stopCh = make(chan struct{}) // fresh channel for each start cycle
+	inj.mu.Unlock()
+
+	go inj.run(target)
+	return nil
+}
+
+// Stop halts the audio injection. This method is safe to call multiple times.
+func (inj *Injector) Stop() {
+	inj.mu.Lock()
+	defer inj.mu.Unlock()
+
+	if !inj.active {
+		return
+	}
+	inj.active = false
+	close(inj.stopCh)
+
+	slog.Info("injector: stopped", "id", inj.ID)
+}
+
+// IsActive returns whether the injector is currently running.
+func (inj *Injector) IsActive() bool {
+	inj.mu.Lock()
+	defer inj.mu.Unlock()
+	return inj.active
+}
+
+func (inj *Injector) run(target InjectorTarget) {
+	defer func() {
+		inj.mu.Lock()
+		inj.active = false
+		inj.mu.Unlock()
+	}()
+
+	slog.Info("injector: started",
+		"id", inj.ID,
+		"source", inj.Source,
+		"mode", inj.Mode,
+		"loop", inj.Loop,
+	)
+
+	for {
+		if err := inj.playFile(target); err != nil {
+			slog.Error("injector: playback error",
+				"id", inj.ID,
+				"error", err,
+			)
+			return
+		}
+
+		if !inj.Loop {
+			slog.Info("injector: playback complete (no loop)", "id", inj.ID)
+			return
+		}
+
+		// Check stop signal between loops.
+		select {
+		case <-inj.stopCh:
+			return
+		default:
+		}
+	}
+}
+
+func (inj *Injector) playFile(target InjectorTarget) error {
+	f, err := os.Open(inj.Source)
+	if err != nil {
+		return fmt.Errorf("open source file: %w", err)
+	}
+	defer f.Close()
+
+	ogg, _, err := oggreader.NewWith(f)
+	if err != nil {
+		return fmt.Errorf("create OGG reader: %w", err)
+	}
+
+	// Opus frames are 20ms at 48kHz = 960 samples per frame.
+	const opusFrameDuration = 20 * time.Millisecond
+	ticker := time.NewTicker(opusFrameDuration)
+	defer ticker.Stop()
+
+	var sequenceNumber uint16
+	var timestamp uint32
+
+	for {
+		select {
+		case <-inj.stopCh:
+			return nil
+		case <-ticker.C:
+		}
+
+		pageData, _, err := ogg.ParseNextPage()
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return fmt.Errorf("parse OGG page: %w", err)
+		}
+
+		// Build an RTP packet with the Opus payload.
+		pkt := &rtp.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				PayloadType:    111, // Opus dynamic payload type
+				SequenceNumber: sequenceNumber,
+				Timestamp:      timestamp,
+				SSRC:           12345678, // fixed SSRC for injected audio
+			},
+			Payload: pageData,
+		}
+		sequenceNumber++
+		timestamp += 960 // 48kHz * 0.02s
+
+		raw, marshalErr := pkt.Marshal()
+		if marshalErr != nil {
+			slog.Warn("injector: failed to marshal RTP packet",
+				"id", inj.ID,
+				"error", marshalErr,
+			)
+			continue
+		}
+
+		if _, writeErr := target.Write(raw); writeErr != nil {
+			slog.Debug("injector: write failed",
+				"id", inj.ID,
+				"error", writeErr,
+			)
+		}
+	}
+}

enterprise/media-server/internal/media/recorder.go

@@ -0,0 +1,297 @@
+// Package media provides audio bridging, recording, and injection capabilities
+// for the media server's call sessions.
+package media
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4/pkg/media/oggwriter"
+)
+
+// Recorder writes incoming Opus RTP packets to two per-direction OGG files
+// (customer and agent). At finalize time the two files are mixed into a single
+// stereo combined.ogg via ffmpeg so playback and Whisper transcription receive
+// a file with coherent OGG pages and correct duration.
+//
+// Writing both streams to a single oggwriter produces a file with non-monotonic
+// granule positions — the two RTP streams have unrelated clocks and sequence
+// numbers — which breaks both the reported audio duration in browsers and the
+// transcription output.
+type Recorder struct {
+	sessionID string
+	dir       string
+
+	// combinedFile is produced by ffmpeg at finalize; never written to directly.
+	combinedFile string
+
+	// customerWriter writes only customer audio (Meta-side track).
+	customerWriter *oggwriter.OggWriter
+	customerFile   string
+
+	// agentWriter writes only agent audio (browser-side track).
+	agentWriter *oggwriter.OggWriter
+	agentFile   string
+
+	// bothActive gates the first RTP write on either side until the *other*
+	// side has produced at least one packet. This clips the pre-answer ringing
+	// period — browsers send mic RTP as soon as Peer B connects, ~3-5 s before
+	// the contact picks up, which would otherwise inflate the agent OGG and
+	// produce a recording much longer than the real conversation.
+	customerSeen bool
+	agentSeen    bool
+
+	startedAt time.Time
+	finalized bool
+	mu        sync.Mutex
+}
+
+// NewRecorder creates a new recorder that writes OGG/Opus files to the given
+// directory. Three files are tracked:
+//   - {sessionID}_customer.ogg (customer channel only, written live)
+//   - {sessionID}_agent.ogg    (agent channel only, written live)
+//   - {sessionID}.ogg          (combined stereo, produced at Finalize via ffmpeg)
+func NewRecorder(sessionID, dir string) (*Recorder, error) {
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return nil, fmt.Errorf("create recordings directory: %w", err)
+	}
+
+	combinedFile := filepath.Join(dir, sessionID+".ogg")
+	customerFile := filepath.Join(dir, sessionID+"_customer.ogg")
+	agentFile := filepath.Join(dir, sessionID+"_agent.ogg")
+
+	customerWriter, err := oggwriter.New(customerFile, 48000, 1)
+	if err != nil {
+		return nil, fmt.Errorf("create customer OGG writer: %w", err)
+	}
+
+	agentWriter, err := oggwriter.New(agentFile, 48000, 1)
+	if err != nil {
+		customerWriter.Close()
+		return nil, fmt.Errorf("create agent OGG writer: %w", err)
+	}
+
+	slog.Info("recorder: started",
+		"session_id", sessionID,
+		"customer_file", customerFile,
+		"agent_file", agentFile,
+	)
+
+	return &Recorder{
+		sessionID:      sessionID,
+		dir:            dir,
+		combinedFile:   combinedFile,
+		customerWriter: customerWriter,
+		customerFile:   customerFile,
+		agentWriter:    agentWriter,
+		agentFile:      agentFile,
+		startedAt:      time.Now(),
+	}, nil
+}
+
+// WriteCustomerRTP writes an RTP packet from the customer's audio stream
+// (Meta-side, Peer A) to the customer-only OGG file. Drops packets until the
+// agent side has also produced RTP, so the recording spans only the actual
+// conversation — not the pre-answer window.
+func (r *Recorder) WriteCustomerRTP(pkt *rtp.Packet) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.finalized {
+		return nil
+	}
+
+	r.customerSeen = true
+	if !r.agentSeen {
+		return nil
+	}
+
+	if err := r.customerWriter.WriteRTP(pkt); err != nil {
+		return fmt.Errorf("write customer RTP: %w", err)
+	}
+	return nil
+}
+
+// WriteAgentRTP writes an RTP packet from the agent's audio stream
+// (browser-side, Peer B) to the agent-only OGG file. Drops packets until the
+// customer side has also produced RTP so the two streams cover the same
+// wall-clock window and the combined ffmpeg mix has a sensible duration.
+func (r *Recorder) WriteAgentRTP(pkt *rtp.Packet) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.finalized {
+		return nil
+	}
+
+	r.agentSeen = true
+	if !r.customerSeen {
+		return nil
+	}
+
+	if err := r.agentWriter.WriteRTP(pkt); err != nil {
+		return fmt.Errorf("write agent RTP: %w", err)
+	}
+	return nil
+}
+
+// Finalize closes the per-direction writers, then merges them into a single
+// stereo combined.ogg via ffmpeg. If ffmpeg isn't on PATH the combined file
+// is produced by copying the customer-side file as a fallback so that at
+// least one side is playable. Transcription quality degrades in the fallback
+// but the pipeline does not break.
+func (r *Recorder) Finalize() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.finalized {
+		return nil
+	}
+	r.finalized = true
+
+	var errs []error
+	if err := r.customerWriter.Close(); err != nil {
+		errs = append(errs, fmt.Errorf("close customer writer: %w", err))
+	}
+	if err := r.agentWriter.Close(); err != nil {
+		errs = append(errs, fmt.Errorf("close agent writer: %w", err))
+	}
+
+	if err := r.buildCombinedFile(); err != nil {
+		errs = append(errs, err)
+	}
+
+	if len(errs) > 0 {
+		return fmt.Errorf("finalize recorder: %v", errs)
+	}
+
+	slog.Info("recorder: finalized",
+		"session_id", r.sessionID,
+		"duration", time.Since(r.startedAt).Round(time.Second),
+		"combined_file", r.combinedFile,
+	)
+	return nil
+}
+
+// buildCombinedFile mixes the two per-direction OGG files into a single
+// stereo combined.ogg so playback and transcription receive a coherent file
+// with correct duration metadata.
+func (r *Recorder) buildCombinedFile() error {
+	customerExists := fileNonEmpty(r.customerFile)
+	agentExists := fileNonEmpty(r.agentFile)
+	if !customerExists && !agentExists {
+		return errors.New("build combined file: no per-direction recordings to merge")
+	}
+
+	ffmpegPath, err := exec.LookPath("ffmpeg")
+	if err != nil {
+		slog.Warn("recorder: ffmpeg not found, falling back to single-side combined file",
+			"session_id", r.sessionID,
+		)
+		src := r.customerFile
+		if !customerExists {
+			src = r.agentFile
+		}
+		return copyFile(src, r.combinedFile)
+	}
+
+	args := []string{"-y", "-loglevel", "error"}
+	if customerExists {
+		args = append(args, "-i", r.customerFile)
+	}
+	if agentExists {
+		args = append(args, "-i", r.agentFile)
+	}
+
+	switch {
+	case customerExists && agentExists:
+		// Mix two mono streams into one mono track. amix pads the shorter input
+		// with silence so the output duration matches the longer input, and
+		// uses the longer input as the reference for timing so the OGG pages
+		// carry correct granule positions.
+		args = append(args,
+			"-filter_complex", "[0:a][1:a]amix=inputs=2:duration=longest:dropout_transition=0[aout]",
+			"-map", "[aout]",
+		)
+	default:
+		// Only one side — just remux to produce correct OGG duration headers.
+		args = append(args, "-map", "0:a")
+	}
+
+	args = append(args, "-c:a", "libopus", "-b:a", "48000", "-ar", "48000", "-ac", "1", r.combinedFile)
+
+	cmd := exec.Command(ffmpegPath, args...)
+	out, cmdErr := cmd.CombinedOutput()
+	if cmdErr != nil {
+		return fmt.Errorf("ffmpeg mix failed: %w (%s)", cmdErr, string(out))
+	}
+	return nil
+}
+
+func fileNonEmpty(path string) bool {
+	info, err := os.Stat(path)
+	return err == nil && info.Size() > 0
+}
+
+func copyFile(src, dst string) error {
+	data, err := os.ReadFile(src)
+	if err != nil {
+		return fmt.Errorf("read %s: %w", src, err)
+	}
+	if err := os.WriteFile(dst, data, 0o644); err != nil {
+		return fmt.Errorf("write %s: %w", dst, err)
+	}
+	return nil
+}
+
+// CombinedFilePath returns the filesystem path of the combined recording file.
+func (r *Recorder) CombinedFilePath() string {
+	return r.combinedFile
+}
+
+// CustomerFilePath returns the filesystem path of the customer-only recording.
+func (r *Recorder) CustomerFilePath() string {
+	return r.customerFile
+}
+
+// AgentFilePath returns the filesystem path of the agent-only recording.
+func (r *Recorder) AgentFilePath() string {
+	return r.agentFile
+}
+
+// Duration returns the elapsed recording time since the recorder was started.
+func (r *Recorder) Duration() time.Duration {
+	return time.Since(r.startedAt)
+}
+
+// FileSize returns the size of the combined recording file in bytes, or -1 on
+// error.
+func (r *Recorder) FileSize() int64 {
+	info, err := os.Stat(r.combinedFile)
+	if err != nil {
+		return -1
+	}
+	return info.Size()
+}
+
+// Cleanup removes all recording files for this session from disk.
+func (r *Recorder) Cleanup() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	for _, f := range []string{r.combinedFile, r.customerFile, r.agentFile} {
+		if err := os.Remove(f); err != nil && !os.IsNotExist(err) {
+			slog.Warn("recorder: failed to remove file",
+				"file", f,
+				"error", err,
+			)
+		}
+	}
+}

enterprise/media-server/internal/peer/agent_peer.go

@@ -0,0 +1,247 @@
+package peer
+
+import (
+	"fmt"
+	"log/slog"
+	"sync"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/webrtc/v4"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/config"
+)
+
+// PeerRole defines the role of an agent peer in a call session.
+type PeerRole string
+
+const (
+	// RoleActive indicates the peer sends and receives audio (the primary agent).
+	RoleActive PeerRole = "active"
+
+	// RoleListenOnly indicates the peer receives audio but does not send
+	// (supervisory monitoring).
+	RoleListenOnly PeerRole = "listen_only"
+
+	// RoleInjectOnly indicates the peer sends audio but does not receive
+	// (audio injection source).
+	RoleInjectOnly PeerRole = "inject_only"
+)
+
+// AgentPeer represents the WebRTC peer connection to an agent's browser
+// (Peer B). The media server creates an SDP offer for the agent; the browser
+// responds with an SDP answer to complete the handshake.
+type AgentPeer struct {
+	ID   string
+	Role PeerRole
+
+	pc         *webrtc.PeerConnection
+	audioTrack *webrtc.TrackRemote
+	localTrack *webrtc.TrackLocalStaticRTP
+	sender     *webrtc.RTPSender
+
+	// onTrackReady is called when the agent's audio track becomes available.
+	onTrackReady func(track *webrtc.TrackRemote)
+
+	// onICEStateChange is called when the ICE connection state changes.
+	onICEStateChange func(state webrtc.ICEConnectionState)
+
+	mu     sync.Mutex
+	closed bool
+}
+
+// NewAgentPeer creates a new agent-side peer connection and generates an SDP
+// offer to send to the agent's browser. The browser will respond with an SDP
+// answer via SetAnswer. The returned string is the SDP offer.
+func NewAgentPeer(cfg *config.Config, id string, role PeerRole, iceServers []webrtc.ICEServer) (*AgentPeer, string, error) {
+	se := webrtc.SettingEngine{}
+
+	if err := se.SetEphemeralUDPPortRange(cfg.UDPPortMin, cfg.UDPPortMax); err != nil {
+		return nil, "", fmt.Errorf("set UDP port range: %w", err)
+	}
+
+	// NOTE: In pion/webrtc v4.2+, migrate to SetICEAddressRewriteRules.
+	if cfg.PublicIP != "" {
+		se.SetNAT1To1IPs([]string{cfg.PublicIP}, webrtc.ICECandidateTypeSrflx)
+	}
+
+	me := &webrtc.MediaEngine{}
+	if err := me.RegisterDefaultCodecs(); err != nil {
+		return nil, "", fmt.Errorf("register codecs: %w", err)
+	}
+
+	ir := &interceptor.Registry{}
+	if err := webrtc.RegisterDefaultInterceptors(me, ir); err != nil {
+		return nil, "", fmt.Errorf("register interceptors: %w", err)
+	}
+
+	api := webrtc.NewAPI(
+		webrtc.WithMediaEngine(me),
+		webrtc.WithSettingEngine(se),
+		webrtc.WithInterceptorRegistry(ir),
+	)
+
+	pc, err := api.NewPeerConnection(webrtc.Configuration{
+		ICEServers: iceServers,
+	})
+	if err != nil {
+		return nil, "", fmt.Errorf("create peer connection: %w", err)
+	}
+
+	// Create the local audio track that carries customer audio (from Meta)
+	// to the agent's browser.
+	localTrack, err := webrtc.NewTrackLocalStaticRTP(
+		webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus},
+		"audio-to-agent",
+		"chatwoot-media-server",
+	)
+	if err != nil {
+		pc.Close()
+		return nil, "", fmt.Errorf("create local track: %w", err)
+	}
+
+	// Bind the local track to a sendrecv transceiver so the offer advertises
+	// both directions: we send customer audio to the browser, and we expect
+	// the browser's microphone audio back. AddTrack alone yields a sendonly
+	// m=audio, leaving the browser with nowhere to send mic audio.
+	transceiver, err := pc.AddTransceiverFromTrack(localTrack, webrtc.RTPTransceiverInit{
+		Direction: webrtc.RTPTransceiverDirectionSendrecv,
+	})
+	if err != nil {
+		pc.Close()
+		return nil, "", fmt.Errorf("add audio transceiver: %w", err)
+	}
+	sender := transceiver.Sender()
+
+	// Consume RTCP packets from the sender to avoid blocking.
+	go func() {
+		buf := make([]byte, 1500)
+		for {
+			if _, _, rtcpErr := sender.Read(buf); rtcpErr != nil {
+				return
+			}
+		}
+	}()
+
+	ap := &AgentPeer{
+		ID:         id,
+		Role:       role,
+		pc:         pc,
+		localTrack: localTrack,
+		sender:     sender,
+	}
+
+	// Register the OnTrack handler to capture the agent's microphone audio.
+	pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
+		slog.Info("agent peer: remote track received",
+			"peer_id", id,
+			"codec", track.Codec().MimeType,
+			"ssrc", track.SSRC(),
+		)
+		ap.mu.Lock()
+		ap.audioTrack = track
+		cb := ap.onTrackReady
+		ap.mu.Unlock()
+
+		if cb != nil {
+			cb(track)
+		}
+	})
+
+	// Register ICE connection state handler.
+	pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
+		slog.Info("agent peer: ICE state changed",
+			"peer_id", id,
+			"state", state.String(),
+		)
+		ap.mu.Lock()
+		cb := ap.onICEStateChange
+		ap.mu.Unlock()
+
+		if cb != nil {
+			cb(state)
+		}
+	})
+
+	pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
+		slog.Info("agent peer: connection state changed", "peer_id", id, "state", state.String())
+	})
+
+	// Create an SDP offer for the agent's browser.
+	offer, err := pc.CreateOffer(nil)
+	if err != nil {
+		pc.Close()
+		return nil, "", fmt.Errorf("create offer: %w", err)
+	}
+
+	gatherComplete := webrtc.GatheringCompletePromise(pc)
+	if err := pc.SetLocalDescription(offer); err != nil {
+		pc.Close()
+		return nil, "", fmt.Errorf("set local description: %w", err)
+	}
+	<-gatherComplete
+
+	sdpOffer := pc.LocalDescription().SDP
+
+	return ap, sdpOffer, nil
+}
+
+// SetAnswer sets the agent browser's SDP answer on the peer connection,
+// completing the WebRTC handshake.
+func (ap *AgentPeer) SetAnswer(sdpAnswer string) error {
+	ap.mu.Lock()
+	defer ap.mu.Unlock()
+
+	answer := webrtc.SessionDescription{
+		Type: webrtc.SDPTypeAnswer,
+		SDP:  sdpAnswer,
+	}
+	return ap.pc.SetRemoteDescription(answer)
+}
+
+// AudioTrack returns the remote audio track from the agent's browser.
+// Returns nil if the track has not been received yet.
+func (ap *AgentPeer) AudioTrack() *webrtc.TrackRemote {
+	ap.mu.Lock()
+	defer ap.mu.Unlock()
+	return ap.audioTrack
+}
+
+// LocalTrack returns the local RTP track used to send audio to the agent.
+func (ap *AgentPeer) LocalTrack() *webrtc.TrackLocalStaticRTP {
+	return ap.localTrack
+}
+
+// OnTrackReady sets a callback that fires when the agent's microphone audio
+// track becomes available.
+func (ap *AgentPeer) OnTrackReady(fn func(track *webrtc.TrackRemote)) {
+	ap.mu.Lock()
+	defer ap.mu.Unlock()
+	ap.onTrackReady = fn
+}
+
+// OnICEStateChange sets a callback that fires when the ICE connection state
+// changes.
+func (ap *AgentPeer) OnICEStateChange(fn func(state webrtc.ICEConnectionState)) {
+	ap.mu.Lock()
+	defer ap.mu.Unlock()
+	ap.onICEStateChange = fn
+}
+
+// ICEConnectionState returns the current ICE connection state.
+func (ap *AgentPeer) ICEConnectionState() webrtc.ICEConnectionState {
+	return ap.pc.ICEConnectionState()
+}
+
+// Close gracefully shuts down the agent-side peer connection.
+func (ap *AgentPeer) Close() error {
+	ap.mu.Lock()
+	defer ap.mu.Unlock()
+
+	if ap.closed {
+		return nil
+	}
+	ap.closed = true
+
+	slog.Info("agent peer: closing peer connection", "peer_id", ap.ID)
+	return ap.pc.Close()
+}

enterprise/media-server/internal/peer/meta_peer.go

@@ -0,0 +1,297 @@
+// Package peer provides WebRTC peer connection wrappers for the two sides
+// of a call: the Meta-side peer (Peer A) and the Agent-side peer (Peer B).
+package peer
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"runtime/debug"
+	"sync"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/webrtc/v4"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/config"
+)
+
+// MetaPeer represents the WebRTC peer connection to Meta's media servers
+// (Peer A). It receives the customer's audio as an incoming remote track and
+// sends the agent's audio via a local static RTP track.
+type MetaPeer struct {
+	pc         *webrtc.PeerConnection
+	audioTrack *webrtc.TrackRemote
+	localTrack *webrtc.TrackLocalStaticRTP
+	sender     *webrtc.RTPSender
+
+	// onTrackReady is called when the remote audio track from Meta is available.
+	onTrackReady func(track *webrtc.TrackRemote)
+
+	// onICEStateChange is called when the ICE connection state changes.
+	onICEStateChange func(state webrtc.ICEConnectionState)
+
+	mu     sync.Mutex
+	closed bool
+}
+
+// NewMetaPeer creates a new Meta-side peer connection configured for the
+// given ICE servers and UDP port range. For incoming calls, sdpOffer contains
+// Meta's SDP offer; the method sets it as the remote description, creates an
+// answer, and returns the SDP answer string. For outgoing calls, sdpOffer is
+// empty; the method creates an SDP offer to send to Meta.
+func NewMetaPeer(cfg *config.Config, sdpOffer string, iceServers []webrtc.ICEServer) (*MetaPeer, string, error) {
+	se := webrtc.SettingEngine{}
+
+	// Configure the UDP port range for media transport.
+	if err := se.SetEphemeralUDPPortRange(cfg.UDPPortMin, cfg.UDPPortMax); err != nil {
+		return nil, "", fmt.Errorf("set UDP port range: %w", err)
+	}
+
+	// For inbound calls Meta doesn't know our DTLS fingerprint until Rails
+	// delivers the SDP answer via pre_accept_call / accept_call — which runs
+	// *after* create_session returns. If we're DTLS client (pion default when
+	// remote is actpass) we'd send ClientHello before Meta is listening for
+	// us and Meta would respond with a fatal alert. Answer as server so Meta
+	// becomes the client and only starts ClientHello after it has our
+	// fingerprint.
+	if err := se.SetAnsweringDTLSRole(webrtc.DTLSRoleServer); err != nil {
+		return nil, "", fmt.Errorf("set answering DTLS role: %w", err)
+	}
+	se.SetDTLSConnectContextMaker(func() (context.Context, func()) {
+		return context.WithTimeout(context.Background(), 30*time.Second)
+	})
+	se.SetDTLSRetransmissionInterval(200 * time.Millisecond)
+	// Skip the HelloVerify round-trip on the server side. Meta's WhatsApp
+	// calling stack sometimes uses stateless retry which loses the pion
+	// cookie between retransmits; skipping it makes the server accept the
+	// ClientHello in one go.
+	se.SetDTLSInsecureSkipHelloVerify(true)
+
+	// If a public IP is configured, use NAT1To1 so ICE candidates advertise
+	// the correct address instead of a private Docker/container IP.
+	// NOTE: In pion/webrtc v4.2+, migrate to SetICEAddressRewriteRules.
+	if cfg.PublicIP != "" {
+		se.SetNAT1To1IPs([]string{cfg.PublicIP}, webrtc.ICECandidateTypeSrflx)
+	}
+
+	// Build the WebRTC API with a media engine that supports Opus audio.
+	me := &webrtc.MediaEngine{}
+	if err := me.RegisterDefaultCodecs(); err != nil {
+		return nil, "", fmt.Errorf("register codecs: %w", err)
+	}
+
+	// Register default interceptors (NACK, RTCP reports, etc.).
+	ir := &interceptor.Registry{}
+	if err := webrtc.RegisterDefaultInterceptors(me, ir); err != nil {
+		return nil, "", fmt.Errorf("register interceptors: %w", err)
+	}
+
+	api := webrtc.NewAPI(
+		webrtc.WithMediaEngine(me),
+		webrtc.WithSettingEngine(se),
+		webrtc.WithInterceptorRegistry(ir),
+	)
+
+	pc, err := api.NewPeerConnection(webrtc.Configuration{
+		ICEServers: iceServers,
+	})
+	if err != nil {
+		return nil, "", fmt.Errorf("create peer connection: %w", err)
+	}
+
+	// Create a local audio track that will carry the agent's audio to Meta.
+	localTrack, err := webrtc.NewTrackLocalStaticRTP(
+		webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus},
+		"audio-to-meta",
+		"chatwoot-media-server",
+	)
+	if err != nil {
+		pc.Close()
+		return nil, "", fmt.Errorf("create local track: %w", err)
+	}
+
+	// Bind the local track to a single sendrecv transceiver. Using AddTrack +
+	// a separate recvonly transceiver yields two m=audio sections, which Meta
+	// rejects with error 100 "Invalid parameter".
+	transceiver, err := pc.AddTransceiverFromTrack(localTrack, webrtc.RTPTransceiverInit{
+		Direction: webrtc.RTPTransceiverDirectionSendrecv,
+	})
+	if err != nil {
+		pc.Close()
+		return nil, "", fmt.Errorf("add audio transceiver: %w", err)
+	}
+	sender := transceiver.Sender()
+
+	// Consume RTCP packets from the sender to avoid blocking.
+	go func() {
+		buf := make([]byte, 1500)
+		for {
+			if _, _, rtcpErr := sender.Read(buf); rtcpErr != nil {
+				return
+			}
+		}
+	}()
+
+	mp := &MetaPeer{
+		pc:         pc,
+		localTrack: localTrack,
+		sender:     sender,
+	}
+
+	// Register the OnTrack handler to capture the incoming audio from Meta.
+	pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
+		slog.Info("meta peer: remote track received",
+			"codec", track.Codec().MimeType,
+			"ssrc", track.SSRC(),
+		)
+		mp.mu.Lock()
+		mp.audioTrack = track
+		cb := mp.onTrackReady
+		mp.mu.Unlock()
+
+		if cb != nil {
+			cb(track)
+		}
+	})
+
+	// Register ICE connection state handler.
+	pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
+		slog.Info("meta peer: ICE state changed", "state", state.String())
+		mp.mu.Lock()
+		cb := mp.onICEStateChange
+		mp.mu.Unlock()
+
+		if cb != nil {
+			cb(state)
+		}
+	})
+
+	// Log overall peer connection state (covers DTLS + ICE + signaling).
+	// This catches failures that don't surface via ICEConnectionState alone,
+	// such as DTLS handshake errors.
+	pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
+		slog.Info("meta peer: connection state changed", "state", state.String())
+	})
+	pc.OnSignalingStateChange(func(state webrtc.SignalingState) {
+		slog.Debug("meta peer: signaling state changed", "state", state.String())
+	})
+
+	// Also log DTLS transport state — failure here is the likely cause of the
+	// inbound "closed immediately after ICE connected" symptom.
+	if t := pc.SCTP().Transport(); t != nil {
+		t.OnStateChange(func(state webrtc.DTLSTransportState) {
+			slog.Info("meta peer: DTLS state changed", "state", state.String())
+		})
+	}
+
+	// Perform SDP negotiation based on call direction.
+	var sdpResult string
+	if sdpOffer != "" {
+		// Incoming call: Meta sent an offer, we generate an answer.
+		offer := webrtc.SessionDescription{
+			Type: webrtc.SDPTypeOffer,
+			SDP:  sdpOffer,
+		}
+		if err := pc.SetRemoteDescription(offer); err != nil {
+			pc.Close()
+			return nil, "", fmt.Errorf("set remote description (Meta offer): %w", err)
+		}
+
+		answer, err := pc.CreateAnswer(nil)
+		if err != nil {
+			pc.Close()
+			return nil, "", fmt.Errorf("create answer: %w", err)
+		}
+
+		// Wait for ICE gathering to complete before returning the answer.
+		gatherComplete := webrtc.GatheringCompletePromise(pc)
+		if err := pc.SetLocalDescription(answer); err != nil {
+			pc.Close()
+			return nil, "", fmt.Errorf("set local description: %w", err)
+		}
+		<-gatherComplete
+
+		sdpResult = pc.LocalDescription().SDP
+	} else {
+		// Outgoing call: we generate an offer to send to Meta. The sendrecv
+		// transceiver registered above already advertises both directions.
+		offer, err := pc.CreateOffer(nil)
+		if err != nil {
+			pc.Close()
+			return nil, "", fmt.Errorf("create offer: %w", err)
+		}
+
+		gatherComplete := webrtc.GatheringCompletePromise(pc)
+		if err := pc.SetLocalDescription(offer); err != nil {
+			pc.Close()
+			return nil, "", fmt.Errorf("set local description: %w", err)
+		}
+		<-gatherComplete
+
+		sdpResult = pc.LocalDescription().SDP
+	}
+
+	return mp, sdpResult, nil
+}
+
+// SetRemoteAnswer sets Meta's SDP answer on the peer connection, used for
+// outbound calls when Meta responds with an answer.
+func (mp *MetaPeer) SetRemoteAnswer(sdpAnswer string) error {
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+
+	answer := webrtc.SessionDescription{
+		Type: webrtc.SDPTypeAnswer,
+		SDP:  sdpAnswer,
+	}
+	return mp.pc.SetRemoteDescription(answer)
+}
+
+// AudioTrack returns the remote audio track from Meta (customer audio).
+// Returns nil if the track has not been received yet.
+func (mp *MetaPeer) AudioTrack() *webrtc.TrackRemote {
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+	return mp.audioTrack
+}
+
+// LocalTrack returns the local RTP track used to send audio to Meta.
+func (mp *MetaPeer) LocalTrack() *webrtc.TrackLocalStaticRTP {
+	return mp.localTrack
+}
+
+// OnTrackReady sets a callback that fires when the remote audio track from
+// Meta becomes available.
+func (mp *MetaPeer) OnTrackReady(fn func(track *webrtc.TrackRemote)) {
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+	mp.onTrackReady = fn
+}
+
+// OnICEStateChange sets a callback that fires when the ICE connection state
+// changes.
+func (mp *MetaPeer) OnICEStateChange(fn func(state webrtc.ICEConnectionState)) {
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+	mp.onICEStateChange = fn
+}
+
+// ICEConnectionState returns the current ICE connection state.
+func (mp *MetaPeer) ICEConnectionState() webrtc.ICEConnectionState {
+	return mp.pc.ICEConnectionState()
+}
+
+// Close gracefully shuts down the Meta-side peer connection.
+func (mp *MetaPeer) Close() error {
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+
+	if mp.closed {
+		return nil
+	}
+	mp.closed = true
+
+	slog.Info("meta peer: closing peer connection (explicit)", "stack", string(debug.Stack()))
+	return mp.pc.Close()
+}

enterprise/media-server/internal/server/handlers.go

@@ -0,0 +1,753 @@
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/pion/webrtc/v4"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/config"
+	"github.com/chatwoot/chatwoot-media-server/internal/media"
+	"github.com/chatwoot/chatwoot-media-server/internal/peer"
+	"github.com/chatwoot/chatwoot-media-server/internal/session"
+)
+
+// maxRequestBodySize limits JSON request bodies to 1MB to prevent memory exhaustion.
+const maxRequestBodySize = 1 << 20
+
+// startTime is set at server startup for uptime calculations.
+var startTime = time.Now()
+
+// Handlers implements all HTTP API endpoint handlers for the media server.
+type Handlers struct {
+	cfg     *config.Config
+	manager *session.Manager
+}
+
+// NewHandlers creates a new Handlers instance backed by the given session
+// manager and configuration.
+func NewHandlers(cfg *config.Config, mgr *session.Manager) *Handlers {
+	return &Handlers{cfg: cfg, manager: mgr}
+}
+
+// --- Request/Response types ---
+
+// CreateSessionRequest is the JSON body for POST /sessions.
+type CreateSessionRequest struct {
+	CallID       string              `json:"call_id"`
+	AccountID    string              `json:"account_id"`
+	Direction    string              `json:"direction"`
+	MetaSDPOffer string              `json:"meta_sdp_offer"`
+	ICEServers   []ICEServerConfig   `json:"ice_servers"`
+}
+
+// ICEServerConfig mirrors webrtc.ICEServer for JSON deserialization.
+type ICEServerConfig struct {
+	URLs       []string `json:"urls"`
+	Username   string   `json:"username,omitempty"`
+	Credential string   `json:"credential,omitempty"`
+}
+
+// CreateSessionResponse is the JSON response for POST /sessions.
+type CreateSessionResponse struct {
+	SessionID    string `json:"session_id"`
+	MetaSDPAnswer string `json:"meta_sdp_answer,omitempty"`
+	MetaSDPOffer  string `json:"meta_sdp_offer,omitempty"`
+	Status       string `json:"status"`
+}
+
+// AgentOfferRequest is the JSON body for POST /sessions/:id/agent-offer.
+type AgentOfferRequest struct {
+	PeerID     string            `json:"peer_id"`
+	Role       string            `json:"role"`
+	ICEServers []ICEServerConfig `json:"ice_servers"`
+}
+
+// AgentOfferResponse is the JSON response for POST /sessions/:id/agent-offer.
+type AgentOfferResponse struct {
+	SDPOffer   string            `json:"sdp_offer"`
+	PeerID     string            `json:"peer_id"`
+	ICEServers []ICEServerConfig `json:"ice_servers"`
+}
+
+// AgentAnswerRequest is the JSON body for POST /sessions/:id/agent-answer.
+type AgentAnswerRequest struct {
+	PeerID    string `json:"peer_id"`
+	SDPAnswer string `json:"sdp_answer"`
+}
+
+// MetaAnswerRequest is the JSON body for POST /sessions/:id/meta-answer.
+// Used for outgoing calls to deliver Meta's SDP answer to the media server
+// so it can complete the Peer A (Meta-side) WebRTC handshake.
+type MetaAnswerRequest struct {
+	SDPAnswer string `json:"sdp_answer"`
+}
+
+// AgentAnswerResponse is the JSON response for POST /sessions/:id/agent-answer.
+type AgentAnswerResponse struct {
+	Status    string `json:"status"`
+	Recording bool   `json:"recording"`
+}
+
+// AgentReconnectRequest is the JSON body for POST /sessions/:id/agent-reconnect.
+type AgentReconnectRequest struct {
+	OldPeerID  string            `json:"old_peer_id"`
+	NewPeerID  string            `json:"new_peer_id"`
+	Role       string            `json:"role"`
+	ICEServers []ICEServerConfig `json:"ice_servers"`
+}
+
+// AgentReconnectResponse is the JSON response for POST /sessions/:id/agent-reconnect.
+type AgentReconnectResponse struct {
+	SDPOffer   string            `json:"sdp_offer"`
+	PeerID     string            `json:"peer_id"`
+	ICEServers []ICEServerConfig `json:"ice_servers"`
+}
+
+// TerminateResponse is the JSON response for POST /sessions/:id/terminate.
+type TerminateResponse struct {
+	Status            string `json:"status"`
+	RecordingFile     string `json:"recording_file,omitempty"`
+	RecordingSizeBytes int64 `json:"recording_size_bytes,omitempty"`
+	DurationSeconds   int    `json:"duration_seconds"`
+}
+
+// AddPeerRequest is the JSON body for POST /sessions/:id/peers.
+type AddPeerRequest struct {
+	PeerID     string            `json:"peer_id"`
+	Role       string            `json:"role"`
+	ICEServers []ICEServerConfig `json:"ice_servers"`
+}
+
+// ChangePeerRoleRequest is the JSON body for PATCH /sessions/:id/peers/:peer_id/role.
+type ChangePeerRoleRequest struct {
+	Role string `json:"role"`
+}
+
+// InjectAudioRequest is the JSON body for POST /sessions/:id/inject-audio.
+type InjectAudioRequest struct {
+	ID     string `json:"id"`
+	Source string `json:"source"`
+	Mode   string `json:"mode"`
+	Target string `json:"target"`
+	Loop   bool   `json:"loop"`
+}
+
+// HealthResponse is the JSON response for GET /health.
+type HealthResponse struct {
+	Status         string `json:"status"`
+	ActiveSessions int    `json:"active_sessions"`
+	UptimeSeconds  int    `json:"uptime_seconds"`
+}
+
+// --- Handlers ---
+
+// Health returns the server's health status. This endpoint does not require
+// authentication and is used by container orchestrators for liveness checks.
+func (h *Handlers) Health(w http.ResponseWriter, r *http.Request) {
+	metrics := h.manager.GetMetrics()
+	writeJSON(w, http.StatusOK, HealthResponse{
+		Status:         "ok",
+		ActiveSessions: metrics.ActiveSessions,
+		UptimeSeconds:  int(time.Since(startTime).Seconds()),
+	})
+}
+
+// Metrics returns Prometheus-compatible metrics about the media server.
+func (h *Handlers) Metrics(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, h.manager.GetMetrics())
+}
+
+// CreateSession handles POST /sessions. It creates a new call session with a
+// Meta-side peer connection and returns the SDP answer (for incoming calls)
+// or SDP offer (for outgoing calls).
+func (h *Handlers) CreateSession(w http.ResponseWriter, r *http.Request) {
+	var req CreateSessionRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+
+	if req.CallID == "" {
+		writeError(w, http.StatusBadRequest, "call_id is required")
+		return
+	}
+	if req.Direction != "incoming" && req.Direction != "outgoing" {
+		writeError(w, http.StatusBadRequest, "direction must be 'incoming' or 'outgoing'")
+		return
+	}
+
+	iceServers := toWebRTCICEServers(req.ICEServers, h.cfg)
+
+	sess, sdpResult, err := h.manager.CreateSession(req.CallID, req.AccountID, req.Direction, req.MetaSDPOffer, iceServers)
+	if err != nil {
+		slog.Error("handler: failed to create session",
+			"call_id", req.CallID,
+			"error", err,
+		)
+		writeError(w, http.StatusInternalServerError, "failed to create session: "+err.Error())
+		return
+	}
+
+	resp := CreateSessionResponse{
+		SessionID: sess.ID,
+		Status:    string(sess.Status),
+	}
+	if req.Direction == "incoming" {
+		resp.MetaSDPAnswer = sdpResult
+	} else {
+		resp.MetaSDPOffer = sdpResult
+	}
+
+	slog.Info("handler: session created",
+		"session_id", sess.ID,
+		"call_id", req.CallID,
+		"direction", req.Direction,
+	)
+
+	writeJSON(w, http.StatusCreated, resp)
+}
+
+// GetSession handles GET /sessions/{id}. It returns the current status of
+// a call session.
+func (h *Handlers) GetSession(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+	writeJSON(w, http.StatusOK, sess.GetInfo())
+}
+
+// AgentOffer handles POST /sessions/{id}/agent-offer. It creates a new
+// agent-side peer connection and returns the SDP offer to send to the
+// agent's browser.
+func (h *Handlers) AgentOffer(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	var req AgentOfferRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+
+	if req.PeerID == "" {
+		req.PeerID = fmt.Sprintf("agent_%d", time.Now().UnixNano())
+	}
+
+	role := peer.RoleActive
+	if req.Role != "" {
+		role = peer.PeerRole(req.Role)
+	}
+
+	iceServers := toWebRTCICEServers(req.ICEServers, h.cfg)
+
+	sdpOffer, err := sess.CreateAgentPeer(req.PeerID, role, iceServers)
+	if err != nil {
+		slog.Error("handler: failed to create agent peer",
+			"session_id", sessionID,
+			"error", err,
+		)
+		writeError(w, http.StatusInternalServerError, "failed to create agent peer: "+err.Error())
+		return
+	}
+
+	slog.Info("handler: agent offer created",
+		"session_id", sessionID,
+		"peer_id", req.PeerID,
+	)
+
+	writeJSON(w, http.StatusOK, AgentOfferResponse{
+		SDPOffer:   sdpOffer,
+		PeerID:     req.PeerID,
+		ICEServers: req.ICEServers,
+	})
+}
+
+// AgentAnswer handles POST /sessions/{id}/agent-answer. It sets the agent's
+// SDP answer on the peer connection, completing the WebRTC handshake and
+// enabling audio bridging.
+func (h *Handlers) AgentAnswer(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	var req AgentAnswerRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+
+	if req.SDPAnswer == "" {
+		writeError(w, http.StatusBadRequest, "sdp_answer is required")
+		return
+	}
+	// When peer_id is omitted (single-agent sessions), fall back to the only
+	// agent peer attached to the session. Rails doesn't currently surface
+	// peer_id through ActionCable, so browsers just send the SDP answer.
+	if req.PeerID == "" {
+		if only, ok := sess.SoleAgentPeerID(); ok {
+			req.PeerID = only
+		} else {
+			writeError(w, http.StatusBadRequest, "peer_id is required (session has multiple agent peers)")
+			return
+		}
+	}
+
+	if err := sess.SetAgentAnswer(req.PeerID, req.SDPAnswer); err != nil {
+		slog.Error("handler: failed to set agent answer",
+			"session_id", sessionID,
+			"peer_id", req.PeerID,
+			"error", err,
+		)
+		writeError(w, http.StatusInternalServerError, "failed to set agent answer: "+err.Error())
+		return
+	}
+
+	slog.Info("handler: agent answer set",
+		"session_id", sessionID,
+		"peer_id", req.PeerID,
+	)
+
+	writeJSON(w, http.StatusOK, AgentAnswerResponse{
+		Status:    "bridged",
+		Recording: true,
+	})
+}
+
+// MetaAnswer handles POST /sessions/{id}/meta-answer. It sets Meta's SDP
+// answer on the Meta-side peer connection for outbound calls.
+func (h *Handlers) MetaAnswer(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	var req MetaAnswerRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+	if req.SDPAnswer == "" {
+		writeError(w, http.StatusBadRequest, "sdp_answer is required")
+		return
+	}
+
+	if err := sess.SetMetaAnswer(req.SDPAnswer); err != nil {
+		slog.Error("handler: failed to set meta answer",
+			"session_id", sessionID,
+			"error", err,
+		)
+		writeError(w, http.StatusInternalServerError, "failed to set meta answer: "+err.Error())
+		return
+	}
+
+	slog.Info("handler: meta answer set", "session_id", sessionID)
+	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
+}
+
+// AgentReconnect handles POST /sessions/{id}/agent-reconnect. It tears down
+// the old agent peer and creates a new one, returning a fresh SDP offer.
+func (h *Handlers) AgentReconnect(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	var req AgentReconnectRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+
+	if req.NewPeerID == "" {
+		req.NewPeerID = fmt.Sprintf("agent_%d", time.Now().UnixNano())
+	}
+
+	role := peer.RoleActive
+	if req.Role != "" {
+		role = peer.PeerRole(req.Role)
+	}
+
+	iceServers := toWebRTCICEServers(req.ICEServers, h.cfg)
+
+	sdpOffer, err := sess.ReconnectAgent(req.OldPeerID, req.NewPeerID, role, iceServers)
+	if err != nil {
+		slog.Error("handler: failed to reconnect agent",
+			"session_id", sessionID,
+			"error", err,
+		)
+		writeError(w, http.StatusInternalServerError, "failed to reconnect agent: "+err.Error())
+		return
+	}
+
+	slog.Info("handler: agent reconnect complete",
+		"session_id", sessionID,
+		"old_peer_id", req.OldPeerID,
+		"new_peer_id", req.NewPeerID,
+	)
+
+	writeJSON(w, http.StatusOK, AgentReconnectResponse{
+		SDPOffer:   sdpOffer,
+		PeerID:     req.NewPeerID,
+		ICEServers: req.ICEServers,
+	})
+}
+
+// TerminateSession handles POST /sessions/{id}/terminate. It ends the call,
+// closes all peer connections, and finalizes the recording.
+func (h *Handlers) TerminateSession(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	info := sess.GetInfo()
+	sess.Terminate("api_request")
+
+	resp := TerminateResponse{
+		Status:          "terminated",
+		DurationSeconds: info.DurationSeconds,
+	}
+
+	if sess.Recorder != nil {
+		resp.RecordingFile = sess.RecordingFilePath()
+		resp.RecordingSizeBytes = sess.Recorder.FileSize()
+	}
+
+	slog.Info("handler: session terminated",
+		"session_id", sessionID,
+		"duration", info.DurationSeconds,
+	)
+
+	writeJSON(w, http.StatusOK, resp)
+}
+
+// GetRecording handles GET /sessions/{id}/recording. It serves the combined
+// recording file as a binary OGG download. An optional ?side=customer|agent
+// query parameter returns the per-direction recording instead, which Rails
+// uses to produce speaker-separated transcripts. Falls back to looking the
+// files up on disk when the session has already been terminated and removed
+// from the in-memory manager — Rails typically fetches per-side recordings
+// shortly after calling terminate.
+func (h *Handlers) GetRecording(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	side := r.URL.Query().Get("side")
+
+	filePath, filename := h.resolveRecordingPath(sessionID, side)
+
+	if filePath == "" {
+		writeError(w, http.StatusNotFound, "no recording available")
+		return
+	}
+
+	f, err := os.Open(filePath)
+	if err != nil {
+		writeError(w, http.StatusNotFound, "recording file not found")
+		return
+	}
+	defer f.Close()
+
+	stat, err := f.Stat()
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "failed to stat recording file")
+		return
+	}
+
+	w.Header().Set("Content-Type", "audio/ogg")
+	w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename))
+	http.ServeContent(w, r, filePath, stat.ModTime(), f)
+}
+
+// resolveRecordingPath returns the filesystem path + download filename for a
+// session's recording. It prefers the live session's recorder (for sessions
+// still in memory) and falls back to deterministic on-disk paths so Rails can
+// download recordings even after the session was terminated and evicted from
+// the manager.
+func (h *Handlers) resolveRecordingPath(sessionID, side string) (string, string) {
+	var suffix, filename string
+	switch side {
+	case "customer":
+		suffix = "_customer.ogg"
+	case "agent":
+		suffix = "_agent.ogg"
+	default:
+		suffix = ".ogg"
+	}
+	filename = sessionID + suffix
+
+	if sess := h.manager.GetSession(sessionID); sess != nil {
+		switch side {
+		case "customer":
+			return sess.RecorderCustomerPath(), filename
+		case "agent":
+			return sess.RecorderAgentPath(), filename
+		default:
+			return sess.RecordingFilePath(), filename
+		}
+	}
+
+	diskPath := filepath.Join(h.cfg.RecordingsDir, filename)
+	if _, err := os.Stat(diskPath); err == nil {
+		return diskPath, filename
+	}
+	return "", filename
+}
+
+// DeleteSession handles DELETE /sessions/{id}. It terminates the session and
+// removes all associated recording files.
+func (h *Handlers) DeleteSession(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	if err := h.manager.DeleteSession(sessionID); err != nil {
+		writeError(w, http.StatusNotFound, err.Error())
+		return
+	}
+
+	slog.Info("handler: session deleted", "session_id", sessionID)
+	writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"})
+}
+
+// AddPeer handles POST /sessions/{id}/peers. It adds a new participant peer
+// to an existing session (multi-participant support).
+func (h *Handlers) AddPeer(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	var req AddPeerRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+
+	if req.PeerID == "" {
+		req.PeerID = fmt.Sprintf("peer_%d", time.Now().UnixNano())
+	}
+
+	role := peer.RoleActive
+	if req.Role != "" {
+		role = peer.PeerRole(req.Role)
+	}
+
+	iceServers := toWebRTCICEServers(req.ICEServers, h.cfg)
+
+	sdpOffer, err := sess.CreateAgentPeer(req.PeerID, role, iceServers)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "failed to add peer: "+err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusCreated, AgentOfferResponse{
+		SDPOffer:   sdpOffer,
+		PeerID:     req.PeerID,
+		ICEServers: req.ICEServers,
+	})
+}
+
+// RemovePeer handles DELETE /sessions/{id}/peers/{peer_id}. It removes a
+// specific participant from the session.
+func (h *Handlers) RemovePeer(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	peerID := r.PathValue("peer_id")
+
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	if err := sess.RemoveAgentPeer(peerID); err != nil {
+		writeError(w, http.StatusNotFound, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]string{"status": "removed"})
+}
+
+// ChangePeerRole handles PATCH /sessions/{id}/peers/{peer_id}/role. It
+// changes the role of a connected participant.
+func (h *Handlers) ChangePeerRole(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	peerID := r.PathValue("peer_id")
+
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	var req ChangePeerRoleRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+
+	if err := sess.ChangeAgentRole(peerID, peer.PeerRole(req.Role)); err != nil {
+		writeError(w, http.StatusNotFound, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]string{"status": "updated", "role": req.Role})
+}
+
+// InjectAudio handles POST /sessions/{id}/inject-audio. It starts playing
+// an audio file into the call's RTP stream.
+func (h *Handlers) InjectAudio(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	var req InjectAudioRequest
+	if err := readJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
+		return
+	}
+
+	if req.ID == "" {
+		req.ID = fmt.Sprintf("inj_%d", time.Now().UnixNano())
+	}
+	if req.Mode == "" {
+		req.Mode = "replace"
+	}
+	if req.Target == "" {
+		req.Target = "meta"
+	}
+
+	injector := media.NewInjector(req.ID, req.Source, req.Mode, req.Target, req.Loop)
+
+	// Determine the target track based on the target parameter.
+	target := sess.GetInjectorTarget(req.Target)
+	if target == nil {
+		writeError(w, http.StatusBadRequest, "target track not available")
+		return
+	}
+
+	if err := injector.Start(target); err != nil {
+		writeError(w, http.StatusInternalServerError, "failed to start injection: "+err.Error())
+		return
+	}
+
+	sess.AddInjector(req.ID, injector)
+
+	writeJSON(w, http.StatusCreated, map[string]string{
+		"id":     req.ID,
+		"status": "started",
+	})
+}
+
+// StopInjectAudio handles DELETE /sessions/{id}/inject-audio/{inj_id}. It
+// stops an active audio injection.
+func (h *Handlers) StopInjectAudio(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.PathValue("id")
+	injID := r.PathValue("inj_id")
+
+	sess := h.manager.GetSession(sessionID)
+	if sess == nil {
+		writeError(w, http.StatusNotFound, "session not found")
+		return
+	}
+
+	if err := sess.StopInjector(injID); err != nil {
+		writeError(w, http.StatusNotFound, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]string{"status": "stopped"})
+}
+
+// --- Helpers ---
+
+func readJSON(r *http.Request, v any) error {
+	defer r.Body.Close()
+	limited := io.LimitReader(r.Body, maxRequestBodySize)
+	err := json.NewDecoder(limited).Decode(v)
+	// Tolerate empty bodies — handlers with all-optional fields treat this
+	// as "use defaults" rather than failing.
+	if err == io.EOF {
+		return nil
+	}
+	return err
+}
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	json.NewEncoder(w).Encode(v)
+}
+
+func writeError(w http.ResponseWriter, status int, msg string) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	json.NewEncoder(w).Encode(map[string]string{"error": msg})
+}
+
+// toWebRTCICEServers converts the request ICE server configs to Pion's
+// ICEServer type, merging with any STUN/TURN servers from the global config.
+func toWebRTCICEServers(reqServers []ICEServerConfig, cfg *config.Config) []webrtc.ICEServer {
+	servers := make([]webrtc.ICEServer, 0, len(reqServers)+2)
+
+	// Add request-provided servers.
+	for _, s := range reqServers {
+		server := webrtc.ICEServer{URLs: s.URLs}
+		if s.Username != "" {
+			server.Username = s.Username
+			server.Credential = s.Credential
+			server.CredentialType = webrtc.ICECredentialTypePassword
+		}
+		servers = append(servers, server)
+	}
+
+	// Add global STUN servers if no STUN was provided in the request.
+	hasSTUN := false
+	for _, s := range reqServers {
+		for _, u := range s.URLs {
+			if strings.HasPrefix(u, "stun:") {
+				hasSTUN = true
+				break
+			}
+		}
+	}
+	if !hasSTUN && len(cfg.STUNServers) > 0 {
+		servers = append(servers, webrtc.ICEServer{URLs: cfg.STUNServers})
+	}
+
+	// Add global TURN servers.
+	if len(cfg.TURNServers) > 0 && cfg.TURNUsername != "" {
+		servers = append(servers, webrtc.ICEServer{
+			URLs:           cfg.TURNServers,
+			Username:       cfg.TURNUsername,
+			Credential:     cfg.TURNPassword,
+			CredentialType: webrtc.ICECredentialTypePassword,
+		})
+	}
+
+	return servers
+}

enterprise/media-server/internal/server/router.go

@@ -0,0 +1,117 @@
+// Package server provides the HTTP API router and handlers for the media server.
+package server
+
+import (
+	"log/slog"
+	"net/http"
+	"strings"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/auth"
+	"github.com/chatwoot/chatwoot-media-server/internal/config"
+	"github.com/chatwoot/chatwoot-media-server/internal/session"
+)
+
+// Router builds the HTTP handler tree with authentication middleware and
+// route matching. It uses the standard library's http.ServeMux for routing.
+type Router struct {
+	handler  *Handlers
+	authToken string
+}
+
+// NewRouter creates a new Router with the given configuration, session
+// manager, and authentication token.
+func NewRouter(cfg *config.Config, mgr *session.Manager) *Router {
+	return &Router{
+		handler:  NewHandlers(cfg, mgr),
+		authToken: cfg.AuthToken,
+	}
+}
+
+// Build constructs and returns the root http.Handler with all routes and
+// middleware applied.
+func (rt *Router) Build() http.Handler {
+	mux := http.NewServeMux()
+
+	// Public endpoints (no auth required).
+	mux.HandleFunc("GET /health", rt.handler.Health)
+
+	// Protected endpoints.
+	authMw := auth.Middleware(rt.authToken)
+
+	// Session CRUD.
+	mux.Handle("POST /sessions", authMw(http.HandlerFunc(rt.handler.CreateSession)))
+	mux.Handle("GET /metrics", authMw(http.HandlerFunc(rt.handler.Metrics)))
+
+	// All session-scoped routes go through a path-parsing handler because
+	// Go 1.22's ServeMux supports {param} patterns.
+	mux.Handle("GET /sessions/{id}", authMw(http.HandlerFunc(rt.handler.GetSession)))
+	mux.Handle("POST /sessions/{id}/agent-offer", authMw(http.HandlerFunc(rt.handler.AgentOffer)))
+	mux.Handle("POST /sessions/{id}/agent-answer", authMw(http.HandlerFunc(rt.handler.AgentAnswer)))
+	mux.Handle("POST /sessions/{id}/meta-answer", authMw(http.HandlerFunc(rt.handler.MetaAnswer)))
+	mux.Handle("POST /sessions/{id}/agent-reconnect", authMw(http.HandlerFunc(rt.handler.AgentReconnect)))
+	mux.Handle("POST /sessions/{id}/terminate", authMw(http.HandlerFunc(rt.handler.TerminateSession)))
+	mux.Handle("GET /sessions/{id}/recording", authMw(http.HandlerFunc(rt.handler.GetRecording)))
+	mux.Handle("DELETE /sessions/{id}", authMw(http.HandlerFunc(rt.handler.DeleteSession)))
+
+	// Multi-participant peer management.
+	mux.Handle("POST /sessions/{id}/peers", authMw(http.HandlerFunc(rt.handler.AddPeer)))
+	mux.Handle("DELETE /sessions/{id}/peers/{peer_id}", authMw(http.HandlerFunc(rt.handler.RemovePeer)))
+	mux.Handle("PATCH /sessions/{id}/peers/{peer_id}/role", authMw(http.HandlerFunc(rt.handler.ChangePeerRole)))
+
+	// Audio injection.
+	mux.Handle("POST /sessions/{id}/inject-audio", authMw(http.HandlerFunc(rt.handler.InjectAudio)))
+	mux.Handle("DELETE /sessions/{id}/inject-audio/{inj_id}", authMw(http.HandlerFunc(rt.handler.StopInjectAudio)))
+
+	// Wrap the mux with global middleware.
+	var handler http.Handler = mux
+	handler = requestLogger(handler)
+	handler = recoverer(handler)
+
+	return handler
+}
+
+// requestLogger is middleware that logs each HTTP request.
+func requestLogger(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Skip logging for health checks to reduce noise.
+		if strings.HasPrefix(r.URL.Path, "/health") {
+			next.ServeHTTP(w, r)
+			return
+		}
+
+		rw := &responseWriter{ResponseWriter: w, statusCode: http.StatusOK}
+		next.ServeHTTP(rw, r)
+
+		// Logging is handled inside handlers for better context; this is
+		// a safety net for unlogged requests.
+	})
+}
+
+// recoverer is middleware that catches panics and returns a 500 response
+// instead of crashing the server.
+func recoverer(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer func() {
+			if rec := recover(); rec != nil {
+				slog.Error("panic recovered",
+					"panic", rec,
+					"path", r.URL.Path,
+					"method", r.Method,
+				)
+				http.Error(w, `{"error":"internal server error"}`, http.StatusInternalServerError)
+			}
+		}()
+		next.ServeHTTP(w, r)
+	})
+}
+
+// responseWriter wraps http.ResponseWriter to capture the status code.
+type responseWriter struct {
+	http.ResponseWriter
+	statusCode int
+}
+
+func (rw *responseWriter) WriteHeader(code int) {
+	rw.statusCode = code
+	rw.ResponseWriter.WriteHeader(code)
+}

enterprise/media-server/internal/session/manager.go

@@ -0,0 +1,292 @@
+package session
+
+import (
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/pion/webrtc/v4"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/callback"
+	"github.com/chatwoot/chatwoot-media-server/internal/config"
+	"github.com/chatwoot/chatwoot-media-server/internal/peer"
+)
+
+// Manager handles the lifecycle of all call sessions, providing thread-safe
+// creation, lookup, termination, and periodic cleanup of expired sessions.
+type Manager struct {
+	sessions    map[string]*Session
+	config      *config.Config
+	railsClient *callback.RailsClient
+
+	mu              sync.RWMutex
+	sessionCounter  atomic.Int64
+	cleanupTicker   *time.Ticker
+	cleanupStopCh   chan struct{}
+}
+
+// Metrics holds observable counters for the session manager, used by the
+// /metrics endpoint.
+type Metrics struct {
+	ActiveSessions    int   `json:"active_sessions"`
+	TotalCreated      int64 `json:"total_created"`
+	TerminatedCount   int   `json:"terminated_count"`
+	MetaConnected     int   `json:"meta_connected"`
+	AgentConnected    int   `json:"agent_connected"`
+	AgentDisconnected int   `json:"agent_disconnected"`
+}
+
+// NewManager creates a new session manager and starts a background goroutine
+// that periodically cleans up expired sessions.
+func NewManager(cfg *config.Config, railsClient *callback.RailsClient) *Manager {
+	m := &Manager{
+		sessions:      make(map[string]*Session),
+		config:        cfg,
+		railsClient:   railsClient,
+		cleanupTicker: time.NewTicker(60 * time.Second),
+		cleanupStopCh: make(chan struct{}),
+	}
+
+	go m.cleanupLoop()
+	return m
+}
+
+// CreateSession creates a new call session with the given parameters. For
+// incoming calls, metaSDPOffer contains Meta's SDP offer and the returned
+// string is the SDP answer. For outgoing calls, metaSDPOffer is empty and
+// the returned string is the SDP offer to send to Meta.
+func (m *Manager) CreateSession(callID, accountID, direction, metaSDPOffer string, iceServers []webrtc.ICEServer) (*Session, string, error) {
+	// Check capacity limit.
+	if m.config.MaxConcurrentSessions > 0 {
+		m.mu.RLock()
+		activeCount := len(m.sessions)
+		m.mu.RUnlock()
+
+		if activeCount >= m.config.MaxConcurrentSessions {
+			return nil, "", fmt.Errorf("max concurrent sessions (%d) reached", m.config.MaxConcurrentSessions)
+		}
+	}
+
+	// Generate a unique session ID.
+	counter := m.sessionCounter.Add(1)
+	sessionID := fmt.Sprintf("sess_%s_%d", time.Now().Format("20060102150405"), counter)
+
+	sess, sdpResult, err := NewSession(m.config, m.railsClient, sessionID, callID, accountID, direction, metaSDPOffer, iceServers)
+	if err != nil {
+		return nil, "", fmt.Errorf("create session: %w", err)
+	}
+
+	m.mu.Lock()
+	m.sessions[sessionID] = sess
+	m.mu.Unlock()
+
+	slog.Info("manager: session created",
+		"session_id", sessionID,
+		"call_id", callID,
+		"direction", direction,
+	)
+
+	return sess, sdpResult, nil
+}
+
+// GetSession returns the session with the given ID, or nil if not found.
+func (m *Manager) GetSession(sessionID string) *Session {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.sessions[sessionID]
+}
+
+// TerminateSession terminates the session with the given ID and removes it
+// from the active sessions map.
+func (m *Manager) TerminateSession(sessionID, reason string) error {
+	m.mu.Lock()
+	sess, ok := m.sessions[sessionID]
+	if !ok {
+		m.mu.Unlock()
+		return fmt.Errorf("session %s not found", sessionID)
+	}
+	delete(m.sessions, sessionID)
+	m.mu.Unlock()
+
+	sess.Terminate(reason)
+	return nil
+}
+
+// DeleteSession removes a session and cleans up its recording files.
+func (m *Manager) DeleteSession(sessionID string) error {
+	m.mu.Lock()
+	sess, ok := m.sessions[sessionID]
+	if ok {
+		delete(m.sessions, sessionID)
+	}
+	m.mu.Unlock()
+
+	if !ok {
+		return fmt.Errorf("session %s not found", sessionID)
+	}
+
+	sess.Terminate("deleted")
+
+	// Clean up recording files.
+	if sess.Recorder != nil {
+		sess.Recorder.Cleanup()
+	}
+
+	return nil
+}
+
+// CreateAgentPeer creates a new agent-side peer for the specified session.
+func (m *Manager) CreateAgentPeer(sessionID, peerID string, role peer.PeerRole, iceServers []webrtc.ICEServer) (string, error) {
+	sess := m.GetSession(sessionID)
+	if sess == nil {
+		return "", fmt.Errorf("session %s not found", sessionID)
+	}
+	return sess.CreateAgentPeer(peerID, role, iceServers)
+}
+
+// SetAgentAnswer sets the agent's SDP answer for the specified peer.
+func (m *Manager) SetAgentAnswer(sessionID, peerID, sdpAnswer string) error {
+	sess := m.GetSession(sessionID)
+	if sess == nil {
+		return fmt.Errorf("session %s not found", sessionID)
+	}
+	return sess.SetAgentAnswer(peerID, sdpAnswer)
+}
+
+// ReconnectAgent creates a new agent peer after tearing down the old one.
+func (m *Manager) ReconnectAgent(sessionID, oldPeerID, newPeerID string, role peer.PeerRole, iceServers []webrtc.ICEServer) (string, error) {
+	sess := m.GetSession(sessionID)
+	if sess == nil {
+		return "", fmt.Errorf("session %s not found", sessionID)
+	}
+	return sess.ReconnectAgent(oldPeerID, newPeerID, role, iceServers)
+}
+
+// GetMetrics returns current observable metrics about the session manager.
+func (m *Manager) GetMetrics() Metrics {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	metrics := Metrics{
+		ActiveSessions: len(m.sessions),
+		TotalCreated:   m.sessionCounter.Load(),
+	}
+
+	for _, sess := range m.sessions {
+		switch sess.Status {
+		case StatusTerminated:
+			metrics.TerminatedCount++
+		case StatusMetaConnected:
+			metrics.MetaConnected++
+		case StatusAgentConnected, StatusActive:
+			metrics.AgentConnected++
+		case StatusAgentDisconnected:
+			metrics.AgentDisconnected++
+		}
+	}
+
+	return metrics
+}
+
+// RecoverOrphanedRecordings scans the recordings directory for files that
+// do not belong to any active session, reporting them to Rails. This handles
+// the case where the media server crashed mid-call and recordings were left
+// on disk.
+func (m *Manager) RecoverOrphanedRecordings() {
+	dir := m.config.RecordingsDir
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return
+		}
+		slog.Error("manager: failed to scan recordings directory",
+			"dir", dir,
+			"error", err,
+		)
+		return
+	}
+
+	m.mu.RLock()
+	activeIDs := make(map[string]bool, len(m.sessions))
+	for id := range m.sessions {
+		activeIDs[id] = true
+	}
+	m.mu.RUnlock()
+
+	orphanCount := 0
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".ogg") {
+			continue
+		}
+
+		// Extract session ID from filename (e.g., "sess_20240101_1.ogg" -> "sess_20240101_1").
+		name := strings.TrimSuffix(entry.Name(), ".ogg")
+		// Remove channel suffixes.
+		name = strings.TrimSuffix(name, "_customer")
+		name = strings.TrimSuffix(name, "_agent")
+
+		if !activeIDs[name] {
+			orphanCount++
+			slog.Warn("manager: found orphaned recording",
+				"file", filepath.Join(dir, entry.Name()),
+				"session_id", name,
+			)
+		}
+	}
+
+	if orphanCount > 0 {
+		slog.Info("manager: orphaned recording scan complete",
+			"orphan_count", orphanCount,
+		)
+	}
+}
+
+// Shutdown gracefully terminates all active sessions and stops the cleanup
+// goroutine. It should be called during server shutdown.
+func (m *Manager) Shutdown() {
+	close(m.cleanupStopCh)
+	m.cleanupTicker.Stop()
+
+	m.mu.Lock()
+	sessions := make([]*Session, 0, len(m.sessions))
+	for _, sess := range m.sessions {
+		sessions = append(sessions, sess)
+	}
+	m.sessions = make(map[string]*Session)
+	m.mu.Unlock()
+
+	for _, sess := range sessions {
+		sess.Terminate("server_shutdown")
+	}
+
+	slog.Info("manager: all sessions terminated", "count", len(sessions))
+}
+
+// cleanupLoop runs periodically to remove terminated sessions from the map.
+func (m *Manager) cleanupLoop() {
+	for {
+		select {
+		case <-m.cleanupStopCh:
+			return
+		case <-m.cleanupTicker.C:
+			m.cleanup()
+		}
+	}
+}
+
+func (m *Manager) cleanup() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	for id, sess := range m.sessions {
+		if sess.Status == StatusTerminated {
+			delete(m.sessions, id)
+			slog.Debug("manager: cleaned up terminated session", "session_id", id)
+		}
+	}
+}

enterprise/media-server/internal/session/session.go

@@ -0,0 +1,595 @@
+// Package session manages call session lifecycles, coordinating between the
+// Meta-side peer, agent-side peers, audio bridge, and recording.
+package session
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/pion/webrtc/v4"
+
+	"github.com/chatwoot/chatwoot-media-server/internal/callback"
+	"github.com/chatwoot/chatwoot-media-server/internal/config"
+	"github.com/chatwoot/chatwoot-media-server/internal/media"
+	"github.com/chatwoot/chatwoot-media-server/internal/peer"
+)
+
+// Status represents the current state of a call session.
+type Status string
+
+const (
+	StatusCreated           Status = "created"
+	StatusMetaConnected     Status = "meta_connected"
+	StatusAgentConnected    Status = "agent_connected"
+	StatusActive            Status = "active"
+	StatusAgentDisconnected Status = "agent_disconnected"
+	StatusTerminated        Status = "terminated"
+)
+
+// Session represents a single active call, holding the Meta-side peer
+// connection (Peer A), one or more agent-side peer connections (Peer B),
+// the audio bridge, and the recording engine.
+type Session struct {
+	ID        string
+	CallID    string
+	AccountID string
+	Direction string // "incoming" or "outgoing"
+
+	MetaPeer   *peer.MetaPeer
+	AgentPeers map[string]*peer.AgentPeer
+	Bridge     *media.Bridge
+	Recorder   *media.Recorder
+	Injectors  map[string]*media.Injector
+
+	Status    Status
+	StartedAt time.Time
+	CreatedAt time.Time
+
+	config         *config.Config
+	railsClient    *callback.RailsClient
+	reconnectTimer *time.Timer
+	cancel         context.CancelFunc
+	ctx            context.Context
+
+	mu sync.Mutex
+}
+
+// Info is the JSON-serializable representation of a session's current state,
+// returned by the GET /sessions/:id endpoint.
+type Info struct {
+	ID              string `json:"id"`
+	CallID          string `json:"call_id"`
+	AccountID       string `json:"account_id"`
+	Direction       string `json:"direction"`
+	Status          string `json:"status"`
+	MetaICEState    string `json:"meta_ice_state"`
+	AgentPeerCount  int    `json:"agent_peer_count"`
+	DurationSeconds int    `json:"duration_seconds"`
+	HasRecording    bool   `json:"has_recording"`
+	CreatedAt       string `json:"created_at"`
+}
+
+// NewSession creates a new call session. For incoming calls, the Meta SDP
+// offer is provided and the method returns the SDP answer. For outgoing calls,
+// the Meta SDP offer is empty and the method returns an SDP offer to send to
+// Meta.
+func NewSession(
+	cfg *config.Config,
+	railsClient *callback.RailsClient,
+	id, callID, accountID, direction, metaSDPOffer string,
+	iceServers []webrtc.ICEServer,
+) (*Session, string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), cfg.MaxSessionDuration)
+
+	sess := &Session{
+		ID:         id,
+		CallID:     callID,
+		AccountID:  accountID,
+		Direction:  direction,
+		AgentPeers: make(map[string]*peer.AgentPeer),
+		Injectors:  make(map[string]*media.Injector),
+		Status:     StatusCreated,
+		CreatedAt:  time.Now(),
+		config:     cfg,
+		railsClient: railsClient,
+		ctx:        ctx,
+		cancel:     cancel,
+	}
+
+	// Create the Meta-side peer connection (Peer A).
+	metaPeer, sdpResult, err := peer.NewMetaPeer(cfg, metaSDPOffer, iceServers)
+	if err != nil {
+		cancel()
+		return nil, "", fmt.Errorf("create meta peer: %w", err)
+	}
+	sess.MetaPeer = metaPeer
+
+	// Create the recorder.
+	recorder, err := media.NewRecorder(id, cfg.RecordingsDir)
+	if err != nil {
+		metaPeer.Close()
+		cancel()
+		return nil, "", fmt.Errorf("create recorder: %w", err)
+	}
+	sess.Recorder = recorder
+
+	// Create the audio bridge.
+	sess.Bridge = media.NewBridge(id, metaPeer, recorder)
+
+	// Wire up Meta peer event handlers.
+	metaPeer.OnICEStateChange(func(state webrtc.ICEConnectionState) {
+		sess.handleMetaICEStateChange(state)
+	})
+
+	metaPeer.OnTrackReady(func(track *webrtc.TrackRemote) {
+		slog.Info("session: Meta audio track ready, starting bridge forwarding",
+			"session_id", id,
+		)
+		// Start forwarding Meta audio to agents in its own goroutine.
+		go sess.Bridge.ReadAndForwardMetaTrack(sess.ctx, track)
+	})
+
+	// Start the max duration timer.
+	go func() {
+		<-ctx.Done()
+		sess.mu.Lock()
+		if sess.Status != StatusTerminated {
+			sess.mu.Unlock()
+			slog.Info("session: max duration reached, terminating",
+				"session_id", id,
+			)
+			sess.Terminate("max_duration_exceeded")
+		} else {
+			sess.mu.Unlock()
+		}
+	}()
+
+	return sess, sdpResult, nil
+}
+
+// SetMetaAnswer sets Meta's SDP answer on the Meta peer connection. This is
+// used for outbound calls when Meta responds to the server's SDP offer.
+func (s *Session) SetMetaAnswer(sdpAnswer string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if err := s.MetaPeer.SetRemoteAnswer(sdpAnswer); err != nil {
+		return fmt.Errorf("set meta answer: %w", err)
+	}
+	return nil
+}
+
+// CreateAgentPeer creates a new agent-side peer connection (Peer B) and
+// returns the SDP offer to send to the agent's browser.
+func (s *Session) CreateAgentPeer(peerID string, role peer.PeerRole, iceServers []webrtc.ICEServer) (string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	agentPeer, sdpOffer, err := peer.NewAgentPeer(s.config, peerID, role, iceServers)
+	if err != nil {
+		return "", fmt.Errorf("create agent peer: %w", err)
+	}
+
+	// Wire up agent peer event handlers.
+	agentPeer.OnICEStateChange(func(state webrtc.ICEConnectionState) {
+		s.handleAgentICEStateChange(peerID, state)
+	})
+
+	agentPeer.OnTrackReady(func(track *webrtc.TrackRemote) {
+		slog.Info("session: agent audio track ready, starting bridge forwarding",
+			"session_id", s.ID,
+			"peer_id", peerID,
+		)
+		go s.Bridge.ReadAndForwardAgentTrack(s.ctx, agentPeer, track)
+	})
+
+	s.AgentPeers[peerID] = agentPeer
+	s.Bridge.AddAgentPeer(agentPeer)
+
+	// Start the bridge if Meta is already connected.
+	if s.Status == StatusMetaConnected || s.Status == StatusAgentDisconnected {
+		s.Bridge.Start(s.ctx)
+		s.Status = StatusAgentConnected
+	}
+
+	// Cancel any active reconnect timer.
+	if s.reconnectTimer != nil {
+		s.reconnectTimer.Stop()
+		s.reconnectTimer = nil
+	}
+
+	return sdpOffer, nil
+}
+
+// SoleAgentPeerID returns the peer_id when the session has exactly one agent
+// peer. Used by the agent-answer handler as a fallback for clients that don't
+// track peer ids.
+func (s *Session) SoleAgentPeerID() (string, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if len(s.AgentPeers) != 1 {
+		return "", false
+	}
+	for id := range s.AgentPeers {
+		return id, true
+	}
+	return "", false
+}
+
+// SetAgentAnswer sets the agent browser's SDP answer on the specified agent
+// peer, completing the WebRTC handshake.
+func (s *Session) SetAgentAnswer(peerID, sdpAnswer string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	ap, ok := s.AgentPeers[peerID]
+	if !ok {
+		return fmt.Errorf("agent peer %s not found", peerID)
+	}
+
+	if err := ap.SetAnswer(sdpAnswer); err != nil {
+		return fmt.Errorf("set agent answer: %w", err)
+	}
+
+	return nil
+}
+
+// ReconnectAgent tears down the old agent peer connection and creates a new
+// one, returning a fresh SDP offer. This is used when the agent reloads the
+// page and needs to re-establish their peer connection while the Meta-side
+// connection stays alive.
+func (s *Session) ReconnectAgent(oldPeerID, newPeerID string, role peer.PeerRole, iceServers []webrtc.ICEServer) (string, error) {
+	s.mu.Lock()
+	// Remove the old agent peer from the session if it exists.
+	var oldPeer *peer.AgentPeer
+	if ap, ok := s.AgentPeers[oldPeerID]; ok {
+		oldPeer = ap
+		s.Bridge.RemoveAgentPeer(oldPeerID)
+		delete(s.AgentPeers, oldPeerID)
+	}
+	s.mu.Unlock()
+
+	// Close outside the lock to avoid deadlock from ICE state callbacks.
+	if oldPeer != nil {
+		oldPeer.Close()
+	}
+
+	// Create a new agent peer.
+	return s.CreateAgentPeer(newPeerID, role, iceServers)
+}
+
+// Terminate gracefully ends the call session. It closes all peer connections,
+// finalizes the recording, and sends callbacks to Rails.
+func (s *Session) Terminate(reason string) {
+	s.mu.Lock()
+	if s.Status == StatusTerminated {
+		s.mu.Unlock()
+		return
+	}
+	s.Status = StatusTerminated
+
+	// Collect injectors and agent peers while holding the lock, then release
+	// before calling Close(). PeerConnection.Close() may fire ICE state
+	// callbacks synchronously, which would deadlock if we held s.mu.
+	injectors := make([]*media.Injector, 0, len(s.Injectors))
+	for _, inj := range s.Injectors {
+		injectors = append(injectors, inj)
+	}
+
+	agentPeers := make([]*peer.AgentPeer, 0, len(s.AgentPeers))
+	for id, ap := range s.AgentPeers {
+		agentPeers = append(agentPeers, ap)
+		delete(s.AgentPeers, id)
+	}
+
+	metaPeer := s.MetaPeer
+	s.mu.Unlock()
+
+	slog.Info("session: terminating",
+		"session_id", s.ID,
+		"reason", reason,
+	)
+
+	// Stop the bridge.
+	s.Bridge.Stop()
+
+	// Stop any active injectors.
+	for _, inj := range injectors {
+		inj.Stop()
+	}
+
+	// Close all agent peers (may trigger ICE state callbacks).
+	for _, ap := range agentPeers {
+		ap.Close()
+	}
+
+	// Close Meta peer.
+	if metaPeer != nil {
+		metaPeer.Close()
+	}
+
+	// Finalize recording.
+	if s.Recorder != nil {
+		if err := s.Recorder.Finalize(); err != nil {
+			slog.Error("session: failed to finalize recording",
+				"session_id", s.ID,
+				"error", err,
+			)
+		}
+	}
+
+	// Cancel the session context.
+	if s.cancel != nil {
+		s.cancel()
+	}
+
+	// Notify Rails.
+	s.sendTerminationCallbacks(reason)
+}
+
+// RemoveAgentPeer removes a specific agent peer from the session.
+func (s *Session) RemoveAgentPeer(peerID string) error {
+	s.mu.Lock()
+	ap, ok := s.AgentPeers[peerID]
+	if !ok {
+		s.mu.Unlock()
+		return fmt.Errorf("agent peer %s not found", peerID)
+	}
+
+	s.Bridge.RemoveAgentPeer(peerID)
+	delete(s.AgentPeers, peerID)
+	s.mu.Unlock()
+
+	// Close outside the lock to avoid deadlock from ICE state callbacks.
+	ap.Close()
+	return nil
+}
+
+// ChangeAgentRole changes the role of a connected agent peer.
+func (s *Session) ChangeAgentRole(peerID string, newRole peer.PeerRole) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	ap, ok := s.AgentPeers[peerID]
+	if !ok {
+		return fmt.Errorf("agent peer %s not found", peerID)
+	}
+
+	ap.Role = newRole
+	return nil
+}
+
+// GetInfo returns a snapshot of the session's current state.
+func (s *Session) GetInfo() Info {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	info := Info{
+		ID:             s.ID,
+		CallID:         s.CallID,
+		AccountID:      s.AccountID,
+		Direction:      s.Direction,
+		Status:         string(s.Status),
+		AgentPeerCount: len(s.AgentPeers),
+		HasRecording:   s.Recorder != nil,
+		CreatedAt:      s.CreatedAt.UTC().Format(time.RFC3339),
+	}
+
+	if s.MetaPeer != nil {
+		info.MetaICEState = s.MetaPeer.ICEConnectionState().String()
+	}
+
+	if !s.StartedAt.IsZero() {
+		info.DurationSeconds = int(time.Since(s.StartedAt).Seconds())
+	}
+
+	return info
+}
+
+// RecordingFilePath returns the path to the combined recording file.
+func (s *Session) RecordingFilePath() string {
+	if s.Recorder == nil {
+		return ""
+	}
+	return s.Recorder.CombinedFilePath()
+}
+
+// RecorderCustomerPath returns the path to the customer-only recording.
+func (s *Session) RecorderCustomerPath() string {
+	if s.Recorder == nil {
+		return ""
+	}
+	return s.Recorder.CustomerFilePath()
+}
+
+// RecorderAgentPath returns the path to the agent-only recording.
+func (s *Session) RecorderAgentPath() string {
+	if s.Recorder == nil {
+		return ""
+	}
+	return s.Recorder.AgentFilePath()
+}
+
+// GetInjectorTarget returns the appropriate write target for audio injection
+// based on the target parameter. Returns nil if the target is unavailable.
+func (s *Session) GetInjectorTarget(target string) media.InjectorTarget {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.MetaPeer == nil {
+		return nil
+	}
+	// All injection targets currently route to the Meta peer's local track.
+	return s.MetaPeer.LocalTrack()
+}
+
+// AddInjector registers an active injector with the session in a thread-safe manner.
+func (s *Session) AddInjector(id string, inj *media.Injector) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.Injectors[id] = inj
+}
+
+// StopInjector stops and removes an injector by ID. Returns an error if not found.
+func (s *Session) StopInjector(id string) error {
+	s.mu.Lock()
+	inj, ok := s.Injectors[id]
+	if !ok {
+		s.mu.Unlock()
+		return fmt.Errorf("injector %s not found", id)
+	}
+	delete(s.Injectors, id)
+	s.mu.Unlock()
+
+	inj.Stop()
+	return nil
+}
+
+func (s *Session) handleMetaICEStateChange(state webrtc.ICEConnectionState) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.Status == StatusTerminated {
+		return
+	}
+
+	switch state {
+	case webrtc.ICEConnectionStateConnected:
+		if s.Status == StatusCreated {
+			s.Status = StatusMetaConnected
+			s.StartedAt = time.Now()
+			slog.Info("session: Meta peer connected",
+				"session_id", s.ID,
+			)
+		}
+
+	case webrtc.ICEConnectionStateFailed, webrtc.ICEConnectionStateDisconnected:
+		slog.Warn("session: Meta peer disconnected/failed",
+			"session_id", s.ID,
+			"state", state.String(),
+		)
+		// Meta disconnecting means the call is over.
+		go s.Terminate("meta_disconnected")
+
+	case webrtc.ICEConnectionStateClosed:
+		slog.Info("session: Meta peer closed", "session_id", s.ID)
+	}
+}
+
+func (s *Session) handleAgentICEStateChange(peerID string, state webrtc.ICEConnectionState) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.Status == StatusTerminated {
+		return
+	}
+
+	switch state {
+	case webrtc.ICEConnectionStateConnected:
+		slog.Info("session: agent peer connected",
+			"session_id", s.ID,
+			"peer_id", peerID,
+		)
+		if s.Status == StatusMetaConnected || s.Status == StatusAgentDisconnected {
+			s.Status = StatusActive
+			// Start bridge if not already running.
+			s.Bridge.Start(s.ctx)
+		}
+
+	case webrtc.ICEConnectionStateFailed, webrtc.ICEConnectionStateDisconnected:
+		slog.Warn("session: agent peer disconnected/failed",
+			"session_id", s.ID,
+			"peer_id", peerID,
+			"state", state.String(),
+		)
+
+		// Check if any other agent peers are still connected.
+		hasConnected := false
+		for id, ap := range s.AgentPeers {
+			if id != peerID && ap.ICEConnectionState() == webrtc.ICEConnectionStateConnected {
+				hasConnected = true
+				break
+			}
+		}
+
+		if !hasConnected && s.Status != StatusTerminated {
+			s.Status = StatusAgentDisconnected
+
+			// Start reconnect timer.
+			s.reconnectTimer = time.AfterFunc(s.config.ReconnectTimeout, func() {
+				slog.Info("session: reconnect timeout expired, terminating",
+					"session_id", s.ID,
+				)
+				s.Terminate("agent_reconnect_timeout")
+			})
+
+			// Notify Rails of agent disconnect.
+			go func() {
+				if s.railsClient != nil {
+					err := s.railsClient.NotifyAgentDisconnected(context.Background(), callback.AgentDisconnectedPayload{
+						SessionID: s.ID,
+						CallID:    s.CallID,
+						Reason:    state.String(),
+					})
+					if err != nil {
+						slog.Error("session: failed to notify Rails of agent disconnect",
+							"session_id", s.ID,
+							"error", err,
+						)
+					}
+				}
+			}()
+		}
+
+	case webrtc.ICEConnectionStateClosed:
+		slog.Info("session: agent peer closed",
+			"session_id", s.ID,
+			"peer_id", peerID,
+		)
+	}
+}
+
+func (s *Session) sendTerminationCallbacks(reason string) {
+	if s.railsClient == nil {
+		return
+	}
+
+	ctx := context.Background()
+	durationSec := 0
+	if !s.StartedAt.IsZero() {
+		durationSec = int(time.Since(s.StartedAt).Seconds())
+	}
+
+	// Notify session terminated.
+	if err := s.railsClient.NotifySessionTerminated(ctx, callback.SessionTerminatedPayload{
+		SessionID:   s.ID,
+		CallID:      s.CallID,
+		Reason:      reason,
+		DurationSec: durationSec,
+	}); err != nil {
+		slog.Error("session: failed to notify Rails of termination",
+			"session_id", s.ID,
+			"error", err,
+		)
+	}
+
+	// Notify recording ready if we have one.
+	if s.Recorder != nil && s.Recorder.FileSize() > 0 {
+		if err := s.railsClient.NotifyRecordingReady(ctx, callback.RecordingReadyPayload{
+			SessionID:     s.ID,
+			CallID:        s.CallID,
+			FilePath:      s.Recorder.CombinedFilePath(),
+			DurationSec:   durationSec,
+			FileSizeBytes: s.Recorder.FileSize(),
+		}); err != nil {
+			slog.Error("session: failed to notify Rails of recording",
+				"session_id", s.ID,
+				"error", err,
+			)
+		}
+	}
+}

spec/jobs/webhooks/whatsapp_events_job_spec.rb

@@ -31,7 +31,7 @@ RSpec.describe Webhooks::WhatsappEventsJob do
   it 'enqueues the job' do
     expect { job.perform_later(params) }.to have_enqueued_job(described_class)
       .with(params)
-      .on_queue('low')
+      .on_queue('default')
   end
 
   context 'when whatsapp_cloud provider' do

spec/services/whatsapp/providers/whatsapp_cloud_service_spec.rb

@@ -25,7 +25,7 @@ describe Whatsapp::Providers::WhatsappCloudService do
   describe '#send_message' do
     context 'when called' do
       it 'calls message endpoints for normal messages' do
-        stub_request(:post, 'https://graph.facebook.com/v13.0/123456789/messages')
+        stub_request(:post, 'https://graph.facebook.com/v22.0/123456789/messages')
           .with(
             body: {
               messaging_product: 'whatsapp',
@@ -40,7 +40,7 @@ describe Whatsapp::Providers::WhatsappCloudService do
       end
 
       it 'calls message endpoints for a reply to messages' do
-        stub_request(:post, 'https://graph.facebook.com/v13.0/123456789/messages')
+        stub_request(:post, 'https://graph.facebook.com/v22.0/123456789/messages')
           .with(
             body: {
               messaging_product: 'whatsapp',
@@ -60,7 +60,7 @@ describe Whatsapp::Providers::WhatsappCloudService do
         attachment = message.attachments.new(account_id: message.account_id, file_type: :image)
         attachment.file.attach(io: Rails.root.join('spec/assets/avatar.png').open, filename: 'avatar.png', content_type: 'image/png')
 
-        stub_request(:post, 'https://graph.facebook.com/v13.0/123456789/messages')
+        stub_request(:post, 'https://graph.facebook.com/v22.0/123456789/messages')
           .with(
             body: hash_including({
                                    messaging_product: 'whatsapp',
@@ -79,7 +79,7 @@ describe Whatsapp::Providers::WhatsappCloudService do
 
         # ref: https://github.com/bblimke/webmock/issues/900
         # reason for Webmock::API.hash_including
-        stub_request(:post, 'https://graph.facebook.com/v13.0/123456789/messages')
+        stub_request(:post, 'https://graph.facebook.com/v22.0/123456789/messages')
           .with(
             body: hash_including({
                                    messaging_product: 'whatsapp',
@@ -106,7 +106,7 @@ describe Whatsapp::Providers::WhatsappCloudService do
                                        { title: 'Sushi', value: 'Sushi' }
                                      ]
                                    })
-        stub_request(:post, 'https://graph.facebook.com/v13.0/123456789/messages')
+        stub_request(:post, 'https://graph.facebook.com/v22.0/123456789/messages')
           .with(
             body: {
               messaging_product: 'whatsapp', to: '+123456789',
@@ -133,7 +133,7 @@ describe Whatsapp::Providers::WhatsappCloudService do
           sections: [{ rows: %w[Burito Pasta Sushi Salad].map { |i| { id: i, title: i } } }]
         }.to_json
 
-        stub_request(:post, 'https://graph.facebook.com/v13.0/123456789/messages')
+        stub_request(:post, 'https://graph.facebook.com/v22.0/123456789/messages')
           .with(
             body: {
               messaging_product: 'whatsapp', to: '+123456789',
@@ -181,7 +181,7 @@ describe Whatsapp::Providers::WhatsappCloudService do
 
     context 'when called' do
       it 'calls message endpoints with template params for template messages' do
-        stub_request(:post, 'https://graph.facebook.com/v13.0/123456789/messages')
+        stub_request(:post, 'https://graph.facebook.com/v22.0/123456789/messages')
           .with(
             body: template_body.to_json
           )

spec/services/whatsapp/send_on_whatsapp_service_spec.rb

@@ -138,7 +138,7 @@ describe Whatsapp::SendOnWhatsappService do
           processed_params: { 'body' => { 'last_name' => 'Dale', 'ticket_id' => '2332' } }
         }
 
-        stub_request(:post, "https://graph.facebook.com/v13.0/#{whatsapp_cloud_channel.provider_config['phone_number_id']}/messages")
+        stub_request(:post, "https://graph.facebook.com/v22.0/#{whatsapp_cloud_channel.provider_config['phone_number_id']}/messages")
           .with(
             :headers => {
               'Accept' => '*/*',