From 9679cd2b35f54c4d3580ac8b422ee2c76ab19f5f Mon Sep 17 00:00:00 2001 From: Tanmay Deep Sharma Date: Tue, 21 Apr 2026 18:51:19 +0700 Subject: [PATCH] fix(whatsapp-call): correct audio duration + diarized transcripts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../whatsapp/call_transcription_service.rb | 84 ++++++++-- .../services/whatsapp/media_server_client.rb | 8 +- .../media-server/internal/media/recorder.go | 145 +++++++++++++----- .../media-server/internal/server/handlers.go | 20 ++- .../media-server/internal/session/session.go | 16 ++ 5 files changed, 216 insertions(+), 57 deletions(-) diff --git a/enterprise/app/services/whatsapp/call_transcription_service.rb b/enterprise/app/services/whatsapp/call_transcription_service.rb index fa8918df2..5a7602090 100644 --- a/enterprise/app/services/whatsapp/call_transcription_service.rb +++ b/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('%02d:%02d', mins: mins, secs: secs) + end + def update_call_and_message(transcribed_text) return if transcribed_text.blank? diff --git a/enterprise/app/services/whatsapp/media_server_client.rb b/enterprise/app/services/whatsapp/media_server_client.rb index 8d6af6203..19b5566aa 100644 --- a/enterprise/app/services/whatsapp/media_server_client.rb +++ b/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 diff --git a/enterprise/media-server/internal/media/recorder.go b/enterprise/media-server/internal/media/recorder.go index f759fc8f3..ef37734e0 100644 --- a/enterprise/media-server/internal/media/recorder.go +++ b/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 diff --git a/enterprise/media-server/internal/server/handlers.go b/enterprise/media-server/internal/server/handlers.go index 5719a3fa0..703d01bae 100644 --- a/enterprise/media-server/internal/server/handlers.go +++ b/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) } diff --git a/enterprise/media-server/internal/session/session.go b/enterprise/media-server/internal/session/session.go index 7d819b2ca..b143f177b 100644 --- a/enterprise/media-server/internal/session/session.go +++ b/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 {