fix(whatsapp-call): correct audio duration + diarized transcripts

Two bugs tracked to the same root cause: the media-server's recorder was
writing both the customer and agent RTP streams into a single oggwriter.
The two streams have unrelated SSRCs, sequence numbers, and timestamp
bases, so the resulting OGG carried non-monotonic granule positions.
Browsers compute duration from granule positions → the audio element
reported inflated times. Whisper parsed the broken container and
returned garbage, which showed up as nonsense transcripts.

media-server changes
- Recorder no longer writes to a combined oggwriter during the call.
  Each direction is captured into its own clean OGG (customer-only and
  agent-only).
- At Finalize, the two files are mixed with ffmpeg into a single
  combined.ogg that has coherent OGG pages and correct duration. If
  ffmpeg isn't on PATH we fall back to the customer-side file alone so
  the pipeline doesn't break on bare hosts.
- GET /sessions/:id/recording now accepts ?side=customer|agent so Rails
  can fetch the per-direction files for diarization.

Rails changes
- MediaServerClient#download_recording takes a `side:` keyword.
- CallTranscriptionService prefers per-side transcription when the
  session has a media_session_id: downloads both files, runs Whisper
  with response_format=verbose_json for segment timestamps, interleaves
  by timestamp and labels lines Customer: / Agent:. Falls back to the
  combined recording if per-side downloads fail.
- Lowered Whisper temperature from 0.4 to 0.2 for more deterministic
  transcripts.

Setup: ffmpeg is now required for correct combined playback and
transcription. Install on macOS with `brew install ffmpeg`; on Linux
install the distro's ffmpeg package.

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

enterprise/app/services/whatsapp/call_transcription_service.rb

@@ -28,23 +28,76 @@ class Whatsapp::CallTranscriptionService < Llm::LegacyBaseOpenAiService
       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
-    temp_file_path = fetch_audio_file
-    transcribed_text = nil
-
-    File.open(temp_file_path, 'rb') do |file|
-      response = @client.audio.transcribe(
-        parameters: { model: WHISPER_MODEL, file: file, temperature: 0.4 }
-      )
-      transcribed_text = response['text']
+    if call.media_session_id.present?
+      diarized = diarized_transcript
+      return diarized if diarized.present?
     end
 
-    transcribed_text
-  ensure
-    FileUtils.rm_f(temp_file_path) if temp_file_path.present?
+    transcribe_combined
   end
 
-  def fetch_audio_file
+  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)
@@ -59,6 +112,13 @@ class Whatsapp::CallTranscriptionService < Llm::LegacyBaseOpenAiService
     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?
 

enterprise/app/services/whatsapp/media_server_client.rb

@@ -30,10 +30,12 @@ class Whatsapp::MediaServerClient
     post("/sessions/#{session_id}/terminate")
   end
 
-  def download_recording(session_id)
-    response = execute_request(:get, "/sessions/#{session_id}/recording")
+  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: status=#{response.code}"
+      Rails.logger.error "[MEDIA SERVER] Recording download failed: side=#{side.inspect} status=#{response.code}"
       raise SessionError, "Recording download failed (#{response.code})"
     end
     response.body

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

@@ -3,9 +3,11 @@
 package media
 
 import (
+	"errors"
 	"fmt"
 	"log/slog"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"sync"
 	"time"
@@ -14,23 +16,27 @@ import (
 	"github.com/pion/webrtc/v4/pkg/media/oggwriter"
 )
 
-// Recorder writes incoming Opus RTP packets to an OGG container file in real
-// time. Two separate OGG files are maintained -- one for each audio channel
-// (customer and agent) -- to enable stereo separation for transcription.
-// A combined mono file is also written for playback convenience.
+// 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
 
-	// combinedWriter writes all audio to a single OGG file (for playback).
-	combinedWriter *oggwriter.OggWriter
-	combinedFile   string
+	// combinedFile is produced by ffmpeg at finalize; never written to directly.
+	combinedFile string
 
-	// customerWriter writes only customer audio (for transcription L channel).
+	// customerWriter writes only customer audio (Meta-side track).
 	customerWriter *oggwriter.OggWriter
 	customerFile   string
 
-	// agentWriter writes only agent audio (for transcription R channel).
+	// agentWriter writes only agent audio (browser-side track).
 	agentWriter *oggwriter.OggWriter
 	agentFile   string
 
@@ -40,10 +46,10 @@ type Recorder struct {
 }
 
 // NewRecorder creates a new recorder that writes OGG/Opus files to the given
-// directory. Three files are created:
-//   - {sessionID}.ogg (combined audio for playback)
-//   - {sessionID}_customer.ogg (customer channel only)
-//   - {sessionID}_agent.ogg (agent channel only)
+// 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)
@@ -53,34 +59,26 @@ func NewRecorder(sessionID, dir string) (*Recorder, error) {
 	customerFile := filepath.Join(dir, sessionID+"_customer.ogg")
 	agentFile := filepath.Join(dir, sessionID+"_agent.ogg")
 
-	// Opus at 48kHz, mono for each individual channel.
-	combinedWriter, err := oggwriter.New(combinedFile, 48000, 1)
-	if err != nil {
-		return nil, fmt.Errorf("create combined OGG writer: %w", err)
-	}
-
 	customerWriter, err := oggwriter.New(customerFile, 48000, 1)
 	if err != nil {
-		combinedWriter.Close()
 		return nil, fmt.Errorf("create customer OGG writer: %w", err)
 	}
 
 	agentWriter, err := oggwriter.New(agentFile, 48000, 1)
 	if err != nil {
-		combinedWriter.Close()
 		customerWriter.Close()
 		return nil, fmt.Errorf("create agent OGG writer: %w", err)
 	}
 
 	slog.Info("recorder: started",
 		"session_id", sessionID,
-		"combined_file", combinedFile,
+		"customer_file", customerFile,
+		"agent_file", agentFile,
 	)
 
 	return &Recorder{
 		sessionID:      sessionID,
 		dir:            dir,
-		combinedWriter: combinedWriter,
 		combinedFile:   combinedFile,
 		customerWriter: customerWriter,
 		customerFile:   customerFile,
@@ -91,8 +89,7 @@ func NewRecorder(sessionID, dir string) (*Recorder, error) {
 }
 
 // WriteCustomerRTP writes an RTP packet from the customer's audio stream
-// (Meta-side, Peer A) to the recording. The packet is written to both the
-// combined file and the customer-only channel file.
+// (Meta-side, Peer A) to the customer-only OGG file.
 func (r *Recorder) WriteCustomerRTP(pkt *rtp.Packet) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -101,17 +98,14 @@ func (r *Recorder) WriteCustomerRTP(pkt *rtp.Packet) error {
 		return nil
 	}
 
-	if err := r.combinedWriter.WriteRTP(pkt); err != nil {
-		return fmt.Errorf("write customer RTP to combined: %w", err)
-	}
 	if err := r.customerWriter.WriteRTP(pkt); err != nil {
-		return fmt.Errorf("write customer RTP to channel: %w", err)
+		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 recording.
+// (browser-side, Peer B) to the agent-only OGG file.
 func (r *Recorder) WriteAgentRTP(pkt *rtp.Packet) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -120,17 +114,17 @@ func (r *Recorder) WriteAgentRTP(pkt *rtp.Packet) error {
 		return nil
 	}
 
-	if err := r.combinedWriter.WriteRTP(pkt); err != nil {
-		return fmt.Errorf("write agent RTP to combined: %w", err)
-	}
 	if err := r.agentWriter.WriteRTP(pkt); err != nil {
-		return fmt.Errorf("write agent RTP to channel: %w", err)
+		return fmt.Errorf("write agent RTP: %w", err)
 	}
 	return nil
 }
 
-// Finalize closes all OGG writers and flushes data to disk. After finalization,
-// further writes are silently ignored. This method is idempotent.
+// 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()
@@ -141,9 +135,6 @@ func (r *Recorder) Finalize() error {
 	r.finalized = true
 
 	var errs []error
-	if err := r.combinedWriter.Close(); err != nil {
-		errs = append(errs, fmt.Errorf("close combined writer: %w", err))
-	}
 	if err := r.customerWriter.Close(); err != nil {
 		errs = append(errs, fmt.Errorf("close customer writer: %w", err))
 	}
@@ -151,6 +142,10 @@ func (r *Recorder) Finalize() error {
 		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)
 	}
@@ -158,10 +153,82 @@ func (r *Recorder) Finalize() error {
 	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

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

@@ -445,7 +445,9 @@ func (h *Handlers) TerminateSession(w http.ResponseWriter, r *http.Request) {
 }
 
 // GetRecording handles GET /sessions/{id}/recording. It serves the combined
-// recording file as a binary OGG download.
+// 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.
 func (h *Handlers) GetRecording(w http.ResponseWriter, r *http.Request) {
 	sessionID := r.PathValue("id")
 	sess := h.manager.GetSession(sessionID)
@@ -454,7 +456,19 @@ func (h *Handlers) GetRecording(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	filePath := sess.RecordingFilePath()
+	var filePath, filename string
+	switch r.URL.Query().Get("side") {
+	case "customer":
+		filePath = sess.RecorderCustomerPath()
+		filename = sessionID + "_customer.ogg"
+	case "agent":
+		filePath = sess.RecorderAgentPath()
+		filename = sessionID + "_agent.ogg"
+	default:
+		filePath = sess.RecordingFilePath()
+		filename = sessionID + ".ogg"
+	}
+
 	if filePath == "" {
 		writeError(w, http.StatusNotFound, "no recording available")
 		return
@@ -474,7 +488,7 @@ func (h *Handlers) GetRecording(w http.ResponseWriter, r *http.Request) {
 	}
 
 	w.Header().Set("Content-Type", "audio/ogg")
-	w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s.ogg"`, sessionID))
+	w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename))
 	http.ServeContent(w, r, filePath, stat.ModTime(), f)
 }
 

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

@@ -398,6 +398,22 @@ func (s *Session) RecordingFilePath() string {
 	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 {