Code review

pkg/asr/asr.go

@@ -93,25 +93,38 @@ func (t *GroqTranscriber) Transcribe(ctx context.Context, audioFilePath string)
 
 	audioFile, err := os.Open(audioFilePath)
 	if err != nil {
-		return nil, fmt.Errorf("failed to open audio file: %w", err)
+		return nil, fmt.Errorf("failed to open audio file %s: %w", audioFilePath, err)
 	}
 	defer audioFile.Close()
 
 	fileInfo, err := audioFile.Stat()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to stat audio file %s: %w", audioFilePath, err)
 	}
 
 	var requestBody bytes.Buffer
 	writer := multipart.NewWriter(&requestBody)
+
 	part, err := writer.CreateFormFile("file", filepath.Base(audioFilePath))
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to create form file: %w", err)
+	}
+
+	if _, copyErr := io.Copy(part, audioFile); copyErr != nil {
+		return nil, fmt.Errorf("failed to copy audio data: %w", copyErr)
+	}
+
+	if err = writer.WriteField("model", "whisper-large-v3-turbo"); err != nil {
+		return nil, fmt.Errorf("failed to write model field: %w", err)
+	}
+
+	if err = writer.WriteField("response_format", "json"); err != nil {
+		return nil, fmt.Errorf("failed to write response_format field: %w", err)
+	}
+
+	if err = writer.Close(); err != nil {
+		return nil, fmt.Errorf("failed to close multipart writer: %w", err)
 	}
-	io.Copy(part, audioFile)
-	writer.WriteField("model", "whisper-large-v3")
-	writer.WriteField("response_format", "json")
-	writer.Close()
 
 	return t.doRequest(ctx, &requestBody, writer.FormDataContentType(), fileInfo.Size())
 }

pkg/channels/discord/discord.go

@@ -669,6 +669,13 @@ func (c *DiscordChannel) listenVoiceControl(ctx context.Context) {
 }
 
 func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnection, text string) {
+	// Clear cancelTTS when playback finishes (normal or interrupted)
+	defer func() {
+		c.ttsMu.Lock()
+		c.cancelTTS = nil
+		c.ttsMu.Unlock()
+	}()
+
 	sentences := audio.SplitSentences(text)
 	if len(sentences) == 0 {
 		return
@@ -684,6 +691,16 @@ func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnect
 
 	var prefetch chan ttResult
 
+	// Ensure any in-flight prefetch is drained on exit to prevent stream leaks
+	defer func() {
+		if prefetch != nil {
+			result := <-prefetch
+			if result.stream != nil {
+				result.stream.Close()
+			}
+		}
+	}()
+
 	for i, sentence := range sentences {
 		// Check for cancellation (interruption)
 		select {
@@ -730,12 +747,4 @@ func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnect
 
 		prefetch = nextPrefetch
 	}
-
-	// Drain any leftover prefetch
-	if prefetch != nil {
-		result := <-prefetch
-		if result.stream != nil {
-			result.stream.Close()
-		}
-	}
 }

pkg/channels/discord/voice.go

@@ -91,7 +91,7 @@ func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID str
 
 		time.Sleep(250 * time.Millisecond) // Wait a bit for connection to settle
 
-		// Abort if the context has already been cancelled.
+		// Abort if the context has already been canceled.
 		select {
 		case <-ctx.Done():
 			return

pkg/voice/agent.go

@@ -82,6 +82,19 @@ func (a *Agent) Start(ctx context.Context) {
 	logger.InfoCF("voice-agent", "Started Voice Agent orchestrator", nil)
 	go a.listenChunks(ctx)
 	go a.vadTick(ctx)
+
+	// Cleanup sessions on shutdown
+	go func() {
+		<-ctx.Done()
+		a.mu.Lock()
+		for key, acc := range a.sessions {
+			acc.Close()
+			os.Remove(acc.file)
+			delete(a.sessions, key)
+		}
+		a.mu.Unlock()
+		logger.InfoCF("voice-agent", "Cleaned up voice sessions on shutdown", nil)
+	}()
 }
 
 func (a *Agent) listenChunks(ctx context.Context) {
@@ -100,6 +113,12 @@ func (a *Agent) listenChunks(ctx context.Context) {
 }
 
 func (a *Agent) handleChunk(chunk bus.AudioChunk) {
+	// Only accept Opus-encoded audio
+	if chunk.Format != "opus" {
+		logger.DebugCF("voice-agent", "Ignoring unsupported audio format", map[string]any{"format": chunk.Format})
+		return
+	}
+
 	a.mu.Lock()
 	defer a.mu.Unlock()
 
@@ -197,24 +216,29 @@ func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) {
 
 	text := strings.ToLower(strings.TrimSpace(res.Text))
 	if strings.Contains(text, "leave the voice channel") || strings.Contains(text, "leave voice") ||
-		strings.Contains(text, "disconnect voice") {
+		strings.Contains(text, "disconnect voice") || strings.Contains(text, "leave the channel") ||
+		strings.Contains(text, "leave channel") {
 		logger.InfoCF("voice-agent", "Voice command triggered: leave", nil)
-		a.bus.PublishVoiceControl(ctx, bus.VoiceControl{
+		if err := a.bus.PublishVoiceControl(ctx, bus.VoiceControl{
 			SessionID: acc.sessionID,
 			Type:      "command",
 			Action:    "leave",
-		})
-		a.bus.PublishOutbound(ctx, bus.OutboundMessage{
+		}); err != nil {
+			logger.ErrorCF("voice-agent", "Failed to publish leave control", map[string]any{"error": err})
+		}
+		if err := a.bus.PublishOutbound(ctx, bus.OutboundMessage{
 			Channel: channelType,
 			ChatID:  acc.chatID,
 			Content: "Goodbye! Leaving the voice channel.",
-		})
+		}); err != nil {
+			logger.ErrorCF("voice-agent", "Failed to publish goodbye message", map[string]any{"error": err})
+		}
 		return
 	}
 
 	oralPrompt := "\n\n[SYSTEM]: The user just spoke this to you over voice chat. Please reply in a highly concise, conversational, oral style suitable for text-to-speech. Do not use markdown, emojis, asterisks, or code blocks. Speak naturally."
 
-	a.bus.PublishInbound(ctx, bus.InboundMessage{
+	if err := a.bus.PublishInbound(ctx, bus.InboundMessage{
 		Channel:  channelType,
 		SenderID: acc.speakerID,
 		ChatID:   acc.chatID,
@@ -223,5 +247,7 @@ func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) {
 		Metadata: map[string]string{
 			"is_voice": "true",
 		},
-	})
+	}); err != nil {
+		logger.ErrorCF("voice-agent", "Failed to publish inbound message", map[string]any{"error": err})
+	}
 }