update with streaming support

pkg/audio/sentence.go

@@ -0,0 +1,96 @@
+package audio
+
+import (
+	"strings"
+	"unicode"
+)
+
+// SplitSentences splits text into sentence-sized chunks suitable for TTS synthesis.
+// It splits on sentence-ending punctuation (.!?\n) while avoiding false splits
+// on abbreviations and decimal numbers. Very short fragments are merged with
+// the next sentence to prevent choppy playback.
+func SplitSentences(text string) []string {
+	if text == "" {
+		return nil
+	}
+
+	var sentences []string
+	var current strings.Builder
+	runes := []rune(text)
+
+	for i := 0; i < len(runes); i++ {
+		r := runes[i]
+		current.WriteRune(r)
+
+		if r == '\n' {
+			s := strings.TrimSpace(current.String())
+			if s != "" {
+				sentences = append(sentences, s)
+			}
+			current.Reset()
+			continue
+		}
+
+		if r == '.' || r == '!' || r == '?' {
+			// Avoid splitting on decimal numbers like "3.14"
+			if r == '.' && i > 0 && unicode.IsDigit(runes[i-1]) &&
+				i+1 < len(runes) && unicode.IsDigit(runes[i+1]) {
+				continue
+			}
+
+			// Consume trailing punctuation and spaces (e.g., "..." or "?!")
+			for i+1 < len(runes) && (runes[i+1] == '.' || runes[i+1] == '!' || runes[i+1] == '?' || runes[i+1] == ' ') {
+				i++
+				current.WriteRune(runes[i])
+			}
+
+			s := strings.TrimSpace(current.String())
+			if s != "" {
+				sentences = append(sentences, s)
+			}
+			current.Reset()
+		}
+	}
+
+	// Flush remaining text
+	if s := strings.TrimSpace(current.String()); s != "" {
+		sentences = append(sentences, s)
+	}
+
+	// Merge very short fragments with the next sentence
+	return mergeShorties(sentences, 15)
+}
+
+// mergeShorties merges sentences shorter than minLen characters with the following sentence.
+func mergeShorties(sentences []string, minLen int) []string {
+	if len(sentences) <= 1 {
+		return sentences
+	}
+
+	var merged []string
+	var buf string
+
+	for _, s := range sentences {
+		if buf != "" {
+			buf += " " + s
+			if len([]rune(buf)) >= minLen {
+				merged = append(merged, buf)
+				buf = ""
+			}
+		} else if len([]rune(s)) < minLen {
+			buf = s
+		} else {
+			merged = append(merged, s)
+		}
+	}
+
+	if buf != "" {
+		if len(merged) > 0 {
+			merged[len(merged)-1] += " " + buf
+		} else {
+			merged = append(merged, buf)
+		}
+	}
+
+	return merged
+}

pkg/channels/discord/discord.go

@@ -3,6 +3,7 @@ package discord
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"os"
@@ -14,6 +15,7 @@ import (
 	"github.com/bwmarrin/discordgo"
 	"github.com/gorilla/websocket"
 
+	"github.com/sipeed/picoclaw/pkg/audio"
 	"github.com/sipeed/picoclaw/pkg/bus"
 	"github.com/sipeed/picoclaw/pkg/channels"
 	"github.com/sipeed/picoclaw/pkg/config"
@@ -45,6 +47,10 @@ type DiscordChannel struct {
 	botUserID  string                   // stored for mention checking
 	bus        *bus.MessageBus
 	tts        tts.TTSProvider
+
+	// TTS interruption: cancel active playback when user speaks
+	ttsMu     sync.Mutex
+	cancelTTS context.CancelFunc
 }
 
 func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordChannel, error) {
@@ -151,7 +157,16 @@ func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro
 	if c.tts != nil {
 		if ch, err := c.session.State.Channel(channelID); err == nil && ch.GuildID != "" {
 			if vc, ok := c.session.VoiceConnections[ch.GuildID]; ok && vc != nil {
-				go c.playTTS(context.Background(), vc, msg.Content)
+				// Cancel any previous TTS playback
+				c.ttsMu.Lock()
+				if c.cancelTTS != nil {
+					c.cancelTTS()
+				}
+				ttsCtx, ttsCancel := context.WithCancel(context.Background())
+				c.cancelTTS = ttsCancel
+				c.ttsMu.Unlock()
+
+				go c.playTTS(ttsCtx, vc, msg.Content)
 			}
 		}
 	}
@@ -652,14 +667,73 @@ func (c *DiscordChannel) listenVoiceControl(ctx context.Context) {
 }
 
 func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnection, text string) {
-	stream, err := c.tts.Synthesize(ctx, text)
+	sentences := audio.SplitSentences(text)
-	if err != nil {
+	if len(sentences) == 0 {
-		logger.ErrorCF("discord", "TTS synthesize failed", map[string]any{"error": err.Error()})
 		return
 	}
-	defer stream.Close()
 
-	if err := streamOggOpusToDiscord(vc, stream); err != nil {
+	logger.InfoCF("discord", "Starting streamed TTS", map[string]any{"sentences": len(sentences)})
-		logger.ErrorCF("discord", "TTS playback failed", map[string]any{"error": err.Error()})
+
+	// Pipeline: prefetch next sentence's audio while playing current
+	type ttResult struct {
+		stream io.ReadCloser
+		err    error
+	}
+
+	var prefetch chan ttResult
+
+	for i, sentence := range sentences {
+		// Check for cancellation (interruption)
+		select {
+		case <-ctx.Done():
+			logger.InfoCF("discord", "TTS interrupted", map[string]any{"at_sentence": i})
+			return
+		default:
+		}
+
+		// Start prefetching the NEXT sentence while we process the current one
+		var nextPrefetch chan ttResult
+		if i+1 < len(sentences) {
+			nextPrefetch = make(chan ttResult, 1)
+			nextSentence := sentences[i+1]
+			go func() {
+				s, e := c.tts.Synthesize(ctx, nextSentence)
+				nextPrefetch <- ttResult{s, e}
+			}()
+		}
+
+		// Get the current sentence's audio
+		var stream io.ReadCloser
+		var err error
+
+		if prefetch != nil {
+			// Use prefetched result from previous iteration
+			result := <-prefetch
+			stream, err = result.stream, result.err
+		} else {
+			// First sentence: synthesize directly
+			stream, err = c.tts.Synthesize(ctx, sentence)
+		}
+
+		if err != nil {
+			logger.ErrorCF("discord", "TTS synthesize failed", map[string]any{"error": err.Error(), "sentence": i})
+			prefetch = nextPrefetch
+			continue
+		}
+
+		if err := streamOggOpusToDiscord(ctx, vc, stream); err != nil {
+			logger.ErrorCF("discord", "TTS playback failed", map[string]any{"error": err.Error(), "sentence": i})
+		}
+		stream.Close()
+
+		prefetch = nextPrefetch
+	}
+
+	// Drain any leftover prefetch
+	if prefetch != nil {
+		result := <-prefetch
+		if result.stream != nil {
+			result.stream.Close()
+		}
 	}
 }

pkg/channels/discord/voice.go

@@ -2,6 +2,7 @@ package discord
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"time"
@@ -46,7 +47,7 @@ func VoiceReceiveActive(vc *discordgo.VoiceConnection) bool {
 	return vc != nil && vc.OpusRecv != nil
 }
 
-func streamOggOpusToDiscord(vc *discordgo.VoiceConnection, r io.Reader) error {
+func streamOggOpusToDiscord(ctx context.Context, vc *discordgo.VoiceConnection, r io.Reader) error {
 	// Wait for the speaking transition to register
 	vc.Speaking(true)
 	defer vc.Speaking(false)
@@ -55,6 +56,13 @@ func streamOggOpusToDiscord(vc *discordgo.VoiceConnection, r io.Reader) error {
 	header := make([]byte, 27)
 
 	for {
+		// Check for interruption
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
 		if _, err := io.ReadFull(r, header); err != nil {
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				return nil
@@ -84,8 +92,11 @@ func streamOggOpusToDiscord(vc *discordgo.VoiceConnection, r io.Reader) error {
 				if len(packet) > 0 {
 					// Ignore Ogg Opus headers
 					if !bytes.HasPrefix(packet, []byte("OpusHead")) && !bytes.HasPrefix(packet, []byte("OpusTags")) {
-						// Pacing is handled natively by vc.OpusSend blocking (it has an internal ticker)
+						select {
-						vc.OpusSend <- packet
+						case <-ctx.Done():
+							return ctx.Err()
+						case vc.OpusSend <- packet:
+						}
 					}
 					// Start new packet
 					packet = nil
@@ -118,6 +129,8 @@ func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID str
 	})
 
 	var sequence uint64 = 0
+	var interruptCount int
+	var lastInterruptAt time.Time
 
 	for {
 		select {
@@ -139,6 +152,26 @@ func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID str
 				continue
 			}
 
+			// Interruption detection: if user sends voice while TTS is playing,
+			// cancel TTS after a short debounce (3 packets in 200ms)
+			now := time.Now()
+			if now.Sub(lastInterruptAt) > 500*time.Millisecond {
+				interruptCount = 0
+			}
+			interruptCount++
+			lastInterruptAt = now
+
+			if interruptCount >= 3 {
+				c.ttsMu.Lock()
+				if c.cancelTTS != nil {
+					c.cancelTTS()
+					c.cancelTTS = nil
+					logger.InfoCF("discord", "TTS interrupted by user voice", nil)
+				}
+				c.ttsMu.Unlock()
+				interruptCount = 0
+			}
+
 			sequence++
 
 			chunk := bus.AudioChunk{